211 lines
7.3 KiB
Python
211 lines
7.3 KiB
Python
from http.client import HTTPException
|
|
import base64
|
|
import re
|
|
import logging
|
|
import xml.etree.ElementTree as ET
|
|
|
|
from utils.helpers import soap_error
|
|
from .controllers import edocs_rest_controller, edocs_vu_controller
|
|
|
|
logger = logging.getLogger("app.api")
|
|
|
|
|
|
|
|
# --- FUNCIONES AUXILIARES ---
|
|
|
|
def _decode_base64_content(base64_content):
|
|
try:
|
|
cleaned_content = re.sub(r'&#x[0-9a-fA-F]+;', '', base64_content)
|
|
cleaned_content = re.sub(r'&#[0-9]+;', '', cleaned_content)
|
|
cleaned_content = re.sub(r'[\s\n\r\t]', '', cleaned_content)
|
|
cleaned_content = re.sub(r'[^A-Za-z0-9+/=]', '', cleaned_content)
|
|
|
|
missing_padding = len(cleaned_content) % 4
|
|
if missing_padding:
|
|
cleaned_content += '=' * (4 - missing_padding)
|
|
|
|
return base64.b64decode(cleaned_content)
|
|
except Exception as e:
|
|
logger.error(f"Error al decodificar Base64: {e}")
|
|
try:
|
|
return base64.b64decode(cleaned_content, validate=False)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _extract_edoc_data(soap_response_text: str) -> str:
|
|
try:
|
|
xml_start = soap_response_text.find('<?xml')
|
|
if xml_start == -1:
|
|
return None
|
|
|
|
xml_content = soap_response_text[xml_start:]
|
|
boundary_end = xml_content.find('--uuid:')
|
|
if boundary_end != -1:
|
|
xml_content = xml_content[:boundary_end]
|
|
|
|
root = ET.fromstring(xml_content.strip())
|
|
|
|
namespaces = {
|
|
'S': 'http://schemas.xmlsoap.org/soap/envelope/',
|
|
'ns3': 'http://www.ventanillaunica.gob.mx/ws/consulta/edocs/'
|
|
}
|
|
|
|
edoc_elemento = root.find('.//ns3:responseConsultaEdocumento/documentoBase64', namespaces)
|
|
if edoc_elemento is None:
|
|
edoc_elemento = root.find('.//documentoBase64')
|
|
|
|
return edoc_elemento.text.strip() if edoc_elemento is not None and edoc_elemento.text else None
|
|
except ET.ParseError as e:
|
|
logger.error(f"Error parseando la respuesta SOAP para Edoc: {e}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Error general en extracción de datos Edoc: {e}")
|
|
return None
|
|
|
|
|
|
def _get_file_name(**kwargs) -> str:
|
|
pedimento = kwargs.get('pedimento', {})
|
|
pedimento_app = pedimento.get('pedimento_app', 'N/A')
|
|
idEDocument = kwargs['edoc'].get('numero_edocument', 'N/A')
|
|
return f"vu_ED_{pedimento_app}_{idEDocument}.pdf"
|
|
|
|
# --- FUNCIONES DE SERVICIO ---
|
|
|
|
async def obtener_edoc(**kwargs):
|
|
|
|
credencial = kwargs.get('credencial', {})
|
|
usuario = credencial.get('user', '')
|
|
password = credencial.get('password', '')
|
|
doc = kwargs.get('edoc', {})
|
|
numero_documento = kwargs['edoc'].get('numero_edocument', '')
|
|
soap_headers = {
|
|
'Content-Type': 'text/xml; charset=utf-8',
|
|
'SOAPAction': 'http://tempuri.org/IServicioEdocument/GetDocumento'
|
|
}
|
|
soap_xml = edocs_vu_controller.generate_edocument_template(username=usuario, password=password, idEDocument=numero_documento)
|
|
response = await edocs_vu_controller.make_request_async(
|
|
"Ventanilla-HA/ServicioEdocument/ServicioEdocument.svc",
|
|
data=soap_xml,
|
|
headers=soap_headers
|
|
)
|
|
if response is None:
|
|
raise Exception("No se obtuvo respuesta del servicio SOAP.")
|
|
|
|
if response.status_code != 200:
|
|
raise Exception(f"Error en la solicitud SOAP: {response.status_code}")
|
|
|
|
if soap_error(response):
|
|
raise Exception("Respuesta SOAP contiene error de VUCEM.")
|
|
|
|
edoc_base64 = extract_pdf_bytes_from_xml_content(response.text)
|
|
if edoc_base64 is None:
|
|
raise Exception("No se pudo extraer el documento de la respuesta SOAP.")
|
|
|
|
pdf_bytes = edoc_base64['pdf_bytes']
|
|
if not pdf_bytes:
|
|
raise HTTPException(status_code=500, detail="No se pudo decodificar el documento")
|
|
|
|
if not pdf_bytes.startswith(b'%PDF'):
|
|
logger.warning("El contenido decodificado no parece ser un PDF válido")
|
|
|
|
pedimento = kwargs.get('pedimento', {})
|
|
numero_documento = kwargs['edoc'].get('numero_edocument', '')
|
|
_file_name = _get_file_name(**kwargs)
|
|
|
|
organizacion = pedimento.get("organizacion", None)
|
|
pedimento_id = pedimento.get("id", None)
|
|
|
|
try:
|
|
with open(_file_name, "wb") as f:
|
|
f.write(pdf_bytes)
|
|
logger.info(f"PDF guardado localmente en {_file_name}")
|
|
except Exception as e:
|
|
logger.error(f"Error guardando el PDF localmente: {e}")
|
|
|
|
rest_response = await edocs_rest_controller.post_document(
|
|
binary_content=pdf_bytes,
|
|
organizacion=organizacion,
|
|
pedimento=pedimento_id,
|
|
file_name=_file_name,
|
|
document_type=5
|
|
)
|
|
|
|
if rest_response is None:
|
|
raise Exception("No se pudo enviar el documento a la API interna.")
|
|
if rest_response.get("id") is None:
|
|
raise Exception("La respuesta de la API interna no contiene un ID válido.")
|
|
|
|
logger.info("Documento enviado, actualizando status de Edoc")
|
|
|
|
edoc_status_response = await change_edocument_status(
|
|
edoc=doc,
|
|
status=True,
|
|
pedimento=pedimento
|
|
)
|
|
|
|
print(edoc_status_response)
|
|
|
|
return {
|
|
"document_response": rest_response,
|
|
"file_name": _file_name,
|
|
"numero_documento": numero_documento,
|
|
"edoc_update_response": edoc_status_response if edoc_status_response else None
|
|
}
|
|
|
|
|
|
async def change_edocument_status(edoc: dict, status: bool, pedimento: dict):
|
|
data = {
|
|
"id": edoc.get("id"),
|
|
"acuse_descargado": status,
|
|
"numero_edocument": edoc.get("numero_edocument"),
|
|
"pedimento": pedimento.get("id"),
|
|
"organizacion": pedimento.get("organizacion"),
|
|
}
|
|
|
|
response = await edocs_rest_controller.put_edocument(edocument_id=edoc.get("id"), data=data)
|
|
|
|
return response
|
|
|
|
|
|
|
|
|
|
def extract_pdf_bytes_from_xml_content(xml_content: str):
|
|
"""
|
|
Extrae el PDF y metadatos desde un string XML.
|
|
"""
|
|
root = ET.fromstring(xml_content)
|
|
file_elem = root.find('.//File')
|
|
if file_elem is None:
|
|
for elem in root.iter():
|
|
if elem.tag.endswith('File') and elem.text:
|
|
file_elem = elem
|
|
break
|
|
if file_elem is not None and file_elem.text and file_elem.text.strip():
|
|
base64_data = file_elem.text.strip().replace('\n', '').replace('\r', '')
|
|
pdf_bytes = base64.b64decode(base64_data)
|
|
cadena_original = None
|
|
sello_digital = None
|
|
cadena_elem = root.find('.//CadenaOriginal')
|
|
if cadena_elem is None:
|
|
for elem in root.iter():
|
|
if elem.tag.endswith('CadenaOriginal') and elem.text:
|
|
cadena_elem = elem
|
|
break
|
|
if cadena_elem is not None and cadena_elem.text:
|
|
cadena_original = cadena_elem.text.strip()
|
|
sello_elem = root.find('.//SelloDigital')
|
|
if sello_elem is None:
|
|
for elem in root.iter():
|
|
if elem.tag.endswith('SelloDigital') and elem.text:
|
|
sello_elem = elem
|
|
break
|
|
if sello_elem is not None and sello_elem.text:
|
|
sello_digital = sello_elem.text.strip()
|
|
return {
|
|
"pdf_bytes": pdf_bytes,
|
|
"cadena_original": cadena_original,
|
|
"sello_digital": sello_digital
|
|
}
|
|
else:
|
|
raise ValueError("No se encontró el tag <File> con contenido válido. Verifique que el XML contiene el tag <File> con datos base64.") |