#!/usr/bin/env python3
"""
cierre_mes.py — Cierre mensual automático · Dashboard Talento Humano · Faranda Hotels
═══════════════════════════════════════════════════════════════════════════════════════
Captura el snapshot del mes ANTERIOR desde la API de nómina e IR y lo guarda en
PostgreSQL para que el dashboard lo lea desde ahí en todas las consultas futuras.

Programado con LaunchAgent (macOS) para correr:
  · Día  1 de cada mes  a las 5:00 AM  → primer cierre (datos iniciales)
  · Día  5 de cada mes  a las 5:00 AM  → cierre definitivo (nómina liquidada)

Si falla: escribe log, reintenta 1 hora después y termina.

Uso manual:
  python3 cierre_mes.py                   → cierra el mes anterior automáticamente
  python3 cierre_mes.py --test            → muestra qué mes cerraría (sin escribir)
  python3 cierre_mes.py --empleados-activos-detalle
  python3 cierre_mes.py --solo-empleados-si-vacio
"""

import json
import os
import sys
import time
import logging
import calendar
import urllib.request
import urllib.error
import psycopg2
import psycopg2.extras
from datetime import date, datetime
from concurrent.futures import ThreadPoolExecutor, as_completed
from app_config import API_BASE, NOMINA_API_KEY, db_config, require_secrets, ssl_context

# ══ CONFIGURACIÓN ════════════════════════════════════════════════════════════

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
LOGS_DIR = os.path.join(BASE_DIR, "logs", "cierres")
os.makedirs(LOGS_DIR, exist_ok=True)

# API externa / PostgreSQL
API_KEY = NOMINA_API_KEY
DB_CONFIG = db_config()

# Parámetros de ejecución
MAX_WORKERS   = 8    # peticiones paralelas a la API nómina
RETRY_DELAY_S = 3600  # 1 hora entre reintentos

# Hoteles excluidos (sin NIT / no aplica)
EXCLUIDOS = {'ELMA', 'FNS'}

# SSL: verifica certificados por defecto; APP_VERIFY_SSL=false solo para dev controlado.
SSL_CTX = ssl_context()

# ══ LOGGING ══════════════════════════════════════════════════════════════════

def setup_logger(ano: int, mes: int) -> logging.Logger:
    """Crea logger que escribe en consola Y en archivo con fecha en el nombre."""
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    log_file  = os.path.join(LOGS_DIR, f"cierre_{ano}_{mes:02d}_{timestamp}.log")

    logger = logging.getLogger(f"cierre_{ano}_{mes}")
    logger.setLevel(logging.DEBUG)
    logger.handlers.clear()

    fmt = logging.Formatter("%(asctime)s  %(levelname)-8s  %(message)s",
                             datefmt="%Y-%m-%d %H:%M:%S")

    # Consola
    ch = logging.StreamHandler(sys.stdout)
    ch.setFormatter(fmt)
    logger.addHandler(ch)

    # Archivo
    fh = logging.FileHandler(log_file, encoding="utf-8")
    fh.setFormatter(fmt)
    logger.addHandler(fh)

    logger.info(f"Log iniciado → {log_file}")
    return logger


# ══ HELPERS ══════════════════════════════════════════════════════════════════

def get_prev_month() -> tuple[int, int]:
    """Devuelve (ano, mes) del mes anterior al día de hoy."""
    hoy = date.today()
    if hoy.month == 1:
        return hoy.year - 1, 12
    return hoy.year, hoy.month - 1


def parse_date(val) -> date | None:
    """Parsea una fecha que puede venir como '2026-04-20' o '2026/06/03' o None."""
    if not val:
        return None
    try:
        return date.fromisoformat(str(val).replace("/", "-"))
    except Exception:
        return None


def api_post(path: str, body: dict, timeout: int = 30):
    """Hace POST a la API externa y devuelve el JSON parseado."""
    data = json.dumps(body).encode()
    req  = urllib.request.Request(
        API_BASE + path, data=data,
        headers={"X-API-Key": API_KEY, "Content-Type": "application/json"},
        method="POST"
    )
    with urllib.request.urlopen(req, timeout=timeout, context=SSL_CTX) as resp:
        return json.loads(resp.read())


def db_connect():
    """Abre conexión PostgreSQL."""
    conn = psycopg2.connect(**DB_CONFIG)
    conn.autocommit = False
    return conn


# ══ VERIFICACIÓN DE IDEMPOTENCIA ═════════════════════════════════════════════

def ya_cerrado_empleados(conn, ano: int, mes: int) -> int:
    """Devuelve cuántos registros existen en historico_empleados para ese mes."""
    with conn.cursor() as cur:
        cur.execute(
            "SELECT COUNT(*) FROM historico_empleados WHERE fecha = %s",
            (date(ano, mes, 1),)
        )
        return cur.fetchone()[0]


