diff --git a/src/bot/bot.py b/src/bot/bot.py
index fa1a1af..f7a5a96 100644
--- a/src/bot/bot.py
+++ b/src/bot/bot.py
@@ -278,7 +278,11 @@ async def cmd_generate(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
chat_id = update.effective_chat.id
output_arg = ctx.args[0].lower() if ctx.args else ""
- lang = "en" if len(ctx.args) > 1 and ctx.args[1].lower() == "en" else "es"
+ rest = [a.lower() for a in ctx.args[1:]]
+ lang = "en" if "en" in rest else "es"
+ # `/generate blog en dry` forces SEO dry-run for this one call (proposes SEO to
+ # Telegram, writes a bare draft). Global default still comes from SEO_AUTOFILL.
+ seo_override = "dryrun" if ("dry" in rest or "dryrun" in rest) else None
type_map = {
"podcast": OutputType.PODCAST,
@@ -347,7 +351,8 @@ async def cmd_generate(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
processor = ContentProcessor(db, ollama)
generator = OutputGenerator(db, ollama, processor)
- output = await generator.generate(session_id, output_type, gen_progress, lang=lang)
+ output = await generator.generate(session_id, output_type, gen_progress,
+ lang=lang, seo_override=seo_override)
# Send as file if very long
if len(output) > 8000:
@@ -384,6 +389,18 @@ async def cmd_generate(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
else:
await send_chunked(update.message, output)
+ # SEO autofill / dry-run summary — a SEPARATE short message so it is never
+ # buried inside the long .md document. None on the flag-off path.
+ if getattr(generator, "last_publish_notice", None):
+ try:
+ await update.message.reply_text(
+ generator.last_publish_notice,
+ parse_mode=ParseMode.MARKDOWN,
+ disable_web_page_preview=True,
+ )
+ except Exception as e:
+ logger.warning("Failed to send SEO summary message", error=str(e))
+
try:
stats = await db.get_usage_stats(session_id)
total_cost = sum(s.get("total_cost", 0) for s in stats)
diff --git a/src/config.py b/src/config.py
index 76cbe44..f29730c 100644
--- a/src/config.py
+++ b/src/config.py
@@ -48,6 +48,15 @@ class Settings(BaseSettings):
ghost_url_en: str = Field("", env="GHOST_URL_EN")
ghost_api_key_en: str = Field("", env="GHOST_API_KEY_EN")
+ # SEO autofill — "off" | "on" | "dryrun" (default off).
+ # off = today's exact behavior (bare draft, no second LLM call).
+ # on = adds best-effort meta/OG/Twitter/tags/internal-links to the DRAFT
+ # (status stays "draft" — NEVER auto-publishes; see src/seo/autofill.py).
+ # dryrun = runs the full pipeline + reports the proposed SEO to Telegram, but
+ # writes a BARE draft (no seo fields), for inspection before trusting
+ # the write path. A `/generate blog en dry` arg forces dryrun per-call.
+ seo_autofill: str = Field("off", env="SEO_AUTOFILL")
+
# Alerts
cost_alert_threshold: float = Field(0.15, env="COST_ALERT_THRESHOLD")
@@ -61,6 +70,15 @@ class Settings(BaseSettings):
return []
return [int(uid.strip()) for uid in self.telegram_allowed_users.split(",") if uid.strip()]
+ @property
+ def seo_autofill_enabled(self) -> bool:
+ """True when autofill should run (either live 'on' or 'dryrun')."""
+ return (self.seo_autofill or "").strip().lower() in ("on", "true", "1", "yes", "dryrun")
+
+ @property
+ def seo_autofill_dryrun(self) -> bool:
+ return (self.seo_autofill or "").strip().lower() == "dryrun"
+
class Config:
env_file = ".env"
diff --git a/src/generator/generator.py b/src/generator/generator.py
index 1793af6..5aca152 100644
--- a/src/generator/generator.py
+++ b/src/generator/generator.py
@@ -290,8 +290,94 @@ def _seo_checklist(slug: str) -> str:
)
+# Language-aware default tag when no seo/tags are supplied. Fixes the historical
+# hardcoded "investigacion" that mis-tagged EN posts; EN canonical is "investigation".
+_DEFAULT_TAG = {"en": "investigation", "es": "investigacion"}
+
+
+def _resolve_seo_mode(override: str | None = None) -> str:
+ """Effective SEO autofill mode: 'off' | 'on' | 'dryrun'.
+ A per-call override ('dryrun'/'on', e.g. from `/generate blog en dry`) wins;
+ otherwise it derives from settings.seo_autofill."""
+ if override in ("on", "dryrun"):
+ return override
+ if settings.seo_autofill_dryrun:
+ return "dryrun"
+ if settings.seo_autofill_enabled:
+ return "on"
+ return "off"
+
+
+def _seo_checklist_slim(slug: str) -> str:
+ """Slimmed checklist for the autofill path: meta/OG/excerpt/tags/links are
+ already on the draft, so only the human-only bits remain."""
+ return (
+ "⚠️ Antes de publicar: añade *feature image + alt*, luego:\n"
+ f"`seo-check {slug}`"
+ )
+
+
+def _bare_ghost_notice(ghost: "GhostPublisher", post: dict) -> str:
+ """Today's exact draft-published notice (editor link + full checklist).
+ Used for the flag-off path and as the autofill failure fallback."""
+ return (
+ f"\n\n---\n"
+ f"📤 *Borrador publicado en Ghost*\n"
+ f"Editar: {ghost.url}/ghost/#/editor/post/{post['id']}"
+ f"{_seo_checklist(post.get('slug', ''))}"
+ )
+
+
+def _seo_link_summary(seo: dict, n_links: int) -> str:
+ pairs = [(l["phrase"], l["slug"]) for l in seo.get("internal_links", [])]
+ return f"{n_links} insertados" + (f" — {pairs}" if pairs else "")
+
+
+def _seo_live_message(ghost: "GhostPublisher", post: dict, seo: dict, n_links: int) -> str:
+ """Separate Telegram message for the live autofill path (SEO written to draft)."""
+ slug = post.get("slug", "")
+ warns = seo.get("seo_warnings") or []
+ warn_line = ("\n⚠️ revisar: " + "; ".join(warns)) if warns else ""
+ return (
+ f"📤 *Borrador en Ghost — SEO autorrelleno*\n"
+ f"Editar: {ghost.url}/ghost/#/editor/post/{post['id']}\n\n"
+ f"🔖 *SEO escrito en el draft:*\n"
+ f"• meta title ({len(seo['meta_title'])}): {seo['meta_title']}\n"
+ f"• meta desc ({len(seo['meta_description'])}): {seo['meta_description']}\n"
+ f"• excerpt: {len(seo['custom_excerpt'])} car.\n"
+ f"• tags: {', '.join(seo['tags'])}\n"
+ f"• internal links: {_seo_link_summary(seo, n_links)}"
+ f"{warn_line}\n\n"
+ f"🖼️ *Imagen sugerida:*\n"
+ f"`imagefinder --query \"{seo['image_query']}\" --context \"{seo['image_context']}\" --lang {ghost.lang}`\n\n"
+ f"{_seo_checklist_slim(slug)}"
+ )
+
+
+def _seo_dryrun_message(ghost: "GhostPublisher", post: dict, seo: dict, n_links: int) -> str:
+ """Separate Telegram message for DRY-RUN: bare draft created, SEO only proposed."""
+ slug = post.get("slug", "")
+ warns = seo.get("seo_warnings") or []
+ warn_line = ("\n⚠️ revisar: " + "; ".join(warns)) if warns else ""
+ return (
+ f"🧪 *DRY-RUN SEO* — draft creado SIN escribir SEO (solo revisión)\n"
+ f"Editar: {ghost.url}/ghost/#/editor/post/{post['id']}\n\n"
+ f"🔖 *SEO propuesto (NO escrito en Ghost):*\n"
+ f"• meta title ({len(seo['meta_title'])}): {seo['meta_title']}\n"
+ f"• meta desc ({len(seo['meta_description'])}): {seo['meta_description']}\n"
+ f"• excerpt ({len(seo['custom_excerpt'])} car.): {seo['custom_excerpt']}\n"
+ f"• tags: {', '.join(seo['tags'])}\n"
+ f"• internal links que se insertarían: {_seo_link_summary(seo, n_links)}"
+ f"{warn_line}\n\n"
+ f"🖼️ *Imagen sugerida:*\n"
+ f"`imagefinder --query \"{seo['image_query']}\" --context \"{seo['image_context']}\" --lang {ghost.lang}`\n\n"
+ f"{_seo_checklist_slim(slug)}"
+ )
+
+
class GhostPublisher:
def __init__(self, lang: str = "es"):
+ self.lang = lang
if lang == "en":
self.url = (settings.ghost_url_en or "").rstrip("/")
self.api_key = settings.ghost_api_key_en or ""
@@ -302,6 +388,16 @@ class GhostPublisher:
def is_configured(self) -> bool:
return bool(self.url and self.api_key)
+ def _build_html(self, markdown_content: str) -> str:
+ """Canonical markdown → Ghost body HTML conversion: strip the ResearchOwl
+ header, render markdown, drop the first (Ghost renders the title).
+ Shared so callers that need the linked HTML produce the SAME html that gets
+ posted (no drift between link-insertion and the mobiledoc body)."""
+ import markdown as _md
+ clean = _strip_researchowl_header(markdown_content)
+ html = _md.markdown(clean, extensions=["extra"])
+ return re.sub(r"]*>.*?
", "", html, count=1, flags=re.DOTALL).lstrip()
+
def _make_token(self) -> str:
key_id, secret = self.api_key.split(":", 1)
now = int(time.time())
@@ -314,18 +410,25 @@ class GhostPublisher:
return f"{signing}.{sig}"
async def publish_draft(self, title: str, markdown_content: str,
- tags: list[str] | None = None) -> dict:
+ tags: list[str] | None = None,
+ seo: dict | None = None,
+ body_html: str | None = None) -> dict:
+ """Create a Ghost DRAFT. status is ALWAYS "draft" — never published.
+
+ Purely-additive SEO hooks (both default None → byte-identical to the
+ original behavior):
+ * body_html — pre-converted + internal-linked HTML from the caller; used
+ verbatim when given, else markdown_content is converted as before.
+ * seo — a generate_seo_fields dict; when given, its meta/OG/Twitter fields
+ are added to the post and its (allow-list) tags are used.
+ """
import aiohttp as _aio
- import markdown as _md
- clean = _strip_researchowl_header(markdown_content)
- html = _md.markdown(clean, extensions=["extra"])
-
- # Ghost añade el título automáticamente — eliminar el primer para evitar duplicado
- html = re.sub(r"]*>.*?
", "", html, count=1, flags=re.DOTALL).lstrip()
+ # Caller-supplied linked HTML wins; otherwise convert markdown as before.
+ html = body_html if body_html is not None else self._build_html(markdown_content)
logger.info("Ghost publish_draft", html_length=len(html),
- html_preview=html[:200])
+ html_preview=html[:200], seo=bool(seo))
if not html.strip():
raise ValueError("Ghost: HTML vacío tras conversión markdown — contenido no enviado")
@@ -342,14 +445,25 @@ class GhostPublisher:
})
token = self._make_token()
- body = {
- "posts": [{
- "title": title,
- "mobiledoc": mobiledoc,
- "status": "draft",
- "tags": [{"name": t} for t in (tags or ["investigacion"])],
- }]
+ # Language-aware default tag (fixes the hardcoded "investigacion" for EN).
+ tag_names = tags or [_DEFAULT_TAG.get(self.lang, "investigation")]
+ post_obj = {
+ "title": title,
+ "mobiledoc": mobiledoc,
+ "status": "draft", # NEVER "published" — draft only, always.
+ "tags": [{"name": t} for t in tag_names],
}
+ if seo:
+ post_obj.update({
+ "meta_title": seo["meta_title"],
+ "meta_description": seo["meta_description"],
+ "custom_excerpt": seo["custom_excerpt"],
+ "og_title": seo["og_title"],
+ "og_description": seo["og_description"],
+ "twitter_title": seo["twitter_title"],
+ "twitter_description": seo["twitter_description"],
+ })
+ body = {"posts": [post_obj]}
async with _aio.ClientSession() as sess:
async with sess.post(
f"{self.url}/ghost/api/admin/posts/",
@@ -372,15 +486,86 @@ class OutputGenerator:
self.db = db
self.ollama = ollama
self.processor = processor
+ # Set during a blog generation when the autofill/dry-run path runs: a short
+ # markdown SEO summary the bot sends as a SEPARATE Telegram message (so it is
+ # never buried inside the long .md document). None on the flag-off path.
+ self.last_publish_notice: str | None = None
+
+ async def _publish_blog_to_ghost(self, lang: str, full_output: str, topic: str,
+ session_id: int, seo_override: str | None) -> str:
+ """Publish a blog DRAFT to Ghost, gated by the SEO autofill mode.
+
+ Returns the ghost_notice to APPEND to the returned document (flag-off /
+ fallback bare-draft path only). For the autofill 'on'/'dryrun' paths it sets
+ self.last_publish_notice (a separate Telegram message) and returns "".
+
+ NEVER raises and NEVER blocks publishing: any autofill failure degrades to
+ today's exact bare-draft publish. status is always "draft".
+ """
+ ghost = GhostPublisher(lang=lang)
+ if not ghost.is_configured():
+ return ""
+ title = _extract_title(full_output) or topic
+ mode = _resolve_seo_mode(seo_override)
+
+ if mode in ("on", "dryrun"):
+ try:
+ from src.seo.autofill import (
+ fetch_published_menu, generate_seo_fields, insert_internal_links,
+ )
+ article_md = _strip_researchowl_header(full_output)
+ menu = await fetch_published_menu(lang)
+ seo = await generate_seo_fields(
+ article_md, menu, lang, title=title,
+ db=self.db, session_id=session_id,
+ )
+ if seo is None:
+ raise RuntimeError("generate_seo_fields returned None")
+ # Build the SAME html that will be posted, then link it deterministically.
+ html = ghost._build_html(article_md)
+ linked_html, n_links = insert_internal_links(
+ html, seo["internal_links"], menu, lang)
+
+ if mode == "dryrun":
+ # Bare draft (today's write path) — do NOT write seo fields; only
+ # surface the proposal so Jose can inspect before trusting writes.
+ result = await ghost.publish_draft(title, full_output)
+ post = result["posts"][0]
+ self.last_publish_notice = _seo_dryrun_message(ghost, post, seo, n_links)
+ else:
+ result = await ghost.publish_draft(
+ title, full_output, tags=seo["tags"], seo=seo,
+ body_html=linked_html)
+ post = result["posts"][0]
+ self.last_publish_notice = _seo_live_message(ghost, post, seo, n_links)
+ logger.info("Auto-published blog to Ghost",
+ mode=mode, post_id=post["id"], links=n_links)
+ return ""
+ except Exception as e:
+ logger.warning("SEO autofill failed; falling back to bare draft",
+ error=str(e))
+ # fall through to the bare-draft publish below
+
+ # OFF mode, or autofill failed → today's exact bare-draft publish.
+ try:
+ result = await ghost.publish_draft(title, full_output)
+ post = result["posts"][0]
+ logger.info("Auto-published blog to Ghost (bare)", post_id=post["id"])
+ return _bare_ghost_notice(ghost, post)
+ except Exception as e:
+ logger.warning("Auto-publish to Ghost failed", error=str(e))
+ return ""
async def generate(self, session_id: int, output_type: OutputType,
- progress_callback=None, lang: str = "es") -> str:
+ progress_callback=None, lang: str = "es",
+ seo_override: str | None = None) -> str:
"""Generate an output for a research session"""
+ self.last_publish_notice = None
if output_type in (OutputType.REPORT_EXTENDED,
OutputType.BLOG_EXTENDED,
OutputType.PODCAST_EXTENDED):
return await self.generate_extended(session_id, output_type, progress_callback,
- lang=lang)
+ lang=lang, seo_override=seo_override)
session = await self.db.get_session(session_id)
if not session:
@@ -426,24 +611,11 @@ class OutputGenerator:
# Save to DB
await self.db.save_output(session_id, output_type, full_output)
- # Auto-publish to Ghost for blog outputs
+ # Auto-publish to Ghost for blog outputs (autofill mode gated inside helper).
ghost_notice = ""
if output_type in (OutputType.BLOG, OutputType.BLOG_EXTENDED):
- ghost = GhostPublisher(lang=lang)
- if ghost.is_configured():
- try:
- title = _extract_title(full_output) or topic
- result = await ghost.publish_draft(title, full_output)
- post = result["posts"][0]
- ghost_notice = (
- f"\n\n---\n"
- f"📤 *Borrador publicado en Ghost*\n"
- f"Editar: {ghost.url}/ghost/#/editor/post/{post['id']}"
- f"{_seo_checklist(post.get('slug', ''))}"
- )
- logger.info("Auto-published blog to Ghost", post_id=post["id"])
- except Exception as e:
- logger.warning("Auto-publish to Ghost failed", error=str(e))
+ ghost_notice = await self._publish_blog_to_ghost(
+ lang, full_output, topic, session_id, seo_override)
logger.info("Output generated", type=output_type, length=len(full_output))
return full_output + ghost_notice
@@ -500,7 +672,8 @@ class OutputGenerator:
return systems.get(output_type, "You are a helpful research assistant.")
async def generate_extended(self, session_id: int, output_type: OutputType,
- progress_callback=None, lang: str = "es") -> str:
+ progress_callback=None, lang: str = "es",
+ seo_override: str | None = None) -> str:
"""
Generación por secciones para outputs exhaustivos.
1. Recupera muestra de contexto para el outline
@@ -601,24 +774,11 @@ class OutputGenerator:
await self.db.save_output(session_id, output_type, full_output)
- # Auto-publish to Ghost for extended blog outputs
+ # Auto-publish to Ghost for extended blog outputs (autofill mode gated inside).
ghost_notice = ""
if output_type == OutputType.BLOG_EXTENDED:
- ghost = GhostPublisher(lang=lang)
- if ghost.is_configured():
- try:
- title = _extract_title(full_output) or topic
- result = await ghost.publish_draft(title, full_output)
- post = result["posts"][0]
- ghost_notice = (
- f"\n\n---\n"
- f"📤 *Borrador publicado en Ghost*\n"
- f"Editar: {ghost.url}/ghost/#/editor/post/{post['id']}"
- f"{_seo_checklist(post.get('slug', ''))}"
- )
- logger.info("Auto-published extended blog to Ghost", post_id=post["id"])
- except Exception as e:
- logger.warning("Auto-publish to Ghost failed (extended)", error=str(e))
+ ghost_notice = await self._publish_blog_to_ghost(
+ lang, full_output, topic, session_id, seo_override)
logger.info("Extended output generated", type=output_type,
sections=len(sections), length=len(full_output))
diff --git a/src/seo/autofill.py b/src/seo/autofill.py
new file mode 100644
index 0000000..92cc66e
--- /dev/null
+++ b/src/seo/autofill.py
@@ -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 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"]*>.*?", "", 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 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 ); 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 … (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"(?"):
+ low = tok.lower()
+ if low.startswith(" 0 or not tok:
+ continue
+ m = pat.search(tok)
+ if not m:
+ continue
+ replacement = (
+ f'{m.group(0)}'
+ )
+ 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