descarga_factura_transporte/sunat_schema.py

"""
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

# ============================================================================
# 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
    total_linea: float = 0.0


@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 _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

    # 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)
            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
    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

            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,
                )
            )

    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 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)
# ============================================================================