def ya_cerrado_ir(conn, ano: int, mes: int) -> int:
    """Devuelve cuántos registros existen en historico_ir para ese mes."""
    periodo = f"{ano}-{mes:02d}"
    with conn.cursor() as cur:
        cur.execute("SELECT COUNT(*) FROM historico_ir WHERE periodo = %s", (periodo,))
        return cur.fetchone()[0]


# ══ MAPAS DE REFERENCIA ══════════════════════════════════════════════════════

def cargar_nit_razon_map(conn) -> dict:
    """
    Devuelve tres mapas:
      hotel_bu_map : (hotel_name_api, bu_codigo_lower) → razon_social
      nit_bu_map   : (nit, bu_codigo_lower)            → razon_social  ← clave para NITs compartidos
      nit_razon    : nit                               → razon_social  ← fallback sin bu
    """
    with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
        cur.execute("""
            SELECT e.codigo    AS razon_social,
                   e.nit,
                   u.hotel_name_api,
                   LOWER(COALESCE(u.bu_codigo, 'local')) AS bu_codigo
            FROM empresas e
            LEFT JOIN hotel_unificacion u ON u.razon_social = e.codigo
            WHERE e.nit IS NOT NULL
              AND e.codigo NOT IN %s
        """, (tuple(EXCLUIDOS),))
        rows = cur.fetchall()

    # NITs donde todos los BUs deben quedar bajo la misma razon_social principal
    # (consistencia con datos históricos cargados del Excel)
    NIT_UNICO = {
        '900657316': 'GHS',    # GHS + The Club → todo bajo GHS
        '901418135': 'SERV',   # Servihoteles + submarcas → todo bajo SERV
        '901936953': 'CCAR',   # FC Cartagena + Spiaggia → todo bajo CCAR
    }

    hotel_bu_map = {}   # (hotel_name_api, bu_lower) → razon_social
    nit_bu_map   = {}   # (nit, bu_lower)            → razon_social  (NITs compartidos)
    nit_razon    = {}   # nit                        → razon_social  (fallback)

    for r in rows:
        # Si el NIT tiene una razon_social única forzada, usarla para todos sus BUs
        razon = NIT_UNICO.get(r["nit"], r["razon_social"])

        if r["hotel_name_api"]:
            hotel_bu_map[(r["hotel_name_api"], r["bu_codigo"])] = razon
        if r["nit"]:
            nit_bu_map[(r["nit"], r["bu_codigo"])] = razon
            if r["nit"] not in nit_razon:
                nit_razon[r["nit"]] = razon

    return hotel_bu_map, nit_bu_map, nit_razon


def cargar_nits_activos(conn) -> list[str]:
    """Devuelve lista de NITs únicos activos (sin excluidos, sin nit nulo)."""
    with conn.cursor() as cur:
        cur.execute("""
            SELECT DISTINCT nit
            FROM empresas
            WHERE nit IS NOT NULL
              AND codigo NOT IN %s
            ORDER BY nit
        """, (tuple(EXCLUIDOS),))
        return [r[0] for r in cur.fetchall()]


def cargar_socios_set(conn) -> set:
    """Devuelve set de codigo_empleado marcados como SOCIO activo."""
    with conn.cursor() as cur:
        cur.execute("SELECT codigo_empleado FROM socios WHERE activo = TRUE")
        return {r[0] for r in cur.fetchall()}


def fetch_vence_contrato_map() -> dict:
    """
    Construye mapa { identificacion → fecha_vence_contrato } combinando:
      1. empleados-activos-detalle  (business date — cubre ~95% activos)
      2. IR ingresos últimos 24 meses (campo contrato=identificacion, vencimiento)
    Devuelve el mapa unificado.
    """
    vence_map = {}

    # Fuente 2 primero (menor prioridad) — IR ingresos rango amplio
    # Dos llamadas: tipo_nomina_id=4 (Quincenal) y 5 (Mensual, Aloft/Spiaggia)
    try:
        hoy = date.today()
        fi  = date(hoy.year - 2, hoy.month, 1).isoformat()
        ff  = hoy.isoformat()
        _ir_base = {
            "fecha_inicial": fi, "fecha_final": ff,
            "estado": "T",
            "agrupar":  {"cco": False, "bu": True,  "cla": False, "causa": False, "cargo": False},
            "saltos":   {"cco": False, "bu": False, "cla": False, "causa": False, "cargo": False},
            "filtros":  {"contrato": ["","zzz"], "empleado": ["","zzz"],
                         "centro_costo": ["","zzz"], "bu": ["","zzz"],
                         "clasificacion": ["","zzz"], "cargo": ["","zzz"],
                         "nivel": ["","zzz"], "causa_retiro": ["","zzz"]},
            "orden": ["C","N","I"], "forma_pago": 999,
        }
        for tipo_id in (4, 5):
            data = api_post("/rrhh/ingresos-retiros", {**_ir_base, "tipo_nomina_id": tipo_id}, timeout=60)
            for hotel in data.get("hoteles", []):
                for bu in hotel.get("unidades_negocio", []):
                    for row in bu.get("rows", []):
                        if row.get("tipo") == "I" and row.get("contrato") and row.get("vencimiento"):
                            idf = str(row["contrato"])
                            vence_map[idf] = parse_date(row["vencimiento"])
    except Exception:
        pass  # fallback a fuente 1

    # Fuente 1 — empleados-activos-detalle (mayor prioridad, sobreescribe)
    try:
        req = urllib.request.Request(
            API_BASE + "/rrhh/empleados-activos-detalle",
            headers={"X-API-Key": API_KEY},
            method="GET"
        )
        with urllib.request.urlopen(req, timeout=30, context=SSL_CTX) as resp:
            data = json.loads(resp.read())
        for hotel in data.get("hoteles", []):
            for emp in hotel.get("empleados", []):
                idf = str(emp.get("identificacion") or "")
                if idf:
                    vence_map[idf] = parse_date(emp.get("fecha_vence_contrato"))
    except Exception:
        pass

    return vence_map


