ChemaVXandClaude Fable 5
d31badeb5b
Build & Deploy ResearchOwl / build-and-push (push) Successful in 8s
find_draft_by_title y fetch_published_menu duplicaban línea a línea el GET admin de Ghost (token, URL, ClientSession, dict de headers, status check, resp.json), y el dict Authorization/Accept-Version/Accept-Encoding estaba copiado en 3 sitios — el incidente brotli ya demostró que el fix por copia se deja sitios. Ahora los headers se construyen en un único punto (_admin_headers, con SAFE_ACCEPT_ENCODING de config) y el GET en _admin_get; publish_draft usa los mismos headers. aiohttp pasa a import de módulo en generator.py (era 'import aiohttp as _aio' repetido dentro de dos métodos). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
653 lines
29 KiB
Python
653 lines
29 KiB
Python
"""SEO autofill — best-effort meta/OG/Twitter/tags/internal-links for a draft.
|
||
|
||
Step 2 (this file): pure, testable units. NOTHING here is wired into the live
|
||
publish path yet (that is Step 3, behind `settings.seo_autofill`). Every public
|
||
function is isolation-safe: on ANY failure it degrades (menu → [], fields → None,
|
||
link insertion → unchanged html), so it can never block or break article
|
||
publishing, and it never changes a post's `status` away from `draft`.
|
||
|
||
Three units:
|
||
* fetch_published_menu(lang) — live "menu" of existing posts to link to.
|
||
* generate_seo_fields(...) — one Haiku JSON call → validated SEO dict.
|
||
* insert_internal_links(...) — deterministic, LLM never touches HTML.
|
||
|
||
The SEO rules engine (length limits, link counting) is the SAME vendored module
|
||
the auditor (Tool A) and validator (Tool C) use: src/seo/rules.py.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
|
||
import structlog
|
||
|
||
from src.config import settings
|
||
from src.llm import get_anthropic_client
|
||
from src.seo import rules as R
|
||
|
||
logger = structlog.get_logger(__name__)
|
||
|
||
# Canonical site host per language. The wrapped <a> hrefs use the www host (the
|
||
# site's canonical form); rules.internal_links still counts them because its
|
||
# SITE_HOST ("theexclusionzone.com") is a substring of "www.theexclusionzone.com".
|
||
SITE_BY_LANG = {
|
||
"en": "www.theexclusionzone.com",
|
||
"es": "www.zonadeexclusion.com",
|
||
}
|
||
|
||
# Tag allow-list — the model may ONLY pick from these; invented tags are dropped
|
||
# deterministically after the call (never trust the LLM to self-limit). EN is the
|
||
# live vocabulary Jose confirmed; ES is the live core taxonomy on zonadeexclusion
|
||
# (fetched 2026-07-03), mirroring EN. Deliberately excluded from ES: the one-post
|
||
# long-tail tags the model invented before constraining, and "investigacion-2"
|
||
# (a Ghost case-collision duplicate of "investigacion").
|
||
ALLOWED_TAGS = {
|
||
"en": [
|
||
"uap", "declassified", "military-cases", "classic-cases",
|
||
"investigation", "spain-cases", "latin-america",
|
||
],
|
||
"es": [
|
||
"uap", "desclasificados", "casos-militares", "casos-clasicos",
|
||
"investigacion", "casos-espana",
|
||
],
|
||
}
|
||
|
||
# Language-aware default tag when nothing from the allow-list fits / the model
|
||
# returns none. EN canonical is "investigation" (NOT the legacy "investigacion").
|
||
DEFAULT_TAG = {"en": "investigation", "es": "investigacion"}
|
||
|
||
MAX_INTERNAL_LINKS = 3
|
||
|
||
# Blocking violations at generation time = only the length/missing rules on the
|
||
# fields WE generate. feature_image*, internal_links.too_few, social *.empty and
|
||
# the INFO rules are expected / non-blocking at draft time.
|
||
_BLOCKING_PREFIXES = ("meta_title.", "meta_description.", "custom_excerpt.")
|
||
|
||
|
||
# ─── 1. Published menu ──────────────────────────────────────────────────────
|
||
|
||
async def fetch_published_menu(lang: str) -> list[dict]:
|
||
"""Live list of published posts on the same site/lang: [{slug, title}, ...].
|
||
|
||
Reuses GhostPublisher's per-lang URL + JWT minting. On ANY failure
|
||
(unconfigured, timeout, non-200, bad JSON) → return [] and log; never raise.
|
||
"""
|
||
try:
|
||
# Lazy import to avoid a heavy/circular import at module load.
|
||
from src.generator.generator import GhostPublisher
|
||
|
||
pub = GhostPublisher(lang=lang)
|
||
if not pub.is_configured():
|
||
logger.warning("seo.menu: Ghost not configured for lang", lang=lang)
|
||
return []
|
||
|
||
# _admin_get lleva los headers canónicos (auth, Accept-Version y el
|
||
# Accept-Encoding sin br — ver KNOWN-ISSUES.md) y loguea los non-200.
|
||
data = await pub._admin_get(
|
||
"posts/?filter=status:published&fields=id,slug,title&limit=all",
|
||
timeout=30,
|
||
)
|
||
if data is None:
|
||
return []
|
||
|
||
menu = [
|
||
{"slug": p["slug"], "title": p.get("title", "")}
|
||
for p in data.get("posts", [])
|
||
if p.get("slug")
|
||
]
|
||
logger.info("seo.menu: fetched", lang=lang, count=len(menu))
|
||
return menu
|
||
except Exception as e: # noqa: BLE001 — isolation guarantee
|
||
logger.warning("seo.menu: fetch failed, using empty menu",
|
||
lang=lang, error=str(e))
|
||
return []
|
||
|
||
|
||
# ─── 2. SEO field generation (one Haiku JSON call) ──────────────────────────
|
||
|
||
def _system_prompt(lang: str) -> str:
|
||
allow = ", ".join(ALLOWED_TAGS.get(lang, []))
|
||
out_lang = "SPANISH" if lang == "es" else "ENGLISH"
|
||
# The anti-legacy "never use investigacion" rule is EN-only: on the ES site
|
||
# "investigacion" IS the canonical tag (and the allow-list already excludes
|
||
# the "investigacion-2" duplicate).
|
||
legacy_clause = ", never use \"investigacion\"" if lang == "en" else ""
|
||
tag_clause = (
|
||
f"TAGS: choose 2-4 tags ONLY from this exact list (never invent a tag, "
|
||
f"never translate it{legacy_clause}): {allow}.\n"
|
||
if allow else
|
||
"TAGS: 2-4 lowercase-hyphenated topical tags appropriate to the article.\n"
|
||
)
|
||
return (
|
||
"You are an SEO editor for an investigative blog about UAP/UFO history.\n"
|
||
"You are given a FINISHED article and a MENU of existing published posts on the site.\n"
|
||
"Return ONLY a single JSON object — no prose, no markdown fences — with these fields.\n\n"
|
||
"HARD LIMITS (count characters; never exceed — and aim BELOW the cap for safety):\n"
|
||
f"- meta_title: <= {R.META_TITLE_MAX} characters (aim ~50). Compelling, specific, "
|
||
"front-load the key entity.\n"
|
||
f"- meta_description: <= {R.META_DESC_MAX} characters (aim ~125 — one SHORT sentence). "
|
||
"Earn the click; describe what the article answers, not clickbait.\n"
|
||
f"- custom_excerpt: <= {R.CUSTOM_EXCERPT_MAX} characters (aim ~260). 1-2 sentences "
|
||
"summarizing the article's substance.\n\n"
|
||
+ tag_clause +
|
||
"\nINTERNAL LINKS — quality over count, mirror the site's manual policy:\n"
|
||
"- The \"phrase\" MUST be a SHORT entity name (typically 1-4 words: a person, program, "
|
||
"office, or named incident) copied VERBATIM from the ARTICLE body. NEVER use a menu "
|
||
"title or slug as the phrase — the phrase has to literally appear in the article text.\n"
|
||
"- Link that phrase to a MENU post ONLY if that post's PRIMARY SUBJECT — what it is "
|
||
"actually ABOUT (judge from its slug + title) — IS that same entity. Not merely a post "
|
||
"that mentions it. If unsure the target is really ABOUT the entity, DO NOT link it.\n"
|
||
" GOOD: phrase \"AARO\" → aaro-... post (short phrase from the body; that post is ABOUT AARO).\n"
|
||
" GOOD: phrase \"GIMBAL\" → gimbal-gofast-... post (it is ABOUT the GIMBAL video).\n"
|
||
" BAD: phrase \"AATIP\" → gimbal-gofast-... post (that post is NOT about AATIP). Skip it.\n"
|
||
" BAD: using the full menu title \"AARO's UFO Investigation: What Eight Decades...\" as the "
|
||
"phrase (that string is not in the article body — it will fail to match). Use \"AARO\".\n"
|
||
"- 0 to 3 links. NEVER force a link to hit a number. ONE strong, on-subject link beats two "
|
||
"loose ones. Skip if no target is genuinely about the phrase.\n"
|
||
"- Use only slugs from the MENU. Never link the article to itself.\n"
|
||
"- Give the phrase EXACTLY as it appears in the article (case-sensitive), so it can be matched.\n\n"
|
||
"IMAGE: suggest a concrete image search query (subjects/objects a stock site would have — "
|
||
"NOT the headline) and a one-sentence context describing what the article is about.\n\n"
|
||
f"Write meta_title, meta_description, custom_excerpt, image_query and image_context in {out_lang}.\n\n"
|
||
"JSON schema (all fields required):\n"
|
||
'{"meta_title": "...", "meta_description": "...", "custom_excerpt": "...", '
|
||
'"tags": ["..."], "internal_links": [{"phrase": "...", "slug": "..."}], '
|
||
'"image_query": "...", "image_context": "..."}'
|
||
)
|
||
|
||
|
||
def _user_message(article_text: str, link_menu: list[dict]) -> str:
|
||
menu_lines = "\n".join(
|
||
f"- {m['slug']} — {m.get('title','')}" for m in link_menu
|
||
) or "(no existing posts)"
|
||
return (
|
||
"MENU of existing published posts (slug — title):\n"
|
||
f"{menu_lines}\n\n"
|
||
"ARTICLE:\n"
|
||
f"{article_text}"
|
||
)
|
||
|
||
|
||
def _parse_json_object(text: str) -> dict:
|
||
"""Strip optional ``` fences and parse the first JSON object. Raises on failure."""
|
||
t = text.strip()
|
||
t = re.sub(r"^```(?:json)?\s*", "", t)
|
||
t = re.sub(r"\s*```$", "", t).strip()
|
||
# If extra prose snuck in, grab the outermost {...}.
|
||
if not t.startswith("{"):
|
||
m = re.search(r"\{.*\}", t, re.DOTALL)
|
||
if m:
|
||
t = m.group(0)
|
||
obj = json.loads(t)
|
||
if not isinstance(obj, dict):
|
||
raise ValueError("model did not return a JSON object")
|
||
return obj
|
||
|
||
|
||
_REQUIRED_STR = ("meta_title", "meta_description", "custom_excerpt",
|
||
"image_query", "image_context")
|
||
|
||
|
||
def _coerce(obj: dict, lang: str) -> dict:
|
||
"""Validate shape + constrain tags to the allow-list. Raises on missing required strings."""
|
||
for k in _REQUIRED_STR:
|
||
if not isinstance(obj.get(k), str) or not obj[k].strip():
|
||
raise ValueError(f"missing/empty required field: {k}")
|
||
|
||
raw_tags = obj.get("tags") or []
|
||
if not isinstance(raw_tags, list):
|
||
raw_tags = []
|
||
allow = ALLOWED_TAGS.get(lang)
|
||
if allow:
|
||
seen, tags = set(), []
|
||
allow_set = set(allow)
|
||
for t in raw_tags:
|
||
if isinstance(t, str):
|
||
s = t.strip().lower()
|
||
if s in allow_set and s not in seen:
|
||
seen.add(s)
|
||
tags.append(s)
|
||
if not tags:
|
||
tags = [DEFAULT_TAG.get(lang, "investigation")]
|
||
else:
|
||
tags = [t.strip().lower() for t in raw_tags
|
||
if isinstance(t, str) and t.strip()] or [DEFAULT_TAG.get(lang, "investigacion")]
|
||
|
||
links = []
|
||
for item in (obj.get("internal_links") or []):
|
||
if isinstance(item, dict) and item.get("phrase") and item.get("slug"):
|
||
links.append({"phrase": str(item["phrase"]), "slug": str(item["slug"])})
|
||
|
||
return {
|
||
"meta_title": obj["meta_title"].strip(),
|
||
"meta_description": obj["meta_description"].strip(),
|
||
"custom_excerpt": obj["custom_excerpt"].strip(),
|
||
"tags": tags,
|
||
"internal_links": links,
|
||
"image_query": obj["image_query"].strip(),
|
||
"image_context": obj["image_context"].strip(),
|
||
}
|
||
|
||
|
||
def _markdown_to_html(article_text: str) -> str:
|
||
"""Same conversion publish_draft uses, so validation sees the real body."""
|
||
import markdown as _md
|
||
html = _md.markdown(article_text, extensions=["extra"])
|
||
return re.sub(r"<h1[^>]*>.*?</h1>", "", html, count=1, flags=re.DOTALL).lstrip()
|
||
|
||
|
||
def _synthetic_post(fields: dict, html: str, title: str, slug: str) -> dict:
|
||
mt, md = fields["meta_title"], fields["meta_description"]
|
||
return {
|
||
"html": html,
|
||
"title": title,
|
||
"slug": slug or "",
|
||
"meta_title": mt,
|
||
"meta_description": md,
|
||
"custom_excerpt": fields["custom_excerpt"],
|
||
"og_title": mt, "og_description": md,
|
||
"twitter_title": mt, "twitter_description": md,
|
||
"feature_image": "", # human adds later
|
||
"feature_image_alt": "", # human adds later
|
||
}
|
||
|
||
|
||
def _blocking(violations) -> list:
|
||
return [v for v in violations if v.rule.startswith(_BLOCKING_PREFIXES)]
|
||
|
||
|
||
# Length-limited fields we generate. The retry aims at (limit - margin), well
|
||
# UNDER the hard limit: Haiku cannot count to an exact char count and reliably
|
||
# overshoots its target by 20-50 chars, so the margin must absorb that overshoot.
|
||
# Margins are per-field (bigger for the long free-text fields that overshoot most).
|
||
_LEN_LIMITS = {
|
||
"meta_title": R.META_TITLE_MAX,
|
||
"meta_description": R.META_DESC_MAX,
|
||
"custom_excerpt": R.CUSTOM_EXCERPT_MAX,
|
||
}
|
||
_LEN_MARGIN = {
|
||
"meta_title": 10, # target 50
|
||
"meta_description": 35, # target 110
|
||
"custom_excerpt": 45, # target 255
|
||
}
|
||
|
||
|
||
# Minimum useful length per field. If a CLEAN boundary-trim would drop a field
|
||
# below this, we DON'T trim it — better a slightly-over field a human nudges than
|
||
# a butchered stub. Falls back to path-1 (keep + flag) for that field only.
|
||
_MIN_USEFUL = {
|
||
"meta_title": 25,
|
||
"meta_description": 80,
|
||
"custom_excerpt": 120,
|
||
}
|
||
|
||
# Connective / function words (EN + ES) we must not leave dangling at the end of
|
||
# a word-boundary trim — they all "expect more" after them, so ending on one reads
|
||
# as cut-off. Compared lowercased, after stripping trailing punctuation. Includes
|
||
# correlatives (neither/nor), interrogatives (why/what), and contracted auxiliaries
|
||
# (won't/can't) that surfaced as bad trims in testing.
|
||
_DANGLING_WORDS = {
|
||
# English — articles / prepositions / basic connectives
|
||
"and", "or", "but", "the", "a", "an", "of", "to", "in", "on", "for", "with",
|
||
"at", "by", "from", "as", "that", "which", "is", "was", "were", "this",
|
||
"its", "their", "his", "her", "into", "over", "after", "about", "between",
|
||
"during", "against", "among", "without", "within", "upon", "toward", "towards",
|
||
"via", "per", "amid", "despite", "near", "off", "out", "up", "down",
|
||
# English — correlatives / subordinators / interrogatives / negation
|
||
"nor", "neither", "either", "both", "whether", "while", "since", "though",
|
||
"although", "unless", "until", "because", "if", "than", "then", "yet", "so",
|
||
"not", "no", "why", "how", "when", "where", "what", "who", "whom", "whose",
|
||
# English — contracted auxiliaries (read as mid-clause)
|
||
"won't", "can't", "cannot", "don't", "doesn't", "didn't", "isn't", "aren't",
|
||
"wasn't", "weren't", "hasn't", "haven't", "still", "just", "ever",
|
||
# Spanish
|
||
"y", "o", "u", "pero", "el", "la", "los", "las", "un", "una", "de", "del",
|
||
"en", "con", "por", "para", "que", "su", "sus", "al", "como", "se", "lo",
|
||
"ni", "sino", "porque", "aunque", "mientras", "cuando", "donde", "segun",
|
||
"según", "sin", "entre", "sobre", "tras", "ante", "hacia", "hasta", "desde",
|
||
"ya", "muy", "mas", "más", "menos", "tan", "no",
|
||
}
|
||
|
||
# Characters safe to strip from the end of a trimmed phrase (whitespace, commas,
|
||
# semicolons, colons, dashes/em-dashes). Sentence-final . ! ? are intentionally
|
||
# NOT here — if a trim happens to land on one we keep it.
|
||
_TRAIL_PUNCT = " \t,;:—–- "
|
||
|
||
|
||
def _drop_sentences_to_fit(text: str, limit: int) -> str:
|
||
"""Drop WHOLE trailing sentences until the text fits within `limit`.
|
||
|
||
Splits on sentence terminators (. ! ?) keeping the delimiter, then returns the
|
||
longest run of complete leading sentences that is <= limit. Never a mid-
|
||
sentence cut. Returns "" if even the first sentence is over limit (caller's
|
||
min-useful guardrail then keeps the original)."""
|
||
parts = [p for p in re.findall(r"[^.!?]*[.!?]+|\S[^.!?]*$", text.strip()) if p.strip()]
|
||
if not parts:
|
||
return ""
|
||
out = ""
|
||
for p in parts:
|
||
candidate = out + p
|
||
if len(candidate.strip()) <= limit:
|
||
out = candidate
|
||
else:
|
||
break
|
||
return out.strip()
|
||
|
||
|
||
def _trim_to_word_boundary(text: str, limit: int) -> str:
|
||
"""Trim to the last WORD boundary that fits, then strip trailing punctuation
|
||
and any dangling connective word. No mid-word cut, no ellipsis. The result
|
||
reads as a complete-ish phrase. Returns "" if nothing survives the cleanup."""
|
||
text = text.strip()
|
||
if len(text) <= limit:
|
||
return text
|
||
out = ""
|
||
for w in text.split():
|
||
candidate = w if not out else f"{out} {w}"
|
||
if len(candidate) <= limit:
|
||
out = candidate
|
||
else:
|
||
break
|
||
# Clean the tail: alternately strip trailing punctuation and dangling words
|
||
# until the phrase ends on a content word.
|
||
while out:
|
||
stripped = out.rstrip(_TRAIL_PUNCT)
|
||
if stripped != out:
|
||
out = stripped
|
||
continue
|
||
# Capture the trailing word INCLUDING internal apostrophes, so contractions
|
||
# ("won't", "doesn't") match the dangling list instead of just their tail.
|
||
m = re.search(r"([^\W\d_]+(?:['’][^\W\d_]+)*)$", out, re.UNICODE)
|
||
if m and m.group(1).lower() in _DANGLING_WORDS:
|
||
out = out[:m.start()].rstrip(_TRAIL_PUNCT)
|
||
continue
|
||
break
|
||
return out.strip()
|
||
|
||
|
||
def _shorten_over_limit(fields: dict) -> tuple[dict, list[dict]]:
|
||
"""FINAL, deterministic, boundary-aware shortener — runs only after the LLM +
|
||
retry have tried, only on fields STILL over their hard limit. This is NOT the
|
||
forbidden ugly truncation: custom_excerpt drops whole trailing sentences; the
|
||
single-line metas trim to a word boundary and clean dangling punctuation.
|
||
|
||
Quality guardrail: if a clean shorten would fall below the field's minimum
|
||
useful length, we keep the LLM's over-limit text and flag it (path-1 fallback
|
||
for that field only). Mutates `fields` in place; also returns it.
|
||
|
||
Returns (fields, log) where log entries are
|
||
{field, before, after, applied, reason} for reporting/auditing."""
|
||
log: list[dict] = []
|
||
for field, limit in _LEN_LIMITS.items():
|
||
before = fields.get(field, "")
|
||
if len(before) <= limit:
|
||
continue
|
||
if field == "custom_excerpt":
|
||
after = _drop_sentences_to_fit(before, limit)
|
||
else:
|
||
# Single-line metas: prefer a clean WHOLE-sentence prefix (many metas
|
||
# are 2 sentences — keeping just the first reads as a complete thought).
|
||
# Only fall back to a word-boundary trim if that prefix is too short.
|
||
after = _drop_sentences_to_fit(before, limit)
|
||
if not after or len(after) < _MIN_USEFUL[field]:
|
||
after = _trim_to_word_boundary(before, limit)
|
||
|
||
if after and len(after) <= limit and len(after) >= _MIN_USEFUL[field]:
|
||
fields[field] = after
|
||
log.append({"field": field, "before": before, "after": after,
|
||
"applied": True, "reason": "shortened"})
|
||
logger.info("seo.shorten: field shortened",
|
||
field=field, before_len=len(before), after_len=len(after))
|
||
else:
|
||
reason = "would-be-too-short" if after else "no-clean-boundary"
|
||
log.append({"field": field, "before": before, "after": before,
|
||
"applied": False, "reason": reason})
|
||
logger.info("seo.shorten: kept over-limit (clean trim unsafe)",
|
||
field=field, before_len=len(before),
|
||
trimmed_len=len(after), reason=reason)
|
||
return fields, log
|
||
|
||
|
||
def _retry_instruction(fields: dict) -> str:
|
||
"""Forceful, field-specific shrink instruction listing actual length, hard
|
||
limit, the exact overage, and a sub-limit target. Empty string if nothing over."""
|
||
lines = []
|
||
for field, limit in _LEN_LIMITS.items():
|
||
cur = len(fields.get(field, ""))
|
||
if cur > limit:
|
||
target = limit - _LEN_MARGIN[field]
|
||
cut = cur - target
|
||
lines.append(
|
||
f"- {field} is currently {cur} characters. The HARD limit is {limit}. "
|
||
f"Rewrite it to AT MOST {target} characters (aim for {target}, NOT {limit}; "
|
||
f"shorter is better than longer) — cut at least {cut} characters by removing a "
|
||
f"clause, adjective, or example. Keep the same meaning and language."
|
||
)
|
||
if not lines:
|
||
return ""
|
||
return (
|
||
"\n\nSTOP. YOUR PREVIOUS OUTPUT EXCEEDED A HARD CHARACTER LIMIT. "
|
||
"Fix ONLY the field(s) below; leave every other field exactly as you had it:\n"
|
||
+ "\n".join(lines)
|
||
+ "\nReturn the FULL JSON object again with ALL fields present, only these shortened. "
|
||
"When in doubt, cut MORE — a shorter field is always acceptable, an over-limit one is not."
|
||
)
|
||
|
||
|
||
async def generate_seo_fields(
|
||
article_text: str,
|
||
link_menu: list[dict],
|
||
lang: str,
|
||
*,
|
||
title: str = "",
|
||
slug: str = "",
|
||
db=None,
|
||
session_id: int | None = None,
|
||
) -> dict | None:
|
||
"""Second, dedicated Haiku call → validated SEO dict, or None on hard failure.
|
||
|
||
Returns: meta_title, meta_description, custom_excerpt, tags[],
|
||
internal_links[{phrase,slug}], image_query, image_context, og_/twitter_*
|
||
(derived deterministically), and seo_warnings[] for any non-fatal issues.
|
||
On ANY exception → None (caller falls back to a bare draft).
|
||
"""
|
||
try:
|
||
client = get_anthropic_client()
|
||
system = _system_prompt(lang)
|
||
user = _user_message(article_text, link_menu)
|
||
|
||
async def _raw(messages: list, max_tokens: int = 1024,
|
||
temperature: float | None = None) -> str:
|
||
kwargs = {
|
||
"model": settings.claude_model,
|
||
"max_tokens": max_tokens,
|
||
"system": system,
|
||
"messages": messages,
|
||
}
|
||
if temperature is not None:
|
||
kwargs["temperature"] = temperature
|
||
msg = await client.messages.create(**kwargs)
|
||
if db is not None and session_id is not None:
|
||
try:
|
||
await db.log_api_call(
|
||
session_id, "seo_fields", settings.claude_model,
|
||
msg.usage.input_tokens, msg.usage.output_tokens,
|
||
)
|
||
except Exception as log_err: # noqa: BLE001
|
||
logger.warning("seo.fields: usage log failed", error=str(log_err))
|
||
return msg.content[0].text
|
||
|
||
base_msgs = [{"role": "user", "content": user}]
|
||
text1 = await _raw(base_msgs)
|
||
fields = _coerce(_parse_json_object(text1), lang)
|
||
fields["internal_links"] = _sanitize_links(fields["internal_links"], link_menu)
|
||
|
||
# Validate against the shared engine using the real body (md→html + links).
|
||
body_html = _markdown_to_html(article_text)
|
||
linked_html, _ = insert_internal_links(body_html, fields["internal_links"], link_menu, lang)
|
||
violations = R.check_post(_synthetic_post(fields, linked_html, title, slug))
|
||
blocking = _blocking(violations)
|
||
|
||
if blocking:
|
||
detail = "; ".join(
|
||
f"{v.rule}: {v.message}" for v in blocking
|
||
)
|
||
instr = _retry_instruction(fields)
|
||
logger.info("seo.fields: blocking violation, one stricter retry", detail=detail)
|
||
try:
|
||
# Real edit turn: hand the model its OWN previous JSON to shorten,
|
||
# rather than re-rolling from scratch (which kept overshooting). Lower
|
||
# max_tokens to physically discourage rambling.
|
||
retry_msgs = base_msgs + [
|
||
{"role": "assistant", "content": text1},
|
||
{"role": "user", "content": instr},
|
||
]
|
||
# temperature=0 so the shorten instruction binds deterministically.
|
||
retry = _coerce(_parse_json_object(
|
||
await _raw(retry_msgs, max_tokens=768, temperature=0.0)), lang)
|
||
retry["internal_links"] = _sanitize_links(retry["internal_links"], link_menu)
|
||
rlinked, _ = insert_internal_links(
|
||
_markdown_to_html(article_text), retry["internal_links"], link_menu, lang)
|
||
rviol = R.check_post(_synthetic_post(retry, rlinked, title, slug))
|
||
if not _blocking(rviol):
|
||
fields, violations, blocking = retry, rviol, []
|
||
else:
|
||
# Keep the retry's text but record it stayed over (never truncate).
|
||
fields, violations, blocking = retry, rviol, _blocking(rviol)
|
||
except Exception as e: # noqa: BLE001
|
||
logger.warning("seo.fields: retry failed, keeping first output", error=str(e))
|
||
|
||
# Final boundary-aware shortener — only for fields the LLM + retry left
|
||
# over limit. Clean (sentence-drop / word-boundary), never mid-word, and
|
||
# skipped (kept + flagged) if it would butcher the field below min-useful.
|
||
shorten_log: list[dict] = []
|
||
if blocking:
|
||
fields, shorten_log = _shorten_over_limit(fields)
|
||
slinked, _ = insert_internal_links(
|
||
_markdown_to_html(article_text), fields["internal_links"], link_menu, lang)
|
||
violations = R.check_post(_synthetic_post(fields, slinked, title, slug))
|
||
blocking = _blocking(violations)
|
||
|
||
mt, md = fields["meta_title"], fields["meta_description"]
|
||
fields["og_title"] = fields["twitter_title"] = mt
|
||
fields["og_description"] = fields["twitter_description"] = md
|
||
fields["seo_warnings"] = [
|
||
f"{v.rule}: {v.message}" for v in blocking
|
||
]
|
||
if shorten_log:
|
||
fields["seo_shortened"] = [
|
||
{"field": e["field"], "applied": e["applied"], "reason": e["reason"]}
|
||
for e in shorten_log
|
||
]
|
||
return fields
|
||
except Exception as e: # noqa: BLE001 — isolation guarantee
|
||
logger.warning("seo.fields: generation failed, falling back to bare draft",
|
||
error=str(e))
|
||
return None
|
||
|
||
|
||
# ─── 3. Deterministic internal-link insertion ──────────────────────────────
|
||
|
||
def _sanitize_links(suggestions: list[dict], menu: list[dict]) -> list[dict]:
|
||
"""Defense-in-depth: drop malformed link suggestions before insertion/reporting.
|
||
|
||
The "phrase" must be SHORT verbatim ARTICLE text — never a slug or a menu
|
||
title. A suggestion is dropped when its phrase equals its own slug, any menu
|
||
slug, or any menu title (the slug-as-phrase / title-as-phrase traps). Even if
|
||
the model ignores the prompt rule, the output stays clean. Drops are logged at
|
||
debug and never raise; survivors still go through the verbatim-in-body guard.
|
||
"""
|
||
menu_slugs = {(m.get("slug") or "").strip().casefold() for m in menu}
|
||
menu_titles = {(m.get("title") or "").strip().casefold() for m in menu}
|
||
menu_slugs.discard("")
|
||
menu_titles.discard("")
|
||
clean: list[dict] = []
|
||
for s in suggestions:
|
||
phrase = (s.get("phrase") or "").strip()
|
||
slug = (s.get("slug") or "").strip()
|
||
if not phrase or not slug:
|
||
continue
|
||
pf = phrase.casefold()
|
||
if pf == slug.casefold() or pf in menu_slugs or pf in menu_titles:
|
||
logger.debug("seo.links: dropped malformed suggestion (slug/title as phrase)",
|
||
phrase=phrase, slug=slug)
|
||
continue
|
||
clean.append({"phrase": phrase, "slug": slug})
|
||
return clean
|
||
|
||
|
||
def insert_internal_links(
|
||
html: str,
|
||
suggestions: list[dict],
|
||
menu: list[dict],
|
||
lang: str = "en",
|
||
) -> tuple[str, int]:
|
||
"""Wrap the FIRST verbatim, word-boundary occurrence of each suggested phrase
|
||
in an <a> to its menu slug. Deterministic — the LLM never edits HTML.
|
||
|
||
Rules: slug must be in the menu; phrase must appear verbatim in a TEXT node
|
||
(never inside a tag, never inside an existing <a>); word-boundary match (no
|
||
partial words); each phrase/slug used at most once; cap at MAX_INTERNAL_LINKS.
|
||
A phrase that is missing or already linked is silently skipped (never forced,
|
||
never an error). Returns (modified_html, inserted_pairs) where inserted_pairs
|
||
is the list of {phrase, slug} dicts actually wrapped (len == links inserted).
|
||
"""
|
||
if not html or not suggestions:
|
||
return html, []
|
||
|
||
site = SITE_BY_LANG.get(lang, SITE_BY_LANG["en"])
|
||
menu_slugs = {m["slug"] for m in menu if m.get("slug")}
|
||
|
||
# Split into tag tokens and text tokens; we only ever edit text tokens, and
|
||
# only when not inside an <a>…</a> (anchor depth 0). This guarantees no nested
|
||
# links and no edits inside tag attributes.
|
||
tokens = re.split(r"(<[^>]+>)", html)
|
||
inserted = 0
|
||
inserted_pairs: list[dict] = []
|
||
used_phrases: set[str] = set()
|
||
used_slugs: set[str] = set()
|
||
|
||
for sug in suggestions:
|
||
if inserted >= MAX_INTERNAL_LINKS:
|
||
break
|
||
phrase = (sug.get("phrase") or "").strip()
|
||
slug = (sug.get("slug") or "").strip()
|
||
if not phrase or not slug:
|
||
continue
|
||
if slug not in menu_slugs:
|
||
continue
|
||
if phrase in used_phrases or slug in used_slugs:
|
||
continue
|
||
|
||
pat = re.compile(r"(?<!\w)" + re.escape(phrase) + r"(?!\w)")
|
||
anchor_depth = 0
|
||
done = False
|
||
for i, tok in enumerate(tokens):
|
||
if tok.startswith("<") and tok.endswith(">"):
|
||
low = tok.lower()
|
||
if low.startswith("<a") and not low.startswith("</a"):
|
||
anchor_depth += 1
|
||
elif low.startswith("</a"):
|
||
anchor_depth = max(0, anchor_depth - 1)
|
||
continue
|
||
if anchor_depth > 0 or not tok:
|
||
continue
|
||
m = pat.search(tok)
|
||
if not m:
|
||
continue
|
||
replacement = (
|
||
f'<a href="https://{site}/{slug}/">{m.group(0)}</a>'
|
||
)
|
||
tokens[i] = tok[:m.start()] + replacement + tok[m.end():]
|
||
inserted += 1
|
||
inserted_pairs.append({"phrase": phrase, "slug": slug})
|
||
used_phrases.add(phrase)
|
||
used_slugs.add(slug)
|
||
done = True
|
||
break
|
||
if not done:
|
||
logger.debug("seo.links: phrase not found / already linked, skipped",
|
||
phrase=phrase, slug=slug)
|
||
|
||
return "".join(tokens), inserted_pairs
|