version 001.002 - extracción y creación de campos personalizados, análisis completo del objeto document de Document AI

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__"]
+    default_threshold = default_confidence_override if default_confidence_override is not None else settings.CONFIDENCE_THRESHOLDS.get("__default__", 0.85)
-    full_text_lines = document.text.split('\n')
+
+    # --- 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

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}'")
+    logging.info(f"Analizando línea de importe: '{text}'")
-    normalized_text = " ".join(text.split())
-    logging.info(f"Texto normalizado para la regex: '{normalized_text}'")
-
     result = {'total_amount': None, 'total_tax_amount': None}
     
-    total_match = re.search(r'([\d.,]+)', normalized_text)
+    normalized_text = " ".join(text.split())
+
+    # 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
+    # Regex de alta precisión para el 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.
     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