# ══ EMPLEADOS ════════════════════════════════════════════════════════════════

class NominaNoDisponible(Exception):
    """La nómina de ese NIT/mes aún no ha sido cerrada en el sistema."""
    pass


def fetch_empleados_nit(nit: str, ano: int, mes: int) -> tuple[str, list]:
    """
    Llama la API de nómina para un NIT y devuelve (nit, [empleados]).
    Lanza NominaNoDisponible si el período aún no está cerrado (HTTP 400).
    """
    try:
        data = api_post("/nomina/reporte", {
            "hotel_nit":          nit,
            "ano":                ano,
            "mes":                mes,
            "incluir_empleados":  True,
            "ss_vista":           "ambos",
        })
        return nit, data.get("empleados", [])
    except urllib.error.HTTPError as e:
        # Leer el cuerpo del error para distinguir "sin periodos" de error real
        try:
            body = json.loads(e.read().decode())
            detalle = body.get("detail", "").lower()
        except Exception:
            detalle = ""
        if e.code == 400 and "periodos cerrados" in detalle:
            raise NominaNoDisponible(f"nómina no cerrada para {ano}-{mes:02d}")
        raise  # otro error HTTP → propagar como error real


def fetch_empleados_activos_detalle() -> list[dict]:
    """
    Lee la foto viva de empleados activos.
    Se usa en el cierre del día 1: la foto tomada el día 1 del mes siguiente
    se guarda como cierre del mes anterior.
    """
    req = urllib.request.Request(
        API_BASE + "/rrhh/empleados-activos-detalle",
        headers={"X-API-Key": API_KEY},
        method="GET"
    )
    with urllib.request.urlopen(req, timeout=60, context=SSL_CTX) as resp:
        data = json.loads(resp.read())
    return data.get("hoteles", [])


def validar_hoteles_recibidos(hoteles: list[dict], nits_esperados: list[str],
                              nit_razon: dict, logger) -> None:
    """
    Compara los NITs devueltos por empleados-activos-detalle contra los NITs
    activos en la tabla `empresas`. Si la API omite un hotel completo (como
    pasó con Belvedere/Faranda_Bogota en jul-2026, sin que la API devolviera
    error alguno), lo deja como WARNING explícito en vez de fallar en
    silencio con ese hotel en cero.
    """
    nits_recibidos = {str(h.get("hotel_nit") or "").strip() for h in hoteles}
    faltantes = [nit for nit in nits_esperados if nit and nit not in nits_recibidos]
    if not faltantes:
        return
    logger.warning(f"  ⚠️  La API empleados-activos-detalle NO devolvió {len(faltantes)} NIT(s) esperado(s):")
    for nit in faltantes:
        razon = nit_razon.get(nit, "?")
        logger.warning(f"      - NIT {nit} (razon_social≈{razon}) — quedará SIN empleados en este cierre")


