diff --git a/application/app_prefs.py b/application/app_prefs.py index a623b1c..f4b78de 100644 --- a/application/app_prefs.py +++ b/application/app_prefs.py @@ -3,16 +3,99 @@ from __future__ import annotations import json import logging +import re from pathlib import Path -from typing import Any, Dict +from typing import Any, Dict, Optional from .config import DEFAULT_ALLOWED_DOCUMENT_TYPES +from .csv_cleaning import DEFAULT_MODERADORA_BY_YEAR, normalize_moderadora_by_year_dict logger = logging.getLogger(__name__) PREFS_FILE = Path(__file__).resolve().parent.parent / "app_prefs.json" +def moderadora_by_year_to_json_dict( + data: dict[int, set[int]], +) -> dict[str, list[int]]: + """Serializa año → conjunto de cuotas a JSON (claves str, listas ordenadas).""" + out: dict[str, list[int]] = {} + for y in sorted(data.keys()): + out[str(y)] = sorted(data[y]) + return out + + +def default_moderadora_by_year_payload() -> dict[str, list[int]]: + """Valores por defecto persistibles para cuota moderadora.""" + return moderadora_by_year_to_json_dict(DEFAULT_MODERADORA_BY_YEAR) + + +def merge_moderadora_prefs(raw: Any) -> dict[str, list[int]]: + """Fusiona prefs guardadas con defaults; devuelve año(str) → lista.""" + base = default_moderadora_by_year_payload() + if not isinstance(raw, dict): + return base + parsed = normalize_moderadora_by_year_dict(raw) + for y, vals in parsed.items(): + base[str(y)] = sorted(vals) + return base + + +def format_moderadora_text_from_prefs(payload: dict[str, list[int]]) -> str: + """Una línea por año: `2024: 4500, 18200, 47700`.""" + lines: list[str] = [] + for ks in sorted(payload.keys(), key=lambda x: int(x)): + nums = payload[ks] + lines.append(f"{ks}: {', '.join(str(n) for n in nums)}") + return "\n".join(lines) + + +def parse_moderadora_text(text: str) -> dict[str, list[int]]: + """Parsea líneas `año: n1, n2` o `año = n1, n2`. Ignora vacías y errores.""" + out: dict[str, list[int]] = {} + for line in (text or "").splitlines(): + line = line.strip() + if not line or line.startswith("#"): + continue + m = re.match(r"^\s*(\d{4})\s*[:=]\s*(.+)$", line) + if not m: + continue + yk = m.group(1) + rest = m.group(2) + nums: list[int] = [] + for part in re.split(r"[,;\s]+", rest): + part = part.strip() + if not part: + continue + try: + nums.append(int(round(float(part)))) + except (TypeError, ValueError): + continue + if nums: + out[yk] = nums + return out + + +def load_cop_mod_preferences() -> tuple[dict[str, list[int]], str]: + """Lee prefs para COP/MOD: mapa año→lista y nombre sugerido columna fecha.""" + prefs = load_app_prefs() + merged = merge_moderadora_prefs(prefs.get("moderadora_by_year")) + fecha_default = str(prefs.get("fecha_servicio_column_default") or "").strip() + return merged, fecha_default + + +def build_cop_mod_config_for_apply( + moderadora_payload: Optional[dict[str, list[int]]] = None, +) -> dict[str, Any]: + """Construye el dict opcional para `csv_cleaning.apply_rules(..., cop_mod_config=)`.""" + if moderadora_payload is None: + prefs = load_app_prefs() + merged = merge_moderadora_prefs(prefs.get("moderadora_by_year")) + else: + merged = merge_moderadora_prefs(moderadora_payload) + return {"moderadora_by_year": merged} + + def load_app_prefs() -> Dict[str, Any]: """Carga preferencias desde disco si existen.""" try: diff --git a/application/batch_processor.py b/application/batch_processor.py index 78c6459..bc59df0 100644 --- a/application/batch_processor.py +++ b/application/batch_processor.py @@ -21,7 +21,7 @@ from application.config import settings from application.app_prefs import get_allowed_document_types from application.tracking_manager import TrackingManager from core.utils import collect_radicado_docs, find_first_matching_file, collect_documents_flat -from core.csv_utils import rows_from_data +from core.csv_utils import merge_verbatim_long_strings, rows_from_data logger = logging.getLogger(__name__) @@ -320,6 +320,7 @@ class BatchProcessor: raw_data, selected_paths_for_type, ) + merge_verbatim_long_strings(raw_data, normalized_obj, selected_paths_for_type) self._accumulate_usage(agg_normalize_usage, n_usage) fused_data[tname] = normalized_obj or {} @@ -508,6 +509,7 @@ class BatchProcessor: raw_data, selected_fields, ) + merge_verbatim_long_strings(raw_data, data, selected_fields) data['radicado'] = entry['radicado'] data["_source_pdf"] = entry["path"] diff --git a/application/config.py b/application/config.py index 59e4622..c2ae9f2 100644 --- a/application/config.py +++ b/application/config.py @@ -28,6 +28,8 @@ class ApiSettings(BaseSettings): api_key: str = os.getenv("EXTRACTOR_API_KEY", "ollama") model: str = os.getenv("EXTRACTOR_MODEL", "gpt-5.4") request_timeout_s: int = int(os.getenv("EXTRACTOR_TIMEOUT_S", 120)) + grok_api_key: str = os.getenv("GROK_API_KEY", "") + grok_base_url: str = os.getenv("GROK_BASE_URL", "https://api.x.ai/v1") model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore") @@ -66,6 +68,21 @@ class AppSettings(BaseSettings): settings = AppSettings() + +def extractor_credentials_for_model(model: str) -> tuple[str, str]: + """Devuelve (api_key, base_url) para el extractor según el id de modelo.""" + m = (model or "").strip() + api = settings.api + if m.lower().startswith("grok-"): + return (str(api.grok_api_key or "").strip(), str(api.grok_base_url or "").strip() or "https://api.x.ai/v1") + return (api.api_key, api.base_url) + + +def is_grok_extractor_model(model: str) -> bool: + """True si el modelo de extracción debe usar la API de xAI (Grok).""" + return (model or "").strip().lower().startswith("grok-") + + # Tipos documentales válidos por defecto para el campo especial `documento_forzate`. # NIT se excluye explícitamente porque no debe tratarse como tipo de documento personal. DEFAULT_ALLOWED_DOCUMENT_TYPES: Tuple[str, ...] = ( diff --git a/application/costs.py b/application/costs.py index 1e115de..579847f 100644 --- a/application/costs.py +++ b/application/costs.py @@ -12,11 +12,6 @@ MODEL_PRICING = { "cached_input": 0.10, "output": 1.60, }, - "gpt-5.2": { - "input": 1.25, - "cached_input": 0.125, - "output": 10.00, - }, "gpt-5.4": { "input": 2.50, "cached_input": 0.25, @@ -32,6 +27,11 @@ MODEL_PRICING = { "cached_input": 0.02, "output": 0.80, }, + "grok-4.3": { + "input": 1.25, + "cached_input": 0.0, + "output": 2.50, + }, } # Legacy mapping o fase por defecto (si no se especifica modelo) diff --git a/application/csv_cleaning.py b/application/csv_cleaning.py index 2bf4e8e..8371708 100644 --- a/application/csv_cleaning.py +++ b/application/csv_cleaning.py @@ -26,6 +26,157 @@ import pandas as pd logger = logging.getLogger(__name__) +# --------------------------------------------------------------------------- +# CUOTA MODERADORA (COP/MOD) +# --------------------------------------------------------------------------- + +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}')" + ) + + # --------------------------------------------------------------------------- # MODELO DE DATOS # --------------------------------------------------------------------------- @@ -166,7 +317,8 @@ def _split_date_time(value: str) -> tuple[str, Optional[str]]: if ampm and not (1 <= hh <= 12): return s, None - date_part = s[: m.start()].rstrip(" T") + # Quitar separadores habituales entre fecha y hora (p. ej. "12/27/24, 12:00 AM"). + date_part = s[: m.start()].rstrip(" \t,;T") if not date_part: return s, None @@ -997,19 +1149,21 @@ def _applicable_rules_for_type(inferred_type: str) -> list[str]: """Identificadores de reglas a mostrar según el tipo inferido (sin redundancias).""" base = ["strip_special_spaces", "strip_special_keep_date_seps", "strip_spaces"] if inferred_type == TYPE_INTEGER: - return base + return base + ["cop_mod_split"] if inferred_type == TYPE_DECIMAL: return base + [ "to_integer", "round_decimals_custom", "doc_id_split_tipo_numero", "doc_type_homologate", + "cop_mod_split", ] if inferred_type == TYPE_DATE: return base + [ "doc_id_split_tipo_numero", "doc_type_homologate", "date_to_ddmmyyyy", + "nota_paciente_autorizacion", ] if inferred_type == TYPE_MIXED: return base + [ @@ -1018,6 +1172,8 @@ def _applicable_rules_for_type(inferred_type: str) -> list[str]: "doc_id_split_tipo_numero", "doc_type_homologate", "date_to_ddmmyyyy", + "cop_mod_split", + "nota_paciente_autorizacion", ] # Texto / categórico / boolean / empty → todas las de limpieza de texto + las # conversoras (el usuario sabrá si aplica). @@ -1027,6 +1183,8 @@ def _applicable_rules_for_type(inferred_type: str) -> list[str]: "doc_id_split_tipo_numero", "doc_type_homologate", "date_to_ddmmyyyy", + "cop_mod_split", + "nota_paciente_autorizacion", ] @@ -1576,6 +1734,74 @@ def _apply_doc_id_split_columns( ) +# --- Texto tipo nota (Paciente / Autorización) -------------------------------- + +_PACIENTE_DOC_NOTA_RE = re.compile(r"(?i)Paciente:\s*(\d+)\s*-\s*") +_AUTORIZACION_NOTA_RE = re.compile(r"(?i)Autorizaci[oó]n\s*No\.?\s*(\d+)") + + +def _normalize_nota_text(raw: Any) -> str: + if _is_nullish_scalar(raw): + return "" + try: + if pd.isna(raw): + return "" + except TypeError: + pass + s = str(raw).strip() + if not s or s.lower() in ("nan", "none", "nat", ""): + return "" + return re.sub(r"\s+", " ", s) + + +def extract_paciente_documento_nota(raw: Any) -> Optional[str]: + """Número de documento tras ``Paciente:`` y antes del guión del nombre.""" + t = _normalize_nota_text(raw) + if not t: + return None + m = _PACIENTE_DOC_NOTA_RE.search(t) + return m.group(1) if m else None + + +def extract_autorizacion_nota(raw: Any) -> Optional[str]: + """Número tras ``Autorización No.`` (opcional por fila).""" + t = _normalize_nota_text(raw) + if not t: + return None + m = _AUTORIZACION_NOTA_RE.search(t) + return m.group(1) if m else None + + +def _alloc_nota_extract_column_names( + col_name: str, existing: set[str] +) -> tuple[str, str]: + """Nombres únicos para documento paciente y autorización.""" + 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 "" + dcol, acol = f"{base}{suf}_nota_paciente_doc", f"{base}{suf}_nota_autorizacion" + if dcol not in existing and acol not in existing: + return dcol, acol + raise ValueError("No se pudieron asignar nombres de columnas únicos.") + + +def _apply_nota_extract_columns(result: pd.DataFrame, col: str, log: list[str]) -> None: + """Añade columnas con documento (Paciente:) y número de autorización si existe.""" + dcol, acol = _alloc_nota_extract_column_names(str(col), set(result.columns)) + docs: list[Any] = [] + auths: list[Any] = [] + for v in result[col].tolist(): + dd = extract_paciente_documento_nota(v) + aa = extract_autorizacion_nota(v) + docs.append(dd if dd is not None else pd.NA) + auths.append(aa if aa is not None else pd.NA) + result[dcol] = docs + result[acol] = auths + log.append( + f"[OK] '{str(col)}' -> columnas '{dcol}' y '{acol}' <- Extraer datos de notas" + ) + + # Registro de reglas CLEANING_RULES: list[dict[str, Any]] = [ { @@ -1661,6 +1887,26 @@ CLEANING_RULES: list[dict[str, Any]] = [ ), "fn": _rule_doc_type_homologate, }, + { + "id": "cop_mod_split", + "name": "Separa COP/MOD (copago / moderadora)", + "description": ( + "Según el año en la columna Fecha Servicio que elija en la tarjeta, clasifica " + "el importe en cuota moderadora (valores por año en Configuración) o copago. " + "Se ejecuta al final del proceso de limpieza para usar fechas ya normalizadas." + ), + "fn": _rule_passthrough_series, + }, + { + "id": "nota_paciente_autorizacion", + "name": "Extraer datos de notas (documento y autorización)", + "description": ( + "Desde texto tipo nota hospitalaria, extrae el número tras «Paciente:» (antes del " + "guión del nombre) y, si existe, el de «Autorización No.». Añade dos columnas; " + "no modifica el texto original." + ), + "fn": _rule_passthrough_series, + }, ] _RULE_BY_ID: dict[str, Callable[[pd.Series], pd.Series]] = { @@ -1686,21 +1932,30 @@ def get_rule(rule_id: str) -> Optional[dict[str, Any]]: def apply_rules( df: pd.DataFrame, rules_per_column: dict[str, list[str]], + *, + cop_mod_fecha_column: Optional[dict[str, str]] = None, + cop_mod_config: Optional[dict[str, Any]] = None, ) -> tuple[pd.DataFrame, list[str]]: """Aplica las reglas seleccionadas a cada columna. `rules_per_column` es un dict de `columna -> lista de IDs de reglas en el - orden a aplicar`. Devuelve el DataFrame resultante y un log. + orden a aplicar`. La regla `cop_mod_split` se difiere a una segunda pasada + tras el resto, para que la columna de fecha pueda normalizarse antes. + + `cop_mod_fecha_column`: mapa columna de importe -> nombre de columna Fecha Servicio. + `cop_mod_config`: opcional; puede incluir ``moderadora_by_year`` (dict año → lista). """ result = df.copy() log: list[str] = [] - for col, rule_ids in rules_per_column.items(): + moderadora_years = merge_moderadora_config(cop_mod_config) + + def _run_rules_for_column(col: str, rule_ids: list[str]) -> None: if not rule_ids: - continue + return if col not in result.columns: log.append(f"[WARN] Columna no encontrada: {col}") - continue + return split_has_homologation = ( "doc_id_split_tipo_numero" in rule_ids and "doc_type_homologate" in rule_ids ) @@ -1714,6 +1969,13 @@ def apply_rules( logger.exception("Error aplicando regla %s a %s", rule_id, col) log.append(f"[ERROR] '{col}' regla '{rule_id}': {exc}") continue + if rule_id == "nota_paciente_autorizacion": + try: + _apply_nota_extract_columns(result, col, log) + except Exception as exc: # noqa: BLE001 + logger.exception("Error aplicando regla %s a %s", rule_id, col) + log.append(f"[ERROR] '{col}' regla '{rule_id}': {exc}") + continue if rule_id == "doc_type_homologate" and split_has_homologation: continue fn = _RULE_BY_ID.get(rule_id) @@ -1727,6 +1989,34 @@ def apply_rules( logger.exception("Error aplicando regla %s a %s", rule_id, col) log.append(f"[ERROR] '{col}' regla '{rule_id}': {exc}") + # Fase 1: todas las reglas excepto cop_mod_split + for col, rule_ids in rules_per_column.items(): + filtered = [r for r in rule_ids if r != "cop_mod_split"] + _run_rules_for_column(col, filtered) + + # Fase 2: Separa COP/MOD sobre el DataFrame ya transformado + fecha_map = cop_mod_fecha_column or {} + for col, rule_ids in rules_per_column.items(): + if "cop_mod_split" not in rule_ids: + continue + fecha_col = fecha_map.get(col) + if not fecha_col: + log.append( + f"[WARN] COP/MOD en '{col}': indique la columna «Fecha Servicio» en la tarjeta; se omite." + ) + continue + try: + _apply_cop_mod_split( + result, + col, + fecha_col, + log, + moderadora_by_year=moderadora_years, + ) + except Exception as exc: # noqa: BLE001 + logger.exception("Error aplicando COP/MOD a %s", col) + log.append(f"[ERROR] '{col}' regla 'cop_mod_split': {exc}") + return result, log diff --git a/application/csv_manager.py b/application/csv_manager.py index 553dcb8..d7ec644 100644 --- a/application/csv_manager.py +++ b/application/csv_manager.py @@ -43,7 +43,7 @@ class CSVManager: file_exists = os.path.exists(csv_path) and os.path.getsize(csv_path) > 0 with self._lock: - with open(csv_path, "a", encoding="utf-8", newline="") as f: + with open(csv_path, "a", encoding="utf-8-sig", newline="") as f: writer = csv.DictWriter(f, fieldnames=header, extrasaction="ignore", delimiter="|") if not file_exists: writer.writeheader() diff --git a/application/llm_client.py b/application/llm_client.py index 251eae5..63b8692 100644 --- a/application/llm_client.py +++ b/application/llm_client.py @@ -223,6 +223,7 @@ class LLMClient: response_format: str = "json_object", timeout_s: Optional[float] = None, component: str = "llm_client", + max_completion_tokens: Optional[int] = None, ) -> Tuple[Dict[str, Any], Optional[Dict[str, Any]]]: """Devuelve (objeto JSON parseado, usage).""" import json @@ -231,11 +232,16 @@ class LLMClient: self._log_start(component=component, model=eff_model, timeout_s=eff_timeout, messages_count=len(messages or [])) start = time.time() try: + completion_kwargs: Dict[str, Any] = { + "response_format": {"type": response_format}, + } + if max_completion_tokens is not None: + completion_kwargs["max_completion_tokens"] = max_completion_tokens chat_completion = self._create_completion( messages=messages, model=eff_model, timeout_s=eff_timeout, - response_format={"type": response_format}, + **completion_kwargs, ) elapsed_ms = int((time.time() - start) * 1000) usage = self._extract_usage(chat_completion) @@ -256,6 +262,7 @@ class LLMClient: response_format: str = "json_object", timeout_s: Optional[float] = None, component: str = "llm_client", + max_completion_tokens: Optional[int] = None, ) -> Tuple[Dict[str, Any], Optional[Dict[str, Any]]]: """Devuelve (objeto JSON parseado, usage) de forma asíncrona.""" import json @@ -264,11 +271,16 @@ class LLMClient: self._log_start(component=component, model=eff_model, timeout_s=eff_timeout, messages_count=len(messages or [])) start = time.time() try: + completion_kwargs: Dict[str, Any] = { + "response_format": {"type": response_format}, + } + if max_completion_tokens is not None: + completion_kwargs["max_completion_tokens"] = max_completion_tokens chat_completion = await self._create_completion_async( messages=messages, model=eff_model, timeout_s=eff_timeout, - response_format={"type": response_format}, + **completion_kwargs, ) elapsed_ms = int((time.time() - start) * 1000) usage = self._extract_usage(chat_completion) diff --git a/application/normalizer.py b/application/normalizer.py index 3a25627..6f7ab8c 100644 --- a/application/normalizer.py +++ b/application/normalizer.py @@ -123,6 +123,7 @@ def normalize_to_schema( response_format="json_object", timeout_s=float(eff_timeout), component="normalizer", + max_completion_tokens=16384, ) except Exception: normalized_obj = None diff --git a/application/prompts.py b/application/prompts.py index d333b66..04a810e 100644 --- a/application/prompts.py +++ b/application/prompts.py @@ -75,7 +75,9 @@ class PromptBuilder: "Eres un asistente que normaliza un JSON a un ESQUEMA CANÓNICO dado. " "Debes devolver EXCLUSIVAMENTE un JSON que respete exactamente la estructura y nombres del 'template'. " "Reglas: (1) No inventes claves; (2) No uses sinónimos ni variantes; (3) Respeta listas (si 'items' es lista, devuelve un array de objetos); " - "(4) Si falta un valor, usa null o [] según corresponda; (5) No incluyas texto extra ni comentarios." + "(4) Si falta un valor, usa null o [] según corresponda; (5) No incluyas texto extra ni comentarios. " + "(6) Los campos de texto largo (notas, observaciones, leyendas, pie de página, etc.): copia el texto COMPLETO " + "del source_json sin resumir, sin omitir tramos ni acortar; respeta tildes y caracteres tal cual en el valor de origen." ) user_obj = { "perfil": profile_name, @@ -83,7 +85,9 @@ class PromptBuilder: "source_json": source_json, "instrucciones": ( "Mapea los valores del 'source_json' al 'template'. Si hay claves equivalentes en el source, " - "elige siempre los nombres del template. No devuelvas rutas con '/', sino objetos anidados." + "elige siempre los nombres del template. No devuelvas rutas con '/', sino objetos anidados. " + "Para cualquier cadena larga del documento (p. ej. notas legales o bloques de observación), " + "transcribe el contenido literal completo; no uses elipsis ni versiones abreviadas." ), } return [ diff --git a/core/csv_utils.py b/core/csv_utils.py index 6c048b4..3b8c9f4 100644 --- a/core/csv_utils.py +++ b/core/csv_utils.py @@ -69,6 +69,89 @@ def _get_at_path(obj: Any, path: str) -> Any: return cur +def set_at_path(root: Dict[str, Any], path: str, value: Any) -> None: + """Asigna ``value`` en ``root`` siguiendo la ruta con segmentos ``/`` (crea dicts intermedios).""" + parts = [p for p in path.split('/') if p] + if not parts: + return + cur: Any = root + for p in parts[:-1]: + if not isinstance(cur, dict): + return + nxt = cur.get(p) + if not isinstance(nxt, dict): + cur[p] = {} + nxt = cur[p] + cur = nxt + if isinstance(cur, dict): + cur[parts[-1]] = value + + +def _resolve_raw_string_for_scalar_path(raw: Dict[str, Any], path: str) -> tuple[Optional[str], bool]: + """Resuelve un string en ``raw`` para la ruta canónica ``path``. + + Devuelve ``(valor, ambiguous)``. Si ``ambiguous`` es True, no debe usarse ningún valor + (varias claves aplanadas comparten el último segmento del path). + """ + direct = _get_at_path(raw, path) + if isinstance(direct, str) and direct.strip(): + return direct, False + + parts = [p for p in path.split('/') if p] + if not parts: + return None, False + last = parts[-1] + raw_flat = flatten_non_list_fields(raw) + same_last: List[tuple[str, str]] = [] + for k, v in raw_flat.items(): + if not isinstance(v, str) or not v.strip(): + continue + segs = [s for s in k.split('/') if s] + if not segs: + continue + if segs[-1] == last: + same_last.append((k, v)) + if len(same_last) == 1: + return same_last[0][1], False + if len(same_last) > 1: + return None, True + return None, False + + +def merge_verbatim_long_strings( + raw: Dict[str, Any], + normalized: Dict[str, Any], + paths: Optional[List[str]], +) -> Dict[str, Any]: + """Sustituye en ``normalized`` cadenas truncadas por la versión más larga del ``raw`` (misma ruta o sufijo único). + + Solo actúa sobre rutas escalares (sin ``*``). Evita sustituciones cuando hay varias claves + candidatas con el mismo último segmento o cuando el texto normalizado no parece un prefijo + del crudo (salvo cadena vacía o solo espacios). + """ + if not isinstance(normalized, dict) or not isinstance(raw, dict): + return normalized + for path in paths or []: + if not path or '*' in path: + continue + norm_val = _get_at_path(normalized, path) + if not isinstance(norm_val, str): + continue + raw_val, ambiguous = _resolve_raw_string_for_scalar_path(raw, path) + if ambiguous or raw_val is None: + continue + if len(raw_val) <= len(norm_val): + continue + nt = norm_val.strip() + if nt: + if nt not in raw_val: + lim = min(80, len(nt)) + if lim == 0 or not raw_val.startswith(nt[:lim]): + continue + set_at_path(normalized, path, raw_val) + return normalized + + def aggregate_list_fields(data: Dict[str, Any], aggregate_paths: List[str]) -> Dict[str, Any]: aggregated: Dict[str, Any] = {} for base_path in aggregate_paths: diff --git a/gui/controller.py b/gui/controller.py index 76741ec..658bdbf 100644 --- a/gui/controller.py +++ b/gui/controller.py @@ -13,12 +13,19 @@ import os from core.utils import parse_comma_separated, collect_radicado_docs, collect_documents_flat from application.logging_config import configure_logging -from application.app_prefs import load_app_prefs, save_app_prefs, normalize_document_types +from application.app_prefs import ( + build_cop_mod_config_for_apply, + load_app_prefs, + merge_moderadora_prefs, + normalize_document_types, + parse_moderadora_text, + save_app_prefs, +) from gui.field_selector import FieldSelectorDialog from application.preclassification import PreclassificationService from application.strategies import OllamaExtractor from application.database import DatabaseManager -from application.config import settings +from application.config import settings, extractor_credentials_for_model, is_grok_extractor_model from application.costs import MODEL_PRICING from application import postprocess_agent from application import csv_cleaning @@ -28,6 +35,21 @@ import pandas as pd logger = logging.getLogger(__name__) +# Etiquetas antiguas del combo → etiquetas actuales (prefs / estado previo) +_LEGACY_SETTINGS_MODEL_LABEL = { + "5.4": "gpt-5.4", + "5.4-mini": "gpt-5.4-mini", + "5.4-nano": "gpt-5.4-nano", + "5.2": "gpt-5.4", + "gpt-5.2": "gpt-5.4", + "4.1-mini": "gpt-4.1-mini", +} + + +def _canonical_settings_model_label(name: str) -> str: + s = (name or "").strip() + return _LEGACY_SETTINGS_MODEL_LABEL.get(s, s) + class AppController: """Controlador principal de la GUI que orquesta la vista y el servicio. @@ -123,11 +145,11 @@ class AppController: # Mapeo de nombres de UI a IDs de modelo (y viceversa) self.ui_to_model = { - "5.4": "gpt-5.4", - "5.4-mini": "gpt-5.4-mini", - "5.4-nano": "gpt-5.4-nano", - "5.2": "gpt-5.2", - "4.1-mini": "gpt-4.1-mini", + "gpt-5.4": "gpt-5.4", + "gpt-5.4-mini": "gpt-5.4-mini", + "gpt-5.4-nano": "gpt-5.4-nano", + "gpt-4.1-mini": "gpt-4.1-mini", + "grok-4.3": "grok-4.3", } self.model_to_ui = {v: k for k, v in self.ui_to_model.items()} @@ -168,6 +190,25 @@ class AppController: results.sort(key=lambda x: x["input"], reverse=True) return results + def _apply_extractor_model(self, real_model_id: str, *, interactive: bool = False) -> bool: + """Aplica modelo y credenciales del extractor (Grok usa GROK_API_KEY / GROK_BASE_URL).""" + if not hasattr(self.service.extractor, "update_config"): + return True + if is_grok_extractor_model(real_model_id): + api_key, base_url = extractor_credentials_for_model(real_model_id) + if not api_key: + msg = "Define GROK_API_KEY en el entorno o en .env para usar modelos Grok." + if interactive: + mb.showwarning("Configuración", msg) + else: + logger.warning("Modelo persistido %s no aplicado: %s", real_model_id, msg) + return False + self.service.extractor.update_config(api_key=api_key, base_url=base_url, model=real_model_id) + return True + api_key, base_url = extractor_credentials_for_model(real_model_id) + self.service.extractor.update_config(api_key=api_key, base_url=base_url, model=real_model_id) + return True + def on_save_settings(self): """Actualiza la configuración del servicio (modelo LLM y concurrencia).""" selected_name = self.view.get_selected_model() @@ -182,19 +223,19 @@ class AppController: return # Traducir nombre UI -> ID modelo real - real_model_id = self.ui_to_model.get(selected_name, selected_name) + canonical_ui = _canonical_settings_model_label(selected_name) + real_model_id = self.ui_to_model.get(canonical_ui, canonical_ui) try: - # Actualizar extractor - if hasattr(self.service.extractor, "update_config"): - self.service.extractor.update_config(model=real_model_id) - + if not self._apply_extractor_model(real_model_id, interactive=True): + return + # Actualizar settings globales settings.processing.concurrency_limit = concurrency self.view.set_allowed_document_types_hint(self.view.get_allowed_document_types()) self._save_preferences() - mb.showinfo("Configuración", f"Configuración actualizada.\nModelo: {selected_name}\nConcurrencia: {concurrency}") + mb.showinfo("Configuración", f"Configuración actualizada.\nModelo: {canonical_ui}\nConcurrencia: {concurrency}") self.view.append_output_safe(f"Configuración actualizada: Modelo={real_model_id}, Concurrencia={concurrency}") except Exception as e: logger.exception("Error actualizando configuración") @@ -514,6 +555,8 @@ class AppController: "root_path": self.view.get_root_path(), "csv_path": self.view.get_csv_path(), "doc_types": doc_types, + "moderadora_by_year": parse_moderadora_text(self.view.get_moderadora_cuotas_text()), + "fecha_servicio_column_default": self.view.get_fecha_servicio_column_hint(), } def _save_preferences(self) -> None: @@ -529,10 +572,10 @@ class AppController: try: selected_model = str(prefs.get("selected_model") or "").strip() if selected_model: - self.view.set_selected_model(selected_model) - real_model_id = self.ui_to_model.get(selected_model, selected_model) - if hasattr(self.service.extractor, "update_config"): - self.service.extractor.update_config(model=real_model_id) + canonical_ui = _canonical_settings_model_label(selected_model) + self.view.set_selected_model(canonical_ui) + real_model_id = self.ui_to_model.get(canonical_ui, canonical_ui) + self._apply_extractor_model(real_model_id, interactive=False) except Exception: logger.exception("No se pudo restaurar el modelo persistido") @@ -581,6 +624,15 @@ class AppController: except Exception: pass + try: + merged = merge_moderadora_prefs(prefs.get("moderadora_by_year")) + self.view.set_moderadora_cuotas_ui( + merged, + str(prefs.get("fecha_servicio_column_default") or ""), + ) + except Exception: + pass + def on_app_close(self): """Guarda preferencias antes de cerrar la ventana.""" try: @@ -1246,8 +1298,8 @@ class AppController: pass def _finish_cleaning_csv_load(self, df: pd.DataFrame, summaries): - self.view.populate_cleaning_columns(summaries) self.view.set_cleaning_preview(df) + self.view.populate_cleaning_columns(summaries) self.view.set_cleaning_status( f"CSV cargado: {len(df):,} filas x {len(df.columns)} columnas." ) @@ -1281,6 +1333,15 @@ class AppController: ) return + fecha_map = self.view.get_cleaning_cop_mod_fecha_columns() + for col_name, rule_ids in rules.items(): + if "cop_mod_split" in rule_ids and col_name not in fecha_map: + mb.showwarning( + "COP/MOD", + f"Marque «Separa COP/MOD» y elija la columna «Fecha Servicio» para «{col_name}».", + ) + return + self.view.enable_cleaning_controls(False) self.view.enable_cleaning_save(False) self.view.set_cleaning_status("Aplicando reglas...") @@ -1290,7 +1351,14 @@ class AppController: def worker(): try: - result_df, log = csv_cleaning.apply_rules(df, rules) + cop_cfg = build_cop_mod_config_for_apply() + fecha_map = self.view.get_cleaning_cop_mod_fecha_columns() + result_df, log = csv_cleaning.apply_rules( + df, + rules, + cop_mod_fecha_column=fecha_map or None, + cop_mod_config=cop_cfg, + ) self._cleaning_result_df = result_df ok_count = sum(1 for line in log if line.startswith("[OK]")) err_count = sum(1 for line in log if line.startswith("[ERROR]")) diff --git a/gui/tabs/cleaning_tab.py b/gui/tabs/cleaning_tab.py index 30cfcf6..c7e2684 100644 --- a/gui/tabs/cleaning_tab.py +++ b/gui/tabs/cleaning_tab.py @@ -17,6 +17,7 @@ from application.csv_cleaning import ( ColumnSummary, applicable_rule_ids_for_type, ) +from application.app_prefs import load_cop_mod_preferences from gui.styles import AppStyles @@ -67,6 +68,8 @@ _RULE_SHORT_LABELS: dict[str, str] = { "round_decimals_custom": "Aprox. decimales (>0.5 sube)", "doc_id_split_tipo_numero": "Dividir doc (tipo + cifras)", "doc_type_homologate": "Homologar tipo doc", + "cop_mod_split": "Separa COP/MOD", + "nota_paciente_autorizacion": "Extraer doc. y autorización (notas)", } @@ -96,6 +99,9 @@ class _ColumnUI: rule_vars: list[tuple[str, ctk.CTkCheckBox, ctk.BooleanVar]] summary: ColumnSummary expanded: bool = True + cop_mod_fecha_frame: Optional[ctk.CTkFrame] = None + cop_mod_fecha_menu: Optional[ctk.CTkOptionMenu] = None + cop_mod_fecha_var: Optional[ctk.StringVar] = None class CleaningTab: @@ -109,6 +115,7 @@ class CleaningTab: self._column_uis: dict[str, _ColumnUI] = {} self._search_var: Optional[ctk.StringVar] = None + self._csv_column_names: list[str] = [] # Tipo de dato forzado por el usuario (id interno) por columna; vacío = inferido self._type_overrides: dict[str, str] = {} self._syncing_rule_vars = False @@ -473,6 +480,7 @@ class CleaningTab: def set_preview(self, df: Optional[pd.DataFrame]): """Renderiza el DataFrame original en la tabla de vista previa.""" if df is None or len(df) == 0: + self._csv_column_names = [] self._render_table_placeholder( self.preview_table_holder, "(sin datos)", ) @@ -489,6 +497,7 @@ class CleaningTab: ) self.chip_rows.configure(text=f"Filas: {len(df):,}") self.chip_cols.configure(text=f"Columnas: {len(df.columns)}") + self._csv_column_names = list(df.columns) def set_result(self, df: Optional[pd.DataFrame], info_text: str = ""): """Renderiza el DataFrame resultante en la tabla de resultado.""" @@ -574,6 +583,56 @@ class CleaningTab: self.chip_cols.configure(text=f"Columnas: {len(summaries)}") self._update_global_count() + self._refresh_cop_mod_menus() + + def get_cop_mod_fecha_columns(self) -> dict[str, str]: + """Mapa columna de importe -> columna Fecha Servicio (solo si COP/MOD está marcado).""" + out: dict[str, str] = {} + for col_name, ui in self._column_uis.items(): + if ui.cop_mod_fecha_var is None: + continue + cop_on = any( + rid == "cop_mod_split" and bool(var.get()) + for rid, _chk, var in ui.rule_vars + ) + if not cop_on: + continue + sel = str(ui.cop_mod_fecha_var.get() or "").strip() + if sel: + out[col_name] = sel + return out + + def _date_columns_for_cop_mod_menu(self) -> list[str]: + cols = self._csv_column_names or [] + return cols if cols else [""] + + def _suggest_fecha_column_option(self) -> str: + cols = self._csv_column_names or [] + if not cols: + return "" + _, hint = load_cop_mod_preferences() + if hint and hint in cols: + return hint + for c in cols: + low = c.lower().replace("_", " ") + if "fecha" in low and "servicio" in low: + return c + return cols[0] + + def _refresh_cop_mod_menus(self) -> None: + cols = self._date_columns_for_cop_mod_menu() + suggest = self._suggest_fecha_column_option() + for ui in self._column_uis.values(): + if ui.cop_mod_fecha_menu is None or ui.cop_mod_fecha_var is None: + continue + cur = ui.cop_mod_fecha_var.get() + ui.cop_mod_fecha_menu.configure(values=cols) + if cur and cur in cols: + ui.cop_mod_fecha_var.set(cur) + elif suggest and suggest in cols: + ui.cop_mod_fecha_var.set(suggest) + elif cols and cols[0]: + ui.cop_mod_fecha_var.set(cols[0]) def get_selected_rules(self) -> dict[str, list[str]]: """Devuelve un mapa columna -> lista de rule_ids en el orden visual.""" @@ -700,6 +759,30 @@ class CleaningTab: col_in_row = 0 row_in_rules += 1 + fec_frame = None + fec_menu = None + fec_var = None + if "cop_mod_split" in ordered_ids: + fec_frame = ctk.CTkFrame(body, fg_color="transparent") + fec_frame.grid(row=4, column=0, sticky="ew", padx=4, pady=(4, 0)) + fec_frame.grid_columnconfigure(1, weight=1) + ctk.CTkLabel( + fec_frame, + text="Fecha Servicio (columna):", + font=self.fonts["italic_small"], + text_color=self.colors["gray_dark"], + ).grid(row=0, column=0, sticky="w", padx=(0, 8)) + fec_var = ctk.StringVar(value=self._suggest_fecha_column_option()) + fec_menu = ctk.CTkOptionMenu( + fec_frame, + values=self._date_columns_for_cop_mod_menu(), + variable=fec_var, + font=self.fonts["ui"], + width=260, + ) + fec_menu.grid(row=0, column=1, sticky="ew") + fec_menu.configure(state="disabled") + ui = _ColumnUI( card=card, body=body, @@ -710,6 +793,9 @@ class CleaningTab: rules_frame=rules_frame, rule_vars=vars_list, summary=summary, + cop_mod_fecha_frame=fec_frame, + cop_mod_fecha_menu=fec_menu, + cop_mod_fecha_var=fec_var, ) toggle_btn.configure(command=lambda c=summary.name: self._toggle_column(c)) return ui @@ -910,6 +996,39 @@ class CleaningTab: col_in_row = 0 row_in_rules += 1 ui.rule_vars = new_vars + if ui.cop_mod_fecha_frame is not None: + try: + ui.cop_mod_fecha_frame.destroy() + except Exception: + pass + ui.cop_mod_fecha_frame = None + ui.cop_mod_fecha_menu = None + ui.cop_mod_fecha_var = None + ordered_after = [r["id"] for r in CLEANING_RULES if r["id"] in applic] + if "cop_mod_split" in ordered_after: + fec_frame = ctk.CTkFrame(ui.body, fg_color="transparent") + fec_frame.grid(row=4, column=0, sticky="ew", padx=4, pady=(4, 0)) + fec_frame.grid_columnconfigure(1, weight=1) + ctk.CTkLabel( + fec_frame, + text="Fecha Servicio (columna):", + font=self.fonts["italic_small"], + text_color=self.colors["gray_dark"], + ).grid(row=0, column=0, sticky="w", padx=(0, 8)) + fec_var = ctk.StringVar(value=self._suggest_fecha_column_option()) + fec_menu = ctk.CTkOptionMenu( + fec_frame, + values=self._date_columns_for_cop_mod_menu(), + variable=fec_var, + font=self.fonts["ui"], + width=260, + ) + fec_menu.grid(row=0, column=1, sticky="ew") + fec_menu.configure(state="disabled") + ui.cop_mod_fecha_frame = fec_frame + ui.cop_mod_fecha_menu = fec_menu + ui.cop_mod_fecha_var = fec_var + self._refresh_cop_mod_menus() self.set_status(f"Columna «{col_name}»: tipo «{_TYPE_LABELS.get(eff, eff)}».") self._on_rule_toggle(col_name) @@ -954,6 +1073,12 @@ class CleaningTab: f"Columna «{col_name}»: se activó homologación de tipo de documento." ) break + if ui.cop_mod_fecha_menu is not None: + cop_on = any( + rid == "cop_mod_split" and bool(var.get()) + for rid, _chk, var in ui.rule_vars + ) + ui.cop_mod_fecha_menu.configure(state="normal" if cop_on else "disabled") selected = sum(1 for _rid, _chk, var in ui.rule_vars if bool(var.get())) text = f"{selected} regla{'s' if selected != 1 else ''}" color = self.colors["success"] if selected else self.colors["gray_light"] diff --git a/gui/tabs/settings_tab.py b/gui/tabs/settings_tab.py index e62cb79..4ff24e6 100644 --- a/gui/tabs/settings_tab.py +++ b/gui/tabs/settings_tab.py @@ -1,7 +1,14 @@ """Pestaña de Configuración de la aplicación GUI.""" import customtkinter as ctk from gui.styles import AppStyles -from application.app_prefs import get_allowed_document_types, normalize_document_types +from application.app_prefs import ( + format_moderadora_text_from_prefs, + get_allowed_document_types, + load_cop_mod_preferences, + normalize_document_types, + parse_moderadora_text, + default_moderadora_by_year_payload, +) class SettingsTab: @@ -22,16 +29,25 @@ class SettingsTab: self._create_api_section() self._create_processing_section() self._create_document_types_section() + self._create_cop_mod_section() self._create_save_button() def _setup_layout(self): - """Configura el layout base de la pestaña.""" + """Configura el layout: área con scroll y barra inferior fija con Guardar.""" self.tab.grid_columnconfigure(0, weight=1) self.tab.grid_rowconfigure(0, weight=1) - - self.settings_container = ctk.CTkFrame(self.tab, fg_color="transparent") - self.settings_container.grid(row=0, column=0, sticky="nsew", padx=20, pady=20) + self.tab.grid_rowconfigure(1, weight=0) + + self.settings_container = ctk.CTkScrollableFrame( + self.tab, + fg_color="transparent", + ) + self.settings_container.grid(row=0, column=0, sticky="nsew", padx=20, pady=(20, 8)) self.settings_container.grid_columnconfigure(0, weight=1) + + self.save_footer = ctk.CTkFrame(self.tab, fg_color="transparent") + self.save_footer.grid(row=1, column=0, sticky="ew", padx=20, pady=(4, 16)) + self.save_footer.grid_columnconfigure(0, weight=1) def _create_api_section(self): """Crea la sección de configuración del modelo LLM.""" @@ -50,25 +66,32 @@ class SettingsTab: # Selector de modelo lbl_model = ctk.CTkLabel( self.frame_api, - text="Modelo (Ollama):", + text="Modelo (extracción):", font=self.fonts["ui"] ) lbl_model.grid(row=1, column=0, padx=15, pady=10, sticky="w") self.combo_model = ctk.CTkComboBox( self.frame_api, - values=["5.4", "5.4-mini", "5.4-nano", "5.2", "4.1-mini"], + values=[ + "gpt-5.4", + "gpt-5.4-mini", + "gpt-5.4-nano", + "gpt-4.1-mini", + "grok-4.3", + ], width=250, font=self.fonts["ui"] ) self.combo_model.grid(row=1, column=1, padx=15, pady=10, sticky="w") - self.combo_model.set("5.4") + self.combo_model.set("gpt-5.4") # Texto de ayuda help_text = ( - "ℹ 5.4: Modelo predeterminado\n" - "ℹ 5.4-mini / 4.1-mini: opciones más rápidas\n" - "ℹ 5.4-nano: opción más económica" + "ℹ gpt-5.4: Modelo predeterminado (OpenAI / compatible)\n" + "ℹ gpt-5.4-mini / gpt-4.1-mini: opciones más rápidas\n" + "ℹ gpt-5.4-nano: opción más económica\n" + "ℹ grok-4.3: API xAI (requiere GROK_API_KEY en .env)" ) lbl_help = ctk.CTkLabel( self.frame_api, @@ -151,16 +174,84 @@ class SettingsTab: ) self.allowed_doc_types_entry.grid(row=2, column=0, sticky="ew", padx=15, pady=(0, 10)) self.set_allowed_document_types(list(get_allowed_document_types())) + + def _create_cop_mod_section(self): + """Cuotas moderadora por año (regla Separa COP/MOD).""" + self.frame_cop_mod = ctk.CTkFrame(self.settings_container) + self.frame_cop_mod.grid(row=3, column=0, sticky="ew", padx=5, pady=5) + self.frame_cop_mod.grid_columnconfigure(0, weight=1) + + lbl_title = ctk.CTkLabel( + self.frame_cop_mod, + text="Cuota moderadora (COP/MOD)", + font=self.fonts["section"], + ) + lbl_title.grid(row=0, column=0, sticky="w", padx=15, pady=(10, 4)) + + lbl_help = ctk.CTkLabel( + self.frame_cop_mod, + text=( + "Una línea por año: 2024: 4500, 18200, 47700. " + "La columna «Fecha Servicio» se elige en Proceso de Limpieza al activar la regla." + ), + font=self.fonts["italic_small"], + text_color=self.colors["gray_light"], + justify="left", + ) + lbl_help.grid(row=1, column=0, sticky="w", padx=15, pady=(0, 6)) + + merged, fecha_def = load_cop_mod_preferences() + initial_txt = format_moderadora_text_from_prefs(merged) + + self.moderadora_text = ctk.CTkTextbox( + self.frame_cop_mod, + font=self.fonts["mono_small"], + height=140, + ) + self.moderadora_text.grid(row=2, column=0, sticky="ew", padx=15, pady=(0, 8)) + self.moderadora_text.insert("1.0", initial_txt) + + row_fe = ctk.CTkFrame(self.frame_cop_mod, fg_color="transparent") + row_fe.grid(row=3, column=0, sticky="ew", padx=15, pady=(0, 6)) + row_fe.grid_columnconfigure(1, weight=1) + + ctk.CTkLabel( + row_fe, + text="Sugerencia nombre columna fecha (opcional):", + font=self.fonts["ui"], + ).grid(row=0, column=0, padx=(0, 10), sticky="w") + + self.fecha_column_default_var = ctk.StringVar(value=fecha_def) + self.fecha_column_default_entry = ctk.CTkEntry( + row_fe, + textvariable=self.fecha_column_default_var, + font=self.fonts["ui"], + placeholder_text="Fecha Servicio", + ) + self.fecha_column_default_entry.grid(row=0, column=1, sticky="ew") + + self.btn_moderadora_defaults = ctk.CTkButton( + self.frame_cop_mod, + text="Restaurar valores por defecto", + font=self.fonts["ui"], + command=self._on_restore_moderadora_defaults, + ) + self.btn_moderadora_defaults.grid(row=4, column=0, sticky="w", padx=15, pady=(0, 12)) + + def _on_restore_moderadora_defaults(self) -> None: + txt = format_moderadora_text_from_prefs(default_moderadora_by_year_payload()) + self.moderadora_text.delete("1.0", "end") + self.moderadora_text.insert("1.0", txt) def _create_save_button(self): - """Crea el botón de guardar configuración.""" + """Barra inferior fija: siempre visible aunque el formulario sea largo.""" self.btn_save_settings = ctk.CTkButton( - self.settings_container, - text="Aplicar Configuración", + self.save_footer, + text="Guardar cambios", font=self.fonts["ui"], - height=35 + height=38, ) - self.btn_save_settings.grid(row=3, column=0, padx=15, pady=15, sticky="ew") + self.btn_save_settings.grid(row=0, column=0, sticky="ew") # --- API Pública --- @@ -197,6 +288,19 @@ class SettingsTab: """Establece la lista de tipos documentales válidos.""" self.allowed_doc_types_entry.delete(0, "end") self.allowed_doc_types_entry.insert(0, ", ".join(normalize_document_types(values))) + + def get_moderadora_cuotas_text(self) -> str: + """Texto multilínea año: valores.""" + return self.moderadora_text.get("1.0", "end").strip() + + def set_moderadora_cuotas_ui(self, payload: dict[str, list[int]], fecha_hint: str) -> None: + """Establece cuotas y sugerencia de columna fecha desde prefs.""" + self.moderadora_text.delete("1.0", "end") + self.moderadora_text.insert("1.0", format_moderadora_text_from_prefs(payload)) + self.fecha_column_default_var.set(fecha_hint or "") + + def get_fecha_servicio_column_hint(self) -> str: + return str(self.fecha_column_default_var.get() or "").strip() def enable_controls(self, enabled: bool): """Habilita o deshabilita los controles.""" @@ -204,4 +308,7 @@ class SettingsTab: self.combo_model.configure(state=state) self.concurrency_entry.configure(state=state) self.allowed_doc_types_entry.configure(state=state) + self.moderadora_text.configure(state=state) + self.fecha_column_default_entry.configure(state=state) + self.btn_moderadora_defaults.configure(state=state) self.btn_save_settings.configure(state=state) diff --git a/gui/view.py b/gui/view.py index a0c246f..36d4cd9 100644 --- a/gui/view.py +++ b/gui/view.py @@ -119,6 +119,15 @@ class InvoiceAppView(ctk.CTk): def set_allowed_document_types(self, values: list[str]) -> None: self.settings_tab.set_allowed_document_types(values) + def get_moderadora_cuotas_text(self) -> str: + return self.settings_tab.get_moderadora_cuotas_text() + + def set_moderadora_cuotas_ui(self, payload: dict, fecha_hint: str) -> None: + self.settings_tab.set_moderadora_cuotas_ui(payload, fecha_hint) + + def get_fecha_servicio_column_hint(self) -> str: + return self.settings_tab.get_fecha_servicio_column_hint() + def set_on_run_preclassification(self, callback): self.preclass_tab.set_on_run(callback) @@ -387,6 +396,9 @@ class InvoiceAppView(ctk.CTk): def get_cleaning_selected_rules(self) -> dict[str, list[str]]: return self.cleaning_tab.get_selected_rules() + def get_cleaning_cop_mod_fecha_columns(self) -> dict[str, str]: + return self.cleaning_tab.get_cop_mod_fecha_columns() + def set_cleaning_result(self, df, info_text: str = ""): self.cleaning_tab.set_result(df, info_text) diff --git a/tests/test_csv_cleaning.py b/tests/test_csv_cleaning.py index 530dca5..b88afb5 100644 --- a/tests/test_csv_cleaning.py +++ b/tests/test_csv_cleaning.py @@ -209,6 +209,15 @@ def test_apply_rules_keep_date_seps_preserves_time_then_normalizes() -> None: assert result["fecha"].tolist() == ["30/04/2024"] +def test_apply_rules_us_style_date_comma_then_am_pm_strips_time() -> None: + """Formato tipo Excel/locale US: mes/día/año, coma y hora — solo fecha DD/MM/AAAA.""" + df = pd.DataFrame({"fecha": ["12/27/24, 12:00 AM"]}) + + result, _log = apply_rules(df, {"fecha": ["date_to_ddmmyyyy"]}) + + assert result["fecha"].tolist() == ["27/12/2024"] + + def test_apply_rules_unified_date_rule_noise_and_unparseable_values() -> None: df = pd.DataFrame( { @@ -231,3 +240,108 @@ def test_apply_rules_unified_date_rule_noise_and_unparseable_values() -> None: "no es fecha", "31-febrero-2026", ] + + +def test_apply_rules_cop_mod_moderadora_matches_year() -> None: + df = pd.DataFrame( + {"Fecha Servicio": ["01/01/2024"], "Importe": [4500]}, + ) + result, _log = apply_rules( + df, + {"Importe": ["cop_mod_split"]}, + cop_mod_fecha_column={"Importe": "Fecha Servicio"}, + cop_mod_config={"moderadora_by_year": {"2024": [4500, 18200, 47700]}}, + ) + assert result["Importe_moderadora"].tolist() == [4500] + assert result["Importe_copago"].tolist() == [0] + + +def test_apply_rules_cop_mod_non_moderadora_to_copago() -> None: + df = pd.DataFrame( + {"Fecha Servicio": ["01/01/2024"], "Importe": [4982]}, + ) + result, _log = apply_rules( + df, + {"Importe": ["cop_mod_split"]}, + cop_mod_fecha_column={"Importe": "Fecha Servicio"}, + cop_mod_config={"moderadora_by_year": {"2024": [4500, 18200, 47700]}}, + ) + assert result["Importe_copago"].tolist() == [4982] + assert result["Importe_moderadora"].tolist() == [0] + + +def test_apply_rules_cop_mod_two_phase_dirty_date() -> None: + df = pd.DataFrame( + {"Fecha Servicio": ["15/jun/2024"], "Importe": [4500]}, + ) + result, _log = apply_rules( + df, + { + "Fecha Servicio": ["date_to_ddmmyyyy"], + "Importe": ["cop_mod_split"], + }, + cop_mod_fecha_column={"Importe": "Fecha Servicio"}, + cop_mod_config={"moderadora_by_year": {"2024": [4500, 18200, 47700]}}, + ) + assert result["Fecha Servicio"].iloc[0] == "15/06/2024" + assert result["Importe_moderadora"].iloc[0] == 4500 + + +def test_apply_rules_cop_mod_year_not_in_table_goes_copago() -> None: + df = pd.DataFrame( + {"Fecha Servicio": ["01/01/2030"], "Importe": [4500]}, + ) + result, _log = apply_rules( + df, + {"Importe": ["cop_mod_split"]}, + cop_mod_fecha_column={"Importe": "Fecha Servicio"}, + cop_mod_config={"moderadora_by_year": {"2024": [4500]}}, + ) + assert result["Importe_copago"].tolist() == [4500] + assert result["Importe_moderadora"].tolist() == [0] + + +def test_apply_rules_cop_mod_null_amount_zeros() -> None: + df = pd.DataFrame({"Fecha Servicio": ["01/01/2024"], "Importe": [pd.NA]}) + result, _log = apply_rules( + df, + {"Importe": ["cop_mod_split"]}, + cop_mod_fecha_column={"Importe": "Fecha Servicio"}, + cop_mod_config={"moderadora_by_year": {"2024": [4500]}}, + ) + assert result["Importe_copago"].tolist() == [0] + assert result["Importe_moderadora"].tolist() == [0] + + +def test_apply_rules_nota_extract_one_line_sin_autorizacion() -> None: + texto = ( + "Las retenciones son a nivel informativo no afecta el total a pagar. " + "Paciente: 1063146603 - VILLA BARRIOS NORMA JUDITH Atención No: 63728; " + "Contrato No. URGENCIAS. Valor en Letras: CIENTO OCHENTA Y SIETE MIL " + "QUINIENTOS TREINTA Y SIETE PESOS Fecha ingreso 19/11/2023 fecha egreso 19/11/2023" + ) + df = pd.DataFrame({"notas": [texto]}) + result, log = apply_rules(df, {"notas": ["nota_paciente_autorizacion"]}) + assert result["notas_nota_paciente_doc"].iloc[0] == "1063146603" + assert pd.isna(result["notas_nota_autorizacion"].iloc[0]) + assert any("Extraer datos de notas" in line for line in log) + + +def test_apply_rules_nota_extract_multiline_con_autorizacion() -> None: + texto = ( + "Las retenciones son a nivel informativo no afecta el total a pagar.\n" + "Paciente: 30662603 - LLORENTE HERNANDEZ CARMEN LUCIA Atención No: 69426;\n" + "Autorización No. 227490546; Contrato No. 118. Valor en Letras: CIENTO NOVENTA Y OCHO\n" + "MIL SEISCIENTOS DIEZ PESOS Fecha ingreso 23/01/2024 fecha egreso 23/01/2024" + ) + df = pd.DataFrame({"notas": [texto]}) + result, _log = apply_rules(df, {"notas": ["nota_paciente_autorizacion"]}) + assert result["notas_nota_paciente_doc"].iloc[0] == "30662603" + assert result["notas_nota_autorizacion"].iloc[0] == "227490546" + + +def test_apply_rules_nota_extract_sin_patrones() -> None: + df = pd.DataFrame({"notas": ["sin datos relevantes aquí"]}) + result, _log = apply_rules(df, {"notas": ["nota_paciente_autorizacion"]}) + assert pd.isna(result["notas_nota_paciente_doc"].iloc[0]) + assert pd.isna(result["notas_nota_autorizacion"].iloc[0]) diff --git a/tests/test_csv_manager.py b/tests/test_csv_manager.py index 489db75..c46dd7c 100644 --- a/tests/test_csv_manager.py +++ b/tests/test_csv_manager.py @@ -47,6 +47,6 @@ def test_append_rows_one_physical_line_per_record() -> None: {"radicado": "2", "descripcion": "ok"}, ] mgr.append_rows(str(path), rows, header) - text = path.read_text(encoding="utf-8") + text = path.read_text(encoding="utf-8-sig") physical_lines = text.splitlines() assert len(physical_lines) == 3 diff --git a/tests/test_csv_utils.py b/tests/test_csv_utils.py new file mode 100644 index 0000000..09de2a6 --- /dev/null +++ b/tests/test_csv_utils.py @@ -0,0 +1,59 @@ +from __future__ import annotations + +from core.csv_utils import merge_verbatim_long_strings, set_at_path + + +def test_merge_replaces_truncated_string_same_path() -> None: + raw = { + "Notas": ( + "Las retenciones son informativas. Paciente: 1 - NOMBRE " + "Contrato No. X. Valor en Letras: MIL PESOS Fecha ingreso 01/01/2024" + ), + } + norm = {"Notas": "Las retenciones son informativas. Paciente: 1 - NOMBRE "} + merge_verbatim_long_strings(raw, norm, ["Notas"]) + assert norm["Notas"] == raw["Notas"] + + +def test_merge_nested_path() -> None: + raw = {"cabecera": {"Notas": "A" * 200}} + norm = {"cabecera": {"Notas": "A" * 40}} + merge_verbatim_long_strings(raw, norm, ["cabecera/Notas"]) + assert norm["cabecera"]["Notas"] == "A" * 200 + + +def test_merge_resolves_by_unique_last_segment() -> None: + raw = {"Notas": "texto completo desde raíz cuando el normalizado usa cabecera/Notas"} + norm = {"cabecera": {"Notas": "texto completo"}} + merge_verbatim_long_strings(raw, norm, ["cabecera/Notas"]) + assert norm["cabecera"]["Notas"] == raw["Notas"] + + +def test_merge_skips_ambiguous_same_last_segment() -> None: + raw = { + "a": {"Notas": "primera nota " + "x" * 50}, + "b": {"Notas": "segunda nota " + "y" * 50}, + } + norm = {"cabecera": {"Notas": "primera nota"}} + merge_verbatim_long_strings(raw, norm, ["cabecera/Notas"]) + assert norm["cabecera"]["Notas"] == "primera nota" + + +def test_merge_skips_when_normalized_not_prefix_of_raw() -> None: + raw = {"Notas": "completamente distinto " + "z" * 80} + norm = {"Notas": "otro texto corto"} + merge_verbatim_long_strings(raw, norm, ["Notas"]) + assert norm["Notas"] == "otro texto corto" + + +def test_merge_skips_paths_with_wildcard() -> None: + raw = {"items": [{"x": "long " * 20}]} + norm = {"items": [{"x": "long "}]} + merge_verbatim_long_strings(raw, norm, ["items/*/x"]) + assert norm["items"][0]["x"] == "long " + + +def test_set_at_path_creates_intermediates() -> None: + root: dict = {} + set_at_path(root, "a/b/c", 1) + assert root == {"a": {"b": {"c": 1}}}