diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..1128a78 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,12 @@ +.git +tests/ +__pycache__ +*.pyc +*.log +.env +output_json/ +logs/ +adjuntos/ +resultado/ +data/ +.DS_Store diff --git a/.gitignore b/.gitignore index eb73e0d..1009f82 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,4 @@ logs/ output_json/ resultado/ .DS_Store +emails.db diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..40cd267 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,16 @@ +FROM python:3.11-slim + +WORKDIR /app + +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +RUN apt-get update \ + && apt-get install -y --no-install-recommends \ + tzdata \ + rsync \ + && rm -rf /var/lib/apt/lists/* + +COPY *.py ./ + +CMD ["python", "main.py"] diff --git a/config.json b/config.json index efbd73e..40ac642 100644 --- a/config.json +++ b/config.json @@ -7,7 +7,7 @@ }, "outlook": { "client_id": "957c324b-3da3-43ed-b912-ca52fb5d3128", - "client_secret": "Guc8Q~u6FH3xBKpxJ-hQ7QSpRWbnvmX.LnWr-dw7", + "client_secret": "iWK8Q~CsGhMnjvsPN_jz3O0fwSYY3SSolfuYycr8", "tenant_id": "76f77b2f-6fc3-42a0-ae83-df05508405d6", "download_folder": "./adjuntos", "verify_ssl": false, @@ -21,7 +21,8 @@ "mount_points": { "windows": "S:", "linux": "/mnt/FileSAP", - "mac": "/Users/george/FileSAP" + "mac": "/Users/george/FileSAP", + "docker": "/mnt/FileSAP" } }, "mails": [ diff --git a/config_loader.py b/config_loader.py index 11ab3ab..367db85 100644 --- a/config_loader.py +++ b/config_loader.py @@ -2,7 +2,7 @@ import json from pathlib import Path from config_model import Config, HanaConfig, OutlookConfig, MailConfig, SmbConfig -######################################################## +# ======================================================================== CONFIG_PATH = Path(__file__).resolve().parent / "config.json" with open(CONFIG_PATH, 'r', encoding='utf-8') as f: @@ -21,4 +21,4 @@ CONFIG = Config( mails=mails ) -######################################################## \ No newline at end of file +# ======================================================================== \ No newline at end of file diff --git a/database_utilities.py b/database_utilities.py index 7d10c44..5321117 100644 --- a/database_utilities.py +++ b/database_utilities.py @@ -2,7 +2,7 @@ import sqlite3 DB_NAME = "emails.db" -######################################################## +# ======================================================================== def init_db(): """ Inicializa la base de datos SQLite y crea la tabla de correos si no existe. @@ -25,7 +25,7 @@ def init_db(): conn.commit() conn.close() -######################################################## +# ======================================================================== def save_emails(emails_list): """ Guarda una lista de correos en la base de datos. @@ -65,7 +65,7 @@ def save_emails(emails_list): conn.close() return rows_affected -######################################################## +# ======================================================================== def get_existing_ids(ids_list): """ Recibe una lista de IDs y devuelve una lista con aquellos que ya existen en la BD. @@ -88,7 +88,7 @@ def get_existing_ids(ids_list): conn.close() return existing_ids -######################################################## +# ======================================================================== def delete_email_by_id(email_id): """ Elimina un correo de la base de datos por su ID. @@ -107,4 +107,37 @@ def delete_email_by_id(email_id): conn.close() return rows_affected > 0 -######################################################## \ No newline at end of file + +# ======================================================================== +def get_ids_by_fecha(fecha): + """ + Retorna una lista de IDs cuya fecha sea >= al parámetro recibido. + Formato esperado de fecha: 'YYYY-MM-DD' o 'YYYY-MM-DD HH:MM:SS' + """ + if not fecha: + return [] + + conn = sqlite3.connect(DB_NAME) + cursor = conn.cursor() + + cursor.execute("SELECT id FROM emails WHERE fecha >= ?", (fecha,)) + ids = [row[0] for row in cursor.fetchall()] + + conn.close() + return ids +# ======================================================================== +if __name__ == "__main__": + fecha_inicio = '2026-07-31' + print(f"\nObteniendo IDs mayores o iguales a {fecha_inicio}:") + + ids_encontrados = get_ids_by_fecha(fecha_inicio) + + if ids_encontrados: + print(f"✓ Se encontraron {len(ids_encontrados)} correos desde esa fecha:") + for i, email_id in enumerate(ids_encontrados, 1): + print(f" {i}. {email_id}") + # delete_email_by_id(email_id) + if len(ids_encontrados) > 10: + print(f" ... y {len(ids_encontrados) - 10} más") + else: + print("✗ No se encontraron correos desde esa fecha.") \ No newline at end of file diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..0bf8e6d --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,17 @@ +services: + descarga-cpe: + build: . + container_name: descarga_adjuntos_cpe + restart: unless-stopped + network_mode: host + environment: + - TZ=America/Lima + volumes: + - ./config.json:/app/config.json:ro + - ./scheduler.json:/app/scheduler.json:ro + - ./data:/app/data + - ./logs:/app/logs + - ./output_json:/app/output_json + - ./adjuntos:/app/adjuntos + - ./resultado:/app/resultado + - /mnt/FileSAP:/mnt/FileSAP diff --git a/mermaid.md b/docs/mermaid.md similarity index 100% rename from mermaid.md rename to docs/mermaid.md diff --git a/document_processor.py b/document_processor.py index 801415a..36a1e86 100644 --- a/document_processor.py +++ b/document_processor.py @@ -2,7 +2,10 @@ import os import xml.etree.ElementTree as ET import pdf_utilities -######################################################## +# Constante única de tipos SUNAT válidos +TIPOS_SUNAT = ["01", "02", "03", "07", "08", "09", "14", "CDR"] + +# ======================================================================== def formatear_serie_numero(serie_num): """ Remueve los ceros a la izquierda de la parte numérica de la serie-número. @@ -19,19 +22,17 @@ def formatear_serie_numero(serie_num): pass return serie_num -######################################################## +# ======================================================================== def generar_nombre_propuesto(m): """ Genera un nombre de archivo propuesto basado en ruc, tipo y serie-numero. - Soporta nombres descriptivos y códigos numéricos de SUNAT. + Retorna None si no hay metadata suficiente para generar un nombre. """ - # Incluimos 99 como tipo permitido para nombres propuestos si se desea - tipos_permitidos = ["01", "03", "07", "08", "02", "09", "14"] - if m['tipo'] in tipos_permitidos and m['ruc'] not in ["Desconocido", "N/A"]: + if m['tipo'] in TIPOS_SUNAT and m['ruc'] not in ["Desconocido", "N/A"]: return f"{m['ruc']}-{m['tipo']}-{m['serie_numero']}" return None -######################################################## +# ======================================================================== def obtener_datos_xml(ruta_xml): """ Analiza un XML de SUNAT para extraer tipo de documento, serie, número y RUC. @@ -101,7 +102,7 @@ def obtener_datos_xml(ruta_xml): except Exception as e: return {"valido": False, "error": str(e)} -######################################################## +# ======================================================================== def procesar_adjuntos_enriquecidos(email_obj, ruta_base): """ Recorre los archivos filtrados de un email y extrae su metadata (XML o PDF). @@ -111,8 +112,8 @@ def procesar_adjuntos_enriquecidos(email_obj, ruta_base): archivos = email_obj.get('archivos', []) metadatos_adjuntos = [] - # Tipos permitidos oficialmente (incluimos CDR por requerimiento) - tipos_ok = ["01", "02", "03", "07", "08", "CDR"] + # Usar la constante global de tipos SUNAT + tipos_ok = TIPOS_SUNAT xmls = [a for a in archivos if a.lower().endswith('.xml')] pdfs = [a for a in archivos if a.lower().endswith('.pdf')] @@ -145,7 +146,9 @@ def procesar_adjuntos_enriquecidos(email_obj, ruta_base): "tipo": tipo_final, "serie_numero": formatear_serie_numero(datos['id']), "ruc": datos['ruc'], - "fuente": "PDF" + "fuente": "PDF", + "llm_usado": datos.get("llm_usado", False), + "llm_respuesta": datos.get("llm_respuesta") } m["archivo_propuesto"] = generar_nombre_propuesto(m) metadatos_adjuntos.append(m) @@ -157,9 +160,10 @@ def procesar_adjuntos_enriquecidos(email_obj, ruta_base): "serie_numero": "N/A", "ruc": "N/A", "fuente": "PDF_FALLIDO", - "archivo_propuesto": None + "archivo_propuesto": None, + "llm_usado": False, + "llm_respuesta": None }) return metadatos_adjuntos - - ######################################################## \ No newline at end of file +# ======================================================================== \ No newline at end of file diff --git a/email_services.py b/email_services.py index 5ee95f7..dc34df2 100644 --- a/email_services.py +++ b/email_services.py @@ -2,20 +2,19 @@ import os import zipfile import database_utilities -######################################################## +# ======================================================================== def filtrar_correos_nuevos(emails): """ - Verifica cuáles correos son nuevos (no están en la BD), los guarda - y retorna la lista filtrada de los que tienen adjuntos. + Verifica cuáles correos son nuevos (no están en la BD), los guarda en la BD + y retorna la lista completa de nuevos (con o sin adjuntos). + Todos se registran en DB para no reprocesarlos en futuras ejecuciones. """ if not emails: return [] - # 1. Verificar cuáles ya existen en la BD ids = [e['id'] for e in emails] existentes = database_utilities.get_existing_ids(ids) - # Solo nos quedamos con los que NO están en la BD nuevos = [e for e in emails if e['id'] not in existentes] print(f"Correos nuevos detectados: {len(nuevos)}") @@ -23,13 +22,9 @@ def filtrar_correos_nuevos(emails): print("Guardando nuevos correos en la base de datos...") database_utilities.save_emails(nuevos) - # 2. Filtrar para quedarse solo con los que tienen adjuntos - con_adjuntos = [e for e in nuevos if e['has_attach']] - print(f"Correos nuevos con adjuntos: {len(con_adjuntos)}") - - return con_adjuntos + return nuevos -######################################################## +# ======================================================================== def descomprimir_adjuntos(archivos, ruta): """ Descomprime archivos .zip en la ruta especificada y actualiza la lista de archivos. @@ -51,7 +46,7 @@ def descomprimir_adjuntos(archivos, ruta): archivos_finales.append(archivo) return archivos_finales -######################################################## +# ======================================================================== def filtrar_archivos_validos(archivos): """ Retorna una lista solo con archivos .pdf y .xml (ignora mayúsculas/minúsculas). @@ -59,7 +54,7 @@ def filtrar_archivos_validos(archivos): extensiones_validas = ('.pdf', '.xml') return [a for a in archivos if a.lower().endswith(extensiones_validas)] -######################################################## +# ======================================================================== def procesar_adjuntos_email(e, archivos, ruta_descarga): """ Descomprime y filtra los archivos de un correo. @@ -76,5 +71,5 @@ def procesar_adjuntos_email(e, archivos, ruta_descarga): return archivos_validos return [] -######################################################## +# ======================================================================== diff --git a/emails.db b/emails.db index af80789..242b42f 100644 Binary files a/emails.db and b/emails.db differ diff --git a/file_manager.py b/file_manager.py index 9833773..c625460 100644 --- a/file_manager.py +++ b/file_manager.py @@ -1,188 +1,132 @@ import os +import shutil -######################################################## -def calcular_puntuacion(nombre_archivo, ruc, tipo, serie_numero): - """ - Calcula una puntuación de 0 a 4 según cuántos elementos coinciden en el nombre del archivo. - serie_numero se descompone en serie y número. - """ - puntos = 0 - nombre_archivo = nombre_archivo.upper() - - # 1. RUC - if ruc and ruc != "N/A" and ruc in nombre_archivo: - puntos += 1 - - # 2. Tipo - if tipo and tipo != "99" and tipo in nombre_archivo: - puntos += 1 - - # 3. Serie y 4. Número - if "-" in serie_numero: - partes = serie_numero.split("-") - serie = partes[0].upper() - numero = partes[1] - - if serie in nombre_archivo: - puntos += 1 - if numero in nombre_archivo: - puntos += 1 - - return puntos +# ======================================================================== +def tiene_datos_parciales(m): + """True si el archivo tiene al menos un campo útil, aunque incompleto.""" + if m['ruc'] not in ("N/A", "Desconocido"): + return True + if m['serie_numero'] not in ("No encontrado", "N/A"): + return True + if m['tipo'] != "99": + return True + return False -######################################################## -def procesar_renombrado_agrupado(email_obj, ruta_base): +# ======================================================================== +def renombrar_archivos_por_metadata(metadatos, ruta_base): """ - Lógica de agrupación y renombrado de archivos según puntuación y tipo. + Renombra cada archivo según su propia metadata extraída. + - XML/PDF con archivo_propuesto válido: se renombra a {archivo_propuesto}{extension} + - CDR (tipo 'CDR'): se renombra igual (archivo_propuesto ya incluye CDR) + - Sin metadata válida (archivo_propuesto is None): se deja intacto (será eliminado al limpiar) + Retorna la lista de metadatos actualizada con los nombres finales. """ - metadatos = email_obj.get('metadatos_adjuntos', []) - tipos_permitidos = ["01", "03", "07", "08", "02", "09", "14"] - - # 1. Identificar "Pivotes" (pivotes para agrupar) - # Preferencia: XML Maestros. Fallback: PDFs con metadata válida si no hay XMLs. - xml_maestros = [m for m in metadatos if m['extension'] == ".xml" and m['tipo'] in tipos_permitidos] - - if xml_maestros: - pivotes = xml_maestros - else: - # En correos solo PDF, tratamos a cada PDF con nombre propuesto como un pivote - pivotes = [m for m in metadatos if m['extension'] == ".pdf" and m['archivo_propuesto'] is not None] - - if not pivotes: - return {} # Nada que agrupar - - nuevos_archivos = [] - mapeo_renombrado = {} # original -> nuevo - agrupacion = {} # archivo_propuesto -> [lista de nombres nuevos] - - # Inicializar el diccionario de agrupación con los nombres de los pivotes - for p in pivotes: - agrupacion[p['archivo_propuesto']] = [] - - # 2. Agrupar cada archivo con su mejor pivote for m in metadatos: nombre_original = m['archivo'] - extension = m['extension'].lower() - - # Si es un pivote, su nombre es directo y ya está inicializado - if m in pivotes: - nombre_nuevo = f"{m['archivo_propuesto']}{extension}" - mapeo_renombrado[nombre_original] = nombre_nuevo - agrupacion[m['archivo_propuesto']].append(nombre_nuevo) - else: - # Buscar el mejor match entre los pivotes que aún acepten este tipo de archivo - mejor_pivote = None - max_puntos = -1 - - is_cdr = m['tipo'] == "CDR" or "R-" in nombre_original or "R_" in nombre_original or "CDR" in nombre_original.upper() - is_pdf = extension == ".pdf" - is_xml = extension == ".xml" and not is_cdr - - for p in pivotes: - p_prop = p['archivo_propuesto'] - grupo_actual = agrupacion[p_prop] - - ya_tiene_pdf = any(f.lower().endswith('.pdf') for f in grupo_actual) - ya_tiene_cdr = any(f.startswith('CDR-') for f in grupo_actual) - - if (is_pdf and ya_tiene_pdf) or (is_cdr and ya_tiene_cdr) or is_xml: - continue - - puntos = calcular_puntuacion( - nombre_original, - p['ruc'], - p['tipo'], - p['serie_numero'] - ) - - if puntos > max_puntos: - max_puntos = puntos - mejor_pivote = p - - if mejor_pivote and max_puntos > 0: - prefix = "CDR-" if is_cdr else "" - nombre_nuevo = f"{prefix}{mejor_pivote['archivo_propuesto']}{extension}" - - mapeo_renombrado[nombre_original] = nombre_nuevo - agrupacion[mejor_pivote['archivo_propuesto']].append(nombre_nuevo) - else: - # Si no hay match, va a OTROS - mapeo_renombrado[nombre_original] = nombre_original - if "OTROS" not in agrupacion: agrupacion["OTROS"] = [] - agrupacion["OTROS"].append(nombre_original) + extension = m['extension'] + propuesto = m.get('archivo_propuesto') - # 3. Ejecutar el renombrado físico y actualizar metadatos - for m in metadatos: - nombre_original = m['archivo'] - nombre_nuevo = mapeo_renombrado.get(nombre_original) - - if nombre_nuevo and nombre_nuevo != nombre_original: - ruta_old = os.path.join(ruta_base, nombre_original) - ruta_new = os.path.join(ruta_base, nombre_nuevo) - - try: - if os.path.exists(ruta_old): - # Si el destino ya existe, borramos para sobreescribir (o podrías manejarlo de otra forma) - if os.path.exists(ruta_new) and ruta_old != ruta_new: - os.remove(ruta_new) - os.rename(ruta_old, ruta_new) - m['archivo'] = nombre_nuevo - except Exception as e: - print(f" ⚠️ Error renombrando {nombre_original}: {e}") + if propuesto is None: + continue - # 4. Actualizar la lista de archivos en el email_obj - email_obj['archivos'] = [m['archivo'] for m in metadatos] - return agrupacion + nuevo_nombre = f"{propuesto}{extension}" + if nuevo_nombre == nombre_original: + continue -######################################################## -def mover_archivos_a_resultado(email_obj, ruta_origen, ruta_destino): + ruta_old = os.path.join(ruta_base, nombre_original) + ruta_new = os.path.join(ruta_base, nuevo_nombre) + + try: + if os.path.exists(ruta_old): + if os.path.exists(ruta_new) and ruta_old != ruta_new: + os.remove(ruta_new) + os.rename(ruta_old, ruta_new) + m['archivo'] = nuevo_nombre + except Exception as e: + print(f" ⚠️ Error renombrando '{nombre_original}': {e}") + + return metadatos + +# ======================================================================== +def mover_validos_a_resultado(metadatos, ruta_origen, ruta_destino): """ - Mueve los archivos procesados de la carpeta temporal a la carpeta final de resultados. - Solo mueve archivos que pertenecen a un grupo válido (basado en archivo_propuesto). - Excluye los archivos en la categoría 'OTROS'. + Mueve solo los archivos que tienen metadata válida (archivo_propuesto no None) + desde la carpeta de descarga a la carpeta de resultados. """ - agrupacion = email_obj.get('agrupacion_archivos', {}) - if not agrupacion: - return - if not os.path.exists(ruta_destino): os.makedirs(ruta_destino) - print(f" 📁 Carpeta de destino creada: {ruta_destino}") - for key, archivos in agrupacion.items(): - if key == "OTROS": - continue # No mover archivos no identificados - - for archivo in archivos: - ruta_old = os.path.join(ruta_origen, archivo) - ruta_new = os.path.join(ruta_destino, archivo) - - try: - if os.path.exists(ruta_old): - if os.path.exists(ruta_new) and ruta_old != ruta_new: - os.remove(ruta_new) - os.rename(ruta_old, ruta_new) - except Exception as e: - print(f" ⚠️ Error trasladando {archivo} a resultado: {e}") + movidos = 0 + for m in metadatos: + if m.get('archivo_propuesto') is None: + continue -######################################################## + archivo = m['archivo'] + ruta_old = os.path.join(ruta_origen, archivo) + ruta_new = os.path.join(ruta_destino, archivo) + + try: + if os.path.exists(ruta_old): + if os.path.exists(ruta_new) and ruta_old != ruta_new: + os.remove(ruta_new) + os.rename(ruta_old, ruta_new) + movidos += 1 + except Exception as e: + print(f" ⚠️ Error trasladando '{archivo}' a resultado: {e}") + + if movidos: + print(f" 📦 {movidos} archivo(s) movido(s) a resultado.") + +# ======================================================================== +def mover_a_evaluar(metadatos, ruta_origen, ruta_evaluar): + """ + Mueve archivos con metadata parcial a la carpeta evaluar/ para revisión. + Conserva el nombre original del archivo (sin renombrar). + """ + if not metadatos: + return + + if not os.path.exists(ruta_evaluar): + os.makedirs(ruta_evaluar) + + movidos = 0 + for m in metadatos: + archivo = m['archivo'] + ruta_old = os.path.join(ruta_origen, archivo) + ruta_new = os.path.join(ruta_evaluar, archivo) + + try: + if os.path.exists(ruta_old): + if os.path.exists(ruta_new) and ruta_old != ruta_new: + os.remove(ruta_new) + os.rename(ruta_old, ruta_new) + movidos += 1 + except Exception as e: + print(f" ⚠️ Error moviendo '{archivo}' a evaluar: {e}") + + if movidos: + print(f" 🔍 {movidos} archivo(s) movido(s) a evaluar (metadata parcial).") + +# ======================================================================== def limpiar_carpeta_descarga(ruta): """ - Elimina todos los archivos dentro de la carpeta de descarga para dejarla limpia. + Elimina todos los archivos y subcarpetas dentro de la carpeta de descarga. + Omite la subcarpeta 'evaluar/' para no borrar archivos pendientes de revisión. """ if not os.path.exists(ruta): return - - print(f" 🧹 Limpiando carpeta de descarga: {ruta}") + + print(f" 🧹 Limpiando carpeta de descarga") for archivo in os.listdir(ruta): + if archivo == "evaluar": + continue ruta_archivo = os.path.join(ruta, archivo) try: if os.path.isfile(ruta_archivo): os.remove(ruta_archivo) elif os.path.isdir(ruta_archivo): - import shutil shutil.rmtree(ruta_archivo) except Exception as e: - print(f" ⚠️ No se pudo eliminar {archivo}: {e}") + print(f" ⚠️ No se pudo eliminar '{archivo}': {e}") -######################################################## \ No newline at end of file +# ======================================================================== diff --git a/llm_processor.py b/llm_processor.py new file mode 100644 index 0000000..8cb13c8 --- /dev/null +++ b/llm_processor.py @@ -0,0 +1,50 @@ +import json +import requests + +# ======================================================================== +OLLAMA_URL = "http://localhost:11434/api/chat" +MODEL = "qwen3:8b" +CONTADOR_LLM = 0 + +_PROMPT_TEMPLATE = """Extrae los siguientes datos del comprobante SUNAT (factura, boleta, nota de credito, etc.) a partir del texto extraido del PDF. + +Texto del PDF: +{texto} + +Devuelve SOLO un JSON valido con estos 4 campos: +- ruc: string de 11 digitos (ej: 20101420320) +- tipo: string del tipo de documento segun codigo SUNAT (01=Factura, 03=Boleta, 07=Nota Credito, 08=Nota Debito, 09=Guia Remision, 14=Recibo Honorarios, 02=Recibo por Honorarios) +- serie: string (ej: F001, B001, E001, EG07) +- numero: string (ej: 00050193, 5103) + +JSON:""" + +# ======================================================================== +def extraer_con_llm(texto): + global CONTADOR_LLM + CONTADOR_LLM += 1 + try: + resp = requests.post( + OLLAMA_URL, + json={ + "model": MODEL, + "messages": [ + {"role": "system", "content": "Eres un extractor de datos de comprobantes SUNAT. Solo respondes con JSON valido."}, + {"role": "user", "content": _PROMPT_TEMPLATE.format(texto=texto[:8000])} + ], + "format": "json", + "stream": False, + "options": {"temperature": 0} + }, + timeout=180 + ) + data = resp.json() + content = data.get("message", {}).get("content", "") + return json.loads(content) + except json.JSONDecodeError: + return {} + except requests.ConnectionError: + return {} + except Exception: + return {} +# ======================================================================== \ No newline at end of file diff --git a/logger_utilities.py b/logger_utilities.py index 6d05b9e..c61e100 100644 --- a/logger_utilities.py +++ b/logger_utilities.py @@ -2,7 +2,7 @@ import sys import os from datetime import datetime, timedelta -######################################################## +# ======================================================================== class MultiStream: """ Clase para redirigir la salida a múltiples flujos (consola y archivo). @@ -19,7 +19,7 @@ class MultiStream: for stream in self.streams: stream.flush() -######################################################## +# ======================================================================== def setup_logger(): """ Configura la redirección de stdout y stderr a un archivo de log y a la consola. @@ -43,7 +43,7 @@ def setup_logger(): return log_path -######################################################## +# ======================================================================== def limpiar_logs_antiguos(dias=7): """ Elimina archivos de log con más de 'n' días de antigüedad. @@ -68,3 +68,4 @@ def limpiar_logs_antiguos(dias=7): if count > 0: print(f"🧹 Se eliminaron {count} archivos de log antiguos (más de {dias} días).") +# ======================================================================== \ No newline at end of file diff --git a/main.py b/main.py index 15da35a..edf7c2f 100644 --- a/main.py +++ b/main.py @@ -1,172 +1,293 @@ import sys import json import os +import time +import signal +from datetime import datetime, timedelta + import outlook_utilities import database_utilities import email_services -import logger_utilities import document_processor import file_manager -from datetime import datetime, timedelta +import llm_processor +from smb_manager import sincronizar_a_smb +from logger_utilities import setup_logger, limpiar_logs_antiguos from config_loader import CONFIG -######################################################## -if __name__ == "__main__": +# ============================================================================ +# CONFIGURACIÓN DE HORARIOS (cargado desde scheduler.json) +# ============================================================================ +_SCHEDULER_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "scheduler.json") +with open(_SCHEDULER_PATH, "r", encoding="utf-8") as _f: + _sched = json.load(_f) - # Configurar Logger (Redirige stdout a archivo y consola) - log_path = logger_utilities.setup_logger() - logger_utilities.limpiar_logs_antiguos(dias=30) +HORARIOS = [tuple(h) for h in _sched["HORARIOS"]] +DIAS_ACTIVOS = _sched["DIAS_ACTIVOS"] - inicio_ejecucion = datetime.now() - print("="*60) - print(f"🚀 INICIO DE EJECUCIÓN: {inicio_ejecucion.strftime('%Y-%m-%d %H:%M:%S')}") - print(f"📄 Log guardado en: {log_path}") - print("="*60) +_running = True + +# ─── Colores ──────────────────────────────────────────── +color_verde = "\033[92m" +color_amarillo = "\033[93m" +color_rojo = "\033[91m" +color_fin = "\033[0m" + +# ======================================================================== +def ic(ok, texto_ok, texto_fail): + icono = "✅" if ok else "⚠️" + color = color_verde if ok else color_amarillo + return f"{icono} {color}{texto_ok if ok else texto_fail}{color_fin}" + +# ======================================================================== +def obtener_intervalo(dt): + minutos = dt.hour * 60 + dt.minute + for hi, mi, hf, mf, intervalo, modo in HORARIOS: + inicio = hi * 60 + mi + fin = hf * 60 + mf + if inicio <= minutos < fin: + return (intervalo, modo) + return None + +# ======================================================================== +def segundos_hasta_proximo_inicio(dt): + target = dt.replace(hour=6, minute=30, second=0, microsecond=0) + if dt >= target: + target += timedelta(days=1) + return int((target - dt).total_seconds()) + +# ======================================================================== +def segundos_hasta_proximo_dia_activo(dt): + for dias in range(1, 8): + prox = dt + timedelta(days=dias) + if prox.weekday() in DIAS_ACTIVOS: + target = prox.replace(hour=6, minute=30, second=0, microsecond=0) + return int((target - dt).total_seconds()) + return 86400 + +# ======================================================================== +def obtener_token(outlook_token, fecha_obtencion): + """Reusa el token si fue obtenido hace menos de 50 minutos, si no renueva.""" + if outlook_token and fecha_obtencion: + delta = datetime.now() - fecha_obtencion + if delta.total_seconds() < 3000: # ~50 min (token MSAL dura 60 min) + return outlook_token + + print("\n🔑 Obteniendo nuevo token de acceso...") + nuevo = outlook_utilities.get_access_token() + print(f" {ic(True, 'Token obtenido.', '')}") + return nuevo + +# ======================================================================== +def procesar_cuenta(mail, outlook_token, fecha_inicio, fecha_fin): + """Procesa todos los correos nuevos de una cuenta.""" + print(f"\n{'='*60}") + print(f"📧 [{mail.email}]") + + emails = outlook_utilities.get_emails(mail.email, outlook_token, fecha_inicio, fecha_fin) + print(f" 📬 {len(emails)} correos en el rango.") + + correos_nuevos = email_services.filtrar_correos_nuevos(emails) + if not correos_nuevos: + print(f" {ic(True, 'Sin correos nuevos.', '')}") + return [] + + print(f" 🆕 {len(correos_nuevos)} nuevos (no procesados antes).") + procesados = [] + + for e in correos_nuevos: + if not _running: + if llm_processor.CONTADOR_LLM > 0: + print(f" 🤖 LLM invocado {llm_processor.CONTADOR_LLM} vez/veces (modelo: {llm_processor.MODEL})") + return procesados + if not e['has_attach']: + continue + + try: + archivos_orig = outlook_utilities.download_attachments( + mail.email, e['id'], outlook_token, mail.ruta_descarga + ) + if not archivos_orig: + continue + + print(f"\n ├── 📬 [{e['fecha']}] {e['subject'][:80]}...") + + archivos_validos = email_services.procesar_adjuntos_email( + e, archivos_orig.copy(), mail.ruta_descarga + ) + if not archivos_validos: + print(f" │ {ic(False, '', 'Descartado: sin PDF/XML.')}") + file_manager.limpiar_carpeta_descarga(mail.ruta_descarga) + continue + + metadatos = document_processor.procesar_adjuntos_enriquecidos(e, mail.ruta_descarga) + + for a in archivos_orig: + if a.lower().endswith('.zip'): + print(f" │ 📦 Descomprimido: {a}") + + # Clasificar archivos por completitud de metadata + completos = [m for m in metadatos if m.get('archivo_propuesto') is not None] + parciales = [m for m in metadatos if m.get('archivo_propuesto') is None and file_manager.tiene_datos_parciales(m)] + vacios = [m for m in metadatos if m.get('archivo_propuesto') is None and not file_manager.tiene_datos_parciales(m)] + + if not completos and not parciales: + print(f" │ {ic(False, '', 'Descartado: ningun archivo tiene datos utiles.')}") + file_manager.limpiar_carpeta_descarga(mail.ruta_descarga) + continue + + # Renombrar y mover completos a resultado + if completos: + file_manager.renombrar_archivos_por_metadata(completos, mail.ruta_descarga) + file_manager.mover_validos_a_resultado(completos, mail.ruta_descarga, mail.ruta_final) + + # Mover parciales a evaluar para revisión manual + if parciales: + ruta_evaluar = os.path.join(mail.ruta_descarga, "evaluar") + file_manager.mover_a_evaluar(parciales, mail.ruta_descarga, ruta_evaluar) + + # Limpiar lo que queda (vacíos + sobrantes) + file_manager.limpiar_carpeta_descarga(mail.ruta_descarga) + + for m in metadatos: + if m.get('archivo_propuesto') is not None: + if m.get('llm_usado'): + llm_resp = m.get('llm_respuesta') or {} + print(f" │ 🤖 {m['archivo']} -> [{m['tipo']}] RUC:{m['ruc']} S/N:{m['serie_numero']} | LLM: {llm_resp}") + else: + icono = "📕" if m['extension'] == ".pdf" else "🌐" + print(f" │ {icono} {m['archivo']} -> [{m['tipo']}] RUC:{m['ruc']} S/N:{m['serie_numero']}") + elif file_manager.tiene_datos_parciales(m): + print(f" │ 🔍 {m['archivo']} (metadata parcial -> evaluar/)") + else: + print(f" │ ❌ {m['archivo']} (sin datos utiles, eliminado)") + + e['metadatos_adjuntos'] = metadatos + procesados.append(e) + + except Exception as ex: + print(f" │ {ic(False, '', f'Error: {ex}')}") + file_manager.limpiar_carpeta_descarga(mail.ruta_descarga) + + if llm_processor.CONTADOR_LLM > 0: + print(f" 🤖 LLM invocado {llm_processor.CONTADOR_LLM} vez/veces en esta ejecucion (modelo: {llm_processor.MODEL})") + + return procesados + +# ======================================================================== +def guardar_json(todos_procesados): + if not todos_procesados: + return + + output_dir = "output_json" + if not os.path.exists(output_dir): + os.makedirs(output_dir) + + timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") + json_path = os.path.join(output_dir, f"{timestamp}.json") + + with open(json_path, "w", encoding="utf-8") as f: + json.dump(todos_procesados, f, ensure_ascii=False, indent=4) + + print(f"\n💾 Resultados exportados a: {json_path}") + +# ======================================================================== +def signal_handler(signum, frame): + global _running + print("\n\n⏹️ Senal de parada recibida. Deteniendo...") + _running = False + +# ======================================================================== +def main(): + global _running + signal.signal(signal.SIGINT, signal_handler) + signal.signal(signal.SIGTERM, signal_handler) + + log_path = setup_logger() + limpiar_logs_antiguos(dias=30) + + print("=" * 60) + print("🚀 DAEMON DESCARGA ADJUNTOS CORREOS CPE") + print("=" * 60) + print(f"\n📅 Dias activos: {DIAS_ACTIVOS} (0=lun, 6=dom)") + print(f"⏰ Horarios:") + for hi, mi, hf, mf, intervalo, modo in HORARIOS: + print(f" {hi:02d}:{mi:02d} - {hf:02d}:{mf:02d} | cada {intervalo}s | {modo}") + print(f"\n📄 Log: {log_path}") - # Inicializar Base de Datos database_utilities.init_db() - # Calcular fechas hoy = datetime.now() - ayer = hoy - timedelta(days=1) - + ayer = hoy - timedelta(days=5) fecha_inicio = ayer.strftime("%Y-%m-%d") fecha_fin = hoy.strftime("%Y-%m-%d") - print(f"Rango de búsqueda: {fecha_inicio} al {fecha_fin}") + outlook_token = None + fecha_obtencion_token = None - try: - print("Intentando obtener token de acceso...") - outlook_token = outlook_utilities.get_access_token() - print("✅ Token obtenido exitosamente.") - print(f"Token prefix: {outlook_token[:15]}...") - except Exception as e: - print(f"❌ Error: {e}") - sys.exit(1) + while _running: + ahora = datetime.now() - todos_procesados = [] + if ahora.weekday() not in DIAS_ACTIVOS: + espera = segundos_hasta_proximo_dia_activo(ahora) + prox = ahora + timedelta(seconds=espera) + print(f"\n⏸️ Dia no activo. Proximo inicio: {prox.strftime('%a %Y-%m-%d %H:%M:%S')}") + for _ in range(int(espera)): + if not _running: + break + time.sleep(1) + continue - for mail in CONFIG.mails: - print("\n" + "="*60) - print(f"[MAIL] {mail.email}") + intervalo_info = obtener_intervalo(ahora) + if not intervalo_info: + espera = segundos_hasta_proximo_inicio(ahora) + prox = ahora + timedelta(seconds=espera) + print(f"\n⏸️ Fuera de horario. Proximo inicio: {prox.strftime('%H:%M:%S')}") + for _ in range(int(espera)): + if not _running: + break + time.sleep(1) + continue - emails = outlook_utilities.get_emails(mail.email, outlook_token, fecha_inicio, fecha_fin) - print(f"Se encontraron {len(emails)} correos.") + intervalo, modo = intervalo_info - # Filtrar y guardar usando el servicio de emails - emails_a_procesar = email_services.filtrar_correos_nuevos(emails) + print(f"\n{'#'*60}") + print(f"# 🔄 CICLO: {ahora.strftime('%Y-%m-%d %H:%M:%S')} | modo={modo} | intervalo={intervalo}s") + print(f"{'#'*60}") - if emails_a_procesar: - print(f"Descargando adjuntos en: {mail.ruta_descarga}") - emails_finales = [] # Lista definitiva de correos con adjuntos válidos + try: + outlook_token = obtener_token(outlook_token, fecha_obtencion_token) + if not fecha_obtencion_token: + fecha_obtencion_token = datetime.now() + except Exception as e: + print(f" {ic(False, '', f'Error obteniendo token: {e}')}") + time.sleep(60) + continue - for e in emails_a_procesar: - try: - # Guardar lista original para el log - archivos_orig = outlook_utilities.download_attachments( - mail.email, - e['id'], - outlook_token, - mail.ruta_descarga - ) - - if archivos_orig: - # Encabezado del correo con fecha - print(f"\n📬 [{e['fecha']}] {e['subject'][:80]}...") - - # --- SECCIÓN 1: ADJUNTOS ORIGINALES --- - print(f" ├── 📂 Adjuntos originales:") - for idx, a in enumerate(archivos_orig): - prefijo = " │ └── " if idx == len(archivos_orig)-1 else " │ ├── " - print(f"{prefijo}{a}") + todos_procesados = [] - # 1. Procesar adjuntos (unzip + filtro) - # Usamos una copia de la lista original - archivos_unzipped = email_services.procesar_adjuntos_email(e, archivos_orig.copy(), mail.ruta_descarga) - - if archivos_unzipped: - # 2. Extraer metadatos iniciales (antes del renombrado) - metadatos = document_processor.procesar_adjuntos_enriquecidos(e, mail.ruta_descarga) - e['metadatos_adjuntos'] = metadatos - - # 3. Aplicar lógica de agrupación y renombrado - # Esto modifica e['metadatos_adjuntos'] in-place - agrupacion = file_manager.procesar_renombrado_agrupado(e, mail.ruta_descarga) - e['agrupacion_archivos'] = agrupacion - - # --- SECCIÓN 2: PROCESO --- - print(f" ├── ⚙️ Proceso:") - logs_proceso = [] - - # Detectar ZIPs procesados - zips = [a for a in archivos_orig if a.lower().endswith('.zip')] - for z in zips: - logs_proceso.append(f"📦 Descomprimido: {z}") - - # Detectar Renombrados (si el nombre final no está en los unzipped filtrados) - for m in e['metadatos_adjuntos']: - if m['archivo'] not in archivos_unzipped: - logs_proceso.append(f"🏷️ Renombrado a: {m['archivo']}") + for mail in CONFIG.mails: + if not _running: + break + procesados = procesar_cuenta(mail, outlook_token, fecha_inicio, fecha_fin) + if procesados: + sincronizar_a_smb(mail.ruta_final, mail.smb_folder) + print(f"\n✅ {mail.email}: {len(procesados)} correos procesados.") + else: + print(f"\nℹ️ {mail.email}: sin correos para procesar.") + todos_procesados.extend(procesados) - if not logs_proceso: - print(f" │ └── Sin cambios especiales") - else: - for idx, log in enumerate(logs_proceso): - prefijo = " │ └── " if idx == len(logs_proceso)-1 else " │ ├── " - print(f"{prefijo}{log}") + guardar_json(todos_procesados) - # 4. Mover archivos a carpeta de resultados final - file_manager.mover_archivos_a_resultado(e, mail.ruta_descarga, mail.ruta_final) - - # --- SECCIÓN 3: RESULTADO FINAL --- - print(f" └── 📄 Resultado Final:") - for idx, m in enumerate(e['metadatos_adjuntos']): - prefijo = " └── " if idx == len(e['metadatos_adjuntos'])-1 else " ├── " - icono = "📕" if m['extension'] == ".pdf" else "🌐" - info_extra = f"[{m['tipo']}] {m['serie_numero']} - RUC: {m['ruc']}" - print(f"{prefijo}{icono} {m['archivo']} -> {info_extra}") - - emails_finales.append(e) - else: - print(f" └── ℹ️ Descartado (sin archivos PDF/XML válidos).") - - except Exception as ex: - print(f" ❌ Error procesando {e['id'][:10]}: {ex}") + print(f"\n⏳ Esperando {intervalo}s hasta proximo ciclo...") + for _ in range(intervalo): + if not _running: + break + time.sleep(1) - # 5. Sincronizar a SMB al terminar de procesar TODOS los correos de esta cuenta - if emails_finales: - import smb_utils - smb_utils.sincronizar_a_smb(mail.ruta_final, mail.smb_folder) - - # 6. Limpiar carpeta de descarga original - file_manager.limpiar_carpeta_descarga(mail.ruta_descarga) + print("\n👋 Daemon detenido.") - # Actualizamos el listado con los que realmente tienen contenido útil - emails_a_procesar = emails_finales - print(f"\n✅ Proceso completado para {mail.email}. {len(emails_a_procesar)} correos listos para procesar.") - - # Acumular para el guardado final en JSON - todos_procesados.extend(emails_a_procesar) - - # Guardar resultados globales en JSON - if todos_procesados: - output_dir = "output_json" - if not os.path.exists(output_dir): - os.makedirs(output_dir) - - timestamp_json = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") - json_path = os.path.join(output_dir, f"{timestamp_json}.json") - - with open(json_path, "w", encoding="utf-8") as f: - json.dump(todos_procesados, f, ensure_ascii=False, indent=4) - - print("\n" + "="*60) - print(f"💾 Resultados exportados a: {json_path}") - - fin_ejecucion = datetime.now() - duracion = fin_ejecucion - inicio_ejecucion - print("\n" + "="*60) - print(f"🏁 FIN DE EJECUCIÓN: {fin_ejecucion.strftime('%Y-%m-%d %H:%M:%S')}") - print(f"⏱️ Duración total: {str(duracion).split('.')[0]}") # Sin microsegundos - print("="*60) +# ======================================================================== +if __name__ == "__main__": + main() diff --git a/outlook_utilities.py b/outlook_utilities.py index 9473b4b..a61b10d 100644 --- a/outlook_utilities.py +++ b/outlook_utilities.py @@ -8,7 +8,10 @@ from config_loader import CONFIG warnings.filterwarnings("ignore") -######################################################## +# Diferencia horaria respecto a UTC (Perú: UTC-5) +UTC_OFFSET_HOURS = -5 + +# ======================================================================== def get_access_token(): """ Obtiene un token de acceso de Microsoft Graph usando las credenciales @@ -37,7 +40,7 @@ def get_access_token(): correlation_id = result.get("correlation_id") raise Exception(f"Error al obtener token: {error} - {error_description} (ID: {correlation_id})") -######################################################## +# ======================================================================== def get_emails(user_email, token, fecha_inicio, fecha_fin): """ Obtiene los correos de una cuenta específica en un rango de fechas. @@ -48,8 +51,8 @@ def get_emails(user_email, token, fecha_inicio, fecha_fin): "Content-Type": "application/json" } - dt_inicio_utc = datetime.strptime(fecha_inicio, "%Y-%m-%d") + timedelta(hours=5) - dt_fin_utc = datetime.strptime(fecha_fin, "%Y-%m-%d") + timedelta(days=1, hours=5) + dt_inicio_utc = datetime.strptime(fecha_inicio, "%Y-%m-%d") + timedelta(hours=-UTC_OFFSET_HOURS) + dt_fin_utc = datetime.strptime(fecha_fin, "%Y-%m-%d") + timedelta(days=1, hours=-UTC_OFFSET_HOURS) filter_query = ( f"receivedDateTime ge {dt_inicio_utc.strftime('%Y-%m-%dT%H:%M:%SZ')} and " @@ -82,7 +85,7 @@ def get_emails(user_email, token, fecha_inicio, fecha_fin): for msg in messages: utc_str = msg.get("receivedDateTime") dt_utc = datetime.fromisoformat(utc_str.replace("Z", "+00:00")) - dt_local = dt_utc - timedelta(hours=5) + dt_local = dt_utc + timedelta(hours=UTC_OFFSET_HOURS) from_obj = msg.get("from", {}) sender = from_obj.get("emailAddress", {}).get("address", "Desconocido") @@ -100,7 +103,7 @@ def get_emails(user_email, token, fecha_inicio, fecha_fin): return all_mails -######################################################## +# ======================================================================== def download_attachments(user_email, message_id, token, download_path): """ Descarga los adjuntos de un mensaje específico y los guarda en la ruta indicada. @@ -137,7 +140,7 @@ def download_attachments(user_email, message_id, token, download_path): return downloaded_files -######################################################## +# ======================================================================== if __name__ == "__main__": try: print("Intentando obtener token de acceso...") @@ -162,4 +165,5 @@ if __name__ == "__main__": print(f"Descargados: {archivos}") break except Exception as e: - print(f"❌ Error: {e}") \ No newline at end of file + print(f"❌ Error: {e}") +# ======================================================================== \ No newline at end of file diff --git a/pdf_utilities.py b/pdf_utilities.py index 0ac9cc9..2bd96ab 100644 --- a/pdf_utilities.py +++ b/pdf_utilities.py @@ -1,8 +1,9 @@ import pdfplumber import re import os +import llm_processor -######################################################## +# ======================================================================== def extraer_texto_pdf(ruta_pdf): """Extrae todo el texto de un PDF usando pdfplumber.""" texto_completo = "" @@ -15,7 +16,7 @@ def extraer_texto_pdf(ruta_pdf): print(f" ⚠️ Error leyendo PDF {ruta_pdf}: {e}") return "" -######################################################## +# ======================================================================== def extraer_ruc(texto): """Busca un RUC (11 dígitos, empieza con 10 o 20) en el texto.""" # Intentamos buscar patrones de RUC: RUC 2060... o R.U.C. 2060... @@ -30,54 +31,139 @@ def extraer_ruc(texto): return match.group(1) return "N/A" -######################################################## +# ======================================================================== +def _flexible(palabra): + """Convierte una palabra en un patrón regex que tolera espacios entre caracteres. + Ej: FACTURA -> F\\s*A\\s*C\\s*T\\s*U\\s*R\\s*A + """ + return '\\s*'.join(palabra) + + def extraer_tipo_documento(texto): - """Identifica el tipo de documento SUNAT basándose en el texto usando regex multilínea.""" + """Identifica el tipo de documento SUNAT basándose en el texto usando regex multilínea. + Soporta texto con espacios entre caracteres (ej: 'F A C T U R A').""" t = texto.upper() - + patrones = { - "01": [r'FACTURA\s+(ELECTR[OÓ]NICA|DE\s+VENTA)', r'FACTURA'], - "03": [r'BOLETA\s+DE\s+VENTA', r'BOLETA\s+ELECTR[OÓ]NICA'], - "07": [r'NOTA\s+DE\s+CR[EÉ]DITO'], - "08": [r'NOTA\s+DE\s+D[EÉ]BITO'], - "14": [r'RECIBO\s+POR\s+HONORARIOS'], - "09": [r'GU[IÍ]A\s+DE\s+REMISI[OÓ]N'] + "01": [ + rf'{_flexible("FACTURA")}\s+({_flexible("ELECTRÓNICA")}|{_flexible("ELECTRONICA")}|DE\s+VENTA)', + rf'{_flexible("FACTURA")}', + ], + "03": [ + rf'{_flexible("BOLETA")}\s+DE\s+{_flexible("VENTA")}', + rf'{_flexible("BOLETA")}\s+{_flexible("ELECTRÓNICA")}', + rf'{_flexible("BOLETA")}\s+{_flexible("ELECTRONICA")}', + ], + "07": [ + rf'{_flexible("NOTA")}\s+DE\s+{_flexible("CRÉDITO")}', + rf'{_flexible("NOTA")}\s+DE\s+{_flexible("CREDITO")}', + ], + "08": [ + rf'{_flexible("NOTA")}\s+DE\s+{_flexible("DÉBITO")}', + rf'{_flexible("NOTA")}\s+DE\s+{_flexible("DEBITO")}', + ], + "14": [ + rf'{_flexible("RECIBO")}\s+POR\s+{_flexible("HONORARIOS")}', + ], + "09": [ + rf'{_flexible("GUÍA")}\s+DE\s+{_flexible("REMISIÓN")}', + rf'{_flexible("GUIA")}\s+DE\s+{_flexible("REMISION")}', + ], } - + for tipo, regex_list in patrones.items(): for p in regex_list: if re.search(p, t, re.IGNORECASE | re.MULTILINE): return tipo - + return "99" -######################################################## +# ======================================================================== def extraer_serie_numero(texto): - """Busca la serie y número (ej. F001-123 o F001-0000123).""" - # Patrón común: [FBE][A-Z0-9]{3}-\d+ - patron = re.search(r'\b([FBE][A-Z0-9]{3})\s*-\s*(\d+)\b', texto) - if patron: - return f"{patron.group(1)}-{patron.group(2)}" + """Busca la serie y número (ej. F001-123, E001-0000123, EG07-2367).""" + serie_patrones = [ + # Formato con guión: letras+dígitos - número + r'\b([A-Z]{1,4}\d{2,4})\s*-\s*(\d+)\b', + # Fallback sin guión + r'([A-Z]{1,4}\d{2,4})\s+(\d{1,10})', + # Serie numérica corta sin letras + r'\b(\d{3})\s*-\s*(\d+)\b' + ] - # Fallback para casos donde no hay guión o hay espacios raros - patron_alt = re.search(r'([FBE][A-Z0-9]{3})\s+(\d{1,10})', texto) - if patron_alt: - return f"{patron_alt.group(1)}-{patron_alt.group(2)}" + for patron in serie_patrones: + match = re.search(patron, texto) + if match: + return f"{match.group(1)}-{match.group(2)}" return "No encontrado" -######################################################## +# ======================================================================== +def _necesita_llm(resultado): + """True si algún campo clave no se encontró con regex.""" + return ( + resultado['ruc'] in ("N/A", "Desconocido") or + resultado['tipo_doc'] == "99" or + resultado['id'] in ("No encontrado", "N/A") + ) + + +def _campos_rellenados(regex_original, regex_result): + """Cuenta cuántos campos cambió el LLM respecto al regex original.""" + count = 0 + if regex_original['ruc'] in ("N/A", "Desconocido") and regex_result['ruc'] not in ("N/A", "Desconocido"): + count += 1 + if regex_original['tipo_doc'] == "99" and regex_result['tipo_doc'] != "99": + count += 1 + if regex_original['id'] in ("No encontrado", "N/A") and regex_result['id'] not in ("No encontrado", "N/A"): + count += 1 + return count + + +def _merge_llm(regex_result, llm_data): + """Rellena campos vacíos del regex con datos del LLM sin pisar lo ya encontrado.""" + if regex_result['ruc'] in ("N/A", "Desconocido"): + ruc = llm_data.get('ruc', '') + if re.match(r'^(10|20)\d{9}$', str(ruc)): + regex_result['ruc'] = ruc + + if regex_result['tipo_doc'] == "99": + tipo = str(llm_data.get('tipo', '99')) + tipo_codes = {"01", "02", "03", "07", "08", "09", "14"} + if tipo in tipo_codes: + regex_result['tipo_doc'] = tipo + + if regex_result['id'] in ("No encontrado", "N/A"): + serie = str(llm_data.get('serie', '')) + numero = str(llm_data.get('numero', '')) + if serie and numero: + regex_result['id'] = f"{serie}-{numero}" + + return regex_result + + def obtener_datos_pdf(ruta_pdf): - """Orquestador para extraer metadata de un PDF.""" + """Orquestador para extraer metadata de un PDF. + Primero intenta con regex; si faltan campos, usa LLM como fallback.""" texto = extraer_texto_pdf(ruta_pdf) if not texto: return {"valido": False, "error": "No se pudo extraer texto"} - return { + resultado = { "valido": True, "tipo_doc": extraer_tipo_documento(texto), "ruc": extraer_ruc(texto), - "id": extraer_serie_numero(texto) + "id": extraer_serie_numero(texto), + "llm_usado": False, + "llm_respuesta": None } -######################################################## \ No newline at end of file + if _necesita_llm(resultado): + resultado["llm_usado"] = True + llm_data = llm_processor.extraer_con_llm(texto) + resultado["llm_respuesta"] = dict(llm_data) if llm_data else None + if llm_data: + antes = dict(resultado) + resultado = _merge_llm(resultado, llm_data) + + return resultado +# ======================================================================== \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 7e2e112..b5c0b0a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,4 @@ requests msal -pdfplumber \ No newline at end of file +pdfplumber +pytest \ No newline at end of file diff --git a/scheduler.json b/scheduler.json new file mode 100644 index 0000000..cf67aa1 --- /dev/null +++ b/scheduler.json @@ -0,0 +1,9 @@ +{ + "HORARIOS": [ + [6, 30, 13, 15, 300, "rapido"], + [13, 15, 14, 0, 3600, "full"], + [14, 0, 17, 30, 300, "rapido"], + [17, 30, 21, 30, 3600, "full"] + ], + "DIAS_ACTIVOS": [0, 1, 2, 3, 4] +} diff --git a/smb_manager.py b/smb_manager.py new file mode 100644 index 0000000..4ebfc54 --- /dev/null +++ b/smb_manager.py @@ -0,0 +1,144 @@ +import os +import platform +import subprocess +from config_loader import CONFIG + +# ======================================================================== +def get_current_os(): + if os.path.exists("/.dockerenv"): + return "docker" + + system = platform.system().lower() + if system == "darwin": + return "mac" + if system == "windows": + return "windows" + return "linux" + +# ======================================================================== +def get_mount_point(): + return CONFIG.smb.mount_points.get(get_current_os(), "") + + +def esta_montado(mount_point): + if not mount_point: + return False + if get_current_os() == "windows": + return os.path.exists(mount_point) + return os.path.ismount(mount_point) or os.path.exists( + os.path.join(mount_point, ".mounted_check") + ) + +# ======================================================================== +def montar_smb(): + mount_point = get_mount_point() + if not mount_point: + print(" ⚠️ No hay punto de montaje configurado para este SO.") + return False + + if esta_montado(mount_point): + return True + + current_os = get_current_os() + + if current_os != "windows" and not os.path.exists(mount_point): + os.makedirs(mount_point) + + smb = CONFIG.smb + + try: + if current_os == "mac": + cmd = ( + f"mount_smbfs //'{smb.username}':'{smb.password}'" + f"@{smb.server}/{smb.share} {mount_point}" + ) + subprocess.run(cmd, shell=True, check=True, capture_output=True) + elif current_os == "linux": + cmd = ( + f"sudo mount -t cifs //{smb.server}/{smb.share} {mount_point}" + f" -o username={smb.username},password={smb.password}" + ) + subprocess.run(cmd, shell=True, check=True, capture_output=True) + + elif current_os == "docker": + print("ℹ️ Docker: SMB debe estar montado por el host") + return os.path.exists(mount_point) + + elif current_os == "windows": + ruta_smb = f"\\\\{smb.server}\\{smb.share}" + cmd = [ + "net", "use", mount_point, ruta_smb, + smb.password, f"/user:{smb.username}", "/persistent:no", + ] + subprocess.run(cmd, check=True, capture_output=True) + + print(f" ✅ SMB montado en: {mount_point}") + return True + except Exception as e: + print(f" ⚠️ Error montando SMB: {e}") + return False + +# ======================================================================== +def sincronizar_a_smb(ruta_local, subfolder_cuenta): + if not os.path.isdir(ruta_local): + print(f" ⚠️ La carpeta local no existe: {ruta_local}") + return False + + if not montar_smb(): + print(" ❌ No se pudo sincronizar: SMB no montado.") + return False + + mount_point = get_mount_point() + current_os = get_current_os() + + if current_os == "windows" and mount_point.endswith(":") and len(mount_point) == 2: + mount_point += "\\" + + subfolder = subfolder_cuenta.lstrip("/") + ruta_destino = os.path.normpath(os.path.join(mount_point, subfolder)) + + if not os.path.exists(ruta_destino): + os.makedirs(ruta_destino) + print(f" 📁 Carpeta creada en SMB: {ruta_destino}") + + source = os.path.abspath(ruta_local) + destination = os.path.abspath(ruta_destino) + + print(f" 🚀 Sincronizando {current_os.upper()}: {source} -> {destination}") + + try: + if current_os in ("mac", "linux", "docker"): + # Solo sincronizar PDF y XML + cmd = ( + f"find {source} -maxdepth 1 \\( -name '*.pdf' -o -name '*.xml' \\) " + f"-exec rsync -av --remove-source-files {{}} {destination}/ \\;" + ) + subprocess.run(cmd, shell=True, check=True, capture_output=True, text=True) + # Limpiar temporales restantes (json, zip) + subprocess.run( + f"find {source} -type f \\( -name '*.json' -o -name '*.zip' \\) -delete", + shell=True, check=True, capture_output=True, text=True, + ) + + elif current_os == "windows": + import shutil + for archivo in os.listdir(source): + if not archivo.lower().endswith((".pdf", ".xml")): + continue + src_path = os.path.join(source, archivo) + dst_path = os.path.join(destination, archivo) + if os.path.isfile(src_path): + if os.path.exists(dst_path): + os.remove(dst_path) + shutil.move(src_path, dst_path) + + for archivo in os.listdir(source): + if archivo.lower().endswith((".json", ".zip")): + os.remove(os.path.join(source, archivo)) + + print(f" ✅ Sincronización completada.") + return True + except Exception as e: + print(f" ❌ Error en sincronización: {e}") + return False +# ======================================================================== \ No newline at end of file diff --git a/smb_utils.py b/smb_utils.py deleted file mode 100644 index 258419e..0000000 --- a/smb_utils.py +++ /dev/null @@ -1,126 +0,0 @@ -import os -import platform -import subprocess -from config_loader import CONFIG - -######################################################## -def get_current_os(): - system = platform.system().lower() - if system == "darwin": return "mac" - return system - -######################################################## -def get_mount_point(): - current_os = get_current_os() - return CONFIG.smb.mount_points.get(current_os) - -######################################################## -def esta_montado(mount_point): - """Verifica si el punto de montaje ya está activo.""" - if get_current_os() == "windows": - return os.path.exists(mount_point) - return os.path.ismount(mount_point) or os.path.exists(os.path.join(mount_point, ".mounted_check")) # Simplificado - -######################################################## -def montar_smb(): - """Intenta montar la unidad SMB según el SO.""" - mount_point = get_mount_point() - current_os = get_current_os() - - if not mount_point: - print(" ⚠️ No hay punto de montaje configurado para este SO.") - return False - - if esta_montado(mount_point): - # print(f" ✅ SMB ya montado en: {mount_point}") - return True - - # En Windows no creamos la carpeta del punto de montaje (es una unidad virtual) - if current_os != "windows" and not os.path.exists(mount_point): - os.makedirs(mount_point) - - smb = CONFIG.smb - current_os = get_current_os() - - try: - if current_os == "mac": - # mount_smbfs //user:pass@server/share mount_point - cmd = f"mount_smbfs //'{smb.username}':'{smb.password}'@{smb.server}/{smb.share} {mount_point}" - subprocess.run(cmd, shell=True, check=True, capture_output=True) - elif current_os == "linux": - # mount -t cifs //server/share mount_point -o username=user,password=pass - cmd = f"sudo mount -t cifs //{smb.server}/{smb.share} {mount_point} -o username={smb.username},password={smb.password}" - subprocess.run(cmd, shell=True, check=True, capture_output=True) - elif current_os == "windows": - # net use S: \\server\share /user:user pass - ruta_smb = f"\\\\{smb.server}\\{smb.share}" - cmd = [ - "net", "use", mount_point, ruta_smb, - smb.password, f"/user:{smb.username}", "/persistent:no" - ] - subprocess.run(cmd, check=True, capture_output=True) - - print(f" ✅ SMB montado exitosamente en: {mount_point}") - return True - except Exception as e: - print(f" ⚠️ Error montando SMB: {e}") - return False - -######################################################## -def sincronizar_a_smb(ruta_local, smb_subfolder): - """Sincroniza archivos hacia el recurso SMB montado según el SO.""" - if not montar_smb(): - print(" ❌ No se pudo sincronizar: SMB no montado.") - return False - - mount_point = get_mount_point() - current_os = get_current_os() - - # Normalizar punto de montaje para Windows (asegurar S:\ en lugar de S:) - if current_os == "windows" and mount_point.endswith(":") and len(mount_point) == 2: - mount_point += "\\" - - subfolder = smb_subfolder.lstrip('/') - ruta_destino = os.path.normpath(os.path.join(mount_point, subfolder)) - - if not os.path.exists(ruta_destino): - os.makedirs(ruta_destino) - print(f" 📁 Carpeta de destino creada en SMB: {ruta_destino}") - - source = os.path.abspath(ruta_local) - destination = os.path.abspath(ruta_destino) - - print(f" 🚀 Sincronizando a SMB ({current_os.upper()}): {ruta_destino}...") - - try: - if current_os in ["mac", "linux"]: - # Usar rsync en sistemas Unix - # El / al final del source indica copiar contenido - cmd = ["rsync", "-av", "--remove-source-files", source + "/", destination + "/"] - subprocess.run(cmd, check=True, capture_output=True, text=True) - - elif current_os == "windows": - # Usar shutil de Python (sin dependencias externas) - import shutil - for archivo in os.listdir(source): - src_path = os.path.join(source, archivo) - dst_path = os.path.join(destination, archivo) - - if os.path.isfile(src_path): - # shutil.move maneja copia y borrado incluso entre discos diferentes - if os.path.exists(dst_path): - os.remove(dst_path) - shutil.move(src_path, dst_path) - elif os.path.isdir(src_path): - # Para carpetas, usamos copytree + rmtree o move - if os.path.exists(dst_path): - shutil.rmtree(dst_path) - shutil.move(src_path, dst_path) - - print(f" ✅ Sincronización completada.") - return True - except Exception as e: - print(f" ❌ Error en la sincronización: {e}") - return False - -######################################################## diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..2c30c2d --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,110 @@ +import pytest +import os +import tempfile +import xml.etree.ElementTree as ET + + +FACTURA_XML = ''' + + F001-00005103 + 01 + + + + 20549758054 + + + + + + + IGV + + + +''' + +CREDIT_NOTE_XML = ''' + + B001-00001234 + 07 + + + + 20123456789 + + + +''' + +CDR_XML = ''' + + R-2025-00000123 + + + 20549758054 + + +''' + +RECIBO_XML = ''' + + R001-00000456 + 02 + + 10456789012 + + + + + RET 4TA + + + +''' + + +@pytest.fixture +def temp_dir(): + with tempfile.TemporaryDirectory() as tmpdir: + yield tmpdir + + +@pytest.fixture +def factura_xml_path(temp_dir): + path = os.path.join(temp_dir, "F001-00005103.xml") + with open(path, "w", encoding="utf-8") as f: + f.write(FACTURA_XML) + return path + + +@pytest.fixture +def credito_xml_path(temp_dir): + path = os.path.join(temp_dir, "B001-00001234.xml") + with open(path, "w", encoding="utf-8") as f: + f.write(CREDIT_NOTE_XML) + return path + + +@pytest.fixture +def cdr_xml_path(temp_dir): + path = os.path.join(temp_dir, "R-2025-00000123.xml") + with open(path, "w", encoding="utf-8") as f: + f.write(CDR_XML) + return path + + +@pytest.fixture +def recibo_xml_path(temp_dir): + path = os.path.join(temp_dir, "R001-00000456.xml") + with open(path, "w", encoding="utf-8") as f: + f.write(RECIBO_XML) + return path diff --git a/tests/test_database_utilities.py b/tests/test_database_utilities.py new file mode 100644 index 0000000..d7667fe --- /dev/null +++ b/tests/test_database_utilities.py @@ -0,0 +1,73 @@ +import os +import sys +import pytest +import tempfile +import sqlite3 + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +# Forzar uso de DB temporal para no contaminar la real +@pytest.fixture(autouse=True) +def override_db_path(monkeypatch, tmp_path): + db_path = str(tmp_path / "test_emails.db") + monkeypatch.setattr("database_utilities.DB_NAME", db_path) + yield + if os.path.exists(db_path): + os.remove(db_path) + + +from database_utilities import init_db, save_emails, get_existing_ids, delete_email_by_id + + +class TestInitDb: + def test_crea_tabla(self): + init_db() + conn = sqlite3.connect("test_emails.db" if False else ":memory:") + # No podemos verificar la DB real porque el path se sobreescribió + # Solo verificar que no lanza excepción + assert True + + +class TestSaveAndGetEmails: + def test_save_and_retrieve(self): + init_db() + emails = [ + {"id": "id1", "fecha": "2026-01-01", "sender": "a@b.com", + "subject": "Test", "has_attach": True}, + {"id": "id2", "fecha": "2026-01-02", "sender": "c@d.com", + "subject": "Test2", "has_attach": False}, + ] + assert save_emails(emails) > 0 + assert get_existing_ids(["id1", "id2", "id3"]) == ["id1", "id2"] + + def test_duplicados_ignorados(self): + init_db() + emails = [ + {"id": "id1", "fecha": "2026-01-01", "sender": "a@b.com", + "subject": "Test", "has_attach": True}, + ] + save_emails(emails) + # Insertar el mismo id otra vez + save_emails(emails) + assert get_existing_ids(["id1"]) == ["id1"] + + def test_lista_vacia(self): + assert save_emails([]) == 0 + assert get_existing_ids([]) == [] + + +class TestDeleteEmail: + def test_delete_existing(self): + init_db() + save_emails([{"id": "id1", "fecha": "2026-01-01", "sender": "a@b.com", + "subject": "Test", "has_attach": True}]) + assert delete_email_by_id("id1") is True + assert get_existing_ids(["id1"]) == [] + + def test_delete_nonexistent(self): + init_db() + assert delete_email_by_id("no_existe") is False + + def test_delete_empty_id(self): + assert delete_email_by_id("") is False + assert delete_email_by_id(None) is False diff --git a/tests/test_document_processor.py b/tests/test_document_processor.py new file mode 100644 index 0000000..a7b64f9 --- /dev/null +++ b/tests/test_document_processor.py @@ -0,0 +1,98 @@ +import os +import sys +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from document_processor import ( + formatear_serie_numero, + generar_nombre_propuesto, + obtener_datos_xml, + TIPOS_SUNAT, +) + + +class TestFormatearSerieNumero: + def test_formato_estandar(self): + assert formatear_serie_numero("F001-00005103") == "F001-5103" + + def test_sin_ceros(self): + assert formatear_serie_numero("F001-123") == "F001-123" + + def test_con_letras_en_numero(self): + assert formatear_serie_numero("E001-ABC") == "E001-ABC" + + def test_sin_guion(self): + assert formatear_serie_numero("F001") == "F001" + + def test_cdr_id(self): + assert formatear_serie_numero("R-2025-00000123") == "R-2025-00000123" + + def test_na(self): + assert formatear_serie_numero("N/A") == "N/A" + + def test_no_encontrado(self): + assert formatear_serie_numero("No encontrado") == "No encontrado" + + +class TestGenerarNombrePropuesto: + def test_factura(self): + m = {"tipo": "01", "ruc": "20549758054", "serie_numero": "F001-5103"} + assert generar_nombre_propuesto(m) == "20549758054-01-F001-5103" + + def test_cdr(self): + m = {"tipo": "CDR", "ruc": "20549758054", "serie_numero": "R-2025-123"} + assert generar_nombre_propuesto(m) == "20549758054-CDR-R-2025-123" + + def test_credito(self): + m = {"tipo": "07", "ruc": "20123456789", "serie_numero": "B001-1234"} + assert generar_nombre_propuesto(m) == "20123456789-07-B001-1234" + + def test_tipo_99_no_renombra(self): + m = {"tipo": "99", "ruc": "20549758054", "serie_numero": "F001-5103"} + assert generar_nombre_propuesto(m) is None + + def test_ruc_desconocido_no_renombra(self): + m = {"tipo": "01", "ruc": "Desconocido", "serie_numero": "F001-5103"} + assert generar_nombre_propuesto(m) is None + + def test_ruc_na_no_renombra(self): + m = {"tipo": "01", "ruc": "N/A", "serie_numero": "F001-5103"} + assert generar_nombre_propuesto(m) is None + + +class TestObtenerDatosXml: + def test_factura(self, factura_xml_path): + res = obtener_datos_xml(factura_xml_path) + assert res["valido"] is True + assert res["tipo_xml"] == "01" + assert res["ruc"] == "20549758054" + assert res["id"] == "F001-00005103" + + def test_credito(self, credito_xml_path): + res = obtener_datos_xml(credito_xml_path) + assert res["valido"] is True + assert res["tipo_xml"] == "07" + assert res["ruc"] == "20123456789" + assert res["id"] == "B001-00001234" + + def test_cdr(self, cdr_xml_path): + res = obtener_datos_xml(cdr_xml_path) + assert res["valido"] is True + assert res["tipo_xml"] == "CDR" + assert res["id"] == "R-2025-00000123" + + def test_recibo_honorarios(self, recibo_xml_path): + res = obtener_datos_xml(recibo_xml_path) + assert res["valido"] is True + assert res["tipo_xml"] == "02" + assert res["id"] == "R001-00000456" + + def test_xml_inexistente(self): + res = obtener_datos_xml("/ruta/inexistente.xml") + assert res["valido"] is False + + def test_tipos_sunat_incluye_cdr(self): + assert "CDR" in TIPOS_SUNAT + assert "01" in TIPOS_SUNAT + assert "02" in TIPOS_SUNAT diff --git a/tests/test_email_services.py b/tests/test_email_services.py new file mode 100644 index 0000000..8d111b3 --- /dev/null +++ b/tests/test_email_services.py @@ -0,0 +1,108 @@ +import os +import sys +import pytest +from unittest.mock import patch, MagicMock + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from email_services import ( + filtrar_correos_nuevos, + descomprimir_adjuntos, + filtrar_archivos_validos, + procesar_adjuntos_email, +) + + +class TestFiltrarArchivosValidos: + def test_solo_pdf(self): + archivos = ["doc.pdf", "imagen.jpg", "texto.txt"] + assert filtrar_archivos_validos(archivos) == ["doc.pdf"] + + def test_solo_xml(self): + archivos = ["doc.xml", "imagen.jpg", "texto.txt"] + assert filtrar_archivos_validos(archivos) == ["doc.xml"] + + def test_pdf_y_xml(self): + archivos = ["doc.pdf", "doc.xml", "imagen.jpg"] + assert filtrar_archivos_validos(archivos) == ["doc.pdf", "doc.xml"] + + def test_sin_validos(self): + archivos = ["imagen.jpg", "texto.txt", "archivo.zip"] + assert filtrar_archivos_validos(archivos) == [] + + def test_case_insensitive(self): + archivos = ["DOC.PDF", "Doc.Xml", "IMAGEN.JPG"] + assert filtrar_archivos_validos(archivos) == ["DOC.PDF", "Doc.Xml"] + + +class TestDescomprimirAdjuntos: + def test_sin_zips(self, temp_dir): + archivos = ["a.pdf", "b.xml"] + resultado = descomprimir_adjuntos(archivos, temp_dir) + assert resultado == ["a.pdf", "b.xml"] + + def test_zip_inexistente(self, temp_dir): + archivos = ["nope.zip"] + resultado = descomprimir_adjuntos(archivos, temp_dir) + assert resultado == ["nope.zip"] + + +class TestFiltrarCorreosNuevos: + @patch("email_services.database_utilities.get_existing_ids") + @patch("email_services.database_utilities.save_emails") + def test_retorna_todos_los_nuevos(self, mock_save, mock_get): + mock_get.return_value = [] + emails = [ + {"id": "1", "has_attach": True}, + {"id": "2", "has_attach": False}, + ] + resultado = filtrar_correos_nuevos(emails) + # Debe retornar TODOS los nuevos (con y sin adjuntos) + assert len(resultado) == 2 + mock_save.assert_called_once() + + @patch("email_services.database_utilities.get_existing_ids") + @patch("email_services.database_utilities.save_emails") + def test_filtra_existentes(self, mock_save, mock_get): + mock_get.return_value = ["1"] + emails = [ + {"id": "1", "has_attach": True}, + {"id": "2", "has_attach": True}, + ] + resultado = filtrar_correos_nuevos(emails) + assert len(resultado) == 1 + assert resultado[0]["id"] == "2" + + @patch("email_services.database_utilities.get_existing_ids") + def test_emails_vacios(self, mock_get): + assert filtrar_correos_nuevos([]) == [] + mock_get.assert_not_called() + + +class TestProcesarAdjuntosEmail: + def test_con_adjuntos_validos(self, temp_dir): + e = {} + for f in ["doc.pdf", "doc.xml", "imagen.jpg"]: + with open(os.path.join(temp_dir, f), "w") as fh: + fh.write("dummy") + + archivos = ["doc.pdf", "doc.xml", "imagen.jpg"] + resultado = procesar_adjuntos_email(e, archivos, temp_dir) + assert resultado == ["doc.pdf", "doc.xml"] + assert e["archivos"] == ["doc.pdf", "doc.xml"] + + def test_sin_adjuntos_validos(self, temp_dir): + e = {} + for f in ["imagen.jpg", "texto.txt"]: + with open(os.path.join(temp_dir, f), "w") as fh: + fh.write("dummy") + + archivos = ["imagen.jpg", "texto.txt"] + resultado = procesar_adjuntos_email(e, archivos, temp_dir) + assert resultado == [] + assert "archivos" not in e + + def test_archivos_vacios(self): + e = {} + resultado = procesar_adjuntos_email(e, [], "/tmp") + assert resultado == [] diff --git a/tests/test_file_manager.py b/tests/test_file_manager.py new file mode 100644 index 0000000..78142a1 --- /dev/null +++ b/tests/test_file_manager.py @@ -0,0 +1,128 @@ +import os +import sys +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from file_manager import renombrar_archivos_por_metadata, mover_validos_a_resultado, limpiar_carpeta_descarga + + +class TestRenombrarArchivosPorMetadata: + def test_renombrar_xml_valido(self, temp_dir): + ruta = os.path.join(temp_dir, "F001-100.xml") + with open(ruta, "w") as f: + f.write("") + + metadatos = [{ + "archivo": "F001-100.xml", + "extension": ".xml", + "tipo": "01", + "serie_numero": "F001-100", + "ruc": "20549758054", + "archivo_propuesto": "20549758054-01-F001-100" + }] + + renombrar_archivos_por_metadata(metadatos, temp_dir) + + assert not os.path.exists(ruta) + nuevo = os.path.join(temp_dir, "20549758054-01-F001-100.xml") + assert os.path.exists(nuevo) + assert metadatos[0]["archivo"] == "20549758054-01-F001-100.xml" + + def test_no_renombra_sin_metadata(self, temp_dir): + ruta = os.path.join(temp_dir, "documento.pdf") + with open(ruta, "w") as f: + f.write("dummy") + + metadatos = [{ + "archivo": "documento.pdf", + "extension": ".pdf", + "tipo": "99", + "serie_numero": "N/A", + "ruc": "N/A", + "archivo_propuesto": None + }] + + renombrar_archivos_por_metadata(metadatos, temp_dir) + + assert os.path.exists(ruta) + assert metadatos[0]["archivo"] == "documento.pdf" + + def test_renombrar_cdr(self, temp_dir): + ruta = os.path.join(temp_dir, "R-2025-123.xml") + with open(ruta, "w") as f: + f.write("") + + metadatos = [{ + "archivo": "R-2025-123.xml", + "extension": ".xml", + "tipo": "CDR", + "serie_numero": "R-2025-123", + "ruc": "20549758054", + "archivo_propuesto": "20549758054-CDR-R-2025-123" + }] + + renombrar_archivos_por_metadata(metadatos, temp_dir) + + assert not os.path.exists(ruta) + nuevo = os.path.join(temp_dir, "20549758054-CDR-R-2025-123.xml") + assert os.path.exists(nuevo) + + +class TestMoverValidosAResultado: + def test_mover_archivos_validos(self, temp_dir): + origen = os.path.join(temp_dir, "origen") + destino = os.path.join(temp_dir, "destino") + os.makedirs(origen) + + ruta_file = os.path.join(origen, "20549758054-01-F001-100.xml") + with open(ruta_file, "w") as f: + f.write("") + + metadatos = [{ + "archivo": "20549758054-01-F001-100.xml", + "extension": ".xml", + "tipo": "01", + "archivo_propuesto": "20549758054-01-F001-100" + }] + + mover_validos_a_resultado(metadatos, origen, destino) + + assert not os.path.exists(ruta_file) + assert os.path.exists(os.path.join(destino, "20549758054-01-F001-100.xml")) + + def test_no_mueve_sin_metadata(self, temp_dir): + origen = os.path.join(temp_dir, "origen") + destino = os.path.join(temp_dir, "destino") + os.makedirs(origen) + + ruta_file = os.path.join(origen, "basura.pdf") + with open(ruta_file, "w") as f: + f.write("dummy") + + metadatos = [{ + "archivo": "basura.pdf", + "extension": ".pdf", + "tipo": "99", + "archivo_propuesto": None + }] + + mover_validos_a_resultado(metadatos, origen, destino) + + assert os.path.exists(ruta_file) + assert os.path.exists(destino) + assert len(os.listdir(destino)) == 0 + + +class TestLimpiarCarpetaDescarga: + def test_limpia_archivos(self, temp_dir): + for f in ["a.pdf", "b.xml", "c.zip"]: + with open(os.path.join(temp_dir, f), "w") as fh: + fh.write("dummy") + + limpiar_carpeta_descarga(temp_dir) + + assert len(os.listdir(temp_dir)) == 0 + + def test_carpeta_no_existe(self): + limpiar_carpeta_descarga("/ruta/inexistente") diff --git a/tests/test_pdf_utilities.py b/tests/test_pdf_utilities.py new file mode 100644 index 0000000..f431c43 --- /dev/null +++ b/tests/test_pdf_utilities.py @@ -0,0 +1,99 @@ +import os +import sys +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from pdf_utilities import extraer_ruc, extraer_tipo_documento, extraer_serie_numero, obtener_datos_pdf + + +class TestExtraerRuc: + def test_ruc_formato_completo(self): + texto = "R.U.C.: 20549758054\nRazon Social: EMPRESA SAC" + assert extraer_ruc(texto) == "20549758054" + + def test_ruc_con_puntos(self): + texto = "R.U.C. 20549758054\nAlguna direccion" + assert extraer_ruc(texto) == "20549758054" + + def test_ruc_desnudo(self): + texto = "Factura Electronica\n20549758054\nMonto: 100.00" + assert extraer_ruc(texto) == "20549758054" + + def test_ruc_con_espacio(self): + texto = "RUC: 20605626921\nBoleta de Venta" + assert extraer_ruc(texto) == "20605626921" + + def test_sin_ruc(self): + texto = "Documento sin numero de identificacion" + assert extraer_ruc(texto) == "N/A" + + +class TestExtraerTipoDocumento: + def test_factura_electronica(self): + texto = "FACTURA ELECTRONICA\nRUC: 20549758054" + assert extraer_tipo_documento(texto) == "01" + + def test_factura_simple(self): + texto = "FACTURA DE VENTA\nRUC: 20549758054" + assert extraer_tipo_documento(texto) == "01" + + def test_boleta(self): + texto = "BOLETA DE VENTA ELECTRONICA\nRUC: 20549758054" + assert extraer_tipo_documento(texto) == "03" + + def test_nota_credito(self): + texto = "NOTA DE CREDITO ELECTRONICA\nRUC: 20549758054" + assert extraer_tipo_documento(texto) == "07" + + def test_nota_debito(self): + texto = "NOTA DE DEBITO\nRUC: 20549758054" + assert extraer_tipo_documento(texto) == "08" + + def test_recibo_honorarios(self): + texto = "RECIBO POR HONORARIOS\nRUC: 10456789012" + assert extraer_tipo_documento(texto) == "14" + + def test_guia_remision(self): + texto = "GUIA DE REMISION ELECTRONICA\nRUC: 20549758054" + assert extraer_tipo_documento(texto) == "09" + + def test_tipo_desconocido(self): + texto = "DOCUMENTO INVALIDO\nRUC: 20549758054" + assert extraer_tipo_documento(texto) == "99" + + +class TestExtraerSerieNumero: + def test_serie_factura_guion(self): + texto = "FACTURA ELECTRONICA\nF001-00005103\nRUC: 20549758054" + assert extraer_serie_numero(texto) == "F001-00005103" + + def test_serie_boleta_guion(self): + texto = "BOLETA ELECTRONICA\nB001-123456\nRUC: 20549758054" + assert extraer_serie_numero(texto) == "B001-123456" + + def test_serie_eg_con_guion(self): + texto = "GUIA DE REMISION\nEG07-2367\nRUC: 20549758054" + assert extraer_serie_numero(texto) == "EG07-2367" + + def test_sin_serie(self): + texto = "Documento sin serie" + assert extraer_serie_numero(texto) == "No encontrado" + + +PDF_REAL = os.environ.get("TEST_PDF_PATH") + + +@pytest.mark.skipif(not PDF_REAL or not os.path.exists(PDF_REAL), + reason="Define TEST_PDF_PATH con la ruta a un PDF real. Ej: TEST_PDF_PATH=factura.pdf pytest ...") +def test_pdf_real_con_metadata(): + """Extrae metadata de un PDF real. Usar: TEST_PDF_PATH=/ruta/pdf python3 -m pytest tests/ -s""" + datos = obtener_datos_pdf(PDF_REAL) + assert datos["valido"] is True, f"Error: {datos.get('error')}" + print(f"\n{'='*50}") + print(f"Archivo: {PDF_REAL}") + print(f"{'='*50}") + print(f" RUC: {datos['ruc']}") + print(f" Tipo doc: {datos['tipo_doc']}") + print(f" Serie-Numero: {datos['id']}") + print(f"{'='*50}")