""" Servidor proxy local para Dashboard Talento Humano. Sirve el HTML y enruta /proxy/* hacia la API real (evita CORS). También expone /proxy/historico/* consultando PostgreSQL local. Uso: python3 servidor.py Luego abrir: http://localhost:8081 """ import http.server import socketserver import urllib.request import urllib.error import json import hmac import os import time import psycopg2 import psycopg2.extras import psycopg2.pool import unicodedata from urllib.parse import urlparse, parse_qs from datetime import date as date_type from app_config import ( API_BASE, NOMINA_API_KEY, PYG_API_KEY, PYG_URL, db_config, env_bool, require_secrets, ssl_context, ) def norm_name(s): """Normaliza nombre: quita tildes, uppercase, ordena palabras alfabéticamente. El orden permite comparar nombres con apellido-primero (API) vs nombre-primero (historico).""" nfkd = unicodedata.normalize('NFKD', s or '') ascii_s = ''.join(c for c in nfkd if not unicodedata.combining(c)) return ' '.join(sorted(ascii_s.upper().split())) def razon_where(razon, col='razon_social'): """Devuelve (fragmento_sql, params) para filtrar por razon_social, admitiendo valores separados por coma (p.ej. 'SERV,SERVMYS,SERVMAN').""" razones = [r.strip() for r in razon.split(',') if r.strip()] if razon else [] if not razones: return '', [] if len(razones) == 1: return f"{col} = %s", [razones[0]] placeholders = ','.join(['%s'] * len(razones)) return f"{col} IN ({placeholders})", razones # Mismos patrones que PR_HOTEL_SERV_PATTERNS en dashboard-empleados.html: # empleados de Servihoteles (razon_social='SERV') cuyo centro_costo indica que # en realidad trabajan para uno de estos hoteles con NIT propio. Usado solo en # activos-count para que las tarjetas KPI de Análisis coincidan con el conteo # ya corregido en la tabla de Empleados/gráficos (que sí leen por centro_costo). SERV_HOTEL_PATTERNS = { # Excluidos por decisión del usuario (no son el mismo hotel pese al # nombre similar): 'GHS' NO incluye 'THE CLUB CTG' (venta de # multipropiedad en Cartagena) ni 'MYS' incluye 'HCARIBE'/'H CARIBE'. 'MAN': ['MANTILLA'], 'CDON': ['CDL'], 'CCAR': ['FARANDA COLLECTION CARTAGENA'], } SSL_CTX = ssl_context() PORT = int(os.getenv("PORT", "8080")) API_KEY = NOMINA_API_KEY PYG_KEY = PYG_API_KEY CORS_ALLOWED_ORIGINS = { origin.strip() for origin in os.getenv( "CORS_ALLOWED_ORIGINS", "http://localhost:8080,http://localhost:8081", ).split(",") if origin.strip() } ADMIN_API_KEY = os.getenv("ADMIN_API_KEY", "") REQUIRE_ADMIN_AUTH = env_bool("APP_REQUIRE_ADMIN_AUTH", True) DEBUG_ERRORS = env_bool("APP_DEBUG_ERRORS", False) # Mapping hotel_name_api → NIT para API de Nómina # HOTEL_NIT_MAP ya no se usa — los NITs se leen de empresas JOIN hotel_unificacion en BD DIR = os.path.dirname(os.path.abspath(__file__)) DB_CONFIG = db_config() # Pool de conexiones PostgreSQL — evita el overhead de crear una conexión nueva # por cada petición (ahorra ~100-200ms por request) DB_POOL = psycopg2.pool.ThreadedConnectionPool(minconn=2, maxconn=10, **DB_CONFIG) # Caché del último período cargado en historico_nomina_pyg — evita hardcodear # el corte caché-PG vs API-en-vivo. Se refresca cada 5 min (cierre_mes corre 1-2x/mes). _PERIODO_NOMINA_CACHE = {"value": None, "ts": 0.0} _PERIODO_NOMINA_TTL = 300 # segundos def _get_max_periodo_nomina(): """(ano, mes) del último período con datos en historico_nomina_pyg (fuente='nomina').""" now = time.time() if _PERIODO_NOMINA_CACHE["value"] and (now - _PERIODO_NOMINA_CACHE["ts"] < _PERIODO_NOMINA_TTL): return _PERIODO_NOMINA_CACHE["value"] conn = DB_POOL.getconn() try: cur = conn.cursor() cur.execute("SELECT MAX(periodo) FROM historico_nomina_pyg WHERE fuente='nomina'") row = cur.fetchone() periodo = row[0] if row and row[0] else "2024-01" finally: DB_POOL.putconn(conn) ano, mes = (int(p) for p in periodo.split('-')) _PERIODO_NOMINA_CACHE["value"] = (ano, mes) _PERIODO_NOMINA_CACHE["ts"] = now return (ano, mes) class ProxyHandler(http.server.SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=DIR, **kwargs) def _send_cors(self): origin = self.headers.get("Origin") if "*" in CORS_ALLOWED_ORIGINS: self.send_header("Access-Control-Allow-Origin", "*") elif origin and origin in CORS_ALLOWED_ORIGINS: self.send_header("Access-Control-Allow-Origin", origin) self.send_header("Vary", "Origin") def _is_admin_path(self): return ( self.path.startswith("/proxy/admin/") or self.path == "/proxy/historico/cecos-sync" ) def _require_admin_auth(self): if not self._is_admin_path() or not REQUIRE_ADMIN_AUTH: return True if not ADMIN_API_KEY: self._send_json(503, { "ok": False, "error": "ADMIN_API_KEY no configurado para endpoints administrativos", }) return False supplied = self.headers.get("X-Admin-Key", "") if not hmac.compare_digest(supplied, ADMIN_API_KEY): self._send_json(403, {"ok": False, "error": "No autorizado"}) return False return True def do_GET(self): if self.path == "/proxy/nomina/hoteles-nit": self._nomina_hoteles_nit() return elif self.path == "/proxy/nomina/conceptos-map": self._nomina_conceptos_map() return elif self.path.startswith("/proxy/stats/"): self._stats() return elif self.path.startswith("/proxy/historico/"): self._historico() elif self.path.startswith("/proxy/"): self._proxy() else: super().do_GET() def do_POST(self): if not self._require_admin_auth(): return if self.path == "/proxy/historico/cecos-sync": self._cecos_sync() elif self.path == "/proxy/admin/sync-hoteles": self._sync_hoteles() elif self.path == "/proxy/admin/sync-cargos": self._sync_cargos() elif self.path == "/proxy/admin/sync-cecos": self._sync_cecos_xlsx() elif self.path == "/proxy/admin/sync-ir": self._sync_ir() elif self.path == "/proxy/rrhh/ingresos-retiros": self._proxy_post() elif self.path == "/proxy/nomina/reporte": self._nomina_reporte() elif self.path == "/proxy/pyg/reporte": self._pyg_reporte() elif self.path == "/proxy/admin/sync-conceptos": self._sync_conceptos_xlsx() else: self.send_response(404) self.end_headers() def do_OPTIONS(self): self.send_response(204) self._send_cors() self.send_header("Access-Control-Allow-Methods", "POST, GET, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type, X-Admin-Key") self.end_headers() # ─── Sincronizar Hoteles desde Excel ──────────────────────────────────── def _sync_hoteles(self): try: import pandas as pd xlsx = os.path.join(DIR, "Unificacion_Hoteles.xlsx") if not os.path.exists(xlsx): raise FileNotFoundError(f"No se encontró {xlsx}") df = pd.read_excel(xlsx) df.columns = [ 'razon_social','hotel_name_api','bu_codigo','bu_nombre', 'registros_historico','meses_historico','activos_api', 'nombre_personalizado','notas' ] # Fila 0 = instrucciones, saltar df = df.iloc[1:].reset_index(drop=True) conn = DB_POOL.getconn() cur = conn.cursor() updated = 0 for _, r in df.iterrows(): rs = str(r['razon_social']).strip() if not pd.isna(r['razon_social']) else None if not rs: continue han_raw = r['hotel_name_api'] han = None if (pd.isna(han_raw) or str(han_raw).startswith('—')) else str(han_raw).strip() buc = str(r['bu_codigo']).strip() if not pd.isna(r['bu_codigo']) else None bun = str(r['bu_nombre']).strip() if not pd.isna(r['bu_nombre']) else None rh = int(r['registros_historico']) if not pd.isna(r['registros_historico']) else None mh = int(r['meses_historico']) if not pd.isna(r['meses_historico']) else None aa = int(r['activos_api']) if not pd.isna(r['activos_api']) else 0 np_ = str(r['nombre_personalizado']).strip() if not pd.isna(r['nombre_personalizado']) else None nt = str(r['notas']).strip() if not pd.isna(r['notas']) else None cur.execute(""" INSERT INTO hotel_unificacion (razon_social, hotel_name_api, bu_codigo, bu_nombre, registros_historico, meses_historico, activos_api, nombre_personalizado, notas, actualizado_en) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,NOW()) ON CONFLICT (razon_social) DO UPDATE SET hotel_name_api = EXCLUDED.hotel_name_api, bu_codigo = EXCLUDED.bu_codigo, bu_nombre = EXCLUDED.bu_nombre, registros_historico = EXCLUDED.registros_historico, meses_historico = EXCLUDED.meses_historico, activos_api = EXCLUDED.activos_api, nombre_personalizado = EXCLUDED.nombre_personalizado, notas = EXCLUDED.notas, actualizado_en = NOW() """, (rs, han, buc, bun, rh, mh, aa, np_, nt)) updated += 1 conn.commit() cur.close() DB_POOL.putconn(conn) print(f" ✅ hotel_unificacion actualizada: {updated} filas") resp = json.dumps({"ok": True, "actualizadas": updated}).encode() self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self._send_cors() self.end_headers() self.wfile.write(resp) except Exception as e: self.send_response(500) self.send_header("Content-Type", "application/json") self._send_cors() self.end_headers() self.wfile.write(json.dumps(self._error_payload(e, ok=False)).encode()) # ─── Sincronizar Cargos desde Excel ───────────────────────────────────── def _sync_cargos(self): try: import pandas as pd xlsx = os.path.join(DIR, "Unificacion_Cargos.xlsx") if not os.path.exists(xlsx): raise FileNotFoundError(f"No se encontró {xlsx}") df = pd.read_excel(xlsx, header=1) df.columns = ['cargo_original','cargo_unificado','num_hoteles', 'registros_pg','ultima_fecha','hoteles'] df = df.dropna(subset=['cargo_original']).reset_index(drop=True) conn = DB_POOL.getconn() cur = conn.cursor() updated = 0 for _, r in df.iterrows(): co = str(r['cargo_original']).strip() cu = str(r['cargo_unificado']).strip() if not pd.isna(r['cargo_unificado']) else None nh = int(r['num_hoteles']) if not pd.isna(r['num_hoteles']) else None rpg = int(r['registros_pg']) if not pd.isna(r['registros_pg']) else None uf = str(r['ultima_fecha'])[:10] if not pd.isna(r['ultima_fecha']) else None hot = str(r['hoteles']).strip() if not pd.isna(r['hoteles']) else None cur.execute(""" INSERT INTO cargo_unificacion (cargo_original, cargo_unificado, num_hoteles, registros_pg, ultima_fecha, hoteles, actualizado_en) VALUES (%s,%s,%s,%s,%s,%s,NOW()) ON CONFLICT (cargo_original) DO UPDATE SET cargo_unificado = EXCLUDED.cargo_unificado, num_hoteles = EXCLUDED.num_hoteles, registros_pg = EXCLUDED.registros_pg, ultima_fecha = EXCLUDED.ultima_fecha, hoteles = EXCLUDED.hoteles, actualizado_en = NOW() """, (co, cu, nh, rpg, uf, hot)) updated += 1 conn.commit() cur.close(); DB_POOL.putconn(conn) print(f" ✅ cargo_unificacion actualizada: {updated} filas") resp = json.dumps({"ok": True, "actualizadas": updated}).encode() self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self._send_cors() self.end_headers() self.wfile.write(resp) except Exception as e: self.send_response(500) self.send_header("Content-Type", "application/json") self._send_cors() self.end_headers() self.wfile.write(json.dumps(self._error_payload(e, ok=False)).encode()) # ─── Sincronizar mappings CECOs desde Excel ───────────────────────────── def _sync_cecos_xlsx(self): """Lee Unificacion_CECOs.xlsx y actualiza ceco_unificado en ceco_unificacion.""" try: import pandas as pd xlsx = os.path.join(DIR, "Unificacion_CECOs.xlsx") if not os.path.exists(xlsx): raise FileNotFoundError(f"No se encontró {xlsx}") # El Excel tiene: fila 0 = header real, fila 1 = leyenda → usar header=0 y skip leyenda df = pd.read_excel(xlsx, header=0) # Renombrar columnas por posición (independiente de tildes/saltos de línea) df.columns = ['ceco_original', 'fuente', 'empleados_api', 'registros_pg', 'ceco_unificado', 'notas'] # Saltar fila de leyenda (primera fila de datos suele ser "Leyenda:") df = df[df['ceco_original'].notna()] df = df[df['ceco_original'].astype(str).str.strip().str.upper() != 'LEYENDA:'] df = df.reset_index(drop=True) conn = DB_POOL.getconn() cur = conn.cursor() updated = 0 skipped = 0 for _, r in df.iterrows(): co = str(r['ceco_original']).strip() cu = str(r['ceco_unificado']).strip() \ if pd.notna(r['ceco_unificado']) and str(r['ceco_unificado']).strip() not in ('', 'nan') \ else None if not co: continue cur.execute(""" INSERT INTO ceco_unificacion (ceco_original, ceco_unificado, actualizado_en) VALUES (%s, %s, NOW()) ON CONFLICT (ceco_original) DO UPDATE SET ceco_unificado = EXCLUDED.ceco_unificado, actualizado_en = NOW() """, (co, cu)) if cu: updated += 1 else: skipped += 1 conn.commit() cur.close(); DB_POOL.putconn(conn) print(f" ✅ ceco_unificacion: {updated} mappings aplicados, {skipped} sin unificado") resp = json.dumps({"ok": True, "mapeados": updated, "sin_unificado": skipped}).encode() self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self._send_cors() self.end_headers() self.wfile.write(resp) except Exception as e: self.send_response(500) self.send_header("Content-Type", "application/json") self._send_cors() self.end_headers() self.wfile.write(json.dumps(self._error_payload(e, ok=False)).encode()) # ─── Sincronizar CECOs nuevos desde la API ────────────────────────────── def _cecos_sync(self): try: length = int(self.headers.get("Content-Length", 0)) body = json.loads(self.rfile.read(length).decode("utf-8")) cecos = body.get("cecos", []) # lista de {key, fuente, empleados_api} conn = DB_POOL.getconn() cur = conn.cursor() inserted = [] for item in cecos: key = item.get("key", "").strip() fuente = item.get("fuente", "API") count = item.get("empleados_api", 0) if not key: continue cur.execute(""" INSERT INTO ceco_unificacion (ceco_original, ceco_unificado, fuente, empleados_api, notas) VALUES (%s, %s, %s, %s, 'Auto-detectado') ON CONFLICT (ceco_original) DO UPDATE SET empleados_api = EXCLUDED.empleados_api, actualizado_en = NOW() RETURNING (xmax = 0) AS es_nuevo """, (key, key, fuente, count)) row = cur.fetchone() if row and row[0]: inserted.append(key) conn.commit() cur.close() DB_POOL.putconn(conn) if inserted: print(f" ✅ CECOs nuevos registrados: {inserted}") resp = json.dumps({"insertados": len(inserted), "nuevos": inserted}).encode() self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self._send_cors() self.end_headers() self.wfile.write(resp) except Exception as e: self.send_response(500) self.send_header("Content-Type", "application/json") self._send_cors() self.end_headers() self.wfile.write(json.dumps(self._error_payload(e)).encode()) # ─── Sincronizar historico IR desde Excel ─────────────────────────────── def _sync_ir(self): XLSX_IR = ( "/Users/ricardomartinez/Library/CloudStorage/" "OneDrive-farandahotels.com/Business Intelligence/" "Talento Humano/Base de Datos Historicos/" "Ingresos y Retiros Empleados Historicos.xlsx" ) try: import pandas as pd, numpy as np if not os.path.exists(XLSX_IR): raise FileNotFoundError(f"No se encontró: {XLSX_IR}") df = pd.read_excel(XLSX_IR, sheet_name="Hoja1") df.columns = ['razon_social','nombre_completo','ingreso','cargo', 'tipo_movimiento','sueldo_base','tipo_contrato', 'termino','operacion','causa_retiro'] def cl(v): if v is None or (isinstance(v, float) and np.isnan(v)): return None return str(v).strip() or None def cd(v): try: ts = pd.Timestamp(v) return None if pd.isna(ts) else ts.date() except: return None def cn(v): try: f = float(v) return None if np.isnan(f) else f except: return None # Derivar periodo: I → mes ingreso, R → mes termino, IR → mes ingreso def cp(tipo, ingreso_val, termino_val): try: if str(tipo).upper() == 'R': ts = pd.Timestamp(termino_val) else: ts = pd.Timestamp(ingreso_val) if pd.isna(ts): return None return f"{ts.year}-{ts.month:02d}" except: return None conn = DB_POOL.getconn() cur = conn.cursor() cur.execute(""" CREATE TABLE IF NOT EXISTS historico_ir ( id SERIAL PRIMARY KEY, razon_social TEXT, nombre_completo TEXT, ingreso DATE, cargo TEXT, tipo_movimiento TEXT, sueldo_base NUMERIC(14,2), tipo_contrato TEXT, termino DATE, operacion TEXT, causa_retiro TEXT, periodo VARCHAR(7) ) """) # Asegurar que la columna periodo existe en tablas creadas antes cur.execute(""" ALTER TABLE historico_ir ADD COLUMN IF NOT EXISTS periodo VARCHAR(7) """) conn.commit() cur.execute("TRUNCATE TABLE historico_ir RESTART IDENTITY") rows = [ (cl(r['razon_social']), cl(r['nombre_completo']), cd(r['ingreso']), cl(r['cargo']), cl(r['tipo_movimiento']), cn(r['sueldo_base']), cl(r['tipo_contrato']), cd(r['termino']), cl(r['operacion']), cl(r['causa_retiro']), cp(r['tipo_movimiento'], r['ingreso'], r['termino'])) for _, r in df.iterrows() ] psycopg2.extras.execute_batch(cur, """ INSERT INTO historico_ir (razon_social,nombre_completo,ingreso,cargo,tipo_movimiento, sueldo_base,tipo_contrato,termino,operacion,causa_retiro,periodo) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) """, rows) conn.commit() cur.execute("SELECT COUNT(*) FROM historico_ir") total = cur.fetchone()[0] cur.close(); DB_POOL.putconn(conn) print(f" ✅ historico_ir actualizado: {total} filas") resp = json.dumps({"ok": True, "total": total}).encode() self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self._send_cors() self.end_headers() self.wfile.write(resp) except Exception as e: self.send_response(500) self.send_header("Content-Type", "application/json") self._send_cors() self.end_headers() self.wfile.write(json.dumps(self._error_payload(e, ok=False)).encode()) # ─── Helpers de respuesta y cuerpo ────────────────────────────────────── def _error_payload(self, exc, *, ok=None): payload = {} if ok is not None: payload["ok"] = ok payload["error"] = str(exc) if DEBUG_ERRORS else "Error interno" if DEBUG_ERRORS: import traceback payload["trace"] = traceback.format_exc() return payload def _send_json(self, code, obj): body = json.dumps(obj, default=str, ensure_ascii=False).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self._send_cors() self.end_headers() self.wfile.write(body) def _read_json_body(self): length = int(self.headers.get("Content-Length", 0)) raw = self.rfile.read(length) if length else b"{}" return json.loads(raw or b"{}"), raw @staticmethod def _is_periodo_historico(ano, mes): """Retorna True si (ano, mes) <= último período cargado en historico_nomina_pyg → usar caché PostgreSQL.""" return (int(ano), int(mes)) <= _get_max_periodo_nomina() def _proxy_post_body(self, api_path, body): """Proxy POST a la API externa con body ya leído.""" url = API_BASE + api_path try: req = urllib.request.Request( url, data=body, headers={"X-API-Key": API_KEY, "Content-Type": "application/json"}, method="POST" ) with urllib.request.urlopen(req, timeout=60, context=SSL_CTX) as resp: data = resp.read() self.send_response(200) self.send_header("Content-Type", "application/json") self._send_cors() self.end_headers() self.wfile.write(data) except urllib.error.HTTPError as e: self.send_response(e.code) self.send_header("Content-Type", "application/json") self._send_cors() self.end_headers() self.wfile.write(json.dumps({"error": str(e)}).encode()) except Exception as e: self.send_response(502) self.send_header("Content-Type", "application/json") self._send_cors() self.end_headers() self.wfile.write(json.dumps({"error": str(e)}).encode()) # ─── Nómina reporte con caché PostgreSQL ──────────────────────────────── def _nomina_reporte(self): """POST /proxy/nomina/reporte con caché histórica PostgreSQL hasta 2026-05.""" try: payload, raw_body = self._read_json_body() nit = str(payload.get("hotel_nit", "")).strip() ano = int(payload.get("ano", 0)) mes = int(payload.get("mes", 0)) if nit and ano and mes and self._is_periodo_historico(ano, mes): conn = DB_POOL.getconn() cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) cur.execute(""" SELECT raw_data FROM historico_nomina_pyg WHERE fuente='nomina' AND registro_tipo='respuesta' AND ano=%s AND mes=%s AND nit=%s ORDER BY cargado_en DESC LIMIT 1 """, (ano, mes, nit)) row = cur.fetchone() cur.close(); DB_POOL.putconn(conn) if row: return self._send_json(200, row["raw_data"]) return self._proxy_post_body("/nomina/reporte", raw_body) except Exception as e: return self._send_json(502, {"error": str(e)}) # ─── POST proxy API externa ───────────────────────────────────────────── # ─── Hoteles con NIT desde BD ─────────────────────────────────────────────── def _nomina_hoteles_nit(self): """ Lee hotel_unificacion JOIN empresas para devolver la lista de hoteles con NIT configurado, uno por fila (separados por BU cuando aplica). Responde: { hotels: [{label, nit, hotel_api, bu_codigo, bu_nombre}] } """ try: import psycopg2, psycopg2.extras conn = DB_POOL.getconn() cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) cur.execute(""" SELECT DISTINCT ON (h.nombre_personalizado, e.nit, h.bu_codigo) h.nombre_personalizado AS label, e.nit, h.hotel_name_api AS hotel_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.nombre_personalizado NOT IN ('No mostrar') ORDER BY h.nombre_personalizado, e.nit, h.bu_codigo """) rows = cur.fetchall() cur.close(); DB_POOL.putconn(conn) data = {"hotels": [dict(r) for r in rows]} except Exception as ex: data = {"hotels": [], "error": str(ex)} resp = json.dumps(data).encode() self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self._send_cors() self.end_headers() self.wfile.write(resp) # ─── Mapeo de conceptos nómina desde BD ───────────────────────────────────── def _nomina_conceptos_map(self): """GET /proxy/nomina/conceptos-map — devuelve tabla nomina_conceptos como JSON.""" try: import psycopg2, psycopg2.extras conn = DB_POOL.getconn() cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) cur.execute(""" SELECT codigo, regexp_replace(codigo, '_v[0-9]+$', '') AS codigo_api, nombre, tipo, grupo, subgrupo, nombre_display, incluir FROM nomina_conceptos ORDER BY tipo, grupo NULLS LAST, subgrupo NULLS LAST, codigo """) rows = cur.fetchall() cur.close(); DB_POOL.putconn(conn) data = {"conceptos": [dict(r) for r in rows]} except Exception as ex: data = {"conceptos": [], "error": str(ex)} resp = json.dumps(data, default=str).encode() self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self._send_cors() self.end_headers() self.wfile.write(resp) # ─── Sync conceptos desde Excel ───────────────────────────────────────────── def _sync_conceptos_xlsx(self): """POST /proxy/admin/sync-conceptos — lee Conceptos_Nomina_Mapeo.xlsx y upsert BD.""" try: import pandas as pd, psycopg2 xlsx = os.path.join(DIR, "Conceptos_Nomina_Mapeo.xlsx") if not os.path.exists(xlsx): # Directorio principal del proyecto (3 niveles arriba del worktree) xlsx = os.path.abspath(os.path.join(DIR, "../../..", "Conceptos_Nomina_Mapeo.xlsx")) df = pd.read_excel(xlsx, sheet_name="Mapeo Agrupación", header=0) # Detectar formato: 11 cols (nuevo con Código BD) o 7 cols (viejo) cols = list(df.columns) if len(cols) >= 8: # Nuevo: CódigoBD, CódigoAPI, NombreAPI, Tipo, Incluir, Grupo, Subgrupo, NombreDisplay... col_map = {0:"codigo", 2:"nombre", 3:"tipo", 4:"incluir", 5:"grupo", 6:"subgrupo", 7:"nombre_display"} else: # Viejo: Código, Nombre, Tipo, Incluir, Grupo, Subgrupo, NombreDisplay col_map = {0:"codigo", 1:"nombre", 2:"tipo", 3:"incluir", 4:"grupo", 5:"subgrupo", 6:"nombre_display"} rename = {cols[i]: name for i, name in col_map.items() if i < len(cols)} df = df.rename(columns=rename) df = df.dropna(subset=["codigo"]) conn = DB_POOL.getconn() cur = conn.cursor() updated = 0 for _, row in df.iterrows(): incluir = False if str(row.get("incluir","S") or "S").strip().upper() == "N" else True cur.execute(""" INSERT INTO nomina_conceptos (codigo, nombre, tipo, grupo, subgrupo, nombre_display, incluir) VALUES (%s,%s,%s,%s,%s,%s,%s) ON CONFLICT (codigo) DO UPDATE SET nombre = EXCLUDED.nombre, tipo = EXCLUDED.tipo, grupo = COALESCE(NULLIF(EXCLUDED.grupo,''), nomina_conceptos.grupo), subgrupo = COALESCE(NULLIF(EXCLUDED.subgrupo,''), nomina_conceptos.subgrupo), nombre_display = COALESCE(NULLIF(EXCLUDED.nombre_display,''), nomina_conceptos.nombre_display), incluir = EXCLUDED.incluir, actualizado_en = NOW() """, ( str(row["codigo"]).strip(), str(row["nombre"]).strip() if pd.notna(row["nombre"]) else "", str(row["tipo"]).strip() if pd.notna(row["tipo"]) else "Devengo", str(row["grupo"]).strip() if pd.notna(row.get("grupo")) else None, str(row["subgrupo"]).strip() if pd.notna(row.get("subgrupo")) else None, str(row["nombre_display"]).strip() if pd.notna(row.get("nombre_display")) else None, incluir, )) updated += 1 conn.commit() cur.execute("SELECT COUNT(*) FROM nomina_conceptos WHERE grupo IS NOT NULL") mapeados = cur.fetchone()[0] cur.close(); DB_POOL.putconn(conn) msg = {"ok": True, "procesados": updated, "con_grupo": mapeados} except Exception as ex: msg = {"ok": False, "error": str(ex)} resp = json.dumps(msg).encode() self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self._send_cors() self.end_headers() self.wfile.write(resp) # ─── P&G reporte ──────────────────────────────────────────────────────────── def _pyg_reporte(self): """POST /proxy/pyg/reporte con caché histórica PostgreSQL hasta 2026-03. Body: { nit, ano, mes, company_id, bu_codigos } """ try: payload, raw_body = self._read_json_body() ano = int(payload.get("ano", 0)) mes = int(payload.get("mes", 0)) company_id = payload.get("company_id") bu_codigos = payload.get("bu_codigos", ["Local"]) or ["Local"] bu_codigos = [str(b).lower() for b in bu_codigos if str(b).strip()] # ── Caché PostgreSQL para períodos históricos ────────────────── if ano and mes and company_id and self._is_periodo_historico(ano, mes): conn = DB_POOL.getconn() cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) cur.execute(""" SELECT concepto_codigo, concepto_nombre, grupo, subgrupo, SUM(valor)::float AS val, SUM(valor_ly)::float AS ly, BOOL_OR(COALESCE((raw_data->>'nomina_backed')::boolean, false)) AS nomina_backed FROM historico_nomina_pyg WHERE fuente='pyg' AND registro_tipo='concepto' AND ano=%s AND mes=%s AND company_id=%s AND LOWER(COALESCE(bu_codigo,'')) = ANY(%s) GROUP BY concepto_codigo, concepto_nombre, grupo, subgrupo ORDER BY grupo, subgrupo, concepto_nombre """, (ano, mes, int(company_id), bu_codigos)) rows = cur.fetchall() if rows: cur.close(); DB_POOL.putconn(conn) nit = str(payload.get("nit", "")).strip() conceptos = [{ "id": r["concepto_codigo"], "nombre": r["concepto_nombre"], "grupo": r["grupo"], "subgrupo": r["subgrupo"], "val": r["val"] or 0, "ly": r["ly"] or 0, "fuente": "pyg", "nomina_backed": r["nomina_backed"], } for r in rows] return self._send_json(200, { "nit": nit, "company_id": int(company_id), "period": f"{ano}{mes:02d}", "conceptos": conceptos, }) # Sin concepto rows — intentar extraer desde respuesta cacheada cur.execute(""" SELECT raw_data, bu_codigo FROM historico_nomina_pyg WHERE fuente='pyg' AND registro_tipo='respuesta' AND ano=%s AND mes=%s AND company_id=%s ORDER BY cargado_en DESC """, (ano, mes, int(company_id))) resp_rows = cur.fetchall() cur.close(); DB_POOL.putconn(conn) if resp_rows: def _flatten(nodes, out): for n in nodes: nid = n.get("id", "") monthly = n.get("monthly", {}) real = monthly.get("real", {}) val = 0.0 if isinstance(real, dict): for k, v in real.items(): if isinstance(v, (int, float)): val = v; break elif isinstance(v, dict): val = v.get("mes", v.get("value", 0.0)); break ly_data = monthly.get("last_year", {}) ly_val = 0.0 if isinstance(ly_data, dict): for k, v in ly_data.items(): if isinstance(v, (int, float)): ly_val = v; break elif isinstance(v, dict): ly_val = v.get("mes", v.get("value", 0.0)); break out[nid] = {"id": nid, "label": n.get("label",""), "val": val or 0.0, "ly": ly_val or 0.0} _flatten(n.get("children", []), out) flat = {} for rrow in resp_rows: bu_k = (rrow.get("bu_codigo") or "local").lower() if bu_k not in bu_codigos and bu_codigos: continue pnl = (rrow["raw_data"] or {}).get("pnl", {}) for bk in [bu_k] + list(pnl.keys()): tree = pnl.get(bk, {}).get("tree", []) if tree: _flatten(tree, flat) break if not flat: for rrow in resp_rows: pnl = (rrow["raw_data"] or {}).get("pnl", {}) for bk in list(pnl.keys()): tree = pnl.get(bk, {}).get("tree", []) if tree: _flatten(tree, flat) break 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), ("dotación_y_suministro_a_trabajadores", "Dotación y Suministro", "Otros Gastos Personal", "Dotación", False), ("dotacion_y_suministro_a_trabajadores", "Dotación y Suministro", "Otros Gastos Personal", "Dotación", False), ("comida_empleados", "Comida Empleados", "Otros Gastos Personal", "Bienestar", False), ("transporte_de_personal", "Transporte de Personal", "Otros Gastos Personal", "Bienestar", False), ("capacitación_al_personal", "Capacitación al Personal", "Otros Gastos Personal", "Bienestar", False), ("capacitacion_al_personal", "Capacitación al Personal", "Otros Gastos Personal", "Bienestar", False), ("gastos_medicos_y_medicamentos", "Gastos Médicos y Medicamentos", "Otros Gastos Personal", "Gastos Médicos", False), ("salud_ocupacional_y_examenes_d", "Salud Ocupacional y Exámenes", "Otros Gastos Personal", "Gastos Médicos", False), ] # Indexar también por segmento final del ID (ej. "grupo/concepto" → "concepto") flat_short = {} for k, v in flat.items(): short = k.split("/")[-1] if short not in flat_short: flat_short[short] = v seen, conceptos_out = set(), [] for (node_id, nombre, grupo, subgrupo, nomina_backed) in CONCEPTOS: if node_id in seen: continue entry = flat.get(node_id) or flat_short.get(node_id) if entry is None: continue seen.add(node_id) seen.add(node_id.replace("ó","o").replace("á","a").replace("é","e").replace("ú","u")) conceptos_out.append({ "id": node_id, "nombre": nombre, "grupo": grupo, "subgrupo": subgrupo, "val": entry["val"], "ly": entry["ly"], "fuente": "pyg", "nomina_backed": nomina_backed }) if conceptos_out: return self._send_json(200, { "nit": str(payload.get("nit","")).strip(), "company_id": int(company_id), "period": f"{ano}{mes:02d}", "conceptos": conceptos_out, }) # ── Llamada a la API P&G ─────────────────────────────────────── return self._pyg_reporte_api(payload, bu_codigos, company_id, ano, mes) except Exception as e: return self._send_json(502, {"error": str(e)}) def _pyg_reporte_api(self, payload, bu_codigos, company_id, ano, mes): """Llama a la API P&G real y devuelve conceptos de gasto personal.""" nit = str(payload.get("nit", "")).strip() if not company_id: if not nit: return self._send_json(400, {"error": "Faltan parámetros: company_id o nit"}) con = DB_POOL.getconn() cur = con.cursor() cur.execute("SELECT DISTINCT pyg_company_id FROM empresas WHERE nit=%s AND pyg_company_id IS NOT NULL LIMIT 1", (nit,)) row = cur.fetchone() cur.close(); DB_POOL.putconn(con) if not row: return self._send_json(404, {"error": f"No hay pyg_company_id para NIT {nit}"}) company_id = row[0] if not ano or not mes: return self._send_json(400, {"error": "Faltan parámetros: ano, mes"}) period = f"{ano}{mes:02d}" pyg_body = json.dumps({ "company_id": company_id, "period": period, "scenario_id": None, "business_units": bu_codigos, "include_kpis": False }).encode() def _request_pyg(bu_list): body = json.dumps({ "company_id": company_id, "period": period, "scenario_id": None, "business_units": bu_list, "include_kpis": False }).encode() req = urllib.request.Request( PYG_URL, data=body, headers={"X-API-Key": PYG_KEY, "Content-Type": "application/json"}, method="POST" ) with urllib.request.urlopen(req, timeout=30, context=SSL_CTX) as resp: return json.loads(resp.read()) raw = _request_pyg(bu_codigos) def flatten_tree(nodes, out): for n in nodes: nid = n.get("id", "") monthly = n.get("monthly", {}) real = monthly.get("real", {}) val = 0.0 if isinstance(real, dict): for k, v in real.items(): if isinstance(v, (int, float)): val = v; break elif isinstance(v, dict): val = v.get("mes", v.get("value", 0.0)); break ly_data = monthly.get("last_year", {}) ly_val = 0.0 if isinstance(ly_data, dict): for k, v in ly_data.items(): if isinstance(v, (int, float)): ly_val = v; break elif isinstance(v, dict): ly_val = v.get("mes", v.get("value", 0.0)); break out[nid] = {"id": nid, "label": n.get("label", ""), "val": val or 0.0, "ly": ly_val or 0.0} flatten_tree(n.get("children", []), out) pnl = raw.get("pnl", {}) flat = {} for bu_key in bu_codigos: bu_section = pnl.get(bu_key, {}) tree = bu_section.get("tree", []) if not tree: continue bu_flat = {} flatten_tree(tree, bu_flat) for nid, entry in bu_flat.items(): if nid not in flat: flat[nid] = {"id": nid, "label": entry["label"], "val": 0.0, "ly": 0.0} flat[nid]["val"] += entry.get("val", 0.0) flat[nid]["ly"] += entry.get("ly", 0.0) if not flat and "local" in pnl: flatten_tree(pnl["local"].get("tree", []), flat) if not flat and "local" not in bu_codigos: raw = _request_pyg(["local"]) pnl = raw.get("pnl", {}) flatten_tree(pnl.get("local", {}).get("tree", []), flat) CONCEPTOS = [ # ── Con contrapartida en nómina ────────────────────────────────── ("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), # ── Solo en P&G (sin empleados en nómina) ─────────────────────── ("temporales", "Temporales", "Personal Externo", "Temporales", False), ("honorarios_laborales", "Honorarios Laborales", "Personal Externo", "Honorarios", False), ("comisiones", "Comisiones Personal", "Personal Externo", "Honorarios", False), ("dotación_y_suministro_a_trabajadores", "Dotación y Suministro", "Otros Gastos Personal", "Dotación", False), ("dotacion_y_suministro_a_trabajadores", "Dotación y Suministro", "Otros Gastos Personal", "Dotación", False), ("comida_empleados", "Comida Empleados", "Otros Gastos Personal", "Bienestar", False), ("transporte_de_personal", "Transporte de Personal", "Otros Gastos Personal", "Bienestar", False), ("capacitación_al_personal", "Capacitación al Personal", "Otros Gastos Personal", "Bienestar", False), ("capacitacion_al_personal", "Capacitación al Personal", "Otros Gastos Personal", "Bienestar", False), ("gastos_medicos_y_medicamentos", "Gastos Médicos y Medicamentos", "Otros Gastos Personal", "Gastos Médicos", False), ("salud_ocupacional_y_examenes_d", "Salud Ocupacional y Exámenes", "Otros Gastos Personal", "Gastos Médicos", False), ] seen = set() conceptos_out = [] for (node_id, nombre, grupo, subgrupo, nomina_backed) in CONCEPTOS: if node_id in seen: continue entry = flat.get(node_id) if entry is None: continue seen.add(node_id) alt = node_id.replace("ó","o").replace("á","a").replace("é","e").replace("ú","u") seen.add(alt) conceptos_out.append({ "id": node_id, "nombre": nombre, "grupo": grupo, "subgrupo": subgrupo, "val": entry["val"], "ly": entry["ly"], "fuente": "pyg", "nomina_backed": nomina_backed }) return self._send_json(200, { "nit": nit, "company_id": company_id, "period": period, "conceptos": conceptos_out }) def _proxy_post(self): api_path = self.path[len("/proxy"):] url = API_BASE + api_path try: length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(length) req = urllib.request.Request( url, data=body, headers={"X-API-Key": API_KEY, "Content-Type": "application/json"}, method="POST" ) with urllib.request.urlopen(req, timeout=60, context=SSL_CTX) as resp: data = resp.read() self.send_response(200) self.send_header("Content-Type", "application/json") self._send_cors() self.end_headers() self.wfile.write(data) except urllib.error.HTTPError as e: self.send_response(e.code) self.send_header("Content-Type", "application/json") self._send_cors() self.end_headers() self.wfile.write(json.dumps({"error": str(e)}).encode()) except Exception as e: self.send_response(502) self.send_header("Content-Type", "application/json") self._send_cors() self.end_headers() self.wfile.write(json.dumps({"error": str(e)}).encode()) # ─── API externa ──────────────────────────────────────────────────────── def _proxy(self): api_path = self.path[len("/proxy"):] url = API_BASE + api_path req = urllib.request.Request(url, headers={"X-API-Key": API_KEY}) try: with urllib.request.urlopen(req, timeout=30, context=SSL_CTX) as resp: body = resp.read() self.send_response(200) self.send_header("Content-Type", "application/json") self._send_cors() self.end_headers() self.wfile.write(body) except urllib.error.HTTPError as e: self.send_response(e.code) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write(json.dumps({"error": str(e)}).encode()) except Exception as e: self.send_response(502) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write(json.dumps({"error": str(e)}).encode()) # ─── PostgreSQL históricos ─────────────────────────────────────────────── @staticmethod def _parse_list(qs, key): """Devuelve lista de valores en mayúsculas; acepta comma-separated.""" raw = qs.get(key, [""])[0].strip().upper() return [v.strip() for v in raw.split(",") if v.strip()] if raw else [] @staticmethod def _where_in(col, vals, a): """Agrega cláusula = o IN según cantidad de valores.""" if len(vals) == 1: a.append(vals[0]) return f"UPPER({col}) = %s" placeholders = ",".join(["%s"] * len(vals)) a.extend(vals) return f"UPPER({col}) IN ({placeholders})" @staticmethod def _where_cargo(col, vals, a): """Cláusula cargo con lookup en cargo_unificacion; soporta múltiples valores. Usa UPPER() en ambos lados para manejar datos con capitalización mixta.""" if len(vals) == 1: a.extend([vals[0], vals[0]]) return f"""(UPPER({col}) = %s OR UPPER({col}) IN ( SELECT UPPER(cargo_original) FROM cargo_unificacion WHERE UPPER(cargo_unificado) = %s))""" ph = ",".join(["%s"] * len(vals)) a.extend(vals + vals) return f"""(UPPER({col}) IN ({ph}) OR UPPER({col}) IN ( SELECT UPPER(cargo_original) FROM cargo_unificacion WHERE UPPER(cargo_unificado) IN ({ph})))""" def _historico(self): parsed = urlparse(self.path) qs = parse_qs(parsed.query) endpoint = parsed.path.split("/proxy/historico/")[-1] conn = DB_POOL.getconn() try: cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) data = {} # ── /proxy/historico/max-periodo ──────────────────────────────── # Último período con datos por tabla — el frontend lo usa para fijar # dinámicamente SAL_MAX / IR_HIST_MAX en vez de hardcodearlos. if endpoint == "max-periodo": cur.execute("SELECT MAX(fecha)::text AS m FROM historico_empleados") max_empleados = cur.fetchone()["m"] cur.execute("SELECT MAX(periodo) AS m FROM historico_ir") max_ir = cur.fetchone()["m"] cur.execute("SELECT MAX(periodo) AS m FROM historico_nomina_pyg WHERE fuente='nomina'") max_nomina = cur.fetchone()["m"] data = { "empleados": max_empleados, # "2026-05-01" "ir": max_ir, # "2026-05" "nomina_pyg": max_nomina, # "2026-05" } # ── /proxy/historico/salarios ───────────────────────────────── elif endpoint == "salarios": fecha = qs.get("fecha", ["2026-03-01"])[0] razon = qs.get("razon_social", [""])[0] operac = qs.get("operacion", [""])[0].upper() centros = self._parse_list(qs, "centro") cargos = self._parse_list(qs, "cargo") nombre = qs.get("nombre", [""])[0].upper() d = date_type.fromisoformat(fecha) fecha_ly = date_type(d.year - 1, d.month, 1).isoformat() def q_cecos(f): w, a = ["fecha = %s"], [f] rw, ra = razon_where(razon) if rw: w.append(rw); a.extend(ra) if operac: w.append("UPPER(operacion) = %s"); a.append(operac) if centros: w.append(self._where_in("centro_costo", centros, a)) if cargos: w.append(self._where_cargo("cargo", cargos, a)) if nombre: w.append("UPPER(nombre_completo) LIKE %s"); a.append(f"%{nombre}%") sql = f""" SELECT TRIM(centro_costo) AS ceco, SUM(sueldo_base)::bigint AS salario_base, ROUND(AVG(sueldo_base))::int AS salario_promedio, COUNT(*)::int AS empleados, COUNT(*) FILTER (WHERE INITCAP(sexo) = 'Masculino')::int AS masculino, COUNT(*) FILTER (WHERE INITCAP(sexo) = 'Femenino')::int AS femenino FROM historico_empleados WHERE {' AND '.join(w)} AND centro_costo IS NOT NULL GROUP BY TRIM(centro_costo) ORDER BY salario_base DESC LIMIT 20 """ cur.execute(sql, a) return {r["ceco"]: dict(r) for r in cur.fetchall()} curr = q_cecos(fecha) ly = q_cecos(fecha_ly) # Razones disponibles cur.execute("""SELECT DISTINCT razon_social FROM historico_empleados WHERE razon_social IS NOT NULL ORDER BY 1""") razones = [r["razon_social"] for r in cur.fetchall()] data = { "fecha": fecha, "fecha_ly": fecha_ly, "razones": razones, "cecos": [ { "ceco": ceco, "salario_base": r["salario_base"], "salario_promedio": r["salario_promedio"], "empleados": r["empleados"], "masculino": r["masculino"], "femenino": r["femenino"], "salario_base_ly": ly.get(ceco, {}).get("salario_base", 0), "salario_promedio_ly": ly.get(ceco, {}).get("salario_promedio", 0), "empleados_ly": ly.get(ceco, {}).get("empleados", 0), } for ceco, r in curr.items() ], } # ── /proxy/historico/matriz ─────────────────────────────────── elif endpoint == "matriz": fecha = qs.get("fecha", ["2026-03-01"])[0] razon = qs.get("razon_social", [""])[0] operac = qs.get("operacion", [""])[0].upper() centros = self._parse_list(qs, "centro") cargos = self._parse_list(qs, "cargo") nombre = qs.get("nombre", [""])[0].upper() w, a = ["fecha = %s", "centro_costo IS NOT NULL"], [fecha] rw, ra = razon_where(razon) if rw: w.append(rw); a.extend(ra) if operac: w.append("UPPER(operacion) = %s"); a.append(operac) if centros: w.append(self._where_in("centro_costo", centros, a)) if cargos: w.append(self._where_cargo("cargo", cargos, a)) if nombre: w.append("UPPER(nombre_completo) LIKE %s"); a.append(f"%{nombre}%") cur.execute(f""" SELECT TRIM(centro_costo) AS ceco, COALESCE(cargo, 'Sin Cargo') AS cargo, nombre_completo, sueldo_base FROM historico_empleados WHERE {' AND '.join(w)} ORDER BY TRIM(centro_costo), cargo, nombre_completo """, a) from collections import defaultdict tree = {} for r in cur.fetchall(): c, g, n, s = r["ceco"], r["cargo"], r["nombre_completo"], int(r["sueldo_base"] or 0) if c not in tree: tree[c] = {} if g not in tree[c]: tree[c][g] = [] tree[c][g].append({"nombre": n, "sueldo": s}) result = [] for ceco in sorted(tree): cargos_list, ct, ce = [], 0, 0 for cargo in sorted(tree[ceco]): personas = sorted(tree[ceco][cargo], key=lambda x: x["nombre"]) st = sum(p["sueldo"] for p in personas) em = len(personas) ct += st; ce += em cargos_list.append({ "cargo": cargo, "salario_base": st, "salario_promedio": st // em if em else 0, "empleados": em, "personas": personas, }) result.append({ "ceco": ceco, "salario_base": ct, "salario_promedio": ct // ce if ce else 0, "empleados": ce, "cargos": cargos_list, }) data = {"fecha": fecha, "cecos": result} # ── /proxy/historico/cargos-map ────────────────────────────── elif endpoint == "cargos-map": cur.execute(""" SELECT cargo_original, cargo_unificado FROM cargo_unificacion ORDER BY cargo_original """) rows = cur.fetchall() data = { "total": len(rows), "map": [dict(r) for r in rows] } # ── /proxy/historico/cecos-map ──────────────────────────────── elif endpoint == "cecos-map": cur.execute(""" SELECT ceco_original, ceco_unificado, fuente, empleados_api, registros_pg, notas FROM ceco_unificacion ORDER BY ceco_original """) rows = cur.fetchall() data = { "total": len(rows), "map": [dict(r) for r in rows] } # ── /proxy/historico/socios ─────────────────────────────────── # Cédulas de socios (tabla `socios`, la misma que usa cierre_mes.py # para clasificar operacion='SOCIO' al cerrar cada mes). Se expone # para que el frontend pueda filtrar por Operación en el mes en # vivo — la API externa de IR no trae ese dato por fila, y cruzar # contra empleados activos no sirve para los ya retirados ese mes. elif endpoint == "socios": cur.execute(""" SELECT codigo_empleado FROM socios WHERE activo = true """) rows = cur.fetchall() data = { "total": len(rows), "codigos": [r["codigo_empleado"] for r in rows] } # ── /proxy/historico/hoteles-map ───────────────────────────── elif endpoint == "hoteles-map": cur.execute(""" SELECT razon_social, hotel_name_api, bu_codigo, bu_nombre, nombre_personalizado, activos_api FROM hotel_unificacion ORDER BY razon_social """) rows = cur.fetchall() data = { "total": len(rows), "map": [dict(r) for r in rows] } # ── /proxy/historico/ingreso-real-map ────────────────────────── # Fecha real de ingreso a la empresa ACTUAL de cada empleado (por # codigo_empleado), ignorando los reinicios de fecha_ingreso que # ocurren en el sistema fuente ante CUALQUIER cambio de cargo o # tipo de contrato (no solo al cambiar de razón social). Se toma # el inicio del bloque continuo más reciente de mismo razon_social # (los cambios de empresa sí generan un bloque nuevo — ese caso no # se suma, es una decisión de negocio distinta, ver PENDIENTES.md). elif endpoint == "ingreso-real-map": cur.execute(""" WITH base AS ( SELECT codigo_empleado, fecha, razon_social, fecha_ingreso, centro_costo, LAG(razon_social) OVER ( PARTITION BY codigo_empleado ORDER BY fecha ) AS prev_razon, LAG(fecha) OVER ( PARTITION BY codigo_empleado ORDER BY fecha ) AS prev_fecha FROM historico_empleados ), grp AS ( SELECT *, SUM(CASE WHEN razon_social IS DISTINCT FROM prev_razon -- Hueco real de empleo (salió y fue recontratado -- después) — no es un simple cierre mensual con -- 1 mes de retraso, así que no debe sumarse como -- antigüedad continua aunque sea la misma empresa. OR (prev_fecha IS NOT NULL AND fecha - prev_fecha > 62) THEN 1 ELSE 0 END) OVER (PARTITION BY codigo_empleado ORDER BY fecha) AS grupo FROM base ), block_start AS ( SELECT codigo_empleado, grupo, (array_agg(fecha_ingreso ORDER BY fecha ASC))[1] AS ingreso_real FROM grp GROUP BY codigo_empleado, grupo ), latest AS ( SELECT DISTINCT ON (codigo_empleado) codigo_empleado, fecha AS ultima_fecha, razon_social, grupo, TRIM(centro_costo) AS centro_costo FROM grp ORDER BY codigo_empleado, fecha DESC ) SELECT l.codigo_empleado, l.razon_social, l.ultima_fecha, l.centro_costo, bs.ingreso_real AS fecha_ingreso_real FROM latest l JOIN block_start bs ON bs.codigo_empleado = l.codigo_empleado AND bs.grupo = l.grupo """) rows = cur.fetchall() data = { "total": len(rows), "map": [dict(r) for r in rows] } # ── /proxy/historico/empleados ──────────────────────────────── elif endpoint == "empleados": fecha = qs.get("fecha", ["2026-03-01"])[0] razon = qs.get("razon_social", [""])[0] operac = qs.get("operacion", [""])[0].upper() w, a = ["h.fecha = %s"], [fecha] rw, ra = razon_where(razon, col='h.razon_social') if rw: w.append(rw); a.extend(ra) if operac: w.append("UPPER(h.operacion) = %s"); a.append(operac) cur.execute(f""" SELECT h.codigo_empleado, h.nombre_completo, COALESCE(h.fecha_nacimiento, fn.fecha_nacimiento) AS fecha_nacimiento, INITCAP(h.sexo) AS sexo, h.fecha_ingreso, h.fecha_vencimiento_contrato, TRIM(h.centro_costo) AS centro_costo, h.cargo, h.sueldo_base::float AS sueldo_base, h.operacion, h.tipo_contrato, h.razon_social FROM historico_empleados h LEFT JOIN ( SELECT codigo_empleado, MAX(fecha_nacimiento) AS fecha_nacimiento FROM historico_empleados WHERE fecha_nacimiento IS NOT NULL GROUP BY codigo_empleado ) fn ON h.codigo_empleado = fn.codigo_empleado WHERE {' AND '.join(w)} ORDER BY h.nombre_completo """, a) rows = cur.fetchall() data = { "fecha": fecha, "total": len(rows), "empleados": [dict(r) for r in rows] } # ── /proxy/historico/trayectoria ───────────────────────────── elif endpoint == "trayectoria": nombre = qs.get("nombre", [""])[0].strip() if not nombre: data = {"snapshots": []} else: nombre_norm = norm_name(nombre) # ORDER BY fecha, id (no solo fecha): algunos empleados tienen # 2 filas para el mismo mes en la fuente (dato inconsistente, # ej. un cargo/empresa "fantasma" de 1-2 meses junto al cargo # real de años) — el id más bajo es determinístico y en la # práctica corresponde al registro real/dominante. cur.execute(""" SELECT DISTINCT ON (fecha) fecha, cargo, sueldo_base::float AS sueldo_base, TRIM(centro_costo) AS centro_costo, tipo_contrato, fecha_ingreso, razon_social FROM historico_empleados WHERE ( SELECT string_agg(w, ' ' ORDER BY w) FROM unnest(string_to_array( upper(translate(nombre_completo, 'ÁÉÍÓÚÜÑáéíóúüñÀÈÌÒÙàèìòùÂÊÎÔÛâêîôûÃÕãõÄÖäö', 'AEIOUUNaeiouunAEIOUaeiouAEIOUaeiouAOaoAOao')), ' ')) AS w ) = %s ORDER BY fecha, id """, (nombre_norm,)) rows = cur.fetchall() data = {"snapshots": [dict(r) for r in rows]} # ── /proxy/historico/nomina-empleado ──────────────────────────── # Desglose completo de nómina (conceptos, deducciones, provisiones, # seguridad social) por empleado y mes, ya cargado en # historico_nomina_pyg (fuente='nomina', registro_tipo='empleado') # desde el cierre mensual — no requiere ninguna llamada a la API en # vivo. Cubre ene-2024 en adelante hasta el último mes cerrado; el # mes en curso no está acá (se sigue consultando en vivo aparte). # empleado_codigo (columna) no está poblada — se filtra por la # cédula dentro de raw_data (sí está poblada al 100%). elif endpoint == "nomina-empleado": identificacion = qs.get("identificacion", [""])[0].strip() if not identificacion: data = {"periodos": []} else: cur.execute(""" SELECT ano, mes, raw_data FROM historico_nomina_pyg WHERE fuente='nomina' AND registro_tipo='empleado' AND raw_data->>'identificacion' = %s ORDER BY ano, mes """, (identificacion,)) rows = cur.fetchall() periodos = [] for r in rows: rd = r["raw_data"] or {} periodos.append({ "ano": r["ano"], "mes": r["mes"], "cargo": rd.get("cargo"), "centro_costo": rd.get("centro_costo"), "bu_nombre": rd.get("bu_nombre"), "sueldo_base": rd.get("s_basico"), "dias": rd.get("dias"), "total_devengos": rd.get("total_devengos"), "total_deducciones": rd.get("total_deducciones"), "aportes_patronales": rd.get("aportes_patronales"), "neto": rd.get("neto"), "conceptos": rd.get("conceptos") or [], "provisiones": rd.get("provisiones") or [], "seguridad_social": rd.get("seguridad_social") or [], }) data = { "identificacion": identificacion, "total": len(periodos), "periodos": periodos, } # ── /proxy/historico/ir ─────────────────────────────────────── elif endpoint == "ir": fi = qs.get("fecha_ini", ["2026-01-01"])[0] ff = qs.get("fecha_fin", ["2026-01-31"])[0] razon = qs.get("razon_social", [""])[0] operac = qs.get("operacion", [""])[0].upper() cargos = self._parse_list(qs, "cargo") nombre = qs.get("nombre", [""])[0].upper() sexo = qs.get("sexo", [""])[0].upper() EXCL = "('ELMA','FNS')" w = [f"h.razon_social NOT IN {EXCL}"] a = [] rw, ra = razon_where(razon, col='h.razon_social') if rw: w.append(rw); a.extend(ra) if operac: w.append("UPPER(h.operacion) = %s"); a.append(operac) if cargos: w.append(self._where_cargo("h.cargo", cargos, a)) if nombre: w.append("UPPER(h.nombre_completo) LIKE %s") a.append(f"%{nombre}%") if sexo: w.append("UPPER(emp_sexo.sexo) = %s"); a.append(sexo) where_extra = (" AND " + " AND ".join(w)) if w else "" cur.execute(f""" SELECT h.nombre_completo AS nombres, COALESCE(u.hotel_name_api, h.razon_social) AS hotel_name, h.razon_social, h.cargo, h.tipo_contrato, h.sueldo_base::float AS sueldo, h.ingreso::text, h.termino::text AS vencimiento, h.causa_retiro, h.operacion, h.tipo_movimiento AS tipo FROM historico_ir h LEFT JOIN hotel_unificacion u ON u.razon_social = h.razon_social LEFT JOIN ( SELECT DISTINCT ON (codigo_empleado) codigo_empleado, sexo FROM historico_empleados WHERE sexo IS NOT NULL ORDER BY codigo_empleado, fecha DESC ) emp_sexo ON emp_sexo.codigo_empleado = h.codigo_empleado WHERE ( (h.tipo_movimiento = 'I' AND h.ingreso BETWEEN %s AND %s) OR (h.tipo_movimiento = 'R' AND h.termino BETWEEN %s AND %s) OR (h.tipo_movimiento = 'IR' AND (h.ingreso BETWEEN %s AND %s OR h.termino BETWEEN %s AND %s)) ) {where_extra} ORDER BY h.nombre_completo """, [fi, ff, fi, ff, fi, ff, fi, ff] + a) rows_ir = cur.fetchall() data = { "fecha_ini": fi, "fecha_fin": ff, "total": len(rows_ir), "rows": [dict(r) for r in rows_ir] } # ── /proxy/historico/activos-count ──────────────────────────── elif endpoint == "activos-count": fecha = qs.get("fecha", ["2026-01-01"])[0] razon = qs.get("razon_social", [""])[0] operac = qs.get("operacion", [""])[0].upper() centros = self._parse_list(qs, "centro") cargos = self._parse_list(qs, "cargo") nombre = qs.get("nombre", [""])[0].upper() sexo = qs.get("sexo", [""])[0].upper() w, a = ["fecha = %s"], [fecha] rw, ra = razon_where(razon) razones_list = [r.strip() for r in razon.split(',') if r.strip()] serv_patterns_here = [p for code in razones_list for p in SERV_HOTEL_PATTERNS.get(code, [])] all_serv_patterns = [p for pats in SERV_HOTEL_PATTERNS.values() for p in pats] if rw and all_serv_patterns: # Reclasificación real (no duplicado): un empleado SERV cuyo # centro_costo coincide con un hotel específico deja de contar # como SERV "genérico" y pasa a contar solo en ese hotel. excl_ors = " OR ".join(["UPPER(centro_costo) LIKE %s"] * len(all_serv_patterns)) clause = f"(({rw}) AND NOT (razon_social = 'SERV' AND ({excl_ors})))" params = list(ra) + [f"%{p}%" for p in all_serv_patterns] if serv_patterns_here: cc_ors = " OR ".join(["UPPER(centro_costo) LIKE %s"] * len(serv_patterns_here)) clause = f"({clause} OR (razon_social = 'SERV' AND ({cc_ors})))" params += [f"%{p}%" for p in serv_patterns_here] w.append(clause) a.extend(params) elif rw: w.append(rw); a.extend(ra) if operac: w.append("UPPER(operacion) = %s"); a.append(operac) if centros: w.append(self._where_in("centro_costo", centros, a)) if cargos: w.append(self._where_cargo("cargo", cargos, a)) if nombre: w.append("UPPER(nombre_completo) LIKE %s"); a.append(f"%{nombre}%") if sexo: w.append("UPPER(sexo) = %s"); a.append(sexo) cur.execute(f""" SELECT COUNT(*)::int AS activos FROM historico_empleados WHERE {' AND '.join(w)} """, a) row = cur.fetchone() data = {"activos": row["activos"] if row else 0} # ── /proxy/historico/antiguedad ─────────────────────────────── # ⚠️ Esta es la 2da de 3 formas de calcular antigüedad en el # proyecto (ver el comentario junto a calcMeses() en # dashboard-empleados.html para las otras 2). Alimenta la # gráfica "Distribución por Antigüedad". A propósito usa # MIN(fecha) agrupado SOLO por codigo_empleado — a diferencia # del endpoint ingreso-real-map (usado por la tabla/panel), NO # respeta cambios de razon_social ni huecos de empleo: un # empleado que salió y volvió años después puede aparecer con # antigüedad "continua" desde su primer ingreso histórico. No # es un bug sin arreglar — es una definición distinta, no # unificada todavía. No cambiar sin confirmar con las otras 2. elif endpoint == "antiguedad": fecha = qs.get("fecha", ["2026-03-01"])[0] razon = qs.get("razon_social", [""])[0] operac = qs.get("operacion", [""])[0].upper() centros = self._parse_list(qs, "centro") cargos = self._parse_list(qs, "cargo") nombre = qs.get("nombre", [""])[0].upper() w, a = ["fecha = %s"], [fecha] rw, ra = razon_where(razon) if rw: w.append(rw); a.extend(ra) if operac: w.append("UPPER(operacion) = %s"); a.append(operac) if centros: w.append(self._where_in("centro_costo", centros, a)) if cargos: w.append(self._where_cargo("cargo", cargos, a)) if nombre: w.append("UPPER(nombre_completo) LIKE %s"); a.append(f"%{nombre}%") cur.execute(f""" WITH activos AS ( SELECT DISTINCT codigo_empleado, INITCAP(sexo) AS sexo FROM historico_empleados WHERE {' AND '.join(w)} ), primera AS ( SELECT h.codigo_empleado, MIN(h.fecha) AS primera_fecha FROM historico_empleados h INNER JOIN activos a ON h.codigo_empleado = a.codigo_empleado GROUP BY h.codigo_empleado ), bins AS ( SELECT p.codigo_empleado, act.sexo, EXTRACT(YEAR FROM AGE(%s::date, p.primera_fecha)) * 12 + EXTRACT(MONTH FROM AGE(%s::date, p.primera_fecha)) AS meses FROM primera p JOIN activos act ON act.codigo_empleado = p.codigo_empleado ) SELECT CASE WHEN meses < 6 THEN '0–6 meses' WHEN meses < 12 THEN '6–12 meses' WHEN meses < 24 THEN '1–2 años' WHEN meses < 36 THEN '2–3 años' WHEN meses < 48 THEN '3–4 años' ELSE '4+ años' END AS rango, CASE WHEN meses < 6 THEN 0 WHEN meses < 12 THEN 1 WHEN meses < 24 THEN 2 WHEN meses < 36 THEN 3 WHEN meses < 48 THEN 4 ELSE 5 END AS orden, sexo, COUNT(*)::int AS total FROM bins GROUP BY rango, orden, sexo ORDER BY orden """, a + [fecha, fecha]) RANGOS = ['0–6 meses', '6–12 meses', '1–2 años', '2–3 años', '3–4 años', '4+ años'] rows_ant = cur.fetchall() lookup = {} gender_lookup = {r: {"Masculino": 0, "Femenino": 0} for r in RANGOS} for r in rows_ant: lookup[r['rango']] = lookup.get(r['rango'], 0) + r['total'] if r['sexo'] in ('Masculino', 'Femenino'): gender_lookup[r['rango']][r['sexo']] += r['total'] data = { "rangos": RANGOS, "counts": [lookup.get(r, 0) for r in RANGOS], "genero": [gender_lookup[r] for r in RANGOS] } # ── /proxy/historico/ir-mensual ─────────────────────────────── elif endpoint == "ir-mensual": razon = qs.get("razon_social", [""])[0] operac = qs.get("operacion", [""])[0].upper() cargos = self._parse_list(qs, "cargo") EXCL = "('ELMA','FNS')" wi, ai = [f"h.razon_social NOT IN {EXCL}", "h.tipo_movimiento IN ('I','IR')", "h.ingreso IS NOT NULL"], [] wr, ar = [f"h.razon_social NOT IN {EXCL}", "h.tipo_movimiento IN ('R','IR')", "h.termino IS NOT NULL"], [] rw, ra = razon_where(razon, col='h.razon_social') if rw: wi.append(rw); ai.extend(ra) wr.append(rw); ar.extend(ra) if operac: wi.append("UPPER(h.operacion) = %s"); ai.append(operac) wr.append("UPPER(h.operacion) = %s"); ar.append(operac) if cargos: wi.append(self._where_cargo("h.cargo", cargos, ai)) wr.append(self._where_cargo("h.cargo", cargos, ar)) cur.execute(f""" WITH ing AS ( SELECT DATE_TRUNC('month', h.ingreso)::date AS mes, COUNT(*)::int AS n FROM historico_ir h WHERE {' AND '.join(wi)} GROUP BY 1 ), ret AS ( SELECT DATE_TRUNC('month', h.termino)::date AS mes, COUNT(*)::int AS n FROM historico_ir h WHERE {' AND '.join(wr)} GROUP BY 1 ), meses AS ( SELECT mes FROM ing UNION SELECT mes FROM ret ) SELECT TO_CHAR(m.mes, 'YYYY-MM') AS mes, COALESCE(i.n, 0) AS ingresos, COALESCE(r.n, 0) AS retiros FROM meses m LEFT JOIN ing i ON i.mes = m.mes LEFT JOIN ret r ON r.mes = m.mes ORDER BY m.mes """, ai + ar) data = {"rows": [dict(r) for r in cur.fetchall()]} # ── /proxy/historico/cohorte-retencion ─────────────────────── elif endpoint == "cohorte-retencion": razon = qs.get("razon_social", [""])[0] operac = qs.get("operacion", [""])[0].upper() EXCL = "('ELMA','FNS')" w, a = [f"razon_social NOT IN {EXCL}", "fecha >= '2024-01-01'"], [] rw, ra = razon_where(razon) if rw: w.append(rw); a.extend(ra) if operac: w.append("UPPER(operacion) = %s"); a.append(operac) where_base = ' AND '.join(w) # Cohorte = empleados que aparecen en snapshot mes X pero NO en mes X-1 cur.execute(f""" WITH nuevos AS ( SELECT e.codigo_empleado, e.fecha AS mes_cohort FROM historico_empleados e WHERE {where_base} AND NOT EXISTS ( SELECT 1 FROM historico_empleados p WHERE p.codigo_empleado = e.codigo_empleado AND p.fecha = (e.fecha - INTERVAL '1 month')::date AND p.razon_social NOT IN {EXCL} ) ), snaps AS ( SELECT DISTINCT codigo_empleado, fecha FROM historico_empleados WHERE {where_base} ), cohortes AS ( SELECT n.mes_cohort, COUNT(DISTINCT n.codigo_empleado) AS total, COUNT(DISTINCT n.codigo_empleado) FILTER (WHERE s1.codigo_empleado IS NOT NULL) AS m1, COUNT(DISTINCT n.codigo_empleado) FILTER (WHERE s2.codigo_empleado IS NOT NULL) AS m2, COUNT(DISTINCT n.codigo_empleado) FILTER (WHERE s3.codigo_empleado IS NOT NULL) AS m3, COUNT(DISTINCT n.codigo_empleado) FILTER (WHERE s6.codigo_empleado IS NOT NULL) AS m6, COUNT(DISTINCT n.codigo_empleado) FILTER (WHERE s9.codigo_empleado IS NOT NULL) AS m9, COUNT(DISTINCT n.codigo_empleado) FILTER (WHERE s12.codigo_empleado IS NOT NULL) AS m12 FROM nuevos n LEFT JOIN snaps s1 ON s1.codigo_empleado = n.codigo_empleado AND s1.fecha = (n.mes_cohort + INTERVAL '1 month')::date LEFT JOIN snaps s2 ON s2.codigo_empleado = n.codigo_empleado AND s2.fecha = (n.mes_cohort + INTERVAL '2 months')::date LEFT JOIN snaps s3 ON s3.codigo_empleado = n.codigo_empleado AND s3.fecha = (n.mes_cohort + INTERVAL '3 months')::date LEFT JOIN snaps s6 ON s6.codigo_empleado = n.codigo_empleado AND s6.fecha = (n.mes_cohort + INTERVAL '6 months')::date LEFT JOIN snaps s9 ON s9.codigo_empleado = n.codigo_empleado AND s9.fecha = (n.mes_cohort + INTERVAL '9 months')::date LEFT JOIN snaps s12 ON s12.codigo_empleado = n.codigo_empleado AND s12.fecha = (n.mes_cohort + INTERVAL '12 months')::date GROUP BY n.mes_cohort ) SELECT TO_CHAR(mes_cohort, 'YYYY-MM') AS cohorte, total, ROUND(m1 * 100.0 / NULLIF(total,0), 1)::float AS pct_1, ROUND(m2 * 100.0 / NULLIF(total,0), 1)::float AS pct_2, ROUND(m3 * 100.0 / NULLIF(total,0), 1)::float AS pct_3, ROUND(m6 * 100.0 / NULLIF(total,0), 1)::float AS pct_6, ROUND(m9 * 100.0 / NULLIF(total,0), 1)::float AS pct_9, ROUND(m12 * 100.0 / NULLIF(total,0), 1)::float AS pct_12 FROM cohortes WHERE total >= 8 ORDER BY mes_cohort """, a + a) # where_base usado 2 veces (nuevos + snaps) data = {"rows": [dict(r) for r in cur.fetchall()]} # ── /proxy/historico/rotacion-heatmap ──────────────────────── elif endpoint == "rotacion-heatmap": razon = qs.get("razon_social", [""])[0] EXCL_H = "('ELMA','FNS')" w, a = [f"razon_social NOT IN {EXCL_H}", "fecha >= '2024-01-01'"], [] rw, ra = razon_where(razon) if rw: w.append(rw); a.extend(ra) cur.execute(f""" WITH base AS ( SELECT razon_social, fecha, COUNT(DISTINCT codigo_empleado) FILTER (WHERE operacion ILIKE '%%retiro%%') AS retiros, COUNT(DISTINCT codigo_empleado) FILTER (WHERE operacion ILIKE '%%operacion%%') AS activos FROM historico_empleados WHERE {' AND '.join(w)} GROUP BY razon_social, fecha ) SELECT razon_social, TO_CHAR(fecha, 'YYYY-MM') AS mes, ROUND(retiros * 100.0 / NULLIF(activos, 0), 1)::float AS pct FROM base ORDER BY razon_social, fecha """, a) data = {"rows": [dict(r) for r in cur.fetchall()]} # ── /proxy/historico/activos-hotel ──────────────────────────── elif endpoint == "activos-hotel": fecha = qs.get("fecha", ["2026-01-01"])[0] razon = qs.get("razon_social", [""])[0] operac = qs.get("operacion", [""])[0].upper() centros = self._parse_list(qs, "centro") cargos = self._parse_list(qs, "cargo") nombre = qs.get("nombre", [""])[0].upper() EXCL_H = "('ELMA','FNS')" w, a = ["fecha = %s", f"razon_social NOT IN {EXCL_H}"], [fecha] rw, ra = razon_where(razon) if rw: w.append(rw); a.extend(ra) if operac: w.append("UPPER(operacion) = %s"); a.append(operac) if centros: w.append(self._where_in("centro_costo", centros, a)) if cargos: w.append(self._where_cargo("cargo", cargos, a)) if nombre: w.append("UPPER(nombre_completo) LIKE %s"); a.append(f"%{nombre}%") cur.execute(f""" SELECT razon_social, COUNT(DISTINCT codigo_empleado)::int AS activos FROM historico_empleados WHERE {' AND '.join(w)} GROUP BY razon_social ORDER BY activos DESC """, a) data = {"rows": [dict(r) for r in cur.fetchall()]} cur.close() body = json.dumps(data, default=str, ensure_ascii=False).encode("utf-8") self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self._send_cors() self.end_headers() self.wfile.write(body) except Exception as e: self.send_response(500) self.send_header("Content-Type", "application/json") self._send_cors() self.end_headers() self.wfile.write(json.dumps(self._error_payload(e)).encode()) finally: DB_POOL.putconn(conn) # ─── Estadísticas comparativas por hotel ──────────────────────────────── def _stats(self): """GET /proxy/stats/resumen-hoteles?ano=2025&mes=1 Devuelve un resumen por hotel: empleados, costo nómina, ingresos P&G. Usa nómina viva cuando está disponible y cae a la caché local como respaldo. """ from urllib.parse import urlparse, parse_qs parsed = urlparse(self.path) qs = parse_qs(parsed.query) endpoint = parsed.path.split("/proxy/stats/")[-1] conn = DB_POOL.getconn() _conn_returned = False try: cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) data = {} if endpoint == "resumen-hoteles": ano = int(qs.get("ano", ["2025"])[0]) mes = int(qs.get("mes", ["1"])[0]) label = qs.get("label", [""])[0].strip() # filtro por nombre_personalizado fecha = date_type(ano, mes, 1).isoformat() # 1. Hoteles — keyed by label para soportar múltiples hoteles con el mismo NIT # (Belvedere/FC Bogotá NIT 900345450, CasaCanabal/FC Cucutá NIT 900361237, # GHS/The Club NIT 900657316) hotel_where = "" hotel_args = [] if label: hotel_where = "AND LOWER(h.nombre_personalizado) = LOWER(%s)" hotel_args = [label] cur.execute(f""" WITH hotel_bus AS ( SELECT h2.nombre_personalizado, ARRAY_AGG(LOWER(COALESCE(h2.bu_codigo, 'local')) ORDER BY h2.bu_codigo ASC NULLS LAST) AS all_bus FROM hotel_unificacion h2 JOIN empresas e2 ON h2.razon_social = e2.codigo WHERE e2.nit IS NOT NULL AND e2.nit <> '' AND h2.hotel_name_api IS NOT NULL AND h2.nombre_personalizado NOT IN ('No mostrar') GROUP BY h2.nombre_personalizado ) SELECT DISTINCT ON (h.nombre_personalizado) h.nombre_personalizado AS label, e.nit, e.codigo AS razon_social, h.bu_codigo, LOWER(h.bu_codigo) AS bu_lower, h.bu_nombre, e.pyg_company_id AS company_id, hb.all_bus FROM hotel_unificacion h JOIN empresas e ON h.razon_social = e.codigo JOIN hotel_bus hb ON h.nombre_personalizado = hb.nombre_personalizado WHERE e.nit IS NOT NULL AND e.nit <> '' AND h.hotel_name_api IS NOT NULL AND h.nombre_personalizado NOT IN ('No mostrar') {hotel_where} ORDER BY h.nombre_personalizado, h.bu_codigo ASC NULLS LAST """, hotel_args) hoteles = {r["label"]: dict(r) for r in cur.fetchall()} nit_counts = {} for h in hoteles.values(): nit_counts[h["nit"]] = nit_counts.get(h["nit"], 0) + 1 # 2. Mapa de conceptos para replicar el pool del tab Payroll. cur.execute(""" SELECT codigo, regexp_replace(codigo, '_v[0-9]+$', '') AS codigo_api, nombre, grupo, subgrupo, incluir FROM nomina_conceptos """) conceptos_rows = [dict(r) for r in cur.fetchall()] conceptos_by_code = {str(r["codigo"]): r for r in conceptos_rows} conceptos_by_api_name = {} for r in conceptos_rows: api_key = str(r.get("codigo_api") or r.get("codigo") or "").upper() nom_key = str(r.get("nombre") or "").upper().strip() conceptos_by_api_name[f"{api_key}|{nom_key}"] = r def _nom_lookup(codigo, nombre): code = str(codigo or "") api_key = code.upper() nom_key = str(nombre or "").upper().strip() return conceptos_by_api_name.get(f"{api_key}|{nom_key}") or conceptos_by_code.get(code) def _nom_incluir(codigo, nombre): m = _nom_lookup(codigo, nombre) return bool(m and m.get("incluir") is not False) def _nom_grupo(codigo, nombre): m = _nom_lookup(codigo, nombre) return (m or {}).get("grupo") or "" def _nom_subgrupo(codigo, nombre): m = _nom_lookup(codigo, nombre) return (m or {}).get("subgrupo") or "" def _build_ss_patronal(emps): ss_map = {} for emp in emps: for ss in emp.get("seguridad_social") or []: ap = float(ss.get("aporte_empleador") or 0) if ap <= 0: continue code = "SS_" + str(ss.get("codigo") or "") if code not in ss_map: ss_map[code] = {"codigo": code, "nombre": ss.get("nombre") or "", "valor": 0.0} ss_map[code]["valor"] += ap return list(ss_map.values()) def _nomina_pool_from_raw(raw, bu, use_raw_aggregates=False): raw = raw or {} bu_lower = str(bu or "local").lower() emps = [ e for e in (raw.get("empleados") or []) if str(e.get("bu_codigo") or "").lower() == bu_lower ] if not emps and bu_lower == "local": emps = [ e for e in (raw.get("empleados") or []) if str(e.get("bu_codigo") or "local").lower() in ("", "local") ] total_devengos = sum(float(e.get("total_devengos") or 0) for e in emps) raw_dev = float((raw.get("resumen") or {}).get("total_devengos") or 0) ratio = total_devengos / raw_dev if raw_dev > 0 else 0 pool = {} c_map = {} if use_raw_aggregates: for c in raw.get("conceptos") or []: code = str(c.get("codigo") or "") name = c.get("nombre") or "" key = code + "|" + name.upper() c_map[key] = { "codigo": code, "nombre": name, "devengos": float(c.get("devengos") or 0), "deducciones": float(c.get("deducciones") or 0), } else: for emp in emps: for c in emp.get("conceptos") or []: code = str(c.get("codigo") or "") name = c.get("nombre") or "" key = code + "|" + name.upper() if key not in c_map: c_map[key] = {"codigo": code, "nombre": name, "devengos": 0.0, "deducciones": 0.0} if code.startswith("DX"): c_map[key]["deducciones"] += float(c.get("valor") or 0) else: c_map[key]["devengos"] += float(c.get("valor") or 0) for item in c_map.values(): if not _nom_incluir(item["codigo"], item["nombre"]): continue val = item["devengos"] if item["devengos"] > 0 else item["deducciones"] if val <= 0: continue key = item["codigo"] + "|" + item["nombre"].upper() pool[key] = { "codigo": item["codigo"], "nombre": item["nombre"], "val": val, "grupo": _nom_grupo(item["codigo"], item["nombre"]), "subgrupo": _nom_subgrupo(item["codigo"], item["nombre"]), } provision_ratio = 1 if use_raw_aggregates else ratio for p in raw.get("provisiones") or []: code = str(p.get("codigo") or "") name = p.get("nombre") or "" if not _nom_incluir(code, name): continue val = round(float(p.get("valor") or 0) * provision_ratio) if val <= 0: continue key = code + "|" + name.upper() pool[key] = { "codigo": code, "nombre": name, "val": val, "grupo": _nom_grupo(code, name), "subgrupo": _nom_subgrupo(code, name), } ss_patronal = _build_ss_patronal(emps) for ss in ss_patronal: code = ss["codigo"] name = ss["nombre"] if not _nom_incluir(code, name): continue val = float(ss.get("valor") or 0) if val <= 0: continue key = code + "|" + name.upper() pool[key] = { "codigo": code, "nombre": name, "val": val, "grupo": _nom_grupo(code, name), "subgrupo": _nom_subgrupo(code, name), } return { "pool": pool, "num_empleados": len(emps), "dias_nomina": sum(float(e.get("dias") or 0) for e in emps), "devengos": total_devengos, "ss_patronal": sum(float(x.get("valor") or 0) for x in ss_patronal), } def _payroll_total(pool, pyg_items): pool = {k: dict(v) for k, v in (pool or {}).items()} for x in pyg_items or []: if x.get("nomina_backed"): continue val = float(x.get("val") or 0) if val <= 0: continue pool["PYG_" + str(x.get("id"))] = { "codigo": "PYG_" + str(x.get("id")), "nombre": x.get("nombre") or "", "val": val, "grupo": x.get("grupo") or "", "subgrupo": x.get("subgrupo") or "", } return sum(float(x.get("val") or 0) for x in pool.values()) # 3. Nómina cruda por NIT para reconstruir conceptos igual que Payroll cur.execute(""" SELECT DISTINCT ON (nit) nit, raw_data FROM historico_nomina_pyg WHERE fuente='nomina' AND registro_tipo='respuesta' AND ano=%s AND mes=%s ORDER BY nit, cargado_en DESC """, (ano, mes)) nomina_raw_map = {r["nit"]: r["raw_data"] for r in cur.fetchall()} def _fetch_nomina_live(nit): body = json.dumps({ "hotel_nit": nit, "ano": ano, "mes": mes, "incluir_empleados": True, "ss_vista": "ambos", }).encode() req = urllib.request.Request( API_BASE + "/nomina/reporte", data=body, headers={"X-API-Key": API_KEY, "Content-Type": "application/json"}, method="POST" ) with urllib.request.urlopen(req, timeout=30, context=SSL_CTX) as resp: return json.loads(resp.read()) if not self._is_periodo_historico(ano, mes): for nit in {h["nit"] for h in hoteles.values() if h.get("nit")}: try: nomina_raw_map[nit] = _fetch_nomina_live(nit) except Exception as e: print(f" ⚠️ nómina en vivo falló para NIT {nit} ({ano}-{mes:02d}): {e}") # 4. Empleados, días, devengos y SS patronal por (nit, bu) desde respuesta nómina # Filtra correctamente hoteles con NIT compartido (Belvedere=be, FC Bogotá=local…) cur.execute(""" WITH latest_nomina AS ( SELECT DISTINCT ON (nit) nit, raw_data FROM historico_nomina_pyg WHERE fuente='nomina' AND registro_tipo='respuesta' AND ano=%s AND mes=%s ORDER BY nit, cargado_en DESC ) SELECT nit, LOWER(COALESCE(emp->>'bu_codigo', 'local')) AS bu, COUNT(*)::int AS num_empleados, COALESCE(SUM((emp->>'dias')::float), 0) AS dias_nomina, COALESCE(SUM((emp->>'total_devengos')::float), 0) AS devengos, COALESCE(SUM( (SELECT COALESCE(SUM((ss->>'aporte_empleador')::float), 0) FROM jsonb_array_elements(emp->'seguridad_social') AS ss) ), 0) AS ss_patronal FROM latest_nomina, jsonb_array_elements(raw_data->'empleados') AS emp GROUP BY nit, bu """, (ano, mes)) nomina_bu_map = {(r["nit"], r["bu"]): dict(r) for r in cur.fetchall()} # 5. Ingresos del P&G por (company_id, bu) — itera todas las BU del árbol P&G cur.execute(""" SELECT company_id::int AS cid, LOWER(bu_kv.key) AS bu, MAX(CASE WHEN t->>'id' = '1-ingresos' THEN (t->'monthly'->'real'->>'value')::float END) AS ingresos_mes FROM historico_nomina_pyg, jsonb_each(raw_data->'pnl') AS bu_kv, jsonb_array_elements(bu_kv.value->'tree') AS t WHERE fuente='pyg' AND registro_tipo='respuesta' AND ano=%s AND mes=%s GROUP BY company_id, bu_kv.key """, (ano, mes)) ingresos_map = {(r["cid"], r["bu"]): r["ingresos_mes"] or 0 for r in cur.fetchall()} # 6. Conceptos P&G cacheados, misma fuente que /proxy/pyg/reporte usa en Payroll. cur.execute(""" SELECT company_id::int AS cid, LOWER(COALESCE(bu_codigo,'')) AS bu, concepto_codigo AS id, concepto_nombre AS nombre, grupo, subgrupo, SUM(valor)::float AS val, BOOL_OR(COALESCE((raw_data->>'nomina_backed')::boolean, false)) AS nomina_backed FROM historico_nomina_pyg WHERE fuente='pyg' AND registro_tipo='concepto' AND ano=%s AND mes=%s GROUP BY company_id, LOWER(COALESCE(bu_codigo,'')), concepto_codigo, concepto_nombre, grupo, subgrupo """, (ano, mes)) pyg_conceptos_map = {} for r in cur.fetchall(): pyg_conceptos_map.setdefault((r["cid"], r["bu"] or "local"), []).append({ "id": r["id"], "nombre": r["nombre"], "grupo": r["grupo"], "subgrupo": r["subgrupo"], "val": r["val"] or 0, "nomina_backed": r["nomina_backed"], }) def _fetch_pyg_live(cid, bu): bu = (bu or "local").lower() period = f"{ano}{mes:02d}" def _request_live(bu_code): body = json.dumps({ "company_id": cid, "period": period, "scenario_id": None, "business_units": [bu_code], "include_kpis": False, }).encode() req = urllib.request.Request( PYG_URL, data=body, headers={"X-API-Key": PYG_KEY, "Content-Type": "application/json"}, method="POST" ) with urllib.request.urlopen(req, timeout=30, context=SSL_CTX) as resp: return json.loads(resp.read()) raw = _request_live(bu) def _flat_val(node): real = node.get("monthly", {}).get("real", {}) if isinstance(real, dict): for v in real.values(): if isinstance(v, (int, float)): return float(v) if isinstance(v, dict): return float(v.get("mes", v.get("value", 0)) or 0) return 0.0 def _flatten(nodes, out): for n in nodes: out[n.get("id", "")] = _flat_val(n) _flatten(n.get("children", []), out) flat = {} pnl = raw.get("pnl", {}) tree = (pnl.get(bu) or pnl.get("local") or {}).get("tree", []) if not tree and bu != "local": raw = _request_live("local") pnl = raw.get("pnl", {}) tree = pnl.get("local", {}).get("tree", []) _flatten(tree, flat) conceptos_def = [ ("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), ("dotación_y_suministro_a_trabajadores", "Dotación y Suministro", "Otros Gastos Personal", "Dotación", False), ("dotacion_y_suministro_a_trabajadores", "Dotación y Suministro", "Otros Gastos Personal", "Dotación", False), ("comida_empleados", "Comida Empleados", "Otros Gastos Personal", "Bienestar", False), ("transporte_de_personal", "Transporte de Personal", "Otros Gastos Personal", "Bienestar", False), ("capacitación_al_personal", "Capacitación al Personal", "Otros Gastos Personal", "Bienestar", False), ("capacitacion_al_personal", "Capacitación al Personal", "Otros Gastos Personal", "Bienestar", False), ("gastos_medicos_y_medicamentos", "Gastos Médicos y Medicamentos", "Otros Gastos Personal", "Gastos Médicos", False), ("salud_ocupacional_y_examenes_d", "Salud Ocupacional y Exámenes", "Otros Gastos Personal", "Gastos Médicos", False), ] seen = set() conceptos = [] for node_id, nombre, grupo, subgrupo, nomina_backed in conceptos_def: if node_id in seen or node_id not in flat: continue seen.add(node_id) seen.add(node_id.replace("ó", "o").replace("á", "a").replace("é", "e").replace("ú", "u")) conceptos.append({ "id": node_id, "nombre": nombre, "grupo": grupo, "subgrupo": subgrupo, "val": flat.get(node_id, 0) or 0, "nomina_backed": nomina_backed, }) return flat.get("1-ingresos", 0) or 0, conceptos if not self._is_periodo_historico(ano, mes): ingresos_map = {} pyg_conceptos_map = {} for h in hoteles.values(): cid = h.get("company_id") bu = h.get("bu_lower") or "local" if not cid: continue try: ingresos, conceptos = _fetch_pyg_live(cid, bu) ingresos_map[(cid, bu)] = ingresos pyg_conceptos_map[(cid, bu)] = conceptos except Exception as e: print(f" ⚠️ P&G en vivo falló para company_id={cid} bu={bu} ({ano}-{mes:02d}): {e}") else: for h in hoteles.values(): cid = h.get("company_id") bu = h.get("bu_lower") or "local" if not cid or bu == "local": continue if ingresos_map.get((cid, bu), 0) > 0 or pyg_conceptos_map.get((cid, bu)): continue try: ingresos, conceptos = _fetch_pyg_live(cid, "local") if ingresos > 0 or conceptos: ingresos_map[(cid, bu)] = ingresos pyg_conceptos_map[(cid, bu)] = conceptos except Exception as e: print(f" ⚠️ P&G en vivo (fallback local) falló para company_id={cid} bu={bu} ({ano}-{mes:02d}): {e}") # 6b. Completar pyg_conceptos_map desde respuesta cacheada (meses sin concepto records) if self._is_periodo_historico(ano, mes): need_pairs = [ (h["company_id"], h.get("bu_lower") or "local") for h in hoteles.values() if h.get("company_id") and not pyg_conceptos_map.get((h["company_id"], h.get("bu_lower") or "local")) ] if need_pairs: need_cids = list({cid for cid, _ in need_pairs}) cur.execute(""" SELECT company_id::int AS cid, bu_codigo, raw_data FROM historico_nomina_pyg WHERE fuente='pyg' AND registro_tipo='respuesta' AND ano=%s AND mes=%s AND company_id = ANY(%s) ORDER BY company_id, cargado_en DESC """, (ano, mes, need_cids)) resp_by_cid = {} for rr in cur.fetchall(): k = (rr["cid"], (rr["bu_codigo"] or "local").lower()) if k not in resp_by_cid: resp_by_cid[k] = rr["raw_data"] def _fv(node): real = node.get("monthly", {}).get("real", {}) if isinstance(real, dict): for v in real.values(): if isinstance(v, (int, float)): return float(v) if isinstance(v, dict): return float(v.get("mes", v.get("value", 0)) or 0) return 0.0 def _fl(nodes, out): for n in nodes: out[n.get("id", "")] = _fv(n) _fl(n.get("children", []), out) CDEF = [ ("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), ("dotación_y_suministro_a_trabajadores", "Dotación y Suministro", "Otros Gastos Personal", "Dotación", False), ("dotacion_y_suministro_a_trabajadores", "Dotación y Suministro", "Otros Gastos Personal", "Dotación", False), ("comida_empleados", "Comida Empleados", "Otros Gastos Personal", "Bienestar", False), ("transporte_de_personal", "Transporte de Personal", "Otros Gastos Personal", "Bienestar", False), ("capacitación_al_personal", "Capacitación al Personal", "Otros Gastos Personal", "Bienestar", False), ("capacitacion_al_personal", "Capacitación al Personal", "Otros Gastos Personal", "Bienestar", False), ("gastos_medicos_y_medicamentos", "Gastos Médicos y Medicamentos", "Otros Gastos Personal", "Gastos Médicos", False), ("salud_ocupacional_y_examenes_d", "Salud Ocupacional y Exámenes", "Otros Gastos Personal", "Gastos Médicos", False), ] for (cid, bu) in need_pairs: rd = resp_by_cid.get((cid, bu)) or resp_by_cid.get((cid, "local")) if not rd: continue pnl = rd.get("pnl", {}) tree = (pnl.get(bu) or pnl.get("local") or next(iter(pnl.values()), {})).get("tree", []) flat = {} _fl(tree, flat) flat_s = {} for k, v in flat.items(): s = k.split("/")[-1] if s not in flat_s: flat_s[s] = v seen, cs = set(), [] for nid, nombre, grupo, subgrupo, nb in CDEF: if nid in seen: continue val = flat.get(nid) or flat_s.get(nid) if val is None: continue seen.add(nid) seen.add(nid.replace("ó","o").replace("á","a").replace("é","e").replace("ú","u")) cs.append({"id": nid, "nombre": nombre, "grupo": grupo, "subgrupo": subgrupo, "val": float(val) or 0, "nomina_backed": nb}) if cs: pyg_conceptos_map[(cid, bu)] = cs # 7. Outsourcing desde respuesta nómina Servihoteles, filtrado por centro_costo # Mismos patrones que PR_HOTEL_SERV_PATTERNS en el frontend (Payroll tab) SERV_NIT = '901418135' SERV_PATTERNS = { 'Hotel Caribe': ['HCARIBE', 'H CARIBE', 'H CARIBE VENTAS'], 'The Club': ['THE CLUB CTG'], 'La Mantilla': ['MANTILLA'], 'Casa Don Luis': ['CDL'], 'FC Cartagena': ['FARANDA COLLECTION CARTAGENA'], } cur.execute(""" SELECT emp->>'centro_costo' AS cc, (emp->>'dias')::float AS dias FROM ( SELECT raw_data FROM historico_nomina_pyg WHERE fuente='nomina' AND registro_tipo='respuesta' AND nit=%s AND ano=%s AND mes=%s ORDER BY cargado_en DESC LIMIT 1 ) latest, jsonb_array_elements(latest.raw_data->'empleados') AS emp """, (SERV_NIT, ano, mes)) serv_by_label = {} # {label: {'emps': int, 'dias': float}} for r in cur.fetchall(): cc = (r["cc"] or "").upper() for lbl, pats in SERV_PATTERNS.items(): if any(p in cc for p in pats): if lbl not in serv_by_label: serv_by_label[lbl] = {'emps': 0, 'dias': 0.0} serv_by_label[lbl]['emps'] += 1 serv_by_label[lbl]['dias'] += float(r.get('dias') or 0) break # Servihoteles "Otros": total nómina − asignados a otros hoteles serv_total_nom = sum( v["num_empleados"] for (n, b), v in nomina_bu_map.items() if n == SERV_NIT ) serv_otros = max(0, serv_total_nom - sum(v['emps'] for v in serv_by_label.values())) cur.close(); DB_POOL.putconn(conn); _conn_returned = True # 6. Combinar todo — iterar por label para no perder hoteles con NIT compartido dias_mes = (date_type(ano, mes % 12 + 1, 1) - date_type(ano, mes, 1)).days if mes < 12 \ else (date_type(ano + 1, 1, 1) - date_type(ano, 12, 1)).days rows = [] for lbl, h in hoteles.items(): nit = h["nit"] cid = h.get("company_id") bu = h.get("bu_lower") or "local" nom_bu = nomina_bu_map.get((nit, bu), nomina_bu_map.get((nit, "local"), {})) # Dedup all_bus para hoteles con múltiples filas en hotel_unificacion raw_bus = h.get("all_bus") or [bu] all_bus = list(dict.fromkeys(b or 'local' for b in raw_bus)) if nit == SERV_NIT: emps_direct = serv_otros emps_serv = 0 dias_dir = 0 dias_serv_nom = 0.0 else: if len(all_bus) > 1: # Sumar empleados/días de todas las BUs (p.ej. GHS Local+TC) emps_direct = sum(nomina_bu_map.get((nit, b), {}).get("num_empleados") or 0 for b in all_bus) dias_dir = sum(nomina_bu_map.get((nit, b), {}).get("dias_nomina") or 0 for b in all_bus) else: emps_direct = nom_bu.get("num_empleados") or 0 dias_dir = nom_bu.get("dias_nomina") or 0 serv_info = serv_by_label.get(lbl, {'emps': 0, 'dias': 0.0}) emps_serv = serv_info['emps'] dias_serv_nom = serv_info['dias'] # días reales de outsourcing desde su nómina emps = (emps_direct + emps_serv) if (emps_direct + emps_serv) > 0 else None # Días Laborados = total empleados × días del mes dias_laborados = int((emps_direct + emps_serv) * dias_mes) if emps else 0 # Días Promedio = suma de días reales de nómina (directos + outsourcing) / total empleados dias_nom_total = dias_dir + dias_serv_nom dias_promedio = round(dias_nom_total / (emps_direct + emps_serv)) \ if (emps_direct + emps_serv) > 0 else None nom_calc = _nomina_pool_from_raw( nomina_raw_map.get(nit), bu, use_raw_aggregates=(nit_counts.get(nit, 0) == 1) ) pyg_items = (pyg_conceptos_map.get((cid, bu), []) if cid else []) costo = _payroll_total(nom_calc["pool"], pyg_items) devengos = nom_calc["devengos"] or nom_bu.get("devengos") or 0 ingresos = (ingresos_map.get((cid, bu), 0) or 0) if cid else 0 rows.append({ "label": lbl, "nit": nit, "company_id": cid, "bu_codigo": h["bu_codigo"], "empleados": emps, "emps_direct": emps_direct, "emps_outsourc": emps_serv, "dias_laborados": dias_laborados, "dias_promedio": dias_promedio, "costo_total": costo, "devengos": devengos, "ingresos": ingresos, "pct_nomina": round(costo / ingresos * 100, 2) if ingresos > 0 else None, }) # Ordenar por costo total desc rows.sort(key=lambda r: r["costo_total"] or 0, reverse=True) # Totales tot_costo = sum(r["costo_total"] for r in rows) tot_devengos = sum(r["devengos"] for r in rows) tot_ingresos = sum(r["ingresos"] for r in rows) tot_emps = sum(r["empleados"] or 0 for r in rows) tot_dias = sum(r["dias_laborados"] or 0 for r in rows) # Días promedio total: promedio ponderado (suma días nómina reales / total empleados) tot_dias_nom = sum( (r["dias_promedio"] or 0) * (r["empleados"] or 0) for r in rows ) tot_dias_prom = round(tot_dias_nom / tot_emps) if tot_emps > 0 else None data = { "ano": ano, "mes": mes, "rows": rows, "totales": { "empleados": tot_emps, "dias_laborados": tot_dias, "dias_promedio": tot_dias_prom, "costo_total": tot_costo, "devengos": tot_devengos, "ingresos": tot_ingresos, "pct_nomina": round(tot_costo / tot_ingresos * 100, 2) if tot_ingresos > 0 else None, } } elif endpoint == "kpis-agregado": # GET /proxy/stats/kpis-agregado?ano=2026&mes=5 # Fetches KPIs for all hotels sequentially (to avoid 429) and returns aggregated sums import time as _time, calendar as _cal ano_k = int(qs.get("ano", ["2025"])[0]) mes_k = int(qs.get("mes", ["1"])[0]) cur.execute(""" SELECT DISTINCT h.nombre_personalizado AS label, e.pyg_company_id AS company_id, LOWER(h.bu_codigo) AS bu_codigo FROM hotel_unificacion h JOIN empresas e ON h.razon_social = e.codigo WHERE e.pyg_company_id IS NOT NULL AND h.nombre_personalizado NOT IN ('No mostrar') AND h.hotel_name_api IS NOT NULL ORDER BY h.nombre_personalizado """) hoteles_kpi = cur.fetchall() cur.close(); DB_POOL.putconn(conn); _conn_returned = True period = f"{ano_k}{mes_k:02d}" agg = {"rooms_total": 0, "rooms_occupied": 0, "hab_fis": 0, "guests": 0, "days_in_month": _cal.monthrange(ano_k, mes_k)[1]} valid_count = 0 for h in hoteles_kpi: cid = h["company_id"] bu_code = (h["bu_codigo"] or "local").lower() try: body_p = json.dumps({ "company_id": int(cid), "period": period, "scenario_id": None, "business_units": [bu_code], "include_kpis": True }).encode() req2 = urllib.request.Request( PYG_URL, data=body_p, headers={"X-API-Key": PYG_KEY, "Content-Type": "application/json"}, method="POST" ) with urllib.request.urlopen(req2, timeout=30, context=SSL_CTX) as resp2: raw2 = json.loads(resp2.read()) snaps2 = raw2.get("kpi_snapshots_by_bu", {}) snap_bu2 = snaps2.get(bu_code, snaps2.get("local", {})) snap2 = snap_bu2.get("real", {}).get(period, {}) def kv2(k): return (snap2.get(k) or {}).get("mes", 0) or 0 agg["rooms_total"] += kv2("rooms_total") agg["rooms_occupied"] += kv2("rooms_occupied") agg["guests"] += kv2("guests_total") agg["hab_fis"] += raw2.get("habitaciones_fisicas", 0) or 0 valid_count += 1 _time.sleep(0.15) # evitar 429 — 150 ms entre llamadas except Exception as e: print(f" ⚠️ KPI ocupación falló para {h.get('label')} ({ano_k}-{mes_k:02d}): {e}") _time.sleep(0.3) # back-off en error continue occ = round(agg["rooms_occupied"] / agg["rooms_total"] * 100, 2) \ if agg["rooms_total"] > 0 else None data = { "hoteles": valid_count, "days_in_month": agg["days_in_month"], "habitaciones_fisicas": agg["hab_fis"], "rooms_total": agg["rooms_total"], "rooms_occupied": agg["rooms_occupied"], "occupancy_pct": occ, "guests_total": agg["guests"], } elif endpoint == "kpis-hotel": # GET /proxy/stats/kpis-hotel?company_id=7&ano=2025&mes=1&bu_codigo=Local # Llama a la API P&G real con include_kpis=true y devuelve el snapshot operativo company_id = qs.get("company_id", [""])[0] ano_k = int(qs.get("ano", ["2025"])[0]) mes_k = int(qs.get("mes", ["1"])[0]) bu_codigo = qs.get("bu_codigo", ["local"])[0].lower() or "local" if not company_id: data = {"error": "Falta company_id"} else: period = f"{ano_k}{mes_k:02d}" pyg_body = json.dumps({ "company_id": int(company_id), "period": period, "scenario_id": None, "business_units": [bu_codigo], "include_kpis": True }).encode() cur.close(); DB_POOL.putconn(conn); _conn_returned = True req = urllib.request.Request( PYG_URL, data=pyg_body, headers={"X-API-Key": PYG_KEY, "Content-Type": "application/json"}, method="POST" ) with urllib.request.urlopen(req, timeout=30, context=SSL_CTX) as resp: raw = json.loads(resp.read()) # Extraer snapshot del período snaps = raw.get("kpi_snapshots_by_bu", {}) snap_bu = snaps.get(bu_codigo, snaps.get("local", {})) snap = snap_bu.get("real", {}).get(period, {}) hab_fis = raw.get("habitaciones_fisicas", 0) def kv(key): return (snap.get(key) or {}).get("mes", 0) or 0 import calendar days_in_month = calendar.monthrange(ano_k, mes_k)[1] data = { "company_id": int(company_id), "period": period, "habitaciones_fisicas": hab_fis, "days_in_month": days_in_month, "rooms_total": kv("rooms_total"), "rooms_occupied": kv("rooms_occupied"), "occupancy_pct": kv("occupancy_pct"), "guests_total": kv("guests_total"), "rrhh_empleados": kv("rrhh_empleados_activos"), "rrhh_costo": kv("rrhh_costo_total"), "rrhh_ausentismo":kv("rrhh_ausentismo_mes"), "rrhh_rotacion": kv("rrhh_tasa_rotacion"), } body = json.dumps(data, default=str, ensure_ascii=False).encode("utf-8") self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self._send_cors() self.end_headers() self.wfile.write(body) except Exception as e: self.send_response(500) self.send_header("Content-Type", "application/json") self._send_cors() self.end_headers() self.wfile.write(json.dumps(self._error_payload(e)).encode()) finally: if not _conn_returned: DB_POOL.putconn(conn) def log_message(self, fmt, *args): a0 = str(args[0]) if args else "" a1 = str(args[1]) if len(args) > 1 else "" if "/proxy/" in a0 or a1.startswith("4") or fmt.startswith("code "): super().log_message(fmt, *args) class ThreadingHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer): """Servidor HTTP con soporte multi-hilo para atender peticiones en paralelo.""" daemon_threads = True allow_reuse_address = True if __name__ == "__main__": require_secrets("NOMINA_API_KEY", "PYG_API_KEY") os.chdir(DIR) with ThreadingHTTPServer(("", PORT), ProxyHandler) as httpd: print(f"✅ Servidor corriendo en http://localhost:{PORT}") print(f" Dashboard activos: /dashboard-empleados.html") print(f" Dashboard salarios: /dashboard-salarios.html") print(f" Proxy API: /proxy/rrhh/... → {API_BASE}/rrhh/...") print(f" Proxy históricos: /proxy/historico/salarios") print(f" Ctrl+C para detener.\n") try: httpd.serve_forever() except KeyboardInterrupt: print("\n🛑 Servidor detenido.")