"""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))