import os import base64 import json from datetime import datetime from typing import Dict, Any, Optional, Tuple import openai from PIL import Image import io class OCRService: """Servicio para procesamiento OCR de facturas usando OpenAI Vision API""" def __init__(self): self.client = openai.OpenAI() def process_invoice_image(self, image_path: str) -> Tuple[Dict[str, Any], float]: """ Procesa una imagen de factura y extrae los datos relevantes Args: image_path: Ruta de la imagen a procesar Returns: Tuple con (datos_extraidos, confianza) """ try: # Leer y codificar la imagen encoded_image = self._encode_image(image_path) # Crear el prompt para extracción de datos prompt = self._create_extraction_prompt() # Llamar a la API de OpenAI response = self.client.chat.completions.create( model="gpt-4o", messages=[ { "role": "user", "content": [ { "type": "text", "text": prompt }, { "type": "image_url", "image_url": { "url": f"data:image/jpeg;base64,{encoded_image}" } } ] } ], max_tokens=1000, temperature=0.1 ) # Extraer y parsear la respuesta content = response.choices[0].message.content extracted_data = self._parse_ocr_response(content) # Calcular confianza basada en campos completados confidence = self._calculate_confidence(extracted_data) return extracted_data, confidence except Exception as e: print(f"Error en OCR: {str(e)}") return {}, 0.0 def _encode_image(self, image_path: str) -> str: """Codifica una imagen en base64""" try: # Abrir y optimizar la imagen with Image.open(image_path) as img: # Convertir a RGB si es necesario if img.mode != 'RGB': img = img.convert('RGB') # Redimensionar si es muy grande (máximo 2048px) max_size = 2048 if max(img.size) > max_size: img.thumbnail((max_size, max_size), Image.Resampling.LANCZOS) # Guardar en buffer buffer = io.BytesIO() img.save(buffer, format='JPEG', quality=85) buffer.seek(0) # Codificar en base64 return base64.b64encode(buffer.read()).decode('utf-8') except Exception as e: raise Exception(f"Error al codificar imagen: {str(e)}") def _create_extraction_prompt(self) -> str: """Crea el prompt para extracción de datos de facturas dominicanas""" return """ Analiza esta factura dominicana y extrae los siguientes datos en formato JSON. Si algún campo no está visible o no es claro, usa null. IMPORTANTE: Responde ÚNICAMENTE con el JSON, sin texto adicional. { "supplier_name": "Nombre del proveedor/empresa", "supplier_rnc": "RNC del proveedor (formato: 101-12345-6)", "ncf": "Número de Comprobante Fiscal (NCF)", "invoice_date": "Fecha de la factura (formato: YYYY-MM-DD)", "invoiced_amount": "Monto total facturado (número decimal)", "itbis_invoiced": "ITBIS facturado (número decimal)", "itbis_rate": "Tasa de ITBIS aplicada (generalmente 18%)", "subtotal": "Subtotal antes de impuestos", "description": "Descripción de productos/servicios", "expense_type_suggestion": "Tipo de gasto sugerido (01-11 según clasificación DGII)", "confidence_notes": "Notas sobre la calidad de la imagen y legibilidad" } Tipos de gasto DGII para referencia: 01 - Gastos de Personal 02 - Gastos por Trabajos, Suministros y Servicios 03 - Arrendamientos 04 - Gastos de Activos Fijo 05 - Gastos de Representación 06 - Otras Deducciones Admitidas 07 - Gastos Financieros 08 - Gastos Extraordinarios 09 - Compras y Gastos que Formaran Parte del Costo de Venta 10 - Adquisiciones de Activos 11 - Gastos de Seguros Extrae los datos con la mayor precisión posible: """ def _parse_ocr_response(self, content: str) -> Dict[str, Any]: """Parsea la respuesta del OCR y la convierte en un diccionario""" try: # Limpiar la respuesta (remover markdown si existe) content = content.strip() if content.startswith('```json'): content = content[7:] if content.endswith('```'): content = content[:-3] content = content.strip() # Parsear JSON data = json.loads(content) # Validar y limpiar datos cleaned_data = self._clean_extracted_data(data) return cleaned_data except json.JSONDecodeError as e: print(f"Error al parsear JSON: {str(e)}") print(f"Contenido recibido: {content}") return {} except Exception as e: print(f"Error al procesar respuesta OCR: {str(e)}") return {} def _clean_extracted_data(self, data: Dict[str, Any]) -> Dict[str, Any]: """Limpia y valida los datos extraídos""" cleaned = {} # Campos de texto text_fields = ['supplier_name', 'ncf', 'description', 'confidence_notes'] for field in text_fields: value = data.get(field) if value and isinstance(value, str): cleaned[field] = value.strip() # RNC (validar formato) rnc = data.get('supplier_rnc') if rnc and isinstance(rnc, str): # Limpiar RNC (remover espacios, guiones extra) rnc_clean = ''.join(c for c in rnc if c.isdigit() or c == '-') if len(rnc_clean) >= 9: # RNC mínimo válido cleaned['supplier_rnc'] = rnc_clean # Fecha invoice_date = data.get('invoice_date') if invoice_date: try: # Intentar parsear la fecha if isinstance(invoice_date, str): # Intentar diferentes formatos for fmt in ['%Y-%m-%d', '%d/%m/%Y', '%d-%m-%Y']: try: parsed_date = datetime.strptime(invoice_date, fmt) cleaned['invoice_date'] = parsed_date.strftime('%Y-%m-%d') break except ValueError: continue except Exception: pass # Montos (convertir a float) amount_fields = ['invoiced_amount', 'itbis_invoiced', 'subtotal'] for field in amount_fields: value = data.get(field) if value is not None: try: # Limpiar el valor (remover símbolos de moneda, comas) if isinstance(value, str): value = value.replace('$', '').replace(',', '').replace('RD', '').strip() amount = float(value) if amount >= 0: # Solo montos positivos cleaned[field] = amount except (ValueError, TypeError): pass # Tipo de gasto sugerido expense_type = data.get('expense_type_suggestion') if expense_type: # Extraer solo el número (01-11) if isinstance(expense_type, str): expense_num = ''.join(c for c in expense_type if c.isdigit())[:2] if expense_num and 1 <= int(expense_num) <= 11: cleaned['expense_type_suggestion'] = expense_num.zfill(2) # Tasa de ITBIS itbis_rate = data.get('itbis_rate') if itbis_rate is not None: try: if isinstance(itbis_rate, str): itbis_rate = itbis_rate.replace('%', '').strip() rate = float(itbis_rate) if 0 <= rate <= 100: cleaned['itbis_rate'] = rate except (ValueError, TypeError): pass return cleaned def _calculate_confidence(self, data: Dict[str, Any]) -> float: """Calcula el nivel de confianza basado en los datos extraídos""" if not data: return 0.0 # Campos críticos y sus pesos critical_fields = { 'supplier_name': 0.2, 'supplier_rnc': 0.2, 'ncf': 0.2, 'invoice_date': 0.15, 'invoiced_amount': 0.25 } confidence = 0.0 for field, weight in critical_fields.items(): if field in data and data[field] is not None: confidence += weight # Bonus por campos adicionales bonus_fields = ['itbis_invoiced', 'subtotal', 'description'] bonus_per_field = 0.05 for field in bonus_fields: if field in data and data[field] is not None: confidence += bonus_per_field return min(confidence, 1.0) # Máximo 1.0 def classify_expense_type(self, description: str, supplier_name: str = "") -> str: """ Clasifica automáticamente el tipo de gasto basado en la descripción Args: description: Descripción del producto/servicio supplier_name: Nombre del proveedor (opcional) Returns: Código del tipo de gasto (01-11) """ try: prompt = f""" Clasifica el siguiente gasto según los tipos oficiales de la DGII dominicana. Descripción: {description} Proveedor: {supplier_name} Tipos disponibles: 01 - Gastos de Personal 02 - Gastos por Trabajos, Suministros y Servicios 03 - Arrendamientos 04 - Gastos de Activos Fijo 05 - Gastos de Representación 06 - Otras Deducciones Admitidas 07 - Gastos Financieros 08 - Gastos Extraordinarios 09 - Compras y Gastos que Formaran Parte del Costo de Venta 10 - Adquisiciones de Activos 11 - Gastos de Seguros Responde ÚNICAMENTE con el número de dos dígitos (01-11) del tipo más apropiado. """ response = self.client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], max_tokens=10, temperature=0.1 ) result = response.choices[0].message.content.strip() # Validar que sea un número válido if result.isdigit() and 1 <= int(result) <= 11: return result.zfill(2) elif len(result) == 2 and result.isdigit() and 1 <= int(result) <= 11: return result else: return "06" # Default: Otras Deducciones Admitidas except Exception as e: print(f"Error en clasificación automática: {str(e)}") return "06" # Default en caso de error