def insertar_empleados_activos_detalle(conn, ano: int, mes: int,
                                       hotel_bu_map: dict, nit_bu_map: dict,
                                       nit_razon: dict, socios_set: set,
                                       logger, nits_esperados: list[str] | None = None) -> int:
    """Inserta empleados activos desde GET /rrhh/empleados-activos-detalle."""
    fecha = date(ano, mes, 1)
    hoteles = fetch_empleados_activos_detalle()
    if nits_esperados:
        validar_hoteles_recibidos(hoteles, nits_esperados, nit_razon, logger)
    rows = []
    sin_razon = 0

    for hotel in hoteles:
        hotel_api = hotel.get("hotel_name") or ""
        nit = str(hotel.get("hotel_nit") or "").strip()
        for emp in hotel.get("empleados", []):
            bu = (emp.get("bu_codigo") or "local").lower()
            razon = (
                hotel_bu_map.get((hotel_api, bu))
                or hotel_bu_map.get((hotel_api, "local"))
                or nit_bu_map.get((nit, bu))
                or nit_razon.get(nit)
            )
            if not razon:
                sin_razon += 1
                razon = nit or None

            idf = str(emp.get("identificacion") or "")
            operacion = "SOCIO" if idf in socios_set else "EN OPERACION"

            sueldo = emp.get("sueldo_base")
            if sueldo is None:
                sueldo = emp.get("s_basico")

            rows.append((
                fecha,
                razon,
                idf or None,
                emp.get("nombre") or "",
                parse_date(emp.get("fecha_nacimiento")),
                (emp.get("sexo") or "").capitalize(),
                parse_date(emp.get("fecha_ingreso")),
                parse_date(emp.get("fecha_vence_contrato")),
                emp.get("centro_costo"),
                emp.get("cargo"),
                int(sueldo or 0),
                operacion,
                # `contrato` del endpoint es un FOLIO numérico, NO el tipo de contrato: usarlo de
                # respaldo dejó el documento/folio en esta columna (99,9 % de agosto 2026). Sin
                # `tipo_contrato` real se guarda NULL; el dashboard lo deriva con `tipoContrato()`.
                emp.get("tipo_contrato"),
            ))

    if sin_razon:
        logger.warning(f"  Empleados activos sin razón social mapeada: {sin_razon}")
    if not rows:
        logger.warning("  No se encontraron empleados activos para insertar")
        return 0

    with conn.cursor() as cur:
        psycopg2.extras.execute_values(cur, """
            INSERT INTO historico_empleados
                (fecha, razon_social, codigo_empleado, nombre_completo,
                 fecha_nacimiento, sexo, fecha_ingreso, fecha_vencimiento_contrato,
                 centro_costo, cargo, sueldo_base, operacion, tipo_contrato)
            VALUES %s
        """, rows, page_size=500)

    conn.commit()
    logger.info(f"  ✅ Empleados activos insertados desde detalle vivo: {len(rows)}")
    return len(rows)


def insertar_empleados(conn, ano: int, mes: int,
                       hotel_bu_map: dict, nit_bu_map: dict, nit_razon: dict,
                       nits_emps: list[tuple], socios_set: set,
                       vence_map: dict, logger) -> int:
    """
    Inserta empleados ACTIVOS del mes en historico_empleados.

    Reglas:
    - Solo empleados con fecha_retiro IS NULL o fecha_retiro > último día del mes
    - operacion: 'SOCIO' si identificacion in socios_set, 'EN OPERACION' si no
    - fecha_vencimiento_contrato:
        1. vence_map[identificacion] (empleados-activos-detalle / IR)
        2. fecha_retiro de nómina si tipo_contrato='fijo' y fecha futura
        3. NULL para contratos indefinidos
    """
    _, ultimo_dia = calendar.monthrange(ano, mes)
    ultimo        = date(ano, mes, ultimo_dia)
    fecha         = date(ano, mes, 1)
    rows          = []
    excluidos     = 0

    for nit, empleados in nits_emps:
        for emp in empleados:
            # ── Filtro: excluir retirados durante el mes ──────────────────
            fecha_retiro = parse_date(emp.get("fecha_retiro"))
            if fecha_retiro is not None and fecha_retiro <= ultimo:
                excluidos += 1
                continue

            bu    = (emp.get("bu_codigo") or "local").lower()
            razon = nit_bu_map.get((nit, bu)) or nit_razon.get(nit, nit)
            idf   = str(emp.get("identificacion") or "")

            # ── Operación ─────────────────────────────────────────────────
            operacion = "SOCIO" if idf in socios_set else "EN OPERACION"

            # ── Fecha vencimiento contrato ─────────────────────────────────
            tipo_contrato = (emp.get("tipo_contrato") or "").lower()
            if tipo_contrato == "indefinido":
                vence = None
            else:
                # Fuente 1+2: mapa de APIs
                vence = vence_map.get(idf)
                # Fuente 3: fecha_retiro de nómina si es futura (contrato fijo aún vigente)
                if vence is None and fecha_retiro is not None and fecha_retiro > ultimo:
                    vence = fecha_retiro

            rows.append((
                fecha,
                razon,
                idf or None,
                emp.get("nombre") or "",
                parse_date(emp.get("fecha_nacimiento")),
                (emp.get("sexo") or "").capitalize(),
                parse_date(emp.get("fecha_ingreso")),
                vence,
                emp.get("centro_costo"),
                emp.get("cargo"),
                int(emp.get("s_basico") or 0),
                operacion,
                emp.get("tipo_contrato"),
            ))

    logger.info(f"  Empleados excluidos (retirados en el mes): {excluidos}")
    if not rows:
        logger.warning("  No se encontraron empleados para insertar")
        return 0

    with conn.cursor() as cur:
        psycopg2.extras.execute_values(cur, """
            INSERT INTO historico_empleados
                (fecha, razon_social, codigo_empleado, nombre_completo,
                 fecha_nacimiento, sexo, fecha_ingreso, fecha_vencimiento_contrato,
                 centro_costo, cargo, sueldo_base, operacion, tipo_contrato)
            VALUES %s
        """, rows, page_size=500)

    conn.commit()
    logger.info(f"  ✅ Empleados insertados: {len(rows)}")
    return len(rows)


