diff --git a/seo_semantic.py b/seo_semantic.py new file mode 100644 index 0000000..f87026c --- /dev/null +++ b/seo_semantic.py @@ -0,0 +1,333 @@ +#!/usr/bin/env python3 +""" +Capa semántica sobre los blogs — enlaces internos afines y canibalización. + +Embebe cada artículo con bge-m3 (el que ya corre en ollama, 1024d) y sobre los +vectores contesta dos preguntas: + + · afines → los N artículos PUBLICADOS más parecidos, para sugerir + enlaces internos al rematar un borrador. Es la palanca + de indexación que queda después de IndexNow. + · canibalizacion → pares demasiado parecidos DENTRO de un sitio: compiten + por la misma búsqueda y se hacen sombra. + +NUNCA escribe en el blog. Solo imprime. Quien escribe el enlace es seo_link.py, +y publicar es decisión de Jose. + +⚠️ NUNCA se cruzan los dos sitios. bge-m3 es multilingüe y da 0,97 entre un +artículo y su traducción: cruzarlos marcaría todo el espejo ES↔EN como +canibalización, que es exactamente lo contrario de la verdad (son sitios +distintos, idiomas distintos, no compiten entre sí). + +⚠️ Los destinos que se sugieren son solo los PUBLICADOS. Enlazar a un post +programado da 404 en la ventana intermedia (cicatriz del caso Wilson-Davis → +MJ-12, ver seo_link.py). + +Uso: + python3 seo_semantic.py afines --site es [-n 4] [--json] + python3 seo_semantic.py canibalizacion --site es [--umbral 0.86] [--json] + python3 seo_semantic.py estado + python3 seo_semantic.py distribucion --site es # para calibrar el umbral + +El caché vive en ~/.local/state/seo-semantic/.json y se re-embebe solo lo +que cambia (hash del texto). 80 artículos enteros son ~3 s en la iGPU. +""" +import argparse +import hashlib +import html as H +import json +import os +import re +import subprocess +import sys +import urllib.error +import urllib.request + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from seo_link import fetch_corpus # noqa: E402 (mismo corpus y misma guarda de truncado) + +MODELO = "bge-m3" +MAX_CHARS = 24000 # ~7k tokens; el contexto de bge-m3 son 8192 +ESTADO = os.path.expanduser("~/.local/state/seo-semantic") +SITIOS = { + "es": {"cli": "ghst-es", "base": "https://zonadeexclusion.com"}, + "en": {"cli": "ghst-en", "base": "https://www.theexclusionzone.com"}, +} + + +# ---------- endpoint de ollama ---------- + +def endpoint(): + """ClusterIP del Service, NO la IP del pod. + + La del pod cambia en cada reinicio (era el gotcha apuntado el 2026-07-24). + La ClusterIP es estable y el host la alcanza por las reglas de kube-proxy, + sin NodePort ni tocar la red. El Ingress no sirve: está tras Authentik. + """ + if os.environ.get("OLLAMA_URL"): + return os.environ["OLLAMA_URL"].rstrip("/") + r = subprocess.run( + ["kubectl", "get", "svc", "ollama", "-n", "ollama", + "-o", "jsonpath={.spec.clusterIP}"], + capture_output=True, text=True, timeout=30) + ip = r.stdout.strip() + if not ip: + sys.exit("✗ no se pudo resolver la ClusterIP de ollama " + f"(kubectl dijo: {r.stderr.strip()[:200]})") + return f"http://{ip}:11434" + + +def embed(textos, url): + """Un solo POST en lote. Sin keep_alive: bge-m3 es de producción (lo usa + researchowl) y mandar keep_alive:0 lo DESALOJARÍA, cobrándole a otro la + recarga.""" + datos = json.dumps({"model": MODELO, "input": textos}).encode() + req = urllib.request.Request(f"{url}/api/embed", data=datos, + headers={"Content-Type": "application/json"}) + try: + with urllib.request.urlopen(req, timeout=300) as r: + vs = json.load(r)["embeddings"] + except urllib.error.URLError as e: + sys.exit(f"✗ ollama no responde en {url}: {e}") + if len(vs) != len(textos): + sys.exit(f"✗ ollama devolvió {len(vs)} vectores para {len(textos)} textos") + return vs + + +# ---------- corpus ---------- + +def enlaces_de(post, base): + """Slugs internos ya enlazados desde el cuerpo. + + Sin esto la herramienta sugiere enlazar a sitios que YA están enlazados, que + es ruido y encima invita a meter un dentro de otro. Se mira el `html` + que renderiza Ghost, así que da igual el formato de origen (lexical, + mobiledoc…), el mismo motivo por el que seo_link.py lee html para analizar. + """ + h = post.get("html") or "" + dominio = base.split("//", 1)[1].replace("www.", "") + out = set() + for href in re.findall(r'href=["\']([^"\']+)["\']', h): + if href.startswith("/"): + out.add(href.strip("/").split("/")[0]) + elif dominio in href: + resto = href.split(dominio, 1)[1].strip("/") + if resto: + out.add(resto.split("/")[0].split("?")[0].split("#")[0]) + return sorted(out) + + +def texto_de(post): + """Título + entradilla + cuerpo en plano. El título pesa poco en un texto + de 10k caracteres, así que se repite para que cuente algo.""" + h = post.get("html") or "" + cuerpo = re.sub(r"\s+", " ", H.unescape(re.sub(r"<[^>]+>", " ", h))).strip() + cabeza = " ".join(filter(None, [ + post.get("title"), post.get("title"), + post.get("custom_excerpt") or post.get("excerpt") or "", + ])) + return (cabeza + " " + cuerpo)[:MAX_CHARS] + + +def carga(sitio, refrescar=False, verboso=True): + """Devuelve {slug: registro} con los vectores al día. + + Solo se re-embebe lo que cambió: el hash es del texto que se embebe, no del + post entero, así que un cambio de metadatos no dispara trabajo inútil. + """ + cfg = SITIOS[sitio] + os.makedirs(ESTADO, exist_ok=True) + ruta = os.path.join(ESTADO, f"{sitio}.json") + cache = {} + if os.path.exists(ruta) and not refrescar: + with open(ruta) as f: + cache = json.load(f) + + posts = fetch_corpus(cfg["cli"]) + vivos, pendientes = {}, [] + for p in posts: + slug = p.get("slug") + txt = texto_de(p) + hh = hashlib.sha256(txt.encode()).hexdigest() + reg = { + "slug": slug, + "titulo": p.get("title"), + "estado": p.get("status"), + "url": f"{cfg['base']}/{slug}/", + "hash": hh, + "chars": len(txt), + # se recalcula en cada pasada (es gratis) para que no envejezca + "enlaza_a": enlaces_de(p, cfg["base"]), + } + viejo = cache.get(slug) + if viejo and viejo.get("hash") == hh and viejo.get("vec"): + reg["vec"] = viejo["vec"] + else: + pendientes.append((slug, txt)) + vivos[slug] = reg + + if pendientes: + if verboso: + print(f" embebiendo {len(pendientes)} de {len(vivos)} artículos " + f"({sitio.upper()})…", file=sys.stderr) + url = endpoint() + vecs = embed([t for _, t in pendientes], url) + for (slug, _), v in zip(pendientes, vecs): + vivos[slug]["vec"] = v + + with open(ruta, "w") as f: + json.dump(vivos, f) + return vivos + + +# ---------- similitud ---------- + +def matriz(regs): + import numpy as np + slugs = sorted(regs) + M = np.array([regs[s]["vec"] for s in slugs], dtype="float32") + M /= np.linalg.norm(M, axis=1, keepdims=True) + return slugs, M @ M.T + + +def afines(sitio, slug, n, refrescar=False): + regs = carga(sitio, refrescar) + if slug not in regs: + sys.exit(f"✗ '{slug}' no está en el corpus de {sitio.upper()} " + f"({len(regs)} artículos). ¿Sitio equivocado?") + slugs, S = matriz(regs) + i = slugs.index(slug) + fila = [(S[i][j], slugs[j]) for j in range(len(slugs)) + if j != i and regs[slugs[j]]["estado"] == "published"] + fila.sort(reverse=True) + ya = set(regs[slug].get("enlaza_a", [])) + return [{"slug": s, "similitud": round(float(v), 4), + "titulo": regs[s]["titulo"], "url": regs[s]["url"], + "ya_enlazado": s in ya} + for v, s in fila[:n]] + + +def canibalizacion(sitio, umbral, refrescar=False): + regs = carga(sitio, refrescar) + slugs, S = matriz(regs) + out = [] + for i in range(len(slugs)): + for j in range(i + 1, len(slugs)): + v = float(S[i][j]) + if v >= umbral: + # La DIRECCIÓN importa y por eso no basta un booleano: un + # enlace A→B le dice a Google cuál manda. Decir "se enlazan + # entre sí" cuando el enlace va en un solo sentido es mentir + # sobre lo único accionable del hallazgo. + ab = slugs[j] in regs[slugs[i]].get("enlaza_a", []) + ba = slugs[i] in regs[slugs[j]].get("enlaza_a", []) + enlace = ("mutuo" if ab and ba else + "A→B" if ab else "B→A" if ba else "ninguno") + out.append({"similitud": round(v, 4), + "enlace": enlace, + "a": {"slug": slugs[i], "titulo": regs[slugs[i]]["titulo"], + "estado": regs[slugs[i]]["estado"]}, + "b": {"slug": slugs[j], "titulo": regs[slugs[j]]["titulo"], + "estado": regs[slugs[j]]["estado"]}}) + out.sort(key=lambda x: -x["similitud"]) + return out + + +# ---------- CLI ---------- + +def main(): + ap = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + sub = ap.add_subparsers(dest="cmd", required=True) + + a = sub.add_parser("afines", help="artículos publicados más parecidos a uno") + a.add_argument("slug") + a.add_argument("--site", required=True, choices=SITIOS) + a.add_argument("-n", type=int, default=4) + + c = sub.add_parser("canibalizacion", help="pares que compiten por la misma búsqueda") + c.add_argument("--site", required=True, choices=SITIOS) + # 0.78 calibrado el 2026-07-25 contra el corpus real, no a ojo. El reparto + # de similitudes NO es el mismo en los dos sitios (ES: mediana 0,58 y p99 + # 0,75; EN: 0,64 y 0,80 — el inglés se parece más entre sí), así que 0,78 + # deja fuera el ruido de los dos y solo saca lo que de verdad solapa. + # A propósito un umbral ABSOLUTO y no un percentil: uno relativo siempre + # encontraría "el 1% más parecido" aunque no hubiera ningún problema, y un + # vigilante que siempre avisa no sirve. + c.add_argument("--umbral", type=float, default=0.78) + + d = sub.add_parser("distribucion", help="reparto de similitudes, para calibrar") + d.add_argument("--site", required=True, choices=SITIOS) + + sub.add_parser("estado", help="qué hay en el caché") + + for p in (a, c, d): + p.add_argument("--json", action="store_true") + p.add_argument("--refrescar", action="store_true", + help="re-embeber todo, ignorando el caché") + + ar = ap.parse_args() + + if ar.cmd == "estado": + if not os.path.isdir(ESTADO): + print("caché vacío (aún no se ha ejecutado nada)") + return + for f in sorted(os.listdir(ESTADO)): + d = json.load(open(os.path.join(ESTADO, f))) + con = sum(1 for r in d.values() if r.get("vec")) + pub = sum(1 for r in d.values() if r["estado"] == "published") + print(f" {f:10} {len(d):3} artículos ({pub} publicados), " + f"{con} con vector") + return + + if ar.cmd == "afines": + res = afines(ar.site, ar.slug, ar.n, ar.refrescar) + if ar.json: + print(json.dumps(res, ensure_ascii=False, indent=2)) + else: + print(f"\nArtículos publicados más afines a «{ar.slug}» ({ar.site.upper()}):\n") + for r in res: + marca = " (YA ENLAZADO)" if r["ya_enlazado"] else "" + print(f" {r['similitud']:.3f} {r['titulo']}{marca}") + print(f" {r['url']}") + return + + if ar.cmd == "canibalizacion": + res = canibalizacion(ar.site, ar.umbral, ar.refrescar) + if ar.json: + print(json.dumps(res, ensure_ascii=False, indent=2)) + else: + if not res: + print(f"Sin pares por encima de {ar.umbral} en {ar.site.upper()}. " + f"Ningún artículo le hace sombra a otro.") + for r in res: + nota = {"ninguno": " ⚠️ SIN enlace en ninguna dirección", + "mutuo": " (se enlazan en los dos sentidos)", + "A→B": " (A enlaza a B, no al revés)", + "B→A": " (B enlaza a A, no al revés)"}[r["enlace"]] + print(f"\n {r['similitud']:.3f}{nota}") + print(f" A [{r['a']['estado']}] {r['a']['titulo']}") + print(f" B [{r['b']['estado']}] {r['b']['titulo']}") + return + + if ar.cmd == "distribucion": + regs = carga(ar.site, ar.refrescar) + slugs, S = matriz(regs) + vals = sorted(float(S[i][j]) for i in range(len(slugs)) + for j in range(i + 1, len(slugs))) + import statistics + print(f"\n{ar.site.upper()}: {len(slugs)} artículos, {len(vals)} pares") + print(f" mín {vals[0]:.3f} mediana {statistics.median(vals):.3f} " + f"máx {vals[-1]:.3f}") + for p in (50, 75, 90, 95, 99): + print(f" percentil {p}: {vals[int(len(vals)*p/100)]:.3f}") + print("\n los 8 pares más parecidos:") + top = sorted(((float(S[i][j]), slugs[i], slugs[j]) + for i in range(len(slugs)) for j in range(i + 1, len(slugs))), + reverse=True)[:8] + for v, x, y in top: + print(f" {v:.3f} {x[:44]} × {y[:44]}") + + +if __name__ == "__main__": + main()