From 7a995da88f2ae1819c21a4c74ad07f35de197c79 Mon Sep 17 00:00:00 2001 From: ChemaVX Date: Thu, 25 Jun 2026 14:14:39 +0000 Subject: [PATCH] feat(seo): wire autofill into publish path behind SEO_AUTOFILL (default off, dry-run support) Co-Authored-By: Claude Opus 4.8 --- src/bot/bot.py | 21 +- src/config.py | 18 ++ src/generator/generator.py | 260 +++++++++++++--- src/seo/autofill.py | 623 +++++++++++++++++++++++++++++++++++++ 4 files changed, 870 insertions(+), 52 deletions(-) create mode 100644 src/seo/autofill.py 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