def cierre_empleados(conn, ano: int, mes: int,
                     hotel_bu_map: dict, nit_bu_map: dict, nit_razon: dict,
                     nits: list[str], logger) -> int:
    """Orquesta la captura paralela de empleados por NIT e inserción en PG."""

    # ── Cargar datos auxiliares en paralelo ───────────────────────────────
    logger.info("  Cargando socios y mapa de vencimientos de contrato…")
    with ThreadPoolExecutor(max_workers=2) as pool:
        fut_socios = pool.submit(cargar_socios_set, conn)
        fut_vence  = pool.submit(fetch_vence_contrato_map)
        socios_set = fut_socios.result()
        vence_map  = fut_vence.result()
    logger.info(f"  Socios activos cargados: {len(socios_set)} | "
                f"Mapa vencimientos: {len(vence_map)} registros")

    # ── Fetch paralelo de nóminas ─────────────────────────────────────────
    logger.info(f"  Llamando API nómina para {len(nits)} NITs en paralelo…")
    nits_emps       = []
    sin_cerrar      = []
    errores_reales  = []

    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
        futuros = {pool.submit(fetch_empleados_nit, nit, ano, mes): nit
                   for nit in nits}
        for fut in as_completed(futuros):
            nit = futuros[fut]
            try:
                nit_res, emps = fut.result()
                nits_emps.append((nit_res, emps))
                logger.debug(f"    NIT {nit}: {len(emps)} empleados")
            except NominaNoDisponible:
                sin_cerrar.append(nit)
                logger.warning(f"    NIT {nit}: nómina aún no cerrada — se omite")
            except Exception as e:
                errores_reales.append(nit)
                logger.error(f"    NIT {nit}: ERROR inesperado — {e}")

    if sin_cerrar:
        logger.warning(f"  ⚠️  Sin nómina cerrada ({len(sin_cerrar)} NITs): {sin_cerrar}")
        logger.warning(f"     Se incluirán en el cierre del día 5 cuando estén disponibles")
    if errores_reales:
        logger.error(f"  ❌ Errores reales ({len(errores_reales)} NITs): {errores_reales}")

    return insertar_empleados(conn, ano, mes, hotel_bu_map, nit_bu_map,
                               nit_razon, nits_emps, socios_set, vence_map, logger)


# ══ INGRESOS Y RETIROS ═══════════════════════════════════════════════════════

def fetch_ir(ano: int, mes: int) -> list[dict]:
    """
    Llama la API de IR para el mes completo y aplana la respuesta.
    Devuelve lista de rows listos para insertar.
    """
    _, ultimo_dia = calendar.monthrange(ano, mes)
    fi  = date(ano, mes, 1).isoformat()
    ff  = date(ano, mes, ultimo_dia).isoformat()

    _ir_base = {
        "fecha_inicial":  fi,
        "fecha_final":    ff,
        "estado":         "T",
        "agrupar":  {"cco": False, "bu": True,  "cla": False,
                     "causa": False, "cargo": False},
        "saltos":   {"cco": False, "bu": False, "cla": False,
                     "causa": False, "cargo": False},
        "filtros": {
            "contrato":      ["", "zzz"],
            "empleado":      ["", "zzz"],
            "centro_costo":  ["", "zzz"],
            "bu":            ["", "zzz"],
            "clasificacion": ["", "zzz"],
            "cargo":         ["", "zzz"],
            "nivel":         ["", "zzz"],
            "causa_retiro":  ["", "zzz"],
        },
        "orden":      ["C", "N", "I"],
        "forma_pago": 999,
    }
    # Dos llamadas: tipo_nomina_id=4 (Quincenal) y 5 (Mensual, Aloft/Spiaggia)
    hoteles_combinados: list[dict] = []
    for tipo_id in (4, 5):
        data = api_post("/rrhh/ingresos-retiros", {**_ir_base, "tipo_nomina_id": tipo_id})
        hoteles_combinados.extend(data.get("hoteles", []))

    rows = []
    for hotel in hoteles_combinados:
        hotel_api = hotel.get("hotel_name", "")
        for bu in hotel.get("unidades_negocio", []):
            bu_codigo = (bu.get("bu_codigo") or bu.get("bu", "") or "local").upper()
            for row in bu.get("rows", []):
                # "contrato" trae la cédula del empleado, a veces con sufijo de
                # renovación de contrato (ej. "1137222929-1") — se limpia para
                # que coincida con codigo_empleado en historico_empleados.
                contrato_raw = (row.get("contrato") or "").strip()
                codigo_empleado = contrato_raw.split("-")[0] if contrato_raw else None
                rows.append({
                    "hotel_api":      hotel_api,
                    "bu_codigo":      bu_codigo,
                    "codigo_empleado": codigo_empleado,
                    "nombre_completo": row.get("nombres") or row.get("nombre_completo", ""),
                    "cargo":          row.get("cargo", ""),
                    "tipo_contrato":  row.get("tipo_contrato", ""),
                    "sueldo_base":    row.get("sueldo") or row.get("s_basico") or 0,
                    "ingreso":        parse_date(row.get("ingreso")),
                    "termino":        parse_date(row.get("vencimiento") or row.get("termino")),
                    "causa_retiro":   row.get("causa_retiro"),
                    "operacion":      row.get("operacion", "En Operacion"),
                    "tipo_movimiento": row.get("tipo", "I"),
                })
    return rows


