#!/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. · padrinos → lo INVERSO: qué páginas fuertes deberían enlazar aquí. `afines` sirve al lector, `padrinos` sirve a Google, y casi nunca dan la misma respuesta. · descubrimiento → publicados a los que solo se llega desde páginas que Google apenas visita. · canibalizacion → pares demasiado parecidos DENTRO de un sitio: compiten por la misma búsqueda y se hacen sombra. ⚠️ Para contar enlaces internos, la verdad está en el `html` que devuelve el CLI de Ghost, NO en la página renderizada: el tema mete navegación anterior / siguiente y un feed de relacionados que parecen enlaces del cuerpo y no lo son. Contarlos desde el HTML público dice «cero huérfanos» cuando la realidad son 21 de 32 (comprobado el 2026-07-28, y la primera medición salió mal por esto). 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 padrinos --site en [-n 5] [--json] python3 seo_semantic.py descubrimiento --site en [--todos] [--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), # Ghost excluye del sitemap los posts con canonical_url propio: son # versiones consolidadas en otra URL. Siguen publicados y pueden # arrastrar tráfico histórico, pero NO sirven de padrino — el enlace # saldría de una página que el propio sitio declara secundaria. "canonico": p.get("canonical_url"), # 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 # ---------- tráfico real (Search Console) ---------- # Sin esto, "enlázalo desde un artículo afín" es un consejo a ciegas: da igual # que el origen sea el artículo con más tráfico del sitio o uno que no ha # recibido una impresión en su vida. Medido el 2026-07-28, esa diferencia es la # que explica que dos posts de EN publicados el 21-jul siguieran siendo «URL # desconocida para Google» una semana después: estaban enlazados, sí, pero solo # desde páginas con 8 y 12 impresiones DE POR VIDA, que Google casi no visita. BD_GSC = os.path.expanduser("~/.local/state/seo-gsc/gsc.db") def impresiones(sitio): """{slug: impresiones acumuladas}. Vacío si aún no hay marcador.""" if not os.path.exists(BD_GSC): return {} import sqlite3 with sqlite3.connect(f"file:{BD_GSC}?mode=ro", uri=True) as c: return {k.rstrip("/").rsplit("/", 1)[-1]: v for k, v in c.execute( "SELECT clave, SUM(impresiones) FROM filas " "WHERE sitio=? AND dim='page' GROUP BY clave", (sitio,))} def tercio_fuerte(regs, impr): """Los slugs publicados del tercio superior por impresiones. Es el listón de «página fuerte»: relativo al sitio y no un número absoluto, porque 200 impresiones son mucho en ES y poco en EN. """ pub = [s for s, r in regs.items() if r.get("estado") == "published" and not r.get("canonico")] if not pub: return set() orden = sorted(pub, key=lambda s: -impr.get(s, 0)) return set(orden[:max(1, len(orden) // 3)]) def entrantes(regs): """Invierte enlaza_a: {slug: [slugs publicados que lo enlazan]}.""" dentro = {s: [] for s in regs} for origen, r in regs.items(): if r.get("estado") != "published": continue for destino in r.get("enlaza_a", []): if destino in dentro and destino != origen: dentro[destino].append(origen) return dentro # ---------- 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 post_suelto(sitio, slug): """Un post por slug, INCLUIDOS los borradores. Hace falta porque `ghst post list` sin filtro devuelve solo publicados y programados: el corpus NO tiene borradores. Y el caso de uso principal — sugerirle enlaces a remate-watch mientras prepara un borrador— es exactamente ese. Se embebe al vuelo contra el caché, sin guardarlo. """ import tempfile cli = SITIOS[sitio]["cli"] fd, path = tempfile.mkstemp(suffix=".json", prefix="seo_semantic_") os.close(fd) try: subprocess.run(f'{cli} post list --filter "slug:{slug}" --limit all ' f'--formats html --json > {path}', shell=True, capture_output=True, text=True, timeout=240) with open(path) as f: d = json.load(f) posts = d.get("posts", d) if isinstance(d, dict) else d return posts[0] if posts else None except (json.JSONDecodeError, IndexError): return None finally: os.unlink(path) def afines(sitio, slug, n, refrescar=False): regs = carga(sitio, refrescar) externo = None if slug not in regs: p = post_suelto(sitio, slug) if p is None: sys.exit(f"✗ '{slug}' no existe en {sitio.upper()} " f"(ni publicado, ni programado, ni borrador). " f"¿Sitio equivocado?") externo = {"vec": embed([texto_de(p)], endpoint())[0], "enlaza_a": enlaces_de(p, SITIOS[sitio]["base"]), "estado": p.get("status")} print(f" («{slug}» está en {externo['estado']}, fuera del corpus: " f"embebido al vuelo)", file=sys.stderr) regs = dict(regs) regs[slug] = {**externo, "slug": slug, "titulo": p.get("title"), "url": f"{SITIOS[sitio]['base']}/{slug}/"} 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]] SIM_MIN_PADRINO = 0.55 # mismo listón que usa el remate para no forzar enlaces def padrinos(sitio, slug, n, refrescar=False): """Lo inverso de `afines`: qué páginas FUERTES deberían enlazar a este post. `afines` contesta «a quién enlazo yo», que es lo que necesita un borrador para dar contexto al lector. Esto contesta «quién me enlaza a mí», que es lo que necesita Google para llegar hasta aquí. No es la misma pregunta y la respuesta rara vez coincide: el artículo más parecido suele ser otro post pequeño del mismo rincón temático, y ése no arrastra a nadie. Se descartan los que YA enlazan al destino, y se ordena por tráfico del origen — no por similitud — porque entre dos candidatos razonables el que sirve es el que Google visita. """ regs = carga(sitio, refrescar) if slug not in regs: p = post_suelto(sitio, slug) if p is None: sys.exit(f"✗ '{slug}' no existe en {sitio.upper()}. ¿Sitio equivocado?") regs = dict(regs) regs[slug] = {"slug": slug, "titulo": p.get("title"), "estado": p.get("status"), "url": f"{SITIOS[sitio]['base']}/{slug}/", "enlaza_a": enlaces_de(p, SITIOS[sitio]["base"]), "vec": embed([texto_de(p)], endpoint())[0]} impr = impresiones(sitio) fuertes = tercio_fuerte(regs, impr) slugs, S = matriz(regs) i = slugs.index(slug) out = [] for j, otro in enumerate(slugs): if j == i or regs[otro].get("estado") != "published": continue if regs[otro].get("canonico"): continue # consolidado en otra URL: no está ni en el sitemap if slug in regs[otro].get("enlaza_a", []): continue # ya nos enlaza: no hay nada que pedirle sim = float(S[i][j]) if sim < SIM_MIN_PADRINO: continue out.append({"slug": otro, "titulo": regs[otro]["titulo"], "url": regs[otro]["url"], "similitud": round(sim, 4), "impresiones": impr.get(otro, 0), "fuerte": otro in fuertes}) out.sort(key=lambda r: (-r["impresiones"], -r["similitud"], r["slug"])) return out[:n] def descubrimiento(sitio, refrescar=False): """Posts publicados a los que solo llega enlace desde páginas flojas. Un post puede no ser huérfano y aun así ser invisible: si los únicos caminos que llevan hasta él pasan por artículos que Google apenas rastrea, tarda semanas en descubrirlo — o no lo descubre. """ regs = carga(sitio, refrescar) impr = impresiones(sitio) fuertes = tercio_fuerte(regs, impr) dentro = entrantes(regs) out = [] for s, r in regs.items(): if r.get("estado") != "published" or r.get("canonico"): continue orig = sorted(dentro[s], key=lambda o: -impr.get(o, 0)) out.append({"slug": s, "titulo": r["titulo"], "impresiones": impr.get(s, 0), "entrantes": len(orig), "apadrinado": any(o in fuertes for o in orig), "mejor_origen": orig[0] if orig else None, "mejor_origen_impr": impr.get(orig[0], 0) if orig else 0}) out.sort(key=lambda r: (r["apadrinado"], r["mejor_origen_impr"], r["slug"])) return out 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) pa = sub.add_parser("padrinos", help="qué páginas fuertes deberían enlazar a este post") pa.add_argument("slug") pa.add_argument("--site", required=True, choices=SITIOS) pa.add_argument("-n", type=int, default=5) de = sub.add_parser("descubrimiento", help="posts a los que solo llega enlace desde páginas flojas") de.add_argument("--site", required=True, choices=SITIOS) de.add_argument("--todos", action="store_true", help="listar también los que sí están apadrinados") 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, pa, de): 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 == "padrinos": res = padrinos(ar.site, ar.slug, ar.n, ar.refrescar) if ar.json: print(json.dumps(res, ensure_ascii=False, indent=2)) elif not res: print(f"Nadie afín (≥{SIM_MIN_PADRINO}) que no lo enlace ya. " f"O está bien apadrinado, o el corpus no da para un enlace honesto.") else: print(f"\nQuién debería enlazar a «{ar.slug}» ({ar.site.upper()}), " f"por tráfico del origen:\n") for r in res: print(f" {r['impresiones']:6} impr sim {r['similitud']:.3f}" f"{' ← tercio fuerte' if r['fuerte'] else ''}") print(f" {r['titulo']}") print(f" {r['slug']}") print("\n El enlace hay que escribirlo en el post ORIGEN:") print(f" python3 ~/seo-tools/seo_link.py {res[0]['slug']} " f"--site {ar.site} --link \"={ar.slug}\"") return if ar.cmd == "descubrimiento": res = descubrimiento(ar.site, ar.refrescar) if ar.json: print(json.dumps(res, ensure_ascii=False, indent=2)) return malos = [r for r in res if not r["apadrinado"]] print(f"\n{ar.site.upper()}: {len(res)} publicados, " f"{len(malos)} sin un solo enlace desde el tercio fuerte\n") for r in (res if ar.todos else malos): marca = "✓" if r["apadrinado"] else "⚠" origen = (f"mejor origen: {r['mejor_origen'][:40]} " f"({r['mejor_origen_impr']} impr)" if r["mejor_origen"] else "HUÉRFANO: nadie lo enlaza") print(f" {marca} {r['impresiones']:6} impr propias " f"{r['entrantes']} entrantes {r['slug'][:44]}") print(f" {origen}") 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()