feat: check de colisión de tema (topic.collision) en rules + validate

Dos posts sobre el mismo caso se canibalizan en la SERP (2026-07-10: se
publicó un segundo Kecksburg con otro ya programado, y había un título
casi gemelo del de Malmstrom en la cola). Nueva función pura
topic_collision(post, corpus) en seo_rules.py — corpus-aware, fuera de
RULES — que dispara por: caso+año compartidos (años < 2020; los de la
era de noticias no son identidad de caso), hooks de título ≥70%
similares (calibrado: el par nuclear da 0.742, el par legítimo más
cercano 0.65), o slugs ≥50% solapados. seo_validate.py obtiene el corpus
via ghst-en (--limit all) y lo reporta como grupo topic_collision;
--no-collision lo omite para follow-ups intencionados.

Calibrado contra los 36 posts reales: dispara solo el duplicado
Kecksburg (HIGH) y el par Grusch preview/evento (MED, intencionado).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
ChemaVX
2026-07-10 11:06:39 +00:00
co-authored by Claude Fable 5
parent beae7c154a
commit bae1cc949c
2 changed files with 137 additions and 3 deletions
+31 -3
View File
@@ -39,9 +39,31 @@ GROUPS = [
("feature_image", ("feature_image.",)),
("feature_image_alt", ("feature_image_alt.",)),
("internal_links", ("internal_links.",)),
("topic_collision", ("topic.",)),
]
def fetch_corpus():
"""All published+scheduled posts (id, title, slug, status) via ghst-en, for
the topic-collision check. Returns [] (with a warning) if the fetch fails —
the validator still runs the per-post rules."""
fd, path = tempfile.mkstemp(suffix=".json", prefix="seo_corpus_")
os.close(fd)
try:
cmd = (f"ghst-en --json post list --limit all "
f"--fields id,title,slug,status > {path}")
r = subprocess.run(cmd, shell=True, stderr=subprocess.PIPE, text=True)
if r.returncode != 0:
print(f" [warn] corpus fetch failed — topic collision NOT checked:\n{r.stderr}",
file=sys.stderr)
return []
data = json.load(open(path))
posts = data.get("posts", data) if isinstance(data, dict) else data
return [p for p in posts if p.get("status") in ("published", "scheduled")]
finally:
os.unlink(path)
def fetch_one(selector, value):
"""Fetch a single post (draft or published) via ghst-en → post dict.
@@ -76,10 +98,13 @@ def load_json(path):
return _unwrap(json.load(open(path)))
def validate(post):
"""Run the shared engine and print a per-group PASS/FAIL report.
def validate(post, corpus=None):
"""Run the shared engine (plus the corpus-aware topic-collision check when a
corpus is given) and print a per-group PASS/FAIL report.
Returns the number of actionable (non-INFO) violations."""
violations = R.check_post(post)
if corpus:
violations.extend(R.topic_collision(post, corpus))
actionable = [v for v in violations if v.severity > R.INFO]
info = [v for v in violations if v.severity == R.INFO]
@@ -129,6 +154,8 @@ def main():
g.add_argument("--slug")
g.add_argument("--id")
g.add_argument("--json", metavar="FILE")
ap.add_argument("--no-collision", action="store_true",
help="skip the topic-collision check (intentional follow-up piece)")
args = ap.parse_args()
if args.json:
@@ -138,7 +165,8 @@ def main():
else:
post = fetch_one("id", args.id)
n = validate(post)
corpus = [] if args.no_collision else fetch_corpus()
n = validate(post, corpus)
sys.exit(1 if n else 0)