def insertar_ir(conn, rows: list[dict], hotel_bu_map: dict,
                nit_razon: dict, ano: int, mes: int, logger) -> int:
    """Inserta los movimientos IR en historico_ir."""
    if not rows:
        logger.warning("  No hay movimientos IR para insertar")
        return 0

    periodo   = f"{ano}-{mes:02d}"
    insert_rows = []
    for r in rows:
        bu_low = r["bu_codigo"].lower()
        razon  = hotel_bu_map.get((r["hotel_api"], bu_low))

        insert_rows.append((
            razon,
            r["nombre_completo"],
            r["ingreso"],
            r["cargo"],
            r["tipo_movimiento"],
            float(r["sueldo_base"]) if r["sueldo_base"] else None,
            r["tipo_contrato"],
            r["termino"],
            r["operacion"],
            r["causa_retiro"],
            periodo,
            r.get("codigo_empleado"),
        ))

    with conn.cursor() as cur:
        psycopg2.extras.execute_values(cur, """
            INSERT INTO historico_ir
                (razon_social, nombre_completo, ingreso, cargo,
                 tipo_movimiento, sueldo_base, tipo_contrato,
                 termino, operacion, causa_retiro, periodo, codigo_empleado)
            VALUES %s
        """, insert_rows, page_size=500)

    conn.commit()
    logger.info(f"  ✅ Movimientos IR insertados: {len(insert_rows)}")
    return len(insert_rows)


# Razones sociales cuya IR la API /rrhh/ingresos-retiros NO devuelve.
# CLUB (FC Bogota) + FEB (Belvedere) comparten el NIT 900345450 / bucket
# "Faranda_Bogota", que dejo de aparecer en la respuesta del endpoint a partir
# de jun-2026 (ene-may si venia). Los datos SI existen en Zeus y en
# empleados-activos-detalle, asi que se reconstruyen desde el snapshot de
# historico_empleados. Si NominaApp corrige su API, los datos reales mandan
# (ver guardia en completar_ir_desde_snapshot).
IR_FALLBACK_SNAPSHOT_RAZONES = {"CLUB", "FEB"}


def _periodo_anterior(ano: int, mes: int) -> tuple[int, int]:
    return (ano - 1, 12) if mes == 1 else (ano, mes - 1)


def completar_ir_desde_snapshot(conn, ano: int, mes: int, razones, logger) -> int:
    """
    Sintetiza movimientos IR para las `razones` cuya API no entrega datos,
    comparando el snapshot de historico_empleados del mes vs. el del mes anterior:
      - cedula en el mes y no en el anterior  -> ingreso ('I')
      - cedula en el anterior y no en el mes   -> retiro  ('R')
    Solo actua si esa razon no tiene NINGUNA fila IR del periodo (si la API
    empieza a devolverla, esos datos reales mandan). Filas marcadas origen='snapshot'.
    Limitacion: una transferencia entre dos razones del mismo grupo se ve como
    retiro en una + ingreso en la otra (no como traslado).
    """
    periodo = f"{ano}-{mes:02d}"
    pa, pm = _periodo_anterior(ano, mes)
    _, ultimo_dia = calendar.monthrange(ano, mes)
    ini_mes = date(ano, mes, 1)
    fin_mes = date(ano, mes, ultimo_dia)
    total = 0

    with conn.cursor() as cur:
        cur.execute(
            "ALTER TABLE historico_ir ADD COLUMN IF NOT EXISTS origen VARCHAR(16) DEFAULT 'api'"
        )
        for rs in sorted(razones):
            cur.execute(
                "SELECT COUNT(*) FROM historico_ir WHERE periodo=%s AND razon_social=%s",
                (periodo, rs),
            )
            if cur.fetchone()[0] > 0:
                continue  # la API si trajo datos de esta razon: no tocar

            cur.execute("""
                SELECT codigo_empleado, nombre_completo, cargo, tipo_contrato,
                       sueldo_base, operacion, fecha_ingreso, fecha_vencimiento_contrato
                FROM historico_empleados WHERE fecha=%s AND razon_social=%s
            """, (ini_mes, rs))
            cur_snap = {r[0]: r for r in cur.fetchall()}

            cur.execute("""
                SELECT codigo_empleado, nombre_completo, cargo, tipo_contrato,
                       sueldo_base, operacion, fecha_ingreso, fecha_vencimiento_contrato
                FROM historico_empleados WHERE fecha=%s AND razon_social=%s
            """, (date(pa, pm, 1), rs))
            prev_snap = {r[0]: r for r in cur.fetchall()}

            if not prev_snap:
                if cur_snap:
                    logger.warning(
                        f"  IR fallback {rs} {periodo}: sin snapshot de {pa}-{pm:02d}, "
                        f"no se puede derivar"
                    )
                continue

            filas = []
            for cod, r in cur_snap.items():
                if cod in prev_snap:
                    continue
                ing = r[6] if (r[6] and ini_mes <= r[6] <= fin_mes) else ini_mes
                filas.append((rs, r[1], ing, r[2], "I",
                              float(r[4]) if r[4] else None, r[3], r[7],
                              r[5], None, periodo, cod, "snapshot"))
            for cod, r in prev_snap.items():
                if cod in cur_snap:
                    continue
                filas.append((rs, r[1], r[6], r[2], "R",
                              float(r[4]) if r[4] else None, r[3], fin_mes,
                              r[5], None, periodo, cod, "snapshot"))

            if not filas:
                continue

            psycopg2.extras.execute_values(cur, """
                INSERT INTO historico_ir
                  (razon_social, nombre_completo, ingreso, cargo, tipo_movimiento,
                   sueldo_base, tipo_contrato, termino, operacion, causa_retiro,
                   periodo, codigo_empleado, origen)
                VALUES %s
            """, filas, page_size=500)
            ing_n = sum(1 for f in filas if f[4] == "I")
            ret_n = sum(1 for f in filas if f[4] == "R")
            total += len(filas)
            logger.warning(
                f"  ⚠️  IR de {rs} {periodo}: la API no la entrego — sintetizados "
                f"{ing_n} ingresos + {ret_n} retiros desde el snapshot (origen='snapshot')"
            )

    conn.commit()
    return total


