Se agregaron los moduloes de api_v2
This commit is contained in:
379
api/api_v2/modules/pedimentos/controllers.py
Normal file
379
api/api_v2/modules/pedimentos/controllers.py
Normal file
@@ -0,0 +1,379 @@
|
||||
from controllers.RESTController import APIRESTController
|
||||
from controllers.SOAPController import VUCEMController
|
||||
from typing import List, Dict, Any
|
||||
import xml.etree.ElementTree as ET
|
||||
from dataclasses import dataclass
|
||||
from typing import List, Dict
|
||||
|
||||
class PedimentoController(APIRESTController):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
|
||||
async def put_pedimento(self, pedimento_id: str, data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""
|
||||
Método para actualizar un pedimento en la API.
|
||||
"""
|
||||
return await self._make_request_async('PUT', f'customs/pedimentos/{pedimento_id}/', data=data)
|
||||
|
||||
async def post_edocument(self, data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""
|
||||
Método para enviar un documento digitalizado a la API.
|
||||
|
||||
Args:
|
||||
data: Diccionario con los datos del documento a enviar
|
||||
"""
|
||||
return await self._make_request_async('POST', 'customs/edocuments/', data=data)
|
||||
|
||||
async def post_cove(self, data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""
|
||||
Método para enviar un número de COVE a la API.
|
||||
|
||||
Args:
|
||||
data: Diccionario con los datos del COVE a enviar
|
||||
"""
|
||||
return await self._make_request_async('POST', 'customs/coves/', data=data)
|
||||
|
||||
async def put_edocument(self, edocument_id: str, data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""
|
||||
Método para actualizar un documento digitalizado en la API.
|
||||
|
||||
Args:
|
||||
edocument_id: UUID del documento a actualizar
|
||||
data: Diccionario con los datos a actualizar
|
||||
"""
|
||||
return await self._make_request_async('PUT', f'customs/edocuments/{edocument_id}/', data=data)
|
||||
|
||||
class PedimentoVUController(VUCEMController):
|
||||
def __init__(self):
|
||||
super().__init__() # Implementación específica para Coves VU
|
||||
|
||||
def generate_remesas_template(self, username: str, password: str, aduana: str, patente: str, numero_operacion: str, pedimento: str) -> str:
|
||||
"""
|
||||
Genera el template SOAP para consultar remesas
|
||||
|
||||
Args:
|
||||
username: Usuario de VUCEM
|
||||
password: Contraseña de VUCEM
|
||||
aduana: Código de aduana
|
||||
patente: Número de patente
|
||||
|
||||
Returns:
|
||||
str: Template SOAP XML completo
|
||||
"""
|
||||
soap_template = f'''
|
||||
<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"
|
||||
xmlns:con="http://www.ventanillaunica.gob.mx/pedimentos/ws/oxml/consultarremesas"
|
||||
xmlns:com="http://www.ventanillaunica.gob.mx/pedimentos/ws/oxml/comunes">
|
||||
<soapenv:Header>
|
||||
<wsse:Security soapenv:mustUnderstand="1"
|
||||
xmlns:wsse="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-secext-1.0.xsd">
|
||||
<wsse:UsernameToken>
|
||||
<wsse:Username>{username}</wsse:Username>
|
||||
<wsse:Password Type="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-username-token-profile-1.0#PasswordText">{password}</wsse:Password>
|
||||
</wsse:UsernameToken>
|
||||
</wsse:Security>
|
||||
</soapenv:Header>
|
||||
<soapenv:Body>
|
||||
<con:consultarRemesasPeticion>
|
||||
<con:numeroOperacion>{numero_operacion}</con:numeroOperacion>
|
||||
<con:peticion>
|
||||
<com:aduana>{aduana}</com:aduana>
|
||||
<com:patente>{patente}</com:patente>
|
||||
<com:pedimento>{pedimento}</com:pedimento>
|
||||
</con:peticion>
|
||||
</con:consultarRemesasPeticion>
|
||||
</soapenv:Body>
|
||||
</soapenv:Envelope>'''
|
||||
return soap_template
|
||||
|
||||
def generate_pedimento_completo_template(self, username: str, password: str, aduana: str, patente: str, pedimento: str) -> str:
|
||||
"""
|
||||
Genera el template SOAP para consultar pedimento completo
|
||||
|
||||
Args:
|
||||
username: Usuario de VUCEM
|
||||
password: Contraseña de VUCEM
|
||||
aduana: Código de aduana
|
||||
patente: Número de patente
|
||||
pedimento: Número de pedimento
|
||||
|
||||
Returns:
|
||||
str: Template SOAP XML completo
|
||||
"""
|
||||
soap_template = f'''<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"
|
||||
xmlns:con="http://www.ventanillaunica.gob.mx/pedimentos/ws/oxml/consultarpedimentocompleto"
|
||||
xmlns:com="http://www.ventanillaunica.gob.mx/pedimentos/ws/oxml/comunes">
|
||||
<soapenv:Header>
|
||||
<wsse:Security soapenv:mustUnderstand="1"
|
||||
xmlns:wsse="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-secext-1.0.xsd">
|
||||
<wsse:UsernameToken>
|
||||
<wsse:Username>{username}</wsse:Username>
|
||||
<wsse:Password
|
||||
Type="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-username-token-profile-1.0#PasswordText">{password}</wsse:Password>
|
||||
</wsse:UsernameToken>
|
||||
</wsse:Security>
|
||||
</soapenv:Header>
|
||||
<soapenv:Body>
|
||||
<con:consultarPedimentoCompletoPeticion>
|
||||
<con:peticion>
|
||||
<com:aduana>{aduana}</com:aduana>
|
||||
<com:patente>{patente}</com:patente>
|
||||
<com:pedimento>{pedimento}</com:pedimento>
|
||||
</con:peticion>
|
||||
</con:consultarPedimentoCompletoPeticion>
|
||||
</soapenv:Body>
|
||||
</soapenv:Envelope>'''
|
||||
|
||||
return soap_template
|
||||
|
||||
# Pedimento Completo
|
||||
@dataclass
|
||||
class PedimentoXMLScraper: # Clase me extrae datos de Pedimento
|
||||
"""
|
||||
Clase para manejar la extracción de datos de un XML.
|
||||
"""
|
||||
|
||||
def _get_numero_operacion(self, root: ET.Element) -> str:
|
||||
"""
|
||||
Método para obtener el número de operación del XML.
|
||||
|
||||
Args:
|
||||
root: Elemento raíz del XML.
|
||||
|
||||
Returns:
|
||||
Número de operación como string.
|
||||
"""
|
||||
numero_operacion = root.find('.//ns2:numeroOperacion', namespaces={'ns2': 'http://www.ventanillaunica.gob.mx/pedimentos/ws/oxml/consultarpedimentocompleto'})
|
||||
return numero_operacion.text if numero_operacion is not None else None
|
||||
|
||||
def _get_pedimento(self, root: ET.Element) -> str:
|
||||
"""
|
||||
Método para obtener el pedimento del XML.
|
||||
|
||||
Args:
|
||||
root: Elemento raíz del XML.
|
||||
|
||||
Returns:
|
||||
Pedimento como string.
|
||||
"""
|
||||
pedimento = root.find('.//ns2:pedimento/ns2:pedimento', namespaces={'ns2': 'http://www.ventanillaunica.gob.mx/pedimentos/ws/oxml/consultarpedimentocompleto'})
|
||||
return pedimento.text if pedimento is not None else None
|
||||
|
||||
def _get_curp_apoderado(self, root: ET.Element) -> str:
|
||||
"""
|
||||
Método para obtener el CURP del apoderado del XML.
|
||||
|
||||
Args:
|
||||
root: Elemento raíz del XML.
|
||||
|
||||
Returns:
|
||||
CURP del apoderado como string.
|
||||
"""
|
||||
curp_apoderado = root.find('.//ns2:curpApoderadomandatario', namespaces={'ns2': 'http://www.ventanillaunica.gob.mx/pedimentos/ws/oxml/consultarpedimentocompleto'})
|
||||
return curp_apoderado.text if curp_apoderado is not None else None
|
||||
|
||||
def _get_agente_aduanal(self, root: ET.Element) -> str:
|
||||
"""
|
||||
Método para obtener el RFC del agente aduanal del XML.
|
||||
|
||||
Args:
|
||||
root: Elemento raíz del XML.
|
||||
|
||||
Returns:
|
||||
RFC del agente aduanal como string.
|
||||
"""
|
||||
agente_aduanal = root.find('.//ns2:rfcAgenteAduanalSocFactura', namespaces={'ns2': 'http://www.ventanillaunica.gob.mx/pedimentos/ws/oxml/consultarpedimentocompleto'})
|
||||
return agente_aduanal.text if agente_aduanal is not None else None
|
||||
|
||||
def _get_partidas(self, root: ET.Element) -> int:
|
||||
"""
|
||||
Método para obtener el número máximo de partidas del XML.
|
||||
|
||||
Args:
|
||||
root: Elemento raíz del XML.
|
||||
|
||||
Returns:
|
||||
Número máximo de partidas como entero.
|
||||
"""
|
||||
partidas_elements = root.findall('.//ns2:partidas', namespaces={'ns2': 'http://www.ventanillaunica.gob.mx/pedimentos/ws/oxml/consultarpedimentocompleto'})
|
||||
partidas_values = []
|
||||
for elem in partidas_elements:
|
||||
try:
|
||||
if elem.text is not None:
|
||||
partidas_values.append(int(elem.text))
|
||||
except ValueError:
|
||||
continue
|
||||
|
||||
return max(partidas_values) if partidas_values else None
|
||||
|
||||
def _get_identificadores_ed(self, root: ET.Element) -> list:
|
||||
"""
|
||||
Método para obtener todos los identificadores con clave 'ED' del XML.
|
||||
|
||||
Args:
|
||||
root: Elemento raíz del XML.
|
||||
|
||||
Returns:
|
||||
Lista de diccionarios con los datos de identificadores ED.
|
||||
"""
|
||||
namespaces = {
|
||||
'ns2': 'http://www.ventanillaunica.gob.mx/pedimentos/ws/oxml/consultarpedimentocompleto',
|
||||
'ns': 'http://www.ventanillaunica.gob.mx/pedimentos/ws/oxml/comunes'
|
||||
}
|
||||
identificadores_ed = []
|
||||
|
||||
# Buscar todos los elementos identificadores
|
||||
identificadores_elements = root.findall('.//ns2:identificadores/ns2:identificadores', namespaces)
|
||||
|
||||
for identificador in identificadores_elements:
|
||||
try:
|
||||
# Extraer la clave del identificador (está dentro de claveIdentificador con namespace)
|
||||
clave_elem = identificador.find('ns:claveIdentificador/ns:clave', namespaces)
|
||||
clave = clave_elem.text if clave_elem is not None else None
|
||||
|
||||
# Solo procesar si la clave es 'ED'
|
||||
if clave == 'ED':
|
||||
# Extraer descripción (con namespace)
|
||||
descripcion_elem = identificador.find('ns:claveIdentificador/ns:descripcion', namespaces)
|
||||
descripcion = descripcion_elem.text if descripcion_elem is not None else None
|
||||
|
||||
# Extraer complemento1 (con namespace)
|
||||
complemento1_elem = identificador.find('ns:complemento1', namespaces)
|
||||
complemento1 = complemento1_elem.text if complemento1_elem is not None else None
|
||||
|
||||
# Agregar a la lista si tenemos los datos básicos
|
||||
if clave and complemento1:
|
||||
identificadores_ed.append({
|
||||
'clave': clave,
|
||||
'descripcion': descripcion,
|
||||
'complemento1': complemento1
|
||||
})
|
||||
|
||||
except Exception as e:
|
||||
# Log del error pero continuar procesando otros identificadores
|
||||
print(f"Error procesando identificador: {e}")
|
||||
continue
|
||||
|
||||
return identificadores_ed
|
||||
|
||||
def _remesas(self, root: ET.Element) -> bool:
|
||||
"""
|
||||
Método para verificar si el pedimento tiene remesas.
|
||||
Busca identificadores con clave 'RC' (REMESAS DE CONSOLIDADO).
|
||||
|
||||
Args:
|
||||
root: Elemento raíz del XML.
|
||||
|
||||
Returns:
|
||||
True si encuentra identificadores con clave 'RC', False en caso contrario.
|
||||
"""
|
||||
namespaces = {
|
||||
'ns2': 'http://www.ventanillaunica.gob.mx/pedimentos/ws/oxml/consultarpedimentocompleto',
|
||||
'ns': 'http://www.ventanillaunica.gob.mx/pedimentos/ws/oxml/comunes'
|
||||
}
|
||||
|
||||
# Buscar todos los elementos identificadores
|
||||
identificadores_elements = root.findall('.//ns2:identificadores/ns2:identificadores', namespaces)
|
||||
|
||||
for identificador in identificadores_elements:
|
||||
try:
|
||||
# Extraer la clave del identificador
|
||||
clave_elem = identificador.find('ns:claveIdentificador/ns:clave', namespaces)
|
||||
clave = clave_elem.text if clave_elem is not None else None
|
||||
|
||||
# Si encontramos una clave 'RC', el pedimento tiene remesas
|
||||
if clave == 'RC':
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
# Log del error pero continuar procesando otros identificadores
|
||||
print(f"Error procesando identificador para remesas: {e}")
|
||||
continue
|
||||
|
||||
print("No se encontraron remesas (sin identificadores RC)")
|
||||
return False
|
||||
|
||||
def _get_tipo_operacion(self, root: ET.Element) -> str:
|
||||
"""
|
||||
Método para obtener el tipo de operación del XML.
|
||||
|
||||
Args:
|
||||
root: Elemento raíz del XML.
|
||||
|
||||
Returns:
|
||||
Tipo de operación como string.
|
||||
"""
|
||||
tipo_operacion = root.find('.//ns2:tipoOperacion/ns2:clave', namespaces={'ns2': 'http://www.ventanillaunica.gob.mx/pedimentos/ws/oxml/consultarpedimentocompleto'})
|
||||
return tipo_operacion.text if tipo_operacion is not None else None
|
||||
|
||||
def _get_cove(self, root: ET.Element) -> str:
|
||||
"""
|
||||
Método para obtener el número de COVE del XML.
|
||||
|
||||
Args:
|
||||
root: Elemento raíz del XML.
|
||||
|
||||
Returns:
|
||||
Número de COVE como string.
|
||||
"""
|
||||
namespaces = {
|
||||
'ns2': 'http://www.ventanillaunica.gob.mx/pedimentos/ws/oxml/consultarpedimentocompleto',
|
||||
'ns': 'http://www.ventanillaunica.gob.mx/pedimentos/ws/oxml/comunes'
|
||||
}
|
||||
facturas = root.findall('.//ns2:facturas', namespaces=namespaces)
|
||||
coves = []
|
||||
for factura in facturas:
|
||||
cove = factura.find('ns2:numero', namespaces)
|
||||
if cove is not None:
|
||||
coves.append(cove.text)
|
||||
else:
|
||||
print("No se encontró <ns2:numero> en la factura.")
|
||||
|
||||
return coves if coves else None
|
||||
|
||||
def extract_data(self, xml_content: str) -> dict:
|
||||
"""
|
||||
Método para extraer datos específicos del XML.
|
||||
|
||||
Args:
|
||||
xml_content: Contenido del XML como string.
|
||||
|
||||
Returns:
|
||||
Diccionario con los datos extraídos.
|
||||
"""
|
||||
try:
|
||||
root = ET.fromstring(xml_content)
|
||||
|
||||
# Extraer datos con manejo de errores individual
|
||||
data = {}
|
||||
|
||||
data['numero_operacion'] = self._get_numero_operacion(root)
|
||||
data['pedimento'] = self._get_pedimento(root)
|
||||
data['curp_apoderado'] = self._get_curp_apoderado(root)
|
||||
data['agente_aduanal'] = self._get_agente_aduanal(root)
|
||||
data['numero_partidas'] = self._get_partidas(root)
|
||||
data['identificadores_ed'] = self._get_identificadores_ed(root)
|
||||
data['remesas'] = self._remesas(root)
|
||||
data['tipo_operacion'] = self._get_tipo_operacion(root)
|
||||
data['coves'] = self._get_cove(root)
|
||||
|
||||
# Verificar que se extrajeron los datos esenciales
|
||||
if not any([data['numero_operacion'], data['pedimento'], data['curp_apoderado'], data['agente_aduanal'], data['coves']]):
|
||||
return {}
|
||||
|
||||
return data
|
||||
|
||||
except ET.ParseError as e:
|
||||
print(f"Error al parsear el XML: {e}")
|
||||
return {}
|
||||
except Exception as e:
|
||||
print(f"Error inesperado al extraer datos del XML: {e}")
|
||||
return {}
|
||||
|
||||
|
||||
|
||||
pedimento_rest_controller = PedimentoController()
|
||||
pedimento_vu_controller = PedimentoVUController()
|
||||
pedimento_xml_scraper = PedimentoXMLScraper()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user