cargando primer proyecto a gitea
This commit is contained in:
George committed
commit
533105e7c3
16 files changed
+1327
No files matched your search
@@ -0,0 +1,7 @@
|
||||
__pycache__/
|
||||
.env/
|
||||
adjuntos/
|
||||
logs/
|
||||
output_json/
|
||||
resultado/
|
||||
.DS_Store
|
||||
+38
@@ -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"
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -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
|
||||
)
|
||||
|
||||
########################################################
|
||||
@@ -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]
|
||||
@@ -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
|
||||
########################################################
|
||||
@@ -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
|
||||
|
||||
########################################################
|
||||
@@ -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 []
|
||||
|
||||
########################################################
|
||||
|
||||
+188
@@ -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}")
|
||||
|
||||
########################################################
|
||||
@@ -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).")
|
||||
@@ -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)
|
||||
+53
@@ -0,0 +1,53 @@
|
||||
```mermaid
|
||||
graph TD
|
||||
Start((<b>INICIO</b><br/>main.py)) --> Init[Configurar Logger y<br/>Base de Datos SQLITE]
|
||||
Init --> Auth[Obtener Token de Acceso<br/>Outlook API]
|
||||
|
||||
subgraph MailConfigLoop [Bucle por cada Cuenta de Correo en config.json]
|
||||
direction TB
|
||||
GetEmails[Consultar correos del día<br/>outlook_utilities.get_emails]
|
||||
GetEmails --> FilterNew[Filtrar correos no procesados<br/>email_services.filtrar_correos_nuevos]
|
||||
|
||||
subgraph EmailLoop [Bucle por cada Correo Nuevo con Adjuntos]
|
||||
direction TB
|
||||
Download[Descargar adjuntos a /adjuntos/temp<br/>outlook_utilities.download_attachments]
|
||||
Download --> Unzip[Descomprimir ZIPs y filtrar archivos<br/>email_services.procesar_adjuntos_email]
|
||||
|
||||
subgraph DataExtraction [Extracción de Metadatos]
|
||||
direction LR
|
||||
ParseXML[Parsear XML<br/>document_processor.obtener_datos_xml]
|
||||
ParsePDF[Parsear PDF multilínea<br/>pdf_utilities.obtener_datos_pdf]
|
||||
end
|
||||
|
||||
Unzip --> DataExtraction
|
||||
|
||||
DataExtraction --> Grouping[Lógica de Agrupación y Renombrado<br/>file_manager.procesar_renombrado_agrupado]
|
||||
|
||||
Grouping --> Rename[Renombrado físico de archivos<br/>F001.xml -> RUC-01-F001.xml]
|
||||
|
||||
Rename --> MoveLocal[Mover archivos agrupados a /resultado/<br/>file_manager.mover_archivos_a_resultado]
|
||||
end
|
||||
|
||||
FilterNew --> EmailLoop
|
||||
|
||||
EmailLoop --> SMBSync[<b>Sincronización SMB</b><br/>smb_utils.sincronizar_a_smb]
|
||||
|
||||
subgraph SMBSteps [Proceso SMB]
|
||||
direction LR
|
||||
Mount[Montar Unidad según SO<br/>mount_smbfs / net use] --> Rsync[Ejecutar rsync -av<br/>--remove-source-files]
|
||||
end
|
||||
|
||||
SMBSync --- SMBSteps
|
||||
end
|
||||
|
||||
Auth --> MailConfigLoop
|
||||
|
||||
MailConfigLoop --> ExportJSON[Exportar resultados finales a JSON<br/>/output_json/YYYY-MM-DD.json]
|
||||
ExportJSON --> End((<b>FIN</b>))
|
||||
|
||||
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
|
||||
```
|
||||
@@ -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}")
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
########################################################
|
||||
@@ -0,0 +1,3 @@
|
||||
requests
|
||||
msal
|
||||
pdfplumber
|
||||
+126
@@ -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
|
||||
|
||||
########################################################
|
||||
Reference in new issue
Block a user