219 lines
6.2 KiB
Python
219 lines
6.2 KiB
Python
"""Separación de copago / cuota moderadora por año."""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from typing import Any, Literal, Optional
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
from .dates import (
|
||
_detect_compact_date_formats,
|
||
_parse_any_date_to_tuple,
|
||
detect_date_format,
|
||
)
|
||
from .numbers import _infer_decimal_hint_from_column, _parse_lenient_one
|
||
|
||
|
||
DEFAULT_MODERADORA_BY_YEAR: dict[int, set[int]] = {
|
||
2023: {4100, 16400, 43000},
|
||
2024: {4500, 18200, 47700},
|
||
2025: {4700, 19200, 50300},
|
||
2026: {5000, 20100, 52800},
|
||
}
|
||
|
||
|
||
def normalize_moderadora_by_year_dict(raw: Any) -> dict[int, set[int]]:
|
||
"""Normaliza un dict JSON año → lista de enteros (cuotas moderadora)."""
|
||
if not isinstance(raw, dict):
|
||
return {}
|
||
out: dict[int, set[int]] = {}
|
||
for k, v in raw.items():
|
||
try:
|
||
yi = int(k)
|
||
except (TypeError, ValueError):
|
||
continue
|
||
if not isinstance(v, (list, tuple)):
|
||
continue
|
||
nums: set[int] = set()
|
||
for x in v:
|
||
try:
|
||
nums.add(int(round(float(x))))
|
||
except (TypeError, ValueError):
|
||
continue
|
||
if nums:
|
||
out[yi] = nums
|
||
return out
|
||
|
||
|
||
def merge_moderadora_config(
|
||
cop_mod_config: Optional[dict[str, Any]],
|
||
) -> dict[int, set[int]]:
|
||
"""Fusiona defaults con `moderadora_by_year` opcional del usuario."""
|
||
merged = {y: set(vals) for y, vals in DEFAULT_MODERADORA_BY_YEAR.items()}
|
||
if cop_mod_config and isinstance(cop_mod_config.get("moderadora_by_year"), dict):
|
||
custom = normalize_moderadora_by_year_dict(
|
||
cop_mod_config["moderadora_by_year"]
|
||
)
|
||
for y, vals in custom.items():
|
||
merged[y] = set(vals)
|
||
return merged
|
||
|
||
|
||
def _alloc_cop_mod_column_names(
|
||
col_name: str, existing: set[str]
|
||
) -> tuple[str, str]:
|
||
"""Nombres únicos para copago y moderadora."""
|
||
base = re.sub(r"[^\w]+", "_", str(col_name).strip(), flags=re.ASCII) or "col"
|
||
for i in range(0, 200):
|
||
suf = f"_{i}" if i else ""
|
||
ccol, mcol = f"{base}{suf}_copago", f"{base}{suf}_moderadora"
|
||
if ccol not in existing and mcol not in existing:
|
||
return ccol, mcol
|
||
raise ValueError("No se pudieron asignar nombres de columnas únicos.")
|
||
|
||
|
||
def _amount_cell_to_optional_int(
|
||
val: Any,
|
||
decimal_hint: Optional[Literal[",", "."]],
|
||
) -> Optional[int]:
|
||
try:
|
||
f = _parse_lenient_one(val, decimal_hint)
|
||
except Exception:
|
||
return None
|
||
if f is None or (isinstance(f, float) and not np.isfinite(f)) or (
|
||
isinstance(f, (float, np.floating)) and np.isnan(f)
|
||
):
|
||
return None
|
||
try:
|
||
return int(round(float(f)))
|
||
except (TypeError, ValueError, OverflowError):
|
||
return None
|
||
|
||
|
||
def _apply_cop_mod_split(
|
||
result: pd.DataFrame,
|
||
amount_col: str,
|
||
fecha_column: str,
|
||
log: list[str],
|
||
*,
|
||
moderadora_by_year: dict[int, set[int]],
|
||
) -> None:
|
||
"""Añade columnas copago y moderadora según cuotas por año de la fecha."""
|
||
if fecha_column not in result.columns:
|
||
log.append(
|
||
f"[WARN] COP/MOD: columna de fecha '{fecha_column}' no existe; se omite."
|
||
)
|
||
return
|
||
if amount_col not in result.columns:
|
||
log.append(f"[WARN] COP/MOD: columna de importe '{amount_col}' no existe; se omite.")
|
||
return
|
||
|
||
fec_ser = result[fecha_column]
|
||
amt_ser = result[amount_col]
|
||
stripped = fec_ser.astype(str).str.replace(r"[\s\/\-\.]", "", regex=True)
|
||
stripped = stripped.where(fec_ser.notna(), other=pd.NA)
|
||
sep_fmt = detect_date_format(fec_ser) or detect_date_format(stripped)
|
||
compact8_fmt, compact6_fmt = _detect_compact_date_formats(fec_ser)
|
||
amt_hint = _infer_decimal_hint_from_column(amt_ser.astype(str))
|
||
|
||
ccol, mcol = _alloc_cop_mod_column_names(str(amount_col), set(result.columns))
|
||
cop_vals: list[Any] = []
|
||
mod_vals: list[Any] = []
|
||
|
||
for i in range(len(result)):
|
||
amt_raw = amt_ser.iloc[i]
|
||
fec_raw = fec_ser.iloc[i]
|
||
amt_i = _amount_cell_to_optional_int(amt_raw, amt_hint)
|
||
if amt_i is None:
|
||
cop_vals.append(0)
|
||
mod_vals.append(0)
|
||
continue
|
||
|
||
parsed_d = _parse_any_date_to_tuple(
|
||
fec_raw,
|
||
sep_fmt=sep_fmt,
|
||
compact8_fmt=compact8_fmt,
|
||
compact6_fmt=compact6_fmt,
|
||
)
|
||
if parsed_d is None:
|
||
cop_vals.append(amt_i)
|
||
mod_vals.append(0)
|
||
continue
|
||
_d, _m, year = parsed_d
|
||
tiers = moderadora_by_year.get(year)
|
||
if tiers is None:
|
||
cop_vals.append(amt_i)
|
||
mod_vals.append(0)
|
||
continue
|
||
|
||
if amt_i in tiers:
|
||
cop_vals.append(0)
|
||
mod_vals.append(amt_i)
|
||
else:
|
||
cop_vals.append(amt_i)
|
||
mod_vals.append(0)
|
||
|
||
result[ccol] = cop_vals
|
||
result[mcol] = mod_vals
|
||
log.append(
|
||
f"[OK] '{amount_col}' -> '{ccol}', '{mcol}' <- Separa COP/MOD "
|
||
f"(fecha: '{fecha_column}')"
|
||
)
|
||
|
||
|
||
def _alloc_cop_mod_diff_column_name(existing: set[str]) -> str:
|
||
"""Nombre único para la columna resultado (preferido: cop/mod)."""
|
||
preferred = "cop/mod"
|
||
if preferred not in existing:
|
||
return preferred
|
||
for i in range(2, 200):
|
||
name = f"cop/mod_{i}"
|
||
if name not in existing:
|
||
return name
|
||
raise ValueError("No se pudo asignar nombre de columna único para cop/mod.")
|
||
|
||
|
||
def _apply_cop_mod_bruto_neto(
|
||
result: pd.DataFrame,
|
||
bruto_col: str,
|
||
neto_col: str,
|
||
log: list[str],
|
||
) -> None:
|
||
"""Crea columna ``cop/mod`` = valor bruto − valor neto."""
|
||
if bruto_col not in result.columns:
|
||
log.append(
|
||
f"[WARN] COP/MOD (bruto−neto): columna bruto '{bruto_col}' no existe; se omite."
|
||
)
|
||
return
|
||
if neto_col not in result.columns:
|
||
log.append(
|
||
f"[WARN] COP/MOD (bruto−neto): columna neto '{neto_col}' no existe; se omite."
|
||
)
|
||
return
|
||
if bruto_col == neto_col:
|
||
log.append(
|
||
f"[WARN] COP/MOD (bruto−neto): bruto y neto son la misma columna '{bruto_col}'; se omite."
|
||
)
|
||
return
|
||
|
||
bruto_ser = result[bruto_col]
|
||
neto_ser = result[neto_col]
|
||
bruto_hint = _infer_decimal_hint_from_column(bruto_ser.astype(str))
|
||
neto_hint = _infer_decimal_hint_from_column(neto_ser.astype(str))
|
||
|
||
out_col = _alloc_cop_mod_diff_column_name(set(result.columns))
|
||
values: list[Any] = []
|
||
for i in range(len(result)):
|
||
bruto_i = _amount_cell_to_optional_int(bruto_ser.iloc[i], bruto_hint)
|
||
neto_i = _amount_cell_to_optional_int(neto_ser.iloc[i], neto_hint)
|
||
if bruto_i is None or neto_i is None:
|
||
values.append("")
|
||
continue
|
||
values.append(bruto_i - neto_i)
|
||
|
||
result[out_col] = values
|
||
log.append(
|
||
f"[OK] '{out_col}' = '{bruto_col}' − '{neto_col}' <- COP/MOD (bruto − neto)"
|
||
)
|