# services/invoice_processor_service.py import logging import re from typing import Dict, List, Any, Optional from google.cloud.documentai_v1.types import Document from .gcp_document_ai_client import process_document_gcp from .utils import data_cleaner from core.config import settings def _extract_specific_fields( document: Document, default_confidence_override: Optional[float] = None ) -> Dict[str, str]: """ Extrae datos usando una lógica de "Texto Bruto Unificado" para los importes, haciéndolo inmune a errores de formato del OCR como saltos de línea inesperados. """ extracted_data = {field: "Not found or low confidence" for field in settings.REQUIRED_FIELDS} default_threshold = default_confidence_override if default_confidence_override is not None else settings.CONFIDENCE_THRESHOLDS.get("__default__", 0.85) # --- PASO 1: LÓGICA DE BÚSQUEDA POR PATRONES EN TEXTO BRUTO UNIFICADO --- # Normalizamos el texto completo, reemplazando saltos de línea por espacios. full_text_normalized = " ".join(document.text.split()) logging.info(f"Texto completo normalizado para búsqueda: '{full_text_normalized}'") amounts_calculated_by_rule = False # Buscamos el patrón maestro en el texto completo normalizado. # Este patrón busca "Total", un número, y el desglose del IVA entre paréntesis. master_pattern = re.search(r'Total\s*([\d.,]+)€?\s*\(\s*incluye\s+([\d.,]+)€?\s*IVA\s*\)', full_text_normalized, re.IGNORECASE) if master_pattern: logging.info("¡ÉXITO! Patrón maestro de importes encontrado en el texto completo.") total_str_raw = master_pattern.group(1) tax_str_raw = master_pattern.group(2) total_str = data_cleaner.clean_numeric_value(total_str_raw) tax_str = data_cleaner.clean_numeric_value(tax_str_raw) try: total_float = float(total_str) tax_float = float(tax_str) subtotal = total_float - tax_float extracted_data['total_amount'] = f"{total_float:.2f}" extracted_data['total_tax_amount'] = f"{tax_float:.2f}" extracted_data['subtotal_amount'] = f"{subtotal:.2f}" extracted_data['net_amount'] = f"{subtotal:.2f}" amounts_calculated_by_rule = True logging.info("Importes calculados con éxito por regla de negocio.") except (ValueError, TypeError) as e: logging.error(f"Error al convertir importes para el cálculo: {e}") else: logging.warning("Patrón maestro de importes NO encontrado en el texto completo normalizado.") # --- PASO 2: PROCESAR EL RESTO DE ENTIDADES SIMPLES --- for entity in document.entities: entity_type = entity.type_ # Saltamos los campos que ya hemos calculado por reglas o que no nos interesan if entity_type in ['total_amount', 'net_amount', 'subtotal_amount', 'total_tax_amount'] or entity_type not in settings.REQUIRED_FIELDS: continue threshold = settings.CONFIDENCE_THRESHOLDS.get(entity_type, default_threshold) if entity.confidence >= threshold: raw_text = entity.mention_text.strip() if entity_type == 'invoice_date': extracted_data[entity_type] = data_cleaner.normalize_date(raw_text) or f"Unparseable Date: '{raw_text}'" else: extracted_data[entity_type] = raw_text.replace('\n', ' ') # --- PASO 3 (Fallback): Si las reglas fallaron, usamos los valores de las entidades --- if not amounts_calculated_by_rule: logging.warning("No se aplicaron reglas de negocio para importes. Usando entidades como fallback.") for entity in document.entities: if entity.type_ in ['total_amount', 'net_amount', 'subtotal_amount']: if extracted_data.get(entity.type_) == "Not found or low confidence": extracted_data[entity.type_] = data_cleaner.clean_numeric_value(entity.mention_text) return extracted_data def process_invoice_from_bytes( file_bytes: bytes, mime_type: str, default_confidence_override: Optional[float] = None ) -> Dict[str, str]: """ Orquesta el proceso completo. """ try: document = process_document_gcp( project_id=settings.GCP_PROJECT_ID, location=settings.GCP_LOCATION, processor_id=settings.DOCAI_PROCESSOR_ID, file_bytes=file_bytes, mime_type=mime_type, ) validated_data = _extract_specific_fields(document, default_confidence_override) logging.info(f"Datos finales procesados: {validated_data}") return validated_data except Exception as e: logging.error(f"Error en el flujo de procesamiento de factura: {e}", exc_info=True) raise