""" ERGON F10 - Cargador DB -> cfg para generate_dg_civil.py Convierte los datos de una obra persistida en ergon.db al formato dict `cfg` que consume `generate_dg_civil.generate()`. Mapea: DB (api_obras) cfg (generate_dg_civil) ------------------------------ ----------------------------------- obras.codigo codigo obras.nombre proyecto obras.cliente cliente obras.ubicacion ubicacion obras.tipo tipo_obra obras.moneda (PYG/USD/EUR) moneda ("Gs."/"USD"/"EUR") obras.presupuesto_total presupuesto obras.fecha_inicio (YYYY-MM-DD) inicio (date) obras.fecha_fin_plan fin_plan (date) obras.fecha_corte_actual corte (date) obras.umbral_desvio_pct umbral_desvio obras.umbral_critico_pct umbral_critico obras.dg_nivel_servicio nivel obras.dg_fiscalizador fiscalizador obras.dg_responsable responsable_dg rubros_obra.peso_pct (0-100) rubros[i][1] (0-1 decimal) subcontratistas.nombre subcontratistas[i] avance_mensual + presupuesto avance_mensual {idx:(plan,real,costo_acum)} Uso: from db_loader import load_cfg_from_db cfg = load_cfg_from_db("TDG1") """ from __future__ import annotations import os import sys from datetime import date from typing import Any _MONEDA_DB_TO_CFG = {"PYG": "Gs.", "USD": "USD", "EUR": "EUR"} class ObraNotFound(Exception): """La obra con el codigo dado no existe en la DB.""" def _import_api_obras(): """Import diferido para que este modulo funcione como CLI y como import.""" here = os.path.dirname(os.path.abspath(__file__)) db_dir = os.path.join(here, "db") if db_dir not in sys.path: sys.path.insert(0, db_dir) import api_obras # type: ignore return api_obras def _parse_iso_date(s: str) -> date: # fecha_inicio, fecha_fin_plan y fecha_corte_actual tienen CHECK del schema # y el flujo F7 siempre los persiste en formato YYYY-MM-DD. return date.fromisoformat(s[:10]) def _build_avance_mensual(avance_agg: list[dict], presu_agg: list[dict]) -> dict[int, tuple[float, float, float]]: """Une avance (plan/real ponderado %) + costo acumulado por mes. Devuelve {idx_mes_1based: (plan_acum_dec, real_acum_dec, costo_real_acum)}. Reglas: - plan/real vienen de la DB como 0-100 (%), se pasan a 0-1 decimal. - costo_real_acum = sum(presu_agg[mes].real_mes hasta mes actual). Si un mes tiene real=NULL, se toma 0 para ese mes (no rompe el acum). - Se cubren todos los meses que aparecen en cualquiera de las dos series, ordenados cronologicamente por la clave "YYYY-MM". idx 1-based. """ avance_by_mes = {r["mes"]: r for r in (avance_agg or [])} presu_by_mes = {r["mes"]: r for r in (presu_agg or [])} meses = sorted(set(avance_by_mes.keys()) | set(presu_by_mes.keys())) out: dict[int, tuple[float, float, float]] = {} costo_acum = 0.0 for idx, mes in enumerate(meses, start=1): plan_pct = float(avance_by_mes.get(mes, {}).get("plan_global") or 0.0) real_pct = float(avance_by_mes.get(mes, {}).get("real_global") or 0.0) real_mes = float(presu_by_mes.get(mes, {}).get("real_mes") or 0.0) costo_acum += real_mes out[idx] = (round(plan_pct / 100.0, 6), round(real_pct / 100.0, 6), round(costo_acum, 2)) return out def load_cfg_from_db(codigo: str, db_path: str | None = None) -> dict[str, Any]: """Construye el cfg esperado por `generate_dg_civil.generate()` para una obra. `db_path`: ruta opcional al archivo SQLite. Por defecto api_obras apunta a `/db/ergon.db`. El arg existe para tests o backups. """ api = _import_api_obras() # Permitir override de DB_PATH via arg (para tests/backups). if db_path: api.DB_PATH = db_path # type: ignore[assignment] conn = api.get_conn() try: obra = api.get_obra(conn, codigo) if obra is None: raise ObraNotFound(f"Obra con codigo '{codigo}' no existe en la DB.") rubros_db = api.get_rubros(conn, codigo) if not rubros_db: raise ValueError( f"Obra '{codigo}' no tiene rubros cargados. Cargue PARAMETROS antes de exportar." ) subs_db = api.get_subcontratistas(conn, codigo) avance = api.get_avance(conn, codigo) presu = api.get_presupuesto(conn, codigo) moneda_cfg = _MONEDA_DB_TO_CFG.get( (obra.get("moneda") or "PYG").upper(), "Gs." ) cfg: dict[str, Any] = { "codigo": obra["codigo"], "proyecto": obra["nombre"], "cliente": obra["cliente"], "ubicacion": obra.get("ubicacion") or "", "tipo_obra": obra.get("tipo") or "", "moneda": moneda_cfg, "presupuesto": float(obra["presupuesto_total"]), "inicio": _parse_iso_date(obra["fecha_inicio"]), "fin_plan": _parse_iso_date(obra["fecha_fin_plan"]), "corte": _parse_iso_date(obra["fecha_corte_actual"]), "umbral_desvio": float(obra["umbral_desvio_pct"]), "umbral_critico": float(obra["umbral_critico_pct"]), "nivel": int(obra["dg_nivel_servicio"]), "fiscalizador": obra.get("dg_fiscalizador") or "", "responsable_dg": obra.get("dg_responsable") or "", # Subcontratistas: lista plana de nombres, ordenados por campo orden. "subcontratistas": [s["nombre"] for s in subs_db], # Rubros: tuplas (nombre, peso_decimal 0-1) ordenadas por orden. "rubros": [(r["nombre"], float(r["peso_pct"]) / 100.0) for r in rubros_db], # Avance mensual consolidado para CURVA_S/DASHBOARD. "avance_mensual": _build_avance_mensual( avance.get("agregado") or [], presu.get("agregado") or [], ), # Metadatos para naming del archivo de salida. "_obra_id": obra["id"], "_es_demo": bool(obra.get("es_demo")), } return cfg finally: conn.close() if __name__ == "__main__": # CLI de inspeccion: python db_loader.py TDG1 import json if len(sys.argv) != 2: print("Uso: python db_loader.py CODIGO_OBRA", file=sys.stderr) sys.exit(2) try: cfg = load_cfg_from_db(sys.argv[1]) except ObraNotFound as e: print(f"[ERROR] {e}", file=sys.stderr) sys.exit(1) # date no es JSON-serializable; convertir. preview = {k: (v.isoformat() if isinstance(v, date) else v) for k, v in cfg.items()} preview["rubros"] = f"<{len(cfg['rubros'])} rubros>" preview["subcontratistas"] = f"<{len(cfg['subcontratistas'])} subs>" preview["avance_mensual"] = f"<{len(cfg['avance_mensual'])} meses>" print(json.dumps(preview, indent=2, ensure_ascii=False))