211 lines
8.7 KiB
Python
211 lines
8.7 KiB
Python
from rest_framework.decorators import api_view, permission_classes
|
|
from rest_framework.permissions import IsAuthenticated
|
|
from rest_framework.response import Response
|
|
from rest_framework.pagination import PageNumberPagination
|
|
from django.db.models import Value, CharField, Q, Exists, OuterRef, Subquery
|
|
from django.db.models.functions import Cast
|
|
from datetime import datetime, timedelta
|
|
from api.customs.models import Pedimento, Cove, EDocument, Partida
|
|
|
|
class CustomPagination(PageNumberPagination):
|
|
page_size = 50
|
|
page_size_query_param = 'page_size'
|
|
max_page_size = 1000
|
|
|
|
@api_view(['GET'])
|
|
@permission_classes([IsAuthenticated])
|
|
def table_summary(request):
|
|
# ...existing code...
|
|
|
|
# Si se solicita CSV, generar archivo y devolverlo (después de definir pedimentos_filters)
|
|
|
|
|
|
my_tags = ['Reportes']
|
|
"""
|
|
Endpoint que devuelve un resumen tabulado de pedimentos y sus documentos asociados.
|
|
"""
|
|
|
|
org_id = request.query_params.get('organizacion_id')
|
|
if not org_id:
|
|
return Response({"error": "organizacion_id es requerido"}, status=400)
|
|
|
|
# Obtener filtros de query params
|
|
tipo_documento = request.query_params.get('tipo_documento')
|
|
rfc = request.query_params.get('contribuyente__rfc')
|
|
fecha_pago_gte = request.query_params.get('fecha_pago__gte')
|
|
fecha_pago_lte = request.query_params.get('fecha_pago__lte')
|
|
patente = request.query_params.get('patente')
|
|
aduana = request.query_params.get('aduana')
|
|
pedimento = request.query_params.get('pedimento')
|
|
pedimento_app = request.query_params.get('pedimento_app')
|
|
regimen = request.query_params.get('regimen')
|
|
tipo_operacion = request.query_params.get('tipo_operacion')
|
|
|
|
# Si no se proporcionan fechas, establecer un rango por defecto de los últimos 30 días
|
|
if not fecha_pago_gte and not fecha_pago_lte:
|
|
fecha_pago_lte = datetime.now().date()
|
|
fecha_pago_gte = fecha_pago_lte - timedelta(days=30)
|
|
|
|
# Construir filtros base para pedimentos
|
|
pedimentos_filters = Q()
|
|
pedimentos_filters &= Q(organizacion_id=org_id)
|
|
if fecha_pago_gte:
|
|
pedimentos_filters &= Q(fecha_pago__gte=fecha_pago_gte)
|
|
if fecha_pago_lte:
|
|
pedimentos_filters &= Q(fecha_pago__lte=fecha_pago_lte)
|
|
if rfc:
|
|
pedimentos_filters &= Q(contribuyente__rfc=rfc)
|
|
if patente:
|
|
pedimentos_filters &= Q(patente=patente)
|
|
if aduana:
|
|
pedimentos_filters &= Q(aduana=aduana)
|
|
if pedimento:
|
|
pedimentos_filters &= Q(pedimento=pedimento)
|
|
if pedimento_app:
|
|
pedimentos_filters &= Q(pedimento_app=pedimento_app)
|
|
if regimen:
|
|
pedimentos_filters &= Q(regimen=regimen)
|
|
if tipo_operacion:
|
|
pedimentos_filters &= Q(tipo_operacion_id=tipo_operacion)
|
|
|
|
# Si se solicita los últimos 100 registros actualizados
|
|
if request.query_params.get('ultimos') == '1':
|
|
pedimentos = Pedimento.objects.filter(pedimentos_filters).order_by('-updated_at')[:100]
|
|
else:
|
|
pedimentos = Pedimento.objects.filter(pedimentos_filters)
|
|
|
|
# Serializar pedimentos con documentos relacionados
|
|
results = []
|
|
for ped in pedimentos:
|
|
ped_dict = {
|
|
'aduana': ped.aduana,
|
|
'patente': ped.patente,
|
|
'regimen': ped.regimen,
|
|
'pedimento': ped.pedimento,
|
|
'pedimento_app': ped.pedimento_app,
|
|
'clave_pedimento': ped.clave_pedimento,
|
|
'tipo_operacion_id': ped.tipo_operacion_id,
|
|
'contribuyente_id': ped.contribuyente_id,
|
|
'documentos': []
|
|
}
|
|
# COVEs
|
|
for cove in Cove.objects.filter(pedimento=ped):
|
|
ped_dict['documentos'].append({
|
|
'tipo': 'COVE',
|
|
'numero': cove.numero_cove,
|
|
'estado': cove.cove_descargado,
|
|
'acuse_estado': cove.acuse_cove_descargado
|
|
})
|
|
# EDOCs
|
|
for edoc in EDocument.objects.filter(pedimento=ped):
|
|
ped_dict['documentos'].append({
|
|
'tipo': 'EDOC',
|
|
'numero': edoc.numero_edocument,
|
|
'estado': edoc.edocument_descargado,
|
|
'acuse_estado': edoc.acuse_descargado,
|
|
})
|
|
# PARTIDAS
|
|
for partida in Partida.objects.filter(pedimento=ped):
|
|
ped_dict['documentos'].append({
|
|
'tipo': 'PARTIDA',
|
|
'numero': partida.numero_partida,
|
|
'estado': partida.descargado
|
|
})
|
|
results.append(ped_dict)
|
|
|
|
if request.query_params.get('csv') == '1':
|
|
import csv
|
|
from django.http import HttpResponse
|
|
headers = [
|
|
'aduana', 'patente', 'regimen', 'pedimento', 'pedimento_app', 'clave_pedimento',
|
|
'tipo_operacion_id', 'contribuyente_id', 'tipo_documento', 'numero_documento', 'estado', 'acuse_estado'
|
|
]
|
|
response = HttpResponse(content_type='text/csv')
|
|
response['Content-Disposition'] = 'attachment; filename=table_summary.csv'
|
|
writer = csv.writer(response)
|
|
writer.writerow(headers)
|
|
# Llenar filas
|
|
if request.query_params.get('ultimos') == '1':
|
|
pedimentos = Pedimento.objects.filter(pedimentos_filters).order_by('-updated_at')[:100]
|
|
else:
|
|
pedimentos = Pedimento.objects.filter(pedimentos_filters)
|
|
for ped in pedimentos:
|
|
# COVEs
|
|
for cove in Cove.objects.filter(pedimento=ped):
|
|
writer.writerow([
|
|
ped.aduana, ped.patente, ped.regimen, ped.pedimento, ped.pedimento_app,
|
|
ped.clave_pedimento, ped.tipo_operacion_id, ped.contribuyente_id,
|
|
'COVE', cove.numero_cove, cove.cove_descargado, cove.acuse_cove_descargado
|
|
])
|
|
# EDOCs
|
|
for edoc in EDocument.objects.filter(pedimento=ped):
|
|
writer.writerow([
|
|
ped.aduana, ped.patente, ped.regimen, ped.pedimento, ped.pedimento_app,
|
|
ped.clave_pedimento, ped.tipo_operacion_id, ped.contribuyente_id,
|
|
'EDOC', edoc.numero_edocument, edoc.edocument_descargado, edoc.acuse_descargado
|
|
])
|
|
# PARTIDAS
|
|
for partida in Partida.objects.filter(pedimento=ped):
|
|
writer.writerow([
|
|
ped.aduana, ped.patente, ped.regimen, ped.pedimento, ped.pedimento_app,
|
|
ped.clave_pedimento, ped.tipo_operacion_id, ped.contribuyente_id,
|
|
'PARTIDA', partida.numero_partida, partida.descargado, ''
|
|
])
|
|
return response
|
|
# Si se solicita Excel, generar archivo y devolverlo
|
|
if request.query_params.get('excel') == '1':
|
|
import openpyxl
|
|
from openpyxl.utils import get_column_letter
|
|
from django.http import HttpResponse
|
|
wb = openpyxl.Workbook()
|
|
ws = wb.active
|
|
ws.title = "Resumen"
|
|
# Encabezados
|
|
headers = [
|
|
'aduana', 'patente', 'regimen', 'pedimento', 'pedimento_app', 'clave_pedimento',
|
|
'tipo_operacion_id', 'contribuyente_id', 'tipo_documento', 'numero_documento', 'estado', 'acuse_estado'
|
|
]
|
|
ws.append(headers)
|
|
# Llenar filas
|
|
for ped in results:
|
|
for doc in ped['documentos']:
|
|
ws.append([
|
|
ped['aduana'], ped['patente'], ped['regimen'], ped['pedimento'], ped['pedimento_app'],
|
|
ped['clave_pedimento'], ped['tipo_operacion_id'], ped['contribuyente_id'],
|
|
doc.get('tipo'), doc.get('numero'), doc.get('estado'), doc.get('acuse_estado')
|
|
])
|
|
# Ajustar ancho de columnas
|
|
for i, col in enumerate(headers, 1):
|
|
ws.column_dimensions[get_column_letter(i)].width = max(12, len(col) + 2)
|
|
# Guardar en memoria y devolver como respuesta
|
|
from io import BytesIO
|
|
output = BytesIO()
|
|
wb.save(output)
|
|
output.seek(0)
|
|
response = HttpResponse(
|
|
output.read(),
|
|
content_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
|
|
)
|
|
response['Content-Disposition'] = 'attachment; filename=table_summary.xlsx'
|
|
return response
|
|
|
|
# Aplicar paginación manual sobre results
|
|
paginator = CustomPagination()
|
|
page = paginator.paginate_queryset(results, request)
|
|
|
|
return paginator.get_paginated_response({
|
|
"results": page,
|
|
"filtros_aplicados": {
|
|
"organizacion_id": org_id,
|
|
"tipo_documento": tipo_documento,
|
|
"contribuyente__rfc": rfc,
|
|
"fecha_pago__gte": fecha_pago_gte,
|
|
"fecha_pago__lte": fecha_pago_lte,
|
|
"patente": patente,
|
|
"aduana": aduana,
|
|
"pedimento": pedimento,
|
|
"pedimento_app": pedimento_app,
|
|
"regimen": regimen,
|
|
"tipo_operacion": tipo_operacion
|
|
}
|
|
}) |