from __future__ import annotations

from pathlib import Path

from typing import Dict, List

import pandas as pd

import math

from openpyxl import load_workbook

#
==========================================
===============

# CONFIGURACIÓN

#
==========================================
===============

RUTA_ARCHIVO = Path(r"C:\Users\dcastilloo\OneDrive - Fonafe\

Actualización PIB 2025 - 2026\2026\CumplimientoHitos\1-Personas\1-
SolicitudPlazosFijos\DesarrolloPy\Programacion PF2026 - [Link]")

CARPETA_SALIDA = RUTA_ARCHIVO.parent

TABLAS_REQUERIDAS = {

"xpu_ProyectoMesTag": [

"CodigoIniciativa",

"Empresa",

"NombreCorto",

"Tipologia",

"Ciclo",

"Comentario",

"Ppto2026",
 "Mes",

"MesIdx",

"Tag",

"FuenteTabla",

"ProyectoMesKey",

"ProyectoMesTagKey",

],

"xpu_RolReglas": [

"Rol",

"TagsVisibles",

"TipologiasPermitidas",

"ExcluyeTipologia",

"Activo",

"OrdenVisual",

"Comentario",

],

"xpu_CapacidadRolEmpresa": [

"Empresa",

"Rol",

"CapacidadParalelo",

"Activo",

"Comentario",

],

"xpu_CatalogoMeses": [

"Mes",

"MesIdx",

"Activo",

"Comentario",

],
 "xpu_ParametrosModelo": [

"Parametro",

"Valor",

"TipoDato",

"Activo",

"Comentario",

],

"xpu_DotacionActual": [

"Empresa",

"Rol",

"DotacionActual",

"Activo",

"Comentario",

],

"xpu_RuteoBalanceRol": [

"Empresa",

"RolBase",

"RolApoyo",

"Tag",

"Tipologia",

"TipoRuteo",

"PrioridadRuteo",

"LimitarBasePorDotacion",

"HeatmapApoyoSinTopeDotacion",

"PermiteDesborde",

"Activo",

"Comentario",

],

}
PRIORIDAD_TIPOLOGIA = {

"Distribución": 1,

"Transmisión": 2,

"Generación": 3,

"Edificaciones": 4,

"TIC": 5,

"Innovación": 6,

#
==========================================
===============

# UTILIDADES

#
==========================================
===============

def limpiar_texto(valor):

if [Link](valor):

return None

return str(valor).strip()

def normalizar_dataframe(df: [Link]) -> [Link]:

df = [Link]()

[Link] = [str(c).strip() for c in [Link]]

for col in [Link]:

if df[col].dtype == "object":

df[col] = df[col].apply(limpiar_texto)
 return df

def partir_lista_pipe(texto: object) -> List[str]:

if [Link](texto) or texto is None:

return []

return [[Link]() for x in str(texto).split("|") if str(x).strip() != ""]

def obtener_tabla_excel(ruta_archivo: Path, nombre_tabla: str) ->

[Link]:

wb = load_workbook(ruta_archivo, data_only=True)

tabla_encontrada = None

ws_tabla = None

for ws in [Link]:

if nombre_tabla in [Link]:

tabla_encontrada = [Link][nombre_tabla]

ws_tabla = ws

break

if tabla_encontrada is None or ws_tabla is None:

raise ValueError(f"No se encontró la tabla '{nombre_tabla}' en el

archivo.")

ref = tabla_encontrada.ref

filas = ws_tabla[ref]

data = [[[Link] for cell in fila] for fila in filas]

if not data or len(data) < 1:

raise ValueError(f"La tabla '{nombre_tabla}' no contiene datos.")

 encabezados = data[0]

registros = data[1:] if len(data) > 1 else []

df = [Link](registros, columns=encabezados)

df = normalizar_dataframe(df)

return df

def exportar_debug(df: [Link], nombre_archivo: str) -> None:

ruta = CARPETA_SALIDA / nombre_archivo

df.to_csv(ruta, index=False, encoding="utf-8-sig")

print(f"Archivo debug generado: {ruta}")

#
==========================================
===============

# VALIDACIONES

#
==========================================
===============

def validar_columnas(nombre: str, df: [Link],

columnas_esperadas: List[str]) -> List[str]:

errores = []

columnas_reales = list([Link])

faltantes = [c for c in columnas_esperadas if c not in

columnas_reales]

if faltantes:

[Link](f"[{nombre}] Faltan columnas: {faltantes}")

return errores
def validar_vacios_criticos(tablas: Dict[str, [Link]]) ->
List[str]:

errores = []

df = tablas["xpu_ProyectoMesTag"]

for col in ["CodigoIniciativa", "Empresa", "Tipologia", "Mes",

"MesIdx", "Tag"]:

nulos = df[col].isna().sum()

if nulos > 0:

[Link](f"[xpu_ProyectoMesTag] La columna '{col}'

tiene {nulos} nulos.")

df = tablas["xpu_CapacidadRolEmpresa"]

for col in ["Empresa", "Rol", "CapacidadParalelo"]:

nulos = df[col].isna().sum()

if nulos > 0:

[Link](f"[xpu_CapacidadRolEmpresa] La columna
'{col}' tiene {nulos} nulos.")

df = tablas["xpu_RuteoBalanceRol"]

for col in ["Empresa", "RolBase", "Tag", "Tipologia", "TipoRuteo"]:

nulos = df[col].isna().sum()

if nulos > 0:

[Link](f"[xpu_RuteoBalanceRol] La columna '{col}'

tiene {nulos} nulos.")

return errores

def validar_duplicados(tablas: Dict[str, [Link]]) -> List[str]:

errores = []
 df = tablas["xpu_CapacidadRolEmpresa"]

dup = [Link](subset=["Empresa", "Rol"], keep=False).sum()

if dup > 0:

[Link](f"[xpu_CapacidadRolEmpresa] Hay {dup} filas

duplicadas por Empresa+Rol.")

df = tablas["xpu_DotacionActual"]

dup = [Link](subset=["Empresa", "Rol"], keep=False).sum()

if dup > 0:

[Link](f"[xpu_DotacionActual] Hay {dup} filas

duplicadas por Empresa+Rol.")

df = tablas["xpu_RuteoBalanceRol"]

dup = [Link](subset=["Empresa", "RolBase", "Tag",

"Tipologia"], keep=False).sum()

if dup > 0:

[Link](f"[xpu_RuteoBalanceRol] Hay {dup} filas

duplicadas por Empresa+RolBase+Tag+Tipologia.")

return errores

def construir_parametros(df_parametros: [Link]) -> Dict[str,

object]:

parametros = {}

activos = df_parametros[df_parametros["Activo"] == 1].copy()

for _, row in [Link]():

nombre = row["Parametro"]

valor = row["Valor"]
 tipo = str(row["TipoDato"]).strip().lower() if not
[Link](row["TipoDato"]) else "text"

if tipo == "int":

try:

valor = int(valor)

except Exception:

pass

elif tipo in ("number", "float", "decimal"):

try:

valor = float(valor)

except Exception:

pass

else:

valor = limpiar_texto(valor)

parametros[nombre] = valor

return parametros

def imprimir_resumen_tablas(tablas: Dict[str, [Link]]) ->

None:

print("\nRESUMEN DE TABLAS CARGADAS")

print("-" * 70)

for nombre, df in [Link]():

print(f"{nombre:<24} -> {[Link][0]:>8} filas x

{[Link][1]:>3} columnas")

print("-" * 70)
#
==========================================
===============

# FASE 2 - EXPANSIÓN DE REGLAS

#
==========================================
===============

def expandir_reglas_rol(df_roles: [Link]) -> [Link]:

filas = []

for _, row in df_roles.iterrows():

rol = row["Rol"]

tags = partir_lista_pipe(row["TagsVisibles"])

tips = partir_lista_pipe(row["TipologiasPermitidas"])

excluye = limpiar_texto([Link]("ExcluyeTipologia"))

orden_visual = [Link]("OrdenVisual")

comentario = limpiar_texto([Link]("Comentario"))

if not tags or not tips:

continue

for tag in tags:

for tip in tips:

[Link](

"Rol": rol,

"TagVisible": limpiar_texto(tag),

"TipologiaPermitida": limpiar_texto(tip),

"ExcluyeTipologia": excluye,
 "OrdenVisual": orden_visual,

"ComentarioRegla": comentario,

"ReglaAplicada": f"{rol}|{tag}|{tip}",

df = [Link](filas)

if not [Link]:

df = normalizar_dataframe(df)

df = df.sort_values(

by=["OrdenVisual", "Rol", "TagVisible", "TipologiaPermitida"],

ascending=[True, True, True, True],

na_position="last"

).reset_index(drop=True)

return df

def expandir_ruteo(df_ruteo: [Link]) -> [Link]:

filas = []

for _, row in df_ruteo.iterrows():

empresa = row["Empresa"]

rol_base = row["RolBase"]

rol_apoyo = row["RolApoyo"]

tags = partir_lista_pipe(row["Tag"])

tips = partir_lista_pipe(row["Tipologia"])

tipo_ruteo = row["TipoRuteo"]

prioridad = row["PrioridadRuteo"]

limitar = row["LimitarBasePorDotacion"]
 heatmap = row["HeatmapApoyoSinTopeDotacion"]

desborde = row["PermiteDesborde"]

comentario = row["Comentario"]

if not tags or not tips:

continue

for tag in tags:

for tip in tips:

[Link](

"Empresa": limpiar_texto(empresa),

"RolBase": limpiar_texto(rol_base),

"RolApoyo": limpiar_texto(rol_apoyo),

"Tag": limpiar_texto(tag),

"Tipologia": limpiar_texto(tip),

"TipoRuteo": limpiar_texto(tipo_ruteo),

"PrioridadRuteo": prioridad,

"LimitarBasePorDotacion": limitar,

"HeatmapApoyoSinTopeDotacion": heatmap,

"PermiteDesborde": desborde,

"Comentario": limpiar_texto(comentario),

"ClaveRuteoExpandida": f"{empresa}|{rol_base}|
{tag}|{tip}",

df = [Link](filas)

if not [Link]:
 df = normalizar_dataframe(df)

df = df.sort_values(

by=["Empresa", "RolBase", "PrioridadRuteo", "Tag",

"Tipologia"],

ascending=[True, True, True, True, True],

na_position="last"

).reset_index(drop=True)

return df

def excluir_roles_apoyo_directos(

df_candidatos: [Link],

df_ruteo_expandido: [Link],

) -> [Link]:

"""

Excluye candidatos que entran por visibilidad directa en un rol de

apoyo

cuando esa misma combinación Empresa + RolApoyo + Tag +

Tipologia está

gobernada por una regla de desborde.

"""

apoyo = df_ruteo_expandido.copy()

apoyo = apoyo[

apoyo["TipoRuteo"].astype(str).[Link]().eq("Desborde")

][["Empresa", "RolApoyo", "Tag", "Tipologia"]].drop_duplicates()

apoyo = [Link](columns={"RolApoyo": "Rol"})

apoyo["EsApoyoCondicionado"] = 1
 salida = df_candidatos.merge(

apoyo,

on=["Empresa", "Rol", "Tag", "Tipologia"],

how="left"

salida = salida[salida["EsApoyoCondicionado"].isna()].copy()

salida = [Link](columns=["EsApoyoCondicionado"])

return salida

def restringir_candidatos_a_roles_base_explicitos(

df_candidatos: [Link],

df_ruteo_expandido: [Link],

) -> [Link]:

"""

Si existe una regla explícita de ruteo para Empresa + Tag +

Tipologia,

entonces solo se permiten como candidatos directos los roles que

figuran

como RolBase para esa combinación.

"""

candidatos = df_candidatos.copy()

# 1) Combinaciones que tienen ruteo explícito

combos_existentes = (

df_ruteo_expandido[["Empresa", "Tag", "Tipologia"]]

.drop_duplicates()

.copy()
 )

combos_existentes["__flag_ruteo_explicito__"] = 1

# 2) Roles base permitidos para esas combinaciones

combos_roles_base = (

df_ruteo_expandido[["Empresa", "Tag", "Tipologia", "RolBase"]]

.drop_duplicates()

.copy()

.rename(columns={"RolBase": "__rol_base_permitido__"})

# 3) Marcar si la combinación Empresa+Tag+Tipologia tiene ruteo

explícito

salida = [Link](

combos_existentes,

on=["Empresa", "Tag", "Tipologia"],

how="left"

# 4) Marcar si el rol candidato coincide con un RolBase permitido

salida = [Link](

combos_roles_base,

left_on=["Empresa", "Tag", "Tipologia", "Rol"],

right_on=["Empresa", "Tag", "Tipologia",

"__rol_base_permitido__"],

how="left"

# 5) Regla:

# - si NO hay ruteo explícito -> conservar

 # - si SÍ hay ruteo explícito -> conservar solo si el rol es RolBase
permitido

salida = salida[

(salida["__flag_ruteo_explicito__"].isna())

| (salida["__rol_base_permitido__"].notna())

].copy()

# 6) Limpiar columnas técnicas

salida = [Link](columns=["__flag_ruteo_explicito__",
"__rol_base_permitido__"])

return salida

#
==========================================
===============

# FASE 3 - GENERACIÓN DE CANDIDATOS

#
==========================================
===============

def generar_candidatos(

df_proyecto_mes_tag: [Link],

df_reglas_expandidas: [Link],

df_ruteo_expandido: [Link],

parametros: Dict[str, object],

) -> [Link]:

usar_tags_visibles = int([Link]("usar_tags_visibles", 1))

== 1

incluir_tipologias_no_configuradas =
int([Link]("incluir_tipologias_no_configuradas", 0)) == 1
 base = df_proyecto_mes_tag.copy()

if not usar_tags_visibles:

raise ValueError("La versión actual del motor espera

usar_tags_visibles = 1.")

reglas = df_reglas_expandidas.copy()

df = [Link](

reglas,

left_on=["Tag", "Tipologia"],

right_on=["TagVisible", "TipologiaPermitida"],

how="inner",

if "ExcluyeTipologia" in [Link]:

mask_excluir = (

df["ExcluyeTipologia"].notna()

& (df["ExcluyeTipologia"].astype(str).[Link]() != "")

& (df["Tipologia"].astype(str).[Link]() ==
df["ExcluyeTipologia"].astype(str).[Link]())

df = [Link][~mask_excluir].copy()

if incluir_tipologias_no_configuradas:

tips_base =
set(base["Tipologia"].dropna().astype(str).[Link]().unique())

tips_reglas =
set(reglas["TipologiaPermitida"].dropna().astype(str).[Link]().unique(
))

tips_no_config = sorted(tips_base - tips_reglas)

 if tips_no_config:

print("\nALERTA DE CONTROL")

print("-" * 70)

print("Tipologías presentes en calendario sin regla explícita en

tblRolReglas:")

for t in tips_no_config:

print(f"- {t}")

print("-" * 70)

df["ProyectoCuenta"] = 1

df["ProyectoMesRolKey"] = (

df["CodigoIniciativa"].astype(str)

+ "|"

+ df["Empresa"].astype(str)

+ "|"

+ df["Mes"].astype(str)

+ "|"

+ df["Rol"].astype(str)

columnas_salida = [

"CodigoIniciativa",

"Empresa",

"NombreCorto",

"Tipologia",

"Ciclo",

"Comentario",

"Ppto2026",
 "Mes",

"MesIdx",

"Tag",

"Rol",

"ProyectoCuenta",

"ReglaAplicada",

"OrdenVisual",

"ComentarioRegla",

"FuenteTabla",

"ProyectoMesKey",

"ProyectoMesTagKey",

"ProyectoMesRolKey",

df = df[columnas_salida].copy()

# Corrección 1:

# un rol de apoyo no debe cargar directo si esa combinación está

condicionada por desborde

df = excluir_roles_apoyo_directos(df, df_ruteo_expandido)

# Corrección 2:

# si existe ruteo explícito para Empresa + Tag + Tipologia,

# solo deben quedar como candidatos directos los RolBase

definidos

df = restringir_candidatos_a_roles_base_explicitos(df,
df_ruteo_expandido)

df = df.sort_values(

by=["Empresa", "Rol", "MesIdx", "CodigoIniciativa"],

 ascending=[True, True, True, True],

na_position="last"

).reset_index(drop=True)

return df

#
==========================================
===============

# FASE 4 - CAPACIDAD Y RUTEO

#
==========================================
===============

def agregar_capacidad_y_dotacion(

df_candidatos: [Link],

df_capacidad: [Link],

df_dotacion: [Link],

parametros: Dict[str, object],

) -> [Link]:

df = df_candidatos.copy()

cap = df_capacidad[["Empresa", "Rol", "CapacidadParalelo"]].copy()

dot = df_dotacion[["Empresa", "Rol", "DotacionActual"]].copy()

df = [Link](cap, on=["Empresa", "Rol"], how="left")

df = [Link](dot, on=["Empresa", "Rol"], how="left")

usar_999 = int([Link]("capacidad_999_sin_restriccion", 1))

== 1
 df["EsSinRestriccion"] = 0

if usar_999:

df["EsSinRestriccion"] = df["CapacidadParalelo"].apply(

lambda x: 1 if [Link](x) and float(x) >= 900 else 0

df["EsTagCargaTrabajo"] = df["Tag"].apply(lambda x: 1 if x in
{"EP", "ETO", "OB"} else 0)

df["CapacidadFuente"] = "xpu_CapacidadRolEmpresa"

df["DotacionFuente"] = "xpu_DotacionActual"

return df

def calcular_cupos_apoyo_proporcional(

gg: [Link],

limite_base: int | None,

) -> dict[str, int]:

"""

Calcula cuántos proyectos de cada tipología deben ir al rol de

apoyo

según reparto proporcional del exceso.

"""

if limite_base is None:

return {}

total_grupo = len(gg)

exceso_total = max(0, total_grupo - int(limite_base))

 if exceso_total <= 0:

return {}

conteo_tip = (

[Link]("Tipologia", dropna=False)

.size()

.reset_index(name="ProyectosGrupo")

conteo_tip["ProporcionGrupo"] = conteo_tip["ProyectosGrupo"] /
total_grupo

conteo_tip["ExcesoIdeal"] = conteo_tip["ProporcionGrupo"] *
exceso_total

conteo_tip["ExcesoEntero"] =
conteo_tip["ExcesoIdeal"].apply([Link]).astype(int)

conteo_tip["ParteDecimal"] = conteo_tip["ExcesoIdeal"] -
conteo_tip["ExcesoEntero"]

conteo_tip["PrioridadTipologia"] =
conteo_tip["Tipologia"].map(PRIORIDAD_TIPOLOGIA).fillna(999).astype
(int)

asignado_entero = int(conteo_tip["ExcesoEntero"].sum())

saldo = int(exceso_total - asignado_entero)

if saldo > 0:

conteo_tip = conteo_tip.sort_values(

by=["ParteDecimal", "PrioridadTipologia", "Tipologia"],

ascending=[False, True, True],

na_position="last"

).reset_index(drop=True)
 conteo_tip["Extra"] = 0

for i in range(saldo):

conteo_tip.loc[i, "Extra"] = 1

conteo_tip["ExcesoAsignadoApoyo"] =
conteo_tip["ExcesoEntero"] + conteo_tip["Extra"]

else:

conteo_tip["ExcesoAsignadoApoyo"] =
conteo_tip["ExcesoEntero"]

# Seguridad: no exceder el volumen real de cada tipología

conteo_tip["ExcesoAsignadoApoyo"] = conteo_tip.apply(

lambda r: min(int(r["ExcesoAsignadoApoyo"]),
int(r["ProyectosGrupo"])),

axis=1

cupos = dict(

zip(

conteo_tip["Tipologia"].astype(str),

conteo_tip["ExcesoAsignadoApoyo"].astype(int),

return cupos

def asignar_grupo_desborde_proporcional(

gg: [Link],

rol_base: str,

rol_apoyo: str,
 limite_base: int | None,

permite_desborde: int,

) -> [Link]:

"""

Asigna proyectos de un grupo de ruteo con desborde proporcional

por tipología.

"""

g = [Link]()

g["PrioridadTipologia"] =
g["Tipologia"].map(PRIORIDAD_TIPOLOGIA).fillna(999).astype(int)

g["RondaBarrido"] = None

g["RolOriginal"] = g["Rol"]

g["FlagRuteado"] = 1

g["LimiteBaseGrupo"] = limite_base

if limite_base is None:

g["RolAsignadoFinal"] = rol_base

g["OrigenAsignacion"] = "RuteoBaseSinLimite"

g["SecuenciaGrupo"] = range(1, len(g) + 1)

return g

if permite_desborde != 1:

g["RolAsignadoFinal"] = rol_base

g["OrigenAsignacion"] = "RuteoBaseExcesoSinDesborde"

g["SecuenciaGrupo"] = range(1, len(g) + 1)

return g

cupos_apoyo = calcular_cupos_apoyo_proporcional(g, limite_base)

 piezas = []

secuencia_global = 1

for tip in sorted(g["Tipologia"].dropna().astype(str).unique(),

key=lambda x: PRIORIDAD_TIPOLOGIA.get(x, 999)):

sub = g[g["Tipologia"] == tip].copy()

sub = sub.sort_values(

by=["CodigoIniciativa", "ProyectoMesTagKey"],

ascending=[True, True],

na_position="last"

).reset_index(drop=True)

apoyo_tip = int(cupos_apoyo.get(str(tip), 0))

total_tip = len(sub)

base_tip = max(0, total_tip - apoyo_tip)

sub["SecuenciaGrupo"] = range(secuencia_global,
secuencia_global + total_tip)

secuencia_global += total_tip

# El base se queda con los primeros proyectos ordenados; el

resto va a apoyo

sub["RolAsignadoFinal"] = [Link].to_series().apply(

lambda i: rol_base if i < base_tip else rol_apoyo

sub["OrigenAsignacion"] = [Link].to_series().apply(

lambda i: "RuteoBase" if i < base_tip else "RuteoApoyo"

)
 sub["CupoApoyoTipologia"] = apoyo_tip

sub["CupoBaseTipologia"] = base_tip

[Link](sub)

if not piezas:

return [Link][0:0].copy()

salida = [Link](piezas, ignore_index=True)

# Reorden final estable

salida = salida.sort_values(

by=["PrioridadTipologia", "CodigoIniciativa",
"ProyectoMesTagKey"],

ascending=[True, True, True],

na_position="last"

).reset_index(drop=True)

return salida

def aplicar_ruteo_base_apoyo(

df_con_capacidad: [Link],

df_ruteo_expandido: [Link],

) -> [Link]:

base = df_con_capacidad.copy()

ruteo = df_ruteo_expandido.copy()

df = [Link](
 ruteo,

left_on=["Empresa", "Rol", "Tag", "Tipologia"],

right_on=["Empresa", "RolBase", "Tag", "Tipologia"],

how="left",

suffixes=("", "_rut"),

sin_regla = df[df["TipoRuteo"].isna()].copy()

sin_regla["RolOriginal"] = sin_regla["Rol"]

sin_regla["RolAsignadoFinal"] = sin_regla["Rol"]

sin_regla["OrigenAsignacion"] = "VisibleSinRuteo"

sin_regla["FlagRuteado"] = 0

sin_regla["SecuenciaGrupo"] = None

sin_regla["PrioridadTipologia"] =
sin_regla["Tipologia"].map(PRIORIDAD_TIPOLOGIA).fillna(999).astype(i
nt)

sin_regla["RondaBarrido"] = None

sin_regla["LimiteBaseGrupo"] = None

sin_regla["CupoApoyoTipologia"] = None

sin_regla["CupoBaseTipologia"] = None

con_regla = df[df["TipoRuteo"].notna()].copy()

resultados = []

if not con_regla.empty:

con_regla["PrioridadRuteo"] =
pd.to_numeric(con_regla["PrioridadRuteo"], errors="coerce")

con_regla["LimitarBasePorDotacion"] =
pd.to_numeric(con_regla["LimitarBasePorDotacion"],
errors="coerce").fillna(0).astype(int)
 con_regla["PermiteDesborde"] =
pd.to_numeric(con_regla["PermiteDesborde"],
errors="coerce").fillna(0).astype(int)

con_regla["DotacionActual"] =
pd.to_numeric(con_regla["DotacionActual"], errors="coerce")

con_regla["CapacidadParalelo"] =
pd.to_numeric(con_regla["CapacidadParalelo"], errors="coerce")

con_regla["PrioridadTipologia"] =
con_regla["Tipologia"].map(PRIORIDAD_TIPOLOGIA).fillna(999).astype(
int)

cols_group = [

"Empresa",

"Mes",

"MesIdx",

"Tag",

"RolBase",

"RolApoyo",

"TipoRuteo",

"PrioridadRuteo",

for _, g in con_regla.groupby(cols_group, dropna=False):

gg = [Link]()

tipo_ruteo = limpiar_texto(gg["TipoRuteo"].iloc[0])

rol_base = limpiar_texto(gg["RolBase"].iloc[0])

rol_apoyo = limpiar_texto(gg["RolApoyo"].iloc[0])

permite_desborde = int(gg["PermiteDesborde"].iloc[0]) if
[Link](gg["PermiteDesborde"].iloc[0]) else 0
 limitar_base_por_dotacion =
int(gg["LimitarBasePorDotacion"].iloc[0]) if
[Link](gg["LimitarBasePorDotacion"].iloc[0]) else 0

capacidad_base = gg["CapacidadParalelo"].dropna()

capacidad_base = float(capacidad_base.iloc[0]) if not

capacidad_base.empty else None

dotacion_base = gg["DotacionActual"].dropna()

dotacion_base = float(dotacion_base.iloc[0]) if not

dotacion_base.empty else None

limite_base = None

if tipo_ruteo == "SinApoyo":

limite_base = None

else:

if limitar_base_por_dotacion == 1 and dotacion_base is not

None and capacidad_base is not None and capacidad_base < 900:

limite_base = int(dotacion_base * capacidad_base)

elif capacidad_base is not None and capacidad_base < 900:

limite_base = int(capacidad_base)

else:

limite_base = None

if tipo_ruteo == "SinApoyo":

gg["RolOriginal"] = gg["Rol"]

gg["FlagRuteado"] = 1

gg["RolAsignadoFinal"] = rol_base

gg["OrigenAsignacion"] = "RuteoSinApoyo"

gg["SecuenciaGrupo"] = range(1, len(gg) + 1)

 gg["RondaBarrido"] = None

gg["LimiteBaseGrupo"] = limite_base

gg["CupoApoyoTipologia"] = None

gg["CupoBaseTipologia"] = None

[Link](gg)

elif tipo_ruteo == "Desborde":

gg_res = asignar_grupo_desborde_proporcional(

gg=gg,

rol_base=rol_base,

rol_apoyo=rol_apoyo,

limite_base=limite_base,

permite_desborde=permite_desborde,

[Link](gg_res)

else:

gg["RolOriginal"] = gg["Rol"]

gg["FlagRuteado"] = 1

gg["RolAsignadoFinal"] = rol_base

gg["OrigenAsignacion"] = "RuteoBaseDefault"

gg["SecuenciaGrupo"] = range(1, len(gg) + 1)

gg["RondaBarrido"] = None

gg["LimiteBaseGrupo"] = limite_base

gg["CupoApoyoTipologia"] = None

gg["CupoBaseTipologia"] = None

[Link](gg)
 con_regla_res = [Link](resultados, ignore_index=True) if
resultados else [Link](columns=[Link]())

detalle = [Link]([sin_regla, con_regla_res], ignore_index=True,

sort=False)

detalle["Rol"] = detalle["RolAsignadoFinal"]

detalle["ProyectoMesRolKey"] = (

detalle["CodigoIniciativa"].astype(str)

+ "|"

+ detalle["Empresa"].astype(str)

+ "|"

+ detalle["Mes"].astype(str)

+ "|"

+ detalle["Rol"].astype(str)

columnas_finales = [

"CodigoIniciativa",

"Empresa",

"NombreCorto",

"Tipologia",

"Ciclo",

"Comentario",

"Ppto2026",

"Mes",

"MesIdx",

"Tag",

"Rol",
 "RolOriginal",

"RolAsignadoFinal",

"ProyectoCuenta",

"EsTagCargaTrabajo",

"CapacidadParalelo",

"DotacionActual",

"EsSinRestriccion",

"CapacidadFuente",

"DotacionFuente",

"TipoRuteo",

"PrioridadRuteo",

"PrioridadTipologia",

"RondaBarrido",

"SecuenciaGrupo",

"OrigenAsignacion",

"FlagRuteado",

"LimiteBaseGrupo",

"CupoBaseTipologia",

"CupoApoyoTipologia",

"FuenteTabla",

"ProyectoMesKey",

"ProyectoMesTagKey",

"ProyectoMesRolKey",

for col in columnas_finales:

if col not in [Link]:

detalle[col] = None
 detalle = detalle[columnas_finales].copy()

detalle = detalle.sort_values(

by=["Empresa", "Rol", "MesIdx", "PrioridadTipologia",

"CodigoIniciativa", "Tag"],

ascending=[True, True, True, True, True, True],

na_position="last"

).reset_index(drop=True)

return detalle

def construir_secuencia_barrido_iterativo(gg: [Link]) ->

[Link]:

"""

Construye una secuencia round-robin por tipología:

1 Distribución, 2 Transmisión, 3 Generación, 4 Edificaciones, 5 TIC,

6 Innovación

y vuelve a empezar hasta agotar todos los proyectos del grupo.

"""

orden_tipologias = [

"Distribución",

"Transmisión",

"Generación",

"Edificaciones",

"TIC",

"Innovación",

grupos = {}
 for tip in orden_tipologias:

sub = gg[gg["Tipologia"] == tip].copy()

sub = sub.sort_values(

by=["CodigoIniciativa", "ProyectoMesTagKey"],

ascending=[True, True],

na_position="last"

).reset_index(drop=True)

grupos[tip] = sub

piezas = []

contador_global = 1

ronda = 1

while True:

hubo_fila = False

for tip in orden_tipologias:

sub = grupos[tip]

if ronda <= len(sub):

fila = [Link][[ronda - 1]].copy()

fila["RondaBarrido"] = ronda

fila["SecuenciaGrupo"] = contador_global

fila["PrioridadTipologia"] = PRIORIDAD_TIPOLOGIA.get(tip,
999)

[Link](fila)

contador_global += 1

hubo_fila = True
 if not hubo_fila:

break

ronda += 1

if not piezas:

return [Link][0:0].copy()

salida = [Link](piezas, ignore_index=True)

salida = salida.reset_index(drop=True)

return salida

def clasificar_cobertura_capacidad(df_detalle_preliminar:
[Link]) -> [Link]:

df = df_detalle_preliminar.copy()

df["EstadoCapacidad"] = "ConCapacidad"

[Link][df["CapacidadParalelo"].isna(), "EstadoCapacidad"] =
"SinCapacidad"

[Link][

(df["CapacidadParalelo"].notna()) & (df["EsSinRestriccion"] ==

1),

"EstadoCapacidad"

] = "SinRestriccion999"

return df
def construir_detalle_calculo(df_detalle_preliminar: [Link]) ->
[Link]:

df = clasificar_cobertura_capacidad(df_detalle_preliminar)

# Base válida para cálculo:

# - Con capacidad operativa

# - O sin restricción 999

df_calculo = df[

(df["EstadoCapacidad"] == "ConCapacidad")

| (df["EstadoCapacidad"] == "SinRestriccion999")

].copy()

return df, df_calculo

def calcular_bolsa_rol_mes(

df_detalle_calculo: [Link],

parametros: Dict[str, object],

) -> [Link]:

df = df_detalle_calculo.copy()

# Solo tags de carga real

df = df[df["EsTagCargaTrabajo"] == 1].copy()

modo_conteo = int([Link]("modo_conteo_proyecto_mes",
1))

mostrar_trab_999 =
int([Link]("mostrar_trabajadores_requeridos_en_cap999", 0))

if modo_conteo != 1:
 raise ValueError("La versión actual espera
modo_conteo_proyecto_mes = 1.")

agrupado = (

[Link](["Empresa", "Rol", "Mes", "MesIdx"], dropna=False)

.agg(

TotalProyectosCargaRolMes=("CodigoIniciativa", lambda s:
[Link]().astype(str).nunique()),

CapacidadParalelo=("CapacidadParalelo", "max"),

EsSinRestriccion=("EsSinRestriccion", "max"),

.reset_index()

def calcular_trabajadores(row):

cap = row["CapacidadParalelo"]

total = row["TotalProyectosCargaRolMes"]

sinr = row["EsSinRestriccion"]

if [Link](cap):

return None

if sinr == 1:

return None if mostrar_trab_999 == 0 else round(float(total),

2)

if float(cap) <= 0:

return None

return round(float(total) / float(cap), 2)

 agrupado["TrabajadoresRequeridosRolMes"] =
[Link](calcular_trabajadores, axis=1)

def calcular_indicador(row):

if row["EsSinRestriccion"] == 1:

return round(float(row["TotalProyectosCargaRolMes"]), 2)

return row["TrabajadoresRequeridosRolMes"]

agrupado["IndicadorDashboardRolMes"] =
[Link](calcular_indicador, axis=1)

agrupado["TipoIndicadorDashboard"] =
agrupado["EsSinRestriccion"].apply(

lambda x: "Proyectos de carga" if x == 1 else "Trabajadores

requeridos"

agrupado["MesOrden"] = agrupado["MesIdx"].astype(str).[Link](3)
+ "|" + agrupado["Mes"].astype(str)

agrupado["FlagCalculoValido"] =
agrupado["CapacidadParalelo"].apply(

lambda x: 0 if [Link](x) else 1

agrupado = agrupado.sort_values(

by=["Empresa", "Rol", "MesIdx"],

ascending=[True, True, True],

na_position="last"

).reset_index(drop=True)
 return agrupado

def calcular_bolsa_rol_mes_tipologia(

df_detalle_calculo: [Link],

parametros: Dict[str, object],

) -> [Link]:

df = df_detalle_calculo.copy()

df = df[df["EsTagCargaTrabajo"] == 1].copy()

modo_conteo = int([Link]("modo_conteo_proyecto_mes",
1))

mostrar_trab_999 =
int([Link]("mostrar_trabajadores_requeridos_en_cap999", 0))

if modo_conteo != 1:

raise ValueError("La versión actual espera

modo_conteo_proyecto_mes = 1.")

agrupado = (

[Link](["Empresa", "Rol", "Tipologia", "Mes", "MesIdx"],

dropna=False)

.agg(

TotalProyectosCargaRolMesTipologia=("CodigoIniciativa",
lambda s: [Link]().astype(str).nunique()),

CapacidadParalelo=("CapacidadParalelo", "max"),

EsSinRestriccion=("EsSinRestriccion", "max"),

.reset_index()

)
 def calcular_indicador(row):

cap = row["CapacidadParalelo"]

total = row["TotalProyectosCargaRolMesTipologia"]

sinr = row["EsSinRestriccion"]

if [Link](cap):

return None

if sinr == 1:

return round(float(total), 2)

if float(cap) <= 0:

return None

return round(float(total) / float(cap), 2)

agrupado["IndicadorDashboardRolMesTipologia"] =
[Link](calcular_indicador, axis=1)

agrupado["PrioridadTipologia"] =
agrupado["Tipologia"].map(PRIORIDAD_TIPOLOGIA).fillna(999).astype(
int)

agrupado["TipologiaOrdenHeatmap"] =
agrupado["PrioridadTipologia"].astype(str).[Link](2) + " - " +
agrupado["Tipologia"].astype(str)

agrupado = agrupado.sort_values(

by=["Empresa", "Rol", "PrioridadTipologia", "MesIdx"],

ascending=[True, True, True, True],

na_position="last"

).reset_index(drop=True)
 return agrupado

def calcular_necesidad_mensual(

df_detalle_calculo: [Link],

df_bolsa_rol_mes: [Link],

) -> [Link]:

df = df_detalle_calculo.copy()

df = df[df["EsTagCargaTrabajo"] == 1].copy()

agrupado_tipologia = (

[Link](["Empresa", "Rol", "Tipologia", "Mes", "MesIdx"],

dropna=False)

.agg(

ProyectosVisiblesTipologia=("CodigoIniciativa", lambda s:
[Link]().astype(str).nunique())

.reset_index()

salida = agrupado_tipologia.merge(

df_bolsa_rol_mes,

on=["Empresa", "Rol", "Mes", "MesIdx"],

how="left"

salida = salida.sort_values(

by=["Empresa", "Rol", "Tipologia", "MesIdx"],

ascending=[True, True, True, True],

 na_position="last"

).reset_index(drop=True)

return salida

def resumen_cobertura_capacidad(df_detalle_clasificado:
[Link]) -> [Link]:

resumen = (

df_detalle_clasificado.groupby("EstadoCapacidad",
dropna=False)

.size()

.reset_index(name="Filas")

.sort_values(by="EstadoCapacidad")

.reset_index(drop=True)

return resumen

def calcular_resumen_rol_empresa(df_bolsa_rol_mes: [Link]) -

> [Link]:

filas = []

for (empresa, rol), g in df_bolsa_rol_mes.groupby(["Empresa",

"Rol"], dropna=False):

gg = g.sort_values(

by=["TotalProyectosCargaRolMes", "MesIdx"],

ascending=[False, True],

na_position="last"

).reset_index(drop=True)

mes_pico = [Link][0]["Mes"] if not [Link] else None

 proy = gg["TotalProyectosCargaRolMes"].dropna()

trab = gg["TrabajadoresRequeridosRolMes"].dropna()

sinr = gg["EsSinRestriccion"].dropna()

meses_activos = (gg["TotalProyectosCargaRolMes"] > 0).sum()

[Link](

"Empresa": empresa,

"Rol": rol,

"MesPico": mes_pico,

"PicoProyectosCarga": [Link]() if not [Link] else

None,

"PromedioProyectosCarga": round([Link](), 2) if not

[Link] else None,

"PicoTrabajadoresRequeridos": [Link]() if not [Link]

else None,

"PromedioTrabajadoresRequeridos": round([Link](), 2) if
not [Link] else None,

"MesesConActividad": int(meses_activos),

"EsSinRestriccion": [Link]() if not [Link] else None,

df = [Link](filas)

df = df.sort_values(by=["Empresa", "Rol"], ascending=[True, True],

na_position="last").reset_index(drop=True)

return df
def calcular_resumen_rol_empresa_tipologia(df_necesidad_mensual:
[Link]) -> [Link]:

filas = []

for (empresa, rol, tipologia), g in

df_necesidad_mensual.groupby(["Empresa", "Rol", "Tipologia"],
dropna=False):

proy_tip = g["ProyectosVisiblesTipologia"].dropna()

proy_bolsa = g["TotalProyectosCargaRolMes"].dropna()

trab = g["TrabajadoresRequeridosRolMes"].dropna()

sinr = g["EsSinRestriccion"].dropna()

meses_act = (g["ProyectosVisiblesTipologia"] > 0).sum()

[Link](

"Empresa": empresa,

"Rol": rol,

"Tipologia": tipologia,

"PicoProyectosCargaTipologia": proy_tip.max() if not

proy_tip.empty else None,

"PromedioProyectosCargaTipologia": round(proy_tip.mean(),
2) if not proy_tip.empty else None,

"MesesConActividad": int(meses_act),

"VolumenTotalProyectoMesTipologia": proy_tip.sum() if not

proy_tip.empty else None,

"PicoBolsaCargaRolMes": proy_bolsa.max() if not

proy_bolsa.empty else None,

"PicoTrabajadoresRequeridosRolMes": [Link]() if not

[Link] else None,

"PromedioTrabajadoresRequeridosRolMes":
round([Link](), 2) if not [Link] else None,
 "EsSinRestriccion": [Link]() if not [Link] else None,

df = [Link](filas)

df = df.sort_values(by=["Empresa", "Rol", "Tipologia"],

ascending=[True, True, True],
na_position="last").reset_index(drop=True)

return df

def calcular_plazas_requeridas(

df_bolsa_rol_mes: [Link],

df_resumen_rol_empresa: [Link],

) -> [Link]:

bolsa = (

df_bolsa_rol_mes.groupby(["Empresa", "Rol"], dropna=False)

.agg(

CapacidadParalelo=("CapacidadParalelo", "max"),

EsSinRestriccion=("EsSinRestriccion", "max"),

PicoProyectosCarga=("TotalProyectosCargaRolMes", "max"),

PromedioProyectosCarga=("TotalProyectosCargaRolMes",
"mean"),

.reset_index()

salida = [Link](

df_resumen_rol_empresa[["Empresa", "Rol", "MesPico"]],

on=["Empresa", "Rol"],

how="left"
)

def tipo_regla_plaza(rol):

if rol == "Coord. BIM":

return "BIM_2"

if rol in {

"Esp. Servidumbre",

"Esp. Ambiental",

"Esp. Legal",

"Esp. Relaciones Comunitarias",

"Coord. Corporativo BIM",

}:

return "FIJA_1"

return "CAPACIDAD"

salida["TipoReglaPlaza"] = salida["Rol"].apply(tipo_regla_plaza)

def plazas(row):

regla = row["TipoReglaPlaza"]

cap = row["CapacidadParalelo"]

pico = row["PicoProyectosCarga"]

if regla == "BIM_2":

return 2

if regla == "FIJA_1":

return 1

if [Link](cap) or float(cap) <= 0 or [Link](pico):

return None

if float(cap) >= 900:

 return None

return int(-(-float(pico) // float(cap)))

salida["PlazasRequeridas"] = [Link](plazas, axis=1)

salida["PromedioProyectosCarga"] =
salida["PromedioProyectosCarga"].round(2)

salida = salida.sort_values(by=["Empresa", "Rol"],

ascending=[True, True], na_position="last").reset_index(drop=True)

return salida

def calcular_asignacion_proyecto_plaza(

df_detalle_calculo: [Link],

df_plazas_requeridas: [Link],

) -> [Link]:

source = df_detalle_calculo.copy()

proyecto_visible = (

[Link](["Empresa", "Rol", "CodigoIniciativa",

"NombreCorto", "Tipologia"], dropna=False)

.agg(

MesInicioVisibleIdx=("MesIdx", "min"),

MesFinVisibleIdx=("MesIdx", "max"),

DuracionVisibleMeses=("MesIdx", lambda s:
[Link]().nunique()),

TagsProyecto=("Tag", lambda s: sorted(set([x for x in

[Link]().astype(str)]))),

.reset_index()

)
 proyecto_carga = (

source[source["EsTagCargaTrabajo"] == 1]

.groupby(["Empresa", "Rol", "CodigoIniciativa"], dropna=False)

.agg(

MesInicioCargaIdx=("MesIdx", "min"),

MesFinCargaIdx=("MesIdx", "max"),

DuracionCargaMeses=("MesIdx", lambda s:
[Link]().nunique()),

.reset_index()

salida = proyecto_visible.merge(

proyecto_carga,

on=["Empresa", "Rol", "CodigoIniciativa"],

how="left"

salida["DuracionCargaMeses"] =
salida["DuracionCargaMeses"].fillna(0)

def tag_principal(tags):

tags = tags if isinstance(tags, list) else []

prioridad = ["OB", "ETO", "EP", "CO", "CE2", "CE1", "TR"]

for t in prioridad:

if t in tags:

return t

return tags[0] if tags else None

 salida["TagPrincipalBIM"] =
salida["TagsProyecto"].apply(tag_principal)

salida = [Link](

df_plazas_requeridas[["Empresa", "Rol", "PlazasRequeridas",

"CapacidadParalelo", "EsSinRestriccion", "TipoReglaPlaza"]],

on=["Empresa", "Rol"],

how="left"

grupos = []

for (empresa, rol), g in [Link](["Empresa", "Rol"],

dropna=False):

gg = [Link]()

regla = gg["TipoReglaPlaza"].dropna()

regla = [Link][0] if not [Link] else None

plazas = gg["PlazasRequeridas"].dropna()

plazas = int([Link][0]) if not [Link] else None

if regla == "FIJA_1":

gg = gg.sort_values(

by=["Tipologia", "CodigoIniciativa"],

ascending=[True, True],

na_position="last"

elif regla == "BIM_2":

gg = gg.sort_values(
 by=["Tipologia", "TagPrincipalBIM", "DuracionVisibleMeses",
"CodigoIniciativa"],

ascending=[True, True, False, True],

na_position="last"

else:

gg = gg.sort_values(

by=["DuracionCargaMeses", "MesInicioCargaIdx",
"MesFinCargaIdx", "CodigoIniciativa"],

ascending=[False, True, False, True],

na_position="last"

gg = gg.reset_index(drop=True)

gg["OrdenAsignacion"] = range(1, len(gg) + 1)

def plaza_n(n):

if plazas is None or plazas <= 0:

return None

if regla == "FIJA_1":

return 1

return ((n - 1) % plazas) + 1

gg["PlazaTeoricaN"] = gg["OrdenAsignacion"].apply(plaza_n)

gg["PlazaTeorica"] = gg["PlazaTeoricaN"].apply(

lambda x: None if [Link](x) else f"Plaza {int(x):02d}"

[Link](gg)
 asignacion = [Link](grupos, ignore_index=True) if grupos else
[Link]()

detalle_mes = source[["Empresa", "Rol", "CodigoIniciativa", "Mes",

"MesIdx", "Tag"]].copy()

asignacion = [Link](

detalle_mes,

on=["Empresa", "Rol", "CodigoIniciativa"],

how="left"

asignacion = asignacion.sort_values(

by=["Empresa", "Rol", "PlazaTeoricaN", "OrdenAsignacion",

"MesIdx"],

ascending=[True, True, True, True, True],

na_position="last"

).reset_index(drop=True)

return asignacion

def escribir_salidas_excel(

ruta_salida: Path,

df_detalle_preliminar: [Link],

df_detalle_clasificado: [Link],

df_detalle_calculo: [Link],

df_bolsa_rol_mes: [Link],

df_bolsa_rol_mes_tipologia: [Link],

df_necesidad_mensual: [Link],

df_resumen_rol_empresa: [Link],
 df_resumen_rol_empresa_tipologia: [Link],

df_plazas_requeridas: [Link],

df_asignacion_proyecto_plaza: [Link],

df_matriz_rol_plaza: [Link],

) -> None:

with [Link](ruta_salida, engine="openpyxl") as writer:

df_detalle_preliminar.to_excel(writer,
sheet_name="out_DetallePreliminar", index=False)

df_detalle_clasificado.to_excel(writer,
sheet_name="out_DetalleClasificado", index=False)

df_detalle_calculo.to_excel(writer,
sheet_name="out_DetalleCalculo", index=False)

df_bolsa_rol_mes.to_excel(writer,
sheet_name="out_BolsaRolMes", index=False)

df_bolsa_rol_mes_tipologia.to_excel(writer,
sheet_name="out_BolsaRolMesTipologia", index=False)

df_necesidad_mensual.to_excel(writer,
sheet_name="out_NecesidadMensual", index=False)

df_resumen_rol_empresa.to_excel(writer,
sheet_name="out_ResumenRolEmpresa", index=False)

df_resumen_rol_empresa_tipologia.to_excel(writer,
sheet_name="out_ResumenRolTipologia", index=False)

df_plazas_requeridas.to_excel(writer,
sheet_name="out_PlazasRequeridas", index=False)

df_asignacion_proyecto_plaza.to_excel(writer,
sheet_name="out_AsignacionPlaza", index=False)

df_matriz_rol_plaza.to_excel(writer,
sheet_name="out_MatrizRolPlaza", index=False)

def construir_matriz_rol_plaza(

df_asignacion_proyecto_plaza: [Link],

df_catalogo_meses: [Link],

) -> [Link]:
source = df_asignacion_proyecto_plaza.copy()

meses_ordenados = (

df_catalogo_meses[["Mes", "MesIdx"]]

.drop_duplicates()

.sort_values("MesIdx")

lista_meses = meses_ordenados["Mes"].tolist()

base = source[

"Empresa",

"Rol",

"PlazaTeorica",

"PlazaTeoricaN",

"OrdenAsignacion",

"CodigoIniciativa",

"NombreCorto",

"Tipologia",

"Mes",

"Tag",

].copy()

matriz = base.pivot_table(

index=[

"Empresa",

"Rol",

"PlazaTeorica",
 "PlazaTeoricaN",

"OrdenAsignacion",

"CodigoIniciativa",

"NombreCorto",

"Tipologia",

],

columns="Mes",

values="Tag",

aggfunc="first",

).reset_index()

columnas_fijas = [

"Empresa",

"Rol",

"PlazaTeorica",

"PlazaTeoricaN",

"OrdenAsignacion",

"CodigoIniciativa",

"NombreCorto",

"Tipologia",

columnas_finales = columnas_fijas + [m for m in lista_meses if m in

[Link]]

matriz = matriz[columnas_finales].copy()

matriz = matriz.sort_values(

by=["Empresa", "Rol", "PlazaTeoricaN", "OrdenAsignacion",

"CodigoIniciativa"],
 ascending=[True, True, True, True, True],

na_position="last"

).reset_index(drop=True)

return matriz

#
==========================================
===============

# MAIN

#
==========================================
===============

def main():

print("Inicio de lectura y validación del monolito balanceo PF")

print(f"Archivo: {RUTA_ARCHIVO}")

if not RUTA_ARCHIVO.exists():

print("ERROR: No existe el archivo indicado.")

return

tablas: Dict[str, [Link]] = {}

errores: List[str] = []

# 1. Lectura

for nombre_tabla, columnas_esperadas in

TABLAS_REQUERIDAS.items():

try:

df = obtener_tabla_excel(RUTA_ARCHIVO, nombre_tabla)

tablas[nombre_tabla] = df
 [Link](validar_columnas(nombre_tabla, df,
columnas_esperadas))

except Exception as e:

[Link](f"[{nombre_tabla}] Error al cargar tabla: {e}")

if tablas:

imprimir_resumen_tablas(tablas)

parametros = {}

if "xpu_ParametrosModelo" in tablas:

parametros =
construir_parametros(tablas["xpu_ParametrosModelo"])

print("\nPARAMETROS ACTIVOS")

print("-" * 70)

for k, v in [Link]():

print(f"{k}: {v}")

print("-" * 70)

if all(nombre in tablas for nombre in ["xpu_ProyectoMesTag",

"xpu_CapacidadRolEmpresa", "xpu_DotacionActual",
"xpu_RuteoBalanceRol"]):

[Link](validar_vacios_criticos(tablas))

[Link](validar_duplicados(tablas))

print("\nRESULTADO DE VALIDACIÓN")

print("-" * 70)

if errores:

for err in errores:

print(f"- {err}")

print("-" * 70)
 print("Se detiene la ejecución por errores de estructura.")

return

else:

print("Sin errores de estructura en esta fase.")

print("-" * 70)

# 2. Expansión

df_reglas_expandidas =
expandir_reglas_rol(tablas["xpu_RolReglas"])

df_ruteo_expandido =
expandir_ruteo(tablas["xpu_RuteoBalanceRol"])

print("\nRESUMEN DE EXPANSIÓN")

print("-" * 70)

print(f"Reglas expandidas: {df_reglas_expandidas.shape[0]} filas x

{df_reglas_expandidas.shape[1]} columnas")

print(f"Ruteo expandido : {df_ruteo_expandido.shape[0]} filas x

{df_ruteo_expandido.shape[1]} columnas")

print("-" * 70)

# 3. Candidatos

df_candidatos = generar_candidatos(

df_proyecto_mes_tag=tablas["xpu_ProyectoMesTag"],

df_reglas_expandidas=df_reglas_expandidas,

df_ruteo_expandido=df_ruteo_expandido,

parametros=parametros,

print("\nRESUMEN DE CANDIDATOS")

print("-" * 70)
 print(f"Candidatos generados: {df_candidatos.shape[0]} filas x
{df_candidatos.shape[1]} columnas")

print(f"Roles en candidatos : {df_candidatos['Rol'].nunique()}")

print(f"Empresas en candidatos:
{df_candidatos['Empresa'].nunique()}")

print(f"Meses en candidatos : {df_candidatos['Mes'].nunique()}")

print("-" * 70)

# Control puntual ELCTO / OB / Generación

control_elcto_generacion = df_candidatos[

(df_candidatos["Empresa"] == "4 ELCTO")

& (df_candidatos["Tag"] == "OB")

& (df_candidatos["Tipologia"] == "Generación")

].copy()

print("\nCONTROL RUTEO EXPLICITO ELCTO / OB / GENERACION")

print("-" * 70)

if control_elcto_generacion.empty:

print("Sin candidatos para la combinación.")

else:

print(control_elcto_generacion["Rol"].value_counts(dropna=False).to_s
tring())

print("-" * 70)

# 4. Capacidad y ruteo

df_con_capacidad = agregar_capacidad_y_dotacion(

df_candidatos=df_candidatos,

df_capacidad=tablas["xpu_CapacidadRolEmpresa"],

df_dotacion=tablas["xpu_DotacionActual"],
 parametros=parametros,

print("\nRESUMEN CON CAPACIDAD")

print("-" * 70)

print(f"Filas con capacidad incorporada:

{df_con_capacidad.shape[0]}")

print(f"Filas sin capacidad :

{df_con_capacidad['CapacidadParalelo'].isna().sum()}")

print(f"Roles sin restricción :

{df_con_capacidad.loc[df_con_capacidad['EsSinRestriccion'] == 1,
'Rol'].nunique()}")

print("-" * 70)

df_detalle_preliminar = aplicar_ruteo_base_apoyo(

df_con_capacidad=df_con_capacidad,

df_ruteo_expandido=df_ruteo_expandido,

print("\nRESUMEN DETALLE PRELIMINAR")

print("-" * 70)

print(f"Detalle preliminar generado :

{df_detalle_preliminar.shape[0]} filas x
{df_detalle_preliminar.shape[1]} columnas")

print(f"Roles finales :
{df_detalle_preliminar['Rol'].nunique()}")

print(f"Filas ruteadas :
{int((df_detalle_preliminar['FlagRuteado'] == 1).sum())}")

print(f"Filas no ruteadas :
{int((df_detalle_preliminar['FlagRuteado'] == 0).sum())}")

print(f"Asignadas a apoyo :
{int((df_detalle_preliminar['OrigenAsignacion'] ==
'RuteoApoyo').sum())}")
 print("-" * 70)

# Control barrido de proyectos

control_estudios_elcto_oct26 = df_detalle_preliminar[

(df_detalle_preliminar["Empresa"] == "4 ELCTO")

& (df_detalle_preliminar["RolOriginal"] == "Coord. Estudios")

& (df_detalle_preliminar["Tag"] == "ETO")

& (df_detalle_preliminar["Mes"] == "Oct-26")

].copy()

print("\nCONTROL BARRIDO ITERATIVO ELCTO / COORD. ESTUDIOS /

OCT-26")

print("-" * 70)

if control_estudios_elcto_oct26.empty:

print("Sin filas para la combinación.")

else:

print(

control_estudios_elcto_oct26[

"CodigoIniciativa",

"Tipologia",

"RondaBarrido",

"SecuenciaGrupo",

"RolAsignadoFinal",

"OrigenAsignacion",

.sort_values(["SecuenciaGrupo", "CodigoIniciativa"],
ascending=[True, True])

.head(40)
 .to_string(index=False)

print("-" * 70)

# Control desborde proporcional

control_desborde_elcto = df_detalle_preliminar[

(df_detalle_preliminar["Empresa"] == "4 ELCTO")

& (df_detalle_preliminar["RolOriginal"] == "Coord. Estudios")

& (df_detalle_preliminar["Tag"] == "ETO")

& (df_detalle_preliminar["Mes"] == "Oct-26")

].copy()

print("\nCONTROL DESBORDE PROPORCIONAL ELCTO / COORD.

ESTUDIOS / OCT-26")

print("-" * 70)

if control_desborde_elcto.empty:

print("Sin filas para la combinación.")

else:

print(

control_desborde_elcto[

"CodigoIniciativa",

"Tipologia",

"RolAsignadoFinal",

"OrigenAsignacion",

"CupoBaseTipologia",

"CupoApoyoTipologia",

]
 .sort_values(["Tipologia", "CodigoIniciativa"], ascending=[True,
True])

.head(60)

.to_string(index=False)

print("-" * 70)

# 5. Clasificación de cobertura de capacidad

df_detalle_clasificado, df_detalle_calculo =
construir_detalle_calculo(df_detalle_preliminar)

df_resumen_cobertura =
resumen_cobertura_capacidad(df_detalle_clasificado)

print("\nRESUMEN COBERTURA CAPACIDAD")

print("-" * 70)

for _, row in df_resumen_cobertura.iterrows():

print(f"{row['EstadoCapacidad']:<20}: {row['Filas']}")

print("-" * 70)

print(f"Base válida para cálculo: {df_detalle_calculo.shape[0]}

filas")

print("-" * 70)

# 6. Bolsa rol-mes

df_bolsa_rol_mes = calcular_bolsa_rol_mes(

df_detalle_calculo=df_detalle_calculo,

parametros=parametros,

print("\nRESUMEN BOLSA ROL MES")

print("-" * 70)
 print(f"Bolsa generada : {df_bolsa_rol_mes.shape[0]} filas
x {df_bolsa_rol_mes.shape[1]} columnas")

print(f"Empresas en bolsa :
{df_bolsa_rol_mes['Empresa'].nunique()}")

print(f"Roles en bolsa :
{df_bolsa_rol_mes['Rol'].nunique()}")

print(f"Meses con bolsa :

{df_bolsa_rol_mes['Mes'].nunique()}")

print(f"Filas sin restricción 999 :

{int((df_bolsa_rol_mes['EsSinRestriccion'] == 1).sum())}")

print("-" * 70)

# 7. Bolsa rol-mes-tipología para heatmap corregido

df_bolsa_rol_mes_tipologia = calcular_bolsa_rol_mes_tipologia(

df_detalle_calculo=df_detalle_calculo,

parametros=parametros,

print("\nRESUMEN BOLSA ROL MES TIPOLOGIA")

print("-" * 70)

print(f"Bolsa rol-tipología generada :

{df_bolsa_rol_mes_tipologia.shape[0]} filas x
{df_bolsa_rol_mes_tipologia.shape[1]} columnas")

print(f"Roles en resultado :
{df_bolsa_rol_mes_tipologia['Rol'].nunique()}")

print(f"Tipologías en resultado :
{df_bolsa_rol_mes_tipologia['Tipologia'].nunique()}")

print("-" * 70)

# 8. Necesidad mensual

df_necesidad_mensual = calcular_necesidad_mensual(

df_detalle_calculo=df_detalle_calculo,
 df_bolsa_rol_mes=df_bolsa_rol_mes,

print("\nRESUMEN NECESIDAD MENSUAL")

print("-" * 70)

print(f"Necesidad mensual generada :

{df_necesidad_mensual.shape[0]} filas x
{df_necesidad_mensual.shape[1]} columnas")

print(f"Tipologías en resultado :
{df_necesidad_mensual['Tipologia'].nunique()}")

print(f"Roles en resultado :
{df_necesidad_mensual['Rol'].nunique()}")

print("-" * 70)

# 9. Resumen rol empresa

df_resumen_rol_empresa =
calcular_resumen_rol_empresa(df_bolsa_rol_mes)

print("\nRESUMEN ROL EMPRESA")

print("-" * 70)

print(f"Resumen rol empresa :

{df_resumen_rol_empresa.shape[0]} filas x
{df_resumen_rol_empresa.shape[1]} columnas")

print(f"Roles en resumen :
{df_resumen_rol_empresa['Rol'].nunique()}")

print("-" * 70)

# 10. Resumen rol empresa tipología

df_resumen_rol_empresa_tipologia =
calcular_resumen_rol_empresa_tipologia(df_necesidad_mensual)

print("\nRESUMEN ROL EMPRESA TIPOLOGIA")

 print("-" * 70)

print(f"Resumen rol tipología :

{df_resumen_rol_empresa_tipologia.shape[0]} filas x
{df_resumen_rol_empresa_tipologia.shape[1]} columnas")

print(f"Tipologías en resumen :
{df_resumen_rol_empresa_tipologia['Tipologia'].nunique()}")

print("-" * 70)

# 11. Plazas requeridas

df_plazas_requeridas = calcular_plazas_requeridas(

df_bolsa_rol_mes=df_bolsa_rol_mes,

df_resumen_rol_empresa=df_resumen_rol_empresa,

print("\nRESUMEN PLAZAS REQUERIDAS")

print("-" * 70)

print(f"Plazas requeridas generadas :

{df_plazas_requeridas.shape[0]} filas x
{df_plazas_requeridas.shape[1]} columnas")

print(f"Filas con plazas definidas :

{int(df_plazas_requeridas['PlazasRequeridas'].notna().sum())}")

print("-" * 70)

# 12. Asignación proyecto plaza

df_asignacion_proyecto_plaza =
calcular_asignacion_proyecto_plaza(

df_detalle_calculo=df_detalle_calculo,

df_plazas_requeridas=df_plazas_requeridas,

print("\nRESUMEN ASIGNACION PROYECTO PLAZA")

 print("-" * 70)

print(f"Asignación generada :
{df_asignacion_proyecto_plaza.shape[0]} filas x
{df_asignacion_proyecto_plaza.shape[1]} columnas")

print(f"Plazas teóricas usadas :

{df_asignacion_proyecto_plaza['PlazaTeorica'].nunique(dropna=True)}
")

print("-" * 70)

# 13. Matriz operativa

df_matriz_rol_plaza = construir_matriz_rol_plaza(

df_asignacion_proyecto_plaza=df_asignacion_proyecto_plaza,

df_catalogo_meses=tablas["xpu_CatalogoMeses"],

print("\nRESUMEN MATRIZ ROL PLAZA")

print("-" * 70)

print(f"Matriz generada : {df_matriz_rol_plaza.shape[0]}

filas x {df_matriz_rol_plaza.shape[1]} columnas")

print(f"Empresas en matriz :
{df_matriz_rol_plaza['Empresa'].nunique()}")

print(f"Roles en matriz :
{df_matriz_rol_plaza['Rol'].nunique()}")

print("-" * 70)

exportar_debug(df_matriz_rol_plaza, "debug_matriz_rol_plaza.csv")

# 14. Escritura a Excel

ruta_excel_salida = CARPETA_SALIDA / "Salidas_Monolito_PF.xlsx"

escribir_salidas_excel(
 ruta_salida=ruta_excel_salida,

df_detalle_preliminar=df_detalle_preliminar,

df_detalle_clasificado=df_detalle_clasificado,

df_detalle_calculo=df_detalle_calculo,

df_bolsa_rol_mes=df_bolsa_rol_mes,

df_bolsa_rol_mes_tipologia=df_bolsa_rol_mes_tipologia,

df_necesidad_mensual=df_necesidad_mensual,

df_resumen_rol_empresa=df_resumen_rol_empresa,

df_resumen_rol_empresa_tipologia=df_resumen_rol_empresa_tipologi
a,

df_plazas_requeridas=df_plazas_requeridas,

df_asignacion_proyecto_plaza=df_asignacion_proyecto_plaza,

df_matriz_rol_plaza=df_matriz_rol_plaza,

print("\nSALIDA EXCEL")

print("-" * 70)

print(f"Archivo generado: {ruta_excel_salida}")

print("-" * 70)

# 15. Export debug adicional

exportar_debug(df_resumen_rol_empresa,
"debug_resumen_rol_empresa.csv")

exportar_debug(df_resumen_rol_empresa_tipologia,
"debug_resumen_rol_empresa_tipologia.csv")

exportar_debug(df_plazas_requeridas,
"debug_plazas_requeridas.csv")

exportar_debug(df_asignacion_proyecto_plaza,
"debug_asignacion_proyecto_plaza.csv")
 # 16. Export debug

exportar_debug(df_detalle_clasificado,
"debug_detalle_clasificado.csv")

exportar_debug(df_detalle_calculo, "debug_detalle_calculo.csv")

exportar_debug(df_bolsa_rol_mes, "debug_bolsa_rol_mes.csv")

exportar_debug(df_necesidad_mensual,
"debug_necesidad_mensual.csv")

# Export debug

exportar_debug(df_reglas_expandidas,
"debug_reglas_expandidas.csv")

exportar_debug(df_ruteo_expandido, "debug_ruteo_expandido.csv")

exportar_debug(df_candidatos, "debug_candidatos.csv")

exportar_debug(df_con_capacidad, "debug_con_capacidad.csv")

exportar_debug(df_detalle_preliminar,
"debug_detalle_preliminar.csv")

exportar_debug(df_detalle_clasificado,
"debug_detalle_clasificado.csv")

exportar_debug(df_detalle_calculo, "debug_detalle_calculo.csv")

exportar_debug(df_bolsa_rol_mes, "debug_bolsa_rol_mes.csv")

exportar_debug(df_necesidad_mensual,
"debug_necesidad_mensual.csv")

exportar_debug(df_resumen_rol_empresa,
"debug_resumen_rol_empresa.csv")

exportar_debug(df_resumen_rol_empresa_tipologia,
"debug_resumen_rol_empresa_tipologia.csv")

exportar_debug(df_plazas_requeridas,
"debug_plazas_requeridas.csv")

exportar_debug(df_asignacion_proyecto_plaza,
"debug_asignacion_proyecto_plaza.csv")

exportar_debug(df_bolsa_rol_mes_tipologia,
"debug_bolsa_rol_mes_tipologia.csv")
if __name__ == "__main__":

main()