Leopoldo Rodríguez
7cbf9556c3
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016X4CAujMuib2tDqt9JRg6X
264 lines
13 KiB
Python
264 lines
13 KiB
Python
"""Tablero de Instalaciones por sitio (22-oct-2026).
|
|
|
|
Lo engancha app.py al final con registrar(app, globals()); vive aparte para no crecer app.py.
|
|
|
|
GET /api/instalaciones[?sitio=eit] (solo lee)
|
|
Una ficha por unidad en cada sitio, armada con las actas de campo: envío → recepción →
|
|
instalación → prueba en sitio → en operación, más «con falla». Trae el vehículo (tipo,
|
|
número económico, voltaje), el nombre en el monitor (H00xx), fechas, técnico y sus actas.
|
|
Las series PRUEBA no cuentan. ?sitio= acepta el id (icave, timsa, eit, lzc) o un alias (lct).
|
|
|
|
Acta nueva «prueba-sitio» (formularios/prueba-sitio.html): la prueba después de instalar.
|
|
registrar() la vuelve repetible (como las demás de campo) y la suma a los tipos que lista
|
|
la página de cada sitio, sin tocar ACTAS_REPETIBLES en actas-hbox.service.
|
|
|
|
Reglas de las etapas (la última acta manda; se ordenan por la fecha que anotó el técnico):
|
|
- Enviada: acta de envío «Enviada» a ese sitio. Un envío nuevo empieza otro ciclo.
|
|
- Recibida en sitio: recepción «Lista para entrar» o «Entra con observaciones».
|
|
- Instalada: acta de instalación que no dice «No se pudo instalar». Una instalación nueva pide
|
|
otra prueba.
|
|
- Probada en sitio: SOLO con el acta «Prueba en sitio» aprobada (resultado OK).
|
|
- En operación: probada hace DIAS_OPERACION días o más, sin falla registrada después.
|
|
- Con falla (aparte de la columna): instalación sin reportar o FALLA, prueba no aprobada,
|
|
diagnóstico con falla, unidad desmontada o reemplazada, recepción rechazada, o marca de daño.
|
|
Cada serie se cuenta en un solo sitio: el de su acta más reciente.
|
|
"""
|
|
import datetime
|
|
import os
|
|
|
|
from flask import jsonify, request
|
|
|
|
TIPO_PRUEBA = "prueba-sitio"
|
|
TIPOS = ("envio", "recepcion", "instalacion-campo", TIPO_PRUEBA, "diagnostico-campo", "dano")
|
|
DIAS_OPERACION = 7
|
|
ORDEN = ("icave", "timsa", "eit", "lzc") # el orden de las pestañas (lo pidió Leo)
|
|
ETAPAS = ("enviada", "recibida", "instalada", "probada", "operacion")
|
|
|
|
|
|
def _resp(d, clave):
|
|
"""Respuesta de un campo del acta (d['respuestas'][clave]['val']) como texto."""
|
|
r = d.get("respuestas")
|
|
v = r.get(clave) if isinstance(r, dict) else None
|
|
v = v.get("val") if isinstance(v, dict) else None
|
|
return str(v).strip() if isinstance(v, str) else ""
|
|
|
|
|
|
def _dia(d):
|
|
f = str(d.get("fecha") or "")
|
|
if len(f) == 10 and f[4] == "-" and f[7] == "-":
|
|
return f
|
|
return str(d.get("_recibido") or "")[:10]
|
|
|
|
|
|
def _evento(tipo, fn, d, sitio, modo):
|
|
ev = {"tipo": tipo, "archivo": fn, "serie": d.get("serie"), "sitio": sitio, "coincidencia": modo,
|
|
"fecha": _dia(d), "recibido": str(d.get("_recibido") or ""), "tecnico": str(d.get("tecnico") or "")[:80],
|
|
"resultado": str(d.get("resultado") or ""), "accion": str(d.get("accion") or "")[:160],
|
|
"motivo": str(d.get("motivo") or "")[:160], "clasificacion": str(d.get("clasificacion") or "")[:120],
|
|
"maquina": str(d.get("maquina_id") or d.get("maquina") or "").strip()[:60],
|
|
"id_monitor": str(d.get("id_monitor") or _resp(d, "id_monitor")).strip().upper()[:20],
|
|
"fotos": len(d.get("fotos") or [])}
|
|
if tipo in ("instalacion-campo", TIPO_PRUEBA):
|
|
ev["tipo_vehiculo"] = (str(d.get("clasificacion") or "") or _resp(d, "tipo_vehiculo"))[:60]
|
|
ev["voltaje"] = (str(d.get("sistema_v") or "") or _resp(d, "sistema_v"))[:20]
|
|
ev["placas"] = _resp(d, "placas")[:40]
|
|
if tipo == "diagnostico-campo":
|
|
ev["repuesto"] = _resp(d, "unidad_repuesto")[:40]
|
|
if tipo == TIPO_PRUEBA:
|
|
ev["gga"] = _resp(d, "gga")[:40]
|
|
return ev
|
|
|
|
|
|
def _falla(ev, motivo):
|
|
return {"motivo": motivo, "tipo": ev["tipo"], "fecha": ev["fecha"], "archivo": ev["archivo"]}
|
|
|
|
|
|
def _ficha(serie, eventos, marca, hoy):
|
|
"""Estado de una serie en su sitio a partir de sus actas ahí (viejas → nuevas)."""
|
|
etapa, falla, instalada, retirada, sin_inst = None, None, False, False, False
|
|
fechas = {}
|
|
prueba = None
|
|
destino = ""
|
|
inicio = 0 # un envío nuevo empieza otro ciclo
|
|
for i, ev in enumerate(eventos):
|
|
t, acc, res = ev["tipo"], ev["accion"], ev["resultado"]
|
|
if t == "envio":
|
|
if acc.startswith("Lista pero detenida"):
|
|
continue
|
|
etapa, falla, instalada, retirada, sin_inst, prueba = "enviada", None, False, False, False, None
|
|
fechas, inicio = {"enviada": ev["fecha"]}, i
|
|
destino = ev["maquina"]
|
|
elif t == "recepcion":
|
|
destino = ev["maquina"] or destino
|
|
mala = acc.startswith(("No entra", "Rechazada")) or res == "FALLA"
|
|
if etapa in (None, "enviada", "recibida"):
|
|
etapa = "recibida"
|
|
fechas["recibida"] = ev["fecha"]
|
|
falla = _falla(ev, acc or "La recepción no pasó la inspección") if mala else None
|
|
elif mala: # ya estaba instalada: una recepción mala solo marca la falla
|
|
falla = _falla(ev, acc or "La recepción no pasó la inspección")
|
|
elif t == "instalacion-campo":
|
|
if acc.startswith("No se pudo instalar"):
|
|
falla = _falla(ev, "No se pudo instalar")
|
|
continue
|
|
etapa, instalada, retirada, sin_inst, prueba = "instalada", True, False, False, None
|
|
fechas["instalada"] = ev["fecha"]
|
|
fechas.pop("probada", None)
|
|
if acc.startswith("Instalada, sin reportar"):
|
|
falla = _falla(ev, "Instalada, sin reportar todavía")
|
|
elif res == "FALLA":
|
|
falla = _falla(ev, "Instalación con falla")
|
|
else:
|
|
falla = None
|
|
elif t == TIPO_PRUEBA:
|
|
if etapa not in ("instalada", "probada"):
|
|
sin_inst, instalada, retirada = True, True, False
|
|
fechas.setdefault("instalada", "")
|
|
if res == "OK" and not acc.startswith("No aprobada"):
|
|
etapa, falla, prueba = "probada", None, ev
|
|
fechas["probada"] = ev["fecha"]
|
|
else:
|
|
etapa = etapa if etapa == "probada" else "instalada"
|
|
falla = _falla(ev, "Prueba en sitio no aprobada")
|
|
elif t == "diagnostico-campo":
|
|
if acc.startswith("Desmontada"):
|
|
falla, instalada, retirada = _falla(ev, acc), False, True
|
|
elif acc.startswith("Reemplazada"):
|
|
falla = _falla(ev, "Reemplazada por otra unidad" + (f" ({ev.get('repuesto')})" if ev.get("repuesto") else ""))
|
|
instalada, retirada = False, True
|
|
elif res == "FALLA" or acc.startswith("Instalada, pero sin resolver"):
|
|
falla = _falla(ev, ev["motivo"] or acc or "Diagnóstico con falla")
|
|
if etapa is None:
|
|
etapa, instalada, sin_inst = "instalada", True, True
|
|
elif res == "OK":
|
|
falla = None
|
|
if etapa is None:
|
|
etapa, instalada, sin_inst = "instalada", True, True
|
|
fechas.setdefault("instalada", "")
|
|
if marca and marca.get("marca"):
|
|
a = marca.get("acta") or {}
|
|
m = "Dada de baja" if marca["marca"] == "BAJA" else "Marcada como dañada"
|
|
if a.get("clasificacion"):
|
|
m += ": " + str(a["clasificacion"])[:80]
|
|
falla = {"motivo": m, "tipo": "dano", "fecha": a.get("fecha") or "", "archivo": a.get("archivo") or ""}
|
|
if etapa is None and not falla:
|
|
return None
|
|
operacion_desde = None
|
|
if prueba and prueba.get("fecha"):
|
|
try:
|
|
operacion_desde = (datetime.date.fromisoformat(prueba["fecha"])
|
|
+ datetime.timedelta(days=DIAS_OPERACION)).isoformat()
|
|
except ValueError:
|
|
operacion_desde = None
|
|
if etapa == "probada" and not falla and operacion_desde and operacion_desde <= hoy:
|
|
etapa = "operacion"
|
|
# vehículo: el último que anotó una instalación, una prueba o un diagnóstico (de este ciclo)
|
|
veh = {"numero": "", "tipo": "", "voltaje": "", "placas": "", "acta": ""}
|
|
monitor = ""
|
|
for ev in eventos[inicio:]:
|
|
if ev["id_monitor"]:
|
|
monitor = ev["id_monitor"]
|
|
if ev["tipo"] in ("instalacion-campo", TIPO_PRUEBA, "diagnostico-campo") and ev["maquina"]:
|
|
veh["numero"], veh["acta"] = ev["maquina"], ev["tipo"]
|
|
for k in ("tipo_vehiculo", "voltaje", "placas"):
|
|
if ev.get(k):
|
|
veh["tipo" if k == "tipo_vehiculo" else k] = ev[k]
|
|
ultima = eventos[-1] if eventos else None
|
|
return {
|
|
"serie": serie,
|
|
"etapa": etapa,
|
|
"columna": "falla" if falla else etapa,
|
|
"falla": falla,
|
|
"instalada": bool(instalada and not retirada),
|
|
"retirada": retirada,
|
|
"sin_acta_instalacion": sin_inst,
|
|
"probada": etapa in ("probada", "operacion") and not falla,
|
|
"vehiculo": veh if (veh["numero"] or veh["tipo"]) else None,
|
|
"vehiculo_destino": destino or None,
|
|
"id_monitor": monitor or None,
|
|
"fechas": fechas,
|
|
"operacion_desde": operacion_desde,
|
|
"marca": (marca or {}).get("marca"),
|
|
"tecnico": ultima["tecnico"] if ultima else "",
|
|
"ultima": {k: ultima[k] for k in ("tipo", "fecha", "accion", "resultado")} if ultima else None,
|
|
"actas": [{k: ev[k] for k in ("tipo", "archivo", "fecha", "tecnico", "resultado", "accion")}
|
|
for ev in reversed(eventos)],
|
|
}
|
|
|
|
|
|
def registrar(app, g):
|
|
"""g = globals() de app.py: se usan sus funciones (lectura de actas y sitios) sin copiarlas."""
|
|
g["TIPOS_REPETIBLES"].add(TIPO_PRUEBA)
|
|
if TIPO_PRUEBA not in g["TIPOS_SITIO"]:
|
|
g["TIPOS_SITIO"] = tuple(g["TIPOS_SITIO"]) + (TIPO_PRUEBA,)
|
|
|
|
def _orden(s):
|
|
sid = s.get("id", "")
|
|
return (ORDEN.index(sid) if sid in ORDEN else len(ORDEN), s.get("orden", 50), s.get("nombre", ""))
|
|
|
|
@app.get("/api/instalaciones")
|
|
def instalaciones():
|
|
slug, leer = g["slug"], g["_leer_json"]
|
|
sitios = [s for s in g["_listar_json"](g["SITIOS_DIR"]) if s.get("activo", True) and s.get("id")]
|
|
sitios.sort(key=_orden)
|
|
ids = [s["id"] for s in sitios]
|
|
pedido = slug(request.args.get("sitio"), 30).lower()
|
|
pedido = g.get("ALIAS_SITIO", {}).get(pedido, pedido)
|
|
if pedido and pedido not in ids:
|
|
return jsonify(ok=False, error=f"No existe el sitio {pedido}."), 404
|
|
|
|
en_sitio = g["_acta_en_sitio"]
|
|
por_serie = {}
|
|
for tipo in TIPOS:
|
|
carpeta = os.path.join(g["BASE"], tipo)
|
|
if not os.path.isdir(carpeta):
|
|
continue
|
|
for fn in sorted(os.listdir(carpeta)):
|
|
if not fn.endswith(".json"):
|
|
continue
|
|
d = leer(os.path.join(carpeta, fn))
|
|
if not isinstance(d, dict) or d.get("_prueba"):
|
|
continue
|
|
serie = str(d.get("serie") or "")
|
|
if not serie or "PRUEBA" in serie.upper():
|
|
continue
|
|
donde, modo = None, None
|
|
for s in sitios: # primero el sitio elegido en el acta; luego por nombre
|
|
m = en_sitio(d, s)
|
|
if m == "sitio":
|
|
donde, modo = s["id"], m
|
|
break
|
|
if m and not donde:
|
|
donde, modo = s["id"], m
|
|
if donde:
|
|
por_serie.setdefault(serie, []).append(_evento(tipo, fn, d, donde, modo))
|
|
|
|
hoy = datetime.date.today().isoformat()
|
|
unidades = []
|
|
for serie, evs in por_serie.items():
|
|
evs.sort(key=lambda e: (e["fecha"], e["recibido"], e["archivo"])) # el archivo lleva microsegundos
|
|
sitio = evs[-1]["sitio"] # cuenta en el sitio de su acta más reciente
|
|
u = _ficha(serie, [e for e in evs if e["sitio"] == sitio], g["estado_serie"](serie), hoy)
|
|
if u:
|
|
u["sitio"] = sitio
|
|
u["coincidencia"] = "sitio" if any(e["coincidencia"] == "sitio" for e in evs if e["sitio"] == sitio) else "texto"
|
|
unidades.append(u)
|
|
|
|
resumen = {}
|
|
for sid in ids:
|
|
us = [u for u in unidades if u["sitio"] == sid]
|
|
r = {c: sum(1 for u in us if u["columna"] == c) for c in ETAPAS + ("falla",)}
|
|
r.update(unidades=len(us), instaladas=sum(1 for u in us if u["instalada"]),
|
|
probadas=sum(1 for u in us if u["probada"]),
|
|
sin_vehiculo=sum(1 for u in us if u["instalada"] and not (u["vehiculo"] or {}).get("numero")),
|
|
sin_prueba=sum(1 for u in us if u["instalada"] and u["etapa"] == "instalada" and not u["falla"]))
|
|
resumen[sid] = r
|
|
|
|
orden_col = {c: i for i, c in enumerate(("falla",) + ETAPAS)}
|
|
unidades.sort(key=lambda u: (ids.index(u["sitio"]), orden_col.get(u["columna"], 9), u["id_monitor"] or u["serie"]))
|
|
if pedido:
|
|
unidades = [u for u in unidades if u["sitio"] == pedido]
|
|
return jsonify(ok=True, generado=g["ahora"](), hoy=hoy, sitio=pedido or None, dias_operacion=DIAS_OPERACION,
|
|
sitios=[{k: s.get(k) for k in ("id", "nombre", "cliente", "ubicacion", "alias")} for s in sitios],
|
|
resumen=resumen, unidades=unidades, total=len(unidades))
|