feat(seo): aviso de colisión de tema en /generate blog en

Al publicar el draft EN en Ghost, el título propuesto se compara contra
los posts published+scheduled del sitio (corpus vía Admin API — incluye
la cola programada, justo el caso del doble Kecksburg del 2026-07-10) con
el topic_collision vendorizado. Si colisiona, el notice de Telegram lleva
un bloque "🚨 Posible colisión de tema" en las tres rutas (live, dryrun,
bare). Nunca bloquea: el draft se crea igual, el humano decide (fusionar,
retitular o enlazar a propósito). Solo lang=en (stopwords inglesas).
Títulos ajenos saneados de entidades Markdown (regla de _safe_send).
Aislamiento: cualquier fallo del check → notice sin bloque y warning en
logs, jamás rompe la publicación.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/generator/generator.py

@@ -612,6 +612,7 @@ class OutputGenerator:
             return ""
         title = _extract_title(full_output) or topic
         mode = _resolve_seo_mode(seo_override)
+        collision_note = await self._collision_note(lang, title)
 
         if mode in ("on", "dryrun"):
             try:
@@ -636,13 +637,17 @@ class OutputGenerator:
                     # 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, inserted_pairs)
+                    self.last_publish_notice = (
+                        _seo_dryrun_message(ghost, post, seo, inserted_pairs)
+                        + collision_note)
                 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, inserted_pairs)
+                    self.last_publish_notice = (
+                        _seo_live_message(ghost, post, seo, inserted_pairs)
+                        + collision_note)
                 logger.info("Auto-published blog to Ghost",
                             mode=mode, post_id=post["id"], links=len(inserted_pairs))
                 return ""
@@ -661,11 +666,27 @@ class OutputGenerator:
             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)
+            return _bare_ghost_notice(ghost, post) + collision_note
         except Exception as e:
             logger.warning("Auto-publish to Ghost failed", error=str(e))
             return ""
 
+    async def _collision_note(self, lang: str, title: str) -> str:
+        """Aviso de canibalización del título propuesto contra los posts
+        published+scheduled del sitio. Solo EN (las stopwords del motor son
+        inglesas). Nunca lanza y nunca bloquea: el draft se publica igual y
+        el aviso viaja en el notice de Telegram.
+        """
+        if lang != "en":
+            return ""
+        try:
+            from src.seo.autofill import collision_notice, fetch_collision_corpus
+            corpus = await fetch_collision_corpus(lang)
+            return collision_notice(title, corpus) or ""
+        except Exception as e:
+            logger.warning("Topic collision check failed — skipped", error=str(e))
+            return ""
+
     async def generate(self, session_id: int, output_type: OutputType,
                        progress_callback=None, lang: str = "es",
                        seo_override: str | None = None) -> str:

src/seo/autofill.py

@@ -650,3 +650,62 @@ def insert_internal_links(
                          phrase=phrase, slug=slug)
 
     return "".join(tokens), inserted_pairs
