IPSExtractor/services/cleaning/domain/csv_cleaning/cop_mod.py

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