georgeandClaude Sonnet 5
15e3b5079c
- hana_manager.get_delivery ahora busca en ODLN y, si no encuentra, en OWTR - sapb1_manager resuelve StockTransfers -> Orders (via U_IDC_NUMOV) para validar precios de flete cuando la guia es un traslado de inventario - se corrige el orden de parametros tipo/serie/numero (sin cruces) en get_delivery, get_delivery_notes y _get_delivery_entries - se agrega sapb1_manager.obtener_condicion_pago (BusinessPartners + PaymentTermsTypes) y pipeline._calcular_fecha_vencimiento para que DocDueDate del draft sea fecha de contabilizacion + dias de la condicion de pago del proveedor - se agrega variante de regex para "VALOR REFER" truncado en sunat_schema.py - se actualiza documentacion (docs/) reflejando estos cambios Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
53 lines
2.0 KiB
Python
53 lines
2.0 KiB
Python
import os
|
|
import zipfile
|
|
|
|
# ============================================================================
|
|
def descomprimir_adjuntos(emails: list[dict], ruta: str) -> list[dict]:
|
|
"""
|
|
Busca adjuntos ZIP, los descomprime y agrega los archivos
|
|
extraídos a la lista de adjuntos del correo.
|
|
"""
|
|
|
|
try:
|
|
for email in emails:
|
|
nuevos_adjuntos = []
|
|
for adjunto in list(email["adjuntos"]):
|
|
if not adjunto.lower().endswith(".zip"):
|
|
continue
|
|
ruta_zip = os.path.join(ruta, adjunto)
|
|
if not os.path.exists(ruta_zip):
|
|
print(f"⚠️ No existe: {ruta_zip}")
|
|
continue
|
|
try:
|
|
with zipfile.ZipFile(ruta_zip, "r") as zip_ref:
|
|
nombres_extraidos = [
|
|
nombre
|
|
for nombre in zip_ref.namelist()
|
|
if not nombre.endswith("/")
|
|
]
|
|
zip_ref.extractall(ruta)
|
|
nuevos_adjuntos.extend(nombres_extraidos)
|
|
print(
|
|
f"📦 {adjunto}: {len(nombres_extraidos)} archivos extraídos"
|
|
)
|
|
|
|
except Exception as e:
|
|
print(f"⚠️ Error descomprimiendo {adjunto}: {e}")
|
|
email["adjuntos"].extend(nuevos_adjuntos)
|
|
return emails
|
|
except Exception as ex:
|
|
print(f"❌ Error general: {ex}")
|
|
return emails
|
|
|
|
# ============================================================================
|
|
def filtrar_archivos_validos(archivos: list[str]) -> list[str]:
|
|
|
|
"""
|
|
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)]
|
|
|
|
# ============================================================================
|
|
|