Leopoldo Rodríguez
a91580e1af
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QCy5VHMVLkAmftYM6swWmx
583 lines
28 KiB
Python
583 lines
28 KiB
Python
#!/usr/bin/env python3
|
||
"""El tablero de pendientes desde la terminal: foto para leerlo y lotes para cambiarlo (21-oct-2026).
|
||
|
||
drogheda.helgen.online es el centro de mando del proyecto: cada tarea termina ahí, también las que
|
||
hace Claude. Claude no llega al servidor, así que deja sus cambios al tablero en un **lote**
|
||
(backend/lotes/AAAA-MM-DD_tema.json); Leo lo revisa y lo aplica con ./scripts/tablero.sh, y Claude lee
|
||
el tablero en la **foto** (_estado/tablero.json y _estado/tablero.md en la Mac de Leo, fuera de git).
|
||
|
||
En el servidor (lo llama ./scripts/tablero.sh por ssh):
|
||
python3 tablero.py foto pendientes, enfoque y actividad de 14 días, en JSON (stdout)
|
||
python3 tablero.py revisar lotes/ qué haría cada lote que falta; no escribe nada
|
||
python3 tablero.py aplicar lotes/ aplica lo que falta; lo ya aplicado se salta
|
||
python3 tablero.py pendientes lotes/ una línea: cuántos cambios faltan por aplicar
|
||
En la Mac:
|
||
python3 backend/tablero.py resumen _estado/tablero.json la foto en texto (tablero.md)
|
||
|
||
Habla con el backend por HTTP local (TABLERO_API, por omisión http://127.0.0.1:8081) y escribe con
|
||
X-Remote-User = «autor» del lote: pasa por las mismas validaciones que el sitio y cada cambio queda en
|
||
el historial del pendiente, con su autor y su nota. Lo aplicado se anota en
|
||
$DATOS_BASE/lotes-aplicados.json (la misma carpeta de datos que usa app.py). Un lote aplicado no se
|
||
vuelve a aplicar aunque alguien deshaga el cambio a mano, y no se edita después de aplicarse: se escribe
|
||
otro. Un cambio nunca toca un pendiente que ya está en Hecho (salvo que el lote lo pida con si_estado).
|
||
Formato de los lotes: backend/lotes/README.md. Solo usa la biblioteca estándar de Python.
|
||
"""
|
||
import argparse
|
||
import datetime
|
||
import glob
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import unicodedata
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
|
||
API = os.environ.get("TABLERO_API", "http://127.0.0.1:8081").rstrip("/")
|
||
DATOS_BASE = os.environ.get("DATOS_BASE") or (
|
||
os.path.dirname(os.environ.get("ACTAS_DIR", "/srv/projects/drogheda/actas").rstrip("/")) or "/srv/projects/drogheda")
|
||
REGISTRO = os.environ.get("TABLERO_REGISTRO") or os.path.join(DATOS_BASE, "lotes-aplicados.json")
|
||
LECTOR = "claude" # X-Remote-User para las lecturas (no cambia nada; queda claro quién consulta)
|
||
|
||
ESTADOS = ("por-hacer", "en-curso", "bloqueado", "hecho")
|
||
ABIERTOS = ("por-hacer", "en-curso", "bloqueado")
|
||
NOMBRE = {"por-hacer": "Por hacer", "en-curso": "En curso", "bloqueado": "Bloqueado", "hecho": "Hecho"}
|
||
CAMPOS = ("titulo", "descripcion", "pista", "estado", "responsable", "fecha_limite", "enlaces", "etiquetas",
|
||
"bloqueada_por", "version_tarjeta")
|
||
LOTE_ID = re.compile(r"^\d{4}-\d{2}-\d{2}_[a-z0-9][a-z0-9-]{1,60}$")
|
||
TAREA_ID = re.compile(r"^[a-z0-9][a-z0-9-]{2,59}$")
|
||
MES = ("ene", "feb", "mar", "abr", "may", "jun", "jul", "ago", "sep", "oct", "nov", "dic")
|
||
|
||
|
||
class SinBackend(Exception):
|
||
pass
|
||
|
||
|
||
# ------------------------------------------------------------------ utilidades
|
||
|
||
def ahora():
|
||
return datetime.datetime.now().replace(microsecond=0).isoformat()
|
||
|
||
|
||
def norm(s):
|
||
"""'Pasada 6 — V5.6' → 'pasada 6 — v5.6': minúsculas, sin acentos, espacios simples."""
|
||
s = unicodedata.normalize("NFKD", str(s or "")).encode("ascii", "ignore").decode().lower()
|
||
return re.sub(r"\s+", " ", s).strip()
|
||
|
||
|
||
def clave(s):
|
||
"""Para buscar por título sin que estorben acentos, mayúsculas ni signos:
|
||
'Pasada 3 — parte A (V5.6)' → 'pasada 3 parte a v5.6'; 'D-21-2' → 'd 21 2'."""
|
||
return re.sub(r"\s+", " ", re.sub(r"[^a-z0-9.]+", " ", norm(s))).strip(" .")
|
||
|
||
|
||
def corto(v, n=48):
|
||
if isinstance(v, list):
|
||
v = ", ".join(str(x.get("titulo", x)) if isinstance(x, dict) else str(x) for x in v) or "—"
|
||
s = "—" if v in (None, "") else str(v)
|
||
return s if len(s) <= n else s[:n - 1] + "…"
|
||
|
||
|
||
def fecha_corta(iso):
|
||
"""'2026-10-22…' → '22-oct' (con hora si la trae: '22-oct 19:40')."""
|
||
s = str(iso or "")
|
||
if not re.match(r"^\d{4}-\d{2}-\d{2}", s):
|
||
return s or "—"
|
||
f = f"{int(s[8:10])}-{MES[int(s[5:7]) - 1]}"
|
||
return f + (" " + s[11:16] if len(s) >= 16 else "")
|
||
|
||
|
||
def huella(cambio):
|
||
return hashlib.sha256(json.dumps(cambio, sort_keys=True, ensure_ascii=False).encode("utf-8")).hexdigest()[:12]
|
||
|
||
|
||
def llamar(metodo, ruta, usuario, cuerpo=None):
|
||
datos = None if cuerpo is None else json.dumps(cuerpo, ensure_ascii=False).encode("utf-8")
|
||
req = urllib.request.Request(API + ruta, data=datos, method=metodo, headers={
|
||
"X-Remote-User": usuario, "Content-Type": "application/json", "Accept": "application/json"})
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=20) as r:
|
||
return r.status, json.loads(r.read().decode("utf-8") or "{}")
|
||
except urllib.error.HTTPError as e:
|
||
try:
|
||
return e.code, json.loads(e.read().decode("utf-8") or "{}")
|
||
except ValueError:
|
||
return e.code, {"ok": False, "error": f"HTTP {e.code}"}
|
||
except (urllib.error.URLError, OSError) as e:
|
||
raise SinBackend(f"No contesta el backend en {API} ({getattr(e, 'reason', e)}). "
|
||
"¿Está arriba el servicio actas-hbox?")
|
||
|
||
|
||
VALORES = {} # pista y version_tarjeta válidas, tal como las publica el backend en GET /api/tareas
|
||
|
||
|
||
def leer_tareas():
|
||
st, j = llamar("GET", "/api/tareas", LECTOR)
|
||
if st != 200 or not j.get("ok"):
|
||
raise SinBackend(f"No pude leer los pendientes: {j.get('error') or st}")
|
||
VALORES.update(pista=list(j.get("pistas") or []), version_tarjeta=list(j.get("versiones") or []))
|
||
return j.get("tareas") or [], j.get("equipo") or {}
|
||
|
||
|
||
def valores_malos(campos):
|
||
"""Lo que el backend rechazaría: se ve en «revisar» en vez de fallar al aplicar."""
|
||
malos = []
|
||
for k, validos in VALORES.items():
|
||
if k in campos and validos and campos[k] not in validos:
|
||
malos.append(f"{k} «{campos[k]}» no existe (usa {', '.join(validos)})")
|
||
return malos
|
||
|
||
|
||
def leer_registro():
|
||
try:
|
||
with open(REGISTRO, encoding="utf-8") as f:
|
||
r = json.load(f)
|
||
return r if isinstance(r, dict) and isinstance(r.get("lotes"), dict) else {"lotes": {}}
|
||
except FileNotFoundError:
|
||
return {"lotes": {}}
|
||
|
||
|
||
def escribir_registro(r):
|
||
os.makedirs(os.path.dirname(REGISTRO) or ".", exist_ok=True)
|
||
tmp = REGISTRO + ".tmp"
|
||
with open(tmp, "w", encoding="utf-8") as f:
|
||
json.dump(r, f, ensure_ascii=False, indent=1)
|
||
os.replace(tmp, REGISTRO)
|
||
|
||
|
||
# ------------------------------------------------------------------ lotes: formato
|
||
|
||
def validar_cambio(c):
|
||
e = []
|
||
if not isinstance(c, dict):
|
||
return ["debe ser un objeto"]
|
||
if "crear" in c:
|
||
t = c["crear"]
|
||
sobra = set(c) - {"crear", "si_no_hay"}
|
||
if sobra:
|
||
e.append("llaves que no van junto a «crear»: " + ", ".join(sorted(sobra)))
|
||
if not isinstance(t, dict):
|
||
return e + ["«crear» debe ser un objeto"]
|
||
malas = set(t) - set(CAMPOS) - {"id"}
|
||
if malas:
|
||
e.append("campos que no existen: " + ", ".join(sorted(malas)))
|
||
if not TAREA_ID.match(str(t.get("id") or "")):
|
||
e.append("«crear.id» es obligatorio: minúsculas, números y guiones (3 a 60)")
|
||
for k in ("titulo", "pista", "responsable"):
|
||
if not str(t.get(k) or "").strip():
|
||
e.append(f"falta «crear.{k}»")
|
||
if t.get("estado") and t["estado"] not in ESTADOS:
|
||
e.append(f"estado inválido: {t['estado']}")
|
||
if t.get("fecha_limite") and not re.match(r"^\d{4}-\d{2}-\d{2}$", str(t["fecha_limite"])):
|
||
e.append("«fecha_limite» va como AAAA-MM-DD")
|
||
sn = c.get("si_no_hay")
|
||
if sn is not None and not (isinstance(sn, str) or (isinstance(sn, list) and all(isinstance(x, str) for x in sn))):
|
||
e.append("«si_no_hay» es un texto o una lista de textos")
|
||
return e
|
||
sobra = set(c) - {"id", "buscar", "todos", "si_estado", "poner", "nota"}
|
||
if sobra:
|
||
e.append("llaves que no existen: " + ", ".join(sorted(sobra)))
|
||
if bool(c.get("id")) == bool(c.get("buscar")):
|
||
e.append("lleva «id» o «buscar» (uno de los dos)")
|
||
if c.get("todos") and not c.get("buscar"):
|
||
e.append("«todos» solo va con «buscar»")
|
||
b = c.get("buscar")
|
||
if b is not None:
|
||
textos = b if isinstance(b, list) else [b]
|
||
if not textos or not all(isinstance(x, str) for x in textos):
|
||
e.append("«buscar» es un texto o una lista de textos")
|
||
elif sum(len(clave(x)) for x in textos if not x.startswith("-")) < 4:
|
||
e.append("«buscar» necesita 4 letras o más (sin contar los «-texto»)")
|
||
nota = str(c.get("nota") or "").strip()
|
||
if not nota:
|
||
e.append("falta «nota»: el porqué del cambio, queda en el historial del pendiente")
|
||
elif len(nota) > 300:
|
||
e.append(f"«nota» de {len(nota)} caracteres; el backend guarda 300")
|
||
p = c.get("poner", {})
|
||
if not isinstance(p, dict):
|
||
e.append("«poner» debe ser un objeto")
|
||
else:
|
||
malas = set(p) - set(CAMPOS)
|
||
if malas:
|
||
e.append("campos que no existen en «poner»: " + ", ".join(sorted(malas)))
|
||
if "estado" in p and p["estado"] not in ESTADOS:
|
||
e.append(f"estado inválido en «poner»: {p['estado']}")
|
||
if p.get("fecha_limite") and not re.match(r"^\d{4}-\d{2}-\d{2}$", str(p["fecha_limite"])):
|
||
e.append("«fecha_limite» va como AAAA-MM-DD (o null para quitarla)")
|
||
se = c.get("si_estado")
|
||
if se is not None and (not isinstance(se, list) or not se or any(x not in ESTADOS for x in se)):
|
||
e.append("«si_estado» es una lista de estados (por-hacer, en-curso, bloqueado, hecho)")
|
||
return e
|
||
|
||
|
||
def cargar_lote(ruta):
|
||
try:
|
||
with open(ruta, encoding="utf-8") as f:
|
||
lote = json.load(f)
|
||
except ValueError as ex:
|
||
raise ValueError(f"{ruta}: no es JSON válido ({ex})")
|
||
if not isinstance(lote, dict):
|
||
raise ValueError(f"{ruta}: el lote debe ser un objeto JSON")
|
||
err = []
|
||
lid = str(lote.get("lote") or "")
|
||
if not LOTE_ID.match(lid):
|
||
err.append("«lote» debe ser AAAA-MM-DD_tema (minúsculas y guiones)")
|
||
elif os.path.splitext(os.path.basename(ruta))[0] != lid:
|
||
err.append(f"el archivo debe llamarse {lid}.json")
|
||
if not TAREA_ID.match(norm(lote.get("autor")) or "-"):
|
||
err.append("falta «autor»: el usuario con el que se escribe (p. ej. claude)")
|
||
if not str(lote.get("motivo") or "").strip():
|
||
err.append("falta «motivo»: qué tarea cerró este lote")
|
||
cambios = lote.get("cambios")
|
||
if not isinstance(cambios, list) or not cambios:
|
||
err.append("«cambios» debe ser una lista con al menos un cambio")
|
||
else:
|
||
for i, c in enumerate(cambios, 1):
|
||
err += [f"cambio {i}: {x}" for x in validar_cambio(c)]
|
||
if err:
|
||
raise ValueError(f"{ruta}:\n - " + "\n - ".join(err))
|
||
lote["autor"] = norm(lote["autor"])
|
||
return lote
|
||
|
||
|
||
def archivos_de(rutas):
|
||
out = []
|
||
for r in rutas:
|
||
if os.path.isdir(r):
|
||
out += sorted(f for f in glob.glob(os.path.join(r, "*.json")))
|
||
elif os.path.isfile(r):
|
||
out.append(r)
|
||
else:
|
||
raise ValueError(f"No existe {r}")
|
||
return out
|
||
|
||
|
||
# ------------------------------------------------------------------ lotes: plan y aplicación
|
||
|
||
def _ids(responsable):
|
||
return sorted({norm(x).replace(" ", "") for x in re.split(r"[,;/]+", str(responsable or "")) if norm(x)})
|
||
|
||
|
||
def igual(campo, actual, nuevo):
|
||
if campo == "responsable":
|
||
return _ids(actual) == _ids(nuevo)
|
||
if campo in ("etiquetas", "bloqueada_por"):
|
||
return sorted(norm(x) for x in (actual or [])) == sorted(norm(x) for x in (nuevo or []))
|
||
if campo == "enlaces":
|
||
par = lambda lista: [(str(x.get("titulo", "")).strip(), str(x.get("url", "")).strip())
|
||
for x in (lista or []) if isinstance(x, dict)]
|
||
return par(actual) == par(nuevo)
|
||
if campo == "fecha_limite":
|
||
return (actual or None) == (nuevo or None)
|
||
return str("" if actual is None else actual).strip() == str("" if nuevo is None else nuevo).strip()
|
||
|
||
|
||
def _quien_es(t):
|
||
return f"{t['id']} «{corto(t.get('titulo'), 70)}»"
|
||
|
||
|
||
def contiene(titulo, texto):
|
||
q = clave(texto)
|
||
return bool(q) and f" {q} " in f" {clave(titulo)} "
|
||
|
||
|
||
def coincide(titulo, buscar):
|
||
"""«buscar» es un texto o una lista de textos que el título debe tener todos; «-texto» = que no lo tenga
|
||
(["piloto", "v5.6", "-banco"] encuentra «Piloto V5.6: 5 tarjetas» y no «Banco del piloto V5.6»)."""
|
||
for x in (buscar if isinstance(buscar, list) else [buscar]):
|
||
if x.startswith("-"):
|
||
if contiene(titulo, x[1:]):
|
||
return False
|
||
elif not contiene(titulo, x):
|
||
return False
|
||
return True
|
||
|
||
|
||
def _texto_busqueda(buscar):
|
||
return " + ".join(f"«{x}»" for x in (buscar if isinstance(buscar, list) else [buscar]))
|
||
|
||
|
||
def _cambio_a(t, c):
|
||
"""Un pendiente existente → (detalle, petición) o (motivo, None) si no aplica."""
|
||
si = c.get("si_estado") or list(ABIERTOS)
|
||
if t.get("estado") not in si:
|
||
return (f"{_quien_es(t)} está en {NOMBRE.get(t.get('estado'), t.get('estado'))}; "
|
||
f"este cambio solo aplica en {', '.join(NOMBRE[x] for x in si)}"), None
|
||
poner = {k: v for k, v in (c.get("poner") or {}).items() if not igual(k, t.get(k), v)}
|
||
if valores_malos(poner):
|
||
return "✗ " + f"{_quien_es(t)}: " + "; ".join(valores_malos(poner)), None
|
||
partes = [f"{k}: {corto(t.get(k), 30)} → {corto(v, 30)}" for k, v in poner.items()]
|
||
detalle = f"{_quien_es(t)}: " + ("; ".join(partes) + " · " if partes else "") + f"nota «{corto(c['nota'], 80)}»"
|
||
return detalle, ("PATCH", "/api/tarea?id=" + urllib.parse.quote(t["id"]), dict(poner, _nota=c["nota"].strip()), t["id"])
|
||
|
||
|
||
def planear(c, tareas):
|
||
"""Qué haría un cambio contra el tablero de ahora → (acción, detalle, [peticiones]).
|
||
acción: crear · cambiar · saltar · invalido. Cada petición: (método, ruta, cuerpo, id del pendiente)."""
|
||
por_id = {t.get("id"): t for t in tareas}
|
||
if "crear" in c:
|
||
t = c["crear"]
|
||
if t["id"] in por_id:
|
||
return "saltar", f"ya existe {_quien_es(por_id[t['id']])}", []
|
||
evitar = c.get("si_no_hay")
|
||
for x in ([evitar] if isinstance(evitar, str) else (evitar or [])):
|
||
for o in tareas:
|
||
if o.get("estado") != "hecho" and contiene(o.get("titulo"), x):
|
||
return "saltar", f"ya hay uno abierto parecido (dice «{x}»): {_quien_es(o)}", []
|
||
cuerpo = dict(t)
|
||
cuerpo.setdefault("estado", "por-hacer")
|
||
if valores_malos(cuerpo):
|
||
return "invalido", f"nuevo {cuerpo['id']}: " + "; ".join(valores_malos(cuerpo)), []
|
||
datos = [cuerpo["pista"], NOMBRE.get(cuerpo["estado"], cuerpo["estado"]), cuerpo["responsable"]]
|
||
if cuerpo.get("fecha_limite"):
|
||
datos.append("vence " + fecha_corta(cuerpo["fecha_limite"]))
|
||
return "crear", f"nuevo {cuerpo['id']} «{corto(cuerpo['titulo'], 70)}» ({', '.join(datos)})", \
|
||
[("POST", "/api/tareas", cuerpo, cuerpo["id"])]
|
||
if c.get("id"):
|
||
t = por_id.get(c["id"])
|
||
if not t:
|
||
return "saltar", f"no existe el pendiente {c['id']} (¿se borró?)", []
|
||
objetivo = [t]
|
||
else:
|
||
q = _texto_busqueda(c["buscar"])
|
||
abiertos = [o for o in tareas if o.get("estado") != "hecho" and coincide(o.get("titulo"), c["buscar"])]
|
||
if c.get("todos"):
|
||
if not abiertos:
|
||
return "saltar", f"ningún pendiente abierto dice {q} en el título", []
|
||
objetivo = abiertos
|
||
else:
|
||
cand = abiertos or [o for o in tareas if o.get("estado") == "hecho" and coincide(o.get("titulo"), c["buscar"])]
|
||
if len(cand) != 1:
|
||
if not cand:
|
||
return "saltar", f"ningún pendiente dice {q} en el título", []
|
||
return "saltar", (f"{q} coincide con {len(cand)} pendientes (usa «todos» o un texto más preciso): "
|
||
+ "; ".join(_quien_es(o) for o in cand[:4])), []
|
||
objetivo = cand
|
||
planes = [_cambio_a(t, c) for t in objetivo]
|
||
peticiones = [p for _, p in planes if p]
|
||
if len(objetivo) == 1:
|
||
detalle, p = planes[0]
|
||
if not p:
|
||
malo = detalle.startswith("✗ ")
|
||
return ("invalido" if malo else "saltar"), (detalle[2:] if malo else detalle), []
|
||
return "cambiar", detalle, peticiones
|
||
detalle = f"{_texto_busqueda(c['buscar'])}, {len(objetivo)} pendientes:\n" + "\n".join(
|
||
f" – {d}" if p else f" – se salta: {d}" for d, p in planes)
|
||
malos = [d for d, p in planes if not p and d.startswith("✗ ")]
|
||
if malos:
|
||
return "invalido", detalle, []
|
||
return ("cambiar" if peticiones else "saltar"), detalle, peticiones
|
||
|
||
|
||
def simular(tareas, peticion):
|
||
"""En «revisar», que los cambios siguientes del mismo lote vean los anteriores."""
|
||
metodo, ruta, cuerpo, rid = peticion
|
||
if metodo == "POST":
|
||
return tareas + [dict(cuerpo)]
|
||
return [dict(t, **{k: v for k, v in cuerpo.items() if not k.startswith("_")}) if t.get("id") == rid else t
|
||
for t in tareas]
|
||
|
||
|
||
def correr(rutas, aplicar, silencioso=False):
|
||
archivos = archivos_de(rutas)
|
||
lotes = [(a, cargar_lote(a)) for a in archivos] # todos válidos o ninguno se aplica
|
||
ids = [lote["lote"] for _, lote in lotes]
|
||
if len(ids) != len(set(ids)):
|
||
raise ValueError("Hay dos lotes con el mismo «lote».")
|
||
reg = leer_registro()
|
||
tareas, _ = leer_tareas()
|
||
por_aplicar = errores = 0
|
||
di = (lambda *a: None) if silencioso else print
|
||
if not lotes:
|
||
di("No hay lotes en " + ", ".join(rutas) + ".")
|
||
for archivo, lote in lotes:
|
||
lid = lote["lote"]
|
||
r = reg["lotes"].get(lid)
|
||
previos = (r or {}).get("cambios", {})
|
||
cambiado = [i for i, c in enumerate(lote["cambios"], 1)
|
||
if str(i) in previos and previos[str(i)].get("huella") != huella(c)]
|
||
if r and r.get("completo"):
|
||
di(f"✓ {lid}: aplicado el {fecha_corta(r.get('fin'))} ({r.get('motivo', '')[:70]})"
|
||
+ (f"\n ⚠ el archivo cambió después de aplicarse (cambios {', '.join(map(str, cambiado))}); "
|
||
"no se vuelve a aplicar: lo nuevo va en otro lote" if cambiado else ""))
|
||
continue
|
||
if cambiado:
|
||
di(f"✗ {lid}: los cambios {', '.join(map(str, cambiado))} ya no son los que se aplicaron. "
|
||
"Un lote aplicado no se edita: escribe uno nuevo.")
|
||
errores += 1
|
||
continue
|
||
di(f"\n▶ {lid} — {lote['motivo']} (lo escribe: {lote['autor']})")
|
||
if aplicar and not r:
|
||
r = reg["lotes"][lid] = {"archivo": os.path.basename(archivo), "autor": lote["autor"],
|
||
"motivo": lote["motivo"], "inicio": ahora(), "fin": None, "completo": False,
|
||
"aplicado_por": os.environ.get("SUDO_USER") or os.environ.get("USER") or "?",
|
||
"cambios": {}}
|
||
for i, c in enumerate(lote["cambios"], 1):
|
||
previo = previos.get(str(i))
|
||
if previo and previo.get("resultado") in ("hecho", "saltado"):
|
||
di(f" {i}. ya {previo['resultado']}: {previo['detalle']}")
|
||
continue
|
||
accion, detalle, peticiones = planear(c, tareas)
|
||
signo = {"crear": "+", "cambiar": "~", "saltar": "·", "invalido": "✗"}[accion]
|
||
if not aplicar:
|
||
di(f" {i}. {signo} {'se salta: ' if accion == 'saltar' else ''}{detalle}")
|
||
por_aplicar += len(peticiones)
|
||
for p in peticiones:
|
||
tareas = simular(tareas, p)
|
||
errores += accion == "invalido"
|
||
continue
|
||
if accion == "invalido":
|
||
r["cambios"][str(i)] = {"resultado": "error", "detalle": detalle, "huella": huella(c), "cuando": ahora()}
|
||
di(f" {i}. ✗ no se manda: {detalle}")
|
||
errores += 1
|
||
elif accion == "saltar":
|
||
r["cambios"][str(i)] = {"resultado": "saltado", "detalle": detalle, "huella": huella(c), "cuando": ahora()}
|
||
di(f" {i}. · se salta: {detalle}")
|
||
else:
|
||
ya = set((previo or {}).get("hechos_en") or []) # en un reintento, lo que ya salió no se repite
|
||
fallas = []
|
||
for metodo, ruta, cuerpo, rid in peticiones:
|
||
if rid in ya:
|
||
continue
|
||
st, resp = llamar(metodo, ruta, lote["autor"], cuerpo)
|
||
if st in (200, 201) and resp.get("ok"):
|
||
t = resp.get("tarea") or {}
|
||
tareas = [x for x in tareas if x.get("id") != t.get("id")] + [t]
|
||
ya.add(rid)
|
||
if resp.get("aviso"):
|
||
di(f" aviso del backend ({rid}): {resp['aviso']}")
|
||
else:
|
||
fallas.append(f"{rid}: {resp.get('error') or st}")
|
||
r["cambios"][str(i)] = {"resultado": "error" if fallas else "hecho",
|
||
"detalle": detalle + (" → " + "; ".join(fallas) if fallas else ""),
|
||
"hechos_en": sorted(ya), "huella": huella(c), "cuando": ahora()}
|
||
if fallas:
|
||
di(f" {i}. ✗ {detalle}\n el backend dijo: " + "; ".join(fallas))
|
||
errores += 1
|
||
else:
|
||
di(f" {i}. {signo} {detalle}")
|
||
escribir_registro(reg) # después de cada cambio: si algo se cae, no se pierde lo aplicado
|
||
if aplicar:
|
||
r["completo"] = all(r["cambios"].get(str(i), {}).get("resultado") in ("hecho", "saltado")
|
||
for i in range(1, len(lote["cambios"]) + 1))
|
||
r["fin"] = ahora() if r["completo"] else None
|
||
escribir_registro(reg)
|
||
if not aplicar and not silencioso:
|
||
print(f"\n{por_aplicar} cambio(s) por aplicar." + (" Para aplicarlos: ./scripts/tablero.sh aplicar" if por_aplicar else ""))
|
||
return por_aplicar, errores
|
||
|
||
|
||
# ------------------------------------------------------------------ foto y resumen
|
||
|
||
def foto():
|
||
tareas, equipo = leer_tareas()
|
||
out = {"tomada": ahora(), "api": API, "tareas": tareas, "equipo": equipo}
|
||
st, h = llamar("GET", "/api/hoy", LECTOR)
|
||
out["enfoque"] = h.get("enfoque") if st == 200 else None
|
||
st, a = llamar("GET", "/api/actividad?dias=14&limite=150", LECTOR)
|
||
out["actividad"] = a.get("eventos", []) if st == 200 else []
|
||
out["lotes"] = {k: {"completo": v.get("completo"), "fin": v.get("fin"), "motivo": v.get("motivo")}
|
||
for k, v in leer_registro()["lotes"].items()}
|
||
json.dump(out, sys.stdout, ensure_ascii=False, indent=1)
|
||
sys.stdout.write("\n")
|
||
|
||
|
||
def resumen(ruta):
|
||
with open(ruta, encoding="utf-8") as f:
|
||
F = json.load(f)
|
||
tareas, eq = F.get("tareas") or [], F.get("equipo") or {}
|
||
hoy = str(F.get("tomada") or ahora())[:10]
|
||
apodo = {"Leopoldo": "Leo"} # como en el sitio (Kanban.nombreCorto)
|
||
nombre = lambda i: (lambda n: apodo.get(n, n))(((eq.get(i) or {}).get("nombre") or i or "—").split(" ")[0])
|
||
estado_de = {t.get("id"): t.get("estado") for t in tareas}
|
||
abiertos = [t for t in tareas if t.get("estado") != "hecho"]
|
||
vencido = lambda t: bool(t.get("fecha_limite")) and t.get("estado") != "hecho" and t["fecha_limite"] < hoy
|
||
cuenta = {e: sum(1 for t in tareas if t.get("estado") == e) for e in ESTADOS}
|
||
o = [f"# Tablero de pendientes — foto del {fecha_corta(F.get('tomada'))}", "",
|
||
"Generado por `./scripts/tablero.sh foto` (no se edita: se vuelve a bajar). Fuente: " + str(F.get("api")), "",
|
||
f"**Abiertos: {len(abiertos)}** — por hacer {cuenta['por-hacer']} · en curso {cuenta['en-curso']} · "
|
||
f"bloqueados {cuenta['bloqueado']} · **vencidos {sum(map(vencido, abiertos))}** · hechos {cuenta['hecho']} en total", ""]
|
||
en = F.get("enfoque")
|
||
if en:
|
||
o += [f"## Enfoque de la semana ({nombre(en.get('editado_por'))}, {fecha_corta(en.get('editado_en'))})", "",
|
||
en.get("titulo") or "", ""] + [f"- {p}" for p in (en.get("puntos") or [])] + [""]
|
||
else:
|
||
o += ["## Enfoque de la semana", "", "Nadie lo ha escrito en el sitio (se ve el de docs/datos/proyecto.json).", ""]
|
||
|
||
def linea(t):
|
||
bits = [NOMBRE.get(t.get("estado"), t.get("estado")), t.get("pista") or "—"]
|
||
if t.get("version_tarjeta") not in (None, "", "—"):
|
||
bits.append(t["version_tarjeta"])
|
||
if t.get("fecha_limite"):
|
||
bits.append(("**VENCIDO** " if vencido(t) else "vence ") + fecha_corta(t["fecha_limite"]))
|
||
espera = [x for x in (t.get("bloqueada_por") or []) if estado_de.get(x) != "hecho"]
|
||
if espera:
|
||
bits.append("espera a " + ", ".join(espera))
|
||
return f"- {t.get('titulo')} — `{t.get('id')}` · " + " · ".join(bits)
|
||
|
||
orden = lambda t: (ESTADOS.index(t.get("estado")) if t.get("estado") in ESTADOS else 9, t.get("fecha_limite") or "9999")
|
||
personas = {}
|
||
for t in abiertos:
|
||
for p in (t.get("responsable_ids") or _ids(t.get("responsable")) or ["(sin responsable)"]):
|
||
personas.setdefault(p, []).append(t)
|
||
o += ["## Abiertos por persona", ""]
|
||
for p in sorted(personas, key=lambda x: (-len(personas[x]), x)):
|
||
o += [f"### {nombre(p)} ({len(personas[p])})", ""] + [linea(t) for t in sorted(personas[p], key=orden)] + [""]
|
||
hace14 = (datetime.date.fromisoformat(hoy) - datetime.timedelta(days=14)).isoformat()
|
||
hechos = sorted((t for t in tareas if t.get("estado") == "hecho" and str(t.get("modificado_en") or "") >= hace14),
|
||
key=lambda t: str(t.get("modificado_en")), reverse=True)
|
||
o += ["## Hechos en los últimos 14 días", ""] + \
|
||
[f"- {fecha_corta(t.get('modificado_en'))} · {t.get('titulo')} — `{t.get('id')}`" for t in hechos] + \
|
||
([] if hechos else ["- (ninguno)"]) + [""]
|
||
o += ["## Actividad (14 días, lo más nuevo arriba)", ""]
|
||
for e in (F.get("actividad") or [])[:60]:
|
||
if e.get("tipo") == "pendiente":
|
||
o.append(f"- {fecha_corta(e.get('cuando'))} · {nombre(e.get('quien'))}: {corto(e.get('cambio'), 140)} "
|
||
f"— «{corto(e.get('titulo'), 60)}» `{e.get('id')}`")
|
||
else:
|
||
o.append(f"- {fecha_corta(e.get('cuando'))} · {e.get('tecnico') or nombre(e.get('quien'))}: acta "
|
||
f"{e.get('acta')} de {e.get('serie') or 'sin serie'} {e.get('resultado') or ''}".rstrip())
|
||
lotes = F.get("lotes") or {}
|
||
o += ["", "## Lotes de Claude", ""] + [
|
||
f"- `{k}` — {'aplicado el ' + fecha_corta(v.get('fin')) if v.get('completo') else 'a medias'}: {v.get('motivo') or ''}"
|
||
for k, v in sorted(lotes.items(), reverse=True)] + ([] if lotes else ["- (ninguno aplicado todavía)"])
|
||
print("\n".join(o))
|
||
|
||
|
||
# ------------------------------------------------------------------ main
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0],
|
||
formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__)
|
||
ap.add_argument("accion", choices=("foto", "revisar", "aplicar", "pendientes", "resumen"))
|
||
ap.add_argument("rutas", nargs="*", help="lotes (archivos o carpetas); para «resumen», la foto")
|
||
a = ap.parse_args()
|
||
aqui = os.path.join(os.path.dirname(os.path.abspath(__file__)), "lotes")
|
||
try:
|
||
if a.accion == "foto":
|
||
foto()
|
||
return 0
|
||
if a.accion == "resumen":
|
||
resumen(a.rutas[0] if a.rutas else "_estado/tablero.json")
|
||
return 0
|
||
rutas = a.rutas or [aqui]
|
||
if a.accion == "pendientes":
|
||
n, err = correr(rutas, aplicar=False, silencioso=True)
|
||
print(f"Tablero: {n} cambio(s) de lotes por aplicar → ./scripts/tablero.sh revisar" if n
|
||
else "Tablero: no hay lotes por aplicar.")
|
||
return 3 if err else 0
|
||
n, err = correr(rutas, aplicar=(a.accion == "aplicar"))
|
||
return 3 if err else 0
|
||
except ValueError as e:
|
||
print("✗ " + str(e), file=sys.stderr)
|
||
return 1
|
||
except SinBackend as e:
|
||
print("✗ " + str(e), file=sys.stderr)
|
||
return 2
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|