104 lines
4.7 KiB
Python
104 lines
4.7 KiB
Python
# 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 |