""" ERGON - Seed de la obra fixture "Obra Demo (DEMO)". Lee metadata + rubros + subcontratistas desde TDG1_N2_Dashboard_DG.xlsx y sintetiza data demo coherente (avance, presupuesto, asistencia, contratos) para que la API devuelva un caso completo sin violar Valor 1: - La obra queda marcada es_demo=1 - El nombre visible es "Obra Demo (DEMO)" - Los datos son ficticios y coherentes entre si Idempotente: si la obra TDG1 ya existe con es_demo=1, borra su data dependiente y la regenera. Obras reales (es_demo=0) nunca se tocan. Uso: python seed_torres_dg_demo.py """ from __future__ import annotations import random import sqlite3 import sys from dataclasses import dataclass from datetime import date, timedelta from pathlib import Path from typing import Iterable try: from openpyxl import load_workbook except ImportError: print("openpyxl no instalado. pip install openpyxl", file=sys.stderr) sys.exit(1) if sys.stdout.encoding and sys.stdout.encoding.lower() != "utf-8": sys.stdout.reconfigure(encoding="utf-8") HERE = Path(__file__).parent DB_PATH = HERE / "ergon.db" TDG1_PATH = HERE.parent / "TDG1_N2_Dashboard_DG.xlsx" DEMO_CODIGO = "TDG1" DEMO_NOMBRE_UI = "Obra Demo (DEMO)" # --------------------------------------------------------------------------- # Lectura de TDG1 # --------------------------------------------------------------------------- @dataclass class ObraMeta: codigo: str nombre: str cliente: str ubicacion: str tipo: str moneda: str presupuesto_total: float fecha_inicio: date fecha_fin_plan: date fecha_corte_actual: date umbral_desvio_pct: float umbral_critico_pct: float dg_nivel_servicio: int dg_fiscalizador: str dg_responsable: str dg_version_template: str rubros: list[tuple[int, str, float]] # (orden, nombre, peso_pct) subcontratistas: list[tuple[int, str]] # (orden, nombre) def _as_date(v) -> date: if hasattr(v, "date"): return v.date() if isinstance(v, date): return v raise TypeError(f"no se pudo convertir {v!r} a date") def _map_moneda(raw: str) -> str: """Normaliza etiqueta visual a codigo ISO.""" s = (raw or "").strip().lower().rstrip(".") if s in ("gs", "guarani", "guaranies", "pyg"): return "PYG" if s in ("usd", "us$", "dolar", "dolares"): return "USD" if s in ("eur", "euro", "euros"): return "EUR" return "PYG" def read_tdg1(path: Path) -> ObraMeta: wb = load_workbook(path, data_only=True) # Named ranges: lectura defensiva def nv(name: str): dn = wb.defined_names[name] sheet, coord = next(iter(dn.destinations)) return wb[sheet][coord].value def nrange(name: str) -> list: dn = wb.defined_names[name] sheet, coord = next(iter(dn.destinations)) out = [] for row in wb[sheet][coord]: for c in row: out.append(c.value) return out lista_rubros = nrange("lista_rubros") pesos_rubros = nrange("pesos_rubros") lista_subs = nrange("lista_subcontratistas") # peso_pct: en TDG1 vienen como fraccion 0..1; la DB guarda 0..100 raw = [] for i, (nombre, peso) in enumerate(zip(lista_rubros, pesos_rubros), start=1): if nombre is None or peso is None: continue raw.append((i, str(nombre).strip(), float(peso) * 100.0)) # Normalizacion a 100% exacto con reporte de discrepancia. # TDG1/FCAT1 contiene pesos que suman 101% por redondeo en el template. # Para que los calculos EVM sean correctos, reescalamos. suma_raw = sum(p for _, _, p in raw) if raw and abs(suma_raw - 100.0) > 1e-6: print(f" [!] pesos_rubros en Excel suman {suma_raw:.4f}% — normalizando a 100%") print(f" factor = {100.0/suma_raw:.6f} (reportar a Ruben para corregir template)") factor = 100.0 / suma_raw rubros = [(i, n, p * factor) for i, n, p in raw] else: rubros = raw subs = [] for i, nombre in enumerate(lista_subs, start=1): if nombre is None: continue subs.append((i, str(nombre).strip())) return ObraMeta( codigo=DEMO_CODIGO, nombre=DEMO_NOMBRE_UI, cliente=str(nv("proj_cliente") or "Grupo DG Desarrollo"), ubicacion=str(nv("proj_ubicacion") or "Asuncion, Paraguay"), tipo=str(nv("proj_tipo") or "Torre residencial"), moneda=_map_moneda(str(nv("proj_moneda") or "Gs.")), presupuesto_total=float(nv("proj_presupuesto")), fecha_inicio=_as_date(nv("proj_inicio")), fecha_fin_plan=_as_date(nv("proj_fin_plan")), fecha_corte_actual=_as_date(nv("proj_corte")), umbral_desvio_pct=float(nv("umbral_desvio")), umbral_critico_pct=float(nv("umbral_critico")), dg_nivel_servicio=int(nv("dg_nivel")), dg_fiscalizador=str(nv("dg_fiscalizador") or ""), dg_responsable=str(nv("dg_responsable") or ""), dg_version_template=str(nv("dg_version") or "v1.0"), rubros=rubros, subcontratistas=subs, ) # --------------------------------------------------------------------------- # Generacion de data sintetica coherente # --------------------------------------------------------------------------- def months_between(start: date, end: date) -> Iterable[date]: """Genera primer dia de cada mes entre start y end, inclusive.""" d = date(start.year, start.month, 1) while d <= end: yield d if d.month == 12: d = date(d.year + 1, 1, 1) else: d = date(d.year, d.month + 1, 1) def synth_avance(rubros, meses, corte) -> list[tuple]: """Avance acumulado plan vs real por (rubro, mes). Plan: curva S simple sobre meses totales. Real: plan con +-3pp de ruido. """ rng = random.Random(42) total_meses = len(meses) out = [] for orden, nombre, peso in rubros: # Rubros empiezan en mes distinto segun su naturaleza (proxy: por orden) start_offset = min(int((orden - 1) * 0.15 * total_meses / len(rubros)), total_meses // 3) duration_months = max(6, total_meses - start_offset - rng.randint(0, 2)) for i, mes in enumerate(meses): if i < start_offset: plan = 0.0 real = None else: pos = min((i - start_offset) / max(1, duration_months - 1), 1.0) # Curva S simplificada: 3*pos^2 - 2*pos^3 ajustado a 0..100 plan = max(0.0, min(100.0, (3 * pos * pos - 2 * pos * pos * pos) * 100.0)) if mes <= corte: noise = rng.uniform(-3.0, 3.0) real = max(0.0, min(100.0, plan + noise)) # Algunos rubros con desvio mayor para generar alertas if orden in (3, 10, 18): # Albanileria, Inst. Electrica, Pisos real = max(0.0, real - rng.uniform(3.0, 10.0)) else: real = None out.append((orden, mes.isoformat()[:7], plan, real)) return out def synth_presupuesto(rubros, meses, corte, presupuesto_total) -> list[tuple]: """Presupuesto mensual por rubro: plan = distribucion uniforme en duracion, real = plan * (1 + desvio_aleatorio). """ rng = random.Random(17) total_meses = len(meses) out = [] for orden, nombre, peso in rubros: presup_rubro = presupuesto_total * (peso / 100.0) start_offset = min(int((orden - 1) * 0.15 * total_meses / len(rubros)), total_meses // 3) duration = max(6, total_meses - start_offset - rng.randint(0, 2)) plan_mes = presup_rubro / duration for i, mes in enumerate(meses): if i < start_offset or i >= start_offset + duration: plan = 0.0 real = 0.0 if mes <= corte else None else: plan = plan_mes if mes <= corte: real = plan * (1.0 + rng.uniform(-0.08, 0.12)) else: real = None out.append((orden, mes.isoformat()[:7], plan, real)) return out def synth_asistencia(corte: date, dias: int = 60) -> list[tuple]: """Asistencia diaria: 40-55 personas en dias habiles, 0-10 en fin de semana, con algunos dias de lluvia que reducen personal. """ rng = random.Random(7) out = [] for i in range(dias): d = corte - timedelta(days=dias - 1 - i) finde = d.weekday() >= 5 lluvia = rng.uniform(0, 25) if rng.random() < 0.18 else 0.0 if finde: base = rng.randint(0, 10) else: base = rng.randint(40, 55) # Lluvia fuerte reduce personal if lluvia > 15: base = max(0, int(base * 0.3)) elif lluvia > 5: base = max(0, int(base * 0.7)) dia_perdido = 1 if lluvia > 15 else 0 notas = None if lluvia > 15: notas = f"Lluvia fuerte {lluvia:.1f} mm - obra detenida" out.append((d.isoformat(), base, round(lluvia, 1), dia_perdido, notas)) return out def synth_contratos(corte: date) -> list[tuple]: """Set minimo coherente de contratos con estados variados. sub_idx apunta al campo 'orden' del subcontratista en TDG1: 5 = Carlos Benitez (Albanileria) 7 = Jose Aranda (Rev. Exterior) 11 = GyC Instalaciones """ base_contratos = [ ("Contrato Principal", "Acuerdo marco cliente-DG", "FIRMADO", None), ("Permiso Municipal", "Permiso de obra Municipalidad de Asuncion", "FIRMADO", None), ("Planos Aprobados", "Planos definitivos visados", "FIRMADO", None), ("Proyecto Ejecutivo Estructura", "Revision final estructura", "OBSERVADO", None), ("Contrato Albanileria", "Subcontrato mano de obra albanileria", "PENDIENTE", 5), ("Contrato Instalaciones", "Subcontrato GyC Instalaciones", "FIRMADO", 11), ("Orden de Trabajo Fachada", "OT revestimiento exterior", "PENDIENTE", 7), ] out = [] for i, (tipo, desc, estado, sub_idx) in enumerate(base_contratos): fecha_emision = corte - timedelta(days=180 + i * 20) if estado == "FIRMADO": fecha_firma = fecha_emision + timedelta(days=15 + i * 3) else: fecha_firma = None fecha_vence = fecha_emision + timedelta(days=365) monto = None if "Subcontrato" in desc: monto = 500_000_000 + i * 100_000_000 out.append((tipo, desc, fecha_emision.isoformat(), fecha_firma.isoformat() if fecha_firma else None, fecha_vence.isoformat(), estado, monto, sub_idx)) return out # --------------------------------------------------------------------------- # Insercion en DB # --------------------------------------------------------------------------- def seed(conn: sqlite3.Connection, meta: ObraMeta) -> int: cur = conn.cursor() cur.execute("SELECT id, es_demo FROM obras WHERE codigo = ?", (meta.codigo,)) row = cur.fetchone() if row: obra_id, es_demo = row if not es_demo: print(f"ABORT: obra {meta.codigo} existe con es_demo=0 (obra real). Seed no la toca.") return -1 print(f" obra demo {meta.codigo} ya existe (id={obra_id}) - regenerando data dependiente") cur.execute("DELETE FROM avance_mensual WHERE obra_id = ?", (obra_id,)) cur.execute("DELETE FROM presupuesto_mensual WHERE obra_id = ?", (obra_id,)) cur.execute("DELETE FROM asistencia_diaria WHERE obra_id = ?", (obra_id,)) cur.execute("DELETE FROM contratos WHERE obra_id = ?", (obra_id,)) cur.execute("DELETE FROM materiales WHERE obra_id = ?", (obra_id,)) cur.execute("DELETE FROM subcontratistas WHERE obra_id = ?", (obra_id,)) cur.execute("DELETE FROM rubros_obra WHERE obra_id = ?", (obra_id,)) 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=?, updated_at=datetime('now') WHERE id=?""", (meta.nombre, meta.cliente, meta.ubicacion, meta.tipo, meta.moneda, meta.presupuesto_total, meta.fecha_inicio.isoformat(), meta.fecha_fin_plan.isoformat(), meta.fecha_corte_actual.isoformat(), meta.umbral_desvio_pct, meta.umbral_critico_pct, meta.dg_nivel_servicio, meta.dg_fiscalizador, meta.dg_responsable, meta.dg_version_template, obra_id)) else: print(f" insertando obra {meta.codigo}") 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 (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,1)""", (meta.codigo, meta.nombre, meta.cliente, meta.ubicacion, meta.tipo, meta.moneda, meta.presupuesto_total, meta.fecha_inicio.isoformat(), meta.fecha_fin_plan.isoformat(), meta.fecha_corte_actual.isoformat(), meta.umbral_desvio_pct, meta.umbral_critico_pct, meta.dg_nivel_servicio, meta.dg_fiscalizador, meta.dg_responsable, meta.dg_version_template)) obra_id = cur.lastrowid # RUBROS rubro_id_by_orden: dict[int, int] = {} for orden, nombre, peso in meta.rubros: cur.execute("""INSERT INTO rubros_obra (obra_id, orden, nombre, peso_pct) VALUES (?,?,?,?)""", (obra_id, orden, nombre, peso)) rubro_id_by_orden[orden] = cur.lastrowid print(f" rubros insertados: {len(meta.rubros)}") # SUBCONTRATISTAS sub_id_by_orden: dict[int, int] = {} for orden, nombre in meta.subcontratistas: cur.execute("""INSERT INTO subcontratistas (obra_id, orden, nombre) VALUES (?,?,?)""", (obra_id, orden, nombre)) sub_id_by_orden[orden] = cur.lastrowid print(f" subcontratistas insertados: {len(meta.subcontratistas)}") # AVANCE y PRESUPUESTO: meses desde inicio hasta min(corte+6m, fin_plan) horizonte = min(meta.fecha_fin_plan, date(meta.fecha_corte_actual.year + (meta.fecha_corte_actual.month + 5) // 12, ((meta.fecha_corte_actual.month + 5) % 12) + 1, 1)) meses = list(months_between(meta.fecha_inicio, horizonte)) avance = synth_avance(meta.rubros, meses, meta.fecha_corte_actual) for orden, mes, plan, real in avance: cur.execute("""INSERT INTO avance_mensual (obra_id, rubro_id, mes, plan_acum_pct, real_acum_pct) VALUES (?,?,?,?,?)""", (obra_id, rubro_id_by_orden[orden], mes, plan, real)) print(f" avance_mensual filas: {len(avance)}") presu = synth_presupuesto(meta.rubros, meses, meta.fecha_corte_actual, meta.presupuesto_total) for orden, mes, plan, real in presu: cur.execute("""INSERT INTO presupuesto_mensual (obra_id, rubro_id, mes, plan, real) VALUES (?,?,?,?,?)""", (obra_id, rubro_id_by_orden[orden], mes, plan, real)) print(f" presupuesto_mensual filas: {len(presu)}") # ASISTENCIA asist = synth_asistencia(meta.fecha_corte_actual, dias=60) for fecha, total, lluvia, perdido, notas in asist: cur.execute("""INSERT INTO asistencia_diaria (obra_id, fecha, total_personal, lluvia_mm, dia_perdido, notas) VALUES (?,?,?,?,?,?)""", (obra_id, fecha, total, lluvia, perdido, notas)) print(f" asistencia_diaria filas: {len(asist)}") # CONTRATOS contratos = synth_contratos(meta.fecha_corte_actual) for tipo, desc, emision, firma, vence, estado, monto, sub_idx in contratos: cid = sub_id_by_orden.get(sub_idx) if sub_idx else None cur.execute("""INSERT INTO contratos (obra_id, contratista_id, tipo, descripcion, fecha_emision, fecha_firma, fecha_vence, estado, monto) VALUES (?,?,?,?,?,?,?,?,?)""", (obra_id, cid, tipo, desc, emision, firma, vence, estado, monto)) print(f" contratos filas: {len(contratos)}") # MATERIALES: set minimo materiales_base = [ ("Cemento CPC40 - 50kg", "bolsa", 12000, 9800), ("Hierro 12mm", "kg", 85000, 71000), ("Ladrillo comun", "un", 180000, 142000), ("Arena gruesa", "m3", 450, 380), ] for item, unidad, recib, cons in materiales_base: cur.execute("""INSERT INTO materiales (obra_id, item, unidad, recibido, consumido) VALUES (?,?,?,?,?)""", (obra_id, item, unidad, recib, cons)) print(f" materiales filas: {len(materiales_base)}") conn.commit() return obra_id def main() -> int: if not DB_PATH.exists(): print(f"ERROR: {DB_PATH} no existe. Corre init_db.py primero.", file=sys.stderr) return 2 if not TDG1_PATH.exists(): print(f"ERROR: {TDG1_PATH} no existe.", file=sys.stderr) return 2 print(f"leyendo metadata desde {TDG1_PATH.name}") meta = read_tdg1(TDG1_PATH) print(f" obra: {meta.codigo} '{meta.nombre}' ({meta.cliente})") print(f" periodo: {meta.fecha_inicio} -> {meta.fecha_fin_plan} (corte {meta.fecha_corte_actual})") print(f" presupuesto: {meta.presupuesto_total:,.0f} {meta.moneda}") print(f" rubros: {len(meta.rubros)} subcontratistas: {len(meta.subcontratistas)}") print(f"\naplicando seed sobre {DB_PATH.name}") conn = sqlite3.connect(str(DB_PATH)) try: conn.execute("PRAGMA foreign_keys = ON") obra_id = seed(conn, meta) if obra_id < 0: return 1 finally: conn.close() print("\nOK. Seed completado.") return 0 if __name__ == "__main__": sys.exit(main())