From 4078c702dd4f384f14d7aa6122d94e7329f6618c Mon Sep 17 00:00:00 2001 From: erik Date: Sun, 1 Mar 2026 21:00:28 -0500 Subject: [PATCH] docs: Enhance docstrings across multiple modules for improved clarity and consistency, detailing parameters, return values, and exceptions. Remove obsolete test files related to LLM and PDF utilities. --- application/database.py | 49 ++++++++++++- application/normalizer.py | 15 +++- application/preclassification.py | 44 +++++++++-- application/service.py | 114 ++++++++++++++++++++++++----- core/pdf_utils.py | 77 +++++++++++++++---- tests/test_llm_smoke.py | 24 ------ tests/test_pdf_utils_tiff_range.py | 67 ----------------- tests/test_service_tiff_range.py | 72 ------------------ tests/test_utils_patterns.py | 43 ----------- 9 files changed, 255 insertions(+), 250 deletions(-) delete mode 100644 tests/test_llm_smoke.py delete mode 100644 tests/test_pdf_utils_tiff_range.py delete mode 100644 tests/test_service_tiff_range.py delete mode 100644 tests/test_utils_patterns.py diff --git a/application/database.py b/application/database.py index c558dd7..f54c7e1 100644 --- a/application/database.py +++ b/application/database.py @@ -8,12 +8,23 @@ import os logger = logging.getLogger(__name__) class DatabaseManager: + """ + Gestiona la persistencia de resultados de preclasificación en una base de datos SQLite. + + Estructura de tablas: + - runs: Registra cada ejecución global (fecha, ruta raíz, configuración). + - results: Almacena el estado de procesamiento de cada archivo individual. + - matches: Detalle de campos específicos encontrados en cada archivo. + """ def __init__(self, db_path: str = "preclassification.db"): self.db_path = db_path self._init_db() def _init_db(self): - """Inicializa las tablas si no existen.""" + """ + Inicializa el esquema de la base de datos si no existe. + Crea las tablas 'runs', 'results' y 'matches' con sus respectivas llaves foráneas. + """ try: conn = sqlite3.connect(self.db_path) c = conn.cursor() @@ -60,7 +71,16 @@ class DatabaseManager: logger.error(f"Error inicializando base de datos: {e}") def create_run(self, root_path: str, config: Dict[str, Any]) -> int: - """Crea una nueva ejecución y devuelve su ID.""" + """ + Registra una nueva sesión de procesamiento. + + Args: + root_path: Directorio raíz procesado. + config: Diccionario de configuración usado (se guarda como JSON). + + Returns: + ID de la ejecución creada. + """ conn = sqlite3.connect(self.db_path) c = conn.cursor() timestamp = datetime.now().isoformat() @@ -72,7 +92,19 @@ class DatabaseManager: return run_id def add_result(self, run_id: int, radicado: str, file_path: str, status: str, raw_response: str) -> int: - """Añade un resultado de archivo procesado y devuelve su ID.""" + """ + Añade el resultado del procesamiento de un archivo. + + Args: + run_id: ID de la ejecución padre. + radicado: Identificador del radicado. + file_path: Ruta al archivo físico. + status: Estado final ('MATCH', 'NO_MATCH', 'ERROR'). + raw_response: Respuesta completa del modelo. + + Returns: + ID del resultado insertado. + """ conn = sqlite3.connect(self.db_path) c = conn.cursor() c.execute('INSERT INTO results (run_id, radicado, file_path, status, raw_response) VALUES (?, ?, ?, ?, ?)', @@ -83,7 +115,16 @@ class DatabaseManager: return result_id def add_match(self, result_id: int, key: str, expected_value: Optional[str], found_value: Optional[str], exists: bool): - """Añade el detalle de un campo encontrado (o no).""" + """ + Registra el hallazgo (o ausencia) de un campo específico en un documento. + + Args: + result_id: ID del resultado al que pertenece. + key: Nombre del campo/llave. + expected_value: Valor que se esperaba encontrar (si aplica). + found_value: Valor extraído por el modelo. + exists: Booleano que indica si el campo fue efectivamente localizado. + """ conn = sqlite3.connect(self.db_path) c = conn.cursor() c.execute('INSERT INTO matches (result_id, key, expected_value, found_value, exists_bool) VALUES (?, ?, ?, ?, ?)', diff --git a/application/normalizer.py b/application/normalizer.py index 81eee4e..3a25627 100644 --- a/application/normalizer.py +++ b/application/normalizer.py @@ -76,9 +76,20 @@ def normalize_to_schema( paths: List[str], profile_name: str, ) -> Optional[Dict[str, Any]]: - """Usa un LLM para mapear `data` a un objeto que siga el template canónico. + """ + Usa un LLM para mapear un JSON extraído a un esquema canónico definido por rutas. - Retorna None si no hay API Key o si el LLM falla. + Este proceso asegura que, independientemente de cómo el extractor haya devuelto los datos, + el resultado final tenga la estructura jerárquica esperada (ej. 'cabecera/fecha', 'items/*/descripcion'). + + Args: + data: El diccionario con los datos extraídos del documento. + paths: Lista de rutas canónicas (ej. ["cliente/nombre", "totales/iva"]). + profile_name: Nombre del perfil de extracción (ej. 'invoice') para cargar instrucciones. + + Returns: + Un diccionario con la clave "normalized" conteniendo el objeto transformado + y "_usage" con las estadísticas de tokens. Retorna None si falla. """ try: t0 = time.perf_counter() diff --git a/application/preclassification.py b/application/preclassification.py index be5e694..b761ecd 100644 --- a/application/preclassification.py +++ b/application/preclassification.py @@ -16,12 +16,27 @@ from core.pdf_utils import document_to_images_bs64 logger = logging.getLogger(__name__) class PreclassificationService: + """ + Servicio para la preclasificación masiva de documentos. + + Permite identificar si un documento cumple con ciertos criterios (patterns) + y extraer información clave para organizar los archivos físicamente en carpetas. + """ def __init__(self, extractor: OllamaExtractor): self.extractor = extractor self.db = DatabaseManager() def export_to_excel(self, run_id: int, output_path: str) -> bool: - """Exporta los resultados de una ejecución a Excel (xlsx) o CSV.""" + """ + Exporta los resultados de una ejecución de preclasificación a Excel o CSV. + + Args: + run_id: ID de la ejecución en la base de datos. + output_path: Ruta del archivo de salida (.xlsx o .csv). + + Returns: + True si la exportación fue exitosa, False si no hay datos. + """ try: results = self.db.get_run_results(run_id) if not results: @@ -74,8 +89,15 @@ class PreclassificationService: def organize_files(self, run_id: int, base_output_path: str) -> tuple[int, int]: """ - Organiza los archivos MATCH en carpetas jerárquicas. - Devuelve (total_copiados, total_errores). + Copia físicamente los archivos que hicieron 'MATCH' a una nueva estructura de carpetas. + La jerarquía se determina automáticamente basada en los valores extraídos. + + Args: + run_id: ID de la ejecución. + base_output_path: Directorio donde se creará la nueva estructura. + + Returns: + Tupla (total_copiados, total_errores). """ run = self.db.get_run(run_id) mode = "radicados" @@ -173,7 +195,7 @@ class PreclassificationService: return count_ok, count_err def _sanitize_filename(self, name: str) -> str: - """Limpia caracteres inválidos para rutas de archivo.""" + """Limpia caracteres inválidos para rutas de archivo en Windows/Linux.""" # Reemplazar caracteres no alfanuméricos (excepto espacio, guion, punto) clean = re.sub(r'[\\/*?:"<>|]', "", name) clean = clean.replace("\n", " ").replace("\r", "").strip() @@ -191,8 +213,18 @@ class PreclassificationService: mode: str = "radicados" # "radicados" | "general" ) -> int: """ - Ejecuta el flujo de preclasificación. - Devuelve el run_id. + Ejecuta el flujo completo de preclasificación sobre un directorio. + + Args: + root_path: Directorio con los documentos. + patterns: Reglas de validación (ej. buscar NIT, buscar Fecha). + target_info: Filtro de archivos por nombre/patrón. + page_range: Rango de páginas a procesar por documento. + on_progress: Callback para reportar avance a la GUI. + mode: 'radicados' (carpetas) o 'general' (archivos sueltos). + + Returns: + ID de la ejecución (run_id) generada. """ # 1. Crear Run en BD mode = str(mode or "radicados").strip().lower() diff --git a/application/service.py b/application/service.py index 734e318..d1bd8f1 100644 --- a/application/service.py +++ b/application/service.py @@ -13,10 +13,8 @@ import os import time import asyncio from .strategies import InvoiceExtractorStrategy -from .prompts import PromptBuilder from .profiles import get_profile from .config import settings -from .normalizer import normalize_to_schema from .logging_config import log_execution_time from .file_discovery import get_sorted_radicados, find_file_for_type_in_radicados, get_files_as_radicados from .batch_processor import BatchProcessor @@ -35,6 +33,15 @@ class InvoiceProcessingService: batch_processor_factory: Callable[[InvoiceProcessingService], BatchProcessor], estimator_factory: Callable[[InvoiceProcessingService], Estimator] ) -> None: + """ + Inicializa el servicio de procesamiento de facturas. + + Args: + extractor: Estrategia para extraer datos (ej. GPT-4o, Claude). + batch_processor_factory: Fábrica para crear el procesador por lotes, + evitando dependencias circulares. + estimator_factory: Fábrica para crear el estimador de costos y tokens. + """ self.extractor = extractor # Componentes delegados inyectados mediante factories para evitar dependencia circular directa en construcción self.batch_processor = batch_processor_factory(self) @@ -48,8 +55,20 @@ class InvoiceProcessingService: except Exception: self._csv_lock = None - def _convert_to_images(self, pdf_path: str, dpi: int, thread_count: int, tiff_pages, tiff_page_range) -> List[str]: - """Convierte el documento a imágenes base64.""" + def _convert_to_images(self, pdf_path: str, dpi: int, thread_count: int, tiff_pages: Optional[Tuple[int, int]], tiff_page_range: Optional[Tuple[int, int]]) -> List[str]: + """ + Convierte un documento (PDF o TIFF) a una lista de imágenes codificadas en base64. + + Args: + pdf_path: Ruta al archivo. + dpi: Resolución de la conversión. + thread_count: Número de hilos para procesamiento paralelo (solo PDF). + tiff_pages: Páginas por defecto para TIFF si no se especifica rango. + tiff_page_range: Rango específico de páginas (inicio, fin) para TIFF. + + Returns: + Lista de strings con las imágenes en base64 (formato PNG). + """ filename = os.path.basename(pdf_path) tiff_pages_cfg = getattr(settings.imaging, 'tiff_pages', (2, 3)) if tiff_pages is None else tiff_pages poppler = getattr(settings.imaging, 'poppler_path', None) @@ -64,7 +83,18 @@ class InvoiceProcessingService: raise def _apply_extractor_fallback(self, pdf_path: str) -> List[str]: - """Carga imágenes con DPI reducido (máx 5 páginas) para el fallback.""" + """ + Intento de recuperación ante errores de extracción usando parámetros reducidos. + + Carga imágenes con DPI fijo (400) y limita a las primeras 5 páginas para + reducir el consumo de tokens y evitar límites de contexto. + + Args: + pdf_path: Ruta al archivo. + + Returns: + Lista de hasta 5 imágenes en base64. + """ fb_dpi = 400 tiff_pages_fb = getattr(settings.imaging, 'tiff_pages', (2, 3)) poppler = getattr(settings.imaging, 'poppler_path', None) @@ -72,7 +102,20 @@ class InvoiceProcessingService: return images[:5] if len(images) > 5 else images def _parse_model_output(self, model_output: dict, pdf_path: str, dpi_cfg: int) -> Tuple[Any, Any]: - """Extrae y parsea el JSON del output del modelo.""" + """ + Extrae y valida el bloque JSON de la respuesta del modelo de lenguaje. + + Args: + model_output: Diccionario con la respuesta cruda ("raw") y el uso de tokens ("usage"). + pdf_path: Ruta del archivo original (para logging). + dpi_cfg: DPI usado (para logging). + + Returns: + Tupla (datos_dict, usage_dict) con el JSON parseado y estadísticas de uso. + + Raises: + ValueError: Si no se encuentra un JSON válido en la respuesta. + """ filename = os.path.basename(pdf_path) raw_text = model_output.get("raw", "") usage = model_output.get("usage") @@ -101,9 +144,23 @@ class InvoiceProcessingService: tiff_page_range: Optional[Tuple[int, int]] = None, history: Optional[List[Dict[str, Any]]] = None, ) -> Tuple[Any, Any]: - """Extrae un JSON completo desde un documento (PDF/TIFF) usando el extractor. + """ + Extrae un JSON completo desde un documento (PDF/TIFF) usando el extractor. - En caso de error, reintenta con DPI reducido y máximo 5 páginas. + El proceso incluye: + 1. Conversión del documento a imágenes base64. + 2. Llamada al modelo de extracción. + 3. En caso de error, reintento automático con parámetros de recuperación (DPI reducido). + 4. Parseo del resultado JSON. + + Args: + pdf_path: Ruta al archivo PDF o TIFF. + instruction: Instrucción personalizada para el modelo (opcional). + tiff_page_range: Rango de páginas específico si es un TIFF. + history: Historial de mensajes para el modelo (opcional). + + Returns: + Tupla (datos_extraidos, estadisticas_uso). """ filename = os.path.basename(pdf_path) logger.info(f"Iniciando extracción de documento: {filename}", extra={"pdf": pdf_path}) @@ -139,7 +196,11 @@ class InvoiceProcessingService: tiff_page_range: Optional[Tuple[int, int]] = None, history: Optional[List[Dict[str, Any]]] = None, ) -> Tuple[Any, Any]: - """Versión asíncrona de extract_full_from_pdf.""" + """ + Versión asíncrona de extract_full_from_pdf. + + Realiza la conversión de imágenes en un hilo separado para no bloquear el loop de eventos. + """ filename = os.path.basename(pdf_path) logger.info(f"Iniciando extracción asíncrona de documento: {filename}", extra={"pdf": pdf_path}) t0 = time.time() @@ -179,7 +240,19 @@ class InvoiceProcessingService: *, tiff_page_range: Optional[Tuple[int, int]] = None, ) -> Tuple[Dict[str, str], Dict[str, Any]]: - """Devuelve (schema, data) generados a partir del primer documento.""" + """ + Genera un esquema de datos (rutas de campos) basándose en el primer documento procesado. + + Útil para permitir al usuario seleccionar qué campos desea exportar a CSV. + + Args: + pdf_path: Ruta al documento de muestra. + instruction: Instrucción para el modelo. + tiff_page_range: Rango de páginas si es TIFF. + + Returns: + Tupla (esquema_dict, datos_ejemplo_dict). + """ data, _usage = self.extract_full_from_pdf(pdf_path, instruction=instruction, tiff_page_range=tiff_page_range) schema = build_schema(data) try: @@ -197,7 +270,19 @@ class InvoiceProcessingService: tiff_page_range: Optional[Tuple[int, int]] = None, structure_mode: str = "radicados", ) -> Tuple[Dict[str, Dict[str, str]], Dict[str, Dict[str, Any]], str | None]: - """Genera schema y muestra para cada Tipo de Documento buscando el primer archivo disponible.""" + """ + Genera esquemas para múltiples tipos de documentos buscando un ejemplo de cada uno. + + Args: + root_path: Directorio raíz de los documentos. + doc_types: Lista de tipos con sus palabras clave de búsqueda. + instruction: Instrucción para el modelo. + tiff_page_range: Rango de páginas si es TIFF. + structure_mode: 'radicados' o 'general'. + + Returns: + Tupla (esquemas_por_tipo, muestras_por_tipo, primer_radicado_encontrado). + """ if structure_mode == "general": radicados = get_files_as_radicados(root_path) else: @@ -292,10 +377,3 @@ class InvoiceProcessingService: doc_types=doc_types, schema_by_type=schema_by_type, selected_paths=selected_paths, structure_mode=structure_mode ) - - # CSV legacy helper (por si algún test lo llama directamente) - def _append_rows_to_csv(self, csv_path: str, rows: List[Dict[str, Any]], header: List[str]) -> None: - """DEPRECATED: Use CSVManager directly or via BatchProcessor.""" - # Delegar temporalmente a un CSVManager local para mantener compatibilidad sin duplicar código - # o simplemente usar la instancia del batch processor - self.batch_processor.csv_manager.append_rows(csv_path, rows, header) diff --git a/core/pdf_utils.py b/core/pdf_utils.py index 4d7a445..44570b5 100644 --- a/core/pdf_utils.py +++ b/core/pdf_utils.py @@ -32,9 +32,21 @@ def pdf_to_images_bs64( thread_count: int = 10, poppler_path: str | None = None, ) -> List[str]: - """Convierte un PDF a una lista de PNGs en base64. + """ + Convierte un archivo PDF completo a una lista de imágenes PNG en base64. - Usa pdf2image + Poppler. Lanza ValueError si una imagen no valida. + Args: + pdf_path: Ruta al archivo PDF. + validate: Si es True, intenta abrir cada imagen generada para verificar su integridad. + dpi: Resolución de la imagen (puntos por pulgada). + thread_count: Número de hilos para la conversión paralela de páginas. + poppler_path: Ruta personalizada a los binarios de Poppler (opcional). + + Returns: + Lista de strings con las imágenes codificadas en base64. + + Raises: + ValueError: Si la validación de alguna imagen falla. """ poppler_bin = _resolve_poppler(poppler_path) t0 = time.time() @@ -60,7 +72,16 @@ def pdf_to_images_bs64( def get_pdf_num_pages(pdf_path: str, poppler_path: str | None = None) -> int: - """Devuelve el número de páginas de un PDF usando pdfinfo_from_path.""" + """ + Obtiene el número total de páginas de un PDF. + + Args: + pdf_path: Ruta al archivo PDF. + poppler_path: Ruta personalizada a los binarios de Poppler (opcional). + + Returns: + Número de páginas (0 si ocurre un error). + """ info = pdfinfo_from_path(pdf_path, poppler_path=_resolve_poppler(poppler_path)) try: return int(info.get('Pages') or info.get('pages') or 0) @@ -79,9 +100,21 @@ def pdf_pages_to_images_bs64( thread_count: int = 10, poppler_path: str | None = None, ) -> List[str]: - """Convierte un rango [first_page, last_page] (1-indexed, inclusivo) a PNGs base64. + """ + Convierte un rango específico de páginas de un PDF a imágenes base64. - Opcionalmente aplica escala de grises y omite validación para acelerar. + Args: + pdf_path: Ruta al archivo PDF. + first_page: Índice de la primera página (1-indexed). + last_page: Índice de la última página (inclusivo). + dpi: Resolución de la imagen. + grayscale: Si es True, convierte las imágenes a escala de grises para ahorrar ancho de banda/tokens. + validate: Si es True, verifica la integridad de las imágenes generadas. + thread_count: Número de hilos para procesamiento paralelo. + poppler_path: Ruta personalizada a Poppler. + + Returns: + Lista de imágenes en base64. """ poppler_bin = _resolve_poppler(poppler_path) t0 = time.time() @@ -119,12 +152,17 @@ def pdf_pages_to_images_bs64( return resultados def tiff_to_images_bs64(tiff_path: str, validate: bool = True, pages_range: Optional[Tuple[int, int]] = (2, 3)) -> List[str]: - """Convierte un TIFF multipágina a PNGs en base64. + """ + Convierte un archivo TIFF multipágina a una lista de imágenes PNG en base64. - Por defecto, procesa un rango de páginas inclusivo cuando se especifica - `pages_range` como `(inicio, fin)`, usando índices 1-based. Use None para - procesar todas las páginas. Puede pasar `None` en uno de los extremos para - indicar “sin inicio” o “sin fin”. + Args: + tiff_path: Ruta al archivo TIFF. + validate: Si es True, verifica la integridad de las imágenes generadas. + pages_range: Tupla (inicio, fin) de páginas a procesar (1-indexed, inclusivo). + Use None para procesar todas las páginas. + + Returns: + Lista de imágenes en base64. """ resultados: List[str] = [] t0 = time.time() @@ -166,11 +204,22 @@ def document_to_images_bs64( tiff_page_range: Optional[Tuple[int, int]] = None, poppler_path: str | None = None, ) -> List[str]: - """Puerta de entrada: decide conversión según la extensión (.pdf, .tif/.tiff). + """ + Función unificada para convertir documentos (PDF o TIFF) a imágenes base64. - `tiff_page_range` permite indicar un rango explícito (inicio, fin) 1-based - para TIFF. Si es None, se usa `tiff_pages` como valor por defecto. - `poppler_path` se pasa a las funciones PDF; si es None se usa el default del proyecto. + Detecta automáticamente el formato por la extensión del archivo. + + Args: + path: Ruta al archivo. + validate: Si es True, valida las imágenes generadas. + dpi: Resolución para archivos PDF. + thread_count: Hilos paralelos para archivos PDF. + tiff_pages: Páginas TIFF por defecto si tiff_page_range es None. + tiff_page_range: Rango explícito de páginas para TIFF (inicio, fin). + poppler_path: Ruta personalizada a Poppler para PDFs. + + Returns: + Lista de imágenes en base64. """ ext = os.path.splitext(path)[1].lower() if ext == ".pdf": diff --git a/tests/test_llm_smoke.py b/tests/test_llm_smoke.py deleted file mode 100644 index 38e6e94..0000000 --- a/tests/test_llm_smoke.py +++ /dev/null @@ -1,24 +0,0 @@ -from __future__ import annotations - -import pytest - -from application.config import settings -from application.llm_client import LLMClient - - -def test_llm_smoke_returns_ok(): - if not settings.api.api_key: - pytest.skip("Sin API key configurada; se omite prueba de integración.") - client = LLMClient() - text, usage = client.chat_text( - messages=[{"role": "user", "content": "Responde exactamente: ok"}], - component="test_llm_smoke", - ) - assert isinstance(text, str) - assert "ok" in text.lower() - if usage is not None: - assert isinstance(usage, dict) - # No todos los backends devuelven 'total_tokens'; validar presencia básica si existe - assert ("total_tokens" in usage) or ("prompt_tokens" in usage) or ("completion_tokens" in usage) - - diff --git a/tests/test_pdf_utils_tiff_range.py b/tests/test_pdf_utils_tiff_range.py deleted file mode 100644 index f313e1b..0000000 --- a/tests/test_pdf_utils_tiff_range.py +++ /dev/null @@ -1,67 +0,0 @@ -from PIL import Image - -from core.pdf_utils import tiff_to_images_bs64, document_to_images_bs64 - - -def _create_multipage_tiff(path: str, pages: int = 4) -> None: - """Crea un TIFF multipágina de prueba.""" - frames = [] - for i in range(pages): - # Colores distintos por página para facilitar depuración si es necesario - color = (50 * (i + 1) % 255, 0, 0) - frames.append(Image.new("RGB", (16, 16), color)) - frames[0].save(path, save_all=True, append_images=frames[1:], format="TIFF") - - -def test_tiff_to_images_bs64_respects_full_and_sub_ranges(tmp_path) -> None: - tiff_path = tmp_path / "sample.tif" - _create_multipage_tiff(str(tiff_path), pages=4) - - # Sin rango -> todas las páginas - imgs_all = tiff_to_images_bs64(str(tiff_path), validate=True, pages_range=None) - assert len(imgs_all) == 4 - - # Rango 2–3 (1-based, inclusivo) - imgs_2_3 = tiff_to_images_bs64(str(tiff_path), validate=True, pages_range=(2, 3)) - assert len(imgs_2_3) == 2 - - -def test_tiff_to_images_bs64_allows_partial_ranges(tmp_path) -> None: - tiff_path = tmp_path / "sample_partial.tif" - _create_multipage_tiff(str(tiff_path), pages=4) - - # Desde la 3 hasta el final - imgs_from_3 = tiff_to_images_bs64(str(tiff_path), validate=True, pages_range=(3, None)) # type: ignore[arg-type] - assert len(imgs_from_3) == 2 - - # Desde el inicio hasta la 2 - imgs_to_2 = tiff_to_images_bs64(str(tiff_path), validate=True, pages_range=(None, 2)) # type: ignore[arg-type] - assert len(imgs_to_2) == 2 - - # Rango completamente fuera de límites -> lista vacía - imgs_out = tiff_to_images_bs64(str(tiff_path), validate=True, pages_range=(10, 20)) - assert imgs_out == [] - - -def test_document_to_images_bs64_uses_explicit_tiff_page_range_over_default(tmp_path) -> None: - tiff_path = tmp_path / "doc.tif" - _create_multipage_tiff(str(tiff_path), pages=4) - - # Por defecto usa `tiff_pages=(2,3)` del helper - imgs_default = document_to_images_bs64(str(tiff_path), validate=True) - assert len(imgs_default) == 2 - - # tiff_page_range tiene prioridad sobre tiff_pages - imgs_range = document_to_images_bs64( - str(tiff_path), - validate=True, - tiff_pages=(1, 4), - tiff_page_range=(2, 3), - ) - assert len(imgs_range) == 2 - - # Sin rango ni valor por defecto -> todas las páginas - imgs_all = document_to_images_bs64(str(tiff_path), validate=True, tiff_pages=None) - assert len(imgs_all) == 4 - - diff --git a/tests/test_service_tiff_range.py b/tests/test_service_tiff_range.py deleted file mode 100644 index 788025b..0000000 --- a/tests/test_service_tiff_range.py +++ /dev/null @@ -1,72 +0,0 @@ -from __future__ import annotations - -from typing import Any, Dict, List, Optional, Tuple - -from application.service import InvoiceProcessingService -from application.strategies import InvoiceExtractorStrategy -import application.service as service_module - - -class DummyExtractor(InvoiceExtractorStrategy): - """Extractor mínimo para pruebas que devuelve un JSON fijo.""" - def __init__(self, fail_first: bool = False) -> None: - self.fail_first = fail_first - self.calls = 0 - - def extract(self, images_b64: List[str], instruction: str) -> Dict[str, Any]: - self.calls += 1 - if self.fail_first and self.calls == 1: - raise RuntimeError("forced failure") - return { - "raw": '{"ok": true}', - "usage": {}, - } - - -def test_extract_full_from_pdf_propagates_tiff_page_range(monkeypatch, tmp_path) -> None: - dummy_pdf = tmp_path / "doc.tif" - dummy_pdf.write_bytes(b"fake") - - captured: List[Dict[str, Any]] = [] - - def fake_document_to_images_bs64(path: str, **kwargs: Any) -> List[str]: - captured.append({"path": path, **kwargs}) - return ["img"] - - # Parchear el símbolo importado en el módulo de servicio - monkeypatch.setattr(service_module, "document_to_images_bs64", fake_document_to_images_bs64) - - service = InvoiceProcessingService(DummyExtractor()) - data, usage = service.extract_full_from_pdf(str(dummy_pdf), instruction="x", tiff_page_range=(2, 3)) - - assert data == {"ok": True} - assert isinstance(usage, dict) - assert captured, "Se esperaba al menos una llamada a document_to_images_bs64" - # Primera llamada (flujo normal) debe respetar el rango solicitado - assert captured[0].get("tiff_page_range") == (2, 3) - - -def test_extract_full_from_pdf_fallback_ignores_tiff_page_range(monkeypatch, tmp_path) -> None: - dummy_pdf = tmp_path / "doc2.tif" - dummy_pdf.write_bytes(b"fake") - - captured: List[Dict[str, Any]] = [] - - def fake_document_to_images_bs64(path: str, **kwargs: Any) -> List[str]: - captured.append({"path": path, **kwargs}) - return ["img"] - - monkeypatch.setattr(service_module, "document_to_images_bs64", fake_document_to_images_bs64) - - service = InvoiceProcessingService(DummyExtractor(fail_first=True)) - data, usage = service.extract_full_from_pdf(str(dummy_pdf), instruction="x", tiff_page_range=(2, 3)) - - assert data == {"ok": True} - assert isinstance(usage, dict) - # Deben existir dos llamadas: una normal (que falla el extractor) y otra fallback - assert len(captured) == 2 - assert captured[0].get("tiff_page_range") == (2, 3) - # En fallback NO debe propagarse el rango explícito - assert captured[1].get("tiff_page_range") is None - - diff --git a/tests/test_utils_patterns.py b/tests/test_utils_patterns.py deleted file mode 100644 index b157bab..0000000 --- a/tests/test_utils_patterns.py +++ /dev/null @@ -1,43 +0,0 @@ -from core.utils import _matches_pattern, find_first_matching_file - - -def test_matches_pattern_simple_contains(): - assert _matches_pattern("Factura_123.pdf", "factura") - assert _matches_pattern("otro_factura_1.tif", "FACTURA") - assert not _matches_pattern("recibo_123.txt", "factura") - - -def test_matches_pattern_wildcard_without_extension(): - # Patrón sin extensión se aplica sobre el nombre sin extensión - assert _matches_pattern("Factura_1.pdf", "*_1") - assert _matches_pattern("doc_OK1.tif", "doc_*1") - assert not _matches_pattern("Factura_10.pdf", "*_1") # termina en _10 - - -def test_matches_pattern_wildcard_with_extension(): - # Patrón con extensión válida se aplica al nombre completo - assert _matches_pattern("Factura_OK1.pdf", "*_OK?.pdf") - assert not _matches_pattern("Factura_OK1.tif", "*_OK?.pdf") - assert not _matches_pattern("Factura_OK12.pdf", "*_OK?.pdf") # demasiados caracteres - - -def test_matches_pattern_only_allowed_exts(): - # Extensiones que no sean pdf/tif/tiff se ignoran - assert not _matches_pattern("Factura_1.png", "*_1") - - -def test_find_first_matching_file_uses_helper_and_sorts(): - files = [ - "Z_Factura_1.pdf", - "A_Factura_1.pdf", - "B_Factura_2.tif", - "nota.txt", - ] - # Sin comodines: contiene - assert find_first_matching_file(files, "Factura_") == "A_Factura_1.pdf" - # Con comodines sin extensión - assert find_first_matching_file(files, "*_2") == "B_Factura_2.tif" - # Con comodines + extensión explícita - assert find_first_matching_file(files, "*_1.pdf") == "A_Factura_1.pdf" - -