# ══ LÓGICA PRINCIPAL ═════════════════════════════════════════════════════════

def ejecutar_cierre(ano: int, mes: int, modo_test: bool = False,
                    empleados_activos_detalle: bool = False,
                    solo_empleados_si_vacio: bool = False) -> bool:
    """
    Ejecuta el cierre para (ano, mes).
    Devuelve True si todo fue bien, False si hubo error.
    """
    meses_es = ["Enero","Febrero","Marzo","Abril","Mayo","Junio",
                "Julio","Agosto","Septiembre","Octubre","Noviembre","Diciembre"]
    logger = setup_logger(ano, mes)

    logger.info("=" * 70)
    logger.info(f"  CIERRE DE MES — {meses_es[mes-1]} {ano}")
    logger.info(f"  Ejecutado: {datetime.now().strftime('%d/%m/%Y %H:%M:%S')}")
    if modo_test:
        logger.info("  MODO TEST — no se escribe nada en la base de datos")
    if empleados_activos_detalle:
        logger.info("  EMPLEADOS — usando GET /rrhh/empleados-activos-detalle")
    if solo_empleados_si_vacio:
        logger.info("  EMPLEADOS — solo se insertan si el periodo está vacío")
    logger.info("=" * 70)

    try:
        conn = db_connect()

        # ── Mapas de referencia ────────────────────────────────────────────
        hotel_bu_map, nit_bu_map, nit_razon = cargar_nit_razon_map(conn)
        nits = cargar_nits_activos(conn)
        logger.info(f"  NITs activos encontrados: {len(nits)}")

        # ── EMPLEADOS ─────────────────────────────────────────────────────
        logger.info("")
        logger.info("[ 1/2 ] EMPLEADOS")
        existe_emps = ya_cerrado_empleados(conn, ano, mes)
        if existe_emps > 0:
            logger.info(f"  ⚠️  Ya existen {existe_emps} registros para {meses_es[mes-1]} {ano}")
            if solo_empleados_si_vacio:
                logger.info("  Se conserva el cierre existente; no se recargan empleados")
            else:
                logger.info("  Se sobreescribirán (DELETE + INSERT para mantener datos frescos)")
            if not modo_test and not solo_empleados_si_vacio:
                with conn.cursor() as cur:
                    cur.execute("DELETE FROM historico_empleados WHERE fecha = %s",
                                (date(ano, mes, 1),))
                conn.commit()
                logger.info(f"  Eliminados {existe_emps} registros anteriores")
        else:
            logger.info(f"  Sin datos previos — inserción fresca")

        if solo_empleados_si_vacio and existe_emps > 0:
            total_emps = existe_emps
        elif not modo_test:
            if empleados_activos_detalle:
                socios_set = cargar_socios_set(conn)
                logger.info(f"  Socios activos cargados: {len(socios_set)}")
                total_emps = insertar_empleados_activos_detalle(
                    conn, ano, mes, hotel_bu_map, nit_bu_map, nit_razon,
                    socios_set, logger, nits_esperados=nits
                )
            else:
                total_emps = cierre_empleados(conn, ano, mes, hotel_bu_map,
                                               nit_bu_map, nit_razon, nits, logger)
        else:
            if empleados_activos_detalle:
                logger.info("  [TEST] Se llamaría empleados-activos-detalle y se insertarían empleados")
            elif solo_empleados_si_vacio and existe_emps > 0:
                logger.info("  [TEST] Se conservarían los empleados existentes")
            else:
                logger.info("  [TEST] Se llamaría la API nómina y se insertarían empleados")
            total_emps = 0

        # ── INGRESOS Y RETIROS ─────────────────────────────────────────────
        logger.info("")
        logger.info("[ 2/2 ] INGRESOS Y RETIROS")
        existe_ir = ya_cerrado_ir(conn, ano, mes)
        if existe_ir > 0:
            logger.info(f"  ⚠️  Ya existen {existe_ir} movimientos IR para {meses_es[mes-1]} {ano}")
            logger.info("  Se sobreescribirán (DELETE + INSERT para mantener datos frescos)")
            if not modo_test:
                _, ultimo_dia = calendar.monthrange(ano, mes)
                with conn.cursor() as cur:
                    cur.execute(
                        "DELETE FROM historico_ir WHERE periodo = %s",
                        (f"{ano}-{mes:02d}",)
                    )
                conn.commit()
                logger.info(f"  Eliminados {existe_ir} movimientos anteriores")
        else:
            logger.info(f"  Sin datos previos — inserción fresca")

        if not modo_test:
            logger.info("  Llamando API de IR…")
            ir_rows   = fetch_ir(ano, mes)
            logger.info(f"  Movimientos recibidos de la API: {len(ir_rows)}")
            total_ir  = insertar_ir(conn, ir_rows, hotel_bu_map, nit_razon, ano, mes, logger)
            # Fallback para razones que la API /rrhh/ingresos-retiros no entrega
            # (CLUB/FEB = NIT 900345450): reconstruye IR desde el snapshot.
            total_ir += completar_ir_desde_snapshot(
                conn, ano, mes, IR_FALLBACK_SNAPSHOT_RAZONES, logger
            )
        else:
            logger.info("  [TEST] Se llamaría la API IR y se insertarían movimientos")
            total_ir = 0

        # ── Resumen ────────────────────────────────────────────────────────
        logger.info("")
        logger.info("=" * 70)
        if modo_test:
            logger.info(f"  ✅ TEST completado — {meses_es[mes-1]} {ano} listo para cerrar")
        else:
            logger.info(f"  ✅ CIERRE COMPLETADO — {meses_es[mes-1]} {ano}")
            logger.info(f"     Empleados:  {total_emps}")
            logger.info(f"     IR:         {total_ir}")
        logger.info("=" * 70)

        conn.close()
        return True

    except Exception as e:
        import traceback
        logger.error("")
        logger.error("=" * 70)
        logger.error(f"  ❌ ERROR EN CIERRE — {e}")
        logger.error(traceback.format_exc())
        logger.error("=" * 70)
        try:
            conn.rollback()
            conn.close()
        except Exception:
            pass
        return False


