ahora se procesa de inicio a fin correo por correo
This commit is contained in:
George committed
1 parent
99b7caf535
commit
7fa7204442
25 files changed
+2234
-334
No files matched your search
+89
-14
@@ -6,12 +6,12 @@ Soporta los siguientes tipos de documento UBL 2.1:
|
||||
- DespatchAdvice (Guía de Remisión Remitente): tipo 09
|
||||
- DespatchAdvice (Guía de Remisión Transportista): tipo 31
|
||||
"""
|
||||
|
||||
import re
|
||||
import xml.etree.ElementTree as ET
|
||||
from dataclasses import dataclass, asdict, field
|
||||
from typing import Optional, List
|
||||
|
||||
########################################################
|
||||
# ============================================================================
|
||||
# Namespaces UBL 2.1 utilizados en los XML de SUNAT
|
||||
NAMESPACES = {
|
||||
"inv": "urn:oasis:names:specification:ubl:schema:xsd:Invoice-2",
|
||||
@@ -30,7 +30,7 @@ TIPOS_DOCUMENTO = {
|
||||
"09": "Guía de Remisión Remitente",
|
||||
"31": "Guía de Remisión Transportista",
|
||||
}
|
||||
########################################################
|
||||
# ============================================================================
|
||||
@dataclass
|
||||
class DocumentoReferencia:
|
||||
"""Representa un documento de referencia (p.ej. Guía de Remisión asociada)."""
|
||||
@@ -46,6 +46,7 @@ class DetalleLinea:
|
||||
descripcion: str
|
||||
cantidad: float
|
||||
precio_unitario: float
|
||||
total_linea: float = 0.0
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -121,7 +122,7 @@ class DatosComprobante:
|
||||
)
|
||||
|
||||
|
||||
########################################################
|
||||
# ============================================================================
|
||||
def _find_text(element: ET.Element, xpath: str, namespaces: dict) -> Optional[str]:
|
||||
"""Busca un elemento por XPath y devuelve su texto, o None si no existe."""
|
||||
node = element.find(xpath, namespaces)
|
||||
@@ -129,7 +130,7 @@ def _find_text(element: ET.Element, xpath: str, namespaces: dict) -> Optional[st
|
||||
return node.text.strip()
|
||||
return None
|
||||
|
||||
########################################################
|
||||
# ============================================================================
|
||||
def _split_serie_numero(comprobante_id: str) -> tuple[str, str]:
|
||||
"""
|
||||
Separa el ID del comprobante en serie y número, removiendo ceros a la izquierda del número.
|
||||
@@ -144,7 +145,26 @@ def _split_serie_numero(comprobante_id: str) -> tuple[str, str]:
|
||||
return partes[0], numero
|
||||
return comprobante_id, ""
|
||||
|
||||
########################################################
|
||||
# ============================================================================
|
||||
def _extraer_serie_numero_desde_texto(texto: str) -> tuple[str, str] | None:
|
||||
texto = re.sub(r'[–—−]', '-', texto)
|
||||
patron = r'\b([TVE][A-Z0-9]{3})\s*-\s*(\d+)\b'
|
||||
m = re.search(patron, texto)
|
||||
if m:
|
||||
return m.group(1).upper(), m.group(2)
|
||||
return None
|
||||
|
||||
|
||||
def obtener_documentos_referencia_detalle_factura(texto: str) -> list[str]:
|
||||
texto = re.sub(r'[–—−]', '-', texto)
|
||||
|
||||
patron = r'\b([TVE][A-Z0-9]{3})\s*-\s*(\d+)\b'
|
||||
|
||||
return list(dict.fromkeys(
|
||||
f"{serie.upper()}-{numero}"
|
||||
for serie, numero in re.findall(patron, texto, flags=re.IGNORECASE)
|
||||
))
|
||||
# ============================================================================
|
||||
def _parse_invoice(root: ET.Element) -> DatosComprobante:
|
||||
"""Extrae datos de un documento Invoice (Factura/Boleta)."""
|
||||
ns = NAMESPACES
|
||||
@@ -203,8 +223,27 @@ def _parse_invoice(root: ET.Element) -> DatosComprobante:
|
||||
ref_id = _find_text(ref_node, "cbc:ID", ns) or ""
|
||||
ref_tipo = _find_text(ref_node, "cbc:DocumentTypeCode", ns) or ""
|
||||
if ref_id or ref_tipo:
|
||||
documentos_referencia.append(DocumentoReferencia(tipo=ref_tipo, serie=_split_serie_numero(ref_id)[0], numero=_split_serie_numero(ref_id)[1]))
|
||||
parsed = _extraer_serie_numero_desde_texto(ref_id)
|
||||
if parsed:
|
||||
ref_serie, ref_numero = parsed
|
||||
else:
|
||||
ref_serie, ref_numero = _split_serie_numero(ref_id)
|
||||
documentos_referencia.append(DocumentoReferencia(tipo=ref_tipo, serie=ref_serie, numero=ref_numero))
|
||||
|
||||
# si los documentos de referencia non estan en los tags de SUNAT
|
||||
# buscamos estos documentos en el cuerpo de la factura
|
||||
if len(documentos_referencia) == 0:
|
||||
first_invoice_line = root.find('.//cac:InvoiceLine', ns)
|
||||
if first_invoice_line is not None:
|
||||
item = first_invoice_line.find('cac:Item', ns)
|
||||
if item is not None:
|
||||
description = item.find('cbc:Description', ns)
|
||||
if description is not None:
|
||||
guias = obtener_documentos_referencia_detalle_factura(description.text)
|
||||
if guias:
|
||||
for guia in guias:
|
||||
documentos_referencia.append(DocumentoReferencia(tipo="09" if guia[:1] in ("T", "V", "E") else "31", serie=_split_serie_numero(guia)[0], numero=_split_serie_numero(guia)[1]))
|
||||
|
||||
# Inicializar campos de detracción y líneas
|
||||
tiene_detraccion = False
|
||||
codigo_detraccion = None
|
||||
@@ -242,11 +281,15 @@ def _parse_invoice(root: ET.Element) -> DatosComprobante:
|
||||
)
|
||||
precio_unitario = float(price_str) if price_str else 0.0
|
||||
|
||||
total_linea_str = _find_text(line_node, "cbc:LineExtensionAmount", ns)
|
||||
total_linea = float(total_linea_str) if total_linea_str else 0.0
|
||||
|
||||
lineas.append(
|
||||
DetalleLinea(
|
||||
descripcion=descripcion,
|
||||
cantidad=cantidad,
|
||||
precio_unitario=precio_unitario,
|
||||
total_linea=total_linea,
|
||||
)
|
||||
)
|
||||
|
||||
@@ -268,7 +311,7 @@ def _parse_invoice(root: ET.Element) -> DatosComprobante:
|
||||
razon_social_emisor=razon_social_emisor,
|
||||
)
|
||||
|
||||
########################################################
|
||||
# ============================================================================
|
||||
def _parse_despatch_advice(root: ET.Element) -> DatosComprobante:
|
||||
"""Extrae datos de un documento DespatchAdvice (Guía de Remisión)."""
|
||||
ns = NAMESPACES
|
||||
@@ -354,6 +397,37 @@ def _parse_despatch_advice(root: ET.Element) -> DatosComprobante:
|
||||
|
||||
# Las guías de remisión no tienen monto total
|
||||
total_factura = None
|
||||
|
||||
# Documentos de referencia
|
||||
documentos_referencia = []
|
||||
for ref_node in root.findall("cac:AdditionalDocumentReference", ns):
|
||||
ref_id = _find_text(ref_node, "cbc:ID", ns) or ""
|
||||
ref_tipo = _find_text(ref_node, "cbc:DocumentTypeCode", ns) or ""
|
||||
if ref_id or ref_tipo:
|
||||
parsed = _extraer_serie_numero_desde_texto(ref_id)
|
||||
if parsed:
|
||||
ref_serie, ref_numero = parsed
|
||||
else:
|
||||
ref_serie, ref_numero = _split_serie_numero(ref_id)
|
||||
documentos_referencia.append(DocumentoReferencia(tipo=ref_tipo, serie=ref_serie, numero=ref_numero))
|
||||
|
||||
lineas = []
|
||||
# Obtener cada cac:InvoiceLine
|
||||
for line_node in root.findall("cac:DespatchLine", ns):
|
||||
descripcion = _find_text(line_node, "cac:Item/cbc:Description", ns) or ""
|
||||
|
||||
qty_str = _find_text(line_node, "cbc:DeliveredQuantity", ns)
|
||||
cantidad = float(qty_str) if qty_str else 0.0
|
||||
|
||||
precio_unitario = 0.0
|
||||
|
||||
lineas.append(
|
||||
DetalleLinea(
|
||||
descripcion=descripcion,
|
||||
cantidad=cantidad,
|
||||
precio_unitario=precio_unitario,
|
||||
)
|
||||
)
|
||||
|
||||
return DatosComprobante(
|
||||
ruc_emisor=ruc_emisor,
|
||||
@@ -365,6 +439,7 @@ def _parse_despatch_advice(root: ET.Element) -> DatosComprobante:
|
||||
ruc_receptor=ruc_receptor,
|
||||
razon_social_receptor=razon_social_receptor,
|
||||
total_factura=total_factura,
|
||||
documentos_referencia=documentos_referencia,
|
||||
punto_partida=punto_partida,
|
||||
punto_llegada=punto_llegada,
|
||||
peso_bruto=peso_bruto,
|
||||
@@ -375,10 +450,11 @@ def _parse_despatch_advice(root: ET.Element) -> DatosComprobante:
|
||||
licencia_conductor=licencia_conductor,
|
||||
nombre_conductor=nombre_conductor,
|
||||
documento_conductor=documento_conductor,
|
||||
ubigeo_partida=ubigeo_partida
|
||||
ubigeo_partida=ubigeo_partida,
|
||||
lineas=lineas
|
||||
)
|
||||
|
||||
########################################################
|
||||
# ============================================================================
|
||||
def extraer_datos_xml(ruta_xml: str) -> Optional[dict]:
|
||||
"""
|
||||
Recibe la ruta de un archivo XML de comprobante electrónico SUNAT.
|
||||
@@ -414,11 +490,9 @@ def extraer_datos_xml(ruta_xml: str) -> Optional[dict]:
|
||||
return datos.to_dict()
|
||||
return None
|
||||
|
||||
|
||||
########################################################
|
||||
# ---------------------------------------------------------------------------
|
||||
# ============================================================================
|
||||
# Ejecución directa: procesa todos los XML de la carpeta adjuntos como prueba
|
||||
# ---------------------------------------------------------------------------
|
||||
# ============================================================================
|
||||
if __name__ == "__main__":
|
||||
import os
|
||||
import glob
|
||||
@@ -443,3 +517,4 @@ if __name__ == "__main__":
|
||||
except Exception as e:
|
||||
print(f"\nError procesando {os.path.basename(archivo)}: {e}\n")
|
||||
print("=" * 70)
|
||||
# ============================================================================
|
||||
Reference in new issue
Block a user