se agrego el modulo sunat_schema.py que lee el archivo xml y extrae la información
This commit is contained in:
George committed
1 parent
937c4d6260
commit
99b7caf535
4 files changed
+516
-11
No files matched your search
@@ -1,11 +1,11 @@
|
||||
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import os
|
||||
import email_utilities
|
||||
import outlook_utilities
|
||||
import database_utilities
|
||||
import logger_utilities
|
||||
import sunat_schema
|
||||
from datetime import datetime, timedelta
|
||||
from config_loader import CONFIG
|
||||
|
||||
@@ -77,6 +77,67 @@ def dejar_mail_si_tiene_adjunto_pdf_o_xml(emails):
|
||||
return emails_con_pdf_o_xml
|
||||
|
||||
########################################################
|
||||
def leer_xml_obtener_datos_validos(emails):
|
||||
json_final = []
|
||||
for e in emails:
|
||||
print(f"\n📬 [{e['fecha']}] {e['subject'][:80]}...")
|
||||
#adjuntos originales
|
||||
print(f" ├── 📂 Adjuntos originales:")
|
||||
|
||||
for idx, a in enumerate(e['adjuntos']):
|
||||
prefijo = " │ └── " if idx == len(e['adjuntos'])-1 else " │ ├── "
|
||||
print(f"{prefijo}{a}")
|
||||
if a.lower().endswith(".xml"):
|
||||
# procesar archivo xml de SUNAT
|
||||
carpeta_adjuntos = os.path.join(os.path.dirname(__file__), CONFIG.client.ruta_descarga)
|
||||
archivo_xml = os.path.join(carpeta_adjuntos, a)
|
||||
print(f" │ 📄 Procesando XML: {a}")
|
||||
data = sunat_schema.extraer_datos_xml(archivo_xml)
|
||||
if data:
|
||||
print(f" │ ✅ Factura extraída: {data.get('serie')}-{data.get('numero')}")
|
||||
json_final.append(data)
|
||||
else:
|
||||
print(" │ ℹ️ XML sin datos (no es factura tipo 01)")
|
||||
|
||||
# opcional: guardar JSON junto al XML
|
||||
archivo_json = f"ejecucion_{datetime.now().strftime('%Y-%m-%d_%H-%M-%S')}.json"
|
||||
with open(archivo_json, "w", encoding="utf-8") as f:
|
||||
json.dump(json_final, f, ensure_ascii=False, indent=4)
|
||||
|
||||
return json_final
|
||||
|
||||
########################################################
|
||||
def anidar_guias_en_facturas(json_final):
|
||||
# obtener solo facturas
|
||||
facturas = [item for item in json_final if item.get("tipo") == "01"]
|
||||
|
||||
# obtener solo guias
|
||||
guias = [item for item in json_final if item.get("tipo") == "31"]
|
||||
|
||||
# recorrer cada factura y buscar guias relacionadas de campo "documento_relacionado" que tenga el mismo numero de serie y numero
|
||||
for factura in facturas:
|
||||
# obtener la guia de tipo ='31' de los documemtos_relacionados
|
||||
guias_transporte = [item for item in factura.get("documentos_referencia", []) if item.get("tipo") == "31"]
|
||||
|
||||
# buscar en la lista de guias si existe una guia que tenga el mismo numero de serie y numero que la guia de transporte
|
||||
for guia in guias_transporte:
|
||||
guia_encontrada = next((g for g in guias if g.get("serie")== guia.get("serie")
|
||||
and g.get("numero")== guia.get("numero")), None)
|
||||
if guia_encontrada:
|
||||
# si se encuentra, anidar la guia dentro de la factura en un nuevo campo "guia_trasporte"
|
||||
factura["guia_transporte"] = guia_encontrada
|
||||
print(f" ├── 📦 Guía {guia_encontrada.get('serie')}-{guia_encontrada.get('numero')} anidada en factura {factura.get('serie')}-{factura.get('numero')}")
|
||||
else:
|
||||
print(f" ├── ⚠️ Guía {guia.get('serie')}-{guia.get('numero')} no encontrada para factura {factura.get('serie')}-{factura.get('numero')}")
|
||||
|
||||
# opcional: guardar JSON junto al XML
|
||||
archivo_json = f"anidado_{datetime.now().strftime('%Y-%m-%d_%H-%M-%S')}.json"
|
||||
with open(archivo_json, "w", encoding="utf-8") as f:
|
||||
json.dump(facturas, f, ensure_ascii=False, indent=4)
|
||||
|
||||
|
||||
return facturas
|
||||
########################################################
|
||||
def main():
|
||||
|
||||
# Inicializar Base de Datos
|
||||
@@ -84,7 +145,7 @@ def main():
|
||||
|
||||
# Calcular fechas
|
||||
hoy = datetime.now()
|
||||
ayer = hoy - timedelta(days=25)
|
||||
ayer = hoy - timedelta(days=30)
|
||||
|
||||
fecha_inicio = ayer.strftime("%Y-%m-%d")
|
||||
fecha_fin = hoy.strftime("%Y-%m-%d")
|
||||
@@ -116,16 +177,12 @@ def main():
|
||||
# también se eliminar los adjuntos no que sean xml y pdf
|
||||
emails = dejar_mail_si_tiene_adjunto_pdf_o_xml(emails)
|
||||
|
||||
# obtener datos de los archivos xml y generar JSON final
|
||||
json_final = leer_xml_obtener_datos_validos(emails)
|
||||
|
||||
anidar_guias_en_facturas(json_final)
|
||||
|
||||
|
||||
for e in emails:
|
||||
print(f"\n📬 [{e['fecha']}] {e['subject'][:80]}...")
|
||||
|
||||
#adjuntos originales
|
||||
print(f" ├── 📂 Adjuntos originales:")
|
||||
for idx, a in enumerate(e['adjuntos']):
|
||||
prefijo = " │ └── " if idx == len(e['adjuntos'])-1 else " │ ├── "
|
||||
print(f"{prefijo}{a}")
|
||||
|
||||
########################################################
|
||||
if __name__ == "__main__":
|
||||
|
||||
+445
@@ -0,0 +1,445 @@
|
||||
"""
|
||||
Módulo para extraer datos de comprobantes electrónicos SUNAT desde archivos XML.
|
||||
|
||||
Soporta los siguientes tipos de documento UBL 2.1:
|
||||
- Invoice (Factura/Boleta): tipos 01, 03, etc.
|
||||
- DespatchAdvice (Guía de Remisión Remitente): tipo 09
|
||||
- DespatchAdvice (Guía de Remisión Transportista): tipo 31
|
||||
"""
|
||||
|
||||
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",
|
||||
"desp": "urn:oasis:names:specification:ubl:schema:xsd:DespatchAdvice-2",
|
||||
"cac": "urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2",
|
||||
"cbc": "urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2",
|
||||
"ext": "urn:oasis:names:specification:ubl:schema:xsd:CommonExtensionComponents-2",
|
||||
}
|
||||
|
||||
# Catálogo 01 SUNAT - Tipos de documento
|
||||
TIPOS_DOCUMENTO = {
|
||||
"01": "Factura",
|
||||
"03": "Boleta de Venta",
|
||||
"07": "Nota de Crédito",
|
||||
"08": "Nota de Débito",
|
||||
"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)."""
|
||||
#id: str
|
||||
tipo: str
|
||||
serie: Optional[str] = None
|
||||
numero: Optional[str] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class DetalleLinea:
|
||||
"""Representa una línea de detalle del comprobante."""
|
||||
descripcion: str
|
||||
cantidad: float
|
||||
precio_unitario: float
|
||||
|
||||
|
||||
@dataclass
|
||||
class DatosComprobante:
|
||||
"""Representa los datos extraídos de un comprobante electrónico SUNAT."""
|
||||
|
||||
ruc_emisor: str
|
||||
razon_social_emisor: str
|
||||
fecha_emision: str
|
||||
tipo: str
|
||||
serie: str
|
||||
numero: str
|
||||
ruc_receptor: str
|
||||
razon_social_receptor: str
|
||||
total_factura: Optional[float]
|
||||
documentos_referencia: List[DocumentoReferencia] = field(default_factory=list)
|
||||
tiene_detraccion: bool = False
|
||||
codigo_detraccion: Optional[str] = None
|
||||
porcentaje_detraccion: Optional[float] = None
|
||||
monto_detraccion: Optional[float] = None
|
||||
punto_partida: Optional[str] = None
|
||||
punto_llegada: Optional[str] = None
|
||||
peso_bruto: Optional[str] = None
|
||||
unidad_peso_bruto: Optional[str] = None
|
||||
fecha_traslado: Optional[str] = None
|
||||
vehiculo_principal: Optional[str] = None
|
||||
vehiculo_secundario: Optional[str] = None
|
||||
licencia_conductor: Optional[str] = None
|
||||
nombre_conductor: Optional[str] = None
|
||||
documento_conductor: Optional[str] = None
|
||||
ubigeo_partida: Optional[str] = None
|
||||
lineas: List[DetalleLinea] = field(default_factory=list)
|
||||
|
||||
@property
|
||||
def tipo_descripcion(self) -> str:
|
||||
"""Devuelve la descripción del tipo de documento según el catálogo 01 SUNAT."""
|
||||
return TIPOS_DOCUMENTO.get(self.tipo, f"Desconocido ({self.tipo})")
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
"""Convierte el dataclass a diccionario."""
|
||||
resultado = asdict(self)
|
||||
resultado["tipo_descripcion"] = self.tipo_descripcion
|
||||
return resultado
|
||||
|
||||
def __str__(self) -> str:
|
||||
total_str = f"S/ {self.total_factura:,.2f}" if self.total_factura is not None else "N/A"
|
||||
refs_str = ""
|
||||
if self.documentos_referencia:
|
||||
refs_str = "\n Documentos Ref: " + ", ".join(
|
||||
f"{ref.id} (Tipo {ref.tipo})" for ref in self.documentos_referencia
|
||||
)
|
||||
detraccion_str = ""
|
||||
if self.tiene_detraccion:
|
||||
detraccion_str = (
|
||||
f"\n Detracción: S/ {self.monto_detraccion:,.2f} "
|
||||
f"({self.porcentaje_detraccion}% - Código: {self.codigo_detraccion})"
|
||||
)
|
||||
lineas_str = ""
|
||||
if self.lineas:
|
||||
lineas_str = "\n Líneas de Detalle:" + "".join(
|
||||
f"\n - Cant: {l.cantidad} | P.U.: S/ {l.precio_unitario:.4f} | Desc: {l.descripcion}"
|
||||
for l in self.lineas
|
||||
)
|
||||
return (
|
||||
f"Comprobante: {self.tipo_descripcion} {self.serie}-{self.numero}\n"
|
||||
f" Emisor: RUC {self.ruc_emisor}\n"
|
||||
f" Receptor: RUC {self.ruc_receptor} - {self.razon_social_receptor}\n"
|
||||
f" Fecha: {self.fecha_emision}\n"
|
||||
f" Total: {total_str}"
|
||||
f"{refs_str}"
|
||||
f"{detraccion_str}"
|
||||
f"{lineas_str}"
|
||||
)
|
||||
|
||||
|
||||
########################################################
|
||||
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)
|
||||
if node is not None and node.text:
|
||||
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.
|
||||
Ejemplo: 'F002-3091' -> ('F002', '3091')
|
||||
'T001-00128143' -> ('T001', '128143')
|
||||
'EG03-1752' -> ('EG03', '1752')
|
||||
"""
|
||||
if "-" in comprobante_id:
|
||||
partes = comprobante_id.split("-", 1)
|
||||
# Remueve ceros a la izquierda, si queda vacío (ej: '000') se mantiene '0'
|
||||
numero = partes[1].lstrip("0") or "0"
|
||||
return partes[0], numero
|
||||
return comprobante_id, ""
|
||||
|
||||
########################################################
|
||||
def _parse_invoice(root: ET.Element) -> DatosComprobante:
|
||||
"""Extrae datos de un documento Invoice (Factura/Boleta)."""
|
||||
ns = NAMESPACES
|
||||
|
||||
# ID del comprobante (ej: F002-3091)
|
||||
comprobante_id = _find_text(root, "inv:ID", ns) or _find_text(root, "cbc:ID", ns) or ""
|
||||
serie, numero = _split_serie_numero(comprobante_id)
|
||||
|
||||
# Fecha de emisión
|
||||
fecha_emision = _find_text(root, "inv:IssueDate", ns) or _find_text(root, "cbc:IssueDate", ns) or ""
|
||||
|
||||
# Tipo de documento (catálogo 01)
|
||||
tipo = _find_text(root, "inv:InvoiceTypeCode", ns) or _find_text(root, "cbc:InvoiceTypeCode", ns) or ""
|
||||
|
||||
# Datos del emisor (AccountingSupplierParty)
|
||||
ruc_emisor = _find_text(
|
||||
root,
|
||||
"cac:AccountingSupplierParty/cac:Party/cac:PartyIdentification/cbc:ID",
|
||||
ns,
|
||||
) or ""
|
||||
|
||||
razon_social_emisor = _find_text(
|
||||
root,
|
||||
"cac:AccountingSupplierParty/cac:Party/cac:PartyName/cbc:Name",
|
||||
ns,
|
||||
) or _find_text(
|
||||
root,
|
||||
"cac:AccountingSupplierParty/cac:Party/cac:PartyLegalEntity/cbc:RegistrationName",
|
||||
ns,
|
||||
) or ""
|
||||
|
||||
# Datos del receptor (AccountingCustomerParty)
|
||||
ruc_receptor = _find_text(
|
||||
root,
|
||||
"cac:AccountingCustomerParty/cac:Party/cac:PartyIdentification/cbc:ID",
|
||||
ns,
|
||||
) or ""
|
||||
|
||||
razon_social_receptor = _find_text(
|
||||
root,
|
||||
"cac:AccountingCustomerParty/cac:Party/cac:PartyLegalEntity/cbc:RegistrationName",
|
||||
ns,
|
||||
) or ""
|
||||
|
||||
# Total (PayableAmount en LegalMonetaryTotal)
|
||||
total_str = _find_text(
|
||||
root,
|
||||
"cac:LegalMonetaryTotal/cbc:PayableAmount",
|
||||
ns,
|
||||
)
|
||||
total_factura = float(total_str) if total_str else None
|
||||
|
||||
# Documentos de referencia
|
||||
documentos_referencia = []
|
||||
for ref_node in root.findall("cac:DespatchDocumentReference", 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:
|
||||
documentos_referencia.append(DocumentoReferencia(tipo=ref_tipo, serie=_split_serie_numero(ref_id)[0], numero=_split_serie_numero(ref_id)[1]))
|
||||
|
||||
# Inicializar campos de detracción y líneas
|
||||
tiene_detraccion = False
|
||||
codigo_detraccion = None
|
||||
porcentaje_detraccion = None
|
||||
monto_detraccion = None
|
||||
lineas = []
|
||||
|
||||
# Solo si el tipo es Factura (01) se extraen detracciones y líneas detalladas
|
||||
if tipo == "01":
|
||||
# Buscar detracción en PaymentTerms
|
||||
for pt_node in root.findall("cac:PaymentTerms", ns):
|
||||
pt_id = _find_text(pt_node, "cbc:ID", ns)
|
||||
if pt_id == "Detraccion":
|
||||
tiene_detraccion = True
|
||||
codigo_detraccion = _find_text(pt_node, "cbc:PaymentMeansID", ns)
|
||||
|
||||
percent_str = _find_text(pt_node, "cbc:PaymentPercent", ns)
|
||||
porcentaje_detraccion = float(percent_str) if percent_str else None
|
||||
|
||||
amount_str = _find_text(pt_node, "cbc:Amount", ns)
|
||||
monto_detraccion = float(amount_str) if amount_str else None
|
||||
break
|
||||
|
||||
# Obtener cada cac:InvoiceLine
|
||||
for line_node in root.findall("cac:InvoiceLine", ns):
|
||||
descripcion = _find_text(line_node, "cac:Item/cbc:Description", ns) or ""
|
||||
|
||||
qty_str = _find_text(line_node, "cbc:InvoicedQuantity", ns)
|
||||
cantidad = float(qty_str) if qty_str else 0.0
|
||||
|
||||
# Buscar el precio en cac:PricingReference/cac:AlternativeConditionPrice/cbc:PriceAmount o cac:Price/cbc:PriceAmount
|
||||
price_str = (
|
||||
_find_text(line_node, "cac:PricingReference/cac:AlternativeConditionPrice/cbc:PriceAmount", ns)
|
||||
or _find_text(line_node, "cac:Price/cbc:PriceAmount", ns)
|
||||
)
|
||||
precio_unitario = float(price_str) if price_str else 0.0
|
||||
|
||||
lineas.append(
|
||||
DetalleLinea(
|
||||
descripcion=descripcion,
|
||||
cantidad=cantidad,
|
||||
precio_unitario=precio_unitario,
|
||||
)
|
||||
)
|
||||
|
||||
return DatosComprobante(
|
||||
ruc_emisor=ruc_emisor,
|
||||
fecha_emision=fecha_emision,
|
||||
tipo=tipo,
|
||||
serie=serie,
|
||||
numero=numero,
|
||||
ruc_receptor=ruc_receptor,
|
||||
razon_social_receptor=razon_social_receptor,
|
||||
total_factura=total_factura,
|
||||
documentos_referencia=documentos_referencia,
|
||||
tiene_detraccion=tiene_detraccion,
|
||||
codigo_detraccion=codigo_detraccion,
|
||||
porcentaje_detraccion=porcentaje_detraccion,
|
||||
monto_detraccion=monto_detraccion,
|
||||
lineas=lineas,
|
||||
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
|
||||
|
||||
# ID del comprobante (ej: T001-00128143, EG03-1752)
|
||||
comprobante_id = _find_text(root, "desp:ID", ns) or _find_text(root, "cbc:ID", ns) or ""
|
||||
serie, numero = _split_serie_numero(comprobante_id)
|
||||
|
||||
# Fecha de emisión
|
||||
fecha_emision = _find_text(root, "desp:IssueDate", ns) or _find_text(root, "cbc:IssueDate", ns) or ""
|
||||
|
||||
# Tipo de documento (catálogo 01)
|
||||
tipo = (
|
||||
_find_text(root, "desp:DespatchAdviceTypeCode", ns)
|
||||
or _find_text(root, "cbc:DespatchAdviceTypeCode", ns)
|
||||
or ""
|
||||
)
|
||||
|
||||
# Datos del emisor (DespatchSupplierParty)
|
||||
ruc_emisor = _find_text(
|
||||
root,
|
||||
"cac:DespatchSupplierParty/cac:Party/cac:PartyIdentification/cbc:ID",
|
||||
ns,
|
||||
) or ""
|
||||
|
||||
razon_social_emisor = _find_text(
|
||||
root,
|
||||
"cac:DespatchSupplierParty/cac:Party/cac:PartyName/cbc:Name",
|
||||
ns,
|
||||
) or _find_text(
|
||||
root,
|
||||
"cac:DespatchSupplierParty/cac:Party/cac:PartyLegalEntity/cbc:RegistrationName",
|
||||
ns,
|
||||
) or ""
|
||||
|
||||
# Datos del receptor (DeliveryCustomerParty)
|
||||
ruc_receptor = _find_text(
|
||||
root,
|
||||
"cac:DeliveryCustomerParty/cac:Party/cac:PartyIdentification/cbc:ID",
|
||||
ns,
|
||||
) or ""
|
||||
|
||||
razon_social_receptor = _find_text(
|
||||
root,
|
||||
"cac:DeliveryCustomerParty/cac:Party/cac:PartyLegalEntity/cbc:RegistrationName",
|
||||
ns,
|
||||
) or ""
|
||||
|
||||
# Punto de partida
|
||||
punto_partida = _find_text(root, ".//cac:Delivery/cac:Despatch/cac:DespatchAddress/cac:AddressLine/cbc:Line", ns)
|
||||
|
||||
# Punto de llegada
|
||||
punto_llegada = _find_text(root, ".//cac:Delivery/cac:DeliveryAddress/cac:AddressLine/cbc:Line", ns)
|
||||
|
||||
# Vehículo principal
|
||||
vehiculo_principal = _find_text(root, ".//cac:TransportHandlingUnit/cac:TransportEquipment/cbc:ID", ns)
|
||||
|
||||
# Vehículo secundario
|
||||
vehiculo_secundario = _find_text(root, ".//cac:TransportHandlingUnit/cac:TransportEquipment/cac:AttachedTransportEquipment/cbc:ID", ns)
|
||||
|
||||
# Licencia del conductor
|
||||
licencia_conductor = _find_text(root, ".//cac:ShipmentStage/cac:DriverPerson/cac:IdentityDocumentReference/cbc:ID", ns)
|
||||
|
||||
# Nombre del conductor
|
||||
nombre_conductor = _find_text(root, ".//cac:ShipmentStage/cac:DriverPerson/cbc:FirstName", ns)
|
||||
|
||||
# Documento del conductor
|
||||
documento_conductor = _find_text(root, ".//cac:ShipmentStage/cac:DriverPerson/cbc:ID", ns)
|
||||
|
||||
# Ubigeo de partida
|
||||
ubigeo_partida = _find_text(root, ".//cac:Delivery/cac:Despatch/cac:DespatchAddress/cbc:ID", ns)
|
||||
|
||||
|
||||
# Peso bruto
|
||||
peso_bruto = _find_text(root, ".//cbc:GrossWeightMeasure", ns)
|
||||
|
||||
# Unidad de peso bruto
|
||||
unidad_peso_bruto = root.find(".//cbc:GrossWeightMeasure", ns).attrib.get('unitCode') if root.find(".//cbc:GrossWeightMeasure", ns) is not None else ""
|
||||
|
||||
# Fecha de traslado
|
||||
fecha_traslado = _find_text(root, ".//cac:ShipmentStage/cac:TransitPeriod/cbc:StartDate", ns)
|
||||
|
||||
|
||||
# Las guías de remisión no tienen monto total
|
||||
total_factura = None
|
||||
|
||||
return DatosComprobante(
|
||||
ruc_emisor=ruc_emisor,
|
||||
razon_social_emisor=razon_social_emisor,
|
||||
fecha_emision=fecha_emision,
|
||||
tipo=tipo,
|
||||
serie=serie,
|
||||
numero=numero,
|
||||
ruc_receptor=ruc_receptor,
|
||||
razon_social_receptor=razon_social_receptor,
|
||||
total_factura=total_factura,
|
||||
punto_partida=punto_partida,
|
||||
punto_llegada=punto_llegada,
|
||||
peso_bruto=peso_bruto,
|
||||
unidad_peso_bruto=unidad_peso_bruto,
|
||||
fecha_traslado=fecha_traslado,
|
||||
vehiculo_principal=vehiculo_principal,
|
||||
vehiculo_secundario=vehiculo_secundario,
|
||||
licencia_conductor=licencia_conductor,
|
||||
nombre_conductor=nombre_conductor,
|
||||
documento_conductor=documento_conductor,
|
||||
ubigeo_partida=ubigeo_partida
|
||||
)
|
||||
|
||||
########################################################
|
||||
def extraer_datos_xml(ruta_xml: str) -> Optional[dict]:
|
||||
"""
|
||||
Recibe la ruta de un archivo XML de comprobante electrónico SUNAT.
|
||||
Si el comprobante es una Factura (InvoiceTypeCode == "01"), extrae y retorna
|
||||
sus datos en un diccionario listo para ser convertido a JSON.
|
||||
De lo contrario, retorna None.
|
||||
|
||||
Args:
|
||||
ruta_xml: Ruta al archivo XML del comprobante.
|
||||
|
||||
Returns:
|
||||
Diccionario con los campos extraídos o None si no es una factura.
|
||||
|
||||
Raises:
|
||||
FileNotFoundError: Si el archivo no existe.
|
||||
ET.ParseError: Si el XML no es válido.
|
||||
"""
|
||||
# print(f"Ruta XML: {ruta_xml}")
|
||||
tree = ET.parse(ruta_xml)
|
||||
root = tree.getroot()
|
||||
|
||||
tag = root.tag
|
||||
|
||||
if "Invoice" in tag:
|
||||
tipo = _find_text(root, "inv:InvoiceTypeCode", NAMESPACES) or _find_text(root, "cbc:InvoiceTypeCode", NAMESPACES) or ""
|
||||
if tipo == "01":
|
||||
datos = _parse_invoice(root)
|
||||
return datos.to_dict()
|
||||
if "DespatchAdvice" in tag:
|
||||
tipo = _find_text(root, "desp:DespatchAdviceTypeCode", NAMESPACES) or _find_text(root, "cbc:DespatchAdviceTypeCode", NAMESPACES) or ""
|
||||
if tipo in ["31"]:
|
||||
datos = _parse_despatch_advice(root)
|
||||
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
|
||||
import json
|
||||
|
||||
carpeta_adjuntos = os.path.join(os.path.dirname(__file__), "adjuntos")
|
||||
archivos_xml = sorted(glob.glob(os.path.join(carpeta_adjuntos, "*.xml")))
|
||||
|
||||
if not archivos_xml:
|
||||
print(f"No se encontraron archivos XML en: {carpeta_adjuntos}")
|
||||
else:
|
||||
print(f"Procesando Facturas (01) en {carpeta_adjuntos}...\n")
|
||||
print("=" * 70)
|
||||
for archivo in archivos_xml:
|
||||
try:
|
||||
print(f"Procesando: {archivo}")
|
||||
datos_dict = extraer_datos_xml(archivo)
|
||||
if datos_dict:
|
||||
print("-" * 50)
|
||||
print(json.dumps(datos_dict, indent=2, ensure_ascii=False))
|
||||
print()
|
||||
except Exception as e:
|
||||
print(f"\nError procesando {os.path.basename(archivo)}: {e}\n")
|
||||
print("=" * 70)
|
||||
@@ -0,0 +1,3 @@
|
||||
{
|
||||
"ngrok":"2L3lPgEfu0YX5O0xPeqm4FKgPwP_PwxXGpJn3Uow7JwiS12h"
|
||||
}
|
||||
Reference in new issue
Block a user