Primera version estable de microservicios
This commit is contained in:
298
controllers/RESTController.py
Normal file
298
controllers/RESTController.py
Normal file
@@ -0,0 +1,298 @@
|
||||
import requests
|
||||
import asyncio
|
||||
import logging
|
||||
logger = logging.getLogger("app.api")
|
||||
from typing import List, Dict, Any
|
||||
import os
|
||||
import httpx
|
||||
from core.config import settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class APIController:
|
||||
"""
|
||||
Controlador para manejar las peticiones a la API.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.base_url = settings.API_URL # URL base de la API
|
||||
|
||||
self.headers = {
|
||||
'Content-Type': 'application/json',
|
||||
'Authorization': f'Token {settings.API_TOKEN}' # Token de autenticación
|
||||
}
|
||||
self.timeout = 5 # Timeout para las peticiones a la API
|
||||
|
||||
def _make_request(self, method, endpoint, data=None):
|
||||
"""
|
||||
Método para hacer peticiones a la API.
|
||||
"""
|
||||
|
||||
url = f"{self.base_url}/{endpoint}"
|
||||
try:
|
||||
|
||||
response = requests.request(method, url, json=data, headers=self.headers, timeout=self.timeout)
|
||||
response.raise_for_status() # Lanza un error si la respuesta no es 200
|
||||
result = response.json()
|
||||
return result # Retorna el JSON de la respuesta
|
||||
except requests.RequestException as e:
|
||||
if hasattr(e, 'response') and e.response is not None:
|
||||
print(f"Status code del error: {e.response.status_code}")
|
||||
print(f"Contenido del error: {e.response.text}")
|
||||
return None
|
||||
|
||||
async def get_pedimento_services(self, pedimento, service_type=3) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Método para obtener la lista de servicios desde la API.
|
||||
"""
|
||||
return await self._make_request_async('GET', f'customs/procesamientopedimentos/?pedimento={pedimento}&estado=1&servicio={service_type}')
|
||||
|
||||
async def get_pedimento(self, pedimento_id: str) -> Dict[str, Any]:
|
||||
"""
|
||||
Método para obtener un pedimento específico desde la API.
|
||||
|
||||
Args:
|
||||
pedimento: UUID del pedimento a consultar
|
||||
"""
|
||||
return self._make_request('GET', f'customs/pedimentos/{pedimento_id}/')
|
||||
|
||||
async def get_vucem_credentials(self, importador) -> Dict[str, Any]:
|
||||
"""
|
||||
Método para obtener las credenciales de VUCEM desde la API.
|
||||
"""
|
||||
return await self._make_request_async('GET', f'vucem/vucem/?usuario={importador}')
|
||||
|
||||
async def post_pedimento_service(self, data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""
|
||||
Método para crear un nuevo servicio de pedimento en la API.
|
||||
|
||||
Args:
|
||||
data: Diccionario con los datos del servicio a crear
|
||||
"""
|
||||
return await self._make_request_async('POST', 'customs/procesamientopedimentos/', data=data)
|
||||
|
||||
async def put_pedimento_service(self, service_id: int, data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""
|
||||
Método para actualizar un servicio de pedimento en la API.
|
||||
"""
|
||||
return await self._make_request_async('PUT', f'customs/procesamientopedimentos/{service_id}/', data=data)
|
||||
|
||||
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_document(self, soap_response=None, organizacion: str = None, pedimento: str = None, file_name: str = None, document_type: int = 2, binary_content: bytes = None) -> Dict[str, Any]:
|
||||
"""
|
||||
Método para enviar documentos (XML, PDF, etc.) a la API.
|
||||
|
||||
Args:
|
||||
soap_response: Respuesta del servicio SOAP (para archivos XML)
|
||||
organizacion: UUID de la organización (requerido)
|
||||
pedimento: UUID del pedimento (requerido)
|
||||
file_name: Nombre del archivo con extensión (requerido)
|
||||
document_type: Tipo de documento
|
||||
binary_content: Contenido binario del archivo (para PDFs, etc.)
|
||||
"""
|
||||
import datetime
|
||||
import tempfile
|
||||
import mimetypes
|
||||
|
||||
if not soap_response and not binary_content:
|
||||
print("Error: Debe proporcionar soap_response o binary_content")
|
||||
return None
|
||||
|
||||
if not file_name:
|
||||
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
file_name = f"documento_{timestamp}.bin"
|
||||
|
||||
try:
|
||||
# Extraer extensión del nombre del archivo
|
||||
file_extension = os.path.splitext(file_name)[1].lstrip('.').lower()
|
||||
if not file_extension:
|
||||
file_extension = 'bin' # Extensión por defecto
|
||||
|
||||
# Determinar Content-Type basado en la extensión
|
||||
content_type_map = {
|
||||
'xml': 'application/xml',
|
||||
'pdf': 'application/pdf',
|
||||
'jpg': 'image/jpeg',
|
||||
'jpeg': 'image/jpeg',
|
||||
'png': 'image/png',
|
||||
'zip': 'application/zip',
|
||||
'doc': 'application/msword',
|
||||
'docx': 'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
|
||||
'xls': 'application/vnd.ms-excel',
|
||||
'xlsx': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
|
||||
}
|
||||
|
||||
content_type = content_type_map.get(file_extension, 'application/octet-stream')
|
||||
|
||||
# Determinar modo de archivo y contenido
|
||||
if binary_content:
|
||||
# Para archivos binarios (PDFs, imágenes, etc.)
|
||||
file_mode = 'wb'
|
||||
temp_suffix = f'.{file_extension}'
|
||||
content = binary_content
|
||||
is_binary = True
|
||||
else:
|
||||
# Para archivos de texto (XML)
|
||||
file_mode = 'w'
|
||||
temp_suffix = f'.{file_extension}'
|
||||
is_binary = False
|
||||
|
||||
# Obtener contenido de la respuesta SOAP
|
||||
if hasattr(soap_response, 'content'):
|
||||
content = soap_response.content.decode('utf-8')
|
||||
elif hasattr(soap_response, 'text'):
|
||||
content = soap_response.text
|
||||
else:
|
||||
content = str(soap_response)
|
||||
|
||||
# Crear archivo temporal con la extensión correcta
|
||||
encoding = None if is_binary else 'utf-8'
|
||||
with tempfile.NamedTemporaryFile(mode=file_mode, suffix=temp_suffix, delete=False, encoding=encoding) as temp_file:
|
||||
temp_file.write(content)
|
||||
temp_file_path = temp_file.name
|
||||
|
||||
# Preparar headers para multipart/form-data (sin Content-Type)
|
||||
headers = {
|
||||
'Authorization': f'Token {settings.API_TOKEN}'
|
||||
}
|
||||
|
||||
# Calcular tamaño del archivo
|
||||
file_size = os.path.getsize(temp_file_path)
|
||||
|
||||
# Preparar datos del documento
|
||||
document_data = {
|
||||
'organizacion': organizacion,
|
||||
'pedimento': pedimento,
|
||||
'extension': file_extension,
|
||||
'document_type': document_type,
|
||||
'size': file_size
|
||||
}
|
||||
|
||||
# Subir archivo
|
||||
url = f"{self.base_url}/record/documents/"
|
||||
|
||||
# Usar httpx AsyncClient para la petición asíncrona
|
||||
import httpx
|
||||
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
||||
with open(temp_file_path, 'rb') as file:
|
||||
files = {
|
||||
'archivo': (file_name, file.read(), content_type)
|
||||
}
|
||||
|
||||
response = await client.post(
|
||||
url,
|
||||
data=document_data, # Datos van como form-data
|
||||
files=files, # Archivo va como multipart
|
||||
headers=headers
|
||||
)
|
||||
|
||||
# Limpiar archivo temporal
|
||||
os.unlink(temp_file_path)
|
||||
|
||||
response.raise_for_status()
|
||||
result = response.json()
|
||||
|
||||
print(f"Documento {file_extension.upper()} enviado exitosamente: {file_name} (tamaño: {file_size} bytes)")
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error al enviar documento: {e}")
|
||||
# Limpiar archivo temporal en caso de error
|
||||
if 'temp_file_path' in locals() and os.path.exists(temp_file_path):
|
||||
os.unlink(temp_file_path)
|
||||
return None
|
||||
|
||||
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 get_edocs(self, pedimento: str) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Método para obtener los documentos digitalizados de un pedimento.
|
||||
|
||||
Args:
|
||||
pedimento: UUID del pedimento a consultar
|
||||
"""
|
||||
return await self._make_request_async('GET', f'customs/edocuments/?pedimento={pedimento}')
|
||||
|
||||
async def get_coves(self, pedimento: str) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Método para obtener los COVES de un pedimento.
|
||||
|
||||
Args:
|
||||
pedimento: UUID del pedimento a consultar
|
||||
"""
|
||||
return await self._make_request_async('GET', f'customs/coves/?pedimento={pedimento}')
|
||||
|
||||
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)
|
||||
|
||||
async def _make_request_async(self, method: str, endpoint: str, data=None):
|
||||
"""
|
||||
Método asíncrono para hacer peticiones a la API usando httpx.
|
||||
"""
|
||||
|
||||
|
||||
url = f"{self.base_url.rstrip('/')}/{endpoint.lstrip('/')}"
|
||||
logger.warning(f"Realizando petición {method} a {url}")
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
||||
logger.info(f"Haciendo petición {method} a {url}")
|
||||
|
||||
if method.upper() == 'GET':
|
||||
response = await client.get(url, headers=self.headers)
|
||||
elif method.upper() == 'POST':
|
||||
response = await client.post(url, json=data, headers=self.headers)
|
||||
elif method.upper() == 'PUT':
|
||||
response = await client.put(url, json=data, headers=self.headers)
|
||||
elif method.upper() == 'DELETE':
|
||||
response = await client.delete(url, headers=self.headers)
|
||||
else:
|
||||
raise ValueError(f"Método HTTP no soportado: {method}")
|
||||
|
||||
response.raise_for_status()
|
||||
logger.info(f"Respuesta exitosa: {response.status_code}")
|
||||
|
||||
result = response.json() if response.content else {}
|
||||
return result
|
||||
|
||||
except httpx.TimeoutException as e:
|
||||
logger.error(f"Timeout en petición a {url}: {e}")
|
||||
return None
|
||||
except httpx.HTTPStatusError as e:
|
||||
logger.error(f"Error HTTP {e.response.status_code} en {url}: {e}")
|
||||
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.error(f"Error inesperado en petición a {url}: {e}")
|
||||
import traceback
|
||||
return None
|
||||
|
||||
|
||||
rest_controller = APIController() # Instancia global del controlador REST
|
||||
271
controllers/SOAPController.py
Normal file
271
controllers/SOAPController.py
Normal file
@@ -0,0 +1,271 @@
|
||||
from core.config import settings
|
||||
from dataclasses import dataclass
|
||||
import requests
|
||||
import httpx
|
||||
import datetime
|
||||
import time
|
||||
|
||||
class SOAPController:
|
||||
"""
|
||||
Controlador para manejar las peticiones SOAP.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.base_url = settings.SOAP_SERVICE_URL
|
||||
self.timeout = settings.TIMEOUT # Timeout por default
|
||||
|
||||
async def make_request(self, endpoint, data=None, headers=None, max_retries=5):
|
||||
intento = 0
|
||||
while intento < settings.MAX_RETRIES:
|
||||
try:
|
||||
with httpx.Client(verify=settings.context, timeout=self.timeout) as client:
|
||||
content = data.encode('utf-8') if data else None
|
||||
response = client.post(
|
||||
f"{self.base_url}/{endpoint}",
|
||||
content=content,
|
||||
headers=headers
|
||||
)
|
||||
response.raise_for_status()
|
||||
return response # ✅ éxito
|
||||
except Exception as e:
|
||||
intento += 1
|
||||
wait_time = 0
|
||||
print(f"[{endpoint}] Error intento {intento}: {e}. Reintentando en {settings.WAIT_TIME}s...")
|
||||
time.sleep(settings.WAIT_TIME)
|
||||
|
||||
print(f"[{endpoint}] Fallo tras {settings.MAX_RETRIES} intentos.")
|
||||
return None
|
||||
|
||||
async def make_request_async(self, endpoint, data=None, headers=None, max_retries=5):
|
||||
"""
|
||||
Método asíncrono para hacer peticiones SOAP sin bloquear el event loop
|
||||
|
||||
Args:
|
||||
endpoint: El endpoint al que se va a hacer la petición
|
||||
data: Los datos a enviar en la petición
|
||||
headers: Los headers HTTP a incluir en la petición
|
||||
max_retries: Número máximo de reintentos en caso de fallo
|
||||
|
||||
Returns:
|
||||
La respuesta de la petición, o None si falla tras los reintentos
|
||||
"""
|
||||
import asyncio
|
||||
intento = 0
|
||||
while intento < settings.MAX_RETRIES:
|
||||
try:
|
||||
async with httpx.AsyncClient(verify=settings.context, timeout=self.timeout) as client:
|
||||
content = data.encode('utf-8') if data else None
|
||||
response = await client.post(
|
||||
f"{self.base_url}/{endpoint}",
|
||||
content=content,
|
||||
headers=headers
|
||||
)
|
||||
response.raise_for_status()
|
||||
return response # ✅ éxito
|
||||
except Exception as e:
|
||||
intento += 1
|
||||
print(f"[{endpoint}] Error intento {intento}: {e}. Reintentando en {settings.WAIT_TIME}s...")
|
||||
if intento < settings.MAX_RETRIES:
|
||||
await asyncio.sleep(settings.WAIT_TIME) # ASYNC SLEEP!
|
||||
|
||||
print(f"[{endpoint}] Fallo tras {settings.MAX_RETRIES} intentos.")
|
||||
return None
|
||||
|
||||
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
|
||||
|
||||
def generate_partidas_template(self, username: str, password: str, aduana: str, patente: str, pedimento: str, numero_operacion: str, partida: str) -> str:
|
||||
"""
|
||||
Genera el template SOAP para consultar partidas de un pedimento
|
||||
|
||||
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/consultarpartida" 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:consultarPartidaPeticion>
|
||||
<con:peticion>
|
||||
<com:aduana>{aduana}</com:aduana>
|
||||
<com:patente>{patente}</com:patente>
|
||||
<com:pedimento>{pedimento}</com:pedimento>
|
||||
<con:numeroOperacion>{numero_operacion}</con:numeroOperacion>
|
||||
<con:numeroPartida>{partida}</con:numeroPartida>
|
||||
</con:peticion>
|
||||
</con:consultarPartidaPeticion>
|
||||
</soapenv:Body>
|
||||
</soapenv:Envelope>
|
||||
'''
|
||||
|
||||
return soap_template
|
||||
|
||||
def generate_acuse_template(self, username: str, password: str, idEDocument: str) -> str:
|
||||
soap_template = f'''
|
||||
<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:oxml="http://www.ventanillaunica.gob.mx/consulta/acuses/oxml">
|
||||
<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>
|
||||
<oxml:consultaAcusesPeticion>
|
||||
<idEdocument>{idEDocument}</idEdocument>
|
||||
</oxml:consultaAcusesPeticion>
|
||||
</soapenv:Body>
|
||||
</soapenv:Envelope>
|
||||
'''
|
||||
return soap_template
|
||||
|
||||
def generate_estado_pedimento_template(self, username: str, password: str, aduana: str, patente: str, pedimento: str, numero_operacion: str) -> str:
|
||||
soap_template = f'''
|
||||
<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"
|
||||
xmlns:con="http://www.ventanillaunica.gob.mx/pedimentos/ws/oxml/consultarestadopedimentos"
|
||||
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:consultarEstadoPedimentosPeticion>
|
||||
<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:consultarEstadoPedimentosPeticion>
|
||||
</soapenv:Body>
|
||||
</soapenv:Envelope>
|
||||
'''
|
||||
return soap_template
|
||||
|
||||
def generate_edocument_template(self, username: str, password: str, idEDocument: str) -> str:
|
||||
"""
|
||||
Genera el template SOAP para consultar un EDocument específico
|
||||
|
||||
Args:
|
||||
username: Usuario de VUCEM
|
||||
password: Contraseña de VUCEM
|
||||
idEDocument: ID del EDocument
|
||||
|
||||
Returns:
|
||||
str: Template SOAP XML completo
|
||||
"""
|
||||
soap_template = f'''
|
||||
<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"
|
||||
xmlns:tem="http://tempuri.org/">
|
||||
<soapenv:Header>
|
||||
<tem:UserName>{username}</tem:UserName>
|
||||
<tem:Password>{password}</tem:Password>
|
||||
</soapenv:Header>
|
||||
<soapenv:Body>
|
||||
<tem:DocumentoIn>
|
||||
<tem:Edocument>{idEDocument}</tem:Edocument>
|
||||
<tem:IsCertificado>1</tem:IsCertificado>
|
||||
</tem:DocumentoIn>
|
||||
</soapenv:Body>
|
||||
</soapenv:Envelope>
|
||||
'''
|
||||
return soap_template
|
||||
|
||||
soap_controller = SOAPController() # Instancia global del controlador SOAP
|
||||
257
controllers/XMLController.py
Normal file
257
controllers/XMLController.py
Normal file
@@ -0,0 +1,257 @@
|
||||
import xml.etree.ElementTree as ET
|
||||
from dataclasses import dataclass
|
||||
|
||||
# Pedimento Completo
|
||||
@dataclass
|
||||
class XMLScraper: # 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 {}
|
||||
|
||||
return extract_xml_data(xml_content)
|
||||
|
||||
|
||||
class XMLControllerRemesas:
|
||||
pass
|
||||
|
||||
class XMLControllerPartidas:
|
||||
pass
|
||||
|
||||
xml_controller = XMLScraper()
|
||||
0
controllers/__init__.py
Normal file
0
controllers/__init__.py
Normal file
Reference in New Issue
Block a user