- backend Flask (backend/): recibe formularios firmados, guarda JSON por numero de serie en /srv/projects/drogheda/actas, indice CSV, GET /api/actas + systemd unit + README de instalacion (nginx /api proxy) - formularios (docs/formularios/): acta.css/acta.js (motor con autoguardado y envio), V5 fiel al procedimiento verificado, V4 borrador adaptado, checklist de ensamble borrador; firma = nombre + conformidad - docs/actas/index.md + guia-ensamble-qc.md; nav actualizada - probado end-to-end (headless): envio + validacion + guardado por serie Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01YJ4eAB6zjQT49RVwpgfKdt
109 lines
4.1 KiB
Python
109 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Actas HBOX — servicio mínimo para recibir formularios firmados de los técnicos.
|
|
|
|
Guarda cada envío como un JSON por número de serie, y mantiene un índice CSV por tipo
|
|
de acta. Sin base de datos: un archivo por envío, fácil de respaldar y de leer.
|
|
|
|
Rutas:
|
|
POST /api/actas -> recibe un acta (JSON), la guarda, devuelve {ok, archivo}
|
|
GET /api/actas -> lista los envíos (para ingeniería)
|
|
GET /api/actas/<tipo>/<archivo> -> devuelve un acta concreta
|
|
GET /api/salud -> healthcheck
|
|
|
|
Variables de entorno:
|
|
ACTAS_DIR carpeta donde se guardan las actas (default /srv/projects/drogheda/actas)
|
|
"""
|
|
import os, re, json, csv, datetime
|
|
from flask import Flask, request, jsonify, abort
|
|
|
|
app = Flask(__name__)
|
|
BASE = os.environ.get("ACTAS_DIR", "/srv/projects/drogheda/actas")
|
|
|
|
SAFE = re.compile(r"[^A-Za-z0-9_\-]")
|
|
|
|
def slug(s, n=40):
|
|
return SAFE.sub("", str(s or "").strip())[:n]
|
|
|
|
@app.get("/api/salud")
|
|
def salud():
|
|
return jsonify(ok=True, servicio="actas-hbox")
|
|
|
|
@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")) # p.ej. prueba-v5, prueba-v4, ensamble
|
|
serie = slug(d.get("serie"), 20) # p.ej. V5-07
|
|
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
|
|
|
|
ts = datetime.datetime.now().replace(microsecond=0)
|
|
iso = ts.isoformat()
|
|
d["_recibido"] = iso
|
|
d["_ip"] = request.headers.get("X-Forwarded-For", request.remote_addr)
|
|
|
|
carpeta = os.path.join(BASE, tipo)
|
|
os.makedirs(carpeta, exist_ok=True)
|
|
archivo = f"{serie}_{iso.replace(':', '-')}.json"
|
|
ruta = os.path.join(carpeta, archivo)
|
|
with open(ruta, "w", encoding="utf-8") as f:
|
|
json.dump(d, f, ensure_ascii=False, indent=2)
|
|
|
|
# índice CSV acumulado por tipo (fácil de abrir en Excel)
|
|
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", "serie", "tecnico", "resultado", "archivo"])
|
|
w.writerow([iso, serie, tecnico, d.get("resultado", ""), archivo])
|
|
|
|
return jsonify(ok=True, archivo=f"{tipo}/{archivo}")
|
|
|
|
@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
|
|
filas.append({
|
|
"tipo": tipo, "archivo": f"{tipo}/{fn}",
|
|
"serie": d.get("serie"), "tecnico": d.get("tecnico"),
|
|
"resultado": d.get("resultado"), "recibido": d.get("_recibido"),
|
|
})
|
|
filas.sort(key=lambda r: (r.get("recibido") or ""), reverse=True)
|
|
return jsonify(ok=True, total=len(filas), actas=filas)
|
|
|
|
@app.get("/api/actas/<tipo>/<archivo>")
|
|
def leer(tipo, archivo):
|
|
tipo = slug(tipo); archivo = slug(archivo, 120)
|
|
if not archivo.endswith("json"):
|
|
archivo += ".json" if not archivo.endswith(".json") else ""
|
|
ruta = os.path.join(BASE, tipo, archivo)
|
|
if not os.path.isfile(ruta):
|
|
abort(404)
|
|
with open(ruta, encoding="utf-8") as f:
|
|
return app.response_class(f.read(), mimetype="application/json")
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="127.0.0.1", port=int(os.environ.get("PORT", "8081")))
|