george/descarga_factura_transporte
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
descarga_factura_transporte/sunat_schema.py
T
georgeandClaude Sonnet 5 15e3b5079c se agrega fallback OWTR y calculo de fecha de vencimiento por condicion de pago 
- hana_manager.get_delivery ahora busca en ODLN y, si no encuentra, en OWTR
- sapb1_manager resuelve StockTransfers -> Orders (via U_IDC_NUMOV) para
  validar precios de flete cuando la guia es un traslado de inventario
- se corrige el orden de parametros tipo/serie/numero (sin cruces) en
  get_delivery, get_delivery_notes y _get_delivery_entries
- se agrega sapb1_manager.obtener_condicion_pago (BusinessPartners +
  PaymentTermsTypes) y pipeline._calcular_fecha_vencimiento para que
  DocDueDate del draft sea fecha de contabilizacion + dias de la condicion
  de pago del proveedor
- se agrega variante de regex para "VALOR REFER" truncado en sunat_schema.py
- se actualiza documentacion (docs/) reflejando estos cambios

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-10 13:21:04 -05:00

722 lines
28 KiB
Python
Raw Permalink Blame History

"""
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 re
import xml.etree.ElementTree as ET
from dataclasses import dataclass, asdict, field
from typing import Optional, List
# ============================================================================
# Patrones regex organizados por tipo de extracción
# ============================================================================
# Serie-Número: busca cualquier patrón T001-123456, V001-789, etc.
SERIE_NUMERO_PATTERNS = [
r'\b([TVE][A-Z0-9]{3})\s*-\s*(\d+)\b',
]
# Guías desde Notes de InvoiceLine: cada entrada es (patron, tipo_documento)
GUIAS_NOTES_PATTERNS = [
# GRE: "GRE REMITENTE: T001-128912", "GRR.T001-129352", "G.R.R:T001-132131"
(r"(?:GRE\s+REMITENTE|GRR|G\.R\.R)\s*[:.]?\s*([TVE][A-Z0-9]{3,4})\s*-\s*(\d+)", "09"),
# GRT: "GRT:V001-194", "GRT.V001-227", "G.R.T: EG03-4643"
(r"\b(?:GRT|G\.R\.T)\s*[:.]?\s*([TVE][A-Z0-9]{3,4})\s*-\s*(\d+)", "31"),
]
# Valor referencial desde la descripción del Item: cada cliente lo redacta distinto,
# por lo que se prueba una lista de patrones en orden hasta encontrar el primero que calce.
# Agregar aquí nuevas variantes a medida que aparezcan en facturas de otros clientes.
VALOR_REFERENCIAL_PATTERNS = [
r"VALOR\s+REFERENCIAL\.?\s*:?\s*(?:S/\.?\s*)?([\d,]+\.\d+)", # "VALOR REFERENCIAL: S/ 2,345.00" / "VALOR REFERENCIAL: 468.86"
r"VALOR\s+REFER\.?\s*:?\s*(?:S/\.?\s*)?([\d,]+\.\d+)", # "VALOR REFER\t:S/ 1431.16"
r"VALOR\s+REF\.?\s*:?\s*(?:S/\.?\s*)?([\d,]+\.\d+)", # "VALOR REF:S/ 1,139.75"
r"VAL\.?\s*REF\.?\s*:?\s*(?:S/\.?\s*)?([\d,]+\.\d+)", # "VAL.REF.:S/700.25"
r"V\.?\s*REF\.?\s*:?\s*(?:S/\.?\s*)?([\d,]+\.\d+)", # "V.REF: S/ 2940.80 TN: 36.85"
r"\bV\.\s*R\.?\s*:?\s*(?:S/\.?\s*)?([\d,]+\.\d+)", # "V.R. S/964.51" / "V.R.: 1045.20"
r"\bVR\s*:?\s*(?:S/\.?\s*)?([\d,]+\.\d+)", # "VR:S/300.10"
]
# ============================================================================
# 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",
"ar": "urn:oasis:names:specification:ubl:schema:xsd:ApplicationResponse-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
total_linea: float = 0.0
notes_text: str = ""
guias_remision: List[dict] = field(default_factory=list)
guia_transportista: Optional[dict] = None
valor_referencial: Optional[float] = None
@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.serie}-{ref.numero} (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 _inferir_tipo_desde_serie(serie: str) -> str:
"""
Infiere el tipo de documento SUNAT desde el prefijo de la serie.
- Serie que empieza con 'T' → tipo 09 (Guía de Remisión Remitente)
- Serie que empieza con 'V' o 'E' → tipo 31 (Guía de Remisión Transportista)
- Otros → tipo 09 por defecto
"""
if serie and serie[0] == "T":
return "09"
if serie and serie[0] in ("V", "E"):
return "31"
return "09"
def _es_referencia_valida(serie: str, numero: str) -> bool:
"""Valida que una referencia tenga serie no vacía y número > 0."""
return bool(serie) and numero not in ("", "0")
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]:
"""Public alias for _split_serie_numero."""
return _split_serie_numero(comprobante_id)
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')
'T001 - 00130229' -> ('T001', '130229')
'0002 - 000608' -> ('0002', '608')
"""
if "-" in comprobante_id:
partes = comprobante_id.split("-", 1)
serie = partes[0].strip()
# Remueve espacios y ceros a la izquierda, si queda vacío (ej: '000') se mantiene '0'
numero = partes[1].strip().lstrip("0") or "0"
return serie, numero
return comprobante_id.strip(), ""
# ============================================================================
def _extraer_serie_numero_desde_texto(texto: str) -> tuple[str, str] | None:
texto = re.sub(r'[–—−]', '-', texto)
for patron in SERIE_NUMERO_PATTERNS:
m = re.search(patron, texto)
if m:
numero = m.group(2).lstrip("0") or "0"
return m.group(1).upper(), numero
return None
def _numeros_extra_guia(texto: str, pos: int) -> list[str]:
"""
Desde una posición devuelve números adicionales que comparten la misma serie
de una guía ya capturada, separados por `,`, `/`, `\\` o `;`.
Ejemplo: ``"T001-133023,133024"`` con ``pos`` tras ``133023`` → ``["133024"]``
"""
numeros = []
resto = texto[pos:]
while True:
m = re.match(r'\s*[,;\\/]\s*(\d+)', resto)
if not m:
break
numeros.append(m.group(1).lstrip("0") or "0")
resto = resto[m.end():]
return numeros
def extraer_guias_desde_notes(notes_text: str) -> list[dict]:
"""
Extrae guías de remisión (GRE/GRT) desde el texto de Notes de una InvoiceLine.
Retorna lista de dicts con ``tipo``, ``serie``, ``numero``.
"""
resultados = []
for patron, tipo in GUIAS_NOTES_PATTERNS:
for m in re.finditer(patron, notes_text, re.IGNORECASE):
serie = m.group(1).upper()
numero = m.group(2).lstrip("0") or "0"
resultados.append({"tipo": tipo, "serie": serie, "numero": numero})
for extra in _numeros_extra_guia(notes_text, m.end()):
resultados.append({"tipo": tipo, "serie": serie, "numero": extra})
return resultados
def extraer_valor_referencial(texto: str) -> Optional[float]:
"""
Extrae el valor referencial (importe) desde el texto de descripción de un Item.
Prueba cada patrón de ``VALOR_REFERENCIAL_PATTERNS`` en orden y devuelve el
primer importe encontrado.
Ejemplos: ``"...VALOR REF:S/ 1,139.75"`` -> ``1139.75``
``"...V.R. S/964.51"`` -> ``964.51``
"""
for patron in VALOR_REFERENCIAL_PATTERNS:
m = re.search(patron, texto, re.IGNORECASE)
if m:
return float(m.group(1).replace(",", ""))
return None
def obtener_documentos_referencia_detalle_factura(texto: str) -> list[str]:
texto = re.sub(r'[–—−]', '-', texto)
resultados = []
for patron in SERIE_NUMERO_PATTERNS:
for m in re.finditer(patron, texto, flags=re.IGNORECASE):
serie = m.group(1).upper()
numero = m.group(2)
resultados.append(f"{serie}-{numero}")
for extra in _numeros_extra_guia(texto, m.end()):
resultados.append(f"{serie}-{extra}")
return list(dict.fromkeys(resultados))
# ============================================================================
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:
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)
if _es_referencia_valida(ref_serie, ref_numero):
tipo_ref = ref_tipo if ref_tipo else _inferir_tipo_desde_serie(ref_serie)
documentos_referencia.append(DocumentoReferencia(tipo=tipo_ref, 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:
serie_parsed, numero_parsed = _split_serie_numero(guia)
if _es_referencia_valida(serie_parsed, numero_parsed):
documentos_referencia.append(DocumentoReferencia(tipo=_inferir_tipo_desde_serie(serie_parsed), serie=serie_parsed, numero=numero_parsed))
# 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":
# Notas a nivel de documento (cbc:Note en la raíz): algunos clientes ponen
# el valor referencial aquí en vez de en la descripción del Item, ej:
# "V.REF: S/ 2940.80 TN: 36.85"
root_notes = root.findall("cbc:Note", ns)
root_notes_text = " | ".join(n.text.strip() for n in root_notes if n.text) if root_notes else ""
# 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 ""
notes = line_node.findall("cbc:Note", ns)
notes_text = " | ".join(n.text.strip() for n in notes if n.text) if notes else ""
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
total_linea_str = _find_text(line_node, "cbc:LineExtensionAmount", ns)
total_linea = float(total_linea_str) if total_linea_str else 0.0
guias = extraer_guias_desde_notes(notes_text)
guias_remision = [g for g in guias if g["tipo"] == "09"]
gt = [g for g in guias if g["tipo"] == "31"]
valor_referencial = (
extraer_valor_referencial(descripcion)
or extraer_valor_referencial(notes_text)
or extraer_valor_referencial(root_notes_text)
)
lineas.append(
DetalleLinea(
descripcion=descripcion,
cantidad=cantidad,
precio_unitario=precio_unitario,
total_linea=total_linea,
notes_text=notes_text,
guias_remision=guias_remision,
guia_transportista=gt[0] if gt else None,
valor_referencial=valor_referencial,
)
)
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
# 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,
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,
documentos_referencia=documentos_referencia,
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,
lineas=lineas
)
# ============================================================================
def _parse_application_response(root: ET.Element) -> dict:
"""Extrae datos de un ApplicationResponse (CDR de SUNAT)."""
ns = NAMESPACES
ruc_emisor = ""
tipo = ""
serie = ""
numero = ""
# RUC del emisor original desde ReceiverParty
ruc_emisor = (
_find_text(root, "cac:ReceiverParty/cac:PartyIdentification/cbc:ID", ns)
or ""
)
# Tipo, serie y número desde DocumentResponse/DocumentReference
for doc_resp in root.findall(".//cac:DocumentResponse", ns):
# DocumentTypeCode del documento original
if not tipo:
tipo = (
_find_text(doc_resp, "cac:DocumentReference/cbc:DocumentTypeCode", ns)
or ""
)
# ID del documento original (ej: FF01-3308, F002-20851)
ref_id = (
_find_text(doc_resp, "cac:DocumentReference/cbc:ID", ns)
or ""
)
if ref_id:
serie, numero = _split_serie_numero(ref_id)
# Si no hay tipo, intentar inferir desde el texto de la descripción
if not tipo:
desc = _find_text(doc_resp, "cac:Response/cbc:Description", ns) or ""
m = re.search(r"\b(?:FACTURA|Factura)\b", desc)
if m:
tipo = "01"
# Si no hay RUC, intentar extraer desde la descripción
if not ruc_emisor:
desc = _find_text(doc_resp, "cac:Response/cbc:Description", ns) or ""
m = re.search(r"\b(\d{11})-\d{2}-", desc)
if m:
ruc_emisor = m.group(1)
return {
"ruc_emisor": ruc_emisor,
"tipo": tipo,
"serie": serie,
"numero": numero,
"es_cdr": True,
"tipo_descripcion": "CDR",
}
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 "ApplicationResponse" in tag:
return _parse_application_response(root)
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 ("09", "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)
# ============================================================================
Reference in new issue View Git Blame Copy Permalink