from __future__ import annotations import argparse import hashlib import json import os import re import tempfile from collections.abc import Callable, Iterable from contextlib import closing from decimal import Decimal from io import BytesIO from pathlib import Path from typing import Any, BinaryIO, Literal from urllib.request import Request, urlopen from zipfile import BadZipFile, ZipFile import openpyxl import xlrd from pydantic import BaseModel, ConfigDict, Field from PyPDF2 import PdfReader from app.soat_tariffs.domain.models import CatalogSource, SoatCatalog from app.soat_tariffs.infrastructure.json_catalog import DEFAULT_CATALOG_DIR, JsonSoatTariffCatalog ROOT = Path(__file__).resolve().parents[2] DEFAULT_SOURCE_DIR = ROOT / "app" / "soat_tariffs" / "sources" DEFAULT_MANIFEST_PATH = DEFAULT_SOURCE_DIR / "manifest.json" CATALOG_YEARS = tuple(range(2022, 2027)) SURGICAL_GROUPS = (*range(2, 14), *range(20, 24)) COMPONENT_ORDER = ("cirujano", "anestesia", "ayudantia", "sala", "materiales") class SourceSpec(BaseModel): model_config = ConfigDict(extra="forbid") id: str = Field(min_length=1) url: str = Field(min_length=1) local_file: str = Field(min_length=1) sha256: str = Field(pattern=r"^[0-9a-f]{64}$") role: Literal["annual_tariffs", "normative_circular", "unit_resolution"] applies_to: list[int] = Field(min_length=1) page_table: str = Field(min_length=1) archive_member: str | None = None def load_manifest(path: Path = DEFAULT_MANIFEST_PATH) -> list[SourceSpec]: try: payload = json.loads(path.read_text(encoding="utf-8")) sources = [SourceSpec.model_validate(item) for item in payload["sources"]] except Exception as exc: raise ValueError(f"Manifiesto SOAT invalido: {path}") from exc ids = [source.id for source in sources] local_files = [source.local_file for source in sources] if len(ids) != len(set(ids)) or len(local_files) != len(set(local_files)): raise ValueError("El manifiesto SOAT contiene identificadores o archivos duplicados") return sources def _digest_bytes(content: bytes) -> str: return hashlib.sha256(content).hexdigest() def _source_path(source: SourceSpec, source_dir: Path) -> Path: path = source_dir / source.local_file if path.parent != source_dir: raise ValueError(f"Ruta de fuente SOAT no permitida: {source.local_file}") return path def validate_source(source: SourceSpec, source_dir: Path = DEFAULT_SOURCE_DIR) -> Path: path = _source_path(source, source_dir) if not path.is_file(): raise FileNotFoundError(f"Fuente SOAT ausente: {source.local_file}") if _digest_bytes(path.read_bytes()) != source.sha256: raise ValueError(f"Hash invalido para fuente SOAT: {source.local_file}") if source.archive_member: try: with ZipFile(path) as archive: if source.archive_member not in archive.namelist(): raise ValueError( f"Miembro ZIP ausente en {source.local_file}: {source.archive_member}" ) except BadZipFile as exc: raise ValueError(f"ZIP SOAT invalido: {source.local_file}") from exc return path def validate_sources( sources: Iterable[SourceSpec], source_dir: Path = DEFAULT_SOURCE_DIR ) -> dict[str, Path]: return {source.id: validate_source(source, source_dir) for source in sources} def _response_stream(response: Any) -> BinaryIO: if hasattr(response, "read"): return response raise TypeError("La descarga SOAT no produjo un flujo binario") def _open_official_source(url: str) -> Any: request = Request(url, headers={"User-Agent": "EpicrisisIA-SOAT-Catalog/2"}) return urlopen(request, timeout=60) def download_sources( *, manifest_path: Path = DEFAULT_MANIFEST_PATH, source_dir: Path = DEFAULT_SOURCE_DIR, opener: Callable[[str], Any] = _open_official_source, ) -> list[str]: source_dir.mkdir(parents=True, exist_ok=True) statuses: list[str] = [] for source in load_manifest(manifest_path): destination = _source_path(source, source_dir) if destination.is_file() and _digest_bytes(destination.read_bytes()) == source.sha256: statuses.append(f"{source.local_file}: present") continue descriptor, temporary_name = tempfile.mkstemp( prefix=f".{destination.name}.", suffix=".tmp", dir=source_dir ) temporary = Path(temporary_name) try: digest = hashlib.sha256() with os.fdopen(descriptor, "wb") as target, closing(opener(source.url)) as response: stream = _response_stream(response) while chunk := stream.read(1024 * 1024): digest.update(chunk) target.write(chunk) target.flush() os.fsync(target.fileno()) if digest.hexdigest() != source.sha256: raise ValueError(f"Hash descargado invalido para {source.local_file}") os.replace(temporary, destination) validate_source(source, source_dir) finally: temporary.unlink(missing_ok=True) statuses.append(f"{source.local_file}: downloaded") return statuses def _archive_member(source: SourceSpec, source_dir: Path) -> bytes: if source.archive_member is None: raise ValueError(f"La fuente {source.id} no declara miembro ZIP") path = validate_source(source, source_dir) with ZipFile(path) as archive: try: return archive.read(source.archive_member) except KeyError as exc: raise ValueError( f"Miembro ZIP ausente en {source.local_file}: {source.archive_member}" ) from exc def _decimal(value: Any) -> Decimal: if value in (None, ""): raise ValueError("Valor tarifario vacio") return Decimal(str(value)).normalize() def _decimal_text(value: Decimal) -> str: text = format(value, "f") return text.rstrip("0").rstrip(".") if "." in text else text def _xls_rows(content: bytes, sheet_name: str) -> list[tuple[Any, ...]]: workbook = xlrd.open_workbook(file_contents=content) try: sheet = workbook.sheet_by_name(sheet_name) except xlrd.biffh.XLRDError as exc: raise ValueError(f"Hoja XLS SOAT ausente: {sheet_name}") from exc return [tuple(sheet.row_values(index)) for index in range(sheet.nrows)] def _xlsx_rows(content: bytes, sheet_name: str) -> list[tuple[Any, ...]]: try: workbook = openpyxl.load_workbook(BytesIO(content), data_only=True, read_only=True) except Exception as exc: raise ValueError("Fuente XLSX SOAT invalida") from exc try: if sheet_name not in workbook.sheetnames: raise ValueError(f"Hoja XLSX SOAT ausente: {sheet_name}") return [tuple(row) for row in workbook[sheet_name].iter_rows(values_only=True)] finally: workbook.close() def _numeric_code(value: Any) -> int | None: if isinstance(value, bool): return None try: number = Decimal(str(value)) except Exception: return None if number != number.to_integral_value(): return None return int(number) def extract_surgical_procedures(content: bytes) -> list[dict[str, Any]]: procedures: list[dict[str, Any]] = [] for row in _xls_rows(content, "SOAT 2022"): if len(row) < 3: continue code = _numeric_code(row[0]) group = _numeric_code(row[2]) description = str(row[1] or "").strip() if code is None or not 0 < code < 19000 or group not in SURGICAL_GROUPS or not description: continue normalized = str(code).zfill(5) if not re.fullmatch(r"\d{5}", normalized): continue procedures.append( {"code": normalized, "description": description, "surgical_group": group} ) codes = [item["code"] for item in procedures] groups = {item["surgical_group"] for item in procedures} if len(procedures) != 1771 or len(codes) != len(set(codes)): raise ValueError("El XLS 2022 no contiene exactamente 1.771 procedimientos quirurgicos unicos") if groups != set(SURGICAL_GROUPS): raise ValueError("El XLS 2022 no contiene los 16 grupos quirurgicos esperados") return procedures def _component_codes() -> dict[int, dict[str, str]]: result: dict[int, dict[str, str]] = {} for position, group in enumerate(SURGICAL_GROUPS): components = { "cirujano": str(39000 + position), "anestesia": str(39100 + position), "sala": str(39204 + position), } if group >= 6: assistant_position = (*range(6, 14), *range(20, 24)).index(group) components["ayudantia"] = str(39117 + assistant_position) if group <= 13: if group <= 3: components["materiales"] = "39301" elif group <= 6: components["materiales"] = "39302" elif group <= 9: components["materiales"] = "39303" else: components["materiales"] = "39304" result[group] = { name: components[name] for name in COMPONENT_ORDER if name in components } return result COMPONENT_CODES = _component_codes() EXPECTED_COMPONENT_CODES = { code for components in COMPONENT_CODES.values() for code in components.values() } def extract_tariff_rows( content: bytes, *, year: int, archive_member: str ) -> dict[str, dict[str, Any]]: if archive_member.lower().endswith(".xls"): rows = _xls_rows(content, "SOAT 2022") else: sheet_name = "DECRETO" if year in {2023, 2024} else "Hoja1" rows = _xlsx_rows(content, sheet_name) extracted: dict[str, dict[str, Any]] = {} for row in rows: if len(row) < 3: continue numeric_code = _numeric_code(row[0]) if numeric_code is None: continue code = str(numeric_code) if code not in EXPECTED_COMPONENT_CODES: continue description = str(row[1] or "").strip() if not description: raise ValueError(f"Descripcion vacia para componente SOAT {code} en {year}") official_value = _numeric_code(row[3]) if len(row) > 3 else None extracted[code] = { "description": description, "coefficient": _decimal(row[2]), "official_value": official_value, } if set(extracted) != EXPECTED_COMPONENT_CODES or len(extracted) != 64: missing = sorted(EXPECTED_COMPONENT_CODES - set(extracted)) raise ValueError(f"La fuente {year} no contiene los 64 componentes esperados; faltan {missing}") if year <= 2025 and any(item["official_value"] is None for item in extracted.values()): raise ValueError(f"La fuente {year} no publica todos los valores oficiales en pesos") if year == 2026 and any(item["official_value"] is not None for item in extracted.values()): raise ValueError("La fuente 2026 debe calcularse desde coeficientes UVB") return extracted def _joined_cells(rows: Iterable[tuple[Any, ...]], limit: int = 12) -> str: return " ".join( str(value) for row in list(rows)[:limit] for value in row if value not in (None, "") ) def extract_header_unit_value(content: bytes, *, year: int, archive_member: str) -> Decimal: if archive_member.lower().endswith(".xls"): text = _joined_cells(_xls_rows(content, "SOAT 2022")) pattern = r"Valor diario\s*\$\s*([\d.]+,[\d]{2})" else: text = _joined_cells(_xlsx_rows(content, "DECRETO")) pattern = rf"UVT\)?\s+de\s+{year}\s*\$\s*([\d.]+,[\d]{{2}})" match = re.search(pattern, text, re.IGNORECASE) if not match: raise ValueError(f"No se pudo extraer la unidad oficial del encabezado {year}") return Decimal(match.group(1).replace(".", "").replace(",", ".")) def extract_uvb_value(content: bytes, year: int) -> Decimal: try: text = "\n".join(page.extract_text() or "" for page in PdfReader(BytesIO(content)).pages) except Exception as exc: raise ValueError(f"Resolucion UVB {year} invalida") from exc pattern = rf"año\s+{year}\s+será.*?\$\s*(\d{{2}})[.,](\d{{3}})[.,]\d{{2}}" match = re.search(pattern, text, re.IGNORECASE | re.DOTALL) if not match: raise ValueError(f"No se pudo extraer el valor UVB oficial para {year}") return Decimal(f"{match.group(1)}{match.group(2)}") def _catalog_source(source: SourceSpec) -> dict[str, Any]: return CatalogSource( url=source.url, sha256=source.sha256, local_file=f"app/soat_tariffs/sources/{source.local_file}", page_table=source.page_table, ).model_dump(exclude_none=True) def _sources_by_id(sources: Iterable[SourceSpec]) -> dict[str, SourceSpec]: return {source.id: source for source in sources} def _year_sources(sources: Iterable[SourceSpec], year: int) -> list[SourceSpec]: applicable = [source for source in sources if year in source.applies_to] procedure_source = next(source for source in sources if source.id == "tariffs-2022") if procedure_source not in applicable: applicable.insert(0, procedure_source) return applicable def build_catalog( year: int, *, manifest_path: Path = DEFAULT_MANIFEST_PATH, source_dir: Path = DEFAULT_SOURCE_DIR, ) -> dict[str, Any]: if year not in CATALOG_YEARS: raise ValueError(f"Vigencia SOAT no soportada: {year}") sources = load_manifest(manifest_path) source_map = _sources_by_id(sources) validate_sources(sources, source_dir) procedure_source = source_map["tariffs-2022"] procedures = extract_surgical_procedures(_archive_member(procedure_source, source_dir)) annual_source = source_map[f"tariffs-{year}"] annual_content = _archive_member(annual_source, source_dir) tariffs = extract_tariff_rows( annual_content, year=year, archive_member=annual_source.archive_member or "" ) if year <= 2024: unit = "SMLDV" if year == 2022 else "UVT" unit_value = extract_header_unit_value( annual_content, year=year, archive_member=annual_source.archive_member or "" ) else: unit = "UVB" resolution = source_map[f"uvb-{year}"] unit_value = extract_uvb_value(validate_source(resolution, source_dir).read_bytes(), year) entries: list[dict[str, Any]] = [] procedure_catalog_source = _catalog_source(procedure_source) for procedure in procedures: group = procedure["surgical_group"] components = [] official_values: list[int] = [] coefficients: list[Decimal] = [] for name, code in COMPONENT_CODES[group].items(): tariff = tariffs[code] coefficient = tariff["coefficient"] coefficients.append(coefficient) component = { "name": name, "code": code, "coefficient": _decimal_text(coefficient), } if tariff["official_value"] is not None: component["official_value"] = tariff["official_value"] official_values.append(tariff["official_value"]) components.append(component) entry = { **procedure, "base_coefficient": _decimal_text(sum(coefficients, Decimal(0))), "components": components, "source": procedure_catalog_source, } if len(official_values) == len(components): entry["official_base_value"] = sum(official_values) entries.append(entry) payload = { "year": year, "unit": unit, "unit_value": _decimal_text(unit_value), "version": f"soat-tarifas-{year}-v2", "generated_from": [_catalog_source(source) for source in _year_sources(sources, year)], "entries": entries, } SoatCatalog.model_validate(payload) return payload def catalog_bytes( year: int, *, manifest_path: Path = DEFAULT_MANIFEST_PATH, source_dir: Path = DEFAULT_SOURCE_DIR, ) -> bytes: payload = build_catalog(year, manifest_path=manifest_path, source_dir=source_dir) return (json.dumps(payload, ensure_ascii=False, indent=2) + "\n").encode() def generate( *, manifest_path: Path = DEFAULT_MANIFEST_PATH, source_dir: Path = DEFAULT_SOURCE_DIR, catalog_dir: Path = DEFAULT_CATALOG_DIR, ) -> None: catalog_dir.mkdir(parents=True, exist_ok=True) for year in CATALOG_YEARS: content = catalog_bytes(year, manifest_path=manifest_path, source_dir=source_dir) destination = catalog_dir / f"soat-{year}.json" descriptor, temporary_name = tempfile.mkstemp( prefix=f".{destination.name}.", suffix=".tmp", dir=catalog_dir ) temporary = Path(temporary_name) try: with os.fdopen(descriptor, "wb") as target: target.write(content) target.flush() os.fsync(target.fileno()) os.replace(temporary, destination) finally: temporary.unlink(missing_ok=True) def check( *, manifest_path: Path = DEFAULT_MANIFEST_PATH, source_dir: Path = DEFAULT_SOURCE_DIR, catalog_dir: Path = DEFAULT_CATALOG_DIR, ) -> dict[int, str]: for year in CATALOG_YEARS: expected = catalog_bytes(year, manifest_path=manifest_path, source_dir=source_dir) path = catalog_dir / f"soat-{year}.json" if not path.is_file(): raise FileNotFoundError(f"Catalogo SOAT {year} no disponible") if path.read_bytes() != expected: raise ValueError(f"Catalogo SOAT {year} alterado o no deterministico") return JsonSoatTariffCatalog(catalog_dir).validate_all() def main() -> int: parser = argparse.ArgumentParser(description="Genera, descarga o valida catalogos tarifarios SOAT.") parser.add_argument("command", choices=["download", "generate", "check"]) args = parser.parse_args() if args.command == "download": for status in download_sources(): print(status) return 0 if args.command == "generate": generate() statuses = JsonSoatTariffCatalog().validate_all() else: statuses = check() for year, status in statuses.items(): print(f"soat-{year}: {status}") return 0 if __name__ == "__main__": raise SystemExit(main())