666 lines
36 KiB
Python
666 lines
36 KiB
Python
"""Compras: las órdenes de compra del cotizador, dentro del centro de mando (28-oct-2026).
|
|
|
|
Fase 1 de la integración cotizador ↔ centro de mando (documento
|
|
`2026-10-27_Drogheda_Integracion-Cotizador-Inventario_ES_v1.0.md`, decisiones D-27-1 a D-27-5).
|
|
Lo engancha app.py al final con registrar(app, globals()); vive aparte para no crecer app.py.
|
|
|
|
Qué hace, en una línea: el cotizador manda cada cambio de una OC (sin montos); aquí se guarda, el
|
|
material aparece **en camino** en la pestaña del sitio del inventario, se abre **un pendiente por OC**
|
|
para el responsable del sitio, y cuando ese material se recibe el pendiente se cierra solo y el
|
|
cotizador se entera en su siguiente consulta.
|
|
|
|
POST /api/integracion/cotizador/evento (firmado, sin sesión)
|
|
oc.autorizada · oc.pagada · oc.en_camino · oc.cancelada · req.cerrada
|
|
GET /api/integracion/inventario/items (firmado) — el selector de ítems del cotizador
|
|
GET /api/integracion/cotizador/recepciones?desde=… (firmado) — lo que ya llegó, para cerrar la requisición
|
|
|
|
GET /api/compras[?sitio=&estado=] (con sesión) — la página Compras
|
|
POST /api/compras/ligar (con sesión) — ligar un renglón a un ítem del inventario
|
|
POST /api/compras/recibir (con sesión) — recibir una OC completa o en parte
|
|
|
|
Seguridad de las rutas de integración: **no** usan la sesión de una persona. Cada llamada va firmada
|
|
con un secreto compartido (HMAC-SHA256 de la hora + "\\n" + el cuerpo). El secreto vive en el servicio
|
|
(`COTIZADOR_SECRETO`) y en las propiedades del script del cotizador (`PORTAL_SECRETO`); nunca en el
|
|
código ni en un chat. Se rechaza una firma de hace más de 5 minutos. Sin secreto configurado, las tres
|
|
rutas contestan 503 y la página lo dice: el sitio sigue funcionando igual.
|
|
|
|
Montos: el centro de mando **no** los recibe ni los guarda. Si un aviso trae costo, importe, subtotal,
|
|
IVA, total, tipo de cambio o margen, se descartan antes de escribir (lo ve todo el equipo).
|
|
|
|
Datos: SQLite en `/srv/projects/drogheda/drogheda.db` (D-27-2), en modo WAL porque el servicio corre con
|
|
dos trabajadores. El inventario sigue en sus JSON: una OK «en camino» crea sus envíos con las mismas
|
|
funciones que usa la página de Inventario, así que las cuentas por sitio no cambian de origen.
|
|
"""
|
|
import datetime
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
|
|
from flask import jsonify, request
|
|
|
|
VENTANA_FIRMA = 300 # segundos de tolerancia de la firma (reloj y red)
|
|
MAX_CUERPO = 256 * 1024 # un aviso más grande que esto no es una OC
|
|
MONTOS = ("costo", "costo_unitario", "precio", "precio_unitario", "importe", "subtotal", "iva",
|
|
"impuestos", "total", "tipo_cambio", "moneda", "margen", "descuento")
|
|
ESTADOS_OC = ("autorizada", "pagada", "en-camino", "recibida-parcial", "recibida", "cancelada")
|
|
TIPOS = ("oc.autorizada", "oc.pagada", "oc.en_camino", "oc.cancelada", "req.cerrada")
|
|
# Quién recibe el material en cada sitio (el que da «Recibir» y a quien se le abre el pendiente).
|
|
# Se puede cambiar sin tocar el código con COMPRAS_RESPONSABLES="VER:alan,ZLO:jesush,…".
|
|
RESPONSABLES = {"VER": "alan", "ZLO": "jesush", "ESE": "miguel", "LZC": "antonio", "QRO": "jesus"}
|
|
RESPONSABLE_OMISION = "adriana"
|
|
|
|
ESQUEMA = """
|
|
CREATE TABLE IF NOT EXISTS ordenes (
|
|
folio TEXT PRIMARY KEY, req_folio TEXT, proyecto_folio TEXT, proyecto_clave TEXT,
|
|
sitio TEXT, sitio_portal TEXT, proveedor TEXT, estado TEXT, fecha_entrega TEXT,
|
|
paqueteria TEXT, guia TEXT, en_camino_en TEXT, solicitante TEXT, autorizo TEXT,
|
|
tarea_id TEXT, prueba INTEGER DEFAULT 0, req_cerrada INTEGER DEFAULT 0,
|
|
recepcion_en TEXT, recepcion_por TEXT, creado_en TEXT, actualizado_en TEXT);
|
|
CREATE TABLE IF NOT EXISTS renglones (
|
|
oc_folio TEXT, n INTEGER, descripcion TEXT, unidad TEXT, cantidad REAL,
|
|
item_codigo TEXT, item_id TEXT, envio_id TEXT, recibido REAL DEFAULT 0,
|
|
recibido_en TEXT, recibido_por TEXT, nota TEXT,
|
|
PRIMARY KEY (oc_folio, n));
|
|
CREATE TABLE IF NOT EXISTS eventos (
|
|
evento_id TEXT PRIMARY KEY, oc_folio TEXT, tipo TEXT, cuando TEXT, quien TEXT,
|
|
origen TEXT, recibido_en TEXT, datos TEXT);
|
|
CREATE INDEX IF NOT EXISTS ix_ordenes_sitio ON ordenes (sitio_portal, estado);
|
|
CREATE INDEX IF NOT EXISTS ix_renglones_envio ON renglones (envio_id);
|
|
"""
|
|
|
|
|
|
def _responsables():
|
|
m = dict(RESPONSABLES)
|
|
for par in os.environ.get("COMPRAS_RESPONSABLES", "").split(","):
|
|
if ":" in par:
|
|
k, v = par.split(":", 1)
|
|
m[k.strip().upper()] = v.strip().lower()
|
|
return m
|
|
|
|
|
|
def _secreto():
|
|
return os.environ.get("COTIZADOR_SECRETO", "").strip()
|
|
|
|
|
|
def _num(v, defecto=0.0):
|
|
try:
|
|
n = float(v)
|
|
except (TypeError, ValueError):
|
|
return defecto
|
|
return n if n == n and abs(n) != float("inf") else defecto
|
|
|
|
|
|
def _entero_exacto(v):
|
|
"""La cantidad del renglón como entero, o None si no lo es (el inventario cuenta piezas y metros)."""
|
|
n = _num(v, -1)
|
|
return int(round(n)) if n > 0 and abs(n - round(n)) < 1e-9 else None
|
|
|
|
|
|
def _sin_montos(d):
|
|
"""Copia del diccionario sin ninguna llave de dinero, a cualquier profundidad."""
|
|
if isinstance(d, dict):
|
|
return {k: _sin_montos(v) for k, v in d.items() if k.lower() not in MONTOS}
|
|
if isinstance(d, list):
|
|
return [_sin_montos(x) for x in d]
|
|
return d
|
|
|
|
|
|
class Compras:
|
|
"""Todo el estado del módulo: la base y las funciones de app.py que necesita."""
|
|
|
|
def __init__(self, g):
|
|
self.g = g
|
|
self.ruta = os.environ.get("COMPRAS_DB", os.path.join(g["DATOS_BASE"], "drogheda.db"))
|
|
|
|
# ---------------------------------------------------------------- base de datos
|
|
def con(self):
|
|
c = sqlite3.connect(self.ruta, timeout=10)
|
|
c.row_factory = sqlite3.Row
|
|
c.execute("PRAGMA journal_mode=WAL")
|
|
c.execute("PRAGMA busy_timeout=8000")
|
|
c.execute("PRAGMA foreign_keys=ON")
|
|
return c
|
|
|
|
def arrancar(self):
|
|
os.makedirs(os.path.dirname(self.ruta) or ".", exist_ok=True)
|
|
with self.con() as c:
|
|
c.executescript(ESQUEMA)
|
|
|
|
# ---------------------------------------------------------------- sitios y equipo
|
|
def sitio_portal(self, codigo):
|
|
"""'ZLO' o 'timsa' → 'timsa'. Devuelve None si no es un sitio del centro de mando."""
|
|
s = self.g["_norm_id"](self.g["_texto"](codigo, 40))
|
|
s = self.g["ALIAS_SITIO"].get(s, s)
|
|
return s if s in self.g["SITIOS_TAREA_IDS"] else None
|
|
|
|
def codigo(self, sitio_portal):
|
|
return self.g["CODIGO_SITIO"].get(sitio_portal, (sitio_portal or "").upper())
|
|
|
|
def nombre_sitio(self, sitio_portal):
|
|
for i, n, _ in self.g["SITIOS_TAREA"]:
|
|
if i == sitio_portal:
|
|
return "Querétaro" if i == "qro" else n
|
|
return sitio_portal or "?"
|
|
|
|
def responsable(self, codigo):
|
|
quien = _responsables().get((codigo or "").upper(), RESPONSABLE_OMISION)
|
|
miembros = self.g["_miembros"]()
|
|
return quien if quien in miembros else (RESPONSABLE_OMISION if RESPONSABLE_OMISION in miembros
|
|
else (sorted(miembros)[0] if miembros else "adriana"))
|
|
|
|
# ---------------------------------------------------------------- inventario
|
|
def items(self):
|
|
return self.g["_listar_json"](self.g["INV_ITEMS"]())
|
|
|
|
def item_por_codigo(self, codigo):
|
|
c = (codigo or "").strip().upper()
|
|
if not c:
|
|
return None
|
|
for it in self.items():
|
|
if (it.get("codigo") or "").upper() == c:
|
|
return it
|
|
return None
|
|
|
|
def envio(self, envio_id):
|
|
if not envio_id:
|
|
return None
|
|
m = self.g["_leer_json"](self.g["_ruta_registro"](self.g["INV_MOVS"](), envio_id))
|
|
return m if isinstance(m, dict) and m.get("tipo") == "envio" else None
|
|
|
|
# ---------------------------------------------------------------- pendiente de la OC
|
|
def abrir_pendiente(self, oc, quien):
|
|
"""Un pendiente por OC: «Recibir OC-… en ZLO», para el responsable del sitio, que vence el día
|
|
de entrega. Si ya existe (un aviso repetido, o alguien lo abrió a mano), no se duplica."""
|
|
g = self.g
|
|
oc = dict(oc) # llega como sqlite3.Row, que no tiene .get()
|
|
tid = "t-oc-" + oc["folio"].lower().replace("/", "-")
|
|
ruta = g["_ruta_registro"](g["TAREAS_DIR"], tid)
|
|
if os.path.exists(ruta):
|
|
return tid
|
|
sitio = oc["sitio_portal"]
|
|
d = {"titulo": f"Recibir {oc['folio']} en {self.codigo(sitio)}"
|
|
+ (f" — {oc['proveedor']}" if oc.get("proveedor") else ""),
|
|
"descripcion": ("Orden de compra del cotizador. Cuando llegue el material, ábrela en "
|
|
f"**Compras** y dale **Recibir** con lo que llegó de verdad.\n\n"
|
|
f"- Proveedor: {oc.get('proveedor') or '—'}\n"
|
|
f"- Requisición: {oc.get('req_folio') or '—'} · pidió {oc.get('solicitante') or '—'}\n"
|
|
f"- Sitio de entrega: {self.nombre_sitio(sitio)}\n"
|
|
"- Los montos y el pago se quedan en el cotizador."),
|
|
"pista": "compras", "estado": "por-hacer",
|
|
"responsable": self.responsable(self.codigo(sitio)),
|
|
"sitios": [sitio], "etiquetas": ["compras", "oc"],
|
|
"enlaces": [{"titulo": "Compras", "url": "/tableros/compras/?oc=" + oc["folio"]}]}
|
|
if oc.get("fecha_entrega"):
|
|
d["fecha_limite"] = oc["fecha_entrega"]
|
|
t, err = g["_tarea_valida"](d)
|
|
if err:
|
|
return None
|
|
t.pop("_aviso", None)
|
|
ahora = g["ahora"]()
|
|
t.update(id=tid, orden=0, creado_por=quien, creado_en=ahora, modificado_por=quien,
|
|
modificado_en=ahora,
|
|
historial=[{"quien": quien, "cuando": ahora,
|
|
"cambio": f"creado en por-hacer desde el cotizador ({oc['folio']})"}])
|
|
g["_guardar_tarea"](t)
|
|
return tid
|
|
|
|
def nota_pendiente(self, tarea_id, cambio, quien, estado=None):
|
|
g = self.g
|
|
if not tarea_id:
|
|
return
|
|
ruta = g["_ruta_registro"](g["TAREAS_DIR"], tarea_id)
|
|
t = g["_leer_json"](ruta)
|
|
if not t:
|
|
return
|
|
ahora = g["ahora"]()
|
|
if estado and t.get("estado") != estado:
|
|
t["estado"] = estado
|
|
t["modificado_por"], t["modificado_en"] = quien, ahora
|
|
t.setdefault("historial", []).append({"quien": quien, "cuando": ahora, "cambio": cambio})
|
|
g["_guardar_tarea"](t)
|
|
|
|
# ---------------------------------------------------------------- envíos «en camino»
|
|
def crear_envios(self, c, folio, quien):
|
|
"""Pone en camino lo que se pueda de esa OC: un envío por renglón que ya tenga su ítem.
|
|
Los renglones sin ítem quedan en «por ligar» y se mandan cuando alguien los liga."""
|
|
oc = c.execute("SELECT * FROM ordenes WHERE folio=?", (folio,)).fetchone()
|
|
if not oc or oc["estado"] not in ("en-camino", "pagada", "autorizada"):
|
|
return []
|
|
hechos = []
|
|
for r in c.execute("SELECT * FROM renglones WHERE oc_folio=? ORDER BY n", (folio,)).fetchall():
|
|
if r["envio_id"] or not r["item_codigo"]:
|
|
continue
|
|
env = self.crear_envio(c, oc, r, quien)
|
|
if env:
|
|
hechos.append(env)
|
|
return hechos
|
|
|
|
def crear_envio(self, c, oc, r, quien):
|
|
it = self.item_por_codigo(r["item_codigo"])
|
|
if not it:
|
|
c.execute("UPDATE renglones SET item_codigo=NULL, nota=? WHERE oc_folio=? AND n=?",
|
|
(f"El ítem {r['item_codigo']} no existe en el inventario: hay que ligarlo a mano.",
|
|
oc["folio"], r["n"]))
|
|
return None
|
|
cant = _entero_exacto(r["cantidad"])
|
|
if cant is None:
|
|
c.execute("UPDATE renglones SET nota=? WHERE oc_folio=? AND n=?",
|
|
(f"Cantidad {r['cantidad']}: el inventario cuenta piezas y metros enteros. "
|
|
"Recíbelo a mano cuando llegue.", oc["folio"], r["n"]))
|
|
return None
|
|
motivo = (f"compra en camino: {oc['folio']}"
|
|
+ (f" ({oc['proveedor']})" if oc["proveedor"] else "")
|
|
+ (f", guía {oc['guia']}" if oc["guia"] else ""))
|
|
m = self.g["_registrar_movimiento"](it, "envio", cant, motivo[:200], oc["folio"], quien,
|
|
sitio=self.g["ORIGEN_FUERA"], destino=oc["sitio_portal"])
|
|
c.execute("UPDATE renglones SET envio_id=?, item_id=?, nota=NULL WHERE oc_folio=? AND n=?",
|
|
(m["id"], it["id"], oc["folio"], r["n"]))
|
|
return m["id"]
|
|
|
|
def cancelar_envios(self, c, folio, quien, motivo):
|
|
"""Una OC cancelada: lo que iba en camino deja de contar. Se cierra cada envío con una
|
|
recepción de 0 (el inventario baja el total, que es justo lo que pasó: nunca llegó)."""
|
|
for r in c.execute("SELECT * FROM renglones WHERE oc_folio=? AND envio_id IS NOT NULL",
|
|
(folio,)).fetchall():
|
|
env = self.envio(r["envio_id"])
|
|
if not env or env.get("estado") != "en-camino":
|
|
continue
|
|
it = self.g["_leer_json"](self.g["_ruta_registro"](self.g["INV_ITEMS"](), env.get("item_id")))
|
|
if it:
|
|
self.g["_registrar_movimiento"](it, "recepcion", 0, motivo[:200], folio, quien,
|
|
sitio=env.get("destino"), envio=env)
|
|
|
|
# ---------------------------------------------------------------- conciliación
|
|
def conciliar(self, c, folio=None):
|
|
"""Se entera de las recepciones hechas desde la pestaña del sitio en Inventario: si el envío de
|
|
un renglón ya se recibió, lo anota aquí y, si llegó todo, cierra la OC y su pendiente."""
|
|
q = "SELECT * FROM renglones WHERE envio_id IS NOT NULL AND recibido_en IS NULL"
|
|
filas = c.execute(q + (" AND oc_folio=?" if folio else ""), (folio,) if folio else ()).fetchall()
|
|
tocadas = set()
|
|
for r in filas:
|
|
env = self.envio(r["envio_id"])
|
|
if not env or env.get("estado") != "recibido":
|
|
continue
|
|
c.execute("UPDATE renglones SET recibido=?, recibido_en=?, recibido_por=? WHERE oc_folio=? AND n=?",
|
|
(_num(env.get("recibido"), 0), env.get("recibido_en") or self.g["ahora"](),
|
|
env.get("recibido_por") or "", r["oc_folio"], r["n"]))
|
|
tocadas.add(r["oc_folio"])
|
|
for f in (tocadas | ({folio} if folio else set())):
|
|
self.cerrar_si_llego_todo(c, f)
|
|
return tocadas
|
|
|
|
def cerrar_si_llego_todo(self, c, folio):
|
|
oc = c.execute("SELECT * FROM ordenes WHERE folio=?", (folio,)).fetchone()
|
|
if not oc or oc["estado"] in ("cancelada", "recibida"):
|
|
return
|
|
rs = c.execute("SELECT * FROM renglones WHERE oc_folio=?", (folio,)).fetchall()
|
|
if not rs:
|
|
return
|
|
con_algo = [r for r in rs if r["recibido_en"]]
|
|
if not con_algo:
|
|
return
|
|
completa = all(r["recibido_en"] and _num(r["recibido"]) >= _num(r["cantidad"]) for r in rs)
|
|
quien = con_algo[-1]["recibido_por"] or "equipo"
|
|
cuando = max(r["recibido_en"] for r in con_algo)
|
|
estado = "recibida" if completa else "recibida-parcial"
|
|
if oc["estado"] != estado:
|
|
c.execute("UPDATE ordenes SET estado=?, recepcion_en=?, recepcion_por=?, actualizado_en=? WHERE folio=?",
|
|
(estado, cuando, quien, self.g["ahora"](), folio))
|
|
falta = [f"{r['descripcion']}: llegaron {_num(r['recibido']):g} de {_num(r['cantidad']):g}"
|
|
for r in rs if _num(r["recibido"]) < _num(r["cantidad"])]
|
|
if completa:
|
|
self.nota_pendiente(oc["tarea_id"], f"llegó todo lo de {folio}: recibido por {quien}",
|
|
quien, estado="hecho")
|
|
else:
|
|
self.nota_pendiente(oc["tarea_id"],
|
|
(f"llegó parte de {folio} ({quien}). Falta: " + "; ".join(falta))[:300], quien)
|
|
|
|
# ---------------------------------------------------------------- recibir desde el centro de mando
|
|
def recibir(self, c, folio, lineas, quien):
|
|
"""lineas = [{n, recibido}]. Recibe por el inventario lo que tiene envío; lo que no tiene ítem
|
|
se anota aquí (así la requisición se puede cerrar igual). Devuelve (hechos, errores)."""
|
|
oc = c.execute("SELECT * FROM ordenes WHERE folio=?", (folio,)).fetchone()
|
|
if not oc:
|
|
return [], [f"No existe {folio}."]
|
|
if oc["estado"] == "cancelada":
|
|
return [], [f"{folio} está cancelada en el cotizador."]
|
|
hechos, errores = [], []
|
|
for ln in lineas:
|
|
n = self.g["_entero"](ln.get("n"), -1)
|
|
r = c.execute("SELECT * FROM renglones WHERE oc_folio=? AND n=?", (folio, n)).fetchone()
|
|
if not r:
|
|
errores.append(f"El renglón {n} no existe en {folio}.")
|
|
continue
|
|
if r["recibido_en"]:
|
|
continue # ya se había recibido: no se recibe dos veces
|
|
cant = _num(ln.get("recibido"), -1)
|
|
if cant < 0 or cant > _num(r["cantidad"]):
|
|
errores.append(f"Renglón {n}: se pidieron {_num(r['cantidad']):g} {r['unidad'] or ''}"
|
|
.strip() + f"; no pueden llegar {ln.get('recibido')}.")
|
|
continue
|
|
ahora = self.g["ahora"]()
|
|
env = self.envio(r["envio_id"])
|
|
if env and env.get("estado") == "en-camino":
|
|
entero = _entero_exacto(cant) if cant else 0
|
|
if entero is None:
|
|
errores.append(f"Renglón {n}: el inventario cuenta enteros.")
|
|
continue
|
|
it = self.g["_leer_json"](self.g["_ruta_registro"](self.g["INV_ITEMS"](), env.get("item_id")))
|
|
if not it:
|
|
errores.append(f"Renglón {n}: ya no existe el ítem del envío.")
|
|
continue
|
|
self.g["_registrar_movimiento"](it, "recepcion", entero,
|
|
f"llegó la compra {folio}"[:200], folio, quien,
|
|
sitio=env.get("destino"), envio=env)
|
|
c.execute("UPDATE renglones SET recibido=?, recibido_en=?, recibido_por=? WHERE oc_folio=? AND n=?",
|
|
(cant, ahora, quien, folio, n))
|
|
hechos.append(n)
|
|
if hechos:
|
|
self.cerrar_si_llego_todo(c, folio)
|
|
return hechos, errores
|
|
|
|
# ---------------------------------------------------------------- avisos del cotizador
|
|
def guardar_oc(self, c, d, tipo, quien, prueba):
|
|
"""Alta o actualización de la OC con lo que trae el aviso. Devuelve (oc, error)."""
|
|
g = self.g
|
|
folio = g["_texto"](d.get("folio"), 40).upper()
|
|
if not folio:
|
|
return None, "Falta el folio de la orden de compra."
|
|
sitio = self.sitio_portal(d.get("sitio"))
|
|
if not sitio:
|
|
return None, (f"Sitio inválido: {g['_texto'](d.get('sitio'), 20) or '(vacío)'} "
|
|
"(usa VER, ZLO, ESE, LZC o QRO)")
|
|
ahora = g["ahora"]()
|
|
campos = {"req_folio": g["_texto"](d.get("req_folio"), 40).upper(),
|
|
"proyecto_folio": g["_texto"](d.get("proyecto_folio"), 40).upper(),
|
|
"proyecto_clave": g["_texto"](d.get("proyecto_clave"), 20).upper(),
|
|
"sitio": self.codigo(sitio), "sitio_portal": sitio,
|
|
"proveedor": g["_texto"](d.get("proveedor"), 120),
|
|
"fecha_entrega": g["_texto"](d.get("fecha_entrega"), 10),
|
|
"paqueteria": g["_texto"](d.get("paqueteria"), 60),
|
|
"guia": g["_texto"](d.get("guia"), 60),
|
|
"solicitante": g["_texto"](d.get("solicitante"), 120),
|
|
"autorizo": g["_texto"](d.get("autorizo"), 120),
|
|
"prueba": 1 if prueba else 0, "actualizado_en": ahora}
|
|
if campos["fecha_entrega"] and not g["FECHA_RE"].match(campos["fecha_entrega"]):
|
|
return None, "La fecha de entrega va como AAAA-MM-DD."
|
|
base = c.execute("SELECT * FROM ordenes WHERE folio=?", (folio,)).fetchone()
|
|
if base and base["sitio_portal"] != sitio and base["estado"] not in ("autorizada", "pagada"):
|
|
return None, (f"{folio} ya iba a {self.codigo(base['sitio_portal'])}: el sitio no se cambia "
|
|
"con el material en camino. Cancélala y haz otra.")
|
|
estado = {"oc.autorizada": "autorizada", "oc.pagada": "pagada", "oc.en_camino": "en-camino",
|
|
"oc.cancelada": "cancelada"}.get(tipo)
|
|
if estado and base and base["estado"] in ("recibida", "recibida-parcial") and estado != "cancelada":
|
|
estado = None # ya llegó: el estado del portal manda
|
|
if estado:
|
|
campos["estado"] = estado
|
|
if estado == "en-camino":
|
|
campos["en_camino_en"] = ahora
|
|
if tipo == "req.cerrada":
|
|
campos["req_cerrada"] = 1
|
|
if base:
|
|
c.execute("UPDATE ordenes SET " + ", ".join(k + "=?" for k in campos) + " WHERE folio=?",
|
|
tuple(campos.values()) + (folio,))
|
|
else:
|
|
campos.update(folio=folio, creado_en=ahora, estado=campos.get("estado") or "autorizada")
|
|
c.execute("INSERT INTO ordenes (" + ", ".join(campos) + ") VALUES ("
|
|
+ ", ".join("?" * len(campos)) + ")", tuple(campos.values()))
|
|
self.guardar_renglones(c, folio, d.get("renglones"))
|
|
return c.execute("SELECT * FROM ordenes WHERE folio=?", (folio,)).fetchone(), None
|
|
|
|
def guardar_renglones(self, c, folio, renglones):
|
|
"""Los renglones se actualizan, no se borran: un renglón que ya tiene envío conserva su envío
|
|
aunque el cotizador vuelva a mandar la OC."""
|
|
g = self.g
|
|
if not isinstance(renglones, list):
|
|
return
|
|
for i, r in enumerate(renglones[:200], start=1):
|
|
if not isinstance(r, dict):
|
|
continue
|
|
n = g["_entero"](r.get("n"), i)
|
|
desc = g["_texto"](r.get("descripcion"), 200)
|
|
unidad = g["_texto"](r.get("unidad"), 10)
|
|
cant = _num(r.get("cantidad"), 0)
|
|
item = g["_texto"](r.get("item"), 40).upper() or None
|
|
vieja = c.execute("SELECT * FROM renglones WHERE oc_folio=? AND n=?", (folio, n)).fetchone()
|
|
if vieja:
|
|
# el ítem ligado en el centro de mando gana: aquí es donde se ve el inventario
|
|
item = vieja["item_codigo"] or item
|
|
if vieja["envio_id"]:
|
|
cant = _num(vieja["cantidad"], cant)
|
|
c.execute("UPDATE renglones SET descripcion=?, unidad=?, cantidad=?, item_codigo=? "
|
|
"WHERE oc_folio=? AND n=?", (desc, unidad, cant, item, folio, n))
|
|
else:
|
|
c.execute("INSERT INTO renglones (oc_folio, n, descripcion, unidad, cantidad, item_codigo) "
|
|
"VALUES (?,?,?,?,?,?)", (folio, n, desc, unidad, cant, item))
|
|
|
|
# ---------------------------------------------------------------- salida para la página
|
|
def ficha(self, c, oc, con_renglones=True):
|
|
d = {k: oc[k] for k in oc.keys()}
|
|
d["prueba"] = bool(d.get("prueba"))
|
|
d["req_cerrada"] = bool(d.get("req_cerrada"))
|
|
d["sitio_nombre"] = self.nombre_sitio(oc["sitio_portal"])
|
|
if con_renglones:
|
|
rs = [{k: r[k] for k in r.keys()} for r in
|
|
c.execute("SELECT * FROM renglones WHERE oc_folio=? ORDER BY n", (oc["folio"],)).fetchall()]
|
|
for r in rs:
|
|
r["pendiente"] = _num(r["cantidad"]) - _num(r["recibido"]) if r["recibido_en"] else _num(r["cantidad"])
|
|
d["renglones"] = rs
|
|
d["por_ligar"] = [r["n"] for r in rs if not r["item_codigo"]]
|
|
d["recibidos"] = sum(1 for r in rs if r["recibido_en"])
|
|
return d
|
|
|
|
|
|
def registrar(app, g):
|
|
C = Compras(g)
|
|
try:
|
|
C.arrancar()
|
|
except Exception as e: # noqa: BLE001 — sin base, el resto del sitio sigue igual
|
|
app.logger.error("compras.py: no se pudo abrir %s: %s", C.ruta, e)
|
|
|
|
# ------------------------------------------------------------ firma
|
|
def firmado():
|
|
"""Revisa la firma del cotizador. Devuelve (cuerpo_en_texto, None) o (None, respuesta de error)."""
|
|
sec = _secreto()
|
|
if not sec:
|
|
return None, (jsonify(ok=False, error="El centro de mando no tiene configurado "
|
|
"COTIZADOR_SECRETO."), 503)
|
|
crudo = request.get_data(cache=True) or b""
|
|
if len(crudo) > MAX_CUERPO:
|
|
return None, (jsonify(ok=False, error="Aviso demasiado grande."), 413)
|
|
tiempo = request.headers.get("X-Hbox-Tiempo", "")
|
|
firma = request.headers.get("X-Hbox-Firma", "")
|
|
if firma.startswith("sha256="):
|
|
firma = firma[7:]
|
|
try:
|
|
t = datetime.datetime.fromisoformat(tiempo)
|
|
except ValueError:
|
|
return None, (jsonify(ok=False, error="Firma inválida o vencida"), 401)
|
|
ahora = datetime.datetime.now(t.tzinfo) if t.tzinfo else datetime.datetime.now()
|
|
if abs((ahora - t).total_seconds()) > VENTANA_FIRMA:
|
|
return None, (jsonify(ok=False, error="Firma inválida o vencida"), 401)
|
|
esperada = hmac.new(sec.encode(), tiempo.encode() + b"\n" + crudo, hashlib.sha256).hexdigest()
|
|
if not hmac.compare_digest(esperada, (firma or "").strip().lower()):
|
|
return None, (jsonify(ok=False, error="Firma inválida o vencida"), 401)
|
|
return crudo.decode("utf-8", "replace"), None
|
|
|
|
# ------------------------------------------------------------ avisos del cotizador
|
|
@app.post("/api/integracion/cotizador/evento")
|
|
def integracion_evento():
|
|
texto, err = firmado()
|
|
if err:
|
|
return err
|
|
try:
|
|
d = json.loads(texto or "{}")
|
|
except ValueError:
|
|
return jsonify(ok=False, error="JSON inválido"), 400
|
|
if not isinstance(d, dict):
|
|
return jsonify(ok=False, error="JSON inválido"), 400
|
|
d = _sin_montos(d) # los montos nunca entran al centro de mando
|
|
tipo = g["_texto"](d.get("tipo"), 30)
|
|
if tipo not in TIPOS:
|
|
return jsonify(ok=False, error=f"Tipo de aviso inválido: {tipo or '(vacío)'}. "
|
|
"Usa " + ", ".join(TIPOS) + "."), 400
|
|
evento_id = g["_texto"](d.get("evento_id"), 120)
|
|
if not evento_id:
|
|
return jsonify(ok=False, error="Falta evento_id."), 400
|
|
oc_in = d.get("oc") if isinstance(d.get("oc"), dict) else {}
|
|
quien = g["_texto"](d.get("quien"), 120) or "cotizador"
|
|
prueba = bool(d.get("prueba"))
|
|
with C.con() as c:
|
|
if c.execute("SELECT 1 FROM eventos WHERE evento_id=?", (evento_id,)).fetchone():
|
|
return jsonify(ok=True, repetido=True)
|
|
oc, error = C.guardar_oc(c, oc_in, tipo, quien, prueba)
|
|
if error:
|
|
return jsonify(ok=False, error=error), 400
|
|
folio = oc["folio"]
|
|
if prueba:
|
|
# las rondas del proyecto de PRUEBAS no tocan el inventario ni el tablero (contrato §prueba)
|
|
c.execute("INSERT INTO eventos (evento_id, oc_folio, tipo, cuando, quien, origen, datos) "
|
|
"VALUES (?,?,?,?,?,?,?)",
|
|
(evento_id, folio, tipo, g["ahora"](), quien, "cotizador", json.dumps(d)[:20000]))
|
|
return jsonify(ok=True, prueba=True, estado_portal=oc["estado"], tarea_id=None,
|
|
por_ligar=[r["n"] for r in c.execute(
|
|
"SELECT n FROM renglones WHERE oc_folio=? AND item_codigo IS NULL", (folio,))])
|
|
tarea_id = oc["tarea_id"]
|
|
if tipo == "oc.autorizada" and not tarea_id:
|
|
tarea_id = C.abrir_pendiente(c.execute("SELECT * FROM ordenes WHERE folio=?",
|
|
(folio,)).fetchone(), quien)
|
|
c.execute("UPDATE ordenes SET tarea_id=? WHERE folio=?", (tarea_id, folio))
|
|
if tipo == "oc.en_camino":
|
|
C.crear_envios(c, folio, quien)
|
|
C.nota_pendiente(tarea_id, (f"el proveedor ya lo mandó"
|
|
+ (f" ({oc['paqueteria']} {oc['guia']})" if oc["guia"] else "")
|
|
+ f"; llega ≈ {oc['fecha_entrega'] or 'sin fecha'}")[:300], quien)
|
|
elif tipo == "oc.pagada":
|
|
C.nota_pendiente(tarea_id, "pagada en el cotizador", quien)
|
|
elif tipo == "oc.cancelada":
|
|
C.cancelar_envios(c, folio, quien, f"{folio} cancelada en el cotizador")
|
|
C.nota_pendiente(tarea_id, f"{folio} se canceló en el cotizador: ya no llega.",
|
|
quien, estado="hecho")
|
|
c.execute("INSERT INTO eventos (evento_id, oc_folio, tipo, cuando, quien, origen, datos) "
|
|
"VALUES (?,?,?,?,?,?,?)",
|
|
(evento_id, folio, tipo, g["ahora"](), quien, "cotizador", json.dumps(d)[:20000]))
|
|
fila = c.execute("SELECT * FROM ordenes WHERE folio=?", (folio,)).fetchone()
|
|
por_ligar = [r["n"] for r in c.execute(
|
|
"SELECT n FROM renglones WHERE oc_folio=? AND item_codigo IS NULL ORDER BY n", (folio,))]
|
|
return jsonify(ok=True, estado_portal=fila["estado"], tarea_id=fila["tarea_id"], por_ligar=por_ligar)
|
|
|
|
@app.get("/api/integracion/inventario/items")
|
|
def integracion_items():
|
|
_, err = firmado()
|
|
if err:
|
|
return err
|
|
items = [{"codigo": it.get("codigo"), "descripcion": it.get("descripcion"),
|
|
"unidad": it.get("unidad"), "categoria": it.get("categoria")}
|
|
for it in C.items() if it.get("codigo")]
|
|
items.sort(key=lambda x: (x["categoria"] or "", x["codigo"] or ""))
|
|
return jsonify(ok=True, items=items)
|
|
|
|
@app.get("/api/integracion/cotizador/recepciones")
|
|
def integracion_recepciones():
|
|
_, err = firmado()
|
|
if err:
|
|
return err
|
|
desde = g["_texto"](request.args.get("desde"), 40)
|
|
hasta = g["ahora"]()
|
|
with C.con() as c:
|
|
C.conciliar(c)
|
|
q = ("SELECT DISTINCT oc_folio FROM renglones WHERE recibido_en IS NOT NULL"
|
|
+ (" AND recibido_en > ?" if desde else ""))
|
|
folios = [r[0] for r in c.execute(q, (desde,) if desde else ()).fetchall()]
|
|
out = []
|
|
for f in sorted(folios):
|
|
oc = c.execute("SELECT * FROM ordenes WHERE folio=?", (f,)).fetchone()
|
|
if not oc or oc["prueba"]:
|
|
continue
|
|
rs = c.execute("SELECT * FROM renglones WHERE oc_folio=? ORDER BY n", (f,)).fetchall()
|
|
con_algo = [r for r in rs if r["recibido_en"]]
|
|
out.append({"oc_folio": f, "req_folio": oc["req_folio"], "sitio": oc["sitio"],
|
|
"completa": all(r["recibido_en"] and _num(r["recibido"]) >= _num(r["cantidad"])
|
|
for r in rs),
|
|
"recibio": (con_algo[-1]["recibido_por"] if con_algo else ""),
|
|
"cuando": max((r["recibido_en"] for r in con_algo), default=""),
|
|
"renglones": [{"n": r["n"], "pedido": _num(r["cantidad"]),
|
|
"recibido": _num(r["recibido"]) if r["recibido_en"] else 0}
|
|
for r in rs]})
|
|
return jsonify(ok=True, hasta=hasta, recepciones=out)
|
|
|
|
# ------------------------------------------------------------ la página Compras
|
|
@app.get("/api/compras")
|
|
def compras_listar():
|
|
sitio = C.sitio_portal(request.args.get("sitio")) if request.args.get("sitio") else None
|
|
estado = g["_texto"](request.args.get("estado"), 20).lower()
|
|
with C.con() as c:
|
|
C.conciliar(c)
|
|
filas = c.execute("SELECT * FROM ordenes WHERE prueba=0 ORDER BY creado_en DESC").fetchall()
|
|
ocs = [C.ficha(c, f) for f in filas]
|
|
if sitio:
|
|
ocs = [o for o in ocs if o["sitio_portal"] == sitio]
|
|
if estado:
|
|
ocs = [o for o in ocs if o["estado"] == estado]
|
|
resumen = {}
|
|
for o in ocs:
|
|
r = resumen.setdefault(o["sitio_portal"], {k: 0 for k in ESTADOS_OC})
|
|
r[o["estado"]] = r.get(o["estado"], 0) + 1
|
|
items = [{"codigo": it.get("codigo"), "descripcion": it.get("descripcion"),
|
|
"unidad": it.get("unidad"), "categoria": it.get("categoria")}
|
|
for it in C.items() if it.get("codigo")]
|
|
items.sort(key=lambda x: (x["categoria"] or "", x["codigo"] or ""))
|
|
return jsonify(ok=True, total=len(ocs), ordenes=ocs, resumen=resumen, items=items,
|
|
estados=ESTADOS_OC, user=g["current_user"](), admin=g["is_admin"](),
|
|
integracion=bool(_secreto()),
|
|
sitios=[{"id": i, "nombre": n, "corto": c2, "codigo": C.codigo(i)}
|
|
for i, n, c2 in g["SITIOS_TAREA"]])
|
|
|
|
@app.post("/api/compras/ligar")
|
|
def compras_ligar():
|
|
u = g["quien"]()
|
|
if not u:
|
|
return g["_sin_usuario"]()
|
|
d = g["_json_body"]()
|
|
if d is None:
|
|
return jsonify(ok=False, error="JSON inválido"), 400
|
|
folio = g["_texto"](d.get("oc"), 40).upper()
|
|
n = g["_entero"](d.get("n"), -1)
|
|
codigo = g["_texto"](d.get("item"), 40).upper()
|
|
if not C.item_por_codigo(codigo):
|
|
return jsonify(ok=False, error=f"No existe el ítem {codigo} en el inventario."), 404
|
|
with C.con() as c:
|
|
oc = c.execute("SELECT * FROM ordenes WHERE folio=?", (folio,)).fetchone()
|
|
r = c.execute("SELECT * FROM renglones WHERE oc_folio=? AND n=?", (folio, n)).fetchone()
|
|
if not oc or not r:
|
|
return jsonify(ok=False, error=f"No existe el renglón {n} de {folio}."), 404
|
|
if r["envio_id"]:
|
|
return jsonify(ok=False, error="Ese renglón ya está en camino en el inventario."), 409
|
|
if r["recibido_en"]:
|
|
return jsonify(ok=False, error="Ese renglón ya se recibió."), 409
|
|
c.execute("UPDATE renglones SET item_codigo=?, nota=NULL WHERE oc_folio=? AND n=?",
|
|
(codigo, folio, n))
|
|
if oc["estado"] == "en-camino": # ya venía en camino: se manda ahora que ya tiene ítem
|
|
C.crear_envio(c, oc, c.execute("SELECT * FROM renglones WHERE oc_folio=? AND n=?",
|
|
(folio, n)).fetchone(), u)
|
|
ficha = C.ficha(c, c.execute("SELECT * FROM ordenes WHERE folio=?", (folio,)).fetchone())
|
|
return jsonify(ok=True, orden=ficha)
|
|
|
|
@app.post("/api/compras/recibir")
|
|
def compras_recibir():
|
|
u = g["quien"]()
|
|
if not u:
|
|
return g["_sin_usuario"]()
|
|
d = g["_json_body"]()
|
|
if d is None:
|
|
return jsonify(ok=False, error="JSON inválido"), 400
|
|
folio = g["_texto"](d.get("oc"), 40).upper()
|
|
lineas = d.get("renglones")
|
|
if not isinstance(lineas, list) or not lineas:
|
|
return jsonify(ok=False, error="Dime qué llegó: al menos un renglón."), 400
|
|
with C.con() as c:
|
|
hechos, errores = C.recibir(c, folio, lineas, u)
|
|
if errores and not hechos:
|
|
return jsonify(ok=False, error=" ".join(errores)), 400
|
|
ficha = C.ficha(c, c.execute("SELECT * FROM ordenes WHERE folio=?", (folio,)).fetchone())
|
|
return jsonify(ok=True, orden=ficha, recibidos=hechos, avisos=errores)
|