"""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)" )