# ══ MAIN CON REINTENTOS ═══════════════════════════════════════════════════════

def main():
    require_secrets("NOMINA_API_KEY")
    modo_test = "--test" in sys.argv
    empleados_activos_detalle = "--empleados-activos-detalle" in sys.argv
    solo_empleados_si_vacio = "--solo-empleados-si-vacio" in sys.argv

    # Soporte para --ano=YYYY --mes=MM (override manual)
    ano, mes = get_prev_month()
    for arg in sys.argv[1:]:
        if arg.startswith("--ano="):
            ano = int(arg.split("=")[1])
        elif arg.startswith("--mes="):
            mes = int(arg.split("=")[1])
    meses_es  = ["Enero","Febrero","Marzo","Abril","Mayo","Junio",
                 "Julio","Agosto","Septiembre","Octubre","Noviembre","Diciembre"]

    print(f"\n🔄 cierre_mes.py — {meses_es[mes-1]} {ano}")
    print(f"   {datetime.now().strftime('%d/%m/%Y %H:%M:%S')}\n")

    # Intento 1
    ok = ejecutar_cierre(ano, mes, modo_test, empleados_activos_detalle, solo_empleados_si_vacio)
    if ok:
        sys.exit(0)

    # Intento 2 — tras 1 hora
    if not modo_test:
        horas = RETRY_DELAY_S // 3600
        print(f"\n⏳ Reintentando en {horas} hora(s)…\n")
        time.sleep(RETRY_DELAY_S)

        ok = ejecutar_cierre(ano, mes, modo_test, empleados_activos_detalle, solo_empleados_si_vacio)
        if ok:
            sys.exit(0)

    # Ambos intentos fallaron → log de error y salida
    logger_alert = logging.getLogger(f"cierre_{ano}_{mes}")
    logger_alert.error("Ambos intentos fallaron — revisar logs en: " + LOGS_DIR)
    logger_alert.error("Ejecutar manualmente: python3 cierre_mes.py")
    sys.exit(1)


if __name__ == "__main__":
    main()