#!/usr/bin/env python3 """Actas HBOX — servicio para recibir y consultar formularios firmados de los técnicos. Reglas: - Un acta real (serie que NO contiene PRUEBA) se puede enviar UNA sola vez por serie. Un segundo envío se rechaza (409) hasta que un administrador la borre. - Al seleccionar una serie ya enviada, el formulario la vuelve a cargar (llena, bloqueada). - Solo el usuario 'leo' (administrador) puede borrar/reabrir un acta. - Las series PRUEBA se guardan aparte (_pruebas), no cuentan y se pueden reenviar. El usuario autenticado llega en la cabecera X-Remote-User (la fija nginx desde $remote_user; el cliente no puede falsificarla porque nginx la sobrescribe). Rutas: GET /api/salud GET /api/whoami -> {user, admin} GET /api/acta?tipo=&serie= -> {exists, acta?} POST /api/actas -> crea (409 si ya existe una real) DELETE /api/acta?tipo=&serie= -> borra (solo admin) GET /api/actas -> listado (ingeniería); incluye 'componentes' en actas compuestas GET /api/historial?tipo=&serie= -> todas las actas de esa serie (tipos repetibles: diagnóstico de campo) POST /api/adjuntos -> sube fotos (multipart) y devuelve sus rutas GET /api/adjunto?ruta= -> descarga una foto guardada DELETE /api/acta?...&archivo= -> borra UNA acta concreta (solo admin; tipos repetibles) Actas compuestas (integracion, ensamble): traen d["componentes"] = {clave: serie_modulo}. En los tipos de TIPOS_UNICOS un módulo no puede aparecer en dos actas reales (409). """ import os, re, json, csv, datetime, uuid, hashlib from flask import Flask, request, jsonify, abort, send_file app = Flask(__name__) BASE = os.environ.get("ACTAS_DIR", "/srv/projects/drogheda/actas") ADJ = os.environ.get("ADJUNTOS_DIR", os.path.join(os.path.dirname(BASE.rstrip("/")) or "/srv/projects/drogheda", "adjuntos")) ADMINS = {u.strip().lower() for u in os.environ.get("ACTAS_ADMINS", "leo").split(",") if u.strip()} TIPOS_UNICOS = {"ensamble"} # tipos en los que un componente solo puede usarse en un acta real # Tipos REPETIBLES: una misma serie puede tener varias actas (cada visita a campo es un evento nuevo). TIPOS_REPETIBLES = {t.strip() for t in os.environ.get( "ACTAS_REPETIBLES", "diagnostico-campo,diagnostico-banco,instalacion-campo,reflasheo").split(",") if t.strip()} # Subida de fotos MAX_MB = int(os.environ.get("ADJUNTO_MAX_MB", "8")) # Archivos de log (capturas por USB del Blackpill, del 4G y del ArduSimple). LOG_MAX_MB = int(os.environ.get("ARCHIVO_MAX_MB", "20")) app.config["MAX_CONTENT_LENGTH"] = (max(MAX_MB, LOG_MAX_MB) + 2) * 1024 * 1024 EXT_OK = {"image/jpeg": ".jpg", "image/png": ".png", "image/webp": ".webp"} # Lista blanca de logs: texto plano y capturas binarias de u-blox. Nada ejecutable. EXT_LOG = {".txt", ".log", ".csv", ".nmea", ".ubx", ".json", ".cfg", ".md"} SAFE = re.compile(r"[^A-Za-z0-9_\-]") SAFE_FILE = re.compile(r"[^A-Za-z0-9_\-.]") def slug(s, n=40): return SAFE.sub("", str(s or "").strip())[:n] def current_user(): return (request.headers.get("X-Remote-User") or "").strip() def is_admin(): return current_user().lower() in ADMINS def componentes_en_uso(tipo, serie_actual, componentes): """Devuelve [(clave, valor, serie_otra)] para componentes ya usados por OTRA acta real del mismo tipo.""" carpeta = os.path.join(BASE, tipo) if not os.path.isdir(carpeta) or not isinstance(componentes, dict): return [] usados = [] for fn in os.listdir(carpeta): if not fn.endswith(".json"): continue try: with open(os.path.join(carpeta, fn), encoding="utf-8") as f: d = json.load(f) except Exception: continue if d.get("serie") == serie_actual or not isinstance(d.get("componentes"), dict): continue for k, v in componentes.items(): if v and d["componentes"].get(k) == v: usados.append((k, v, d.get("serie"))) return usados def es_repetible(tipo): return tipo in TIPOS_REPETIBLES def listar_serie(tipo, serie): """Todas las actas guardadas para (tipo, serie), de la más vieja a la más nueva.""" carpeta = os.path.join(BASE, tipo) if not os.path.isdir(carpeta): return [] pref = serie + "_" return sorted(fn for fn in os.listdir(carpeta) if fn.startswith(pref) and fn.endswith(".json")) def find_latest(tipo, serie): """Ruta del acta real más reciente para (tipo, serie), o None.""" carpeta = os.path.join(BASE, tipo) if not os.path.isdir(carpeta): return None pref = serie + "_" cand = sorted(fn for fn in os.listdir(carpeta) if fn.startswith(pref) and fn.endswith(".json")) return os.path.join(carpeta, cand[-1]) if cand else None @app.get("/api/salud") def salud(): return jsonify(ok=True, servicio="actas-hbox") @app.get("/api/whoami") def whoami(): u = current_user() return jsonify(ok=True, user=u, admin=(u.lower() in ADMINS)) @app.get("/api/acta") def get_acta(): tipo = slug(request.args.get("tipo")); serie = slug(request.args.get("serie"), 24) if not tipo or not serie: return jsonify(ok=False, error="faltan tipo/serie"), 400 archivo = SAFE_FILE.sub("", str(request.args.get("archivo") or ""))[:120] if archivo: if archivo not in listar_serie(tipo, serie): return jsonify(ok=True, exists=False), 404 with open(os.path.join(BASE, tipo, archivo), encoding="utf-8") as f: d = json.load(f) d["_archivo"] = archivo return jsonify(ok=True, exists=True, acta=d) ruta = find_latest(tipo, serie) if not ruta: return jsonify(ok=True, exists=False, repetible=es_repetible(tipo)) with open(ruta, encoding="utf-8") as f: d = json.load(f) d["_archivo"] = os.path.basename(ruta) # En un tipo repetible el acta previa NO bloquea: se devuelve como antecedente. if es_repetible(tipo): return jsonify(ok=True, exists=False, repetible=True, previa=d, visitas=len(listar_serie(tipo, serie))) return jsonify(ok=True, exists=True, acta=d) @app.post("/api/actas") def crear(): try: d = request.get_json(force=True) except Exception: return jsonify(ok=False, error="JSON inválido"), 400 if not isinstance(d, dict): return jsonify(ok=False, error="Se esperaba un objeto"), 400 tipo = slug(d.get("tipo")); serie = slug(d.get("serie"), 24) tecnico = str(d.get("tecnico", "")).strip() if not tipo or not serie: return jsonify(ok=False, error="Faltan 'tipo' o 'serie'"), 400 if not tecnico: return jsonify(ok=False, error="Falta el nombre del técnico"), 400 if not d.get("conforme"): return jsonify(ok=False, error="Falta la conformidad del técnico"), 400 es_prueba = "PRUEBA" in serie.upper() # un acta real solo se puede enviar una vez (salvo tipos repetibles: cada visita es un evento) if not es_prueba and not es_repetible(tipo) and find_latest(tipo, serie): return jsonify(ok=False, existe=True, error=f"Ya existe un acta para {serie}. Pide a Leo que la borre para rehacerla."), 409 # actas compuestas (ensamble): un módulo no puede pertenecer a dos unidades comps = d.get("componentes") if isinstance(comps, dict): comps = {slug(k, 20): slug(v, 20) for k, v in comps.items()} d["componentes"] = comps if not es_prueba and tipo in TIPOS_UNICOS: usados = componentes_en_uso(tipo, serie, comps) if usados: det = "; ".join(f"{v} ya está en {s}" for _, v, s in usados) return jsonify(ok=False, error=f"Módulo ya asignado: {det}. Pide a Leo que revise."), 409 fotos = d.get("fotos") if isinstance(fotos, list): d["fotos"] = [f for f in fotos[:60] if isinstance(f, dict) and isinstance(f.get("ruta"), str) and os.path.isfile(os.path.join(ADJ, f["ruta"]))] archivos = d.get("archivos") if isinstance(archivos, list): d["archivos"] = [a for a in archivos[:40] if isinstance(a, dict) and isinstance(a.get("ruta"), str) and os.path.isfile(os.path.join(ADJ, a["ruta"]))] ts = datetime.datetime.now().replace(microsecond=0) iso = ts.isoformat() d["_recibido"] = iso d["_usuario"] = current_user() d["_ip"] = request.headers.get("X-Forwarded-For", request.remote_addr) d["_prueba"] = es_prueba carpeta = os.path.join(BASE, tipo, "_pruebas") if es_prueba else os.path.join(BASE, tipo) os.makedirs(carpeta, exist_ok=True) archivo = f"{serie}_{iso.replace(':', '-')}.json" if es_repetible(tipo): # cada visita es un registro propio: sufijo único para que dos envíos # en el mismo segundo (doble toque) no se pisen archivo = f"{serie}_{iso.replace(':', '-')}_{uuid.uuid4().hex[:6]}.json" with open(os.path.join(carpeta, archivo), "w", encoding="utf-8") as f: json.dump(d, f, ensure_ascii=False, indent=2) if not es_prueba: idx = os.path.join(carpeta, "_indice.csv") nuevo = not os.path.exists(idx) with open(idx, "a", newline="", encoding="utf-8") as f: w = csv.writer(f) if nuevo: w.writerow(["recibido", "fecha", "serie", "maquina", "tecnico", "resultado", "clasificacion", "accion", "fotos", "logs", "archivo"]) w.writerow([iso, d.get("fecha", ""), serie, d.get("maquina_id", d.get("maquina", "")), tecnico, d.get("resultado", ""), d.get("clasificacion", ""), d.get("accion", ""), len(d.get("fotos") or []), len(d.get("archivos") or []), archivo]) sub = f"{tipo}/_pruebas/{archivo}" if es_prueba else f"{tipo}/{archivo}" return jsonify(ok=True, archivo=sub, prueba=es_prueba) @app.delete("/api/acta") def borrar_acta(): if not is_admin(): return jsonify(ok=False, error="Solo un administrador puede borrar actas."), 403 tipo = slug(request.args.get("tipo")); serie = slug(request.args.get("serie"), 24) archivo = SAFE_FILE.sub("", str(request.args.get("archivo") or ""))[:120] if not tipo or not serie: return jsonify(ok=False, error="faltan tipo/serie"), 400 carpeta = os.path.join(BASE, tipo) borrados = [] # borrar UNA visita concreta (tipos repetibles), no toda la serie if archivo: existentes = listar_serie(tipo, serie) if archivo in existentes: os.remove(os.path.join(carpeta, archivo)); borrados.append(archivo) return jsonify(ok=True, borrados=borrados) if os.path.isdir(carpeta): pref = serie + "_" for fn in list(os.listdir(carpeta)): if fn.startswith(pref) and fn.endswith(".json"): os.remove(os.path.join(carpeta, fn)); borrados.append(fn) idx = os.path.join(carpeta, "_indice.csv") if os.path.exists(idx): rows = list(csv.reader(open(idx, encoding="utf-8"))) keep = ([rows[0]] if rows else []) + [r for r in rows[1:] if not (len(r) > 1 and r[1] == serie)] with open(idx, "w", newline="", encoding="utf-8") as f: csv.writer(f).writerows(keep) return jsonify(ok=True, borrados=borrados) @app.delete("/api/actas") def borrar_todas(): """Reset admin: borra TODAS las actas reales de un tipo (opcional: también las de prueba).""" if not is_admin(): return jsonify(ok=False, error="Solo un administrador puede reiniciar actas."), 403 tipo = slug(request.args.get("tipo")) incl_pruebas = request.args.get("pruebas") == "1" if not tipo: return jsonify(ok=False, error="falta tipo"), 400 carpeta = os.path.join(BASE, tipo) n = 0 if os.path.isdir(carpeta): for fn in list(os.listdir(carpeta)): fp = os.path.join(carpeta, fn) if fn.endswith(".json"): os.remove(fp); n += 1 elif fn == "_indice.csv": os.remove(fp) if incl_pruebas: pdir = os.path.join(carpeta, "_pruebas") if os.path.isdir(pdir): for fn in list(os.listdir(pdir)): if fn.endswith(".json"): os.remove(os.path.join(pdir, fn)); n += 1 return jsonify(ok=True, borradas=n) @app.get("/api/actas") def listar(): filas = [] if os.path.isdir(BASE): for tipo in sorted(os.listdir(BASE)): carpeta = os.path.join(BASE, tipo) if not os.path.isdir(carpeta): continue for fn in sorted(os.listdir(carpeta)): if fn.endswith(".json"): try: with open(os.path.join(carpeta, fn), encoding="utf-8") as f: d = json.load(f) except Exception: continue fila = {"tipo": tipo, "archivo": f"{tipo}/{fn}", "serie": d.get("serie"), "tecnico": d.get("tecnico"), "resultado": d.get("resultado"), "recibido": d.get("_recibido")} if isinstance(d.get("componentes"), dict): fila["componentes"] = d["componentes"] for k in ("clasificacion", "accion", "motivo", "maquina", "terminal", "unidad"): if d.get(k): fila[k] = d[k] if d.get("archivos"): fila["archivos"] = [{"ruta": a.get("ruta"), "nombre": a.get("nombre")} for a in d["archivos"] if a.get("ruta")] if d.get("fotos"): fila["fotos"] = [f.get("ruta") for f in d["fotos"] if f.get("ruta")] fila["archivo_n"] = fn filas.append(fila) filas.sort(key=lambda r: (r.get("recibido") or ""), reverse=True) return jsonify(ok=True, total=len(filas), actas=filas) # ------------------------------------------------------------------ historial @app.get("/api/historial") def historial(): """Todas las actas de una serie, de la más reciente a la más vieja (visitas a campo).""" tipo = slug(request.args.get("tipo")); serie = slug(request.args.get("serie"), 24) if not tipo or not serie: return jsonify(ok=False, error="faltan tipo/serie"), 400 carpeta = os.path.join(BASE, tipo) out = [] for fn in listar_serie(tipo, serie): try: with open(os.path.join(carpeta, fn), encoding="utf-8") as f: d = json.load(f) except Exception: continue out.append({"archivo": fn, "recibido": d.get("_recibido"), "tecnico": d.get("tecnico"), "resultado": d.get("resultado"), "clasificacion": d.get("clasificacion"), "accion": d.get("accion"), "motivo": d.get("motivo"), "fotos": len(d.get("fotos") or []), "archivos": len(d.get("archivos") or []), "fecha": d.get("fecha")}) out.reverse() return jsonify(ok=True, total=len(out), visitas=out) # ------------------------------------------------------------------ adjuntos (fotos) def _tipo_imagen(cab): if cab[:3] == b"\xff\xd8\xff": return ".jpg" if cab[:8] == b"\x89PNG\r\n\x1a\n": return ".png" if cab[:4] == b"RIFF" and cab[8:12] == b"WEBP": return ".webp" return None @app.post("/api/adjuntos") def subir_adjuntos(): """Sube fotos de campo. multipart/form-data: tipo, serie, foto (1..10), etiqueta opcional.""" tipo = slug(request.form.get("tipo")); serie = slug(request.form.get("serie"), 24) etiqueta = str(request.form.get("etiqueta", ""))[:80] if not tipo or not serie: return jsonify(ok=False, error="faltan tipo/serie"), 400 archivos = request.files.getlist("foto") if not archivos: return jsonify(ok=False, error="no llegó ninguna foto"), 400 if len(archivos) > 10: return jsonify(ok=False, error="máximo 10 fotos por envío"), 400 carpeta = os.path.join(ADJ, tipo, serie) os.makedirs(carpeta, exist_ok=True) guardadas, rechazadas = [], [] for fh in archivos: datos = fh.read(MAX_MB * 1024 * 1024 + 1) if len(datos) > MAX_MB * 1024 * 1024: rechazadas.append({"nombre": fh.filename, "error": f"pesa más de {MAX_MB} MB"}); continue ext = _tipo_imagen(datos[:16]) if not ext: rechazadas.append({"nombre": fh.filename, "error": "no es una imagen JPG/PNG/WEBP"}); continue nom = datetime.datetime.now().strftime("%Y%m%d-%H%M%S") + "-" + uuid.uuid4().hex[:6] + ext with open(os.path.join(carpeta, nom), "wb") as f: f.write(datos) guardadas.append({"ruta": f"{tipo}/{serie}/{nom}", "bytes": len(datos), "etiqueta": etiqueta, "usuario": current_user(), "subida": datetime.datetime.now().replace(microsecond=0).isoformat()}) if not guardadas: return jsonify(ok=False, error=(rechazadas[0]["error"] if rechazadas else "no se guardó nada"), rechazadas=rechazadas), 400 return jsonify(ok=True, fotos=guardadas, rechazadas=rechazadas) @app.post("/api/archivos") def subir_archivos(): """Sube archivos de log de campo/banco. multipart: tipo, serie, archivo (1..10), etiqueta.""" tipo = slug(request.form.get("tipo")); serie = slug(request.form.get("serie"), 24) etiqueta = str(request.form.get("etiqueta", ""))[:80] if not tipo or not serie: return jsonify(ok=False, error="faltan tipo/serie"), 400 subidos = request.files.getlist("archivo") if not subidos: return jsonify(ok=False, error="no llegó ningún archivo"), 400 if len(subidos) > 10: return jsonify(ok=False, error="máximo 10 archivos por envío"), 400 carpeta = os.path.join(ADJ, tipo, serie, "logs") os.makedirs(carpeta, exist_ok=True) guardados, rechazados = [], [] for fh in subidos: nombre = SAFE_FILE.sub("_", os.path.basename(fh.filename or "captura"))[:80] ext = os.path.splitext(nombre)[1].lower() if ext not in EXT_LOG: rechazados.append({"nombre": nombre, "error": "extensión no permitida (usa .txt .log .csv .nmea .ubx .json)"}) continue datos = fh.read(LOG_MAX_MB * 1024 * 1024 + 1) if len(datos) > LOG_MAX_MB * 1024 * 1024: rechazados.append({"nombre": nombre, "error": f"pesa más de {LOG_MAX_MB} MB"}); continue if not datos: rechazados.append({"nombre": nombre, "error": "el archivo está vacío"}); continue final = datetime.datetime.now().strftime("%Y%m%d-%H%M%S") + "-" + uuid.uuid4().hex[:4] + "_" + nombre with open(os.path.join(carpeta, final), "wb") as f: f.write(datos) guardados.append({"ruta": f"{tipo}/{serie}/logs/{final}", "nombre": nombre, "bytes": len(datos), "etiqueta": etiqueta, "usuario": current_user(), "subida": datetime.datetime.now().replace(microsecond=0).isoformat()}) if not guardados: return jsonify(ok=False, error=(rechazados[0]["error"] if rechazados else "no se guardó nada"), rechazados=rechazados), 400 return jsonify(ok=True, archivos=guardados, rechazados=rechazados) @app.get("/api/adjunto") def ver_adjunto(): ruta = (request.args.get("ruta") or "").strip().lstrip("/") if not ruta or ".." in ruta: return jsonify(ok=False, error="ruta inválida"), 400 fp = os.path.realpath(os.path.join(ADJ, ruta)) if not fp.startswith(os.path.realpath(ADJ) + os.sep) or not os.path.isfile(fp): return jsonify(ok=False, error="no existe"), 404 return send_file(fp) @app.errorhandler(413) def _muy_grande(e): return jsonify(ok=False, error=f"El archivo pesa demasiado (fotos: máximo {MAX_MB} MB; " f"logs: máximo {LOG_MAX_MB} MB)."), 413 # ------------------------------------------------- contraseñas (HTTP Basic de nginx) # El archivo de usuarios de nginx se escribe aquí mismo, sin invocar `htpasswd`: # pasar una contraseña como argumento de un proceso la deja visible en `ps`. HTPASSWD = os.environ.get("HTPASSWD_FILE", "/etc/nginx/hbox-docs.htpasswd") PASS_MIN = int(os.environ.get("PASS_MIN", "8")) ITOA64 = "./0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz" def _to64(v, n): return "".join(ITOA64[(v >> (6 * i)) & 0x3F] for i in range(n)) def apr1(password, salt): """Hash Apache MD5 ($apr1$): lo que escribe htpasswd por defecto y lo que lee nginx.""" pw = password.encode("utf-8") sl = salt.encode("ascii") ctx = hashlib.md5(pw + b"$apr1$" + sl) final = hashlib.md5(pw + sl + pw).digest() i = len(pw) while i > 0: ctx.update(final[:16] if i > 16 else final[:i]); i -= 16 i = len(pw) while i: ctx.update(b"\0" if (i & 1) else pw[:1]); i >>= 1 final = ctx.digest() for i in range(1000): c = hashlib.md5() c.update(pw if (i & 1) else final) if i % 3: c.update(sl) if i % 7: c.update(pw) c.update(final if (i & 1) else pw) final = c.digest() out = "".join(_to64((final[a] << 16) | (final[b] << 8) | final[c], 4) for a, b, c in ((0, 6, 12), (1, 7, 13), (2, 8, 14), (3, 9, 15), (4, 10, 5))) return "$apr1$" + salt + "$" + out + _to64(final[11], 2) def _leer_htpasswd(): if not os.path.isfile(HTPASSWD): return None with open(HTPASSWD, encoding="utf-8") as f: return f.read().splitlines() def _escribir_htpasswd(lineas): """Reescribe el archivo de forma atómica, conservando permisos y dueño.""" carpeta = os.path.dirname(HTPASSWD) or "." st = os.stat(HTPASSWD) tmp = os.path.join(carpeta, ".htpasswd.tmp-" + uuid.uuid4().hex[:8]) with open(tmp, "w", encoding="utf-8") as f: f.write("\n".join(lineas) + "\n") os.chmod(tmp, st.st_mode & 0o7777) try: os.chown(tmp, st.st_uid, st.st_gid) except PermissionError: pass os.replace(tmp, HTPASSWD) @app.get("/api/usuarios") def usuarios(): """Lista de usuarios (solo administrador), para el restablecimiento.""" if not is_admin(): return jsonify(ok=False, error="Solo un administrador puede ver la lista."), 403 lineas = _leer_htpasswd() if lineas is None: return jsonify(ok=False, error="No encuentro el archivo de usuarios."), 500 return jsonify(ok=True, usuarios=sorted(l.split(":", 1)[0] for l in lineas if ":" in l)) @app.post("/api/password") def cambiar_password(): yo = current_user() if not yo: return jsonify(ok=False, error="No puedo identificarte. Vuelve a iniciar sesión."), 403 try: d = request.get_json(force=True) or {} except Exception: return jsonify(ok=False, error="JSON inválido"), 400 destino = str(d.get("usuario") or yo).strip() actual = str(d.get("actual") or "") nueva = str(d.get("nueva") or "") confirma = str(d.get("confirma") or "") es_reset = (destino.lower() != yo.lower()) if es_reset and not is_admin(): return jsonify(ok=False, error="Solo puedes cambiar tu propia contraseña."), 403 if len(nueva) < PASS_MIN: return jsonify(ok=False, error=f"La nueva contraseña debe tener al menos {PASS_MIN} caracteres."), 400 if nueva != confirma: return jsonify(ok=False, error="Las dos contraseñas nuevas no coinciden."), 400 if not es_reset and nueva == actual: return jsonify(ok=False, error="La nueva contraseña es igual a la actual."), 400 lineas = _leer_htpasswd() if lineas is None: return jsonify(ok=False, error="No encuentro el archivo de usuarios en el servidor."), 500 idx = next((i for i, l in enumerate(lineas) if ":" in l and l.split(":", 1)[0] == destino), None) if idx is None: return jsonify(ok=False, error=f"El usuario {destino} no existe."), 404 guardado = lineas[idx].split(":", 1)[1] if not es_reset: partes = guardado.split("$") if not guardado.startswith("$apr1$") or len(partes) < 4: return jsonify(ok=False, error="Tu contraseña está guardada en un formato que este " "servicio no puede verificar. Pide a Leo que la restablezca."), 400 if apr1(actual, partes[2]) != guardado: return jsonify(ok=False, error="La contraseña actual no es correcta."), 403 salt = "".join(ITOA64[b % 64] for b in os.urandom(8)) lineas[idx] = destino + ":" + apr1(nueva, salt) try: _escribir_htpasswd(lineas) except PermissionError: return jsonify(ok=False, error="El servicio no tiene permiso de escribir el archivo de " "usuarios. Avisa a Leo (permisos de hbox-docs.htpasswd)."), 500 except OSError as e: return jsonify(ok=False, error="No se pudo guardar: " + str(e)), 500 return jsonify(ok=True, usuario=destino, reset=es_reset) # ------------------------------------------------- expediente de una unidad def _listar_adjuntos(serie): """Todas las fotos y logs subidos para esa serie, de cualquier tipo de acta.""" fotos, logs = [], [] if not os.path.isdir(ADJ): return fotos, logs for tipo in sorted(os.listdir(ADJ)): base = os.path.join(ADJ, tipo, serie) if not os.path.isdir(base): continue for fn in sorted(os.listdir(base)): fp = os.path.join(base, fn) if os.path.isfile(fp): fotos.append({"tipo": tipo, "ruta": f"{tipo}/{serie}/{fn}", "bytes": os.path.getsize(fp), "ts": int(os.path.getmtime(fp))}) dlogs = os.path.join(base, "logs") if os.path.isdir(dlogs): for fn in sorted(os.listdir(dlogs)): fp = os.path.join(dlogs, fn) if os.path.isfile(fp): logs.append({"tipo": tipo, "ruta": f"{tipo}/{serie}/logs/{fn}", "nombre": fn[16:] if len(fn) > 16 else fn, "bytes": os.path.getsize(fp), "ts": int(os.path.getmtime(fp))}) return fotos, logs @app.get("/api/unidad") def expediente(): """Todo lo que existe en el servidor sobre una serie: actas de cualquier tipo, fotos, archivos de log, y las actas de otras unidades que la mencionan como componente.""" serie = slug(request.args.get("serie"), 24) if not serie: return jsonify(ok=False, error="falta serie"), 400 actas, relacionadas = [], [] if os.path.isdir(BASE): for tipo in sorted(os.listdir(BASE)): carpeta = os.path.join(BASE, tipo) if not os.path.isdir(carpeta): continue for fn in sorted(os.listdir(carpeta)): if not fn.endswith(".json"): continue try: with open(os.path.join(carpeta, fn), encoding="utf-8") as f: d = json.load(f) except Exception: continue fila = {"tipo": tipo, "archivo": fn, "serie": d.get("serie"), "recibido": d.get("_recibido"), "fecha": d.get("fecha"), "tecnico": d.get("tecnico"), "resultado": d.get("resultado"), "clasificacion": d.get("clasificacion"), "accion": d.get("accion"), "motivo": d.get("motivo"), "maquina": d.get("maquina_id") or d.get("maquina"), "fotos": [f2.get("ruta") for f2 in (d.get("fotos") or []) if f2.get("ruta")], "archivos": [{"ruta": a.get("ruta"), "nombre": a.get("nombre")} for a in (d.get("archivos") or []) if a.get("ruta")]} if d.get("serie") == serie: actas.append(fila) elif isinstance(d.get("componentes"), dict) and serie in d["componentes"].values(): fila["componentes"] = d["componentes"] relacionadas.append(fila) # Orden del expediente: manda la FECHA que anotó el técnico (la de la visita); # la hora de recepción solo desempata. Un acta capturada días después no debe # aparecer como si el trabajo se hubiera hecho ese día. def _clave(r): return ((r.get("fecha") or (r.get("recibido") or "")[:10]), r.get("recibido") or "") actas.sort(key=_clave, reverse=True) relacionadas.sort(key=_clave, reverse=True) fotos, logs = _listar_adjuntos(serie) return jsonify(ok=True, serie=serie, actas=actas, relacionadas=relacionadas, fotos=fotos, logs=logs, total=len(actas), total_fotos=len(fotos), total_logs=len(logs)) @app.get("/api/series") def series(): """Todas las series que tienen algo guardado — para el buscador del expediente.""" vistas = {} if os.path.isdir(BASE): for tipo in sorted(os.listdir(BASE)): carpeta = os.path.join(BASE, tipo) if not os.path.isdir(carpeta): continue for fn in os.listdir(carpeta): if fn.endswith(".json"): ser = fn.rsplit("_", 1)[0] if "_" in fn else fn[:-5] ser = fn.split("_2")[0] if "_2" in fn else ser vistas.setdefault(ser, set()).add(tipo) return jsonify(ok=True, series=[{"serie": k, "tipos": sorted(v)} for k, v in sorted(vistas.items())]) if __name__ == "__main__": app.run(host="127.0.0.1", port=int(os.environ.get("PORT", "8081")))