ChemaVXandClaude Opus 5
2512144be5
1) remate-watch ya no elige los enlaces internos a ojo. El paso 3 del prompt pregunta primero a seo_semantic (afines), que le da los publicados más parecidos y le marca los que el borrador YA enlaza. Se le dice explícitamente que es una sugerencia y no una orden, y que si nada pasa de ~0,55 lo diga en el informe en vez de forzar un enlace malo: un enlace forzado es peor que ninguno. Para que eso funcione, seo_semantic sabe ahora trabajar con un post que NO está en el corpus: `ghst post list` sin filtro devuelve solo publicados y programados, así que el borrador que remate-watch está preparando no aparecía. Se baja suelto por slug y se embebe al vuelo. Verificado por equivalencia: los vecinos de un post embebido al vuelo son idénticos, hasta el cuarto decimal, a los del mismo post cacheado. 2) canibal-watch: a diario a las 07:00 UTC (una hora después de que publique la cola de agosto). Telegram SOLO si hay algo nuevo. Solo avisa de pares que además NO se enlazan. Un par muy parecido pero cosido con un enlace ya le dice a Google cuál manda; avisar de él es ruido. Los enlazados se guardan igual en el estado, porque hacen falta para detectar que MÁS TARDE pierdan el enlace (una edición que se lleve por delante la jerarquía). Con el filtro, el primer aviso baja de 19 líneas a 5 accionables. Probado por el camino real, arrancándolo con systemctl y no desde una shell con el PATH bueno: primera pasada avisa (5 pares), segunda se calla, y con una regresión simulada en el estado la detecta. No escribe en el blog. Coser o fusionar lo decide Jose. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
372 lines
15 KiB
Python
372 lines
15 KiB
Python
#!/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 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]]
|
||
|
||
|
||
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()
|