capa semántica: enlaces internos afines y canibalización
Embebe los 80 artículos de los dos blogs con bge-m3 (ya cargado en ollama, 100% GPU) y sobre los vectores contesta dos cosas: a qué artículos publicados conviene enlazar desde uno nuevo, y qué pares compiten por la misma búsqueda. Nunca escribe en el blog: imprime. Quien escribe el enlace es seo_link.py, y publicar lo decide Jose. Decisiones que no son de adorno: - NUNCA se cruzan los dos sitios. bge-m3 es multilingüe y da 0,966 entre un artículo y su traducción (medido): cruzarlos marcaría todo el espejo ES↔EN como canibalización, justo lo contrario de la verdad. - Endpoint de ollama por la ClusterIP del Service, no por la IP del pod (que cambia en cada reinicio) ni por el Ingress (está tras Authentik). El host la alcanza por kube-proxy, sin NodePort ni tocar la red. - Sin keep_alive en las peticiones: bge-m3 es de producción (lo usa researchowl) y mandar keep_alive:0 lo desalojaría, cobrándole a otro la recarga. - Umbral 0,78 calibrado contra el corpus real, no a ojo: el reparto no es igual en los dos sitios (ES mediana 0,58 / p99 0,75; EN 0,64 / 0,80). Absoluto y no percentil, porque un umbral relativo siempre encuentra "el 1% más parecido" aunque no haya problema, y un vigilante que siempre avisa no sirve. - Se mira qué enlaces EXISTEN ya en el cuerpo, y en qué DIRECCIÓN. Sugerir un enlace que ya está es ruido; y decir "se enlazan entre sí" cuando el enlace va en un solo sentido es mentir sobre lo único accionable. Caché en ~/.local/state/seo-semantic/, re-embebe solo lo que cambia (hash del texto). Corpus vía fetch_corpus de seo_link, que ya trae la guarda de corpus truncado. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
ChemaVX
+333
@@ -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 <slug> → 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 <slug> --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/<sitio>.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 <a> 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()
|
||||
Reference in New Issue
Block a user