commit 533105e7c376285024fa26ecaf52146c77c074e2 Author: George Date: Fri May 15 16:39:08 2026 -0500 cargando primer proyecto a gitea diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..eb73e0d --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +__pycache__/ +.env/ +adjuntos/ +logs/ +output_json/ +resultado/ +.DS_Store diff --git a/config.json b/config.json new file mode 100644 index 0000000..efbd73e --- /dev/null +++ b/config.json @@ -0,0 +1,38 @@ +{ + "hana": { + "server": "192.168.1.175", + "port": 30015, + "user": "SYSTEM", + "password": "SAPB1Admin.25" + }, + "outlook": { + "client_id": "957c324b-3da3-43ed-b912-ca52fb5d3128", + "client_secret": "Guc8Q~u6FH3xBKpxJ-hQ7QSpRWbnvmX.LnWr-dw7", + "tenant_id": "76f77b2f-6fc3-42a0-ae83-df05508405d6", + "download_folder": "./adjuntos", + "verify_ssl": false, + "ca_bundle": null + }, + "smb": { + "server": "192.168.1.231", + "username": "tgmtgm", + "password": "..Pesare33Waras", + "share": "FileSAP", + "mount_points": { + "windows": "S:", + "linux": "/mnt/FileSAP", + "mac": "/Users/george/FileSAP" + } + }, + "mails": [ + { + "email": "cpe@tagumedica.com", + "ruta_descarga": "./adjuntos/tagumedica", + "ruta_final": "./resultado/tagumedica", + "base_datos": "SBO_TAGUMEDICA_21", + "usuario": "manager", + "password": "m1r1", + "smb_folder": "/COMPRAS/TAGUMEDICA/ADJUNTOS_CORREOS/2025-03" + } + ] +} \ No newline at end of file diff --git a/config_loader.py b/config_loader.py new file mode 100644 index 0000000..11ab3ab --- /dev/null +++ b/config_loader.py @@ -0,0 +1,24 @@ +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: + data = json.load(f) + +# Instantiate dataclasses +hana = HanaConfig(**data.get('hana', {})) +outlook = OutlookConfig(**data.get('outlook', {})) +smb = SmbConfig(**data.get('smb', {})) +mails = [MailConfig(**c) for c in data.get('mails', [])] + +CONFIG = Config( + hana=hana, + outlook=outlook, + smb=smb, + mails=mails +) + +######################################################## \ No newline at end of file diff --git a/config_model.py b/config_model.py new file mode 100644 index 0000000..1b81cad --- /dev/null +++ b/config_model.py @@ -0,0 +1,43 @@ +from dataclasses import dataclass, field +from typing import List, Optional, Dict + +@dataclass +class HanaConfig: + server: str + port: int + user: str + password: str + +@dataclass +class OutlookConfig: + client_id: str + client_secret: str + tenant_id: str + download_folder: str + verify_ssl: bool + ca_bundle: Optional[str] + +@dataclass +class SmbConfig: + server: str + username: str + password: str + share: str + mount_points: Dict[str, str] + +@dataclass +class MailConfig: + email: str + ruta_descarga: str + ruta_final: str + base_datos: str + usuario: str + password: str + smb_folder: str + +@dataclass +class Config: + hana: HanaConfig + outlook: OutlookConfig + smb: SmbConfig + mails: List[MailConfig] diff --git a/database_utilities.py b/database_utilities.py new file mode 100644 index 0000000..7d10c44 --- /dev/null +++ b/database_utilities.py @@ -0,0 +1,110 @@ +import sqlite3 + +DB_NAME = "emails.db" + +######################################################## +def init_db(): + """ + Inicializa la base de datos SQLite y crea la tabla de correos si no existe. + """ + conn = sqlite3.connect(DB_NAME) + cursor = conn.cursor() + + # Crear tabla con id como llave primaria para evitar duplicados + cursor.execute(''' + CREATE TABLE IF NOT EXISTS emails ( + id TEXT PRIMARY KEY, + fecha TEXT, + sender TEXT, + subject TEXT, + has_attach INTEGER, + fecha_descarga TEXT + ) + ''') + + conn.commit() + conn.close() + +######################################################## +def save_emails(emails_list): + """ + Guarda una lista de correos en la base de datos. + Retorna la cantidad de registros nuevos insertados. + """ + if not emails_list: + return 0 + + conn = sqlite3.connect(DB_NAME) + cursor = conn.cursor() + + from datetime import datetime + fecha_ahora = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + + # Mapear lista de diccionarios a tuplas para executemany + # has_attach se guarda como 1 (True) o 0 (False) + data = [ + ( + e['id'], + e['fecha'], + e['sender'], + e['subject'], + 1 if e['has_attach'] else 0, + fecha_ahora + ) + for e in emails_list + ] + + # INSERT OR IGNORE evita errores si el ID ya existe + cursor.executemany(''' + INSERT OR IGNORE INTO emails (id, fecha, sender, subject, has_attach, fecha_descarga) + VALUES (?, ?, ?, ?, ?, ?) + ''', data) + + conn.commit() + rows_affected = cursor.rowcount + 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. + """ + if not ids_list: + return [] + + conn = sqlite3.connect(DB_NAME) + cursor = conn.cursor() + + existing_ids = [] + # Procesar por lotes para evitar límites de parámetros en SQLite + for i in range(0, len(ids_list), 900): + chunk = ids_list[i:i + 900] + placeholders = ','.join(['?'] * len(chunk)) + query = f"SELECT id FROM emails WHERE id IN ({placeholders})" + cursor.execute(query, chunk) + existing_ids.extend([row[0] for row in cursor.fetchall()]) + + conn.close() + return existing_ids + +######################################################## +def delete_email_by_id(email_id): + """ + Elimina un correo de la base de datos por su ID. + Retorna True si se eliminó el registro, False si no se encontró. + """ + if not email_id: + return False + + conn = sqlite3.connect(DB_NAME) + cursor = conn.cursor() + + cursor.execute("DELETE FROM emails WHERE id = ?", (email_id,)) + + conn.commit() + rows_affected = cursor.rowcount + conn.close() + + return rows_affected > 0 +######################################################## \ No newline at end of file diff --git a/document_processor.py b/document_processor.py new file mode 100644 index 0000000..801415a --- /dev/null +++ b/document_processor.py @@ -0,0 +1,165 @@ +import os +import xml.etree.ElementTree as ET +import pdf_utilities + +######################################################## +def formatear_serie_numero(serie_num): + """ + Remueve los ceros a la izquierda de la parte numérica de la serie-número. + Ejemplo: F001-00005103 -> F001-5103 + """ + if "-" in serie_num: + partes = serie_num.split("-") + if len(partes) == 2: + try: + # El strip('0') no sirve si el número es '000', mejor usar int() + numero_limpio = str(int(partes[1])) + return f"{partes[0]}-{numero_limpio}" + except: + 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. + """ + # 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"]: + 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. + Usa códigos numéricos para el tipo de documento. + """ + try: + tree = ET.parse(ruta_xml) + root = tree.getroot() + + tag_completo = root.tag + root_tag = tag_completo.split('}')[-1] if '}' in tag_completo else tag_completo + + ns = { + 'cbc': 'urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2', + 'cac': 'urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2', + 'ar': 'urn:oasis:names:specification:ubl:schema:xsd:ApplicationResponse-2' + } + + # Identificación de RUC primero (necesario para validaciones de tipo) + if root_tag == "ApplicationResponse": + ruc_emisor = root.find('.//cac:SenderParty/cac:PartyLegalEntity/cbc:CompanyID', ns) + else: + # Intentar ruta estándar + ruc_emisor = root.find('.//cac:AccountingSupplierParty/cac:Party/cac:PartyIdentification/cbc:ID', ns) + # Intentar CustomerAssignedAccountID (común en Recibos por Honorarios) + if ruc_emisor is None: + ruc_emisor = root.find('.//cac:AccountingSupplierParty/cbc:CustomerAssignedAccountID', ns) + # Intentar ruta de Firma + if ruc_emisor is None: + ruc_emisor = root.find('.//cac:Signature/cac:SignatoryParty/cac:PartyIdentification/cbc:ID', ns) + + ruc = ruc_emisor.text if ruc_emisor is not None else "Desconocido" + + # Tipificación por códigos SUNAT + root_tag_original = root_tag + if root_tag == "ApplicationResponse": + root_tag = "CDR" + elif root_tag in ["Invoice", "CreditNote", "DebitNote"]: + type_code_elem = root.find('cbc:InvoiceTypeCode', ns) + if type_code_elem is not None: + root_tag = type_code_elem.text + else: + # Fallback por tag raíz si no existe InvoiceTypeCode + if root_tag == "CreditNote": root_tag = "07" + elif root_tag == "DebitNote": root_tag = "08" + else: root_tag = "99" + + # Validación especial para Recibo por Honorarios (02) + # Solo si el root es Invoice y el RUC empieza con 10 + if root_tag_original == "Invoice" and ruc.startswith("10"): + for tax_id in root.findall('.//cac:TaxCategory/cbc:ID', ns): + if tax_id is not None and tax_id.text == "RET 4TA": + root_tag = "02" + break + else: + root_tag = "99" + + doc_id = root.find('cbc:ID', ns) + doc_id = doc_id.text if doc_id is not None else "Desconocido" + + return { + "tipo_xml": root_tag, + "id": doc_id, + "ruc": ruc, + "valido": True + } + 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). + Aplica formateo de serie-número y generación de nombre propuesto. + Fuerza tipos no permitidos a '99'. + """ + archivos = email_obj.get('archivos', []) + metadatos_adjuntos = [] + + # Tipos permitidos oficialmente (incluimos CDR por requerimiento) + tipos_ok = ["01", "02", "03", "07", "08", "CDR"] + + xmls = [a for a in archivos if a.lower().endswith('.xml')] + pdfs = [a for a in archivos if a.lower().endswith('.pdf')] + + for xml in xmls: + ruta = os.path.join(ruta_base, xml) + datos = obtener_datos_xml(ruta) + if datos['valido']: + tipo_final = datos['tipo_xml'] if datos['tipo_xml'] in tipos_ok else "99" + m = { + "archivo": xml, + "extension": ".xml", + "tipo": tipo_final, + "serie_numero": formatear_serie_numero(datos['id']), + "ruc": datos['ruc'], + "fuente": "XML" + } + m["archivo_propuesto"] = generar_nombre_propuesto(m) + metadatos_adjuntos.append(m) + + for pdf in pdfs: + ruta = os.path.join(ruta_base, pdf) + datos = pdf_utilities.obtener_datos_pdf(ruta) + + if datos['valido']: + tipo_final = datos['tipo_doc'] if datos['tipo_doc'] in tipos_ok else "99" + m = { + "archivo": pdf, + "extension": ".pdf", + "tipo": tipo_final, + "serie_numero": formatear_serie_numero(datos['id']), + "ruc": datos['ruc'], + "fuente": "PDF" + } + m["archivo_propuesto"] = generar_nombre_propuesto(m) + metadatos_adjuntos.append(m) + else: + metadatos_adjuntos.append({ + "archivo": pdf, + "extension": ".pdf", + "tipo": "99", + "serie_numero": "N/A", + "ruc": "N/A", + "fuente": "PDF_FALLIDO", + "archivo_propuesto": None + }) + + return metadatos_adjuntos + + ######################################################## \ No newline at end of file diff --git a/email_services.py b/email_services.py new file mode 100644 index 0000000..5ee95f7 --- /dev/null +++ b/email_services.py @@ -0,0 +1,80 @@ +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. + """ + 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)}") + + if nuevos: + 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 + +######################################################## +def descomprimir_adjuntos(archivos, ruta): + """ + Descomprime archivos .zip en la ruta especificada y actualiza la lista de archivos. + Remueve el .zip original de la lista y añade los contenidos extraídos. + """ + archivos_finales = [] + for archivo in archivos: + ruta_archivo = os.path.join(ruta, archivo) + if archivo.lower().endswith('.zip') and os.path.exists(ruta_archivo): + try: + with zipfile.ZipFile(ruta_archivo, 'r') as zip_ref: + nombres_extraidos = zip_ref.namelist() + zip_ref.extractall(ruta) + archivos_finales.extend(nombres_extraidos) + except Exception as e: + print(f" ⚠️ Error descomprimiendo {archivo}: {e}") + archivos_finales.append(archivo) + else: + 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). + """ + 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. + Retorna la lista de archivos válidos si existen, de lo contrario lista vacía. + """ + # 1. Descomprimir archivos ZIP si existen + archivos_procesados = descomprimir_adjuntos(archivos, ruta_descarga) + + # 2. Filtrar solo PDF y XML + archivos_validos = filtrar_archivos_validos(archivos_procesados) + + if archivos_validos: + e['archivos'] = archivos_validos + return archivos_validos + return [] + +######################################################## + diff --git a/emails.db b/emails.db new file mode 100644 index 0000000..af80789 Binary files /dev/null and b/emails.db differ diff --git a/file_manager.py b/file_manager.py new file mode 100644 index 0000000..9833773 --- /dev/null +++ b/file_manager.py @@ -0,0 +1,188 @@ +import os + +######################################################## +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 procesar_renombrado_agrupado(email_obj, ruta_base): + """ + Lógica de agrupación y renombrado de archivos según puntuación y tipo. + """ + 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) + + # 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}") + + # 4. Actualizar la lista de archivos en el email_obj + email_obj['archivos'] = [m['archivo'] for m in metadatos] + return agrupacion + +######################################################## +def mover_archivos_a_resultado(email_obj, 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'. + """ + 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}") + +######################################################## +def limpiar_carpeta_descarga(ruta): + """ + Elimina todos los archivos dentro de la carpeta de descarga para dejarla limpia. + """ + if not os.path.exists(ruta): + return + + print(f" 🧹 Limpiando carpeta de descarga: {ruta}") + for archivo in os.listdir(ruta): + 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}") + +######################################################## \ No newline at end of file diff --git a/logger_utilities.py b/logger_utilities.py new file mode 100644 index 0000000..6d05b9e --- /dev/null +++ b/logger_utilities.py @@ -0,0 +1,70 @@ +import sys +import os +from datetime import datetime, timedelta + +######################################################## +class MultiStream: + """ + Clase para redirigir la salida a múltiples flujos (consola y archivo). + """ + def __init__(self, *streams): + self.streams = streams + + def write(self, data): + for stream in self.streams: + stream.write(data) + stream.flush() + + def flush(self): + 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. + Retorna la ruta del archivo de log creado. + """ + log_dir = "logs" + if not os.path.exists(log_dir): + os.makedirs(log_dir) + + # Nombre del archivo: YYYY-MM-DD_HH-MM-SS.log + timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") + log_filename = f"{timestamp}.log" + log_path = os.path.join(log_dir, log_filename) + + # Abrir archivo en modo append con encoding utf-8 para soportar iconos + log_file = open(log_path, "a", encoding="utf-8") + + # Redirigir stdout y stderr + sys.stdout = MultiStream(sys.stdout, log_file) + sys.stderr = MultiStream(sys.stderr, log_file) + + return log_path + +######################################################## +def limpiar_logs_antiguos(dias=7): + """ + Elimina archivos de log con más de 'n' días de antigüedad. + """ + log_dir = "logs" + if not os.path.exists(log_dir): + return + + limite = datetime.now() - timedelta(days=dias) + count = 0 + + for archivo in os.listdir(log_dir): + ruta_archivo = os.path.join(log_dir, archivo) + if os.path.isfile(ruta_archivo) and archivo.endswith(".log"): + fecha_archivo = datetime.fromtimestamp(os.path.getmtime(ruta_archivo)) + if fecha_archivo < limite: + try: + os.remove(ruta_archivo) + count += 1 + except Exception as e: + print(f"⚠️ Error al eliminar log {archivo}: {e}") + + if count > 0: + print(f"🧹 Se eliminaron {count} archivos de log antiguos (más de {dias} días).") diff --git a/main.py b/main.py new file mode 100644 index 0000000..15da35a --- /dev/null +++ b/main.py @@ -0,0 +1,172 @@ +import sys +import json +import os +import outlook_utilities +import database_utilities +import email_services +import logger_utilities +import document_processor +import file_manager +from datetime import datetime, timedelta +from config_loader import CONFIG + + +######################################################## +if __name__ == "__main__": + + # Configurar Logger (Redirige stdout a archivo y consola) + log_path = logger_utilities.setup_logger() + logger_utilities.limpiar_logs_antiguos(dias=30) + + 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) + + # Inicializar Base de Datos + database_utilities.init_db() + + # Calcular fechas + hoy = datetime.now() + ayer = hoy - timedelta(days=1) + + 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}") + + 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) + + todos_procesados = [] + + for mail in CONFIG.mails: + print("\n" + "="*60) + print(f"[MAIL] {mail.email}") + + emails = outlook_utilities.get_emails(mail.email, outlook_token, fecha_inicio, fecha_fin) + print(f"Se encontraron {len(emails)} correos.") + + # Filtrar y guardar usando el servicio de emails + emails_a_procesar = email_services.filtrar_correos_nuevos(emails) + + if emails_a_procesar: + print(f"Descargando adjuntos en: {mail.ruta_descarga}") + emails_finales = [] # Lista definitiva de correos con adjuntos válidos + + 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}") + + # 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']}") + + 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}") + + # 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}") + + # 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) + + # 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) diff --git a/mermaid.md b/mermaid.md new file mode 100644 index 0000000..1b7ab5a --- /dev/null +++ b/mermaid.md @@ -0,0 +1,53 @@ +```mermaid +graph TD + Start((INICIO
main.py)) --> Init[Configurar Logger y
Base de Datos SQLITE] + Init --> Auth[Obtener Token de Acceso
Outlook API] + + subgraph MailConfigLoop [Bucle por cada Cuenta de Correo en config.json] + direction TB + GetEmails[Consultar correos del día
outlook_utilities.get_emails] + GetEmails --> FilterNew[Filtrar correos no procesados
email_services.filtrar_correos_nuevos] + + subgraph EmailLoop [Bucle por cada Correo Nuevo con Adjuntos] + direction TB + Download[Descargar adjuntos a /adjuntos/temp
outlook_utilities.download_attachments] + Download --> Unzip[Descomprimir ZIPs y filtrar archivos
email_services.procesar_adjuntos_email] + + subgraph DataExtraction [Extracción de Metadatos] + direction LR + ParseXML[Parsear XML
document_processor.obtener_datos_xml] + ParsePDF[Parsear PDF multilínea
pdf_utilities.obtener_datos_pdf] + end + + Unzip --> DataExtraction + + DataExtraction --> Grouping[Lógica de Agrupación y Renombrado
file_manager.procesar_renombrado_agrupado] + + Grouping --> Rename[Renombrado físico de archivos
F001.xml -> RUC-01-F001.xml] + + Rename --> MoveLocal[Mover archivos agrupados a /resultado/
file_manager.mover_archivos_a_resultado] + end + + FilterNew --> EmailLoop + + EmailLoop --> SMBSync[Sincronización SMB
smb_utils.sincronizar_a_smb] + + subgraph SMBSteps [Proceso SMB] + direction LR + Mount[Montar Unidad según SO
mount_smbfs / net use] --> Rsync[Ejecutar rsync -av
--remove-source-files] + end + + SMBSync --- SMBSteps + end + + Auth --> MailConfigLoop + + MailConfigLoop --> ExportJSON[Exportar resultados finales a JSON
/output_json/YYYY-MM-DD.json] + ExportJSON --> End((FIN)) + + style Start fill:#f9f,stroke:#333,stroke-width:2px + style End fill:#f9f,stroke:#333,stroke-width:2px + style MailConfigLoop fill:#e1f5fe,stroke:#01579b,stroke-dasharray: 5 5 + style EmailLoop fill:#fff3e0,stroke:#e65100 + style SMBSync fill:#c8e6c9,stroke:#2e7d32,stroke-width:2px +``` diff --git a/outlook_utilities.py b/outlook_utilities.py new file mode 100644 index 0000000..9473b4b --- /dev/null +++ b/outlook_utilities.py @@ -0,0 +1,165 @@ +import msal +import requests +import warnings +import base64 +import os +from datetime import datetime, timedelta +from config_loader import CONFIG + +warnings.filterwarnings("ignore") + +######################################################## +def get_access_token(): + """ + Obtiene un token de acceso de Microsoft Graph usando las credenciales + configuradas en config.json (Client ID, Client Secret y Tenant ID). + """ + if not CONFIG or not CONFIG.outlook: + raise ValueError("La configuración de Outlook no está cargada correctamente.") + + authority = f"https://login.microsoftonline.com/{CONFIG.outlook.tenant_id}" + + app = msal.ConfidentialClientApplication( + CONFIG.outlook.client_id, + authority=authority, + client_credential=CONFIG.outlook.client_secret, + verify=CONFIG.outlook.verify_ssl + ) + + scopes = ["https://graph.microsoft.com/.default"] + result = app.acquire_token_for_client(scopes=scopes) + + if "access_token" in result: + return result["access_token"] + else: + error = result.get("error") + error_description = result.get("error_description") + 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. + Retorna una lista de JSON con: id, fecha (ajustada -5h), sender, subject y has_attach. + """ + headers = { + "Authorization": f"Bearer {token}", + "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) + + filter_query = ( + f"receivedDateTime ge {dt_inicio_utc.strftime('%Y-%m-%dT%H:%M:%SZ')} and " + f"receivedDateTime lt {dt_fin_utc.strftime('%Y-%m-%dT%H:%M:%SZ')}" + ) + + params = { + "$filter": filter_query, + "$select": "id,receivedDateTime,hasAttachments,subject,from", + "$top": 50 + } + + url = f"https://graph.microsoft.com/v1.0/users/{user_email}/messages" + all_mails = [] + + while url: + response = requests.get( + url, + headers=headers, + params=params, + verify=CONFIG.outlook.verify_ssl + ) + + if response.status_code != 200: + raise Exception(f"Error Graph API ({user_email}): {response.status_code} - {response.text}") + + data = response.json() + messages = data.get("value", []) + + 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) + + from_obj = msg.get("from", {}) + sender = from_obj.get("emailAddress", {}).get("address", "Desconocido") + + all_mails.append({ + "id": msg.get("id"), + "fecha": dt_local.strftime("%Y-%m-%d %H:%M:%S"), + "sender": sender, + "subject": msg.get("subject", ""), + "has_attach": msg.get("hasAttachments") + }) + + url = data.get("@odata.nextLink") + params = None + + 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. + Retorna una lista con los nombres de los archivos descargados. + """ + if not os.path.exists(download_path): + os.makedirs(download_path, exist_ok=True) + + headers = { + "Authorization": f"Bearer {token}", + "Content-Type": "application/json" + } + + url = f"https://graph.microsoft.com/v1.0/users/{user_email}/messages/{message_id}/attachments" + + response = requests.get(url, headers=headers, verify=CONFIG.outlook.verify_ssl) + + if response.status_code != 200: + raise Exception(f"Error al obtener adjuntos ({message_id}): {response.status_code} - {response.text}") + + attachments = response.json().get("value", []) + downloaded_files = [] + + for attachment in attachments: + if attachment.get("@odata.type") == "#microsoft.graph.fileAttachment": + file_name = attachment.get("name") + content_bytes = attachment.get("contentBytes") + + if content_bytes: + file_path = os.path.join(download_path, file_name) + with open(file_path, "wb") as f: + f.write(base64.b64decode(content_bytes)) + downloaded_files.append(file_name) + + return downloaded_files + +######################################################## +if __name__ == "__main__": + try: + print("Intentando obtener token de acceso...") + token = get_access_token() + print("✅ Token obtenido exitosamente.") + + if CONFIG.mails: + test_email = CONFIG.mails[0].email + hoy = datetime.now() + ayer = hoy - timedelta(days=1) + f_inicio = ayer.strftime("%Y-%m-%d") + f_fin = hoy.strftime("%Y-%m-%d") + + print(f"\nBuscando correos para {test_email}...") + emails = get_emails(test_email, token, f_inicio, f_fin) + print(f"✅ Se encontraron {len(emails)} correos.") + + for e in emails: + if e['has_attach']: + print(f"Probando descarga para correo ID: {e['id'][:10]}...") + archivos = download_attachments(test_email, e['id'], token, "./test_download") + print(f"Descargados: {archivos}") + break + except Exception as e: + print(f"❌ Error: {e}") \ No newline at end of file diff --git a/pdf_utilities.py b/pdf_utilities.py new file mode 100644 index 0000000..0ac9cc9 --- /dev/null +++ b/pdf_utilities.py @@ -0,0 +1,83 @@ +import pdfplumber +import re +import os + +######################################################## +def extraer_texto_pdf(ruta_pdf): + """Extrae todo el texto de un PDF usando pdfplumber.""" + texto_completo = "" + try: + with pdfplumber.open(ruta_pdf) as pdf: + for page in pdf.pages: + texto_completo += page.extract_text() + "\n" + return texto_completo + except Exception as e: + 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... + patrones = [ + r'R\.?U\.?C\.?\s*[:\-\s]\s*((?:10|20)\d{9})', # R.U.C: 20... + r'\b((?:10|20)\d{9})\b' # 11 dígitos sueltos que empiecen con 10 o 20 + ] + + for patron in patrones: + match = re.search(patron, texto, re.IGNORECASE) + if match: + return match.group(1) + return "N/A" + +######################################################## +def extraer_tipo_documento(texto): + """Identifica el tipo de documento SUNAT basándose en el texto usando regex multilínea.""" + 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'] + } + + 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)}" + + # 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)}" + + return "No encontrado" + +######################################################## +def obtener_datos_pdf(ruta_pdf): + """Orquestador para extraer metadata de un PDF.""" + texto = extraer_texto_pdf(ruta_pdf) + if not texto: + return {"valido": False, "error": "No se pudo extraer texto"} + + return { + "valido": True, + "tipo_doc": extraer_tipo_documento(texto), + "ruc": extraer_ruc(texto), + "id": extraer_serie_numero(texto) + } + +######################################################## \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..7e2e112 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +requests +msal +pdfplumber \ No newline at end of file diff --git a/smb_utils.py b/smb_utils.py new file mode 100644 index 0000000..258419e --- /dev/null +++ b/smb_utils.py @@ -0,0 +1,126 @@ +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 + +########################################################