george/descarga_factura_transporte
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

se divide pipeline.py en módulos (adjuntos, documentos, persistencia, draft_sap, notificaciones)

Browse Source 
Las funciones anidadas de _procesar_un_correo pasan a funciones de módulo
con parámetros explícitos; pipeline.py queda solo como orquestador.
copiar_a_red lee las variables de entorno al ejecutarse. Se actualizan docs.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
GeorgeandClaude Sonnet 5.5 committed 2026-10-20 22:03:38 -05:00
1 parent c5210c8a57
commit 6cc5e03209
9 files changed
+1309 -1228
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

adjuntos.py
+81
View File
@@ -0,0 +1,81 @@
"""Descarga y filtrado de adjuntos de un correo."""
import os
import outlook_utilities
from pipeline_utils import log, ruta_carpeta
# ============================================================================
def descargar_adjuntos(email: dict, email_addr: str, token: str, download_folder: str) -> list[str] | None:
"""
Descarga todos los adjuntos del correo usando Microsoft Graph.
Almacena la lista de nombres de archivo en ``email["adjuntos"]``.
Returns:
Lista de nombres de archivo descargados, o None si no hay adjuntos
o hubo un error.
Example:
Si el correo tiene ``factura.xml`` y ``guia.pdf``::
adjuntos = descargar_adjuntos(email, email_addr, token, download_folder)
# → ["factura.xml", "guia.pdf"]
# email["adjuntos"] = ["factura.xml", "guia.pdf"]
"""
try:
adjuntos = outlook_utilities.download_attachments(
email_addr, email["id"], token, download_folder
)
if adjuntos:
email["adjuntos"] = adjuntos
return adjuntos
return None
except Exception as ex:
log(f"❌ Error descargando adjuntos: {ex}")
return None
# ============================================================================
def filtrar_y_limpiar(email: dict, download_folder: str) -> bool:
"""
Filtra los adjuntos conservando solo archivos PDF y XML.
- Los archivos que no son PDF ni XML se eliminan del disco.
- Si no queda ningún PDF/XML válido, se eliminan todos los archivos
descargados y se retorna False.
- Actualiza ``email["adjuntos"]`` con la lista filtrada.
Returns:
True si hay al menos un PDF o XML, False si no queda ninguno.
Example:
Adjuntos originales: ``[img.png, factura.xml, guia.pdf]``::
ok = filtrar_y_limpiar(email, download_folder)
# → img.png eliminado
# → ok = True
# → email["adjuntos"] = ["factura.xml", "guia.pdf"]
"""
todos = email.get("adjuntos", [])
validos = [a for a in todos if a.lower().endswith((".pdf", ".xml"))]
no_validos = [a for a in todos if a not in validos]
carpeta = ruta_carpeta(download_folder)
if not validos:
for a in todos:
ruta = os.path.join(carpeta, a)
if os.path.exists(ruta):
os.remove(ruta)
log("🗑️ Correo sin PDF/XML válidos, adjuntos eliminados")
return False
for a in no_validos:
ruta = os.path.join(carpeta, a)
if os.path.exists(ruta):
os.remove(ruta)
email["adjuntos"] = validos
return True
docs/inventario.md
+53 -36
View File
@@ -5,7 +5,13 @@
| Archivo | Líneas | Rol | | Archivo | Líneas | Rol |
|---------|--------|-----| |---------|--------|-----|
| `main.py` | 134 | Demonio orquestador (loop de horarios) | | `main.py` | 134 | Demonio orquestador (loop de horarios) |
| `pipeline.py` | 1284 | Pipeline core (procesar un correo completo) | | `pipeline.py` | 230 | Orquestador del pipeline (autenticar, descargar correos, procesar un correo) |
| `pipeline_utils.py` | 15 | `log()` con indentación de árbol y `ruta_carpeta()` |
| `adjuntos.py` | 81 | Descarga de adjuntos (Graph) y filtrado PDF/XML |
| `documentos.py` | 371 | Extracción PDF/XML, asignación de guías, copia a red |
| `persistencia.py` | 163 | Guardado de JSON de resultados y de drafts |
| `draft_sap.py` | 413 | Validación de precios contra SAP, valor referencial y construcción del draft |
| `notificaciones.py` | 117 | Cuerpos HTML de los correos de notificación |
| `sunat_schema.py` | 720 | Parseo de XML SUNAT (Invoice, DespatchAdvice) | | `sunat_schema.py` | 720 | Parseo de XML SUNAT (Invoice, DespatchAdvice) |
| `pdf_utilities.py` | 464 | Parseo de PDFs (factura, guías) | | `pdf_utilities.py` | 464 | Parseo de PDFs (factura, guías) |
| `sapb1_manager.py` | 321 | Comunicación con SAP B1 Service Layer | | `sapb1_manager.py` | 321 | Comunicación con SAP B1 Service Layer |
@@ -32,6 +38,17 @@ main.py
├── outlook_utilities.py ├── outlook_utilities.py
├── email_utilities.py ├── email_utilities.py
└── database_utilities.py └── database_utilities.py
```
Módulos que `pipeline.py` orquesta (todos usan `pipeline_utils.py`):
```
pipeline.py
├── adjuntos.py → outlook_utilities
├── documentos.py → sunat_schema, pdf_utilities
├── persistencia.py → sunat_schema, pdf_utilities
├── draft_sap.py → sapb1_manager, hana_manager
└── notificaciones.py
consulta_rapida.py consulta_rapida.py
├── logger_manager.py ├── logger_manager.py
@@ -50,28 +67,28 @@ main()
├── step_fetch_emails() → emails[] ├── step_fetch_emails() → emails[]
└── _procesar_un_correo() (por cada email) └── _procesar_un_correo() (por cada email)
├── BD: verificar duplicado ├── BD: verificar duplicado
├── _descargar_adjuntos() → archivos ├── adjuntos.descargar_adjuntos() → archivos
├── _descomprimir() ZIPs ├── email_utilities.descomprimir_adjuntos() ZIPs
├── _filtrar_y_limpiar() → solo PDF/XML ├── adjuntos.filtrar_y_limpiar() → solo PDF/XML
├── _procesar_pdfs() → pdf_cache ├── documentos.procesar_pdfs() → pdf_cache
│ └── pdf_utilities.extraer_datos_pdf() │ └── pdf_utilities.extraer_datos_pdf()
├── _procesar_xmls(pdf_cache) → documentos[] ├── documentos.procesar_xmls(pdf_cache) → documentos[]
│ ├── sunat_schema.extraer_datos_xml() │ ├── sunat_schema.extraer_datos_xml()
│ └── _copiar_a_red() → copia PDF/XML a ruta de red (RUTA_RED_FACTURACION) │ └── documentos.copiar_a_red() → copia PDF/XML a ruta de red (RUTA_RED_FACTURACION)
├── _completar_guias(documentos, pdf_cache) → completa guías faltantes desde PDF ├── documentos.completar_guias(documentos, pdf_cache) → completa guías faltantes desde PDF
├── _validar_guias(documentos) → advierte guías con datos incompletos ├── documentos.validar_guias(documentos) → advierte guías con datos incompletos
├── _guardar_resultado(documentos) ├── persistencia.guardar_resultado(documentos)
├── _validate_prices(factura) → (ok, resultados) ├── draft_sap.validate_prices(factura) → (ok, resultados)
│ └── sapb1_manager.get_delivery_notes() │ └── sapb1_manager.get_delivery_notes()
│ ├── SL (Service Layer) GET /DeliveryNotes │ ├── SL (Service Layer) GET /DeliveryNotes
│ ├── HANA fallback (get_delivery: ODLN, luego OWTR) │ ├── HANA fallback (get_delivery: ODLN, luego OWTR)
│ └── si vino de OWTR: GET StockTransfers(DocEntry) │ └── si vino de OWTR: GET StockTransfers(DocEntry)
│ → U_IDC_NUMOV → GET Orders(U_IDC_NUMOV) │ → U_IDC_NUMOV → GET Orders(U_IDC_NUMOV)
├── _obtener_valor_referencial_sap(factura) → guias_valor_ref ├── draft_sap.obtener_valor_referencial_sap(factura) → guias_valor_ref
│ └── hana_manager.obtener_valor_referencial() (Sp_ObtenerValorReferencial) │ └── hana_manager.obtener_valor_referencial() (Sp_ObtenerValorReferencial)
├── _build_sap_draft(factura, resultados) → draft (incluye U_VS_ANEXO_PDF/XML) ├── draft_sap.build_sap_draft(factura, resultados) → draft (incluye U_VS_ANEXO_PDF/XML)
│ └── sapb1_manager.obtener_condicion_pago(CardCode) → DocDueDate │ └── sapb1_manager.obtener_condicion_pago(CardCode) → DocDueDate
├── _guardar_draft_json(draft) ├── persistencia.guardar_draft_json(draft)
├── sapb1_manager.create_draft(draft) ├── sapb1_manager.create_draft(draft)
└── outlook_utilities.send_email(notificación) └── outlook_utilities.send_email(notificación)
``` ```
@@ -109,27 +126,27 @@ main()
--- ---
## Funciones internas en `_procesar_un_correo` (pipeline.py) ## Funciones por módulo
> Nota: `_ruta_carpeta` y `_copiar_a_red` son helpers a nivel de módulo (no anidados en `_procesar_un_correo`); el resto son funciones anidadas dentro de `_procesar_un_correo` (líneas 142-1304). > Antes eran funciones anidadas en `_procesar_un_correo`; ahora son funciones de módulo que reciben explícitamente `email`, `download_folder`, `output_folder`, etc. `pipeline.py` solo las orquesta.
| Función | Líneas | Propósito | | Módulo | Función | Propósito |
|---------|--------|-----------| |--------|---------|-----------|
| `_ruta_carpeta` (módulo) | 33-34 | Resuelve una ruta relativa al directorio del script | | `pipeline_utils` | `ruta_carpeta` | Resuelve una ruta relativa al directorio del proyecto |
| `_copiar_a_red` (módulo) | 37-94 | Copia PDF/XML a la ruta de red de Facturación Electrónica (`RUTA_RED_FACTURACION`) | | `pipeline_utils` | `log` | Imprime con indentación de árbol (niveles 1-4) |
| `_descargar_adjuntos` | 169-196 | Descarga adjuntos vía Graph API | | `adjuntos` | `descargar_adjuntos` | Descarga adjuntos vía Graph API |
| `_descomprimir` | 198-211 | Descomprime ZIPs | | `adjuntos` | `filtrar_y_limpiar` | Filtra PDF/XML, elimina el resto |
| `_filtrar_y_limpiar` | 213-253 | Filtra PDF/XML, elimina el resto | | `documentos` | `copiar_a_red` | Copia PDF/XML a la ruta de red de Facturación Electrónica (`RUTA_RED_FACTURACION`) |
| `_procesar_pdfs` | 255-308 | Extrae datos de PDFs, renombra, genera cache | | `documentos` | `procesar_pdfs` | Extrae datos de PDFs, renombra, genera cache |
| `_procesar_xmls` | 310-463 | Procesa XMLs, cruza con PDF, asigna guías, copia a red | | `documentos` | `procesar_xmls` | Procesa XMLs, cruza con PDF, asigna guías, copia a red |
| `_completar_guias` | 465-522 | Completa guías faltantes en líneas XML desde el PDF cache | | `documentos` | `completar_guias` | Completa guías faltantes en líneas XML desde el PDF cache |
| `_validar_guias` | 524-532 | Advierte (log) guías con datos incompletos | | `documentos` | `validar_guias` | Advierte (log) guías con datos incompletos |
| `_guardar_resultado` | 534-653 | Guarda JSON con datos extraídos (XML y PDF) | | `persistencia` | `guardar_resultado` | Guarda JSON con datos extraídos (XML y PDF) |
| `_calcular_fecha_vencimiento` | 665-692 | Calcula `DocDueDate` = fecha de contabilización + días de la condición de pago (`PaymentTermsTypes`) | | `persistencia` | `guardar_draft_json` | Guarda draft JSON para depuración |
| `_build_sap_draft` | 694-855 | Construye payload del draft SAP B1 | | `draft_sap` | `calcular_fecha_vencimiento` | Calcula `DocDueDate` = fecha de contabilización + días de la condición de pago (`PaymentTermsTypes`) |
| `_validate_prices` | 856-1030 | Valida precios contra DeliveryNotes / Orders (SAP) | | `draft_sap` | `build_sap_draft` | Construye payload del draft SAP B1 |
| `_build_validation_html` | 1032-1061 | Genera tabla HTML de validación de precios | | `draft_sap` | `validate_prices` | Valida precios contra DeliveryNotes / Orders (SAP) |
| `_obtener_valor_referencial_sap` | 1062-1090 | Consulta `Sp_ObtenerValorReferencial` en HANA por guía | | `draft_sap` | `obtener_valor_referencial_sap` | Consulta `Sp_ObtenerValorReferencial` en HANA por guía |
| `_build_valor_referencial_html` | 1091-1126 | Genera tabla HTML comparando valor referencial XML vs SAP | | `notificaciones` | `build_validation_html` | Genera tabla HTML de validación de precios |
| `_build_notification_body` | 1127-1168 | Arma cuerpo del email de notificación | | `notificaciones` | `build_valor_referencial_html` | Genera tabla HTML comparando valor referencial XML vs SAP |
| `_guardar_draft_json` | 1169-1195 | Guarda draft JSON para depuración | | `notificaciones` | `build_notification_body` | Arma cuerpo del email de notificación |
docs/pipeline.md
+61 -55
View File
@@ -4,12 +4,18 @@
``` ```
pipeline.py pipeline.py
├── CONFIG: RUTA_RED_FACTURACION, RUC_EMPRESA (desde .env)
├── HELPERS: _ruta_carpeta(), _copiar_a_red(data, _log)
├── PASO 1: step_authenticate() → token | None ├── PASO 1: step_authenticate() → token | None
├── PASO 2: step_fetch_emails() → list[dict] ├── PASO 2: step_fetch_emails() → list[dict]
├── _procesar_un_correo() → None (core, funciones anidadas) ├── _procesar_un_correo() → None (orquesta los módulos de abajo)
└── run_pipeline() → None └── run_pipeline() → None
Módulos que orquesta _procesar_un_correo():
├── adjuntos.py descargar_adjuntos, filtrar_y_limpiar
├── documentos.py procesar_pdfs, procesar_xmls, completar_guias, validar_guias, copiar_a_red
├── persistencia.py guardar_resultado, guardar_draft_json
├── draft_sap.py validate_prices, obtener_valor_referencial_sap, build_sap_draft
├── notificaciones.py build_validation_html, build_valor_referencial_html, build_notification_body
└── pipeline_utils.py log, ruta_carpeta
``` ```
--- ---
@@ -42,29 +48,29 @@ flowchart TD
A[_procesar_un_correo] --> B[1-2. BD: ¿ya procesado?] A[_procesar_un_correo] --> B[1-2. BD: ¿ya procesado?]
B -- Sí --> C[⏭️ Saltar] B -- Sí --> C[⏭️ Saltar]
B -- No --> D[Guardar en BD] B -- No --> D[Guardar en BD]
D --> E[3. _descargar_adjuntos] D --> E[3. descargar_adjuntos]
E --> F{Hay adjuntos?} E --> F{Hay adjuntos?}
F -- No --> C F -- No --> C
F -- Sí --> G[4. _descomprimir ZIPs] F -- Sí --> G[4. _descomprimir ZIPs]
G --> H[5. _filtrar_y_limpiar] G --> H[5. filtrar_y_limpiar]
H --> I{Hay PDF/XML?} H --> I{Hay PDF/XML?}
I -- No --> C I -- No --> C
I -- Sí --> J[6. _procesar_pdfs]:::highlight I -- Sí --> J[6. procesar_pdfs]:::highlight
J --> K[7. _procesar_xmls]:::highlight J --> K[7. procesar_xmls]:::highlight
K --> K2[8. _completar_guias] K --> K2[8. completar_guias]
K2 --> K3[9. _validar_guias] K2 --> K3[9. validar_guias]
K3 --> L{Hay documentos?} K3 --> L{Hay documentos?}
L -- No --> C L -- No --> C
L -- Sí --> M[10. _guardar_resultado] L -- Sí --> M[10. guardar_resultado]
M --> N[11. Filtrar facturas tipo 01] M --> N[11. Filtrar facturas tipo 01]
N --> O{Hay facturas?} N --> O{Hay facturas?}
O -- No --> C O -- No --> C
O -- Sí --> P[Bucle: por cada factura] O -- Sí --> P[Bucle: por cada factura]
P --> Q[12. _validate_prices]:::highlight P --> Q[12. validate_prices]:::highlight
Q --> Q2[12.5 _obtener_valor_referencial_sap] Q --> Q2[12.5 obtener_valor_referencial_sap]
Q2 --> R[13. _build_sap_draft]:::highlight Q2 --> R[13. build_sap_draft]:::highlight
R --> S[14. _guardar_draft_json] R --> S[14. guardar_draft_json]
S --> T{Todas coinciden?} S --> T{Todas coinciden?}
T -- Sí --> U[15. sapb1_manager.create_draft] T -- Sí --> U[15. sapb1_manager.create_draft]
T -- No --> V[⏭️ Draft NO creado] T -- No --> V[⏭️ Draft NO creado]
@@ -75,38 +81,38 @@ flowchart TD
classDef highlight fill:#fff3e0,stroke:#e65100,stroke-width:3px; classDef highlight fill:#fff3e0,stroke:#e65100,stroke-width:3px;
``` ```
`_procesar_xmls` (paso 7) incluye internamente la copia de PDF/XML a la ruta de red (`_copiar_a_red`, ver más abajo) por cada comprobante procesado. `procesar_xmls` (paso 7) incluye internamente la copia de PDF/XML a la ruta de red (`copiar_a_red`, ver más abajo) por cada comprobante procesado.
--- ---
## Funciones internas de `_procesar_un_correo` ## Funciones por módulo
> `_ruta_carpeta` y `_copiar_a_red` están definidas a nivel de módulo (no anidadas); el resto son funciones anidadas dentro de `_procesar_un_correo`. > Son funciones de módulo (antes anidadas en `_procesar_un_correo`) y reciben explícitamente `email`, `download_folder`, `output_folder`, etc. `_procesar_un_correo` solo las orquesta.
| Función | Línea | Propósito | | Función | Módulo | Propósito |
|---------|-------|-----------| |---------|--------|-----------|
| `_ruta_carpeta()` (módulo) | 33 | Resuelve una ruta relativa al directorio del script | | `ruta_carpeta()` | `pipeline_utils` | Resuelve una ruta relativa al directorio del proyecto |
| `_copiar_a_red(data, _log)` (módulo) | 37 | Copia PDF/XML de un comprobante a la ruta de red configurable | | `log(msg, level, last)` | `pipeline_utils` | Imprime con indentación de árbol |
| `_descargar_adjuntos()` | 169 | Descarga adjuntos vía Microsoft Graph | | `copiar_a_red(data)` | `documentos` | Copia PDF/XML de un comprobante a la ruta de red configurable |
| `_descomprimir()` | 198 | Descomprime archivos ZIP | | `descargar_adjuntos(email, email_addr, token, download_folder)` | `adjuntos` | Descarga adjuntos vía Microsoft Graph |
| `_filtrar_y_limpiar()` | 213 | Filtra solo PDF/XML, elimina el resto | | `filtrar_y_limpiar(email, download_folder)` | `adjuntos` | Filtra solo PDF/XML, elimina el resto |
| **`_procesar_pdfs()`** | **255** | **Extrae datos de PDFs, renombra, genera cache** | | **`procesar_pdfs(email, download_folder)`** | `documentos` | **Extrae datos de PDFs, renombra, genera cache** |
| **`_procesar_xmls(pdf_cache)`** | **310** | **Procesa XMLs, cruza con PDF, asigna guías, copia a red** | | **`procesar_xmls(email, download_folder, pdf_cache)`** | `documentos` | **Procesa XMLs, cruza con PDF, asigna guías, copia a red** |
| `_completar_guias(documentos, pdf_cache)` | 465 | Completa guías faltantes en líneas XML desde el PDF cache | | `completar_guias(documentos, pdf_cache)` | `documentos` | Completa guías faltantes en líneas XML desde el PDF cache |
| `_validar_guias(documentos)` | 524 | Advierte (log) guías con datos incompletos (falta tipo/serie/numero) | | `validar_guias(documentos)` | `documentos` | Advierte (log) guías con datos incompletos (falta tipo/serie/numero) |
| `_guardar_resultado(documentos, datos_pdf)` | 534 | Guarda JSON con datos extraídos (XML y PDF) | | `guardar_resultado(documentos, datos_pdf, output_folder)` | `persistencia` | Guarda JSON con datos extraídos (XML y PDF) |
| `_calcular_fecha_vencimiento(fecha_base, condicion_pago)` | 665 | Calcula `DocDueDate` = fecha de contabilización + días de `PaymentTermsTypes` | | `guardar_draft_json(draft, serie, numero, output_folder)` | `persistencia` | Guarda draft JSON para depuración |
| **`_build_sap_draft(factura, resultados_val)`** | **694** | **Construye payload del draft SAP B1 (incluye `U_VS_ANEXO_PDF`/`U_VS_ANEXO_XML`, `DocDueDate` por condición de pago)** | | `calcular_fecha_vencimiento(fecha_base, condicion_pago)` | `draft_sap` | Calcula `DocDueDate` = fecha de contabilización + días de `PaymentTermsTypes` |
| **`_validate_prices(factura)`** | **856** | **Valida precios contra DeliveryNotes/Orders SAP** | | **`build_sap_draft(factura, resultados_val)`** | `draft_sap` | **Construye payload del draft SAP B1 (incluye `U_VS_ANEXO_PDF`/`U_VS_ANEXO_XML`, `DocDueDate` por condición de pago)** |
| `_build_validation_html(resultados)` | 1032 | Genera tabla HTML de validación de precios | | **`validate_prices(factura)`** | `draft_sap` | **Valida precios contra DeliveryNotes/Orders SAP** |
| `_obtener_valor_referencial_sap(factura)` | 1062 | Consulta `Sp_ObtenerValorReferencial` en HANA por cada guía única | | `obtener_valor_referencial_sap(factura)` | `draft_sap` | Consulta `Sp_ObtenerValorReferencial` en HANA por cada guía única |
| `_build_valor_referencial_html(factura, guias_valor_ref)` | 1091 | Genera tabla HTML comparando valor referencial XML vs SAP | | `build_validation_html(resultados)` | `notificaciones` | Genera tabla HTML de validación de precios |
| `_build_notification_body(status, factura, validation_table, valor_referencial_table, **kwargs)` | 1127 | Arma cuerpo del email de notificación | | `build_valor_referencial_html(factura, guias_valor_ref)` | `notificaciones` | Genera tabla HTML comparando valor referencial XML vs SAP |
| `_guardar_draft_json(draft, serie, numero)` | 1169 | Guarda draft JSON para depuración | | `build_notification_body(status, factura, validation_table, valor_referencial_table, **kwargs)` | `notificaciones` | Arma cuerpo del email de notificación |
--- ---
## `_copiar_a_red(data, _log)` — Copia a ruta de red ## `copiar_a_red(data)` — Copia a ruta de red
Copia el PDF y el XML de cada comprobante procesado a una ruta de red UNC configurable, en paralelo al guardado local existente (no lo reemplaza). No falla el pipeline si la ruta no está configurada o no está disponible: solo registra una advertencia. Copia el PDF y el XML de cada comprobante procesado a una ruta de red UNC configurable, en paralelo al guardado local existente (no lo reemplaza). No falla el pipeline si la ruta no está configurada o no está disponible: solo registra una advertencia.
@@ -118,8 +124,8 @@ Copia el PDF y el XML de cada comprobante procesado a una ruta de red UNC config
{RUC_EMPRESA}-{tipo}-{serie}-{numero}.xml {RUC_EMPRESA}-{tipo}-{serie}-{numero}.xml
``` ```
`RUC_EMPRESA` es el RUC de Aberio (fijo); `ruc_emisor` es el RUC del proveedor que emitió la factura (varía por comprobante). Año/mes/día salen de `fecha_emision`, sin ceros a la izquierda en mes/día. `RUC_EMPRESA` es el RUC de Aberio (fijo); `ruc_emisor` es el RUC del proveedor que emitió la factura (varía por comprobante). Año/mes/día salen de `fecha_emision`, sin ceros a la izquierda en mes/día.
- Guarda las rutas resultantes en `data["ruta_red_pdf"]` / `data["ruta_red_xml"]`, que luego usa `_build_sap_draft` para llenar `U_VS_ANEXO_PDF` / `U_VS_ANEXO_XML` en el draft SAP B1. - Guarda las rutas resultantes en `data["ruta_red_pdf"]` / `data["ruta_red_xml"]`, que luego usa `build_sap_draft` para llenar `U_VS_ANEXO_PDF` / `U_VS_ANEXO_XML` en el draft SAP B1.
- Se invoca desde `_procesar_xmls`, justo después de resolver `data["ruta_pdf"]`. - Se invoca desde `procesar_xmls`, justo después de resolver `data["ruta_pdf"]`.
--- ---
@@ -134,23 +140,23 @@ run_pipeline
│ └── outlook_utilities.get_emails │ └── outlook_utilities.get_emails
└── _procesar_un_correo (por cada email) └── _procesar_un_correo (por cada email)
├── database_utilities.get_existing_ids / save_emails ├── database_utilities.get_existing_ids / save_emails
├── _descargar_adjuntos ├── descargar_adjuntos
│ └── outlook_utilities.download_attachments │ └── outlook_utilities.download_attachments
├── _descomprimir ├── _descomprimir
│ └── email_utilities.descomprimir_adjuntos │ └── email_utilities.descomprimir_adjuntos
├── _filtrar_y_limpiar ├── filtrar_y_limpiar
├── _procesar_pdfs ├── procesar_pdfs
│ └── pdf_utilities.extraer_datos_pdf / renombrar_pdf │ └── pdf_utilities.extraer_datos_pdf / renombrar_pdf
├── _procesar_xmls ├── procesar_xmls
│ ├── sunat_schema.extraer_datos_xml │ ├── sunat_schema.extraer_datos_xml
│ ├── pdf_utilities.extraer_referencias │ ├── pdf_utilities.extraer_referencias
│ └── _copiar_a_red │ └── copiar_a_red
├── _completar_guias ├── completar_guias
├── _validar_guias ├── validar_guias
├── _guardar_resultado ├── guardar_resultado
│ └── sunat_schema.split_serie_numero, pdf_utilities.TIPO_MAP │ └── sunat_schema.split_serie_numero, pdf_utilities.TIPO_MAP
├── (por cada factura tipo 01) ├── (por cada factura tipo 01)
│ ├── _validate_prices │ ├── validate_prices
│ │ ├── sapb1_manager.get_zonas │ │ ├── sapb1_manager.get_zonas
│ │ └── sapb1_manager.get_delivery_notes │ │ └── sapb1_manager.get_delivery_notes
│ │ ├── hana_manager.get_delivery (fallback si SL no responde; │ │ ├── hana_manager.get_delivery (fallback si SL no responde;
@@ -158,15 +164,15 @@ run_pipeline
│ │ └── si el DocEntry vino de OWTR (StockTransfers), resuelve │ │ └── si el DocEntry vino de OWTR (StockTransfers), resuelve
│ │ además la Orden de Venta asociada (Orders vía │ │ además la Orden de Venta asociada (Orders vía
│ │ U_IDC_NUMOV) para poder validar las líneas FL* │ │ U_IDC_NUMOV) para poder validar las líneas FL*
│ ├── _build_validation_html │ ├── build_validation_html
│ ├── _obtener_valor_referencial_sap │ ├── obtener_valor_referencial_sap
│ │ └── hana_manager.obtener_valor_referencial │ │ └── hana_manager.obtener_valor_referencial
│ ├── _build_valor_referencial_html │ ├── build_valor_referencial_html
│ ├── _build_sap_draft │ ├── build_sap_draft
│ │ └── sapb1_manager.obtener_condicion_pago (BusinessPartners + │ │ └── sapb1_manager.obtener_condicion_pago (BusinessPartners +
│ │ PaymentTermsTypes, para calcular DocDueDate) │ │ PaymentTermsTypes, para calcular DocDueDate)
│ ├── _guardar_draft_json │ ├── guardar_draft_json
│ ├── sapb1_manager.create_draft (si todas las líneas coinciden) │ ├── sapb1_manager.create_draft (si todas las líneas coinciden)
│ ├── _build_notification_body │ ├── build_notification_body
│ └── outlook_utilities.send_email │ └── outlook_utilities.send_email
``` ```
documentos.py
+371
View File
@@ -0,0 +1,371 @@
"""Extracción de datos de PDF/XML y asignación de guías de remisión."""
import os
import re
import shutil
import ntpath
from datetime import datetime
import sunat_schema
import pdf_utilities
from pipeline_utils import log, ruta_carpeta
# ============================================================================
def copiar_a_red(data: dict) -> None:
"""
Copia el PDF y el XML de un comprobante ya procesado a la ruta de red
de Facturación Electrónica, organizada como::
{RUTA_RED_FACTURACION}\\{año}\\{mes}\\{día}\\{tipo}\\{RUC_EMPRESA}\\{ruc_emisor}\\
{RUC_EMPRESA}-{tipo}-{serie}-{numero}_FormatoImpreso.pdf
{RUC_EMPRESA}-{tipo}-{serie}-{numero}.xml
``RUC_EMPRESA`` es el RUC de Aberio (fijo, por ahora); ``ruc_emisor``
es el RUC del proveedor que emitió la factura (varía por comprobante).
Año/mes/día se toman de ``fecha_emision`` (mes y día sin ceros a la
izquierda). No falla el pipeline si la ruta de red no está configurada
o no está disponible: solo registra una advertencia.
Guarda las rutas resultantes en ``data["ruta_red_pdf"]`` y
``data["ruta_red_xml"]`` para que ``build_sap_draft`` las incluya en
el draft (``U_VS_ANEXO_PDF`` / ``U_VS_ANEXO_XML``).
"""
RUTA_RED_FACTURACION = os.getenv("RUTA_RED_FACTURACION", "")
RUC_EMPRESA = os.getenv("RUC_EMPRESA", "")
if not RUTA_RED_FACTURACION:
return
ruc_emisor = data.get("ruc_emisor", "")
tipo = data.get("tipo", "")
serie = data.get("serie", "")
numero = data.get("numero", "")
xml_path = data.get("ruta_xml")
pdf_path = data.get("ruta_pdf")
try:
fecha = datetime.strptime(data.get("fecha_emision", ""), "%Y-%m-%d")
except (ValueError, TypeError):
log(f"⚠️ No se pudo copiar a red: fecha_emision inválida ({data.get('fecha_emision')})", 3)
return
carpeta_destino = os.path.join(
RUTA_RED_FACTURACION,
str(fecha.year), str(fecha.month), str(fecha.day),
tipo, RUC_EMPRESA, ruc_emisor,
)
# Ruta guardada en el draft SAP (U_VS_ANEXO_PDF/XML): siempre estilo
# Windows/UNC con backslash, sin importar el separador del SO donde
# corre el script (evita mezclar "\\" y "/" al usar os.path.join en
# un host no-Windows).
carpeta_destino_unc = ntpath.join(
RUTA_RED_FACTURACION,
str(fecha.year), str(fecha.month), str(fecha.day),
tipo, RUC_EMPRESA, ruc_emisor,
)
nombre_base = f"{RUC_EMPRESA}-{tipo}-{serie}-{numero}"
try:
os.makedirs(carpeta_destino, exist_ok=True)
if pdf_path and os.path.exists(pdf_path):
ruta_red_pdf = os.path.join(carpeta_destino, f"{nombre_base}_FormatoImpreso.pdf")
shutil.copy2(pdf_path, ruta_red_pdf)
data["ruta_red_pdf"] = ntpath.join(carpeta_destino_unc, f"{nombre_base}_FormatoImpreso.pdf")
if xml_path and os.path.exists(xml_path):
ruta_red_xml = os.path.join(carpeta_destino, f"{nombre_base}.xml")
shutil.copy2(xml_path, ruta_red_xml)
data["ruta_red_xml"] = ntpath.join(carpeta_destino_unc, f"{nombre_base}.xml")
log(f"📁 Copiado a red: {carpeta_destino}", 3)
except OSError as e:
log(f"⚠️ Error copiando a ruta de red: {e}", 3)
# ============================================================================
def procesar_pdfs(email: dict, download_folder: str) -> dict:
"""
Procesa todos los PDF de los adjuntos del correo.
Por cada PDF:
1. Extrae sus datos estructurados via ``pdf_utilities.extraer_datos_pdf``.
2. Renombra el archivo al formato canónico
``{ruc}-{tipo}-{serie}-{numero}.pdf``.
3. Almacena los datos extraídos en un cache indexado por el nombre
canónico.
Los datos extraídos también se acumulan en ``email["datos_pdf"]``.
Returns:
dict — Cache ``{nombre_canonico.pdf: datos_extraídos}``.
Example:
Si el PDF es ``F002-486.pdf`` y la extracción devuelve
``ruc_emisor=20537929520, tipo_documento=FACTURA``::
cache = procesar_pdfs(email, download_folder)
# → {"20537929520-01-F002-486.pdf": {...datos...}}
# El archivo se renombra a 20537929520-01-F002-486.pdf
"""
carpeta = ruta_carpeta(download_folder)
pdf_cache: dict[str, dict] = {}
datos_pdf = []
for idx, adjunto in enumerate(email.get("adjuntos", [])):
if not adjunto.lower().endswith(".pdf"):
continue
pdf_path = os.path.join(carpeta, adjunto)
data = pdf_utilities.extraer_datos_pdf(pdf_path)
if data and not data.get("error"):
nuevo_path = pdf_utilities.renombrar_pdf(pdf_path, data)
if nuevo_path != pdf_path:
nuevo_nombre = os.path.basename(nuevo_path)
email["adjuntos"][idx] = nuevo_nombre
data["archivo"] = nuevo_nombre
datos_pdf.append(data)
canonical = os.path.basename(nuevo_path)
pdf_cache[canonical] = data
log(f"📄 PDF procesado: {canonical} ({data.get('tipo_documento', '?')})", 3)
else:
log(f"⚠️ PDF no procesado: {adjunto} - {data.get('error', 'error desconocido')}", 2)
if datos_pdf:
email["datos_pdf"] = datos_pdf
return pdf_cache
# ============================================================================
def procesar_xmls(email: dict, download_folder: str, pdf_cache: dict) -> list[dict]:
"""
Procesa todos los XML de los adjuntos del correo.
Por cada XML:
1. Extrae datos estructurados via ``sunat_schema.extraer_datos_xml``.
2. Construye el nombre canónico del PDF asociado
(``{ruc}-{tipo}-{serie}-{numero}.pdf``) y lo busca en el cache.
3. Si el XML no trae ``documentos_referencia``, los completa desde
el PDF cache (usando ``pdf_utilities.extraer_referencias``).
4. Asigna las guías de remisión (``guias_remision[]``) y
transportista (``guia_transportista``) a cada línea de detalle desde
``notes_text`` o ``documentos_referencia``.
Returns:
list[dict] — Lista de documentos extraídos listos para
``guardar_resultado``.
Example:
XML con ``DespatchDocumentReference`` ID sucio
``"GRRE: T001-127773 / GRET :EG03-11"``::
docs = procesar_xmls(email, download_folder, pdf_cache)
# → [{
# "lineas": [{
# "guias_remision": [{"tipo":"09","serie":"T001","numero":"127773"}],
# "guia_transportista": {"tipo":"31","serie":"EG03","numero":"11"}
# }]
# }]
"""
carpeta = ruta_carpeta(download_folder)
documentos = []
last_idx = len(email["adjuntos"]) - 1
for idx, a in enumerate(email["adjuntos"]):
log(a, 2, last=(idx == last_idx))
if not a.lower().endswith(".xml"):
continue
xml_path = os.path.join(carpeta, a)
log(f"📄 Procesando XML: {a}", 3)
data = sunat_schema.extraer_datos_xml(xml_path)
if not data:
log("ℹ️ XML sin datos (no es factura tipo 01 o guía 31)", 3)
continue
# CDR solo se renombra, no se procesa como documento
if data.get("es_cdr"):
prefijo = "CDR-"
cdr_name = f"{prefijo}{data['ruc_emisor']}-{data['tipo']}-{data['serie']}-{data['numero']}.xml"
cdr_dest = os.path.join(carpeta, cdr_name)
if cdr_dest != xml_path:
if os.path.exists(cdr_dest):
os.remove(cdr_dest)
os.rename(xml_path, cdr_dest)
email["adjuntos"][idx] = cdr_name
log(f"ℹ️ {data.get('serie','?')}-{data.get('numero','?')} (CDR) — solo renombrado", 3)
continue
data["ruta_xml"] = xml_path
# Renombrar XML a {prefijo}{ruc}-{tipo}-{serie}-{numero}.xml
prefijo = ""
xml_canonical = f"{prefijo}{data.get('ruc_emisor', '')}-{data.get('tipo', '')}-{data.get('serie', '')}-{data.get('numero', '')}.xml"
xml_destino = os.path.join(carpeta, xml_canonical)
if xml_destino != xml_path:
if os.path.exists(xml_destino):
os.remove(xml_destino)
os.rename(xml_path, xml_destino)
email["adjuntos"][idx] = xml_canonical
data["ruta_xml"] = xml_destino
xml_path = xml_destino
# ruta_pdf: construir nombre canónico {ruc}-{tipo}-{serie}-{numero}.pdf
pdf_canonical = f"{data.get('ruc_emisor', '')}-{data.get('tipo', '')}-{data.get('serie', '')}-{data.get('numero', '')}.pdf"
pdf_path = os.path.join(carpeta, pdf_canonical)
if not os.path.exists(pdf_path):
# Fallback: reemplazar extensión .xml por .pdf
pdf_path = os.path.splitext(xml_path)[0] + ".pdf"
pdf_path = pdf_path if os.path.exists(pdf_path) else None
data["ruta_pdf"] = pdf_path
copiar_a_red(data)
# Cruzar documentos_referencia desde el PDF renombrado (pdf_cache por nombre canónico)
if not data.get("documentos_referencia"):
pdf_data = pdf_cache.get(pdf_canonical)
if pdf_data:
refs = pdf_utilities.extraer_referencias(pdf_data.get("items", []))
if refs:
data["documentos_referencia"] = refs
log(f"🔄 documentos_referencia completados desde PDF ({len(refs)} refs)", 3)
if not data.get("documentos_referencia"):
for pdf_data in email.get("datos_pdf", []):
refs = pdf_utilities.extraer_referencias(pdf_data.get("items", []))
if refs:
data["documentos_referencia"] = refs
log(f"🔄 documentos_referencia completados desde PDF (fallback, {len(refs)} refs)", 3)
break
# Asignar guias_remision[] y guia_transportista por línea
# Prioridad: notes_text (ya parseado) > documentos_referencia
refs_09 = [r for r in data.get("documentos_referencia", []) if r.get("tipo") == "09"]
refs_31 = [r for r in data.get("documentos_referencia", []) if r.get("tipo") == "31"]
for i, linea in enumerate(data.get("lineas", [])):
gr = linea.get("guias_remision") or []
gt = linea.get("guia_transportista")
if not gr and i < len(refs_09):
es_ultima = i == len(data["lineas"]) - 1
if es_ultima:
gr = list(refs_09[i:])
else:
gr = [dict(refs_09[i])]
if not gt and i < len(refs_31):
gt = dict(refs_31[i])
linea["guias_remision"] = gr
if gt:
linea["guia_transportista"] = gt
log(f"✅ Datos extraídos: {data.get('serie')}-{data.get('numero')} ({data.get('tipo_descripcion', '?')})", 3)
documentos.append(data)
# Post-procesamiento: resolver GREs desde guías de transporte referenciadas
guia_lookup = {}
for doc in documentos:
if doc.get("tipo") in ("31", "09"):
guia_lookup[(doc["tipo"], doc["serie"], doc["numero"])] = doc
for doc in documentos:
if doc.get("tipo") != "01":
continue
for ref in doc.get("documentos_referencia", []):
if ref["tipo"] == "31":
guia_doc = guia_lookup.get(("31", ref["serie"], ref["numero"]))
if guia_doc:
gre_list = [g for g in guia_doc.get("documentos_referencia", []) if g.get("tipo") == "09"]
if gre_list:
log(f"🔗 Resueltas {len(gre_list)} GRE(s) desde guía {ref['serie']}-{ref['numero']}", 3)
for i, linea in enumerate(doc.get("lineas", [])):
if not linea.get("guias_remision"):
es_ultima = i == len(doc["lineas"]) - 1
if es_ultima:
linea["guias_remision"] = gre_list[i:]
elif i < len(gre_list):
linea["guias_remision"] = [gre_list[i]]
if not linea.get("guia_transportista"):
linea["guia_transportista"] = dict(ref)
return documentos
# ============================================================================
def completar_guias(documentos: list[dict], pdf_cache: dict) -> int:
"""
Completa guías faltantes en XML desde datos del PDF.
Para cada documento en ``documentos``, verifica si todas las líneas
tienen ``guias_remision``. Si faltan, busca el PDF correspondiente en
el cache y copia las guías de las líneas del PDF.
Args:
documentos: Lista de documentos extraídos de XML.
pdf_cache: Cache de datos PDF (canonical_name → datos).
Returns:
int — Número de guías copiadas.
"""
total = 0
for xd in documentos:
has_all = all(
len(linea.get("guias_remision") or []) > 0
for linea in xd.get("lineas", [])
)
if has_all:
continue
pdf_canonical = f"{xd.get('ruc_emisor', '')}-{xd.get('tipo', '')}-{xd.get('serie', '')}-{xd.get('numero', '')}.pdf"
pd = pdf_cache.get(pdf_canonical)
if not pd:
continue
for i, xl in enumerate(xd.get("lineas", [])):
if xl.get("guias_remision"):
continue
pl = pd["items"][i] if i < len(pd.get("items", [])) else {}
p_gr = pl.get("guias_remision") or []
if isinstance(p_gr[0], str) if p_gr else False:
# Normalizar formato string → dict
p_gr = [
{"tipo": "09", "serie": m.group(1).upper(), "numero": m.group(2).lstrip("0") or "0"}
for g in p_gr
for m in [re.search(r"\b([TVE][A-Z0-9]{3,4})\s*-?\s*(\d+)", g)]
if m
]
if p_gr:
xl["guias_remision"] = list(p_gr)
total += len(p_gr)
if not xl.get("guia_transportista") and pl.get("guia_transportista"):
gt = pl["guia_transportista"]
if isinstance(gt, str):
m = re.search(r"\b([TVE][A-Z0-9]{3,4})\s*-?\s*(\d+)", gt)
if m:
xl["guia_transportista"] = {
"tipo": "31",
"serie": m.group(1).upper(),
"numero": m.group(2).lstrip("0") or "0",
}
elif isinstance(gt, dict):
xl["guia_transportista"] = dict(gt)
return total
# ============================================================================
def validar_guias(documentos: list[dict]) -> None:
for doc in documentos:
for i, linea in enumerate(doc.get("lineas", [])):
for j, g in enumerate(linea.get("guias_remision") or []):
if not isinstance(g, dict) or not all(k in g and g[k] for k in ("tipo", "serie", "numero")):
log(f"⚠️ guias_remision[{j}] inválida en {doc.get('serie','?')}-{doc.get('numero','?')} línea {i}: {g}", 2)
gt = linea.get("guia_transportista")
if gt and (not isinstance(gt, dict) or not all(k in gt and gt[k] for k in ("tipo", "serie", "numero"))):
log(f"⚠️ guia_transportista inválida en {doc.get('serie','?')}-{doc.get('numero','?')} línea {i}: {gt}", 2)
draft_sap.py
+413
View File
@@ -0,0 +1,413 @@
"""Validación de precios contra SAP y construcción del draft SAP B1."""
import re
import math
from datetime import datetime, timedelta
import sapb1_manager
import hana_manager
from pipeline_utils import log
# ============================================================================
def calcular_fecha_vencimiento(fecha_base: str, condicion_pago: dict | None) -> str:
"""
Calcula la fecha de vencimiento sumando a ``fecha_base`` los días de
la condición de pago (PaymentTermsTypes) del proveedor.
Usa ``NumberOfAdditionalDays`` como días a sumar; si ese valor es 0,
usa ``NumberOfAdditionalMonths * 30`` en su lugar.
Args:
fecha_base: Fecha de contabilización, formato "YYYY-MM-DD".
condicion_pago: Resultado de ``sapb1_manager.obtener_condicion_pago``
(campos ``NumberOfAdditionalDays``, ``NumberOfAdditionalMonths``),
o None si el proveedor no tiene condición de pago configurada.
Returns:
str — Fecha de vencimiento, formato "YYYY-MM-DD". Si
``condicion_pago`` es None, retorna ``fecha_base`` sin cambios.
"""
fecha = datetime.strptime(fecha_base, "%Y-%m-%d")
if not condicion_pago:
return fecha.strftime("%Y-%m-%d")
dias = int(condicion_pago.get("NumberOfAdditionalDays") or 0)
if dias == 0:
dias = int(condicion_pago.get("NumberOfAdditionalMonths") or 0) * 30
fecha += timedelta(days=dias)
return fecha.strftime("%Y-%m-%d")
# ============================================================================
def build_sap_draft(factura: dict, resultados_val: list[dict]) -> dict:
"""
Construye el payload JSON para crear un Draft en SAP B1 Service Layer
(endpoint ``POST /Drafts``).
Flujo:
1. Toma los datos de cabecera de la factura (RUCS, fechas, totales).
2. Determina el ``ItemCode`` según la fecha de emisión:
``C103002-A`` si es de un mes anterior, ``C103002`` si es del
mes actual.
3. Prepara las cuotas de detracción (``DocumentInstallments``):
si la factura tiene detracción, genera dos cuotas (detracción y
neto); si no, una sola cuota por el total.
4. Por cada línea de factura, itera sobre las guías validadas
(``guias_lt[]`` de ``resultados_val``) y genera una línea de
documento (``DocumentLine``) por cada guía con:
- ``ItemCode``, ``Quantity``, ``PriceAfterVAT``,
- ``WarehouseCode``, ``U_IDC_GRFLETE`` y campos fijos.
Cuando la validación es ``ok=True`` usa los valores de SAP
(``draft_quantity``, ``draft_price``); cuando no, usa los del XML.
5. Calcula ``U_IDC_VALREF`` como la suma de ``valor_referencial``
(extraído del XML SUNAT por ``sunat_schema``) de todas las
líneas de la factura.
Args:
factura: Diccionario con los datos de la factura extraídos del
XML (claves: ``serie``, ``numero``, ``tipo``, ``lineas[]``,
``guias_remision[]``, ``tiene_detraccion``, etc.).
resultados_val: Lista de resultados de ``validate_prices``,
uno por línea de factura. Cada elemento contiene
``guias_lt[]``, ``ok``, ``use_sap_prices``, etc.
Returns:
Dict con la estructura completa del draft SAP B1 lista para
enviar a ``POST /Drafts``.
Example:
Factura F002-486 con guía T001-127773 y SAP devuelve
``FL10007 qty=350 pav=1.3396``::
draft = build_sap_draft(factura, resultados_val)
# → {
# "DocType": "dDocument_Items",
# "CardCode": "PL20537929520",
# "DocumentLines": [{
# "ItemCode": "C103002",
# "Quantity": 350,
# "PriceAfterVAT": 1.3396,
# "U_IDC_GRFLETE": "09T001-127773",
# ...
# }],
# "DocumentInstallments": [...],
# "U_IDC_VALREF": 31181.5,
# ...
# }
"""
serie = factura.get("serie", "")
numero = factura.get("numero", "")
tipo = factura.get("tipo", "01")
total = factura.get("total_factura", 0) or 0
tiene_detraccion = factura.get("tiene_detraccion", False)
codigo_detraccion = factura.get("codigo_detraccion", "")
porcentaje_detraccion = factura.get("porcentaje_detraccion", 0) or 0
monto_detraccion = factura.get("monto_detraccion", 0) or 0
fecha_emision = factura.get("fecha_emision", "")
ruc_emisor = factura.get("ruc_emisor", "")
razon_social_emisor = factura.get("razon_social_emisor", "")
card_code = f"PL{ruc_emisor}"
fecha_contabilizacion = datetime.now().strftime("%Y-%m-%d")
emision = datetime.strptime(fecha_emision, "%Y-%m-%d")
item_code = "C103002-A" if (emision.year, emision.month) < (datetime.now().year, datetime.now().month) else "C103002"
condicion_pago = sapb1_manager.obtener_condicion_pago(card_code)
fecha_vencimiento = calcular_fecha_vencimiento(fecha_contabilizacion, condicion_pago)
if tiene_detraccion:
# neto = round(total - monto_detraccion, 2)
monto_detraccion = math.ceil(monto_detraccion)
neto = total - monto_detraccion
installments = [
{"DueDate": fecha_emision, "Total": monto_detraccion, "InstallmentId": 1, "U_VS_TIPOCUOTA": "D", "U_VS_TIPOSERVICIO": f"{codigo_detraccion}02"},
{"DueDate": fecha_emision, "Total": round(neto,2), "InstallmentId": 2, "U_VS_TIPOCUOTA": "N", "U_VS_TIPOSERVICIO": ""},
]
else:
installments = [
{"DueDate": fecha_emision, "Total": round(total, 2), "InstallmentId": 1, "U_VS_TIPOCUOTA": "N", "U_VS_TIPOSERVICIO": ""},
]
doc_lines = []
guia_referencia = ""
for i, linea in enumerate(factura.get("lineas", [])):
resultado_validacion = resultados_val[i] if i < len(resultados_val) else None
guias_lt = (resultado_validacion or {}).get("guias_lt", [])
if not guias_lt:
continue
guia_referencia = f"{guias_lt[0]['guia']['tipo']}{guias_lt[0]['guia']['serie']}-{guias_lt[0]['guia']['numero']}"
for item in guias_lt:
codigo_guia = f"{item['guia']['tipo']}{item['guia']['serie']}-{item['guia']['numero']}"
lt_val = math.trunc(item["draft_price"] * 1_000_000) / 1_000_000
qty = item.get("draft_quantity", int(item.get("quantity", 1)))
doc_lines.append({
"ItemCode": item_code,
#"ItemDescription": linea.get("descripcion", "TRANSPORTE DE CARGA MERCADERIA")[:200],
"Quantity": qty,
"PriceAfterVAT": lt_val,
"WarehouseCode": item.get("warehouse_code", "A-CENTRA"),
"CostingCode": "PRO",
"CostingCode2": "PRO03",
"TaxCode": "I18",
"FreeText": linea.get("descripcion", "")[:100],
"MeasureUnit": "UND",
"U_BPP_CTAD": "9231101",
"U_BPP_OPER": "A",
"U_tipoOpT12": "02",
"U_VS_ONEROSO": "1",
"U_VS_TIPAFE": "10",
"U_IDC_GRFLETE": codigo_guia,
})
valor_referencial = sum(l.get("valor_referencial") or 0 for l in factura.get("lineas", []))
return {
"DocType": "dDocument_Items",
"DocDate": fecha_contabilizacion,
"DocDueDate": fecha_vencimiento,
"TaxDate": fecha_emision,
"CardCode": card_code,
"NumAtCard": f"{tipo}{serie}-{numero}",
"U_IDC_NUMGR": "" if not guia_referencia else guia_referencia,
"DocCurrency": "SOL",
"Comments": f"Factura {serie}-{numero} / {razon_social_emisor}"[:254],
"JournalMemo": razon_social_emisor[:254],
"DocObjectCode": "oPurchaseInvoices",
"U_BPV_SERI": f"{serie}-01",
"U_VS_NATNUM": "T",
"U_BPV_NCON2": numero,
"U_BPP_MDTD": tipo,
"U_BPP_MDBI": "A",
"U_BPP_MDCD": numero,
"U_BPP_MDSD": serie,
"U_VS_CODSER": f"{codigo_detraccion}02" if codigo_detraccion else "",
"U_VS_AFEDET": "Y" if tiene_detraccion else "N",
"U_VS_PORDET": porcentaje_detraccion,
"U_VS_APLANT": "Y",
"U_BPP_MDSD2": serie,
"U_IDC_FLETE": "02",
"U_IDC_LINEANE": "01",
"U_IDC_VALREF": valor_referencial,
"U_VS_ANEXO_PDF": factura.get("ruta_red_pdf", ""),
"U_VS_ANEXO_XML": factura.get("ruta_red_xml", ""),
"DocumentLines": doc_lines,
"DocumentInstallments": installments,
}
# ============================================================================
def validate_prices(factura: dict) -> tuple[bool, list[dict]]:
"""
Valida los precios de cada guía contra DeliveryNotes en SAP y
prepara los datos necesarios para construir el draft.
Flujo:
1. Obtiene el maestro de zonas SAP (``U_IDC_ZONAS``) para resolver
el nombre de zona a partir de ``U_IDC_DESTINO``.
2. Por cada línea de factura:
a. Itera sobre sus ``guias_remision[]``.
b. Para cada guía consulta SAP via
``sapb1_manager.get_delivery_notes``.
c. Dentro de cada DeliveryNote, suma los ``LineTotal``,
``PriceAfterVAT`` y obtiene la máxima ``Quantity`` de las
líneas cuyo ``ItemCode`` coincide con ``FL\\d``.
d. Acumula los resultados en ``guias_lt[]`` (uno por guía) con
``line_total``, ``quantity``, ``price_after_vat_sum``,
``warehouse_code`` y ``zona``.
3. Determina si los precios coinciden (``ok``):
``abs(total_linea - sap_total) < 0.01``.
4. Decide si se deben usar precios de SAP en el draft
(``use_sap_prices``):
``ok and int(cantidad) == 1 and int(total_sap_qty) != 1``.
Args:
factura: Diccionario con los datos de la factura (debe contener
``lineas[]``, y cada línea debe tener ``guias_remision[]``
y ``total_linea``).
Returns:
tuple[bool, list[dict]]:
- ``todas_coinciden``: True si TODAS las líneas tienen
``ok=True``.
- ``resultados``: Lista de resultados por línea, cada uno
con claves: ``grflete``, ``ok``, ``draft_price``,
``sap_price``, ``warehouse_code``, ``draft_cantidad``,
``use_sap_prices``, ``guias_lt[]``.
Example:
Factura con guía ``T001-127773`` y SAP devuelve
``FL10007 qty=350 pav=1.3396, FL20007 qty=350 pav=0``::
ok, resultados = validate_prices(factura)
# → ok = False (397.34 != 633.47)
# → resultados[0]["guias_lt"][0] = {
# "guia": {"tipo":"09","serie":"T001","numero":"127773"},
# "line_total": 397.34,
# "quantity": 350.0,
# "price_after_vat_sum": 1.3396,
# "zona": "CAÑETE",
# "warehouse_code": "A-CENTRA"
# }
"""
resultados = []
factura_lines = factura.get("lineas", [])
# listado de zonas para obtener su descripción
zonas_raw = sapb1_manager.get_zonas()
zonas = {z["Code"]: z["Name"] for z in zonas_raw}
for fl in factura_lines:
# por cada linea de factura se obtiene
# cantidad del xml de la factura
cantidad = float(fl.get("cantidad", 0))
# total de linea antes de impuesto del xml de la factura
total_linea = float(fl.get("total_linea", 0) or 0)
# lista de guias asociadas a la linea de factura de xml o pdf
guias = fl.get("guias_remision", [])
# dificil que la cantidad sea 0, pero esos casos los descartamos
if cantidad == 0:
resultados.append({"grflete": "-", "ok": False, "draft_price": None, "sap_price": None, "label": "Cantidad cero, sin validar", "warehouse_code": "-", "draft_cantidad": cantidad, "guias_lt": []})
continue
# si no hay guia asociada, no se puede validar, lo descartamos
if not guias:
resultados.append({"grflete": "-", "ok": False, "draft_price": None, "sap_price": None, "label": "Sin guía de referencia", "warehouse_code": "-", "draft_cantidad": cantidad, "guias_lt": []})
continue
guias_lt = []
all_wh = set()
all_ok = True
# una linea de la factura puede tener varias guias de remision
# hay que validar cada una de ellas
for guia in guias:
codigo_guia = f"{guia['tipo']}{guia['serie']}-{guia['numero']}"
try:
delivery_notes = sapb1_manager.get_delivery_notes(
tipo=guia["tipo"], serie=guia["serie"], numero=guia["numero"]
)
wh_set = set()
suma_line_total = 0.0
max_qty = 0.0
sum_pav = 0.0
weight1 = 0.0
zona = ""
for dn in delivery_notes:
# obtenemos la zona de destino del documento
zona = zonas.get(dn.get("U_IDC_DESTINO",""),"")
# obtenemos el almacen de la primera linea
first_line = dn.get("DocumentLines", [None])[0]
if first_line:
wh = first_line.get("WarehouseCode", "")
if wh:
wh_set.add(wh)
# obtenemos el peso de la primera linea de la guia de remision
# que representa el peso de la guia de remisión del articulo inventariable
# y lo vamos acumulando para obtener el peso total de la linea de la factura
weight1 += float(first_line.get("Weight1", 0) or 0)
# recorremos cada linea del documento
for linea_detalle_dict in dn.get("DocumentLines", []):
# solo procesamos lineas que empiecen con FL\d
if not re.match(r"FL\d", linea_detalle_dict.get("ItemCode", "")):
continue
# obtenemos el total de linea
line_total = float(linea_detalle_dict.get("LineTotal", 0) or 0)
# obtenemos el precio después de impuesto
price_after_vat = float(linea_detalle_dict.get("PriceAfterVAT", 0) or 0)
# obtenemos la cantidad
quantity = float(linea_detalle_dict.get("Quantity", 0) or 0)
# sumamos el total de linea
suma_line_total += line_total
# obtenemos la cantidad maxima
max_qty = max(max_qty, quantity)
sum_pav += price_after_vat
log(f"└─ {dn.get('Series', '?')}-{dn.get('DocNum', '?')} | ItemCode: {linea_detalle_dict.get('ItemCode', '')} | PriceAfterVAT: {price_after_vat:.6f} | Quantity: {quantity:.6f} | LineTotal: {line_total:.6f}", 4)
guias_lt.append({"guia": guia, "line_total": suma_line_total, "warehouse_code": ", ".join(sorted(wh_set)) if wh_set else "A-CENTRA", "quantity": max_qty, "price_after_vat_sum": sum_pav, "zona":zona, "weight1": weight1, "draft_quantity": 0, "draft_price": 0.0})
all_wh.update(wh_set)
except Exception as ex:
log(f"⚠️ Guía {codigo_guia}: error al consultar SAP - {ex}", 2)
guias_lt.append({"guia": guia, "line_total": 0.0, "warehouse_code": "A-CENTRA", "weight1": 0.0, "quantity": 0.0, "price_after_vat_sum": 0.0, "zona": "", "draft_quantity": 0, "draft_price": 0.0})
all_ok = False
sap_total = sum(item["line_total"] for item in guias_lt)
sin_sap = all(item["line_total"] == 0.0 and item["quantity"] == 0.0 for item in guias_lt)
ok = all_ok and abs(total_linea - sap_total) < 0.01
label = "Sin guías SAP" if sin_sap else "LineTotal (FL*) vs total_linea"
grflete_str = "; ".join(f"{item['guia']['tipo']}{item['guia']['serie']}-{item['guia']['numero']}" for item in guias_lt)
wh_str = ", ".join(sorted(all_wh)) if all_wh else "A-CENTRA"
for item in guias_lt:
if ok:
item["draft_quantity"] = int(item["quantity"])
item["draft_price"] = item["price_after_vat_sum"]
else:
item["draft_quantity"] = int(cantidad)
item["draft_price"] = float(total_linea) / max(float(cantidad), 1)
resultados.append({
"grflete": grflete_str,
"ok": ok,
"draft_price": total_linea,
"sap_price": sap_total,
"label": label,
"warehouse_code": wh_str,
"guias_lt": guias_lt,
})
if sin_sap:
log("⏭️ Sin guías en SAP — línea saltada", 3)
else:
log(f"Cantidad: {cantidad} | Total LineTotal (SAP): {sap_total:.6f} vs total_linea (XML): {total_linea:.6f}", 3)
if ok:
log(f"✅ Coincide: {total_linea:.6f} ≈ {sap_total:.6f}", 3)
else:
log(f"❌ NO coincide — XML: {total_linea:.6f} vs SAP: {sap_total:.6f}", 3)
todas_coinciden = all(r["ok"] for r in resultados)
return todas_coinciden, resultados
# ============================================================================
def obtener_valor_referencial_sap(factura: dict) -> list[dict]:
"""
Ejecuta ``Sp_ObtenerValorReferencial`` en SAP HANA por cada guía de
remisión (tipo 09) única de la factura y acumula el resultado.
Una factura puede referenciar la misma guía desde más de una línea;
en ese caso solo se consulta una vez por guía.
Args:
factura: Diccionario de la factura, con ``lineas[]`` y cada
línea con ``guias_remision[]`` (tipo 09).
Returns:
list[dict] — uno por guía única, con ``serie``, ``numero`` y
``valor_sap``.
"""
vistos = set()
resultados = []
for linea in factura.get("lineas", []):
for guia in linea.get("guias_remision", []):
clave = (guia.get("serie"), guia.get("numero"))
if clave in vistos:
continue
vistos.add(clave)
valor_sap = hana_manager.obtener_valor_referencial(guia.get("serie", ""), guia.get("numero", ""))
resultados.append({"serie": guia.get("serie", ""), "numero": guia.get("numero", ""), "valor_sap": valor_sap})
log(f"└─ Guía 09{guia.get('serie')}-{guia.get('numero')} | Valor Referencial SAP: {valor_sap:.2f}", 3)
return resultados
notificaciones.py
+117
View File
@@ -0,0 +1,117 @@
"""Cuerpos HTML de los correos de notificación."""
# ============================================================================
def build_validation_html(resultados: list[dict]) -> str:
"""
Genera una tabla HTML con el resultado de la validación de precios
para incluir en el cuerpo del email de notificación.
Args:
resultados: Lista de resultados de ``validate_prices``.
Returns:
str — Código HTML de la tabla con columnas:
Guía, Estado, Draft Price, SAP, Almacén, Resultado.
Example:
Con un resultado ``ok=True`` y ``grflete="09T001-127773"``::
html = build_validation_html(resultados)
# → "<table ...><tr><td>09T001-127773</td><td>✅</td>...</tr></table>"
"""
rows = ""
for r in resultados:
icono = "✅" if r["ok"] else "❌"
dp = f"{r['draft_price']:.6f}" if r["draft_price"] is not None else "-"
sap_price_str = f"{r['sap_price']:.6f}" if r["sap_price"] is not None else "-"
rows += f"<tr><td>{r['grflete']}</td><td>{icono}</td><td>{dp}</td><td>{sap_price_str}</td><td>{r['warehouse_code']}</td><td>{r['label']}</td></tr>"
return (
"<table border='1' cellpadding='6' cellspacing='0' style='border-collapse:collapse;'>"
"<tr><th>Guía</th><th>Estado</th><th>Draft Price</th><th>SAP</th><th>Almacén</th><th>Resultado</th></tr>"
f"{rows}</table>"
)
# ============================================================================
def build_valor_referencial_html(factura: dict, guias_valor_ref: list[dict]) -> str:
"""
Genera una tabla HTML comparando el valor referencial del XML de la
factura contra el valor referencial obtenido en SAP
(``Sp_ObtenerValorReferencial``) por cada guía de remisión, y el
total acumulado.
Args:
factura: Diccionario de la factura (usa ``lineas[]`` para sumar
``valor_referencial`` extraído del XML).
guias_valor_ref: Resultado de ``obtener_valor_referencial_sap``.
Returns:
str — Tabla HTML con una fila por guía, el total SAP, el valor
del XML y un ícono de coincidencia.
"""
valor_xml = sum(l.get("valor_referencial") or 0 for l in factura.get("lineas", []))
total_sap = sum(g["valor_sap"] for g in guias_valor_ref)
rows = "".join(
f"<tr><td>09{g['serie']}-{g['numero']}</td><td>{g['valor_sap']:.2f}</td></tr>"
for g in guias_valor_ref
)
coincide = abs(valor_xml - total_sap) < 0.01
icono = "✅" if coincide else "❌"
return (
"<table border='1' cellpadding='6' cellspacing='0' style='border-collapse:collapse;'>"
"<tr><th>Guía de Remisión</th><th>Valor Referencial SAP</th></tr>"
f"{rows}"
f"<tr><td><b>Total SAP</b></td><td><b>{total_sap:.2f}</b></td></tr>"
f"<tr><td><b>Valor Referencial XML</b></td><td><b>{valor_xml:.2f}</b></td></tr>"
f"<tr><td colspan='2'>{icono} {'Coincide' if coincide else 'No coincide'}</td></tr>"
"</table>"
)
# ============================================================================
def build_notification_body(status: str, factura: dict, validation_table: str, valor_referencial_table: str, **kwargs) -> str:
"""
Construye el cuerpo HTML del email de notificación del resultado
del proceso SAP.
Args:
status: ``"created"`` (draft creado), ``"error"`` (error al
crear), o cualquier otro valor (validación falló, draft no
creado).
factura: Diccionario de la factura (se usa ``serie``,
``numero``, ``razon_social_emisor``).
validation_table: Tabla HTML generada por
``build_validation_html``.
valor_referencial_table: Tabla HTML generada por
``build_valor_referencial_html``.
**kwargs: Argumentos adicionales:
- ``doc_entry`` (int): DocEntry del draft creado.
- ``error`` (str): Mensaje de error si ocurrió.
Returns:
str — Cuerpo HTML completo del email.
Example:
Draft creado exitosamente::
body = build_notification_body(
"created", factura, tabla_html, valref_html, doc_entry=2021
)
# → "<p>✅ Draft SAP B1 creado exitosamente.</p>..."
"""
base = (
f"<p><b>Factura:</b> {factura.get('serie')}-{factura.get('numero')}</p>"
f"<p><b>Emisor:</b> {factura.get('razon_social_emisor', '')}</p>"
)
valref_section = f"<h3>Valor Referencial (XML vs SAP)</h3>{valor_referencial_table}"
if status == "created":
return f"<p>✅ Draft SAP B1 creado exitosamente.</p><p><b>DocEntry:</b> {kwargs.get('doc_entry')}</p>{base}<h3>Validación de precios</h3>{validation_table}{valref_section}"
elif status == "error":
return f"<p>❌ Error al crear Draft SAP B1.</p>{base}<p><b>Error:</b> {kwargs.get('error')}</p><h3>Validación de precios</h3>{validation_table}{valref_section}"
else:
return f"<p>⏭️ Draft NO creado. Los precios de las guías no coinciden con SAP.</p>{base}<h3>Validación de precios</h3>{validation_table}{valref_section}"
persistencia.py
+163
View File
@@ -0,0 +1,163 @@
"""Persistencia a disco (JSON) de resultados y drafts."""
import os
import re
import json
from datetime import datetime
import sunat_schema
import pdf_utilities
from pipeline_utils import log
# ============================================================================
def guardar_resultado(documentos: list[dict], datos_pdf: list[dict], output_folder: str) -> list[str]:
"""
Guarda los resultados XML y PDF en archivos JSON separados.
El nombre del archivo sigue el patrón
``{tipo}_resultado_{serie}-{numero}_{timestamp}.json`` y se escribe en
``output_folder``.
Args:
documentos: Lista de diccionarios con datos extraídos de XML
(producida por ``procesar_xmls``).
datos_pdf: Lista de diccionarios con datos extraídos de PDF
(producida por ``procesar_pdfs``).
Example:
Si ``documentos[0]`` tiene ``serie="F002", numero="486"``::
guardar_resultado(documentos, pdf_datos, output_folder)
# → Crea xml_resultado_F002-486_2026-07-08_12-00-00.json
# → Crea pdf_resultado_F002-486_2026-07-08_12-00-00.json
"""
if not documentos:
return []
if not os.path.exists(output_folder):
os.makedirs(output_folder, exist_ok=True)
ts = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
serie = documentos[0].get("serie", "unknown")
numero = documentos[0].get("numero", "unknown")
base = f"{serie}-{numero}_{ts}"
saved: list[str] = []
# XML result
xml_name = f"xml_resultado_{base}.json"
xml_path = os.path.join(output_folder, xml_name)
with open(xml_path, "w", encoding="utf-8") as f:
json.dump(documentos, f, ensure_ascii=False, indent=4)
log(f"💾 XML guardado: {xml_name}")
saved.append(xml_path)
# PDF result (normalizado al mismo schema que XML)
pdf_docs = []
for raw in datos_pdf:
if raw.get("error"):
continue
td = (raw.get("tipo_documento") or "").upper()
tipo = pdf_utilities.TIPO_MAP.get(td)
if not tipo:
continue
sn = raw.get("serie_numero") or ""
serie_p, num_p = sunat_schema.split_serie_numero(sn)
if not serie_p:
continue
doc = {
"ruc_emisor": raw.get("ruc_emisor"),
"razon_social_emisor": None, # el PDF no extrae la razón social del emisor
"fecha_emision": raw.get("fecha_emision"),
"tipo": tipo,
"serie": serie_p,
"numero": num_p,
"ruc_receptor": raw.get("ruc_receptor"),
"razon_social_receptor": raw.get("nombre_receptor"),
"total_factura": raw.get("monto_total"),
"documentos_referencia": [],
"tiene_detraccion": False,
"codigo_detraccion": None,
"porcentaje_detraccion": None,
"monto_detraccion": None,
"lineas": [],
"ruta_pdf": raw.get("archivo"),
}
det = raw.get("detraccion") or {}
if det.get("tiene"):
doc["tiene_detraccion"] = True
doc["porcentaje_detraccion"] = det.get("porcentaje")
doc["monto_detraccion"] = det.get("monto")
for item in raw.get("items") or []:
linea = {
"descripcion": item.get("descripcion", ""),
"cantidad": item.get("cantidad"),
"precio_unitario": item.get("precio_unitario"),
"total_linea": item.get("total"),
"guias_remision": [],
"guia_transportista": None,
}
for g in item.get("guias_remision") or []:
if isinstance(g, str):
m = re.search(r"\b([TVE][A-Z0-9]{3,4})\s*-?\s*(\d+)", g)
if m:
linea["guias_remision"].append({
"tipo": "09",
"serie": m.group(1).upper(),
"numero": m.group(2).lstrip("0") or "0",
})
elif isinstance(g, dict):
linea["guias_remision"].append(dict(g))
gt = item.get("guia_transportista")
if gt:
if isinstance(gt, str):
m = re.search(r"\b([TVE][A-Z0-9]{3,4})\s*-?\s*(\d+)", gt)
if m:
linea["guia_transportista"] = {
"tipo": "31",
"serie": m.group(1).upper(),
"numero": m.group(2).lstrip("0") or "0",
}
elif isinstance(gt, dict):
linea["guia_transportista"] = dict(gt)
doc["lineas"].append(linea)
pdf_docs.append(doc)
if pdf_docs:
pdf_name = f"pdf_resultado_{base}.json"
pdf_path = os.path.join(output_folder, pdf_name)
with open(pdf_path, "w", encoding="utf-8") as f:
json.dump(pdf_docs, f, ensure_ascii=False, indent=4)
log(f"💾 PDF guardado: {pdf_name}")
saved.append(pdf_path)
return saved
# ============================================================================
def guardar_draft_json(draft: dict, serie: str, numero: str, output_folder: str) -> None:
"""
Guarda el payload del draft SAP B1 en un archivo JSON para
depuración.
El nombre sigue el patrón
``draft_sapb1_{serie}-{numero}_{timestamp}.json``.
Args:
draft: Payload del draft SAP B1 (producido por
``build_sap_draft``).
serie: Serie de la factura (ej. ``"F002"``).
numero: Número de la factura (ej. ``"486"``).
Example:
Draft de factura F002-486::
guardar_draft_json(draft, "F002", "486", output_folder)
# → Crea draft_sapb1_F002-486_2026-07-08_12-00-00.json
"""
if not os.path.exists(output_folder):
os.makedirs(output_folder, exist_ok=True)
ts = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
filename = f"draft_sapb1_{serie}-{numero}_{ts}.json"
with open(os.path.join(output_folder, filename), "w", encoding="utf-8") as f:
json.dump(draft, f, ensure_ascii=False, indent=4)
log(f"💾 Draft guardado: {filename}")
pipeline.py
+35 -1137
View File
File diff suppressed because it is too large. Load diff
pipeline_utils.py
+15
View File
@@ -0,0 +1,15 @@
"""Utilidades compartidas por los módulos del pipeline."""
import os
def ruta_carpeta(subcarpeta: str) -> str:
"""Retorna la ruta absoluta de ``subcarpeta`` relativa a la raíz del proyecto."""
return os.path.join(os.path.dirname(__file__), subcarpeta)
def log(msg: str, level: int = 1, last: bool = False) -> None:
"""Imprime ``msg`` con la indentación de árbol usada en el log del pipeline."""
branch = " └── " if last else " ├── "
pfx = {1: branch, 2: f" │ {branch}", 3: f" │ │ {branch}", 4: f" │ │ │ {branch}"}.get(level, branch)
print(f"{pfx}{msg}")