Compare commits
2 Commits
only-datas
...
fix-T2025-
| Author | SHA1 | Date | |
|---|---|---|---|
| 04d19118be | |||
| 4ccb5fd718 |
@@ -513,7 +513,8 @@ class ViewSetPedimento(LoggingMixin, viewsets.ModelViewSet, OrganizacionFiltrada
|
||||
"processed_files": 3,
|
||||
"summary": "Procesados 3 archivo(s): 5 pedimento(s) creado(s), 15 documento(s) asociado(s)",
|
||||
"failed_files": [],
|
||||
"errors": []
|
||||
"errors": [],
|
||||
"already_existing": [] # Nuevo campo para pedimentos que ya existían
|
||||
}
|
||||
"""
|
||||
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})$')
|
||||
|
||||
created_pedimentos = []
|
||||
already_existing_pedimentos = [] # Para trackear pedimentos que ya existen
|
||||
failed_files = []
|
||||
errors = []
|
||||
documents_created = 0
|
||||
@@ -599,8 +601,90 @@ class ViewSetPedimento(LoggingMixin, viewsets.ModelViewSet, OrganizacionFiltrada
|
||||
archivo_name = archivo.name.lower()
|
||||
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]
|
||||
|
||||
# Validar nomenclatura del nombre del archivo/folder
|
||||
match = nomenclatura_pattern.match(archivo_name_sin_extension)
|
||||
match_sin_anio = nomenclatura_pattern_sin_anio.match(archivo_name_sin_extension)
|
||||
|
||||
if not match and not match_sin_anio:
|
||||
print(f"Nomenclatura inválida en nombre de archivo: {archivo_name_sin_extension}")
|
||||
failed_files.append({
|
||||
"archivo_original": archivo.name,
|
||||
"error": f"Nomenclatura inválida: {archivo_name_sin_extension}. Esperado: anio-aduana-patente-pedimento"
|
||||
})
|
||||
continue
|
||||
|
||||
# Extraer información del pedimento desde el nombre del archivo
|
||||
if match:
|
||||
anio, aduana, patente, pedimento_num = match.groups()
|
||||
print(f"Extraído del nombre del archivo - Año: {anio}, Aduana: {aduana}, Patente: {patente}, Pedimento: {pedimento_num}")
|
||||
|
||||
try:
|
||||
# Convertir año de 2 dígitos a 4 dígitos
|
||||
anio_completo = 2000 + int(anio) if int(anio) < 50 else 1900 + int(anio)
|
||||
fecha_pago = datetime(anio_completo, 1, 1).date()
|
||||
print(f"Fecha de pago calculada: {fecha_pago}")
|
||||
except ValueError:
|
||||
failed_files.append({
|
||||
"archivo_original": archivo.name,
|
||||
"error": f"Año inválido: {anio}"
|
||||
})
|
||||
continue
|
||||
|
||||
elif match_sin_anio:
|
||||
aduana, patente, pedimento_num = match_sin_anio.groups()
|
||||
print(f"Extraído del nombre del archivo - Aduana: {aduana}, Patente: {patente}, Pedimento: {pedimento_num}")
|
||||
|
||||
# Obtener el primer dígito del pedimento
|
||||
primer_digito_pedimento = int(pedimento_num[0]) if pedimento_num else 0
|
||||
|
||||
# Usar año actual para fecha_pago y ajustar según el dígito del pedimento
|
||||
año_actual = datetime.now().year
|
||||
|
||||
# Crear año con el dígito del pedimento (reemplazando el último dígito)
|
||||
año_con_digito = int(str(año_actual)[:-1] + str(primer_digito_pedimento))
|
||||
|
||||
# Aplicar lógica de comparación
|
||||
if año_con_digito <= año_actual:
|
||||
año_final = año_con_digito
|
||||
else:
|
||||
año_final = año_con_digito - 10
|
||||
|
||||
# Tomar los últimos 2 dígitos del año final
|
||||
anio = año_final % 100
|
||||
|
||||
# Crear fecha de pago (primer día del año)
|
||||
fecha_pago = datetime(año_final, 1, 1).date()
|
||||
print(f"Fecha de pago (año actual) calculada: {fecha_pago}")
|
||||
|
||||
# Generar pedimento_app
|
||||
pedimento_app = f"{anio}-{aduana.zfill(2)}-{patente}-{pedimento_num}"
|
||||
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}")
|
||||
existing_pedimento = Pedimento.objects.filter(
|
||||
pedimento_app=pedimento_app,
|
||||
organizacion=organizacion
|
||||
).first()
|
||||
|
||||
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
|
||||
|
||||
# Si el pedimento no existe, continuar con el procesamiento normal
|
||||
print("📝 Pedimento no existe, continuando con procesamiento...")
|
||||
|
||||
# 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}")
|
||||
@@ -613,18 +697,20 @@ class ViewSetPedimento(LoggingMixin, viewsets.ModelViewSet, OrganizacionFiltrada
|
||||
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
|
||||
)
|
||||
failed_files.append({
|
||||
"archivo_original": archivo.name,
|
||||
"error": f"Archivo ZIP corrupto o inválido: {str(e)}"
|
||||
})
|
||||
continue
|
||||
except Exception as e:
|
||||
return Response(
|
||||
{"error": f"Error al extraer ZIP {archivo.name}: {str(e)}"},
|
||||
status=status.HTTP_400_BAD_REQUEST
|
||||
)
|
||||
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
|
||||
# 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():
|
||||
@@ -635,260 +721,154 @@ class ViewSetPedimento(LoggingMixin, viewsets.ModelViewSet, OrganizacionFiltrada
|
||||
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
|
||||
# )
|
||||
|
||||
failed_files.append({
|
||||
"archivo_original": archivo.name,
|
||||
"error": f"Error al extraer archivo RAR: {error_msg}"
|
||||
})
|
||||
continue
|
||||
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)
|
||||
# Ahora crear el pedimento (ya verificamos que no existe)
|
||||
try:
|
||||
print("🔄 Iniciando creación de pedimento...")
|
||||
|
||||
# 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)
|
||||
# Obtener o crear el importador
|
||||
print(f"🏢 Buscando/creando importador con RFC: {contribuyente}")
|
||||
importador, created = Importador.objects.get_or_create(
|
||||
rfc=contribuyente,
|
||||
defaults={
|
||||
'nombre': f"Importador {contribuyente}",
|
||||
'organizacion': organizacion
|
||||
}
|
||||
)
|
||||
if created:
|
||||
print(f"✅ Importador creado: {importador.rfc} - {importador.nombre}")
|
||||
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"♻️ Importador existente: {importador.rfc} - {importador.nombre}")
|
||||
|
||||
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:
|
||||
print(f"Nomenclatura inválida: {folder_name}")
|
||||
# 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({
|
||||
"file": relative_path,
|
||||
"archivo_original": archivo_original,
|
||||
"error": f"Nomenclatura inválida: {folder_name}. Esperado: anio-aduana-patente-pedimento"
|
||||
})
|
||||
continue
|
||||
|
||||
if match:
|
||||
|
||||
print(f"Nomenclatura válida: {folder_name}")
|
||||
anio, aduana, patente, pedimento_num = match.groups()
|
||||
print(f"Extraído - 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:
|
||||
# Convertir año de 2 dígitos a 4 dígitos
|
||||
anio_completo = 2000 + int(anio) if int(anio) < 50 else 1900 + int(anio)
|
||||
fecha_pago = datetime(anio_completo, 1, 1).date()
|
||||
print(f"Fecha de pago calculada: {fecha_pago}")
|
||||
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({
|
||||
"file": relative_path,
|
||||
"archivo_original": archivo_original,
|
||||
"error": f"Año inválido: {anio}"
|
||||
})
|
||||
continue
|
||||
|
||||
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()
|
||||
print(f"Extraído - Aduana: {aduana}, Patente: {patente}, Pedimento: {pedimento_num}")
|
||||
|
||||
# Obtener el primer dígito del pedimento
|
||||
primer_digito_pedimento = int(pedimento_num[0]) if pedimento_num else 0
|
||||
|
||||
# Usar año actual para fecha_pago y ajustar según el dígito del pedimento
|
||||
año_actual = datetime.now().year
|
||||
|
||||
# Crear año con el dígito del pedimento (reemplazando el último dígito)
|
||||
año_con_digito = int(str(año_actual)[:-1] + str(primer_digito_pedimento))
|
||||
|
||||
# Aplicar lógica de comparación
|
||||
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
|
||||
else:
|
||||
# Si el año con dígito es mayor al año actual, restar 10
|
||||
año_final = año_con_digito - 10
|
||||
|
||||
# Tomar los últimos 2 dígitos del año final
|
||||
anio = año_final % 100
|
||||
|
||||
# Crear fecha de pago (primer día del año)
|
||||
fecha_pago = datetime(año_final , 1, 1).date()
|
||||
|
||||
print(f"Fecha de pago (año actual) calculada: {fecha_pago}")
|
||||
|
||||
# Generar pedimento_app
|
||||
pedimento_app = f"{anio}-{aduana.zfill(2)}-{patente}-{pedimento_num}"
|
||||
print(f"Pedimento_app generado: {pedimento_app}")
|
||||
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(
|
||||
pedimento = Pedimento.objects.create(
|
||||
organizacion=organizacion,
|
||||
contribuyente=importador,
|
||||
# pedimento=int(pedimento_num),
|
||||
pedimento=pedimento_num,
|
||||
aduana=aduana,
|
||||
# aduana=int(aduana),
|
||||
# patente=int(patente),
|
||||
patente=patente,
|
||||
fecha_pago=fecha_pago,
|
||||
pedimento_app=pedimento_app,
|
||||
# organizacion=organizacion
|
||||
).first()
|
||||
|
||||
print(f"Pedimento existente: {existing_pedimento is not None}")
|
||||
agente_aduanal=f"Agente {patente}", # Valor por defecto
|
||||
clave_pedimento="A1" # Valor por defecto
|
||||
)
|
||||
|
||||
if not existing_pedimento:
|
||||
print("📝 Pedimento no existe, creando nuevo...")
|
||||
# Crear nuevo pedimento
|
||||
print(f"✅ Pedimento creado exitosamente: ID {pedimento.id}, pedimento_app: {pedimento_app}")
|
||||
|
||||
created_pedimentos.append({
|
||||
"id": str(pedimento.id),
|
||||
"pedimento_app": pedimento_app,
|
||||
"contribuyente": importador.rfc,
|
||||
"contribuyente_nombre": importador.nombre,
|
||||
"archivo_original": archivo.name
|
||||
})
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Error al crear pedimento: {str(e)}")
|
||||
failed_files.append({
|
||||
"archivo_original": archivo.name,
|
||||
"error": f"Error al crear pedimento: {str(e)}"
|
||||
})
|
||||
continue
|
||||
|
||||
# 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:
|
||||
print("🔄 Iniciando creación de pedimento...")
|
||||
# Leer el archivo desde el directorio temporal
|
||||
with open(file_path, 'rb') as f:
|
||||
file_content = f.read()
|
||||
from api.utils.helpers import extraer_info_pedimento_xml
|
||||
|
||||
# Obtener o crear el importador
|
||||
print(f"🏢 Buscando/creando importador con RFC: {contribuyente}")
|
||||
importador, created = Importador.objects.get_or_create(
|
||||
rfc=contribuyente,
|
||||
defaults={
|
||||
'nombre': f"Importador {contribuyente}",
|
||||
'organizacion': organizacion
|
||||
}
|
||||
)
|
||||
if created:
|
||||
print(f"✅ Importador creado: {importador.rfc} - {importador.nombre}")
|
||||
else:
|
||||
print(f"♻️ Importador existente: {importador.rfc} - {importador.nombre}")
|
||||
# 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
|
||||
extension = os.path.splitext(file_name)[1].lower().lstrip('.')
|
||||
|
||||
pedimento = Pedimento.objects.create(
|
||||
organizacion=organizacion,
|
||||
contribuyente=importador,
|
||||
pedimento=int(pedimento_num),
|
||||
aduana=int(aduana),
|
||||
patente=int(patente),
|
||||
fecha_pago=fecha_pago,
|
||||
pedimento_app=pedimento_app,
|
||||
agente_aduanal=f"Agente {patente}", # Valor por defecto
|
||||
clave_pedimento="A1" # Valor por defecto
|
||||
)
|
||||
|
||||
print(f"✅ Pedimento creado exitosamente: ID {pedimento.id}, pedimento_app: {pedimento_app}")
|
||||
|
||||
created_pedimentos.append({
|
||||
"id": str(pedimento.id),
|
||||
"pedimento_app": pedimento_app,
|
||||
"contribuyente": importador.rfc,
|
||||
"contribuyente_nombre": importador.nombre
|
||||
})
|
||||
|
||||
except Exception as 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({
|
||||
"file": relative_path,
|
||||
"archivo_original": archivo_original,
|
||||
"error": f"Error al crear pedimento: {str(e)}"
|
||||
})
|
||||
continue
|
||||
else:
|
||||
pedimento = existing_pedimento
|
||||
|
||||
try:
|
||||
# Leer el archivo desde el directorio temporal
|
||||
with open(file_path, 'rb') as f:
|
||||
file_content = f.read()
|
||||
|
||||
# Obtener información del archivo
|
||||
extension = os.path.splitext(file_name)[1].lower().lstrip('.')
|
||||
|
||||
# Buscar todos los documentos existentes para este pedimento
|
||||
existing_documents = Document.objects.filter(
|
||||
pedimento_id=pedimento.id,
|
||||
organizacion=organizacion
|
||||
)
|
||||
|
||||
# Buscar si ya existe un documento con el mismo nombre base
|
||||
existing_document = None
|
||||
for doc in existing_documents:
|
||||
if is_same_document(doc, file_name):
|
||||
existing_document = doc
|
||||
print(f"✅ Encontrado documento existente: ID {doc.id}")
|
||||
break
|
||||
|
||||
# Crear ContentFile
|
||||
django_file = ContentFile(file_content, name=file_name)
|
||||
|
||||
if existing_document:
|
||||
# Opcional: Eliminar el archivo físico anterior
|
||||
try:
|
||||
if existing_document.archivo and os.path.exists(existing_document.archivo.path):
|
||||
os.remove(existing_document.archivo.path)
|
||||
except (ValueError, OSError) as e:
|
||||
print(f"No se pudo eliminar archivo físico anterior: {str(e)}")
|
||||
|
||||
# Actualizar el documento existente
|
||||
existing_document.archivo = django_file
|
||||
existing_document.size = len(file_content)
|
||||
existing_document.extension = extension
|
||||
existing_document.updated_at = timezone.now() # Si tienes este campo
|
||||
existing_document.save()
|
||||
|
||||
else:
|
||||
# Crear nuevo documento
|
||||
document = Document.objects.create(
|
||||
organizacion=organizacion,
|
||||
# Buscar si ya existe un documento con el mismo nombre para este pedimento
|
||||
existing_documents = Document.objects.filter(
|
||||
pedimento_id=pedimento.id,
|
||||
document_type=document_type,
|
||||
fuente_id=4,
|
||||
archivo=django_file,
|
||||
size=len(file_content),
|
||||
extension=extension
|
||||
organizacion=organizacion
|
||||
)
|
||||
|
||||
documents_created += 1
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Error al crear documento: {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({
|
||||
"file": relative_path,
|
||||
"archivo_original": archivo_original,
|
||||
"error": f"Error al crear documento: {str(e)}"
|
||||
})
|
||||
continue
|
||||
|
||||
existing_document = None
|
||||
for doc in existing_documents:
|
||||
if is_same_document(doc, file_name):
|
||||
existing_document = doc
|
||||
print(f"✅ Encontrado documento existente: ID {doc.id}")
|
||||
break
|
||||
|
||||
# Crear ContentFile
|
||||
django_file = ContentFile(file_content, name=file_name)
|
||||
|
||||
if existing_document:
|
||||
# Opcional: Eliminar el archivo físico anterior
|
||||
try:
|
||||
if existing_document.archivo and os.path.exists(existing_document.archivo.path):
|
||||
os.remove(existing_document.archivo.path)
|
||||
except (ValueError, OSError) as e:
|
||||
print(f"No se pudo eliminar archivo físico anterior: {str(e)}")
|
||||
|
||||
# Actualizar el documento existente
|
||||
existing_document.archivo = django_file
|
||||
existing_document.size = len(file_content)
|
||||
existing_document.extension = extension
|
||||
existing_document.updated_at = timezone.now() # Si tienes este campo
|
||||
existing_document.save()
|
||||
documents_created += 1
|
||||
print(f"📄 Documento actualizado: {file_name}")
|
||||
|
||||
else:
|
||||
# Crear nuevo documento
|
||||
document = Document.objects.create(
|
||||
organizacion=organizacion,
|
||||
pedimento_id=pedimento.id,
|
||||
document_type=document_type,
|
||||
fuente_id=4,
|
||||
archivo=django_file,
|
||||
size=len(file_content),
|
||||
extension=extension
|
||||
)
|
||||
documents_created += 1
|
||||
print(f"📄 Nuevo documento creado: {file_name}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Error al procesar documento {file_name}: {str(e)}")
|
||||
# Continuar con otros documentos
|
||||
|
||||
print(f"🏁 Procesamiento completado. Archivos procesados en este directorio.")
|
||||
|
||||
except Exception as e:
|
||||
return Response(
|
||||
{"error": f"Error durante el procesamiento: {str(e)}"},
|
||||
@@ -903,21 +883,37 @@ class ViewSetPedimento(LoggingMixin, viewsets.ModelViewSet, OrganizacionFiltrada
|
||||
response_data = {
|
||||
"created_count": len(created_pedimentos),
|
||||
"created_pedimentos": created_pedimentos,
|
||||
"already_existing_count": len(already_existing_pedimentos),
|
||||
"already_existing": already_existing_pedimentos,
|
||||
"documents_created": documents_created,
|
||||
"failed_files": failed_files,
|
||||
"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:
|
||||
response_data.update({
|
||||
"message": "Procesamiento completado con algunos errores",
|
||||
"errors": [item["error"] for item in failed_files]
|
||||
})
|
||||
response_status = status.HTTP_207_MULTI_STATUS
|
||||
else:
|
||||
response_data["message"] = "Pedimentos creados exitosamente"
|
||||
response_status = status.HTTP_201_CREATED
|
||||
|
||||
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({
|
||||
"message": "Procesamiento completado con advertencias",
|
||||
})
|
||||
if failed_files:
|
||||
response_data["errors"] = [item["error"] for item in failed_files]
|
||||
response_status = status.HTTP_207_MULTI_STATUS
|
||||
else:
|
||||
response_data["message"] = "Pedimentos creados exitosamente"
|
||||
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)
|
||||
|
||||
|
||||
0
api/utils/__init__.py
Normal file
0
api/utils/__init__.py
Normal file
194
api/utils/helpers.py
Normal file
194
api/utils/helpers.py
Normal 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)}
|
||||
Reference in New Issue
Block a user