EFC/microservice
5
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

coves y edocuments

Browse Source
This commit is contained in:
Kevin Rosales
2025-08-05 07:36:15 -06:00
parent 975fed6f81
commit 5b492b0caa
4 changed files with 84 additions and 504 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
utils/peticiones.py
View File

@@ -129,36 +129,54 @@ def extract_acuse_documento_from_soap(soap_response_text): # Testeado
logger.error(f"Error extrayendo acuseDocumento: {e}")
return None
def extract_pdf_bytes_from_xml(xml_path):
tree = ET.parse(xml_path)
root = tree.getroot()
# Busca el tag <File> (ajusta el namespace si es necesario)
"""
Igual que extract_pdf_bytes_from_xml_content pero recibe una ruta de archivo.
"""
with open(xml_path, 'r', encoding='utf-8') as f:
xml_content = f.read()
return extract_pdf_bytes_from_xml_content(xml_content)
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 not None and file_elem.text:
# Limpia el contenido base64
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
# Buscar CadenaOriginal y SelloDigital en el XML
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
}
return pdf_bytes
else:
raise ValueError("No se encontró el tag <File> con contenido válido.")
raise ValueError("No se encontró el tag <File> con contenido válido. Verifique que el XML contiene el tag <File> con datos base64.")
def decode_acuse_base64_content(base64_content): # Testeado
"""
@@ -846,32 +864,20 @@ async def get_soap_edocument(credenciales, response_service, soap_controller, ed
# Extraer contenido Base64 del acuse
logger.info("Extrayendo documento binario del edocument...")
response = extract_pdf_bytes_from_xml(soap_response.text)
response = extract_pdf_bytes_from_xml_content(soap_response.text)
pdf_bytes = response.get('pdf_bytes')
# cadena_original = response.get('cadena_original')
# sello_digital = response.get('sello_digital')
if not acuse_base64:
logger.error("No se pudo extraer el contenido del acuseDocumento")
raise HTTPException(status_code=500, detail="No se pudo extraer el documento del acuse")
# Decodificar contenido Base64
response_edoc = rest_controller.put_edocument(edocument_id=ide, data={
"numero_edocument": edocument['numero_edocument'],
"pedimento": response_service['pedimento'],
# "cadena_original": cadena_original,
# "sello_digital": sello_digital
})
if not pdf_bytes:
logger.error("No se pudo decodificar el contenido Base64 del acuse")
raise HTTPException(status_code=500, detail="No se pudo decodificar el documento del acuse")
logger.error("No se pudo decodificar el contenido Base64 del documento e-document")
raise HTTPException(status_code=500, detail="No se pudo decodificar el documento del e-document")
# Verificar que es un PDF válido
if not pdf_bytes.startswith(b'%PDF'):
logger.warning("El contenido decodificado no parece ser un PDF válido")
# Continuar de todos modos, podría ser otro tipo de documento
# Generar nombre del archivo
remesas = 1 if response_service['pedimento'].get('remesas', 0) else 0
patente = response_service['pedimento'].get('patente', 'N/A')
@@ -880,7 +886,7 @@ async def get_soap_edocument(credenciales, response_service, soap_controller, ed
tipo_operacion = response_service['pedimento'].get('tipo_operacion', 'N/A')
pedimento = response_service['pedimento'].get('pedimento', 'N/A')
_file_name = f"vu_EDC_{remesas}{no_partidas}{tipo_operacion}_{aduana}_{patente}_{pedimento}_{idx}.pdf"
# Enviar el documento PDF usando binary_content
logger.info(f"Enviando documento PDF: {_file_name} ({len(pdf_bytes)} bytes)")
document_response = await rest_controller.post_document(
@@ -904,7 +910,7 @@ async def get_soap_edocument(credenciales, response_service, soap_controller, ed
logger.error(f"Error inesperado en get_acuse: {e}")
import traceback
logger.error(f"Traceback: {traceback.format_exc()}")
raise HTTPException(status_code=500, detail=f"Error interno al procesar acuse: {str(e)}")
raise HTTPException(status_code=500, detail=f"Error interno al procesar acuse")
async def get_soap_cove(credenciales, response_service, soap_controller, cove, idx):
"""