+
+# ─── 4. Topic collision (aviso pre-publish) ──────────────────────────────────
+
+_MD_UNSAFE = re.compile(r"[*_`\[\]]")
+
+
+def _slugify_title(title: str) -> str:
+    """Aproximación del slug que Ghost generará del título — el draft aún no
+    tiene slug real, y topic_collision usa el slug como una de sus señales."""
+    return re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-")
+
+
+async def fetch_collision_corpus(lang: str) -> list[dict]:
+    """Posts published+scheduled del sitio (id, slug, title, status) para el
+    check de colisión de tema. A diferencia de fetch_published_menu incluye
+    los PROGRAMADOS: chocar con la cola de vacaciones es justo el caso a
+    cazar (2026-07-10: segundo Kecksburg publicado con otro ya en cola).
+    Aislamiento total: cualquier fallo → [] y log, nunca raise.
+    """
+    try:
+        # Lazy import to avoid a heavy/circular import at module load.
+        from src.generator.generator import GhostPublisher
+
+        pub = GhostPublisher(lang=lang)
+        if not pub.is_configured():
+            return []
+        data = await pub._admin_get(
+            "posts/?filter=status:[published,scheduled]"
+            "&fields=id,slug,title,status&limit=all",
+            timeout=30,
+        )
+        if data is None:
+            return []
+        corpus = [p for p in data.get("posts", []) if p.get("slug")]
+        logger.info("seo.collision: corpus fetched", lang=lang, count=len(corpus))
+        return corpus
+    except Exception as e:  # noqa: BLE001 — isolation guarantee
+        logger.warning("seo.collision: corpus fetch failed", lang=lang, error=str(e))
+        return []
+
+
+def collision_notice(title: str, corpus: list[dict]) -> str | None:
+    """Aviso (Markdown seguro para Telegram) si el título propuesto colisiona
+    con un post existente, vía el motor vendorizado R.topic_collision.
+    None si no hay colisión. Los títulos ajenos se sanean de entidades
+    Markdown para no romper el parseo del mensaje (ver regla de _safe_send).
+    """
+    if not corpus:
+        return None
+    candidate = {"id": None, "title": title, "slug": _slugify_title(title)}
+    violations = R.topic_collision(candidate, corpus)
+    if not violations:
+        return None
+    lines = [_MD_UNSAFE.sub("", v.message) for v in violations[:3]]
+    return (
+        "\n\n🚨 *Posible colisión de tema* — el draft se ha creado igualmente:\n"
+        + "\n".join(f"• {line}" for line in lines)
+        + "\nAntes de publicar: fusionar, retitular a otro ángulo o enlazar a propósito."
+    )

tests/test_seo_autofill.py

@@ -37,3 +37,47 @@ def test_system_prompt_lists_allowed_tags_per_lang():
     assert 'never use "investigacion"' in en
     assert 'never use "investigacion"' not in es
     assert "ONLY from this exact list" in es
+
+
+# ─── topic collision ─────────────────────────────────────────────────────────
+
+from src.seo.autofill import collision_notice, _slugify_title
+
+CORPUS = [
+    {"id": "1", "status": "published", "slug": "kecksburg-1965-acorn-ufo-missing-nasa-files",
+     "title": 'Kecksburg 1965: The Acorn-Shaped Object, the Missing NASA Files, and "Pennsylvania\'s Roswell"'},
+    {"id": "2", "status": "scheduled", "slug": "uss-russell-2019-pyramid-uap-channel-islands",
+     "title": "USS Russell 2019: The Pyramid UAP Video and the Channel Islands Drone Swarm"},
+]
+
+
+def test_collision_fires_on_same_case_and_year():
+    note = collision_notice("Kecksburg 1965: New Acorn Evidence", CORPUS)
+    assert note is not None
+    assert "kecksburg" in note.lower()
+    assert "1965" in note
+
+
+def test_collision_none_on_distinct_case():
+    assert collision_notice("Tehran 1976: The Jet-Disabling Encounter", CORPUS) is None
+
+
+def test_collision_none_on_empty_corpus():
+    assert collision_notice("Kecksburg 1965: Anything", []) is None
+
+
+def test_collision_note_is_markdown_safe():
+    corpus = [{"id": "9", "status": "published", "slug": "weird-1990-case",
+               "title": "Weird *1990* [Case] with_underscores and `ticks`"}]
+    note = collision_notice("Weird 1990: Case Revisited", corpus)
+    assert note is not None
+    # las entidades Markdown de títulos ajenos se sanean (solo quedan las nuestras)
+    bullets = [line for line in note.split("\n") if line.startswith("• ")]
+    assert bullets
+    for line in bullets:
+        for ch in "*_`[]":
+            assert ch not in line
+
+
+def test_slugify_title():
+    assert _slugify_title("USS Russell 2019: The Pyramid UAP!") == "uss-russell-2019-the-pyramid-uap"