Helgen-Services/hbox-docs
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
b41b7710b4
hbox-docs/backend/incidencias.py

405 lines
22 KiB
Python
Raw Blame History

"""Incidencias del sitio (27-oct-2026): lo que falla en drogheda.helgen.online, reportado desde cualquier página
(sección «Problemas del sitio» al pie), desde la página de entrada (sin sesión) o detectado solo (vigía).
Lo engancha app.py al final con registrar(app, globals()), como instalaciones.py.
Datos: DATOS_BASE/incidencias/INC-0001.json (un JSON por incidencia) + _indice.csv; folio con candado en _folio.
DATOS_BASE/incidencias/_errores/cliente-AAAA-MM-DD.jsonl errores de JavaScript y llamadas fallidas (navegador)
DATOS_BASE/incidencias/_errores/servidor-AAAA-MM-DD.jsonl excepciones y respuestas 5xx del backend
Nada se borra: una incidencia se resuelve o se descarta.
Rutas con sesión:
GET /api/incidencias?estado=abiertas|cerradas|todas&tipo=&origen=&q= lista + conteos
GET /api/incidencias/resumen lo que muestra el pie de cada página (abiertas, mías, últimas 3)
GET /api/incidencia?id=INC-0001
POST /api/incidencias {titulo?, descripcion, tipo, pagina, contexto, firma?}
(firma: quien detecta solo —vigia o claude— manda una huella; si ya hay una abierta con esa huella se suma
una ocurrencia en vez de abrir otra)
PATCH /api/incidencia?id= {comentario} cualquiera · {estado, gravedad, asignado, publica,
resumen_publico, duplicado_de, tarea_id, titulo, tipo} quien edita (INCIDENCIAS_EDITORES + administradores);
quien la reportó puede marcarla resuelta o reabrirla
POST /api/incidencias/error errores del navegador (los manda js/incidencias.js solo)
GET /api/incidencias/errores?dias=2 (editores) errores de cliente y servidor agrupados
Rutas sin sesión (nginx las deja pasar sin auth_request y SIN X-Remote-User):
POST /api/incidencias/publica {nombre, contacto, descripcion, tipo} desde /entrar.html
GET /api/incidencias/publicas avisos marcados «publica» (solo título público, estado y fecha)
"""
import datetime
import fcntl
import hashlib
import json
import os
import re
import traceback
from flask import got_request_exception, jsonify, request
TIPOS = ("acceso", "error", "datos", "lento", "sugerencia", "otro")
NOMBRE_TIPO = {"acceso": "No puedo entrar", "error": "Error o pantalla rota", "datos": "Datos incorrectos",
"lento": "Lento", "sugerencia": "Sugerencia", "otro": "Otro"}
ESTADOS = ("nueva", "en-revision", "en-curso", "esperando", "resuelta", "descartada")
ABIERTAS = ("nueva", "en-revision", "en-curso", "esperando")
GRAVEDADES = ("alta", "media", "baja")
ORIGENES = ("sitio", "entrada", "vigia", "agente")
SISTEMA = {"vigia", "claude"} # pueden abrir con origen vigia/agente y con firma
COLS = ["id", "titulo", "tipo", "estado", "gravedad", "origen", "reportado_por", "asignado", "pagina", "ocurrencias",
"publica", "tarea_id", "duplicado_de", "creado_en", "modificado_en", "resuelta_en"]
MAX_ERRORES_DIA = 3000
def registrar(app, g):
carpeta = os.environ.get("INCIDENCIAS_DIR", os.path.join(g["DATOS_BASE"], "incidencias"))
carpeta_err = os.path.join(carpeta, "_errores")
editores = {u.strip().lower() for u in os.environ.get("INCIDENCIAS_EDITORES", "leo,gibranlp,adriana,claude").split(",")
if u.strip()}
texto, ahora, hist = g["_texto"], g["ahora"], g["_hist"]
def yo():
return g["current_user"]()
def puede_editar(u):
return bool(u) and (u.lower() in editores or u.lower() in g["ADMINS"])
def ruta(iid):
return os.path.join(carpeta, iid + ".json")
def todas():
return [d for d in g["_listar_json"](carpeta) if str(d.get("id", "")).startswith("INC-")]
def guardar(d):
g["_escribir_json"](ruta(d["id"]), d)
g["_reescribir_indice"](carpeta, todas(), COLS)
def siguiente_folio():
os.makedirs(carpeta, exist_ok=True)
with open(os.path.join(carpeta, "_folio"), "a+", encoding="utf-8") as f:
fcntl.flock(f, fcntl.LOCK_EX)
f.seek(0)
try:
n = int((f.read() or "0").strip() or 0)
except ValueError:
n = 0
existentes = [int(m.group(1)) for x in os.listdir(carpeta) if (m := re.match(r"INC-(\d+)\.json$", x))]
n = max([n] + existentes) + 1
f.seek(0)
f.truncate()
f.write(str(n))
f.flush()
fcntl.flock(f, fcntl.LOCK_UN)
return f"INC-{n:04d}"
def contexto_limpio(c):
"""El contexto técnico que manda la página: se guarda acotado (nada de cookies ni contraseñas)."""
if not isinstance(c, dict):
return {}
out = {}
for k in ("pantalla", "navegador", "idioma", "referer", "hora_local"):
if c.get(k):
out[k] = texto(c.get(k), 200)
for k in ("errores", "llamadas"):
v = c.get(k)
if isinstance(v, list):
out[k] = [{kk: texto(x.get(kk), 300) for kk in list(x)[:8]} for x in v[-10:] if isinstance(x, dict)]
return out
def nueva(d, u, origen):
tipo = d.get("tipo") if d.get("tipo") in TIPOS else "otro"
desc = texto(d.get("descripcion"), 4000)
titulo = texto(d.get("titulo"), 140) or (desc.split("\n")[0][:120] if desc else NOMBRE_TIPO[tipo])
r = {"id": siguiente_folio(), "titulo": titulo, "descripcion": desc, "tipo": tipo, "estado": "nueva",
"gravedad": d.get("gravedad") if d.get("gravedad") in GRAVEDADES else ("alta" if tipo == "acceso" else "media"),
"origen": origen, "reportado_por": u or "", "nombre": texto(d.get("nombre"), 80),
"contacto": texto(d.get("contacto"), 120), "asignado": "", "pagina": texto(d.get("pagina"), 300),
"navegador": texto(request.headers.get("User-Agent"), 200), "contexto": contexto_limpio(d.get("contexto")),
"firma": texto(d.get("firma"), 80), "ocurrencias": 1, "ultima_vez": ahora(), "evidencia": [],
"publica": False, "resumen_publico": "", "tarea_id": "", "duplicado_de": "", "comentarios": [],
"creado_por": u or "(sin sesión)", "creado_en": ahora(), "resuelta_en": None}
if isinstance(d.get("evidencia"), list):
r["evidencia"] = [texto(x, 600) for x in d["evidencia"][:20]]
hist(r, u or "(sin sesión)", f"reportada desde {origen}")
guardar(r)
return r
def para_lista(r, completo=False):
if completo:
return r
keys = ("id", "titulo", "tipo", "estado", "gravedad", "origen", "reportado_por", "nombre", "asignado", "pagina",
"ocurrencias", "ultima_vez", "publica", "tarea_id", "duplicado_de", "creado_en", "modificado_en",
"resuelta_en")
out = {k: r.get(k) for k in keys}
out["comentarios"] = len(r.get("comentarios") or [])
out["ultimo_comentario"] = (r.get("comentarios") or [{}])[-1] if r.get("comentarios") else None
return out
# ------------------------------------------------------------------ con sesión
@app.get("/api/incidencias")
def incidencias_lista():
u = yo()
if not u:
return g["_sin_usuario"]()
est = (request.args.get("estado") or "abiertas").strip()
tipo = (request.args.get("tipo") or "").strip()
origen = (request.args.get("origen") or "").strip()
q = (request.args.get("q") or "").strip().lower()
regs = todas()
conteo = {"abiertas": sum(1 for r in regs if r.get("estado") in ABIERTAS),
"por_estado": {e: sum(1 for r in regs if r.get("estado") == e) for e in ESTADOS},
"por_tipo": {t: sum(1 for r in regs if r.get("tipo") == t and r.get("estado") in ABIERTAS) for t in TIPOS}}
if est == "abiertas":
regs = [r for r in regs if r.get("estado") in ABIERTAS]
elif est == "cerradas":
regs = [r for r in regs if r.get("estado") not in ABIERTAS]
elif est in ESTADOS:
regs = [r for r in regs if r.get("estado") == est]
if tipo in TIPOS:
regs = [r for r in regs if r.get("tipo") == tipo]
if origen in ORIGENES:
regs = [r for r in regs if r.get("origen") == origen]
if q:
regs = [r for r in regs if q in (r.get("id", "") + " " + r.get("titulo", "") + " " + r.get("descripcion", "")).lower()]
abiertas = sorted([r for r in regs if r.get("estado") in ABIERTAS],
key=lambda r: r.get("modificado_en") or r.get("creado_en") or "", reverse=True)
abiertas.sort(key=lambda r: GRAVEDADES.index(r["gravedad"]) if r.get("gravedad") in GRAVEDADES else 1) # estable
cerradas = sorted([r for r in regs if r.get("estado") not in ABIERTAS],
key=lambda r: r.get("modificado_en") or "", reverse=True)
return jsonify(ok=True, incidencias=[para_lista(r) for r in abiertas + cerradas][:300], conteo=conteo,
tipos=[{"id": t, "nombre": NOMBRE_TIPO[t]} for t in TIPOS], estados=list(ESTADOS),
gravedades=list(GRAVEDADES), puede_editar=puede_editar(u), yo=u)
@app.get("/api/incidencias/resumen")
def incidencias_resumen():
u = yo()
if not u:
return g["_sin_usuario"]()
regs = [r for r in todas() if r.get("estado") in ABIERTAS]
regs.sort(key=lambda r: r.get("modificado_en") or r.get("creado_en") or "", reverse=True)
mias = [r for r in regs if (r.get("reportado_por") or "").lower() == u.lower()]
return jsonify(ok=True, abiertas=len(regs), mias=len(mias),
ultimas=[{k: r.get(k) for k in ("id", "titulo", "estado", "tipo")} for r in regs[:3]])
@app.get("/api/incidencia")
def incidencia_una():
if not yo():
return g["_sin_usuario"]()
r = g["_leer_json"](ruta(g["slug"](request.args.get("id"), 12)))
if not r:
return jsonify(ok=False, error="No existe esa incidencia."), 404
return jsonify(ok=True, incidencia=r, puede_editar=puede_editar(yo()))
@app.post("/api/incidencias")
def incidencias_crear():
u = yo()
if not u:
return g["_sin_usuario"]()
d = g["_json_body"]()
if d is None:
return jsonify(ok=False, error="JSON inválido"), 400
if not texto(d.get("descripcion")) and not texto(d.get("titulo")):
return jsonify(ok=False, error="Cuéntanos qué pasó (aunque sea una línea)."), 400
origen = d.get("origen") if d.get("origen") in ORIGENES else "sitio"
if origen in ("vigia", "agente") and u.lower() not in SISTEMA:
origen = "sitio"
firma = texto(d.get("firma"), 80)
if firma and u.lower() in SISTEMA:
for r in todas():
if r.get("firma") == firma and r.get("estado") in ABIERTAS:
r["ocurrencias"] = int(r.get("ocurrencias") or 1) + 1
r["ultima_vez"] = ahora()
ev = [texto(x, 600) for x in (d.get("evidencia") or [])][:5]
r["evidencia"] = (r.get("evidencia") or [])[-(20 - len(ev)):] + ev if ev else r.get("evidencia") or []
hist(r, u, f"se repitió (ocurrencia {r['ocurrencias']})")
guardar(r)
return jsonify(ok=True, id=r["id"], nueva=False, ocurrencias=r["ocurrencias"])
r = nueva(d, u, origen)
return jsonify(ok=True, id=r["id"], nueva=True, incidencia=r)
@app.patch("/api/incidencia")
def incidencia_cambiar():
u = yo()
if not u:
return g["_sin_usuario"]()
r = g["_leer_json"](ruta(g["slug"](request.args.get("id"), 12)))
if not r:
return jsonify(ok=False, error="No existe esa incidencia."), 404
d = g["_json_body"]()
if d is None:
return jsonify(ok=False, error="JSON inválido"), 400
editor = puede_editar(u)
reporto = (r.get("reportado_por") or "").lower() == u.lower()
cambios = []
if texto(d.get("comentario")):
r.setdefault("comentarios", []).append({"quien": u, "cuando": ahora(), "texto": texto(d.get("comentario"), 4000)})
cambios.append("comentó")
if "estado" in d:
e = d.get("estado")
if e not in ESTADOS:
return jsonify(ok=False, error="Estado inválido."), 400
if not editor and not (reporto and e in ("resuelta", "nueva")):
return jsonify(ok=False, error="Solo quien atiende las incidencias cambia su estado "
"(tú puedes marcar la tuya como resuelta o reabrirla)."), 403
if e != r.get("estado"):
cambios.append(f"estado: {r.get('estado')} → {e}")
r["estado"] = e
r["resuelta_en"] = ahora() if e in ("resuelta", "descartada") else None
for campo, validos, largo in (("gravedad", GRAVEDADES, 0), ("tipo", TIPOS, 0), ("asignado", None, 40),
("tarea_id", None, 60), ("duplicado_de", None, 12), ("titulo", None, 140),
("resumen_publico", None, 200)):
if campo not in d:
continue
if not editor:
return jsonify(ok=False, error=f"Solo quien atiende las incidencias cambia «{campo}»."), 403
v = d.get(campo) if validos else texto(d.get(campo), largo)
if validos and v not in validos:
return jsonify(ok=False, error=f"Valor inválido para {campo}."), 400
if v != r.get(campo):
cambios.append(f"{campo}: {g['_corto'](r.get(campo))} → {g['_corto'](v)}")
r[campo] = v
if "publica" in d:
if not editor:
return jsonify(ok=False, error="Solo quien atiende las incidencias decide qué se ve en la entrada."), 403
v = g["_bool"](d.get("publica"), False)
if v and not texto(r.get("resumen_publico")): # el título puede traer nombres o teléfonos: nunca se publica
return jsonify(ok=False, error="Para mostrarla en la entrada escribe primero el texto corto público "
"(sin nombres, teléfonos ni datos internos)."), 400
if v != bool(r.get("publica")):
cambios.append("se muestra en la página de entrada" if v else "ya no se muestra en la entrada")
r["publica"] = v
if not cambios:
return jsonify(ok=True, incidencia=r, sin_cambios=True)
hist(r, u, "; ".join(cambios))
guardar(r)
return jsonify(ok=True, incidencia=r)
# ------------------------------------------------------------------ errores (navegador y servidor)
def anotar_error(lado, reg):
os.makedirs(carpeta_err, exist_ok=True)
fp = os.path.join(carpeta_err, f"{lado}-{datetime.date.today():%Y-%m-%d}.jsonl")
try:
if os.path.exists(fp) and os.path.getsize(fp) > MAX_ERRORES_DIA * 1200:
return False
with open(fp, "a", encoding="utf-8") as f:
f.write(json.dumps(reg, ensure_ascii=False) + "\n")
return True
except OSError:
return False
@app.post("/api/incidencias/error")
def incidencias_error():
u = yo()
if not u:
return g["_sin_usuario"]()
d = g["_json_body"]() or {}
reg = {"cuando": ahora(), "usuario": u, "tipo": texto(d.get("tipo"), 20) or "js",
"mensaje": texto(d.get("mensaje"), 400), "fuente": texto(d.get("fuente"), 200),
"linea": g["_entero"](d.get("linea"), 0), "pagina": texto(d.get("pagina"), 200),
"url_api": texto(d.get("url_api"), 200), "estado_http": g["_entero"](d.get("estado_http"), 0),
"pila": texto(d.get("pila"), 1200), "navegador": texto(request.headers.get("User-Agent"), 160)}
return jsonify(ok=anotar_error("cliente", reg))
def leer_errores(dias):
out = {"cliente": [], "servidor": []}
for i in range(dias):
dia = datetime.date.today() - datetime.timedelta(days=i)
for lado in out:
fp = os.path.join(carpeta_err, f"{lado}-{dia:%Y-%m-%d}.jsonl")
if os.path.isfile(fp):
with open(fp, encoding="utf-8", errors="ignore") as f:
for linea in f:
try:
out[lado].append(json.loads(linea))
except Exception: # noqa: BLE001
pass
return out
def agrupar(regs, claves):
grupos = {}
for r in regs:
k = hashlib.sha1("|".join(str(r.get(c, "")) for c in claves).encode()).hexdigest()[:12]
gr = grupos.setdefault(k, {"firma": k, "veces": 0, "usuarios": set(), "primero": r.get("cuando"),
"ultimo": r.get("cuando"), "ejemplo": r})
gr["veces"] += 1
gr["usuarios"].add(r.get("usuario") or "")
gr["ultimo"] = max(gr["ultimo"] or "", r.get("cuando") or "")
out = sorted(grupos.values(), key=lambda x: -x["veces"])
for x in out:
x["usuarios"] = sorted(x["usuarios"])
return out
@app.get("/api/incidencias/errores")
def incidencias_errores():
u = yo()
if not puede_editar(u):
return jsonify(ok=False, error="Solo quien atiende las incidencias ve los errores."), 403
dias = max(1, min(14, g["_entero"](request.args.get("dias"), 2)))
e = leer_errores(dias)
return jsonify(ok=True, dias=dias,
cliente=agrupar(e["cliente"], ("tipo", "mensaje", "fuente", "linea", "url_api", "estado_http")),
servidor=agrupar(e["servidor"], ("tipo", "ruta", "error")))
def _excepcion(sender, exception, **extra): # noqa: ARG001
try:
anotar_error("servidor", {"cuando": ahora(), "tipo": "excepcion", "ruta": request.path,
"metodo": request.method, "usuario": yo(),
"error": f"{type(exception).__name__}: {exception}"[:400],
"traza": "".join(traceback.format_exception(exception))[-3000:]})
except Exception: # noqa: BLE001
pass
got_request_exception.connect(_excepcion, app, weak=False)
@app.after_request
def _respuestas_5xx(resp):
try:
if resp.status_code >= 500 and request.path.startswith("/api/"):
cuerpo = resp.get_json(silent=True) if resp.is_json else None
anotar_error("servidor", {"cuando": ahora(), "tipo": f"respuesta-{resp.status_code}", "ruta": request.path,
"metodo": request.method, "usuario": yo(),
"error": texto((cuerpo or {}).get("error") if isinstance(cuerpo, dict) else "", 400)})
except Exception: # noqa: BLE001
pass
return resp
# ------------------------------------------------------------------ sin sesión (página de entrada)
@app.post("/api/incidencias/publica")
def incidencias_publica():
d = g["_json_body"]() or request.form or {}
if texto(d.get("sitio_web")): # campo trampa: solo lo llenan los robots
return jsonify(ok=True, id="INC-0000")
desc = texto(d.get("descripcion"), 2000)
contacto = texto(d.get("contacto"), 120)
nombre = texto(d.get("nombre"), 80)
if len(desc) < 5:
return jsonify(ok=False, error="Cuéntanos qué pasa (aunque sea una línea)."), 400
if not (nombre or contacto):
return jsonify(ok=False, error="Dinos quién eres (tu usuario o tu nombre) para poder ayudarte."), 400
# límite: 5 por IP por hora (además del limit_req de nginx)
ip = (request.headers.get("X-Real-IP") or request.remote_addr or "")[:60]
hace_1h = (datetime.datetime.now() - datetime.timedelta(hours=1)).replace(microsecond=0).isoformat()
recientes = [r for r in todas() if r.get("origen") == "entrada" and (r.get("creado_en") or "") >= hace_1h
and (r.get("contexto") or {}).get("ip") == ip]
if len(recientes) >= 5:
return jsonify(ok=False, error="Ya recibimos varios reportes desde aquí. Te contactamos pronto."), 429
tipo = d.get("tipo") if d.get("tipo") in ("acceso", "error", "otro") else "acceso"
r = nueva({"descripcion": desc, "tipo": tipo, "nombre": nombre, "contacto": contacto,
"titulo": (NOMBRE_TIPO[tipo] + " — " + (nombre or contacto))[:140],
"pagina": "/entrar.html", "contexto": {"pantalla": texto(d.get("pantalla"), 40)}}, "", "entrada")
r["contexto"]["ip"] = ip
r["reportado_por"] = ""
guardar(r)
return jsonify(ok=True, id=r["id"])
@app.get("/api/incidencias/publicas")
def incidencias_publicas():
hace = (datetime.datetime.now() - datetime.timedelta(days=14)).replace(microsecond=0).isoformat()
regs = [r for r in todas() if r.get("publica") and texto(r.get("resumen_publico"))
and (r.get("estado") in ABIERTAS or (r.get("resuelta_en") or "") >= hace)]
regs.sort(key=lambda r: r.get("modificado_en") or "", reverse=True)
return jsonify(ok=True, avisos=[{"id": r["id"], "titulo": r.get("resumen_publico"),
"estado": r.get("estado"),
"resuelta": r.get("estado") in ("resuelta", "descartada"),
"actualizado": (r.get("modificado_en") or r.get("creado_en") or "")[:10]}
for r in regs[:8]])
g["_incidencias"] = {"todas": todas, "nueva": nueva, "guardar": guardar, "leer_errores": leer_errores,
"ABIERTAS": ABIERTAS}
Reference in New Issue View Git Blame Copy Permalink