191 lines
7.7 KiB
Python
191 lines
7.7 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)
|
|
"""
|
|
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")
|
|
ADMINS = {u.strip().lower() for u in os.environ.get("ACTAS_ADMINS", "leo").split(",") if u.strip()}
|
|
|
|
SAFE = 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 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"), 20)
|
|
if not tipo or not serie:
|
|
return jsonify(ok=False, error="faltan tipo/serie"), 400
|
|
ruta = find_latest(tipo, serie)
|
|
if not ruta:
|
|
return jsonify(ok=True, exists=False)
|
|
with open(ruta, encoding="utf-8") as f:
|
|
d = json.load(f)
|
|
d["_archivo"] = os.path.basename(ruta)
|
|
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"), 20)
|
|
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
|
|
if not es_prueba 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
|
|
|
|
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"
|
|
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", "serie", "tecnico", "resultado", "archivo"])
|
|
w.writerow([iso, serie, tecnico, d.get("resultado", ""), 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"), 20)
|
|
if not tipo or not serie:
|
|
return jsonify(ok=False, error="faltan tipo/serie"), 400
|
|
carpeta = os.path.join(BASE, tipo)
|
|
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
|
|
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)
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="127.0.0.1", port=int(os.environ.get("PORT", "8081")))
|