""" ERGON - Funciones de query para la API de obras. Cada funcion recibe una conexion SQLite y devuelve un dict o lista lista para serializar a JSON. No hace networking ni maneja HTTP — es logica pura. Convenciones: - Codigos de obra llegan desde la URL (ej: "TDG1"). - Si la obra no existe, las funciones devuelven None o [] segun corresponda. - Datos demo se etiquetan en el payload con "es_demo": true. """ from __future__ import annotations import sqlite3 from pathlib import Path from typing import Any, Optional import os as _os # DATABASE_PATH env var permite override para Railway volume /data/ergon.db DB_PATH = Path(_os.environ.get("DATABASE_PATH", str(Path(__file__).parent / "ergon.db"))) def get_conn() -> sqlite3.Connection: conn = sqlite3.connect(str(DB_PATH)) conn.row_factory = sqlite3.Row conn.execute("PRAGMA foreign_keys = ON") return conn def _row_to_dict(row: sqlite3.Row) -> dict[str, Any]: return {k: row[k] for k in row.keys()} # --------------------------------------------------------------------------- # OBRAS # --------------------------------------------------------------------------- def list_obras(conn: sqlite3.Connection) -> list[dict]: """Lista compacta de obras para el selector del header.""" rows = conn.execute(""" SELECT id, codigo, nombre, cliente, moneda, es_demo, fecha_inicio, fecha_fin_plan, fecha_corte_actual, dg_nivel_servicio FROM obras ORDER BY es_demo ASC, codigo """).fetchall() return [_row_to_dict(r) for r in rows] def get_obra(conn: sqlite3.Connection, codigo: str) -> Optional[dict]: """Detalle completo de una obra por codigo.""" row = conn.execute(""" SELECT * FROM obras WHERE codigo = ? """, (codigo,)).fetchone() return _row_to_dict(row) if row else None def _obra_id(conn: sqlite3.Connection, codigo: str) -> Optional[int]: row = conn.execute("SELECT id FROM obras WHERE codigo = ?", (codigo,)).fetchone() return row["id"] if row else None # --------------------------------------------------------------------------- # RUBROS + SUBCONTRATISTAS # --------------------------------------------------------------------------- def get_rubros(conn: sqlite3.Connection, codigo: str) -> list[dict]: oid = _obra_id(conn, codigo) if oid is None: return [] rows = conn.execute(""" SELECT orden, nombre, peso_pct FROM rubros_obra WHERE obra_id = ? ORDER BY orden """, (oid,)).fetchall() return [_row_to_dict(r) for r in rows] def get_subcontratistas(conn: sqlite3.Connection, codigo: str) -> list[dict]: oid = _obra_id(conn, codigo) if oid is None: return [] rows = conn.execute(""" SELECT s.orden, s.nombre, s.contacto, r.nombre AS rubro FROM subcontratistas s LEFT JOIN rubros_obra r ON r.id = s.rubro_id WHERE s.obra_id = ? ORDER BY s.orden """, (oid,)).fetchall() return [_row_to_dict(r) for r in rows] # --------------------------------------------------------------------------- # AVANCE (con SV por mes + agregado ponderado) # --------------------------------------------------------------------------- def get_avance(conn: sqlite3.Connection, codigo: str) -> dict: oid = _obra_id(conn, codigo) if oid is None: return {"rubros": [], "meses": [], "agregado": None} rubros = conn.execute(""" SELECT id, orden, nombre, peso_pct FROM rubros_obra WHERE obra_id = ? ORDER BY orden """, (oid,)).fetchall() meses = [r["mes"] for r in conn.execute(""" SELECT DISTINCT mes FROM avance_mensual WHERE obra_id = ? ORDER BY mes """, (oid,)).fetchall()] rubros_out = [] for rub in rubros: serie = conn.execute(""" SELECT mes, plan_acum_pct, real_acum_pct FROM avance_mensual WHERE obra_id = ? AND rubro_id = ? ORDER BY mes """, (oid, rub["id"])).fetchall() rubros_out.append({ "orden": rub["orden"], "nombre": rub["nombre"], "peso_pct": rub["peso_pct"], "serie": [_row_to_dict(s) for s in serie], }) # Agregado ponderado por peso para cada mes (plan y real). # Rubros aun no iniciados (real_acum_pct=NULL) cuentan como 0 — consistente con get_evm. agg = conn.execute(""" SELECT a.mes, ROUND(SUM(a.plan_acum_pct * r.peso_pct / 100.0), 2) AS plan_global, ROUND(SUM(CASE WHEN a.real_acum_pct IS NOT NULL THEN a.real_acum_pct * r.peso_pct / 100.0 ELSE 0 END), 2) AS real_global FROM avance_mensual a JOIN rubros_obra r ON r.id = a.rubro_id WHERE a.obra_id = ? GROUP BY a.mes ORDER BY a.mes """, (oid,)).fetchall() return { "meses": meses, "rubros": rubros_out, "agregado": [_row_to_dict(r) for r in agg], } # --------------------------------------------------------------------------- # PRESUPUESTO (plan / real por rubro + EVM agregado) # --------------------------------------------------------------------------- def get_presupuesto(conn: sqlite3.Connection, codigo: str) -> dict: oid = _obra_id(conn, codigo) if oid is None: return {"rubros": [], "agregado": None} rubros = conn.execute(""" SELECT id, orden, nombre, peso_pct FROM rubros_obra WHERE obra_id = ? ORDER BY orden """, (oid,)).fetchall() out = [] for rub in rubros: serie = conn.execute(""" SELECT mes, plan, real FROM presupuesto_mensual WHERE obra_id = ? AND rubro_id = ? ORDER BY mes """, (oid, rub["id"])).fetchall() out.append({ "orden": rub["orden"], "nombre": rub["nombre"], "peso_pct": rub["peso_pct"], "serie": [_row_to_dict(s) for s in serie], }) agg = conn.execute(""" SELECT mes, ROUND(SUM(plan), 0) AS plan_mes, ROUND(SUM(CASE WHEN real IS NOT NULL THEN real ELSE NULL END), 0) AS real_mes FROM presupuesto_mensual WHERE obra_id = ? GROUP BY mes ORDER BY mes """, (oid,)).fetchall() return {"rubros": out, "agregado": [_row_to_dict(r) for r in agg]} # --------------------------------------------------------------------------- # EVM (Earned Value Management) calculado al corte actual # --------------------------------------------------------------------------- def get_evm(conn: sqlite3.Connection, codigo: str) -> Optional[dict]: """Calculo EVM al `fecha_corte_actual` de la obra. Sigue la formula del CURVA_S del FCAT1: BAC = presupuesto_total PV (t) = BAC * plan_acum_pct_global(t) / 100 EV (t) = BAC * real_acum_pct_global(t) / 100 AC (t) = sum(real de presupuesto_mensual hasta t) SV = EV - PV, CV = EV - AC SPI = EV / PV, CPI = EV / AC EAC (CPI) = BAC / CPI Devuelve None si la obra no existe o no hay data suficiente. """ obra = get_obra(conn, codigo) if not obra: return None corte_mes = obra["fecha_corte_actual"][:7] # "YYYY-MM" bac = float(obra["presupuesto_total"]) # Avance global ponderado en el mes del corte row = conn.execute(""" SELECT ROUND(SUM(a.plan_acum_pct * r.peso_pct / 100.0), 4) AS plan_global, ROUND(SUM(CASE WHEN a.real_acum_pct IS NOT NULL THEN a.real_acum_pct * r.peso_pct / 100.0 ELSE 0 END), 4) AS real_global FROM avance_mensual a JOIN rubros_obra r ON r.id = a.rubro_id WHERE a.obra_id = (SELECT id FROM obras WHERE codigo = ?) AND a.mes = ? """, (codigo, corte_mes)).fetchone() if not row or row["plan_global"] is None: return None plan_global = float(row["plan_global"]) real_global = float(row["real_global"] or 0) pv = bac * plan_global / 100.0 ev = bac * real_global / 100.0 ac_row = conn.execute(""" SELECT ROUND(SUM(real), 0) AS ac FROM presupuesto_mensual WHERE obra_id = (SELECT id FROM obras WHERE codigo = ?) AND mes <= ? AND real IS NOT NULL """, (codigo, corte_mes)).fetchone() ac = float(ac_row["ac"] or 0) sv = ev - pv cv = ev - ac spi = round(ev / pv, 4) if pv > 0 else None cpi = round(ev / ac, 4) if ac > 0 else None eac = round(bac / cpi, 0) if cpi else None # Calibracion (Valor 2): intervalo de confianza heuristico segun meses transcurridos. # A mas meses de data, mas angosto el intervalo; menos meses, mas ancho. meses_con_real = conn.execute(""" SELECT COUNT(DISTINCT mes) AS n FROM avance_mensual WHERE obra_id = (SELECT id FROM obras WHERE codigo = ?) AND real_acum_pct IS NOT NULL """, (codigo,)).fetchone()["n"] # Heuristica simple: +/- 15% sobre EAC cuando hay 1 mes, decreciendo a 3% con 18+ meses. if meses_con_real <= 1: banda_pct = 15.0 elif meses_con_real >= 18: banda_pct = 3.0 else: banda_pct = 15.0 - (meses_con_real - 1) * (12.0 / 17.0) eac_lo = round(eac * (1 - banda_pct / 100.0), 0) if eac else None eac_hi = round(eac * (1 + banda_pct / 100.0), 0) if eac else None return { "codigo": codigo, "corte_mes": corte_mes, "bac": round(bac, 0), "pv": round(pv, 0), "ev": round(ev, 0), "ac": round(ac, 0), "sv": round(sv, 0), "cv": round(cv, 0), "spi": spi, "cpi": cpi, "eac": eac, "calibracion": { "meses_con_data": meses_con_real, "banda_pct": round(banda_pct, 1), "eac_lo": eac_lo, "eac_hi": eac_hi, "nota": ("Intervalo se estrecha a medida que la obra genera datos. " f"Con {meses_con_real} mes(es) de data real, banda de +/- {banda_pct:.1f}%."), }, } # --------------------------------------------------------------------------- # ALERTAS: aplicar umbrales umbral_desvio / umbral_critico # --------------------------------------------------------------------------- def get_alertas(conn: sqlite3.Connection, codigo: str) -> dict: """F11 (2026-04-20): delega a `alertas_engine.evaluar()`. Motor de reglas con 8 reglas (R1-R8): avance/sobrecosto por rubro, fecha fin proyectada (EAC_date via SPI), contratos vencidos y sin firma. Retrocompat: cada alerta incluye `estado` como alias de `severidad` para no romper frontends pre-F11. El alias se puede retirar en sprint 4. """ import alertas_engine # import tardio para evitar ciclos al import time payload = alertas_engine.evaluar(conn, codigo) # Alias retrocompat: estado == severidad. Tambien expone desvio_pp / plan_acum_pct # / real_acum_pct desde contexto cuando el tipo lo justifica (R1/R2). for a in payload.get("alertas", []): a["estado"] = a["severidad"] ctx = a.get("contexto") or {} if "desvio_pp" in ctx: a.setdefault("desvio_pp", ctx["desvio_pp"]) if "plan_acum_pct" in ctx: a.setdefault("plan_acum_pct", ctx["plan_acum_pct"]) if "real_acum_pct" in ctx: a.setdefault("real_acum_pct", ctx["real_acum_pct"]) return payload # --------------------------------------------------------------------------- # ASISTENCIA y CONTRATOS y MATERIALES # --------------------------------------------------------------------------- def get_asistencia(conn: sqlite3.Connection, codigo: str, desde: Optional[str] = None, hasta: Optional[str] = None) -> dict: oid = _obra_id(conn, codigo) if oid is None: return {"dias": [], "resumen": None} q = """SELECT fecha, total_personal, lluvia_mm, dia_perdido, notas FROM asistencia_diaria WHERE obra_id = ?""" args: list[Any] = [oid] if desde: q += " AND fecha >= ?" args.append(desde) if hasta: q += " AND fecha <= ?" args.append(hasta) q += " ORDER BY fecha" rows = conn.execute(q, args).fetchall() resumen_row = conn.execute(""" SELECT ROUND(AVG(total_personal), 1) AS promedio, SUM(dia_perdido) AS dias_perdidos, ROUND(SUM(lluvia_mm), 1) AS lluvia_total_mm, COUNT(*) AS n_dias FROM asistencia_diaria WHERE obra_id = ? """, (oid,)).fetchone() return { "dias": [_row_to_dict(r) for r in rows], "resumen": _row_to_dict(resumen_row) if resumen_row else None, } def get_contratos(conn: sqlite3.Connection, codigo: str) -> list[dict]: oid = _obra_id(conn, codigo) if oid is None: return [] rows = conn.execute(""" SELECT c.tipo, c.descripcion, c.fecha_emision, c.fecha_firma, c.fecha_vence, c.estado, c.monto, s.nombre AS subcontratista FROM contratos c LEFT JOIN subcontratistas s ON s.id = c.contratista_id WHERE c.obra_id = ? ORDER BY c.fecha_emision, c.tipo """, (oid,)).fetchall() return [_row_to_dict(r) for r in rows] def get_materiales(conn: sqlite3.Connection, codigo: str) -> list[dict]: oid = _obra_id(conn, codigo) if oid is None: return [] rows = conn.execute(""" SELECT item, unidad, recibido, consumido, ROUND(recibido - consumido, 2) AS saldo, ultima_update FROM materiales WHERE obra_id = ? ORDER BY item """, (oid,)).fetchall() return [_row_to_dict(r) for r in rows] # --------------------------------------------------------------------------- # CRUD PARAMETROS (F7 sprint 2): crear / editar obra completa # --------------------------------------------------------------------------- import re as _re from datetime import date as _date _CODIGO_RE = _re.compile(r"^[A-Z0-9]{3,6}$") _FECHA_RE = _re.compile(r"^\d{4}-\d{2}-\d{2}$") _OBRA_REQUIRED_FIELDS = ( "codigo", "nombre", "cliente", "presupuesto_total", "fecha_inicio", "fecha_fin_plan", "fecha_corte_actual", "umbral_desvio_pct", "umbral_critico_pct", "dg_nivel_servicio", ) _OBRA_OPTIONAL_FIELDS = ( "ubicacion", "tipo", "moneda", "dg_fiscalizador", "dg_responsable", "dg_version_template", "es_demo", ) def _parse_date(s: str) -> Optional[_date]: if not isinstance(s, str) or not _FECHA_RE.match(s): return None try: y, m, d = s.split("-") return _date(int(y), int(m), int(d)) except (ValueError, TypeError): return None def validate_obra_payload(payload: dict, is_update: bool = False) -> list[str]: """Valida el payload antes de tocar la DB. Devuelve lista de errores (strings). Lista vacia = valido. No hace IO. """ errors: list[str] = [] if not isinstance(payload, dict): return ["Payload debe ser un objeto JSON"] # Campos requeridos for field in _OBRA_REQUIRED_FIELDS: if payload.get(field) in (None, ""): errors.append(f"Campo requerido ausente: {field}") # codigo codigo = payload.get("codigo") if isinstance(codigo, str) and not _CODIGO_RE.match(codigo): errors.append( f"codigo invalido '{codigo}'. Formato esperado: 3 a 6 caracteres A-Z o 0-9, sin espacios ni minusculas." ) # presupuesto try: presu = float(payload.get("presupuesto_total", 0)) if presu <= 0: errors.append("presupuesto_total debe ser mayor que 0.") except (TypeError, ValueError): errors.append("presupuesto_total debe ser numerico.") # fechas f_ini = _parse_date(payload.get("fecha_inicio", "")) f_fin = _parse_date(payload.get("fecha_fin_plan", "")) f_cor = _parse_date(payload.get("fecha_corte_actual", "")) if payload.get("fecha_inicio") and not f_ini: errors.append("fecha_inicio debe ser YYYY-MM-DD.") if payload.get("fecha_fin_plan") and not f_fin: errors.append("fecha_fin_plan debe ser YYYY-MM-DD.") if payload.get("fecha_corte_actual") and not f_cor: errors.append("fecha_corte_actual debe ser YYYY-MM-DD.") if f_ini and f_fin and f_fin <= f_ini: errors.append("fecha_fin_plan debe ser posterior a fecha_inicio.") if f_ini and f_cor and f_cor < f_ini: errors.append("fecha_corte_actual no puede ser anterior a fecha_inicio.") # umbrales try: u_des = float(payload.get("umbral_desvio_pct", 0)) u_cri = float(payload.get("umbral_critico_pct", 0)) if u_des < 0 or u_cri < 0: errors.append("umbrales deben ser >= 0.") if u_cri <= u_des: errors.append("umbral_critico_pct debe ser mayor que umbral_desvio_pct.") except (TypeError, ValueError): errors.append("umbrales deben ser numericos.") # nivel servicio try: nivel = int(payload.get("dg_nivel_servicio", 0)) if nivel not in (1, 2, 3): errors.append("dg_nivel_servicio debe ser 1, 2 o 3.") except (TypeError, ValueError): errors.append("dg_nivel_servicio debe ser 1, 2 o 3.") # rubros rubros = payload.get("rubros", []) if not isinstance(rubros, list) or len(rubros) == 0: errors.append("Debe incluir al menos un rubro.") else: suma_pesos = 0.0 nombres_vistos: set[str] = set() for i, r in enumerate(rubros): if not isinstance(r, dict): errors.append(f"Rubro #{i+1}: formato invalido.") continue nombre = (r.get("nombre") or "").strip() if not nombre: errors.append(f"Rubro #{i+1}: nombre vacio.") continue if nombre.lower() in nombres_vistos: errors.append(f"Rubro duplicado: '{nombre}'.") nombres_vistos.add(nombre.lower()) try: peso = float(r.get("peso_pct", 0)) if peso < 0 or peso > 100: errors.append(f"Rubro '{nombre}': peso fuera de rango (0-100).") suma_pesos += peso except (TypeError, ValueError): errors.append(f"Rubro '{nombre}': peso no numerico.") if abs(suma_pesos - 100.0) > 0.01: errors.append( f"Suma de pesos de rubros = {suma_pesos:.2f}%. Debe ser 100% (± 0.01%). " f"Diferencia actual: {suma_pesos - 100:+.2f}pp." ) # subcontratistas (opcional pero si viene debe ser lista) subs = payload.get("subcontratistas", []) if not isinstance(subs, list): errors.append("subcontratistas debe ser una lista.") else: for i, s in enumerate(subs): if not isinstance(s, dict): errors.append(f"Subcontratista #{i+1}: formato invalido.") continue if not (s.get("nombre") or "").strip(): errors.append(f"Subcontratista #{i+1}: nombre vacio.") return errors def upsert_obra(conn: sqlite3.Connection, payload: dict, codigo_existente: Optional[str] = None) -> dict: """Crea o actualiza una obra con sus rubros y subcontratistas. Si `codigo_existente` es None, INSERT. Si se provee, UPDATE (payload.codigo debe coincidir — si difiere, se usa el del path). Transaccional: rollback atomico si cualquier paso falla. Devuelve: {"ok": True, "codigo": "...", "obra_id": N, "created": bool} o: {"ok": False, "errors": [...], "status": 400|404|409|500} """ errors = validate_obra_payload(payload, is_update=bool(codigo_existente)) if errors: return {"ok": False, "errors": errors, "status": 400} codigo = (payload.get("codigo") or "").strip().upper() if codigo_existente: # En UPDATE, el path gobierna. Si el body trajo otro codigo, normalizarlo. codigo = codigo_existente.upper() existing = _obra_id(conn, codigo) if existing is None: return {"ok": False, "errors": [f"Obra {codigo} no encontrada"], "status": 404} else: if _obra_id(conn, codigo) is not None: return { "ok": False, "errors": [f"Ya existe una obra con codigo {codigo}. Use editar en lugar de crear."], "status": 409, } # Normalizar campos opcionales moneda = (payload.get("moneda") or "PYG").strip() or "PYG" version = (payload.get("dg_version_template") or "v1.0").strip() or "v1.0" es_demo = 1 if payload.get("es_demo") in (1, True, "1") else 0 cur = conn.cursor() try: cur.execute("BEGIN") if codigo_existente: cur.execute(""" UPDATE obras SET nombre = ?, cliente = ?, ubicacion = ?, tipo = ?, moneda = ?, presupuesto_total = ?, fecha_inicio = ?, fecha_fin_plan = ?, fecha_corte_actual = ?, umbral_desvio_pct = ?, umbral_critico_pct = ?, dg_nivel_servicio = ?, dg_fiscalizador = ?, dg_responsable = ?, dg_version_template = ?, es_demo = ?, updated_at = datetime('now') WHERE codigo = ? """, ( payload.get("nombre"), payload.get("cliente"), payload.get("ubicacion"), payload.get("tipo"), moneda, float(payload.get("presupuesto_total")), payload.get("fecha_inicio"), payload.get("fecha_fin_plan"), payload.get("fecha_corte_actual"), float(payload.get("umbral_desvio_pct")), float(payload.get("umbral_critico_pct")), int(payload.get("dg_nivel_servicio")), payload.get("dg_fiscalizador"), payload.get("dg_responsable"), version, es_demo, codigo, )) obra_id = _obra_id(conn, codigo) # Reemplazo completo de rubros y subcontratistas. # Dependencias (avance/presupuesto/contratos) se preservan solo si los # rubros/subs conservan el mismo id. Para simplificar el CRUD inicial # borramos y recreamos — en la proxima sesion se puede hacer diff-merge # si se quiere preservar series historicas por rubro renombrado. cur.execute("DELETE FROM subcontratistas WHERE obra_id = ?", (obra_id,)) cur.execute("DELETE FROM rubros_obra WHERE obra_id = ?", (obra_id,)) created = False else: cur.execute(""" INSERT INTO obras ( codigo, nombre, cliente, ubicacion, tipo, moneda, presupuesto_total, fecha_inicio, fecha_fin_plan, fecha_corte_actual, umbral_desvio_pct, umbral_critico_pct, dg_nivel_servicio, dg_fiscalizador, dg_responsable, dg_version_template, es_demo ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( codigo, payload.get("nombre"), payload.get("cliente"), payload.get("ubicacion"), payload.get("tipo"), moneda, float(payload.get("presupuesto_total")), payload.get("fecha_inicio"), payload.get("fecha_fin_plan"), payload.get("fecha_corte_actual"), float(payload.get("umbral_desvio_pct")), float(payload.get("umbral_critico_pct")), int(payload.get("dg_nivel_servicio")), payload.get("dg_fiscalizador"), payload.get("dg_responsable"), version, es_demo, )) obra_id = cur.lastrowid created = True # Insert rubros (orden 1..N segun llegada si no viene explicito) rubro_id_by_nombre: dict[str, int] = {} for idx, r in enumerate(payload.get("rubros", []), start=1): nombre = (r.get("nombre") or "").strip() peso = float(r.get("peso_pct", 0)) orden = int(r.get("orden", idx)) cur.execute(""" INSERT INTO rubros_obra (obra_id, orden, nombre, peso_pct) VALUES (?, ?, ?, ?) """, (obra_id, orden, nombre, peso)) rubro_id_by_nombre[nombre.lower()] = cur.lastrowid # Insert subcontratistas (orden 1..N, link a rubro por nombre si coincide) for idx, s in enumerate(payload.get("subcontratistas", []), start=1): nombre = (s.get("nombre") or "").strip() rubro_ref = (s.get("rubro") or "").strip().lower() rubro_id = rubro_id_by_nombre.get(rubro_ref) orden = int(s.get("orden", idx)) cur.execute(""" INSERT INTO subcontratistas (obra_id, orden, nombre, rubro_id, contacto, notas) VALUES (?, ?, ?, ?, ?, ?) """, ( obra_id, orden, nombre, rubro_id, s.get("contacto"), s.get("notas"), )) conn.commit() return {"ok": True, "codigo": codigo, "obra_id": obra_id, "created": created} except sqlite3.IntegrityError as e: conn.rollback() msg = str(e) # Caso especifico: el DELETE de rubros_obra / subcontratistas falla con FK # cuando la obra ya tiene avance_mensual, presupuesto_mensual, contratos o # materiales apuntando a esos ids. Pasa al editar obras con historico. if "FOREIGN KEY" in msg.upper() and codigo_existente: return { "ok": False, "errors": [ "No se puede reemplazar la lista de rubros o subcontratistas: la obra " f"{codigo_existente} ya tiene avance, presupuesto o contratos registrados " "que los referencian. Edite los datos escalares (fechas, presupuesto, " "umbrales) sin cambiar la lista de rubros, o solicite un merge asistido." ], "status": 409, } return { "ok": False, "errors": [f"Restriccion de base de datos: {msg}"], "status": 400, } except Exception as e: conn.rollback() return {"ok": False, "errors": [f"Error interno: {e}"], "status": 500} # --------------------------------------------------------------------------- # FOTOS DE OBRA (sprint 2 F9) # --------------------------------------------------------------------------- _FOTO_FIELDS = ( "id", "obra_id", "fecha", "sector", "piso", "nota", "blob_path", "exif_gps_lat", "exif_gps_lon", "exif_fecha", "usuario", "mime_type", "size_bytes", "created_at", ) def create_foto( conn: sqlite3.Connection, codigo: str, meta: dict, blob_path: str, ) -> dict: """Inserta una fila en fotos_obra. Devuelve {ok, foto, status} o {ok:False, errors}.""" obra_id = _obra_id(conn, codigo) if obra_id is None: return {"ok": False, "errors": [f"obra {codigo} no encontrada"], "status": 404} fecha = (meta.get("fecha") or "").strip() if not fecha: return {"ok": False, "errors": ["fecha es obligatoria (YYYY-MM-DD)"], "status": 400} if not _FECHA_RE.match(fecha): return { "ok": False, "errors": [f"fecha invalida '{fecha}'. Formato YYYY-MM-DD."], "status": 400, } try: cur = conn.execute( """ INSERT INTO fotos_obra (obra_id, fecha, sector, piso, nota, blob_path, exif_gps_lat, exif_gps_lon, exif_fecha, usuario, mime_type, size_bytes) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( obra_id, fecha, (meta.get("sector") or None), (meta.get("piso") or None), (meta.get("nota") or None), blob_path, meta.get("exif_gps_lat"), meta.get("exif_gps_lon"), (meta.get("exif_fecha") or None), (meta.get("usuario") or None), (meta.get("mime_type") or "image/jpeg"), int(meta.get("size_bytes") or 0), ), ) conn.commit() foto = get_foto(conn, codigo, cur.lastrowid) return {"ok": True, "foto": foto, "status": 201} except sqlite3.IntegrityError as e: conn.rollback() return {"ok": False, "errors": [f"Error DB: {e}"], "status": 400} def list_fotos( conn: sqlite3.Connection, codigo: str, desde: Optional[str] = None, hasta: Optional[str] = None, sector: Optional[str] = None, ) -> list[dict]: """Lista fotos de una obra con filtros opcionales. Ordena por fecha desc, id desc.""" obra_id = _obra_id(conn, codigo) if obra_id is None: return [] sql = ( "SELECT " + ", ".join(_FOTO_FIELDS) + " FROM fotos_obra " "WHERE obra_id = ?" ) params: list[Any] = [obra_id] if desde: sql += " AND fecha >= ?" params.append(desde) if hasta: sql += " AND fecha <= ?" params.append(hasta) if sector: sql += " AND sector = ?" params.append(sector) sql += " ORDER BY fecha DESC, id DESC" rows = conn.execute(sql, params).fetchall() return [_row_to_dict(r) for r in rows] def get_foto( conn: sqlite3.Connection, codigo: str, foto_id: int, ) -> Optional[dict]: """Devuelve la foto por id, solo si pertenece a la obra codigo.""" obra_id = _obra_id(conn, codigo) if obra_id is None: return None row = conn.execute( "SELECT " + ", ".join(_FOTO_FIELDS) + " FROM fotos_obra " "WHERE id = ? AND obra_id = ?", (foto_id, obra_id), ).fetchone() return _row_to_dict(row) if row else None # --------------------------------------------------------------------------- # DAILY LOG (sprint 2 F9) # --------------------------------------------------------------------------- _DAILY_FIELDS = ( "id", "obra_id", "fecha", "autor", "actividades", "observaciones", "clima", "decisiones_key", "created_at", "parse_estado", "parse_procesado_at", "parse_confirmado_at", ) def create_daily_log( conn: sqlite3.Connection, codigo: str, payload: dict, ) -> dict: """Inserta una fila en daily_log. Devuelve {ok, entry, status} o {ok:False, errors}.""" obra_id = _obra_id(conn, codigo) if obra_id is None: return {"ok": False, "errors": [f"obra {codigo} no encontrada"], "status": 404} fecha = (payload.get("fecha") or "").strip() if not fecha: return {"ok": False, "errors": ["fecha es obligatoria (YYYY-MM-DD)"], "status": 400} if not _FECHA_RE.match(fecha): return { "ok": False, "errors": [f"fecha invalida '{fecha}'. Formato YYYY-MM-DD."], "status": 400, } actividades = (payload.get("actividades") or "").strip() observaciones = (payload.get("observaciones") or "").strip() if not actividades and not observaciones: return { "ok": False, "errors": ["al menos uno de 'actividades' u 'observaciones' debe contener texto"], "status": 400, } try: cur = conn.execute( """ INSERT INTO daily_log (obra_id, fecha, autor, actividades, observaciones, clima, decisiones_key) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( obra_id, fecha, (payload.get("autor") or None), actividades or None, observaciones or None, (payload.get("clima") or None), (payload.get("decisiones_key") or None), ), ) conn.commit() entry = conn.execute( "SELECT " + ", ".join(_DAILY_FIELDS) + " FROM daily_log WHERE id = ?", (cur.lastrowid,), ).fetchone() return {"ok": True, "entry": _row_to_dict(entry), "status": 201} except sqlite3.IntegrityError as e: conn.rollback() return {"ok": False, "errors": [f"Error DB: {e}"], "status": 400} def list_daily_log( conn: sqlite3.Connection, codigo: str, desde: Optional[str] = None, hasta: Optional[str] = None, ) -> list[dict]: """Lista entradas daily_log de una obra. Ordena fecha desc, id desc.""" obra_id = _obra_id(conn, codigo) if obra_id is None: return [] sql = ( "SELECT " + ", ".join(_DAILY_FIELDS) + " FROM daily_log " "WHERE obra_id = ?" ) params: list[Any] = [obra_id] if desde: sql += " AND fecha >= ?" params.append(desde) if hasta: sql += " AND fecha <= ?" params.append(hasta) sql += " ORDER BY fecha DESC, id DESC" rows = conn.execute(sql, params).fetchall() return [_row_to_dict(r) for r in rows] # --------------------------------------------------------------------------- # AUTH (sprint 2 F13): usuarios + sesiones + filtro por obra # --------------------------------------------------------------------------- import secrets as _secrets from datetime import datetime as _dt, timedelta as _td try: import bcrypt as _bcrypt # type: ignore _BCRYPT_OK = True except ImportError: _bcrypt = None # type: ignore _BCRYPT_OK = False _SESSION_TTL_DAYS = 7 _EMAIL_RE = _re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") def bcrypt_available() -> bool: return _BCRYPT_OK def hash_password(plain: str) -> str: """Devuelve hash bcrypt como string utf-8. Lanza si bcrypt no instalado.""" if not _BCRYPT_OK: raise RuntimeError("bcrypt no instalado. Ejecute: pip install bcrypt") return _bcrypt.hashpw(plain.encode("utf-8"), _bcrypt.gensalt()).decode("utf-8") def verify_password(plain: str, hashed: str) -> bool: """Comparacion constant-time via bcrypt. False si bcrypt no esta disponible.""" if not _BCRYPT_OK or not hashed: return False try: return _bcrypt.checkpw(plain.encode("utf-8"), hashed.encode("utf-8")) except (ValueError, TypeError): return False def _usuario_row_to_dict(row: sqlite3.Row) -> dict: """Nunca incluir password_hash en la salida.""" base = { "id": row["id"], "email": row["email"], "nombre": row["nombre"], "rol": row["rol"], "activo": bool(row["activo"]), "created_at": row["created_at"], "last_login_at": row["last_login_at"], } # Campos de tier / trial opcionales (post migrate_trial). Tolera ausencia # para retro-compatibilidad si la migration aun no corrio. try: keys = set(row.keys()) except Exception: keys = set() if "plan_tier" in keys: base["plan_tier"] = row["plan_tier"] if "trial_started_at" in keys: base["trial_started_at"] = row["trial_started_at"] if "trial_ends_at" in keys: base["trial_ends_at"] = row["trial_ends_at"] if "subscription_status" in keys: base["subscription_status"] = row["subscription_status"] if "empresa" in keys: base["empresa"] = row["empresa"] return base # Tier limits canonicos Sprint 5 GTM TIER_LIMITS: dict[str, dict] = { "trial": {"max_obras": 1, "max_usuarios": 3, "storage_gb": 1, "watermark": True}, "n1": {"max_obras": 1, "max_usuarios": 5, "storage_gb": 10, "watermark": False}, "n2": {"max_obras": 3, "max_usuarios": 10, "storage_gb": 30, "watermark": False}, "n3": {"max_obras": 10, "max_usuarios": 25, "storage_gb": 100, "watermark": False}, "enterprise": {"max_obras": 9999, "max_usuarios": 9999, "storage_gb": 9999, "watermark": False}, } def get_trial_status(usuario: dict) -> dict: """Calcula el estado de prueba/suscripcion del usuario. Retorna dict con: plan_tier: tier vigente subscription_status: trial | active | expired | cancelled is_trial: True si plan_tier == 'trial' trial_active: True si trial y dentro de los 14 dias trial_expired: True si trial y vencido days_remaining: enteros >= 0 si trial activo, 0 si expirado, None si no trial watermark_required: True si exports deben llevar marca DEMO limits: dict con max_obras / max_usuarios / storage_gb """ from datetime import datetime, timezone plan_tier = (usuario.get("plan_tier") or "trial").lower() sub_status = (usuario.get("subscription_status") or "trial").lower() limits = TIER_LIMITS.get(plan_tier, TIER_LIMITS["trial"]) is_trial = plan_tier == "trial" trial_active = False trial_expired = False days_remaining: Optional[int] = None if is_trial: ends_at_str = usuario.get("trial_ends_at") if ends_at_str: try: # SQLite datetime() format 'YYYY-MM-DD HH:MM:SS' ends_dt = datetime.fromisoformat(ends_at_str.replace(" ", "T")) if ends_dt.tzinfo is None: ends_dt = ends_dt.replace(tzinfo=timezone.utc) now = datetime.now(timezone.utc) delta = ends_dt - now secs = delta.total_seconds() if secs > 0: trial_active = True days_remaining = max(0, int(secs // 86400)) else: trial_expired = True days_remaining = 0 except (ValueError, TypeError): # Si trial_ends_at esta mal formateado, asumir activo trial_active = True days_remaining = 14 else: # sin fecha de expiracion definida, asumir activo trial_active = True days_remaining = 14 # Subscription cancelada o expirada fuerza expired_state if sub_status in ("expired", "cancelled"): trial_active = False if is_trial: trial_expired = True watermark_required = bool(limits.get("watermark") or trial_expired) return { "plan_tier": plan_tier, "subscription_status": sub_status, "is_trial": is_trial, "trial_active": trial_active, "trial_expired": trial_expired, "days_remaining": days_remaining, "watermark_required": watermark_required, "limits": dict(limits), } def can_create_obra(conn: sqlite3.Connection, usuario: dict) -> tuple[bool, str]: """Verifica si el usuario puede crear una obra mas segun su tier. Retorna (allowed, reason). reason es string vacio si allowed=True. """ status = get_trial_status(usuario) if status["trial_expired"]: return False, "Tu prueba de 14 dias vencio. Activa una suscripcion para seguir creando obras." limit = status["limits"]["max_obras"] # Conteo simple: obras visibles para el usuario (m2m usuarios_obras) usuario_id = usuario.get("id") if not usuario_id: return False, "Usuario invalido." if usuario.get("rol") == "admin": # admin global, sin limite return True, "" count_row = conn.execute( "SELECT COUNT(*) AS c FROM usuarios_obras WHERE usuario_id = ?", (int(usuario_id),), ).fetchone() count = count_row[0] if count_row else 0 if count >= limit: return False, ( f"Tu plan {status['plan_tier'].upper()} permite hasta {limit} obra(s). " f"Ya tenes {count}. Actualiza a un plan superior para sumar mas." ) return True, "" def get_usuario_by_email(conn: sqlite3.Connection, email: str) -> Optional[sqlite3.Row]: """Devuelve la Row cruda (incluye password_hash para uso interno de login).""" if not isinstance(email, str) or not _EMAIL_RE.match(email): return None return conn.execute( "SELECT * FROM usuarios WHERE email = ? AND activo = 1", (email.lower(),), ).fetchone() def get_usuario(conn: sqlite3.Connection, usuario_id: int) -> Optional[dict]: row = conn.execute( "SELECT * FROM usuarios WHERE id = ?", (int(usuario_id),), ).fetchone() return _usuario_row_to_dict(row) if row else None def create_usuario(conn: sqlite3.Connection, payload: dict) -> dict: """Crea un usuario. Devuelve {ok, usuario, status} o {ok:False, errors, status}.""" if not _BCRYPT_OK: return {"ok": False, "errors": ["bcrypt no instalado en el servidor."], "status": 500} email = (payload.get("email") or "").strip().lower() password = payload.get("password") or "" nombre = (payload.get("nombre") or "").strip() rol = (payload.get("rol") or "").strip() errors: list[str] = [] if not _EMAIL_RE.match(email): errors.append("email invalido (formato esperado: usuario@dominio.tld).") if len(password) < 6: errors.append("password debe tener al menos 6 caracteres.") if not nombre: errors.append("nombre requerido.") if rol not in ("admin", "cliente"): errors.append("rol debe ser 'admin' o 'cliente'.") if errors: return {"ok": False, "errors": errors, "status": 400} if conn.execute("SELECT 1 FROM usuarios WHERE email = ?", (email,)).fetchone(): return {"ok": False, "errors": [f"Ya existe usuario con email {email}."], "status": 409} try: cur = conn.execute( "INSERT INTO usuarios (email, password_hash, nombre, rol) " "VALUES (?, ?, ?, ?)", (email, hash_password(password), nombre, rol), ) conn.commit() return {"ok": True, "usuario": get_usuario(conn, cur.lastrowid), "status": 201} except sqlite3.IntegrityError as e: conn.rollback() return {"ok": False, "errors": [f"Error DB: {e}"], "status": 400} def assign_usuario_obra(conn: sqlite3.Connection, usuario_id: int, obra_id: int) -> dict: """M2M usuarios_obras. Idempotente (INSERT OR IGNORE).""" u = conn.execute("SELECT id, rol FROM usuarios WHERE id = ?", (int(usuario_id),)).fetchone() if u is None: return {"ok": False, "errors": [f"usuario_id {usuario_id} no existe."], "status": 404} o = conn.execute("SELECT id FROM obras WHERE id = ?", (int(obra_id),)).fetchone() if o is None: return {"ok": False, "errors": [f"obra_id {obra_id} no existe."], "status": 404} try: conn.execute( "INSERT OR IGNORE INTO usuarios_obras (usuario_id, obra_id) VALUES (?, ?)", (int(usuario_id), int(obra_id)), ) conn.commit() return {"ok": True, "status": 201} except sqlite3.IntegrityError as e: conn.rollback() return {"ok": False, "errors": [f"Error DB: {e}"], "status": 400} def get_obras_ids_de_usuario(conn: sqlite3.Connection, usuario_id: int) -> set[int]: rows = conn.execute( "SELECT obra_id FROM usuarios_obras WHERE usuario_id = ?", (int(usuario_id),), ).fetchall() return {r["obra_id"] for r in rows} def filter_obras_por_usuario( obras_list: list[dict], usuario_id: int, rol: str, ids_permitidas: Optional[set[int]] = None, conn: Optional[sqlite3.Connection] = None, ) -> list[dict]: """Admin ve todas; cliente solo las asignadas en usuarios_obras.""" if rol == "admin": return obras_list if ids_permitidas is None: if conn is None: return [] ids_permitidas = get_obras_ids_de_usuario(conn, usuario_id) return [o for o in obras_list if o.get("id") in ids_permitidas] def usuario_tiene_acceso_obra( conn: sqlite3.Connection, usuario_id: int, rol: str, codigo_obra: str, ) -> bool: if rol == "admin": return True oid = _obra_id(conn, codigo_obra) if oid is None: return False row = conn.execute( "SELECT 1 FROM usuarios_obras WHERE usuario_id = ? AND obra_id = ?", (int(usuario_id), oid), ).fetchone() return row is not None # --- Sesiones -------------------------------------------------------------- def create_session(conn: sqlite3.Connection, usuario_id: int) -> dict: """Genera token urlsafe 32 bytes. TTL 7 dias. Actualiza last_login_at.""" token = _secrets.token_urlsafe(32) expira = (_dt.utcnow() + _td(days=_SESSION_TTL_DAYS)).strftime("%Y-%m-%d %H:%M:%S") conn.execute( "INSERT INTO sesiones (token, usuario_id, expira_at) VALUES (?, ?, ?)", (token, int(usuario_id), expira), ) conn.execute( "UPDATE usuarios SET last_login_at = datetime('now') WHERE id = ?", (int(usuario_id),), ) # Oportunistic cleanup de sesiones vencidas del mismo usuario. conn.execute( "DELETE FROM sesiones WHERE usuario_id = ? AND expira_at < datetime('now')", (int(usuario_id),), ) conn.commit() return {"token": token, "expira_at": expira, "ttl_days": _SESSION_TTL_DAYS} def get_session(conn: sqlite3.Connection, token: str) -> Optional[dict]: """Valida token + no expirado + usuario activo. Devuelve dict con datos del usuario o None.""" if not token or not isinstance(token, str) or len(token) < 20: return None row = conn.execute( """ SELECT s.token, s.usuario_id, s.expira_at, u.email, u.nombre, u.rol, u.activo FROM sesiones s JOIN usuarios u ON u.id = s.usuario_id WHERE s.token = ? AND s.expira_at > datetime('now') AND u.activo = 1 """, (token,), ).fetchone() if row is None: return None return { "token": row["token"], "usuario_id": row["usuario_id"], "email": row["email"], "nombre": row["nombre"], "rol": row["rol"], "expira_at": row["expira_at"], } def delete_session(conn: sqlite3.Connection, token: str) -> bool: if not token: return False cur = conn.execute("DELETE FROM sesiones WHERE token = ?", (token,)) conn.commit() return cur.rowcount > 0 # --------------------------------------------------------------------------- # DOCUMENTOS (sprint 3 F8, 2026-04-20) # --------------------------------------------------------------------------- _DOC_CATEGORIAS_VALIDAS = frozenset({ "plano", "contrato", "certificado", "acta", "orden_pago", "informe", "foto", "garantia", }) _DOC_FIELDS = ( "id", "obra_id", "categoria", "titulo", "tags", "blob_path", "mime_type", "size_bytes", "sha256", "fecha_doc", "subido_por", "notas", "created_at", ) def _normalize_tags_csv(tags_raw: str) -> Optional[str]: """Normaliza CSV: trim, lowercase, dedup, ordenado alfabeticamente. None si vacio.""" if not tags_raw: return None items = [t.strip().lower() for t in tags_raw.split(",") if t.strip()] if not items: return None return ",".join(sorted(set(items))) def find_documento_by_sha( conn: sqlite3.Connection, obra_id: int, sha256: str, ) -> list[dict]: """Devuelve docs existentes en la obra con mismo sha256 (dedup check).""" if not sha256: return [] rows = conn.execute( "SELECT " + ", ".join(_DOC_FIELDS) + " FROM documentos " "WHERE obra_id = ? AND sha256 = ? ORDER BY created_at DESC", (int(obra_id), sha256), ).fetchall() return [_row_to_dict(r) for r in rows] def create_documento( conn: sqlite3.Connection, codigo: str, meta: dict, blob_path: str, ) -> dict: """Inserta una fila en documentos. Valida categoria (enum 8), titulo no vacio, sha256 presente. No escribe a disco — el blob ya fue guardado por el caller. Retrocompat: devuelve `{ok, documento, status, duplicados?: [...]}`. `duplicados` lista docs previos con mismo sha256 (aviso UI, no rechazo). """ obra_id = _obra_id(conn, codigo) if obra_id is None: return {"ok": False, "errors": [f"obra {codigo} no encontrada"], "status": 404} categoria = (meta.get("categoria") or "").strip().lower() if categoria not in _DOC_CATEGORIAS_VALIDAS: return { "ok": False, "errors": [( f"categoria invalida '{categoria}'. Validas: " f"{', '.join(sorted(_DOC_CATEGORIAS_VALIDAS))}." )], "status": 400, } titulo = (meta.get("titulo") or "").strip() if not titulo: return {"ok": False, "errors": ["titulo es obligatorio"], "status": 400} if len(titulo) > 200: return {"ok": False, "errors": ["titulo demasiado largo (max 200 chars)"], "status": 400} sha256 = (meta.get("sha256") or "").strip().lower() if not sha256 or len(sha256) != 64: return { "ok": False, "errors": ["sha256 invalido (hex 64 chars obligatorio)"], "status": 400, } fecha_doc = (meta.get("fecha_doc") or "").strip() or None if fecha_doc and not _FECHA_RE.match(fecha_doc): return { "ok": False, "errors": [f"fecha_doc invalida '{fecha_doc}'. Formato YYYY-MM-DD."], "status": 400, } if not blob_path or not str(blob_path).strip(): return {"ok": False, "errors": ["blob_path es obligatorio"], "status": 400} tags = _normalize_tags_csv(meta.get("tags") or "") # Dedup check: previa al insert, informativo (no rechaza) duplicados = find_documento_by_sha(conn, obra_id, sha256) try: cur = conn.execute( """ INSERT INTO documentos (obra_id, categoria, titulo, tags, blob_path, mime_type, size_bytes, sha256, fecha_doc, subido_por, notas) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( obra_id, categoria, titulo, tags, blob_path, (meta.get("mime_type") or "application/octet-stream"), int(meta.get("size_bytes") or 0), sha256, fecha_doc, (meta.get("subido_por") or None), (meta.get("notas") or None), ), ) conn.commit() doc = get_documento(conn, codigo, cur.lastrowid) resp: dict = {"ok": True, "documento": doc, "status": 201} if duplicados: resp["duplicados"] = duplicados return resp except sqlite3.IntegrityError as e: conn.rollback() return {"ok": False, "errors": [f"Error DB: {e}"], "status": 400} def list_documentos( conn: sqlite3.Connection, codigo: str, categoria: Optional[str] = None, tag: Optional[str] = None, desde: Optional[str] = None, hasta: Optional[str] = None, ) -> list[dict]: """Lista documentos con filtros opcionales. Orden: created_at desc, id desc.""" obra_id = _obra_id(conn, codigo) if obra_id is None: return [] sql = ( "SELECT " + ", ".join(_DOC_FIELDS) + " FROM documentos " "WHERE obra_id = ?" ) params: list[Any] = [obra_id] if categoria: cat_norm = categoria.strip().lower() if cat_norm not in _DOC_CATEGORIAS_VALIDAS: # Filtro invalido: devolver lista vacia en vez de raise (UX-friendly) return [] sql += " AND categoria = ?" params.append(cat_norm) if tag: # Busqueda por tag dentro del CSV normalizado. Tags estan lowercase, # separados por ',' sin espacios — buscar como palabra exacta. tag_norm = tag.strip().lower() sql += " AND (tags = ? OR tags LIKE ? OR tags LIKE ? OR tags LIKE ?)" params.extend([ tag_norm, f"{tag_norm},%", f"%,{tag_norm}", f"%,{tag_norm},%", ]) if desde: sql += " AND (fecha_doc IS NOT NULL AND fecha_doc >= ?)" params.append(desde) if hasta: sql += " AND (fecha_doc IS NOT NULL AND fecha_doc <= ?)" params.append(hasta) sql += " ORDER BY created_at DESC, id DESC" rows = conn.execute(sql, params).fetchall() return [_row_to_dict(r) for r in rows] def get_documento( conn: sqlite3.Connection, codigo: str, doc_id: int, ) -> Optional[dict]: """Devuelve el doc por id, solo si pertenece a la obra codigo.""" obra_id = _obra_id(conn, codigo) if obra_id is None: return None row = conn.execute( "SELECT " + ", ".join(_DOC_FIELDS) + " FROM documentos " "WHERE id = ? AND obra_id = ?", (int(doc_id), obra_id), ).fetchone() return _row_to_dict(row) if row else None def delete_documento( conn: sqlite3.Connection, codigo: str, doc_id: int, ) -> dict: """Elimina fila documentos. Devuelve `{ok, blob_path}` para que el caller borre el archivo.""" doc = get_documento(conn, codigo, doc_id) if doc is None: return {"ok": False, "errors": ["documento no encontrado"], "status": 404} conn.execute("DELETE FROM documentos WHERE id = ?", (int(doc_id),)) conn.commit() return {"ok": True, "blob_path": doc["blob_path"], "status": 204}