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

fix: Se agrega validacion para no intentar crear de nuevo el pedimento en caso de ya existir. tambien se agrega funcion de obtener del xml de pedimento completo el dato de la aduana completo.

Browse Source
This commit is contained in:
Luis Andrade
2026-01-26 15:52:11 -07:00
parent 4ccb5fd718
commit 04d19118be
3 changed files with 446 additions and 256 deletions
Download Patch File Download Diff File Expand all files Collapse all files

286
api/customs/views.py
View File

@@ -513,7 +513,8 @@ class ViewSetPedimento(LoggingMixin, viewsets.ModelViewSet, OrganizacionFiltrada
"processed_files": 3, "processed_files": 3,
"summary": "Procesados 3 archivo(s): 5 pedimento(s) creado(s), 15 documento(s) asociado(s)", "summary": "Procesados 3 archivo(s): 5 pedimento(s) creado(s), 15 documento(s) asociado(s)",
"failed_files": [], "failed_files": [],
"errors": [] "errors": [],
"already_existing": [] # Nuevo campo para pedimentos que ya existían
} }
""" """
print(request.data) print(request.data)
@@ -547,6 +548,7 @@ class ViewSetPedimento(LoggingMixin, viewsets.ModelViewSet, OrganizacionFiltrada
nomenclatura_pattern_sin_anio = re.compile(r'^(\d{2,3})-(\d{4})-(\d{7})$') nomenclatura_pattern_sin_anio = re.compile(r'^(\d{2,3})-(\d{4})-(\d{7})$')
created_pedimentos = [] created_pedimentos = []
already_existing_pedimentos = [] # Para trackear pedimentos que ya existen
failed_files = [] failed_files = []
errors = [] errors = []
documents_created = 0 documents_created = 0
@@ -599,140 +601,41 @@ class ViewSetPedimento(LoggingMixin, viewsets.ModelViewSet, OrganizacionFiltrada
archivo_name = archivo.name.lower() archivo_name = archivo.name.lower()
print(f"Procesando archivo {idx + 1}/{len(archivos)}: {archivo_name}") print(f"Procesando archivo {idx + 1}/{len(archivos)}: {archivo_name}")
# Crear subdirectorio para cada archivo usando el nombre del archivo sin extensión # Extraer nombre base sin extensión para validación
archivo_name_sin_extension = os.path.splitext(archivo.name)[0] archivo_name_sin_extension = os.path.splitext(archivo.name)[0]
sub_dir = os.path.join(temp_dir, archivo_name_sin_extension)
os.makedirs(sub_dir, exist_ok=True)
print(f"Subdirectorio creado: {sub_dir}")
if archivo_name.endswith('.zip'): # Validar nomenclatura del nombre del archivo/folder
# Manejar archivo ZIP match = nomenclatura_pattern.match(archivo_name_sin_extension)
print("Es un archivo ZIP") match_sin_anio = nomenclatura_pattern_sin_anio.match(archivo_name_sin_extension)
try:
with zipfile.ZipFile(archivo, 'r') as zip_ref:
zip_ref.extractall(sub_dir)
print("Archivo ZIP extraído exitosamente")
except zipfile.BadZipFile as e:
return Response(
{"error": f"Archivo ZIP corrupto o inválido: {archivo.name} - {str(e)}"},
status=status.HTTP_400_BAD_REQUEST
)
except Exception as e:
return Response(
{"error": f"Error al extraer ZIP {archivo.name}: {str(e)}"},
status=status.HTTP_400_BAD_REQUEST
)
elif archivo_name.endswith('.rar'):
# Manejar archivo RAR: guardar el archivo en disco y usar helper con fallbacks
# Guardar el archivo subido en un path temporal dentro del sub_dir
archivo_temp_path = os.path.join(sub_dir, archivo.name)
with open(archivo_temp_path, 'wb') as f:
for chunk in archivo.chunks():
f.write(chunk)
try:
extract_rar_to_dir(archivo_temp_path, sub_dir)
print(f"Archivo RAR {archivo.name} extraído en {sub_dir}")
except Exception as e:
error_msg = str(e)
help_msg = "Instale 'unrar' o 'p7zip' (7z) y asegúrese de que estén en PATH, o instale y configure 'rarfile' con un backend."
return Response(
{"error": f"Error al extraer archivo RAR {archivo.name}: {error_msg}. {help_msg}"},
status=status.HTTP_400_BAD_REQUEST
)
# if not RAR_SUPPORT:
# return Response(
# {"error": "Soporte para archivos RAR no disponible. Instalar rarfile: pip install rarfile"},
# status=status.HTTP_400_BAD_REQUEST
# )
# try:
# with rarfile.RarFile(archivo, 'r') as rar_ref:
# rar_ref.extractall(sub_dir)
# print(f"Archivo RAR {archivo.name} extraído en sub_dir")
# except rarfile.Error as e:
# return Response(
# {"error": f"Error al extraer archivo RAR {archivo.name}: {str(e)}"},
# status=status.HTTP_400_BAD_REQUEST
# )
else:
# Asumir que es un archivo individual
# Crear el archivo en el subdirectorio
archivo_path = os.path.join(sub_dir, archivo.name)
with open(archivo_path, 'wb') as f:
for chunk in archivo.chunks():
f.write(chunk)
print(f"Archivo individual {archivo.name} guardado en sub_dir:", archivo_path)
# Recorrer todos los archivos extraídos o el directorio
print("Iniciando recorrido de archivos...")
for root, dirs, files in os.walk(temp_dir):
print(f"Revisando directorio: {root}")
print(f"Archivos encontrados: {files}")
for file_name in files:
print(f"Procesando archivo: {file_name}")
file_path = os.path.join(root, file_name)
# Obtener la ruta relativa para determinar la estructura de carpetas
relative_path = os.path.relpath(file_path, temp_dir)
print(f"Ruta relativa: {relative_path}")
# Determinar si el archivo está en una carpeta que sigue la nomenclatura
folder_name = None
if os.path.dirname(relative_path):
# El archivo está dentro de una carpeta
folder_parts = relative_path.split(os.sep)
folder_name = folder_parts[0] # Primera carpeta (nombre del archivo ZIP/RAR sin extensión)
else:
# El archivo está en la raíz, usar el nombre del archivo sin extensión
folder_name = os.path.splitext(file_name)[0]
print(f"Folder name para validación: {folder_name}")
# Validar nomenclatura
match = nomenclatura_pattern.match(folder_name)
match_sin_anio = nomenclatura_pattern_sin_anio.match(folder_name)
if not match and not match_sin_anio: if not match and not match_sin_anio:
print(f"Nomenclatura inválida: {folder_name}") print(f"Nomenclatura inválida en nombre de archivo: {archivo_name_sin_extension}")
# Determinar el archivo original basado en el subdirectorio
archivo_original = folder_name + ('.zip' if any(f.endswith('.zip') for f in [a.name for a in archivos]) else '.rar')
failed_files.append({ failed_files.append({
"file": relative_path, "archivo_original": archivo.name,
"archivo_original": archivo_original, "error": f"Nomenclatura inválida: {archivo_name_sin_extension}. Esperado: anio-aduana-patente-pedimento"
"error": f"Nomenclatura inválida: {folder_name}. Esperado: anio-aduana-patente-pedimento"
}) })
continue continue
# Extraer información del pedimento desde el nombre del archivo
if match: if match:
print(f"Nomenclatura válida: {folder_name}")
anio, aduana, patente, pedimento_num = match.groups() anio, aduana, patente, pedimento_num = match.groups()
print(f"Extraído - Año: {anio}, Aduana: {aduana}, Patente: {patente}, Pedimento: {pedimento_num}") print(f"Extraído del nombre del archivo - Año: {anio}, Aduana: {aduana}, Patente: {patente}, Pedimento: {pedimento_num}")
# Formato original: anio-aduana-patente-pedimento
# Crear fecha_pago basada en el año
try: try:
# Convertir año de 2 dígitos a 4 dígitos # Convertir año de 2 dígitos a 4 dígitos
anio_completo = 2000 + int(anio) if int(anio) < 50 else 1900 + int(anio) anio_completo = 2000 + int(anio) if int(anio) < 50 else 1900 + int(anio)
fecha_pago = datetime(anio_completo, 1, 1).date() fecha_pago = datetime(anio_completo, 1, 1).date()
print(f"Fecha de pago calculada: {fecha_pago}") print(f"Fecha de pago calculada: {fecha_pago}")
except ValueError: except ValueError:
archivo_original = folder_name + ('.zip' if any(f.endswith('.zip') for f in [a.name for a in archivos]) else '.rar')
failed_files.append({ failed_files.append({
"file": relative_path, "archivo_original": archivo.name,
"archivo_original": archivo_original,
"error": f"Año inválido: {anio}" "error": f"Año inválido: {anio}"
}) })
continue continue
elif match_sin_anio: elif match_sin_anio:
print(f"Nomenclatura válida sin año: {folder_name}")
# Formato sin año: aduana-patente-pedimento
aduana, patente, pedimento_num = match_sin_anio.groups() aduana, patente, pedimento_num = match_sin_anio.groups()
print(f"Extraído - Aduana: {aduana}, Patente: {patente}, Pedimento: {pedimento_num}") print(f"Extraído del nombre del archivo - Aduana: {aduana}, Patente: {patente}, Pedimento: {pedimento_num}")
# Obtener el primer dígito del pedimento # Obtener el primer dígito del pedimento
primer_digito_pedimento = int(pedimento_num[0]) if pedimento_num else 0 primer_digito_pedimento = int(pedimento_num[0]) if pedimento_num else 0
@@ -745,10 +648,8 @@ class ViewSetPedimento(LoggingMixin, viewsets.ModelViewSet, OrganizacionFiltrada
# Aplicar lógica de comparación # Aplicar lógica de comparación
if año_con_digito <= año_actual: if año_con_digito <= año_actual:
# Si el año con dígito es menor o igual al año actual
año_final = año_con_digito año_final = año_con_digito
else: else:
# Si el año con dígito es mayor al año actual, restar 10
año_final = año_con_digito - 10 año_final = año_con_digito - 10
# Tomar los últimos 2 dígitos del año final # Tomar los últimos 2 dígitos del año final
@@ -756,24 +657,84 @@ class ViewSetPedimento(LoggingMixin, viewsets.ModelViewSet, OrganizacionFiltrada
# Crear fecha de pago (primer día del año) # Crear fecha de pago (primer día del año)
fecha_pago = datetime(año_final, 1, 1).date() fecha_pago = datetime(año_final, 1, 1).date()
print(f"Fecha de pago (año actual) calculada: {fecha_pago}") print(f"Fecha de pago (año actual) calculada: {fecha_pago}")
# Generar pedimento_app # Generar pedimento_app
pedimento_app = f"{anio}-{aduana.zfill(2)}-{patente}-{pedimento_num}" pedimento_app = f"{anio}-{aduana.zfill(2)}-{patente}-{pedimento_num}"
print(f"Pedimento_app generado: {pedimento_app}") print(f"Pedimento_app generado: {pedimento_app}")
# VERIFICAR SI EL PEDIMENTO YA EXISTE ANTES DE PROCESAR EL ARCHIVO
print(f"Buscando pedimento existente con pedimento_app: {pedimento_app} y organización ID: {organizacion.id}") print(f"Buscando pedimento existente con pedimento_app: {pedimento_app} y organización ID: {organizacion.id}")
# Verificar si el pedimento ya existe
existing_pedimento = Pedimento.objects.filter( existing_pedimento = Pedimento.objects.filter(
pedimento_app=pedimento_app, pedimento_app=pedimento_app,
# organizacion=organizacion organizacion=organizacion
).first() ).first()
print(f"Pedimento existente: {existing_pedimento is not None}") if existing_pedimento:
print(f"⚠️ Pedimento ya existe: ID {existing_pedimento.id}, pedimento_app: {pedimento_app}")
already_existing_pedimentos.append({
"id": str(existing_pedimento.id),
"pedimento_app": pedimento_app,
"contribuyente": existing_pedimento.contribuyente.rfc if existing_pedimento.contribuyente else None,
"archivo_original": archivo.name
})
# NO procesamos este archivo, pasamos al siguiente
continue
if not existing_pedimento: # Si el pedimento no existe, continuar con el procesamiento normal
print("📝 Pedimento no existe, creando nuevo...") print("📝 Pedimento no existe, continuando con procesamiento...")
# Crear nuevo pedimento
# Crear subdirectorio para cada archivo usando el nombre del archivo sin extensión
sub_dir = os.path.join(temp_dir, archivo_name_sin_extension)
os.makedirs(sub_dir, exist_ok=True)
print(f"Subdirectorio creado: {sub_dir}")
if archivo_name.endswith('.zip'):
# Manejar archivo ZIP
print("Es un archivo ZIP")
try:
with zipfile.ZipFile(archivo, 'r') as zip_ref:
zip_ref.extractall(sub_dir)
print("Archivo ZIP extraído exitosamente")
except zipfile.BadZipFile as e:
failed_files.append({
"archivo_original": archivo.name,
"error": f"Archivo ZIP corrupto o inválido: {str(e)}"
})
continue
except Exception as e:
failed_files.append({
"archivo_original": archivo.name,
"error": f"Error al extraer ZIP: {str(e)}"
})
continue
elif archivo_name.endswith('.rar'):
# Manejar archivo RAR: guardar el archivo en disco y usar helper con fallbacks
archivo_temp_path = os.path.join(sub_dir, archivo.name)
with open(archivo_temp_path, 'wb') as f:
for chunk in archivo.chunks():
f.write(chunk)
try:
extract_rar_to_dir(archivo_temp_path, sub_dir)
print(f"Archivo RAR {archivo.name} extraído en {sub_dir}")
except Exception as e:
error_msg = str(e)
help_msg = "Instale 'unrar' o 'p7zip' (7z) y asegúrese de que estén en PATH, o instale y configure 'rarfile' con un backend."
failed_files.append({
"archivo_original": archivo.name,
"error": f"Error al extraer archivo RAR: {error_msg}"
})
continue
else:
# Asumir que es un archivo individual
archivo_path = os.path.join(sub_dir, archivo.name)
with open(archivo_path, 'wb') as f:
for chunk in archivo.chunks():
f.write(chunk)
print(f"Archivo individual {archivo.name} guardado en sub_dir:", archivo_path)
# Ahora crear el pedimento (ya verificamos que no existe)
try: try:
print("🔄 Iniciando creación de pedimento...") print("🔄 Iniciando creación de pedimento...")
@@ -794,9 +755,12 @@ class ViewSetPedimento(LoggingMixin, viewsets.ModelViewSet, OrganizacionFiltrada
pedimento = Pedimento.objects.create( pedimento = Pedimento.objects.create(
organizacion=organizacion, organizacion=organizacion,
contribuyente=importador, contribuyente=importador,
pedimento=int(pedimento_num), # pedimento=int(pedimento_num),
aduana=int(aduana), pedimento=pedimento_num,
patente=int(patente), aduana=aduana,
# aduana=int(aduana),
# patente=int(patente),
patente=patente,
fecha_pago=fecha_pago, fecha_pago=fecha_pago,
pedimento_app=pedimento_app, pedimento_app=pedimento_app,
agente_aduanal=f"Agente {patente}", # Valor por defecto agente_aduanal=f"Agente {patente}", # Valor por defecto
@@ -809,36 +773,55 @@ class ViewSetPedimento(LoggingMixin, viewsets.ModelViewSet, OrganizacionFiltrada
"id": str(pedimento.id), "id": str(pedimento.id),
"pedimento_app": pedimento_app, "pedimento_app": pedimento_app,
"contribuyente": importador.rfc, "contribuyente": importador.rfc,
"contribuyente_nombre": importador.nombre "contribuyente_nombre": importador.nombre,
"archivo_original": archivo.name
}) })
except Exception as e: except Exception as e:
print(f"❌ Error al crear pedimento: {str(e)}") print(f"❌ Error al crear pedimento: {str(e)}")
archivo_original = folder_name + ('.zip' if any(f.endswith('.zip') for f in [a.name for a in archivos]) else '.rar')
failed_files.append({ failed_files.append({
"file": relative_path, "archivo_original": archivo.name,
"archivo_original": archivo_original,
"error": f"Error al crear pedimento: {str(e)}" "error": f"Error al crear pedimento: {str(e)}"
}) })
continue continue
else:
pedimento = existing_pedimento # Procesar documentos dentro del directorio
print("Procesando documentos del directorio...")
for root, dirs, files in os.walk(sub_dir):
for file_name in files:
file_path = os.path.join(root, file_name)
print(f"Procesando documento: {file_name}")
try: try:
# Leer el archivo desde el directorio temporal # Leer el archivo desde el directorio temporal
with open(file_path, 'rb') as f: with open(file_path, 'rb') as f:
file_content = f.read() file_content = f.read()
from api.utils.helpers import extraer_info_pedimento_xml
# Extraer info del pedimento desde XML si es aplicable
if file_name.lower().endswith('.xml'):
try:
xml_info = extraer_info_pedimento_xml(file_content)
if xml_info:
if 'numero_operacion' in xml_info:
if 'numero_pedimento' in xml_info:
if xml_info['numero_pedimento'] == str(pedimento.pedimento):
Pedimento.objects.filter(id=pedimento.id).update(
aduana=xml_info.get('aduana_clave', pedimento.aduana)
)
print(f"Información extraída del XML: {xml_info}")
except Exception as e:
print(f"No se pudo extraer información del XML {file_name}: {str(e)}")
# Obtener información del archivo # Obtener información del archivo
extension = os.path.splitext(file_name)[1].lower().lstrip('.') extension = os.path.splitext(file_name)[1].lower().lstrip('.')
# Buscar todos los documentos existentes para este pedimento # Buscar si ya existe un documento con el mismo nombre para este pedimento
existing_documents = Document.objects.filter( existing_documents = Document.objects.filter(
pedimento_id=pedimento.id, pedimento_id=pedimento.id,
organizacion=organizacion organizacion=organizacion
) )
# Buscar si ya existe un documento con el mismo nombre base
existing_document = None existing_document = None
for doc in existing_documents: for doc in existing_documents:
if is_same_document(doc, file_name): if is_same_document(doc, file_name):
@@ -863,6 +846,8 @@ class ViewSetPedimento(LoggingMixin, viewsets.ModelViewSet, OrganizacionFiltrada
existing_document.extension = extension existing_document.extension = extension
existing_document.updated_at = timezone.now() # Si tienes este campo existing_document.updated_at = timezone.now() # Si tienes este campo
existing_document.save() existing_document.save()
documents_created += 1
print(f"📄 Documento actualizado: {file_name}")
else: else:
# Crear nuevo documento # Crear nuevo documento
@@ -875,20 +860,15 @@ class ViewSetPedimento(LoggingMixin, viewsets.ModelViewSet, OrganizacionFiltrada
size=len(file_content), size=len(file_content),
extension=extension extension=extension
) )
documents_created += 1 documents_created += 1
print(f"📄 Nuevo documento creado: {file_name}")
except Exception as e: except Exception as e:
print(f"❌ Error al crear documento: {str(e)}") print(f"❌ Error al procesar documento {file_name}: {str(e)}")
archivo_original = folder_name + ('.zip' if any(f.endswith('.zip') for f in [a.name for a in archivos]) else '.rar') # Continuar con otros documentos
failed_files.append({
"file": relative_path,
"archivo_original": archivo_original,
"error": f"Error al crear documento: {str(e)}"
})
continue
print(f"🏁 Procesamiento completado. Archivos procesados en este directorio.") print(f"🏁 Procesamiento completado. Archivos procesados en este directorio.")
except Exception as e: except Exception as e:
return Response( return Response(
{"error": f"Error durante el procesamiento: {str(e)}"}, {"error": f"Error durante el procesamiento: {str(e)}"},
@@ -903,22 +883,38 @@ class ViewSetPedimento(LoggingMixin, viewsets.ModelViewSet, OrganizacionFiltrada
response_data = { response_data = {
"created_count": len(created_pedimentos), "created_count": len(created_pedimentos),
"created_pedimentos": created_pedimentos, "created_pedimentos": created_pedimentos,
"already_existing_count": len(already_existing_pedimentos),
"already_existing": already_existing_pedimentos,
"documents_created": documents_created, "documents_created": documents_created,
"failed_files": failed_files, "failed_files": failed_files,
"processed_files": len(archivos), "processed_files": len(archivos),
"summary": f"Procesados {len(archivos)} archivo(s): {len(created_pedimentos)} pedimento(s) creado(s), {documents_created} documento(s) asociado(s)" "summary": f"Procesados {len(archivos)} archivo(s): {len(created_pedimentos)} pedimento(s) creado(s), {len(already_existing_pedimentos)} ya existían, {documents_created} documento(s) asociado(s)"
} }
if failed_files: try:
# Determinar el mensaje apropiado
if already_existing_pedimentos and not created_pedimentos and not failed_files:
response_data["message"] = "Todos los pedimentos ya existen. No se crearon nuevos pedimentos."
response_status = status.HTTP_200_OK
elif already_existing_pedimentos or failed_files:
response_data.update({ response_data.update({
"message": "Procesamiento completado con algunos errores", "message": "Procesamiento completado con advertencias",
"errors": [item["error"] for item in failed_files]
}) })
if failed_files:
response_data["errors"] = [item["error"] for item in failed_files]
response_status = status.HTTP_207_MULTI_STATUS response_status = status.HTTP_207_MULTI_STATUS
else: else:
response_data["message"] = "Pedimentos creados exitosamente" response_data["message"] = "Pedimentos creados exitosamente"
response_status = status.HTTP_201_CREATED response_status = status.HTTP_201_CREATED
except Exception as e:
return Response(
{"error": f"Error durante el procesamiento: {str(e)}"},
status=status.HTTP_500_INTERNAL_SERVER_ERROR
)
return Response(response_data, status=response_status) return Response(response_data, status=response_status)
@action(detail=False, methods=['post'], url_path='bulk-create-pedimento_desk', parser_classes=[MultiPartParser, FormParser]) @action(detail=False, methods=['post'], url_path='bulk-create-pedimento_desk', parser_classes=[MultiPartParser, FormParser])

0
api/utils/__init__.py Normal file
View File

194
api/utils/helpers.py Normal file
View File

@@ -0,0 +1,194 @@
# auditoria_xml.py
import xml.etree.ElementTree as ET
from datetime import datetime
def extraer_info_pedimento_xml(xml_content):
"""
Extrae información específica de un XML de pedimento.
"""
try:
# Parsear el XML
root = ET.fromstring(xml_content)
# Buscar el namespace (puede variar)
namespaces = {
'S': 'http://schemas.xmlsoap.org/soap/envelope/',
'ns2': 'http://www.ventanillaunica.gob.mx/pedimentos/ws/oxml/consultarpedimentocompleto',
'ns3': 'http://www.ventanillaunica.gob.mx/common/ws/oxml/respuesta'
}
resultado = {}
# Extraer número de operación
num_op = root.find('.//ns2:numeroOperacion', namespaces)
if num_op is not None and num_op.text:
resultado['numero_operacion'] = num_op.text
# Extraer información del pedimento
pedimento_elem = root.find('.//ns2:pedimento', namespaces)
if pedimento_elem is not None:
# Número de pedimento
ped_num = pedimento_elem.find('ns2:pedimento', namespaces)
if ped_num is not None and ped_num.text:
resultado['numero_pedimento'] = ped_num.text
# Número de partidas
partidas = pedimento_elem.find('ns2:partidas', namespaces)
if partidas is not None and partidas.text:
try:
resultado['numero_partidas'] = int(partidas.text)
except (ValueError, TypeError):
pass
# Tipo de operación clave
tipo_op_clave = pedimento_elem.find('.//ns2:tipoOperacion/ns2:clave', namespaces)
if tipo_op_clave is not None and tipo_op_clave.text:
if tipo_op_clave.text.strip() == '1':
resultado['tipo_operacion'] = 'Importacion'
resultado['tipo_operacion_descripcion'] = 'Indica operacion como Importaciones'
elif tipo_op_clave.text.strip() == '2':
resultado['tipo_operacion'] = 'Exportacion'
resultado['tipo_operacion_descripcion'] = 'Indica operacion de exportacion'
# Clave del documento (clave_pedimento)
clave_doc = pedimento_elem.find('.//ns2:claveDocumento/ns2:clave', namespaces)
if clave_doc is not None and clave_doc.text:
resultado['clave_pedimento'] = clave_doc.text.strip()
# Aduana (patente)
aduana = pedimento_elem.find('.//ns2:aduanaEntradaSalida/ns2:clave', namespaces)
if aduana is not None and aduana.text:
resultado['aduana_clave'] = aduana.text.strip()
# Importador/Exportador
importador = pedimento_elem.find('.//ns2:importadorExportador', namespaces)
if importador is not None:
rfc = importador.find('ns2:rfc', namespaces)
if rfc is not None and rfc.text:
resultado['contribuyente_rfc'] = rfc.text.strip()
razon_social = importador.find('ns2:razonSocial', namespaces)
if razon_social is not None and razon_social.text:
resultado['contribuyente_nombre'] = razon_social.text.strip()
# Valor en dólares
valor_dolares = importador.find('ns2:valorDolares', namespaces)
if valor_dolares is not None and valor_dolares.text:
try:
resultado['valor_dolares'] = float(valor_dolares.text)
except (ValueError, TypeError):
pass
# Aduana de despacho
aduana_despacho = importador.find('ns2:aaduanaDespacho/ns2:clave', namespaces)
if aduana_despacho is not None and aduana_despacho.text:
resultado['aduana_despacho'] = aduana_despacho.text.strip()
# Encabezado del pedimento
encabezado = pedimento_elem.find('ns2:encabezado', namespaces)
if encabezado is not None:
# Aduana
aduana = encabezado.find('ns2:aduanaEntradaSalida/ns2:clave', namespaces)
if aduana is not None and aduana.text:
resultado['aduana_clave'] = aduana.text.strip()
# Tipo de cambio
tipo_cambio = encabezado.find('ns2:tipoCambio', namespaces)
if tipo_cambio is not None and tipo_cambio.text:
try:
resultado['tipo_cambio'] = float(tipo_cambio.text)
except (ValueError, TypeError):
pass
# RFC Agente Aduanal
rfc_agente = encabezado.find('ns2:rfcAgenteAduanalSocFactura', namespaces)
if rfc_agente is not None and rfc_agente.text:
resultado['rfc_agente_aduanal'] = rfc_agente.text.strip()
# CURP Apoderado
curp_apoderado = encabezado.find('ns2:curpApoderadomandatario', namespaces)
if curp_apoderado is not None and curp_apoderado.text:
resultado['curp_apoderado'] = curp_apoderado.text.strip()
# Valor Aduanal Total
valor_aduanal = encabezado.find('ns2:valorAduanalTotal', namespaces)
if valor_aduanal is not None and valor_aduanal.text:
try:
resultado['valor_aduanal_total'] = float(valor_aduanal.text)
except (ValueError, TypeError):
pass
# Valor Comercial Total
valor_comercial = encabezado.find('ns2:valorComercialTotal', namespaces)
if valor_comercial is not None and valor_comercial.text:
try:
resultado['valor_comercial_total'] = float(valor_comercial.text)
except (ValueError, TypeError):
pass
# Fechas
fechas = pedimento_elem.findall('.//ns2:fechas', namespaces)
for fecha_elem in fechas:
fecha = fecha_elem.find('ns2:fecha', namespaces)
clave_fecha = fecha_elem.find('ns2:tipo/ns2:clave', namespaces)
if fecha is not None and fecha.text and clave_fecha is not None and clave_fecha.text:
fecha_texto = fecha.text.strip()
clave_fecha_texto = clave_fecha.text.strip()
# Mapeo de claves según especificación
if clave_fecha_texto == '1': # Entrada
resultado['fecha_entrada'] = fecha_texto
elif clave_fecha_texto == '2': # Pago
resultado['fecha_pago'] = fecha_texto
elif clave_fecha_texto == '3': # Extracción
resultado['fecha_extraccion'] = fecha_texto
elif clave_fecha_texto == '5': # Presentación
resultado['fecha_presentacion'] = fecha_texto
elif clave_fecha_texto == '6': # Importación
resultado['fecha_importacion'] = fecha_texto
elif clave_fecha_texto == '7': # Original
resultado['fecha_original'] = fecha_texto
else:
resultado[f'fecha_clave_{clave_fecha_texto}'] = fecha_texto
# Facturas (para COVEs)
facturas = pedimento_elem.findall('.//ns2:facturas', namespaces)
coves_encontrados = []
for factura in facturas:
numero = factura.find('ns2:numero', namespaces)
if numero is not None and numero.text:
coves_encontrados.append(numero.text.strip())
if coves_encontrados:
resultado['coves_en_xml'] = coves_encontrados
# E-Documents
identificadores = pedimento_elem.findall('.//ns2:identificadores/ns2:identificadores', namespaces)
edocs_encontrados = []
for ident in identificadores:
clave = ident.find('claveIdentificador/descripcion', namespaces)
complemento = ident.find('complemento1', namespaces)
if clave is not None and clave.text and 'E_DOCUMENT' in clave.text:
if complemento is not None and complemento.text:
edocs_encontrados.append(complemento.text.strip())
if edocs_encontrados:
resultado['edocuments_en_xml'] = edocs_encontrados
# Verificar si hay error en la respuesta
tiene_error = root.find('.//ns3:tieneError', namespaces)
if tiene_error is not None:
resultado['tiene_error'] = tiene_error.text.lower() == 'true'
return resultado
except ET.ParseError as e:
return {'error_parse': str(e)}
except Exception as e:
return {'error': str(e)}