""" Carga historica Nomina + P&G en PostgreSQL. Rango por defecto: enero 2024 a marzo 2026. Uso: python3 cargar_historico_nomina_pyg.py --init-db python3 cargar_historico_nomina_pyg.py --fuente ambas python3 cargar_historico_nomina_pyg.py --fuente nomina --desde 2024-01 --hasta 2024-03 python3 cargar_historico_nomina_pyg.py --mes-anterior --fuente ambas El dashboard sigue usando /proxy/nomina/reporte y /proxy/pyg/reporte; servidor.py lee esta tabla para periodos historicos y cae a la API para meses vivos. """ import argparse import hashlib import json import os import time import urllib.error import urllib.request from datetime import date import psycopg2 import psycopg2.extras from app_config import ( API_BASE, NOMINA_API_KEY, PYG_API_KEY, PYG_URL, db_config, require_secrets, ssl_context, ) DIR = os.path.dirname(os.path.abspath(__file__)) DB_CONFIG = db_config() API_KEY = NOMINA_API_KEY PYG_KEY = PYG_API_KEY SSL_CTX = ssl_context() PYG_CONCEPTOS = [ ("sueldos_y_salarios_del_personal", "Sueldos y Salarios", "Salarios", "Salario Mes", True), ("horas_extras_y_festivos", "Horas Extras y Festivos", "Salarios", "Horas Extras y Recargos", True), ("auxilio_de_transporte", "Auxilio de Transporte", "Salarios", "Subsidio de Transporte", True), ("auxilios", "Auxilios", "Salarios", "Auxilios", True), ("incapacidades", "Incapacidades", "Salarios", "Incapacidades", True), ("prestaciones_sociales_y_vacaciones", "Prestaciones Sociales y Vacaciones", "Provisiones", "Prestaciones Sociales y Vacaciones", True), ("primas_extralegales", "Primas Extralegales", "Provisiones", "Primas Extralegales", True), ("bonificaciones", "Bonificaciones", "Salarios", "Bonificaciones", True), ("indemnizaciones_laborales", "Indemnizaciones Laborales", "Indemnizaciones", "Indemnizaciones Laborales", True), ("temporales", "Temporales", "Personal Externo", "Temporales", False), ("honorarios_laborales", "Honorarios Laborales", "Personal Externo", "Honorarios", False), ("comisiones", "Comisiones Personal", "Personal Externo", "Honorarios", False), ("dotacion_y_suministro_a_trabajadores", "Dotacion y Suministro", "Otros Gastos Personal", "Dotacion", False), ("dotación_y_suministro_a_trabajadores", "Dotacion y Suministro", "Otros Gastos Personal", "Dotacion", False), ("comida_empleados", "Comida Empleados", "Otros Gastos Personal", "Bienestar", False), ("transporte_de_personal", "Transporte de Personal", "Otros Gastos Personal", "Bienestar", False), ("capacitacion_al_personal", "Capacitacion al Personal", "Otros Gastos Personal", "Bienestar", False), ("capacitación_al_personal", "Capacitacion al Personal", "Otros Gastos Personal", "Bienestar", False), ("gastos_medicos_y_medicamentos", "Gastos Medicos y Medicamentos", "Otros Gastos Personal", "Gastos Medicos", False), ("salud_ocupacional_y_examenes_d", "Salud Ocupacional y Examenes", "Otros Gastos Personal", "Gastos Medicos", False), ] def stable_json(obj): return json.dumps(obj, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str) def make_hash(*parts): return hashlib.sha256("|".join(str(p) for p in parts).encode("utf-8")).hexdigest() def months_between(desde, hasta): y, m = [int(x) for x in desde.split("-")] end_y, end_m = [int(x) for x in hasta.split("-")] while (y, m) <= (end_y, end_m): yield y, m, f"{y:04d}-{m:02d}" m += 1 if m == 13: y += 1 m = 1 def prev_month_period(): today = date.today() if today.month == 1: return f"{today.year - 1:04d}-12" return f"{today.year:04d}-{today.month - 1:02d}" def connect(): return psycopg2.connect(**DB_CONFIG) def init_db(conn): with open(os.path.join(DIR, "historico_nomina_pyg.sql"), encoding="utf-8") as fh: sql = fh.read() with conn.cursor() as cur: cur.execute(sql) conn.commit() def ensure_db_ready(conn): required_tables = ( "historico_nomina_pyg", "historico_nomina_pyg_control", ) with conn.cursor() as cur: cur.execute( """ SELECT table_name FROM information_schema.tables WHERE table_schema = current_schema() AND table_name = ANY(%s) """, (list(required_tables),), ) existing = {row[0] for row in cur.fetchall()} missing = sorted(set(required_tables) - existing) if missing: raise RuntimeError( "Faltan tablas requeridas: " + ", ".join(missing) + ". Ejecutar --init-db una vez con el usuario propietario de la base." ) def post_json(url, payload, api_key, timeout=90): body = json.dumps(payload).encode("utf-8") req = urllib.request.Request( url, data=body, headers={"X-API-Key": api_key, "Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(req, timeout=timeout, context=SSL_CTX) as resp: return json.loads(resp.read()) except urllib.error.HTTPError as exc: # El cuerpo del 4xx trae el motivo real en "detail". Sin esto, en el log # se ven identicos dos casos muy distintos: # - "Cannot open database ... login failed (4060)" -> base de nomina # caida del lado del proveedor: HAY QUE REPORTARLO. # - "No hay periodos cerrados para YYYY-MM" -> la nomina de ese hotel # aun no se ha cerrado ese mes: se resuelve solo en el reintento # dia 5/7. (Ver PENDIENTES: PSOL vs DORA.) try: cuerpo = exc.read().decode("utf-8", errors="replace") except Exception: cuerpo = "" detalle = cuerpo try: detalle = json.loads(cuerpo).get("detail", cuerpo) except Exception: pass detalle = (detalle or "").strip() raise RuntimeError( f"HTTP {exc.code} {exc.reason}" + (f" — {detalle}" if detalle else "") ) from exc def fetch_hoteles(conn): with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: cur.execute(""" SELECT DISTINCT ON (h.nombre_personalizado, e.nit, h.bu_codigo) h.nombre_personalizado AS label, h.razon_social, e.nit, h.hotel_name_api, h.bu_codigo, h.bu_nombre, e.pyg_company_id FROM hotel_unificacion h JOIN empresas e ON h.razon_social = e.codigo WHERE e.nit IS NOT NULL AND e.nit <> '' AND h.hotel_name_api IS NOT NULL AND h.hotel_name_api <> '' AND h.nombre_personalizado NOT IN ('No mostrar') ORDER BY h.nombre_personalizado, e.nit, h.bu_codigo """) return [dict(r) for r in cur.fetchall()] def upsert_control(conn, fuente, ano, mes, nit=None, company_id=None, bu_codigo=None, estado="iniciado", filas=0, error=None): periodo = f"{ano:04d}-{mes:02d}" with conn.cursor() as cur: cur.execute(""" INSERT INTO historico_nomina_pyg_control (fuente, ano, mes, periodo, nit, company_id, bu_codigo, estado, filas, error, terminado_en) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,CASE WHEN %s IN ('ok','error') THEN NOW() ELSE NULL END) ON CONFLICT (fuente, ano, mes, COALESCE(nit, ''), COALESCE(company_id, -1), COALESCE(bu_codigo, '')) DO UPDATE SET estado = EXCLUDED.estado, filas = EXCLUDED.filas, error = EXCLUDED.error, terminado_en = EXCLUDED.terminado_en """, (fuente, ano, mes, periodo, nit, company_id, bu_codigo, estado, filas, error, estado)) conn.commit() def insert_rows(conn, rows): if not rows: return 0 cols = [ "fuente", "registro_tipo", "ano", "mes", "periodo", "nit", "razon_social", "hotel", "bu_codigo", "bu_nombre", "company_id", "concepto_codigo", "concepto_nombre", "grupo", "subgrupo", "empleado_codigo", "empleado_nombre", "cargo", "centro_costo", "valor", "valor_ly", "cantidad", "raw_data", "hash_registro", ] values = [[r.get(c) for c in cols] for r in rows] with conn.cursor() as cur: psycopg2.extras.execute_values( cur, f""" INSERT INTO historico_nomina_pyg ({','.join(cols)}) VALUES %s ON CONFLICT (hash_registro) DO UPDATE SET valor = EXCLUDED.valor, valor_ly = EXCLUDED.valor_ly, cantidad = EXCLUDED.cantidad, raw_data = EXCLUDED.raw_data, cargado_en = NOW() """, values, template=None, page_size=500, ) conn.commit() return len(rows) def clear_slice(conn, fuente, ano, mes, nit=None, company_id=None, bu_codigo=None): where = ["fuente=%s", "ano=%s", "mes=%s"] params = [fuente, ano, mes] if nit is not None: where.append("nit=%s") params.append(nit) if company_id is not None: where.append("company_id=%s") params.append(company_id) if bu_codigo is not None: where.append("LOWER(COALESCE(bu_codigo,''))=%s") params.append(bu_codigo.lower()) with conn.cursor() as cur: cur.execute(f"DELETE FROM historico_nomina_pyg WHERE {' AND '.join(where)}", params) conn.commit() def nomina_rows(raw, hotel_meta, ano, mes, periodo): nit = hotel_meta["nit"] rows = [{ "fuente": "nomina", "registro_tipo": "respuesta", "ano": ano, "mes": mes, "periodo": periodo, "nit": nit, "raw_data": psycopg2.extras.Json(raw), "hash_registro": make_hash("nomina", "respuesta", periodo, nit), }] resumen = raw.get("resumen") or {} for key, val in resumen.items(): if isinstance(val, (int, float)): rows.append({ "fuente": "nomina", "registro_tipo": "resumen", "ano": ano, "mes": mes, "periodo": periodo, "nit": nit, "concepto_codigo": key, "concepto_nombre": key, "valor": val, "raw_data": psycopg2.extras.Json({"campo": key, "valor": val}), "hash_registro": make_hash("nomina", "resumen", periodo, nit, key), }) for c in raw.get("conceptos") or []: val = (c.get("devengos") or 0) - (c.get("deducciones") or 0) + (c.get("aportes_patronales") or 0) rows.append({ "fuente": "nomina", "registro_tipo": "concepto", "ano": ano, "mes": mes, "periodo": periodo, "nit": nit, "concepto_codigo": c.get("codigo"), "concepto_nombre": c.get("nombre"), "valor": val, "cantidad": c.get("num_empleados"), "raw_data": psycopg2.extras.Json(c), "hash_registro": make_hash("nomina", "concepto", periodo, nit, c.get("codigo"), c.get("nombre")), }) for e in raw.get("empleados") or []: rows.append({ "fuente": "nomina", "registro_tipo": "empleado", "ano": ano, "mes": mes, "periodo": periodo, "nit": nit, "hotel": e.get("hotel") or e.get("hotel_name"), "bu_codigo": e.get("bu_codigo"), "empleado_codigo": e.get("codigo") or e.get("codigo_empleado") or e.get("cedula"), "empleado_nombre": e.get("nombre") or e.get("nombre_completo"), "cargo": e.get("cargo"), "centro_costo": e.get("centro_costo"), "valor": e.get("total_devengos"), "raw_data": psycopg2.extras.Json(e), "hash_registro": make_hash("nomina", "empleado", periodo, nit, e.get("codigo") or e.get("codigo_empleado") or e.get("cedula"), e.get("nombre") or e.get("nombre_completo")), }) return rows def cargar_nomina(conn, hoteles, desde, hasta, pausa): nits = {} for h in hoteles: nits.setdefault(h["nit"], h) for ano, mes, periodo in months_between(desde, hasta): for nit, meta in sorted(nits.items()): upsert_control(conn, "nomina", ano, mes, nit=nit) try: payload = {"hotel_nit": nit, "ano": ano, "mes": mes, "incluir_empleados": True, "ss_vista": "ambos"} raw = post_json(f"{API_BASE}/nomina/reporte", payload, API_KEY) clear_slice(conn, "nomina", ano, mes, nit=nit) filas = insert_rows(conn, nomina_rows(raw, meta, ano, mes, periodo)) upsert_control(conn, "nomina", ano, mes, nit=nit, estado="ok", filas=filas) print(f"OK nomina {periodo} nit={nit} filas={filas}", flush=True) except Exception as exc: upsert_control(conn, "nomina", ano, mes, nit=nit, estado="error", error=str(exc)) print(f"ERROR nomina {periodo} nit={nit}: {exc}", flush=True) time.sleep(pausa) def flatten_tree(nodes, out): for node in nodes: node_id = node.get("id", "") monthly = node.get("monthly", {}) real = monthly.get("real", {}) ly_data = monthly.get("last_year", {}) val = first_month_value(real) ly = first_month_value(ly_data) out[node_id] = {"id": node_id, "label": node.get("label", ""), "val": val, "ly": ly} flatten_tree(node.get("children", []), out) def first_month_value(data): if isinstance(data, (int, float)): return data if not isinstance(data, dict): return 0.0 for val in data.values(): if isinstance(val, (int, float)): return val if isinstance(val, dict): return val.get("mes") or val.get("value") or 0.0 return 0.0 def pyg_rows(raw, company_id, bu_codigo, ano, mes, periodo): pnl = raw.get("pnl", {}) bu_key = (bu_codigo or "local").lower() flat = {} flatten_tree(pnl.get(bu_key, {}).get("tree", []), flat) rows = [{ "fuente": "pyg", "registro_tipo": "respuesta", "ano": ano, "mes": mes, "periodo": periodo, "company_id": company_id, "bu_codigo": bu_key, "raw_data": psycopg2.extras.Json(raw), "hash_registro": make_hash("pyg", "respuesta", periodo, company_id, bu_key), }] seen = set() for node_id, nombre, grupo, subgrupo, nomina_backed in PYG_CONCEPTOS: if node_id in seen or node_id not in flat: continue seen.add(node_id) alt = node_id.translate(str.maketrans("áéíóúÁÉÍÓÚ", "aeiouAEIOU")) seen.add(alt) entry = flat[node_id] row_raw = { "id": node_id, "nombre": nombre, "grupo": grupo, "subgrupo": subgrupo, "val": entry["val"], "ly": entry["ly"], "fuente": "pyg", "nomina_backed": nomina_backed, } rows.append({ "fuente": "pyg", "registro_tipo": "concepto", "ano": ano, "mes": mes, "periodo": periodo, "company_id": company_id, "bu_codigo": bu_key, "concepto_codigo": node_id, "concepto_nombre": nombre, "grupo": grupo, "subgrupo": subgrupo, "valor": entry["val"], "valor_ly": entry["ly"], "raw_data": psycopg2.extras.Json(row_raw), "hash_registro": make_hash("pyg", "concepto", periodo, company_id, bu_key, node_id), }) return rows def cargar_pyg(conn, hoteles, desde, hasta, pausa): pairs = sorted({ (int(h["pyg_company_id"]), (h.get("bu_codigo") or "local").lower()) for h in hoteles if h.get("pyg_company_id") }) for ano, mes, periodo in months_between(desde, hasta): for company_id, bu_codigo in pairs: upsert_control(conn, "pyg", ano, mes, company_id=company_id, bu_codigo=bu_codigo) try: payload = { "company_id": company_id, "period": f"{ano}{mes:02d}", "scenario_id": None, "business_units": [bu_codigo], "include_kpis": False, } raw = post_json(PYG_URL, payload, PYG_KEY) clear_slice(conn, "pyg", ano, mes, company_id=company_id, bu_codigo=bu_codigo) filas = insert_rows(conn, pyg_rows(raw, company_id, bu_codigo, ano, mes, periodo)) upsert_control(conn, "pyg", ano, mes, company_id=company_id, bu_codigo=bu_codigo, estado="ok", filas=filas) print(f"OK pyg {periodo} company={company_id} bu={bu_codigo} filas={filas}", flush=True) except Exception as exc: upsert_control(conn, "pyg", ano, mes, company_id=company_id, bu_codigo=bu_codigo, estado="error", error=str(exc)) print(f"ERROR pyg {periodo} company={company_id} bu={bu_codigo}: {exc}", flush=True) time.sleep(pausa) def main(): parser = argparse.ArgumentParser() parser.add_argument("--desde", default="2024-01") parser.add_argument("--hasta", default="2026-03") parser.add_argument("--fuente", choices=["nomina", "pyg", "ambas"], default="ambas") parser.add_argument("--init-db", action="store_true") parser.add_argument("--mes-anterior", action="store_true", help="Carga solo el mes anterior a la fecha actual.") parser.add_argument("--dry-run", action="store_true", help="Muestra el periodo y fuentes que cargaria sin escribir en la base.") parser.add_argument("--pausa", type=float, default=0.2) args = parser.parse_args() required = [] if args.fuente in ("nomina", "ambas"): required.append("NOMINA_API_KEY") if args.fuente in ("pyg", "ambas"): required.append("PYG_API_KEY") if required and not args.dry_run and not args.init_db: require_secrets(*required) if args.mes_anterior: periodo = prev_month_period() args.desde = periodo args.hasta = periodo with connect() as conn: if args.init_db: init_db(conn) print("Tablas listas.", flush=True) return ensure_db_ready(conn) if args.dry_run: hoteles = fetch_hoteles(conn) nits = {h["nit"] for h in hoteles if h.get("nit")} pyg_pairs = { (int(h["pyg_company_id"]), (h.get("bu_codigo") or "local").lower()) for h in hoteles if h.get("pyg_company_id") } print("DRY RUN cargar_historico_nomina_pyg", flush=True) print(f" fuente: {args.fuente}", flush=True) print(f" desde: {args.desde}", flush=True) print(f" hasta: {args.hasta}", flush=True) print(f" nits nomina: {len(nits)}", flush=True) print(f" pares pyg: {len(pyg_pairs)}", flush=True) return hoteles = fetch_hoteles(conn) if args.fuente in ("nomina", "ambas"): cargar_nomina(conn, hoteles, args.desde, args.hasta, args.pausa) if args.fuente in ("pyg", "ambas"): cargar_pyg(conn, hoteles, args.desde, args.hasta, args.pausa) if __name__ == "__main__": main()