diff --git a/services/invoice_processor_service.py b/services/invoice_processor_service.py index c8c32e5..1ad217d 100644 --- a/services/invoice_processor_service.py +++ b/services/invoice_processor_service.py @@ -1,68 +1,84 @@ # services/invoice_processor_service.py import logging +import re from typing import Dict, List, Any, Optional -import json # Necesario para formatear el JSON de salida - -# IMPORTAMOS EL TIPO Document y el MessageToJson para la depuración from google.cloud.documentai_v1.types import Document -from google.protobuf.json_format import MessageToJson from .gcp_document_ai_client import process_document_gcp from .utils import data_cleaner from core.config import settings -# --- La función _extract_specific_fields NO necesita cambios en esta fase de depuración --- -# Puedes dejar la versión anterior, ya que el problema está en los datos de entrada que recibe. def _extract_specific_fields( document: Document, default_confidence_override: Optional[float] = None ) -> Dict[str, str]: - # ... (código de la respuesta anterior, no es necesario cambiarlo ahora) + """ + 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["__default__"] - full_text_lines = document.text.split('\n') + 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_ - if entity_type not in settings.REQUIRED_FIELDS or entity_type in ['total_tax_amount', 'subtotal_amount']: + + # 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}'" - elif entity_type == 'total_amount': - contextual_line = None - logging.info(f"Buscando contexto para '{raw_text}' con la palabra clave 'Total'") - for line in full_text_lines: - if raw_text in line and "total" in line.lower(): - contextual_line = line - logging.info(f"Contexto definitivo para total_amount encontrado: '{contextual_line}'") - break - text_to_parse = contextual_line if contextual_line else raw_text - parsed_amounts = data_cleaner.parse_total_and_tax(text_to_parse) - total_str = parsed_amounts.get('total_amount') - tax_str = parsed_amounts.get('total_tax_amount') - if total_str: - extracted_data['total_amount'] = total_str - if tax_str: - extracted_data['total_tax_amount'] = tax_str - try: - subtotal = float(total_str) - float(tax_str) - subtotal_str = f"{subtotal:.2f}" - extracted_data['subtotal_amount'] = subtotal_str - extracted_data['net_amount'] = subtotal_str - except (ValueError, TypeError): - logging.error("Error de conversión para cálculo de subtotal.") - else: - extracted_data['total_tax_amount'] = '0.00' - extracted_data['subtotal_amount'] = total_str - if extracted_data.get('net_amount') == "Not found or low confidence": - extracted_data['net_amount'] = total_str - elif entity_type in ['net_amount', 'subtotal_amount']: - if extracted_data.get(entity_type) == "Not found or low confidence": - extracted_data[entity_type] = data_cleaner.clean_numeric_value(raw_text) else: - extracted_data[entity_type] = raw_text.replace('\n', ' ').strip() + 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 @@ -71,7 +87,7 @@ def process_invoice_from_bytes( mime_type: str, default_confidence_override: Optional[float] = None ) -> Dict[str, str]: - """ Orquesta el proceso completo e imprime la respuesta de GCP para depuración. """ + """ Orquesta el proceso completo. """ try: document = process_document_gcp( project_id=settings.GCP_PROJECT_ID, @@ -80,23 +96,9 @@ def process_invoice_from_bytes( file_bytes=file_bytes, mime_type=mime_type, ) - - # --- INICIO DEL BLOQUE DE DEPURACIÓN --- - # Convertimos la respuesta completa del objeto 'Document' a un JSON legible. - document_json = MessageToJson(document._pb) - # Lo cargamos como un objeto Python para poder formatearlo bonito (indentado). - document_dict = json.loads(document_json) - - # Imprimimos en el log de la consola con un formato claro. - logging.critical("\n\n" + "="*20 + " INICIO RESPUESTA COMPLETA DE DOCUMENT AI " + "="*20) - logging.critical(json.dumps(document_dict, indent=2, ensure_ascii=False)) - logging.critical("="*20 + " FIN RESPUESTA COMPLETA DE DOCUMENT AI " + "="*20 + "\n\n") - # --- FIN DEL BLOQUE DE DEPURACIÓN --- - 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 \ No newline at end of file diff --git a/services/utils/data_cleaner.py b/services/utils/data_cleaner.py index e18a669..710b0a3 100644 --- a/services/utils/data_cleaner.py +++ b/services/utils/data_cleaner.py @@ -28,7 +28,6 @@ def normalize_date(date_string: str) -> Optional[str]: return parsed_date.strftime('%d/%m/%Y') if parsed_date else None # --- FIN DE LA SECCIÓN SIN CAMBIOS --- - def clean_numeric_value(text: str) -> str: """Función pública para limpiar y normalizar un string numérico.""" if not text: return "0.00" @@ -42,55 +41,24 @@ def parse_total_and_tax(text: str) -> Dict[str, Optional[str]]: """ Versión final y robusta. Parsea un string que contiene el total y el IVA. """ - logging.info(f"Texto original recibido para parsing: '{text}'") - normalized_text = " ".join(text.split()) - logging.info(f"Texto normalizado para la regex: '{normalized_text}'") - + logging.info(f"Analizando línea de importe: '{text}'") result = {'total_amount': None, 'total_tax_amount': None} + + normalized_text = " ".join(text.split()) - total_match = re.search(r'([\d.,]+)', normalized_text) + # Regex para el total: busca el primer número después de la palabra "Total" + total_match = re.search(r'total\s*([\d.,]+)', normalized_text, re.IGNORECASE) if total_match: result['total_amount'] = clean_numeric_value(total_match.group(1)) - # Regex Definitiva: más tolerante con el texto que rodea al número del IVA - tax_match = re.search(r'\(.*?(?:incluye|incluido|iva)\s+([\d.,]+).*?\)', normalized_text, re.IGNORECASE) - - if tax_match: - result['total_tax_amount'] = clean_numeric_value(tax_match.group(1)) - logging.info(f"¡ÉXITO! Importe de IVA encontrado y limpiado: {result['total_tax_amount']}") - else: - logging.warning(f"No se encontró desglose de IVA en el texto normalizado: '{normalized_text}'") - - return result - -def parse_total_and_tax_LEGACY(text: str) -> Dict[str, Optional[str]]: - """ - Versión final y robusta. Parsea un string que contiene el total y el IVA. - Primero normaliza los espacios en blanco y luego aplica una regex de alta precisión. - """ - logging.info(f"Texto original recibido para parsing: '{text}'") - - # --- PASO 1: PRE-PROCESAMIENTO Y NORMALIZACIÓN DEL TEXTO --- - # Esto convierte saltos de línea, tabs y espacios múltiples en un único espacio. - # Ej: "398,00€\n (incluye..." -> "398,00€ (incluye..." - normalized_text = " ".join(text.split()) - logging.info(f"Texto normalizado para la regex: '{normalized_text}'") - - result = {'total_amount': None, 'total_tax_amount': None} - - # 2. Extraer el importe total (el primer número que encuentre del texto normalizado) - total_match = re.search(r'([\d.,]+)', normalized_text) - if total_match: - result['total_amount'] = clean_numeric_value(total_match.group(1)) - logging.info(f"Importe total encontrado y limpiado: {result['total_amount']}") - - # 3. Regex de alta precisión aplicada sobre el texto normalizado. + # Regex de alta precisión para el IVA tax_match = re.search(r'\(.*?(?:incluye|incluido)\s+([\d.,]+)€?\s*IVA.*?\)', normalized_text, re.IGNORECASE) - if tax_match: result['total_tax_amount'] = clean_numeric_value(tax_match.group(1)) - logging.info(f"¡ÉXITO! Importe de IVA encontrado y limpiado: {result['total_tax_amount']}") + + if result['total_amount'] and result['total_tax_amount']: + logging.info(f"¡ÉXITO! Total: {result['total_amount']}, IVA: {result['total_tax_amount']}") else: - logging.warning(f"No se encontró desglose de IVA en el texto normalizado: '{normalized_text}'") + logging.warning(f"Fallo en el parsing de importes para la línea: '{normalized_text}'") return result \ No newline at end of file