ChemaVXandClaude Opus 5
d1d8549089
Sustituir la imagen destacada en Ghost BORRA feature_image_alt: el campo va atado a la imagen, no al post. El vigilante volvía a considerar candidato al post y, como no distinguía casos, iba a relanzar el remate COMPLETO para rellenar 191 caracteres: revisión editorial otra vez, el enlazador sobre un artículo que ya tenía sus cuatro enlaces, posible cambio de slug —que rompe la pareja de hreflang recién declarada— y otra sesión entera del plan, el mismo día en que la cuota ya se había agotado una vez. solo_falta_alt() distingue el caso y build_prompt() sirve un encargo de 2.248 caracteres en vez de 7.237, que además PROHÍBE tocar texto, slug, metas, excerpt, tags y enlaces: ahí cualquier cambio es una regresión, no una mejora. El modo va en el nombre del log y en la cabecera de Telegram, porque al revisar por qué cambió un post lo primero que quieres saber es cuál corrió. Escribe el alt por el MISMO camino que el modo largo (seo_finish prep → Read de la imagen → seo_finish apply), así que no hay una segunda forma de escribirlo que pueda derivar de la primera. Once tests, y los que importan son los del falso positivo: un post recién creado NO debe entrar por el corto (se quedaría sin remate y parecería hecho), ni uno con el remate interrumpido a medias (meta_title sin meta_description). Verificado por el camino real sobre un borrador canario en el estado exacto —imagen y metas puestas, sin alt—: eligió el modo alt, escribió 189/191 caracteres describiendo la imagen, y de doce campos comparados NO tocó ninguno; el cuerpo quedó literalmente idéntico. Canario borrado después. Y hub-en/hub-es con los dos artículos publicados hoy (Socorro 1964 y Cash-Landrum ES), que seo_estado cazó fuera del índice a los minutos de publicarse. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
352 lines
14 KiB
Python
352 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Tool G — remate-watch: la "tacada" automática de borrador a listo-para-publicar.
|
|
|
|
El flujo editorial es: ResearchOwl crea el borrador SIN SEO (dryrun) → Jose
|
|
genera la infografía con ChatGPT y la sube a Ghost → y a partir de ahí este
|
|
vigilante hace el resto solo. Detecta el momento por la firma inequívoca de
|
|
"listo para rematar":
|
|
|
|
status=draft Y feature_image puesta Y metas sin rellenar
|
|
|
|
(un borrador recién creado no tiene imagen; uno ya rematado tiene metas), y
|
|
espera además a que el post lleve un rato sin editarse (--min-age, 10 min por
|
|
defecto) para no saltar mientras Jose aún retoca o está subiendo la imagen.
|
|
|
|
Cuando dispara, lanza `claude -p` (plan Pro, coste API cero — el mismo patrón
|
|
que hermes-bot) con el checklist de ~/seo-tools/remate_prompt.md: revisión
|
|
editorial completa, slug, enlaces internos, metas + alt leyendo la imagen, y
|
|
validación. Después NO se cree el "hecho" del agente: re-lee el post de Ghost y
|
|
comprueba que las metas están de verdad antes de dar el OK por Telegram
|
|
(feedback_canario_camino_real). El post queda en DRAFT siempre: publicar es
|
|
decisión humana.
|
|
|
|
Uso:
|
|
python3 remate_watch.py # lo que corre el timer
|
|
python3 remate_watch.py --dry-run # solo detecta e imprime
|
|
python3 remate_watch.py --site es --min-age 0 # forzar (pruebas)
|
|
|
|
Salvaguardas:
|
|
- estado por (post, updated_at): un remate fallido NO entra en bucle — solo
|
|
se reintenta si el post vuelve a editarse (updated_at cambia).
|
|
- lock con pid: nunca dos remates a la vez (el timer de systemd tampoco
|
|
solapa, esto cubre las ejecuciones manuales).
|
|
- si claude falla, agota el tiempo, o dice "hecho" pero las metas no están
|
|
en Ghost, llega un 🚨 con la ruta del log completo.
|
|
"""
|
|
import argparse
|
|
import base64
|
|
import datetime
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import urllib.parse
|
|
import urllib.request
|
|
from string import Template
|
|
|
|
# Ojo al canónico: EN es www, ES es el APEX — al revés (patrón de seo_watch).
|
|
SITES = {
|
|
"en": {"cli": "ghst-en", "site_name": "The Exclusion Zone (EN)",
|
|
"canonical": "https://www.theexclusionzone.com/",
|
|
"canon_note": "en www, con barra final"},
|
|
"es": {"cli": "ghst-es", "site_name": "Zona de Exclusión (ES)",
|
|
"canonical": "https://zonadeexclusion.com/",
|
|
"canon_note": "en el APEX, con barra final"},
|
|
}
|
|
STATE_DIR = os.path.expanduser("~/.local/state/remate-watch")
|
|
STATE = os.path.join(STATE_DIR, "state.json")
|
|
LOCK = os.path.join(STATE_DIR, "lock")
|
|
LOGS = os.path.join(STATE_DIR, "logs")
|
|
PROMPT_TMPL = os.path.join(os.path.dirname(os.path.abspath(__file__)), "remate_prompt.md")
|
|
# Encargo corto para el post que YA está rematado y solo ha perdido el alt.
|
|
PROMPT_ALT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "remate_alt_prompt.md")
|
|
# Sobreescribible por env para poder ensayar el camino de fallo con un binario
|
|
# que casca, sin tocar código (feedback_canario_camino_real: hay que VER fallar
|
|
# la alerta antes de fiarse de ella).
|
|
CLAUDE = os.environ.get("CLAUDE_BIN", os.path.expanduser("~/.local/bin/claude"))
|
|
CLAUDE_TIMEOUT = 2700 # 45 min: una revisión editorial de verdad tarda
|
|
TG_SECRET = ("monitoring", "grafana-telegram-infisical")
|
|
FIELDS = ("id,slug,title,status,feature_image,feature_image_alt,"
|
|
"meta_title,meta_description,updated_at")
|
|
|
|
|
|
def sh(cmd, timeout=180):
|
|
return subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=timeout)
|
|
|
|
|
|
def telegram(text):
|
|
r = sh(f"kubectl get secret -n {TG_SECRET[0]} {TG_SECRET[1]} -o jsonpath='{{.data.TELEGRAM_BOT_TOKEN}}'")
|
|
token = base64.b64decode(r.stdout).decode()
|
|
r2 = sh(f"kubectl get secret -n {TG_SECRET[0]} {TG_SECRET[1]} -o jsonpath='{{.data.TELEGRAM_CHAT_ID}}'")
|
|
chat = base64.b64decode(r2.stdout).decode()
|
|
data = urllib.parse.urlencode({"chat_id": chat, "text": text[:3900]}).encode()
|
|
req = urllib.request.Request(f"https://api.telegram.org/bot{token}/sendMessage", data=data)
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
return json.loads(resp.read()).get("ok", False)
|
|
|
|
|
|
def fetch_drafts(site):
|
|
"""Borradores del sitio. A fichero, no a PIPE (patrón de seo_watch: un PIPE
|
|
se trunca a 64 KB en silencio; con --fields cortos sobra, pero la copia del
|
|
patrón es gratis)."""
|
|
cli = SITES[site]["cli"]
|
|
fd, path = tempfile.mkstemp(suffix=".json", prefix="remate_watch_")
|
|
os.close(fd)
|
|
try:
|
|
r = sh(f'{cli} post list --filter "status:draft" --limit all '
|
|
f'--fields {FIELDS} --json > {path}', timeout=240)
|
|
if r.returncode != 0:
|
|
raise RuntimeError(f"{cli} post list falló: {r.stderr.strip()[:200]}")
|
|
with open(path) as f:
|
|
data = json.load(f)
|
|
return data.get("posts", data) if isinstance(data, dict) else data
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
|
|
def age_seconds(updated_at):
|
|
dt = datetime.datetime.strptime(updated_at, "%Y-%m-%dT%H:%M:%S.%fZ")
|
|
return (datetime.datetime.utcnow() - dt).total_seconds()
|
|
|
|
|
|
def is_candidate(p):
|
|
return (p.get("status") == "draft"
|
|
and p.get("feature_image")
|
|
and not (p.get("meta_title") and p.get("meta_description")
|
|
and p.get("feature_image_alt")))
|
|
|
|
|
|
def load_state():
|
|
try:
|
|
with open(STATE) as f:
|
|
return json.load(f)
|
|
except (FileNotFoundError, ValueError):
|
|
return {}
|
|
|
|
|
|
def save_state(st):
|
|
os.makedirs(STATE_DIR, exist_ok=True)
|
|
with open(STATE, "w") as f:
|
|
json.dump(st, f, indent=1)
|
|
|
|
|
|
def acquire_lock():
|
|
"""Lock con pid. Si el dueño ya no existe (crash), se roba."""
|
|
os.makedirs(STATE_DIR, exist_ok=True)
|
|
for _ in range(2):
|
|
try:
|
|
fd = os.open(LOCK, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
|
|
os.write(fd, str(os.getpid()).encode())
|
|
os.close(fd)
|
|
return True
|
|
except FileExistsError:
|
|
try:
|
|
with open(LOCK) as f:
|
|
pid = int(f.read().strip() or "0")
|
|
os.kill(pid, 0) # ¿vive?
|
|
return False # sí: otro remate en curso, nos retiramos
|
|
except (ProcessLookupError, ValueError):
|
|
os.unlink(LOCK) # dueño muerto: lock huérfano, se limpia
|
|
except PermissionError:
|
|
return False
|
|
return False
|
|
|
|
|
|
def release_lock():
|
|
try:
|
|
os.unlink(LOCK)
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
|
|
def solo_falta_alt(p):
|
|
"""El post ya está rematado y solo ha perdido el texto alternativo.
|
|
|
|
Pasa cada vez que se SUSTITUYE la imagen destacada en Ghost: el campo
|
|
`feature_image_alt` va atado a la imagen, no al post, y se borra. Sin
|
|
distinguir este caso, el vigilante relanzaba el remate COMPLETO para
|
|
rellenar un campo de 191 caracteres: rehacía la revisión editorial, volvía a
|
|
pasar el enlazador sobre un artículo que ya tenía sus enlaces, podía
|
|
recambiar el slug —rompiendo la pareja de hreflang recién declarada— y se
|
|
comía otra sesión entera del plan. Visto el 2026-07-30 al regenerar la
|
|
infografía de Socorro.
|
|
"""
|
|
return bool(p.get("feature_image")
|
|
and p.get("meta_title") and p.get("meta_description")
|
|
and not p.get("feature_image_alt"))
|
|
|
|
|
|
def build_prompt(site, post):
|
|
cfg = SITES[site]
|
|
plantilla = PROMPT_ALT if solo_falta_alt(post) else PROMPT_TMPL
|
|
with open(plantilla) as f:
|
|
tmpl = Template(f.read())
|
|
return tmpl.substitute(
|
|
site=site, site_name=cfg["site_name"], canonical=cfg["canonical"],
|
|
canon_note=cfg["canon_note"], cli=cfg["cli"],
|
|
id=post["id"], slug=post["slug"], title=post.get("title", ""))
|
|
|
|
|
|
def verify_in_ghost(site, post_id):
|
|
"""La verificación contra la realidad: ¿están las metas EN GHOST?"""
|
|
cli = SITES[site]["cli"]
|
|
r = sh(f"{cli} post get {post_id} --fields {FIELDS} --json", timeout=120)
|
|
try:
|
|
data = json.loads(r.stdout)
|
|
p = (data.get("posts") or [data])[0] if isinstance(data, dict) else data[0]
|
|
except (ValueError, IndexError, KeyError):
|
|
return None, f"no pude re-leer el post ({r.stderr.strip()[:120]})"
|
|
faltan = [k for k in ("meta_title", "meta_description", "feature_image_alt")
|
|
if not p.get(k)]
|
|
if faltan:
|
|
return p, "faltan tras el remate: " + ", ".join(faltan)
|
|
return p, None
|
|
|
|
|
|
def run_remate(site, post):
|
|
"""Lanza claude con el checklist y devuelve (ok, texto_para_telegram, log)."""
|
|
os.makedirs(LOGS, exist_ok=True)
|
|
ts = datetime.datetime.utcnow().strftime("%Y%m%d-%H%M%S")
|
|
# El modo va en el NOMBRE del log: al revisar por qué un post cambió, lo
|
|
# primero que quieres saber es si corrió el checklist entero o solo el alt.
|
|
modo = "alt" if solo_falta_alt(post) else "full"
|
|
log = os.path.join(LOGS, f"{ts}-{site}-{modo}-{post['slug'][:40]}.log")
|
|
prompt = build_prompt(site, post)
|
|
|
|
try:
|
|
proc = subprocess.run(
|
|
[CLAUDE, "-p", prompt, "--output-format", "json",
|
|
"--dangerously-skip-permissions"],
|
|
cwd=os.path.expanduser("~"), text=True, capture_output=True,
|
|
timeout=CLAUDE_TIMEOUT)
|
|
except subprocess.TimeoutExpired:
|
|
with open(log, "w") as f:
|
|
f.write("TIMEOUT tras %ss" % CLAUDE_TIMEOUT)
|
|
return False, f"claude agotó el tiempo ({CLAUDE_TIMEOUT}s)", log
|
|
|
|
with open(log, "w") as f:
|
|
f.write(proc.stdout or "")
|
|
if proc.stderr:
|
|
f.write("\n--- stderr ---\n" + proc.stderr)
|
|
|
|
if proc.returncode != 0:
|
|
return False, f"claude salió con código {proc.returncode}: {proc.stderr.strip()[:200]}", log
|
|
try:
|
|
out = json.loads(proc.stdout)
|
|
result = out.get("result") or ""
|
|
if out.get("is_error"):
|
|
return False, f"claude reportó error: {result[:300]}", log
|
|
except ValueError:
|
|
result = proc.stdout.strip()
|
|
if not result:
|
|
return False, "claude terminó sin informe (salida vacía)", log
|
|
return True, result, log
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--dry-run", action="store_true",
|
|
help="solo detecta e imprime; no lanza, no notifica, no guarda estado")
|
|
ap.add_argument("--site", choices=("en", "es", "both"), default="both")
|
|
ap.add_argument("--min-age", type=int, default=600,
|
|
help="segundos sin ediciones antes de disparar (debounce)")
|
|
# Tope por pasada: con una tanda grande de borradores (p.ej. la cola de
|
|
# vacaciones de agosto) se pueden subir varias imágenes seguidas y quedar
|
|
# 9 candidatos en el mismo tick. Encadenarlos en una sola ejecución
|
|
# rebasaría el TimeoutStartSec del unit y systemd mataría el remate a
|
|
# medias. Se hacen 2 por tick y el resto espera al siguiente (15 min).
|
|
ap.add_argument("--max", type=int, default=2, dest="maximo",
|
|
help="máximo de remates por ejecución (0 = sin límite)")
|
|
a = ap.parse_args()
|
|
sitios = ["en", "es"] if a.site == "both" else [a.site]
|
|
|
|
state = load_state()
|
|
trabajos = []
|
|
for site in sitios:
|
|
try:
|
|
drafts = fetch_drafts(site)
|
|
except Exception as e: # un sitio caído no debe tumbar al otro
|
|
print(f"[{site}] ERROR listando borradores: {e}")
|
|
if not a.dry_run:
|
|
telegram(f"🚨 remate-watch: no pude listar borradores de [{site.upper()}]: {e}")
|
|
continue
|
|
for p in drafts:
|
|
if not is_candidate(p):
|
|
continue
|
|
edad = age_seconds(p["updated_at"])
|
|
prev = state.get(p["id"], {})
|
|
if prev.get("updated_at") == p["updated_at"]:
|
|
print(f"[{site}] {p['slug']}: ya intentado con esta versión "
|
|
f"(resultado {prev.get('result')}), espero a que se edite")
|
|
continue
|
|
if edad < a.min_age:
|
|
print(f"[{site}] {p['slug']}: candidato pero editado hace "
|
|
f"{int(edad)}s (<{a.min_age}s), espero")
|
|
continue
|
|
trabajos.append((site, p))
|
|
print(f"[{site}] CANDIDATO: {p['slug']} (sin editar desde hace {int(edad)}s)")
|
|
|
|
if not trabajos:
|
|
print("sin candidatos")
|
|
return 0
|
|
|
|
# Los más antiguos primero: si hay cola, se remata en el orden en que se
|
|
# dejaron listos, no en el que devuelva la API.
|
|
trabajos.sort(key=lambda t: t[1]["updated_at"])
|
|
pendientes = 0
|
|
if a.maximo and len(trabajos) > a.maximo:
|
|
pendientes = len(trabajos) - a.maximo
|
|
trabajos = trabajos[:a.maximo]
|
|
print(f"[tope] {a.maximo} remate(s) en esta pasada; "
|
|
f"{pendientes} para el siguiente tick")
|
|
|
|
if a.dry_run:
|
|
print(f"[dry-run] lanzaría {len(trabajos)} remate(s)")
|
|
return 0
|
|
|
|
if not acquire_lock():
|
|
print("otro remate en curso; me retiro")
|
|
return 0
|
|
try:
|
|
for site, p in trabajos:
|
|
# El estado se apunta ANTES de lanzar: si esto casca a mitad, el
|
|
# siguiente tick no relanza en bucle sobre la misma versión.
|
|
state[p["id"]] = {"updated_at": p["updated_at"], "result": "lanzado",
|
|
"ts": datetime.datetime.utcnow().isoformat()}
|
|
save_state(state)
|
|
|
|
modo = "alt" if solo_falta_alt(p) else "completo"
|
|
print(f"[{site}] rematando ({modo}) {p['slug']} …")
|
|
ok, texto, log = run_remate(site, p)
|
|
|
|
if ok:
|
|
_, problema = verify_in_ghost(site, p["id"])
|
|
if problema:
|
|
ok, texto = False, (f"el agente dio el remate por hecho pero Ghost "
|
|
f"dice otra cosa: {problema}")
|
|
state[p["id"]]["result"] = "ok" if ok else "fallo"
|
|
save_state(state)
|
|
|
|
if ok:
|
|
cabecera = ("🖼 Alt rehecho [%s] %s\n(la imagen se sustituyó y Ghost borró "
|
|
"el alt; el resto del remate NO se ha tocado)"
|
|
% (site.upper(), p["slug"]) if modo == "alt"
|
|
else "🪄 Remate automático [%s] %s" % (site.upper(), p["slug"]))
|
|
msg = f"{cabecera}\n\n{texto}"
|
|
else:
|
|
msg = (f"🚨 remate-watch: el remate de [{site.upper()}] «{p['slug']}» "
|
|
f"FALLÓ: {texto}\n\nLog: {log}\n"
|
|
f"(no se reintenta hasta que el post se edite de nuevo)")
|
|
enviado = telegram(msg)
|
|
print(f"[{site}] {p['slug']}: {'OK' if ok else 'FALLO'} "
|
|
f"[telegram: {'enviado' if enviado else 'FALLO'}] log: {log}")
|
|
finally:
|
|
release_lock()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|