722 lines
32 KiB
Python
722 lines
32 KiB
Python
#!/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,dano,envio,recepcion,prueba-bateria-modem").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. El nombre lleva microsegundos para que el
|
|
# orden alfabético SEA el orden cronológico (dos envíos en el mismo segundo no
|
|
# se pisan ni se desordenan), y un sufijo aleatorio por si acaso.
|
|
micro = datetime.datetime.now().strftime("%f")
|
|
archivo = f"{serie}_{iso.replace(':', '-')}-{micro}_{uuid.uuid4().hex[:4]}.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", "resumen"):
|
|
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)
|
|
marca = estado_serie(serie)
|
|
return jsonify(ok=True, serie=serie, actas=actas, relacionadas=relacionadas,
|
|
marca=marca["marca"], marca_acta=marca["acta"],
|
|
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())])
|
|
|
|
|
|
# ------------------------------------------------- estado de una serie (daños abiertos)
|
|
# Un acta de tipo 'dano' MARCA la serie. La marca se levanta con otra acta 'dano' cuyo
|
|
# campo 'accion' empiece por "Reparada" o "Baja": lo último que se dijo, manda.
|
|
def estado_serie(serie):
|
|
"""Devuelve {'marca': None|'DANADA'|'BAJA', 'acta': {...}|None} para una serie."""
|
|
carpeta = os.path.join(BASE, "dano")
|
|
ultimo = None
|
|
# listar_serie() ordena por nombre = cronológico (los repetibles llevan microsegundos).
|
|
# Manda la ÚLTIMA acta registrada, no la de fecha de visita más reciente: si alguien
|
|
# captura hoy un daño de la semana pasada, hoy es cuando la serie se marca.
|
|
for fn in listar_serie("dano", serie) if os.path.isdir(carpeta) else []:
|
|
try:
|
|
with open(os.path.join(carpeta, fn), encoding="utf-8") as f:
|
|
d = json.load(f)
|
|
except Exception:
|
|
continue
|
|
if "PRUEBA" in serie.upper():
|
|
continue
|
|
ultimo = {"archivo": fn, "fecha": d.get("fecha"), "recibido": d.get("_recibido"),
|
|
"tecnico": d.get("tecnico"), "clasificacion": d.get("clasificacion"),
|
|
"accion": d.get("accion"), "resumen": d.get("resumen")}
|
|
if not ultimo:
|
|
return {"marca": None, "acta": None}
|
|
acc = str(ultimo.get("accion") or "")
|
|
if acc.startswith("Baja"):
|
|
return {"marca": "BAJA", "acta": ultimo}
|
|
if acc.startswith("Reparada"):
|
|
return {"marca": None, "acta": ultimo}
|
|
return {"marca": "DANADA", "acta": ultimo}
|
|
|
|
|
|
@app.get("/api/estado")
|
|
def estado():
|
|
"""Marca de daño/baja de una serie — lo consultan TODOS los formularios al elegirla."""
|
|
serie = slug(request.args.get("serie"), 24)
|
|
if not serie:
|
|
return jsonify(ok=False, error="falta serie"), 400
|
|
e = estado_serie(serie)
|
|
return jsonify(ok=True, serie=serie, marca=e["marca"], acta=e["acta"])
|
|
|
|
|
|
@app.get("/api/danos")
|
|
def danos():
|
|
"""Todas las series con marca vigente (dañada o baja), para el tablero."""
|
|
out = []
|
|
carpeta = os.path.join(BASE, "dano")
|
|
if os.path.isdir(carpeta):
|
|
vistas = set()
|
|
for fn in sorted(os.listdir(carpeta)):
|
|
if fn.endswith(".json"):
|
|
ser = fn.split("_2")[0] if "_2" in fn else fn[:-5]
|
|
vistas.add(ser)
|
|
for ser in sorted(vistas):
|
|
e = estado_serie(ser)
|
|
if e["marca"]:
|
|
out.append({"serie": ser, "marca": e["marca"], "acta": e["acta"]})
|
|
return jsonify(ok=True, total=len(out), series=out)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="127.0.0.1", port=int(os.environ.get("PORT", "8081")))
|