GitHub
6
.gitignore
vendored
6
.gitignore
vendored
@@ -105,4 +105,8 @@ venv.bak/
|
||||
# mypy
|
||||
.mypy_cache/
|
||||
|
||||
.idea/*
|
||||
.idea/*
|
||||
|
||||
# certs
|
||||
*.cer
|
||||
*.key
|
||||
@@ -9,7 +9,7 @@ from typing import Dict, Any, List, Optional
|
||||
from contextlib import asynccontextmanager
|
||||
from controllers.RESTController import rest_controller
|
||||
from controllers.SOAPController import soap_controller
|
||||
from utils.peticiones import get_soap_acuseCOVE, get_soap_pedimento_completo, get_soap_remesas, get_soap_partidas, get_soap_acuse, get_soap_edocument
|
||||
from utils.peticiones import get_soap_acuseCOVE, get_soap_cove, get_soap_pedimento_completo, get_soap_remesas, get_soap_partidas, get_soap_acuse, get_soap_edocument
|
||||
from fastapi.responses import JSONResponse
|
||||
from utils.servicios import (
|
||||
_validate_request_data,
|
||||
@@ -1175,10 +1175,120 @@ async def get_cove(request: ServiceRemesaSchema):
|
||||
raise HTTPException(status_code=400, detail="ID de contribuyente no encontrado")
|
||||
|
||||
credentials = await _get_vucem_credentials(contribuyente_id, operation_name)
|
||||
|
||||
# Obtener COVES
|
||||
logger.info("Obteniendo COVES...")
|
||||
try:
|
||||
coves = await rest_controller.get_coves(service_data['pedimento']['id'])
|
||||
|
||||
if not coves:
|
||||
logger.warning("No se encontraron COVES para el pedimento")
|
||||
await _update_service_status(service_data['id'], ESTADO_ERROR, service_data, operation_name)
|
||||
raise HTTPException(status_code=404, detail="No se encontraron COVES para el pedimento")
|
||||
|
||||
logger.info(f"Se encontraron {len(coves)} COVES")
|
||||
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Error al obtener COVES: {e}")
|
||||
await _update_service_status(service_data['id'], ESTADO_ERROR, service_data, operation_name)
|
||||
raise HTTPException(status_code=500, detail="Error al obtener COVES")
|
||||
|
||||
# Procesar acuses de documentos digitalizados
|
||||
documentos_procesados = []
|
||||
documentos_exitosos = 0
|
||||
|
||||
logger.info(f"Procesando COVE para {len(coves)} documentos...")
|
||||
|
||||
for idx, cove in enumerate(coves):
|
||||
documento_info = {
|
||||
#"clave": cove.get('clave', 'N/A'),
|
||||
#"descripcion": cove.get('descripcion', 'N/A'),
|
||||
"numero_cove": cove.get('numero_cove', 'N/A'),
|
||||
"procesado": False,
|
||||
"error": None
|
||||
}
|
||||
|
||||
# Verificar que el documento tenga número de cove
|
||||
if not cove.get('numero_cove'):
|
||||
logger.warning(f"Documento {idx + 1} no tiene numero_cove, saltando...")
|
||||
documento_info["error"] = "Sin número de cove"
|
||||
documentos_procesados.append(documento_info)
|
||||
continue
|
||||
|
||||
try:
|
||||
logger.info(f"Procesando cove para documento {idx + 1}: {cove['numero_cove']}")
|
||||
|
||||
soap_response = await get_soap_cove(
|
||||
credenciales=credentials,
|
||||
response_service=service_data,
|
||||
soap_controller=soap_controller,
|
||||
cove=cove,
|
||||
idx=idx + 1
|
||||
)
|
||||
|
||||
if soap_response:
|
||||
documento_info["procesado"] = True
|
||||
documento_info["documento"] = soap_response.get('documento', {})
|
||||
documentos_exitosos += 1
|
||||
logger.info(f"cove del documento {idx + 1} procesado exitosamente")
|
||||
else:
|
||||
documento_info["error"] = "Error en petición SOAP"
|
||||
logger.warning(f"No se pudo procesar el cove del documento {idx + 1}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error al procesar cove del documento {idx + 1}: {e}")
|
||||
documento_info["error"] = str(e)
|
||||
# Continuar con los siguientes documentos
|
||||
|
||||
documentos_procesados.append(documento_info)
|
||||
|
||||
# Verificar si se procesó al menos un documento
|
||||
if documentos_exitosos == 0:
|
||||
logger.error("No se pudo procesar ningún cove de documento digitalizado")
|
||||
await _update_service_status(service_data['id'], ESTADO_ERROR, service_data, operation_name)
|
||||
raise HTTPException(status_code=500, detail="No se pudo procesar ningún acuse cove de documento digitalizado")
|
||||
|
||||
# Finalizar servicio exitosamente
|
||||
await _update_service_status(service_data['id'], ESTADO_FINALIZADO, service_data, operation_name)
|
||||
|
||||
# Crear respuesta estandarizada
|
||||
response_data = await _create_response(
|
||||
service_data=service_data,
|
||||
additional_data={
|
||||
"covesDocs": documentos_procesados,
|
||||
"total_documentos": len(coves),
|
||||
"documentos_exitosos": documentos_exitosos,
|
||||
"documentos_fallidos": len(coves) - documentos_exitosos
|
||||
},
|
||||
success_message=f"Se procesaron {documentos_exitosos}/{len(coves)} cove de documentos exitosamente"
|
||||
)
|
||||
|
||||
# Agregar advertencias si hubo documentos fallidos
|
||||
if documentos_exitosos < len(coves):
|
||||
response_data["warnings"] = [
|
||||
f"Se procesaron solo {documentos_exitosos} de {len(coves)} coves"
|
||||
]
|
||||
|
||||
logger.info(f"Procesamiento de acuses cove completado - Exitosos: {documentos_exitosos}/{len(coves)}")
|
||||
return JSONResponse(content=response_data, status_code=200)
|
||||
|
||||
except HTTPException:
|
||||
# Re-lanzar HTTPExceptions sin modificar
|
||||
raise
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error inesperado en {operation_name}: {e}")
|
||||
logger.error(f"Traceback: {traceback.format_exc()}")
|
||||
|
||||
# Actualizar estado a error si tenemos service_data
|
||||
if service_data:
|
||||
try:
|
||||
await _update_service_status(service_data['id'], ESTADO_ERROR, service_data, operation_name)
|
||||
except Exception as update_error:
|
||||
logger.error(f"Error al actualizar estado del servicio tras fallo: {update_error}")
|
||||
|
||||
raise HTTPException(status_code=500, detail=f"Error interno en {operation_name}: {str(e)}")
|
||||
|
||||
@router.post("/services/acuseCove") # Sin Testear
|
||||
async def get_Acusecove(request: ServiceRemesaSchema):
|
||||
|
||||
@@ -268,4 +268,48 @@ class SOAPController:
|
||||
'''
|
||||
return soap_template
|
||||
|
||||
def generate_cove_template(self, username: str, password: str, certificado: str, firma: str, cove: str) -> str:
|
||||
"""
|
||||
Genera el template SOAP para consultar un COVE específico
|
||||
|
||||
Args:
|
||||
username: Usuario de VUCEM
|
||||
password: Contraseña de VUCEM
|
||||
certificado: certificado base 64
|
||||
firma: firma a base de cadena original base 64
|
||||
cove: COVE
|
||||
|
||||
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/ConsultarEdocument/"
|
||||
xmlns:oxml="http://www.ventanillaunica.gob.mx/cove/ws/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>
|
||||
<con:ConsultarEdocumentRequest>
|
||||
<con:request>
|
||||
<con:firmaElectronica>
|
||||
<oxml:certificado>{certificado}</oxml:certificado>
|
||||
<oxml:cadenaOriginal>|{username}|{cove}|</oxml:cadenaOriginal>
|
||||
<oxml:firma>{firma}</oxml:firma>
|
||||
</con:firmaElectronica>
|
||||
<con:criterioBusqueda>
|
||||
<con:eDocument>{cove}</con:eDocument>
|
||||
</con:criterioBusqueda>
|
||||
</con:request>
|
||||
</con:ConsultarEdocumentRequest>
|
||||
</soapenv:Body>
|
||||
</soapenv:Envelope>
|
||||
'''
|
||||
return soap_template
|
||||
|
||||
soap_controller = SOAPController() # Instancia global del controlador SOAP
|
||||
@@ -33,6 +33,11 @@ class Settings(BaseSettings):
|
||||
SECRET_KEY: str = "your-secret-key-here"
|
||||
ALGORITHM: str = "HS256"
|
||||
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
|
||||
|
||||
# cer key y contrasena uso temporal
|
||||
KEY_PASSWORD: str = ""
|
||||
CERT_PATH: str = ""
|
||||
KEY_PATH: str = ""
|
||||
|
||||
model_config = {"env_file": ".env"}
|
||||
|
||||
|
||||
@@ -19,3 +19,5 @@ typing-inspection==0.4.1
|
||||
typing_extensions==4.14.1
|
||||
urllib3==2.5.0
|
||||
uvicorn==0.35.0
|
||||
python-dotenv
|
||||
cryptography
|
||||
|
||||
@@ -6,6 +6,16 @@ from typing import Dict, Any
|
||||
import xml.etree.ElementTree as ET
|
||||
import base64
|
||||
import re
|
||||
import os
|
||||
from cryptography.hazmat.primitives import serialization, hashes
|
||||
from cryptography.hazmat.primitives.asymmetric import padding
|
||||
from cryptography.hazmat.primitives.serialization import load_der_private_key
|
||||
from dotenv import load_dotenv
|
||||
from pathlib import Path
|
||||
from core.config import settings
|
||||
|
||||
# Cargar variables de entorno
|
||||
load_dotenv()
|
||||
|
||||
from schemas.serviceSchema import ServiceBaseSchema
|
||||
|
||||
@@ -14,6 +24,26 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
from controllers.XMLController import xml_controller
|
||||
|
||||
def load_cert_base64(cert_path: str) -> str:
|
||||
with open(cert_path, 'rb') as cert_file:
|
||||
cert_data = cert_file.read()
|
||||
return base64.b64encode(cert_data).decode()
|
||||
|
||||
def sign_chain_original(key_path: str, password: str, cadena_original: str) -> str:
|
||||
with open(key_path, 'rb') as key_file:
|
||||
private_key = load_der_private_key(
|
||||
key_file.read(),
|
||||
password=password.encode() if password else None
|
||||
)
|
||||
|
||||
signature = private_key.sign(
|
||||
cadena_original.encode(),
|
||||
padding.PKCS1v15(),
|
||||
hashes.SHA256()
|
||||
)
|
||||
|
||||
return base64.b64encode(signature).decode()
|
||||
|
||||
|
||||
def validate_pedimento_data(response_service: Dict[str, Any], credenciales: Dict[str, Any]) -> tuple: # Testeado
|
||||
"""
|
||||
@@ -881,4 +911,94 @@ async def get_soap_edocument(credenciales, response_service, soap_controller, ed
|
||||
logger.error(f"Traceback: {traceback.format_exc()}")
|
||||
raise HTTPException(status_code=500, detail=f"Error interno al procesar acuse: {str(e)}")
|
||||
|
||||
async def get_soap_cove(credenciales, response_service, soap_controller, cove, idx):
|
||||
"""
|
||||
Procesa la petición SOAP para obtener el COVE de un pedimento y guarda el documento.
|
||||
|
||||
Args:
|
||||
credenciales: Diccionario con credenciales VUCEM (usuario, password)
|
||||
response_service: Respuesta del servicio con datos del pedimento
|
||||
soap_controller: Instancia del controlador SOAP
|
||||
|
||||
Returns:
|
||||
dict: Respuesta con el servicio, respuesta SOAP y documento guardado
|
||||
|
||||
Raises:
|
||||
HTTPException: Si hay errores en la petición SOAP o al guardar el documento
|
||||
"""
|
||||
try:
|
||||
# Extraer credenciales
|
||||
username, password, aduana, patente, pedimento, numero_operacion = validate_pedimento_data(response_service, credenciales)
|
||||
|
||||
# Cadena original que vas a firmar
|
||||
cadena_original = f"|{username}|{cove['numero_cove']}|"
|
||||
|
||||
# Obtener certificado base64 y firma
|
||||
certificado = load_cert_base64(settings.CERT_PATH)
|
||||
firma = sign_chain_original(settings.KEY_PATH, settings.KEY_PASSWORD, cadena_original)
|
||||
|
||||
logger.info(f"Datos para SOAP - Usuario: {username}, Aduana: {aduana}, Patente: {patente}, Pedimento: {pedimento}, Numero Operacion: {numero_operacion}")
|
||||
|
||||
# Generar template SOAP
|
||||
soap_xml = soap_controller.generate_cove_template(
|
||||
username=username,
|
||||
password=password,
|
||||
certificado=certificado,
|
||||
firma=firma,
|
||||
cove=cove['numero_cove']
|
||||
)
|
||||
|
||||
### >>> AQUÍ SE AÑADE EL LOGGER.DEBUG <<< ###
|
||||
logger.debug(f"XML SOAP generado: {soap_xml}") # 👈 Registra el XML completo
|
||||
|
||||
# Realizar petición SOAP
|
||||
logger.info("Realizando petición SOAP...")
|
||||
|
||||
# Headers específicos para este servicio SOAP
|
||||
soap_headers = {
|
||||
'Content-Type': 'text/xml; charset=utf-8',
|
||||
'SOAPAction': '',
|
||||
#'Accept-Encoding': 'gzip,deflate',
|
||||
}
|
||||
|
||||
soap_response = await soap_controller.make_request_async(
|
||||
"ventanilla/ConsultarEdocumentService?wsdl",
|
||||
data=soap_xml,
|
||||
headers=soap_headers
|
||||
)
|
||||
|
||||
if (soap_response) and (not soap_error(soap_response)):
|
||||
logger.info(f"Petición SOAP exitosa - Status: {soap_response.status_code}")
|
||||
|
||||
remesas = 1 if response_service['pedimento'].get('remesas', 0) else 0
|
||||
patente = response_service['pedimento'].get('patente', 'N/A')
|
||||
aduana = response_service['pedimento'].get('aduana', 'N/A')
|
||||
no_partidas = response_service['pedimento'].get('numero_partidas', 0)
|
||||
tipo_operacion = response_service['pedimento'].get('tipo_operacion', 'N/A')
|
||||
pedimento = response_service['pedimento'].get('pedimento', 'N/A')
|
||||
|
||||
_file_name = f"vu_COVE_{remesas}{no_partidas}{tipo_operacion}_{aduana}_{patente}_{pedimento}_{idx}.xml"
|
||||
|
||||
document_response = await rest_controller.post_document(
|
||||
soap_response=soap_response,
|
||||
organizacion=response_service['organizacion'],
|
||||
pedimento=response_service['pedimento']['id'],
|
||||
file_name=_file_name,
|
||||
document_type=8
|
||||
)
|
||||
|
||||
return {
|
||||
"servicio": response_service,
|
||||
"documento": document_response
|
||||
}
|
||||
else:
|
||||
logger.error("Error en petición SOAP")
|
||||
raise HTTPException(status_code=500, detail="Error en la petición SOAP al servicio VUCEM")
|
||||
except HTTPException:
|
||||
# Re-lanzar HTTPExceptions sin modificar
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Error inesperado en get_acuse cove: {e}")
|
||||
import traceback
|
||||
logger.error(f"Traceback: {traceback.format_exc()}")
|
||||
raise HTTPException(status_code=500, detail=f"Error interno al procesar acuse cove: {str(e)}")
|
||||
|
||||
Reference in New Issue
Block a user