feat(seo): wire autofill into publish path behind SEO_AUTOFILL (default off, dry-run support)
Build & Deploy ResearchOwl / build-and-push (push) Successful in 7s
Build & Deploy ResearchOwl / build-and-push (push) Successful in 7s
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
ChemaVX
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
89aa8d6030
commit
7a995da88f
@@ -0,0 +1,623 @@
|
||||
"""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 not yet constrained, so we don't enforce
|
||||
# it (model picks freely there until an ES vocab is pinned down).
|
||||
ALLOWED_TAGS = {
|
||||
"en": [
|
||||
"uap", "declassified", "military-cases", "classic-cases",
|
||||
"investigation", "spain-cases", "latin-america",
|
||||
],
|
||||
# "es": [...] # TODO: pin the live ES tag vocabulary before constraining.
|
||||
}
|
||||
|
||||
# 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
|
||||
import aiohttp as _aio
|
||||
|
||||
pub = GhostPublisher(lang=lang)
|
||||
if not pub.is_configured():
|
||||
logger.warning("seo.menu: Ghost not configured for lang", lang=lang)
|
||||
return []
|
||||
|
||||
token = pub._make_token()
|
||||
url = (
|
||||
f"{pub.url}/ghost/api/admin/posts/"
|
||||
"?filter=status:published&fields=id,slug,title&limit=all"
|
||||
)
|
||||
timeout = _aio.ClientTimeout(total=30)
|
||||
async with _aio.ClientSession(timeout=timeout) as sess:
|
||||
async with sess.get(
|
||||
url,
|
||||
headers={
|
||||
"Authorization": f"Ghost {token}",
|
||||
"Accept-Version": "v5.0",
|
||||
},
|
||||
) as resp:
|
||||
if resp.status != 200:
|
||||
body = await resp.text()
|
||||
logger.warning("seo.menu: non-200 from Ghost",
|
||||
status=resp.status, body=body[:200])
|
||||
return []
|
||||
data = await resp.json()
|
||||
|
||||
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"
|
||||
tag_clause = (
|
||||
f"TAGS: choose 2-4 tags ONLY from this exact list (never invent a tag, "
|
||||
f"never translate it, never use \"investigacion\"): {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)
|
||||
|
||||
# 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)
|
||||
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 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, links_inserted).
|
||||
"""
|
||||
if not html or not suggestions:
|
||||
return html, 0
|
||||
|
||||
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
|
||||
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
|
||||
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
|
||||
Reference in New Issue
Block a user