from server_backsemot.models import Certificado, Personal, Cliente, Obra, Transaccion, Despachador, ClientePresencial, ControlQr, TicketsCertificado, TicketPresencial, CertificadoPresencial, TicketsCertificadoPresencial, ControlQrPresencial, Comuna, LogConsultaCertificado from server_backsemot.personal.despachador import validaciones_basicas_despachador, buscar_despachador_activo from server_backsemot.personal.personal import validaciones_basicas, mantener_personal_activo from django.core.paginator import Paginator, EmptyPage, PageNotAnInteger from server_backsemot.api.pdf import generar_pdf_con_certificado from server_backsemot.api.brevo import brevo_correo_certificado from django.db.models import Exists, OuterRef, Sum, Count from datetime import datetime, timedelta, date, time from rest_framework.decorators import api_view from django.db.models.functions import Coalesce from django.db.models import Value, FloatField from django.utils.dateparse import parse_date from django.db.models import Count, Sum, Q from server_backsemot.utiles import Utiles from django.db import transaction from django.db.models import Sum from django.conf import settings import traceback import hashlib import secrets import hmac @api_view(["POST"]) def solicitar_certificado_credito_admin(request): try: validacion = validaciones_basicas(validaciones=[1, 2, 3, 4], perfiles=[1, 2, 3], request=request) if validacion != "ok": return Utiles.responder_500(validacion) campos_requeridos = ["cliente", "obra", "fecha_inicio", "fecha_termino"] validacion_campos = validar_campos_requeridos(request, campos_requeridos) if validacion_campos != "ok": return Utiles.responder_500(validacion_campos) token = request.POST["token"] personal = buscar_personal_activo(token) if not personal: return Utiles.responder_500("No existe personal activo con el token indicado") cliente_id = request.POST["cliente"] obra_id = request.POST["obra"] fechai_str = request.POST["fecha_inicio"] fechat_str = request.POST["fecha_termino"] if not Utiles.validar_fecha_formato(fechai_str): return Utiles.responder_500("El formato de la fecha de inicio es incorrecto") if not Utiles.validar_fecha_formato(fechat_str): return Utiles.responder_500("El formato de la fecha de término es incorrecto") fechai = datetime.strptime(fechai_str, "%Y-%m-%d").date() fechat = datetime.strptime(fechat_str, "%Y-%m-%d").date() hoy = date.today() limite = hoy - timedelta(days=366) if fechai < limite or fechat < limite: return Utiles.responder_500("Las fechas no pueden ser mayores a 1 año hacia atrás") if fechai > fechat: return Utiles.responder_500("La fecha de inicio no puede ser mayor a la fecha de término") cliente = Cliente.objects.get(id=cliente_id) obra = Obra.objects.get(id=obra_id) if obra.cliente_id != cliente.id: return Utiles.responder_500("La obra no pertenece al cliente indicado") certificado_solapado = ( Certificado.objects .filter( cliente=cliente, obra=obra, fecha_desde__lte=fechat, fecha_hasta__gte=fechai, activo=1 ) .order_by("fecha_desde") .first() ) if certificado_solapado: return Utiles.responder_500("Ya existe un certificado para este cliente y obra dentro del rango indicado.") inicio_dt = datetime.combine(fechai, time.min) termino_dt = datetime.combine(fechat + timedelta(days=1), time.min) control_qr_usado = ControlQr.objects.filter( qr_id=OuterRef("qr_id"), fecha__gte=inicio_dt, fecha__lt=termino_dt, puesto_control__index=1, ) ticket_ya_certificado = TicketsCertificado.objects.filter( ticket_id=OuterRef("id") ) transacciones = ( Transaccion.objects .filter( cliente=cliente, obra=obra, activo=1, ) .annotate( usada=Exists(control_qr_usado), ya_certificada=Exists(ticket_ya_certificado), ) .filter( usada=True, ya_certificada=False, ) .select_related( "cliente", "obra", "camion", "qr", "estado_pago", "pesaje", ) .order_by("fecha") ) resumen = transacciones.aggregate( total_transacciones=Count("id"), total_m3=Coalesce( Sum("cantidad", filter=Q(unidad__iexact="m3")), Value(0.0), output_field=FloatField() ), total_toneladas=Coalesce( Sum("cantidad", filter=Q(unidad__iexact="toneladas")), Value(0.0), output_field=FloatField() ), ) total_transacciones = resumen["total_transacciones"] or 0 total_m3 = resumen["total_m3"] or 0 total_toneladas = resumen["total_toneladas"] or 0 if total_transacciones == 0 and (total_m3 + total_toneladas) <= 0: return Utiles.responder_500("No existen tickets utilizados para el rango de fechas indicado") data = { "total_transacciones": total_transacciones, "total_m3": total_m3, "total_toneladas": total_toneladas, "fecha_inicio": fechai_str, "fecha_termino": fechat_str, } with transaction.atomic(): c = Certificado() c.cliente = cliente c.obra = obra c.solicita_personal = personal c.fecha_desde = fechai c.fecha_hasta = fechat c.total_cantidad = total_transacciones c.total_m3 = total_m3 c.total_toneladas = total_toneladas c.save() tickets_certificado = [ TicketsCertificado( certificado=c, ticket=transaccion ) for transaccion in transacciones ] TicketsCertificado.objects.bulk_create(tickets_certificado) return Utiles.responder_200(data) except Cliente.DoesNotExist: return Utiles.responder_500("No existe cliente con el id indicado") except Obra.DoesNotExist: return Utiles.responder_500("No existe obra con el id indicado") except ValueError: return Utiles.responder_500("Error al procesar las fechas") except Exception: traceback.print_exc() return Utiles.responder_500("Error interno") @api_view(["POST"]) def solicitar_certificado_credito_cliente(request): try: validacion = validaciones_basicas_despachador(validaciones=[1, 2, 3, 4], perfiles=[1000, 999], request=request) if validacion != "ok": return Utiles.responder_500(validacion) campos_requeridos = ["obra", "fecha_inicio", "fecha_termino"] validacion_campos = validar_campos_requeridos(request, campos_requeridos) if validacion_campos != "ok": return Utiles.responder_500(validacion_campos) token = request.POST["token"] despachador = buscar_despachador_activo(token) if not despachador: return Utiles.responder_500("No existe cuenta de cliente activa con el token indicado") obra_id = request.POST["obra"] fechai_str = request.POST["fecha_inicio"] fechat_str = request.POST["fecha_termino"] if not Utiles.validar_fecha_formato(fechai_str): return Utiles.responder_500("El formato de la fecha de inicio es incorrecto") if not Utiles.validar_fecha_formato(fechat_str): return Utiles.responder_500("El formato de la fecha de término es incorrecto") fechai = datetime.strptime(fechai_str, "%Y-%m-%d").date() fechat = datetime.strptime(fechat_str, "%Y-%m-%d").date() hoy = date.today() limite = hoy - timedelta(days=366) if fechai < limite or fechat < limite: return Utiles.responder_500("Las fechas no pueden ser mayores a 1 año hacia atrás") if fechai > fechat: return Utiles.responder_500("La fecha de inicio no puede ser mayor a la fecha de término") cliente = despachador.cliente obra = Obra.objects.get(id=obra_id) if obra.cliente_id != cliente.id: return Utiles.responder_500("La obra no pertenece al cliente indicado") certificado_solapado = ( Certificado.objects .filter( cliente=cliente, obra=obra, fecha_desde__lte=fechat, fecha_hasta__gte=fechai, activo=1 ) .order_by("fecha_desde") .first() ) if certificado_solapado: return Utiles.responder_500("Ya existe un certificado para este cliente y obra dentro del rango indicado.") # return Utiles.responder_500( # "Ya existe un certificado para este cliente y obra dentro del rango indicado. " # f"Certificado existente: #{certificado_solapado.id}, " # f"desde {certificado_solapado.fecha_desde} hasta {certificado_solapado.fecha_hasta}." # ) inicio_dt = datetime.combine(fechai, time.min) termino_dt = datetime.combine(fechat + timedelta(days=1), time.min) control_qr_usado = ControlQr.objects.filter( qr_id=OuterRef("qr_id"), fecha__gte=inicio_dt, fecha__lt=termino_dt, puesto_control__index=1, ) ticket_ya_certificado = TicketsCertificado.objects.filter( ticket_id=OuterRef("id") ) transacciones = ( Transaccion.objects .filter( cliente=cliente, obra=obra, activo=1, ) .annotate( usada=Exists(control_qr_usado), ya_certificada=Exists(ticket_ya_certificado), ) .filter( usada=True, ya_certificada=False, ) .select_related( "cliente", "obra", "camion", "qr", "estado_pago", "pesaje", ) .order_by("fecha") ) resumen = transacciones.aggregate( total_transacciones=Count("id"), total_m3=Coalesce( Sum("cantidad", filter=Q(unidad__iexact="m3")), Value(0.0), output_field=FloatField() ), total_toneladas=Coalesce( Sum("cantidad", filter=Q(unidad__iexact="toneladas")), Value(0.0), output_field=FloatField() ), ) total_transacciones = resumen["total_transacciones"] or 0 total_m3 = resumen["total_m3"] or 0 total_toneladas = resumen["total_toneladas"] or 0 if total_transacciones == 0 and (total_m3 + total_toneladas) <= 0: return Utiles.responder_500("No existen tickets utilizados para el rango de fechas indicado") data = { "total_transacciones": total_transacciones, "total_m3": total_m3, "total_toneladas": total_toneladas, "fecha_inicio": fechai_str, "fecha_termino": fechat_str, } with transaction.atomic(): c = Certificado() c.cliente = cliente c.obra = obra c.solicita_cliente = despachador c.fecha_desde = fechai c.fecha_hasta = fechat c.total_cantidad = total_transacciones c.total_m3 = total_m3 c.total_toneladas = total_toneladas c.save() tickets_certificado = [ TicketsCertificado( certificado=c, ticket=transaccion ) for transaccion in transacciones ] TicketsCertificado.objects.bulk_create(tickets_certificado) return Utiles.responder_200(data) except Cliente.DoesNotExist: return Utiles.responder_500("No existe cliente con el id indicado") except Obra.DoesNotExist: return Utiles.responder_500("No existe obra con el id indicado") except ValueError: return Utiles.responder_500("Error al procesar las fechas") except Exception: traceback.print_exc() return Utiles.responder_500("Error interno") @api_view(["POST"]) def solicitar_certificado_presencial_admin(request): try: validacion = validaciones_basicas(validaciones=[1, 2, 3, 4], perfiles=[1, 2, 3], request=request) if validacion != "ok": return Utiles.responder_500(validacion) campos_requeridos = ["cliente", "fecha_inicio", "fecha_termino"] validacion_campos = validar_campos_requeridos(request, campos_requeridos) if validacion_campos != "ok": return Utiles.responder_500(validacion_campos) token = request.POST["token"] personal = buscar_personal_activo(token) if not personal: return Utiles.responder_500("No existe personal activo con el token indicado") cliente_id = request.POST["cliente"] fechai_str = request.POST["fecha_inicio"] fechat_str = request.POST["fecha_termino"] direccion = request.POST.get("direccion", "").strip() obra = request.POST.get("obra", "").strip() comuna_id = request.POST.get("comuna") comuna = None if direccion and len(direccion) < 5: return Utiles.responder_500("Debe ingresar una dirección válida") if not Utiles.validar_fecha_formato(fechai_str): return Utiles.responder_500("El formato de la fecha de inicio es incorrecto") if not Utiles.validar_fecha_formato(fechat_str): return Utiles.responder_500("El formato de la fecha de término es incorrecto") fechai = datetime.strptime(fechai_str, "%Y-%m-%d").date() fechat = datetime.strptime(fechat_str, "%Y-%m-%d").date() hoy = date.today() limite = hoy - timedelta(days=366) if fechai < limite or fechat < limite: return Utiles.responder_500("Las fechas no pueden ser mayores a 1 año hacia atrás") if fechai > fechat: return Utiles.responder_500("La fecha de inicio no puede ser mayor a la fecha de término") cliente = ClientePresencial.objects.get(id=cliente_id) certificados_solapados = CertificadoPresencial.objects.filter( cliente=cliente, fecha_desde__lte=fechat, fecha_hasta__gte=fechai, activo=1 ) if obra: certificados_solapados = certificados_solapados.filter( obra__icontains=obra ) else: certificados_solapados = certificados_solapados.filter( obra__isnull=True ) certificado_solapado = certificados_solapados.order_by("fecha_desde").first() if certificado_solapado: obra_msg = f" y obra '{certificado_solapado.obra}'" if certificado_solapado.obra else "" return Utiles.responder_500(f"Ya existe un certificado para este cliente{obra_msg} dentro del rango indicado.") # return Utiles.responder_500( # "Ya existe un certificado presencial para este cliente" # f"{obra_msg} dentro del rango indicado. " # f"Certificado existente: #{certificado_solapado.id}, " # f"desde {certificado_solapado.fecha_desde} hasta {certificado_solapado.fecha_hasta}." # ) inicio_dt = datetime.combine(fechai, time.min) termino_dt = datetime.combine(fechat + timedelta(days=1), time.min) control_qr_usado = ControlQrPresencial.objects.filter( qr_id=OuterRef("qr_id"), fecha__gte=inicio_dt, fecha__lt=termino_dt, puesto_control__index=1, ) ticket_ya_certificado = TicketsCertificadoPresencial.objects.filter( ticket_id=OuterRef("id") ) filtros = { "cliente": cliente, "activo": 1, } if obra: filtros["obra__icontains"] = obra if not comuna_id: return Utiles.responder_500("Debe enviar el id de la comuna") comuna = Comuna.objects.get(id=comuna_id) transacciones = ( TicketPresencial.objects .filter(**filtros) .annotate( usada=Exists(control_qr_usado), ya_certificada=Exists(ticket_ya_certificado), ) .filter( usada=True, ya_certificada=False, ) .select_related( "cliente", "camion", "qr", "estado", "pesaje", ) .order_by("fecha_creacion") ) resumen = transacciones.aggregate( total_transacciones=Count("id"), total_m3=Coalesce( Sum("cantidad", filter=Q(unidad__iexact="m3")), Value(0.0), output_field=FloatField() ), total_toneladas=Coalesce( Sum("cantidad", filter=Q(unidad__iexact="toneladas")), Value(0.0), output_field=FloatField() ), ) total_transacciones = resumen["total_transacciones"] or 0 total_m3 = resumen["total_m3"] or 0 total_toneladas = resumen["total_toneladas"] or 0 if total_transacciones == 0 and (total_m3 + total_toneladas) <= 0: return Utiles.responder_500("No existen tickets utilizados para el rango de fechas indicado") data = { "total_transacciones": total_transacciones, "total_m3": total_m3, "total_toneladas": total_toneladas, "fecha_inicio": fechai_str, "fecha_termino": fechat_str, } with transaction.atomic(): c = CertificadoPresencial() c.cliente = cliente c.solicita_personal = personal c.fecha_desde = fechai c.fecha_hasta = fechat c.total_cantidad = total_transacciones c.total_m3 = total_m3 c.total_toneladas = total_toneladas c.obra = obra if obra else None c.direccion = direccion if direccion else None c.comuna = comuna if comuna else None c.save() tickets_certificado = [ TicketsCertificadoPresencial( certificado=c, ticket=transaccion ) for transaccion in transacciones ] TicketsCertificadoPresencial.objects.bulk_create(tickets_certificado) return Utiles.responder_200(data) except Comuna.DoesNotExist: return Utiles.responder_500("No existe comuna con el id indicado") except ClientePresencial.DoesNotExist: return Utiles.responder_500("No existe cliente con el id indicado") except ValueError: return Utiles.responder_500("Error al procesar las fechas") except Exception: traceback.print_exc() return Utiles.responder_500("Error interno") @api_view(["POST"]) def aprobar_certificado_credito(request): try: validacion = validaciones_basicas(validaciones=[1, 2, 3, 4], perfiles=[1, 2], request=request) if validacion != "ok": return Utiles.responder_500(validacion) campos_requeridos = ["certificado"] validacion_campos = validar_campos_requeridos(request, campos_requeridos) if validacion_campos != "ok": return Utiles.responder_500(validacion_campos) token = request.POST["token"] personal = buscar_personal_activo(token) if not personal: return Utiles.responder_500("No existe personal activo con el token indicado") cert_id = request.POST["certificado"] c = Certificado.objects.get(id=cert_id) codigo = generar_codigo_unico() fecha = datetime.now() validacion = generar_token_validacion_certificado(fecha, codigo, 1) # 1 = Certificado Credito if c.aprobado == 0 and c.aprobado_por is None and c.fecha_aprobacion is None and c.token is None: c.aprobado = 1 c.aprobado_por = personal c.fecha_aprobacion = fecha c.codigo_validacion = codigo c.token = validacion else: return Utiles.responder_500("El certificado ha sido aprobado anteriormente o ya se intento aprobar") if c.aprobado == 1 and c.aprobado != None and c.fecha_aprobacion != None and c.codigo_validacion != None and c.token != None: c.save() if not c.solicita_cliente: return Utiles.responder_200("Certificado aprobado correctamente.") resp, pdf = generar_pdf_con_certificado(c, 1) if resp: bresp, btext = brevo_correo_certificado(c.solicita_cliente.nombre, c.solicita_cliente.correo, pdf) if bresp: return Utiles.responder_200("Certificado aprobado y enviado al correo correctamente.") else: return Utiles.responder_200("Certificado aprobado correctamente.") else: return Utiles.responder_200("Certificado aprobado correctamente.") else: return Utiles.responder_500("Ocurrió un error al aprobar el certificado") except Certificado.DoesNotExist: return Utiles.responder_500("No existe certificado con el id indicado") except Exception: traceback.print_exc() return Utiles.responder_500("Error interno") @api_view(["POST"]) def aprobar_certificado_presencial(request): try: validacion = validaciones_basicas(validaciones=[1, 2, 3, 4], perfiles=[1, 2], request=request) if validacion != "ok": return Utiles.responder_500(validacion) campos_requeridos = ["certificado"] validacion_campos = validar_campos_requeridos(request, campos_requeridos) if validacion_campos != "ok": return Utiles.responder_500(validacion_campos) token = request.POST["token"] personal = buscar_personal_activo(token) if not personal: return Utiles.responder_500("No existe personal activo con el token indicado") cert_id = request.POST["certificado"] c = CertificadoPresencial.objects.get(id=cert_id) codigo = generar_codigo_unico() fecha = datetime.now() validacion = generar_token_validacion_certificado(fecha, codigo, 2) # 2 = Certificado Presencial if c.aprobado == 0 and c.aprobado_por is None and c.fecha_aprobacion is None and c.token is None: c.aprobado = 1 c.aprobado_por = personal c.fecha_aprobacion = fecha c.codigo_validacion = codigo c.token = validacion else: return Utiles.responder_500("El certificado ha sido aprobado anteriormente o ya se intento aprobar") if c.aprobado == 1 and c.aprobado != None and c.fecha_aprobacion != None and c.codigo_validacion != None and c.token != None: c.save() resp, pdf = generar_pdf_con_certificado(c, 2) if resp: bresp, btext = brevo_correo_certificado(c.cliente.nombre, c.cliente.correo, pdf) if bresp: return Utiles.responder_200("Certificado aprobado y enviado al correo correctamente.") else: return Utiles.responder_200("Certificado aprobado correctamente.") else: return Utiles.responder_200("Certificado aprobado correctamente.") else: return Utiles.responder_500("Ocurrió un error al aprobar el certificado") except Certificado.DoesNotExist: return Utiles.responder_500("No existe certificado con el id indicado") except Exception: traceback.print_exc() return Utiles.responder_500("Error interno") def generar_codigo_unico(length=10): try: alphabet = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" while True: codigo = "".join(secrets.choice(alphabet) for _ in range(length)) existe = ( Certificado.objects.filter( codigo_validacion=codigo ).exists() or CertificadoPresencial.objects.filter( codigo_validacion=codigo ).exists() ) if not existe: return codigo except: traceback.print_exc() @api_view(["GET"]) def consultar_certificado_publico(request): try: campos_requeridos = ["numero", "codigo"] validacion_campos = validar_campos_requeridos(request, campos_requeridos) if validacion_campos != "ok": return Utiles.responder_500(validacion_campos) numero = request.GET["numero"] codigo = request.GET["codigo"] rut = request.GET.get("rut", "").strip() if not rut: return Utiles.responder_500("Debe enviar el rut") ip = None x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR') if x_forwarded_for: # Podría haber múltiples direcciones IP en el encabezado (proxies), tomar la primera ip = x_forwarded_for.split(',')[0].strip() else: ip = request.META.get('REMOTE_ADDR') # IP directa sin pasar por proxy lcc = LogConsultaCertificado() lcc.ip = ip lcc.rut = rut lcc.numero = numero lcc.codigo = codigo lcc.save() try: numero = int(numero) except ValueError: lcc.resultado = "Número no válido" lcc.save() return Utiles.responder_500("Debe ingresar un número válido") if not Utiles.validate_rut(rut): lcc.resultado = "Rut no válido" lcc.save() return Utiles.responder_500("Debe ingresar un rut válido") cert = Certificado.objects.filter(codigo_validacion=codigo).first() if not cert: cert = CertificadoPresencial.objects.filter(codigo_validacion=codigo).first() tipo = 2 else: tipo = 1 if not cert: lcc.resultado = "No se encontró un certificado válido con la información ingresada. Verifique los datos e intente nuevamente. # 1" lcc.save() return Utiles.responder_500("No se encontró un certificado válido con la información ingresada. Verifique los datos e intente nuevamente.") # 1 if cert.id == numero: if cert.cliente.rut != rut: lcc.resultado = "No se encontró un certificado válido con la información ingresada. Verifique los datos e intente nuevamente. # 2" lcc.save() return Utiles.responder_500("No se encontró un certificado válido con la información ingresada. Verifique los datos e intente nuevamente.") # 2 token = generar_token_validacion_certificado(cert.fecha_aprobacion, cert.codigo_validacion, tipo) if token is not None and cert.token == token: if cert.activo != 1: lcc.resultado = "El certificado se encuentra anulado" lcc.save() return Utiles.responder_500("El certificado se encuentra anulado") resp, pdf = generar_pdf_con_certificado(cert, tipo) if resp: lcc.resultado = "Se entrego el certificado para ser visualizado" lcc.save() return Utiles.responder_200_archivo(pdf, "certificado" + str(cert.id) + ".pdf") else: lcc.resultado = f"{pdf} #pdf" lcc.save() return Utiles.responder_500(pdf) else: lcc.resultado = "No se encontró un certificado válido con la información ingresada. Verifique los datos e intente nuevamente. # 3" lcc.save() return Utiles.responder_500("No se encontró un certificado válido con la información ingresada. Verifique los datos e intente nuevamente.") # 3 else: lcc.resultado = "No se encontró un certificado válido con la información ingresada. Verifique los datos e intente nuevamente. # 4" lcc.save() return Utiles.responder_500("No se encontró un certificado válido con la información ingresada. Verifique los datos e intente nuevamente.") # 5 except Exception: lcc.resultado = "Error interno" lcc.save() traceback.print_exc() return Utiles.responder_500("Error interno") @api_view(["GET"]) def consultar_certificado_interno_admin(request): try: validacion = validaciones_basicas(validaciones=[1, 2, 3, 4], perfiles=[1, 2, 3], request=request) if validacion != "ok": return Utiles.responder_500(validacion) campos_requeridos = ["certificado", "tipo"] validacion_campos = validar_campos_requeridos(request, campos_requeridos) if validacion_campos != "ok": return Utiles.responder_500(validacion_campos) cert_id = request.GET["certificado"] tipo = request.GET["tipo"] try: tipo = int(tipo) except ValueError: return Utiles.responder_500("Tipo debe ser un número") if tipo != 1 and tipo != 2: return Utiles.responder_500("Valor de tipo incorrecto") if tipo == 1: cert = Certificado.objects.filter(id=cert_id).first() elif tipo == 2: cert = CertificadoPresencial.objects.filter(id=cert_id).first() if not cert: return Utiles.responder_500("No existe certificado con el id indicado") if cert.activo != 1: return Utiles.responder_500("El certificado se encuentra anulado") resp, pdf = generar_pdf_con_certificado(cert, tipo) if resp: return Utiles.responder_200_archivo(pdf, "certificado" + str(cert.id) + ".pdf") except Exception: traceback.print_exc() return Utiles.responder_500("Error interno") @api_view(["GET"]) def consultar_certificado_interno_cliente(request): try: validacion = validaciones_basicas_despachador(validaciones=[1, 2, 3, 4], perfiles=[1000, 999], request=request) if validacion != "ok": return Utiles.responder_500(validacion) campos_requeridos = ["certificado"] validacion_campos = validar_campos_requeridos(request, campos_requeridos) if validacion_campos != "ok": return Utiles.responder_500(validacion_campos) cert_id = request.GET["certificado"] cert = Certificado.objects.filter(id=cert_id).first() if not cert: return Utiles.responder_500("No existe certificado con el id indicado") token = request.GET["token"] despachador = buscar_despachador_activo(token) if not despachador: return Utiles.responder_500("Debe iniciar sesión") if despachador.cliente.id != cert.cliente.id: return Utiles.responder_500("No tiene permiso para ver el certificado") if cert.activo != 1: return Utiles.responder_500("El certificado se encuentra anulado") resp, pdf = generar_pdf_con_certificado(cert, 1) if resp: return Utiles.responder_200_archivo(pdf, "certificado" + str(cert.id) + ".pdf") except Exception: traceback.print_exc() return Utiles.responder_500("Error interno") @api_view(["GET"]) def consultar_certificados_disponibles_admin(request): try: validacion = validaciones_basicas( validaciones=[1, 2, 3, 4], perfiles=[1, 2, 3], request=request ) if validacion != "ok": return Utiles.responder_500(validacion) cliente_id = request.GET.get("cliente", "").strip() obra_id = request.GET.get("obra", "").strip() fecha_inicio = request.GET.get("fecha_inicio", "").strip() fecha_termino = request.GET.get("fecha_termino", "").strip() tipo = request.GET.get("tipo", "").strip() if not cliente_id or not tipo: return Utiles.responder_500("Cliente y tipo son obligatorios") if bool(fecha_inicio) != bool(fecha_termino): return Utiles.responder_500( "Debe enviar fecha inicio y fecha término juntas" ) try: tipo = int(tipo) except ValueError: return Utiles.responder_500("Tipo debe ser un número") if tipo not in [1, 2]: return Utiles.responder_500("Valor de tipo no válido") if fecha_inicio and fecha_termino: try: fecha_inicio = datetime.strptime( fecha_inicio, "%Y-%m-%d" ).date() fecha_termino = datetime.strptime( fecha_termino, "%Y-%m-%d" ).date() except ValueError: return Utiles.responder_500( "Formato de fecha inválido. Use YYYY-MM-DD" ) if fecha_inicio > fecha_termino: return Utiles.responder_500( "La fecha de inicio no puede ser mayor a la fecha término" ) if (fecha_termino - fecha_inicio).days > 366: return Utiles.responder_500( "El rango de fechas no puede ser superior a un año" ) else: fecha_termino = date.today() fecha_inicio = fecha_termino - timedelta(days=366) if tipo == 1: cliente = Cliente.objects.get(id=cliente_id) else: cliente = ClientePresencial.objects.get(id=cliente_id) obra = None obra_nombre_respuesta = None filtros = { "cliente": cliente, "activo": 1, } if obra_id: if tipo == 1: obra = Obra.objects.get(id=obra_id) filtros["obra"] = obra obra_nombre_respuesta = obra.nombre else: filtros["obra__icontains"] = obra_id obra_nombre_respuesta = obra_id datos = None if tipo == 1: certificados = ( Certificado.objects .filter( **filtros, fecha_desde__lte=fecha_termino, fecha_hasta__gte=fecha_inicio, ) .order_by("fecha_desde") ) datos = [ formatear_certificado_credito(c) for c in certificados ] else: certificados = ( CertificadoPresencial.objects .filter( **filtros, fecha_desde__lte=fecha_termino, fecha_hasta__gte=fecha_inicio, ) .order_by("fecha_desde") ) datos = [ formatear_certificado_presencial(c) for c in certificados ] # Si hay certificados con los datos indicados, se envian los certificados if datos: return Utiles.responder_500(datos) # Si no hay certificados con los datos indicados, se revisa que existan tickets para generar un certificado con esos datos if tipo == 1: inicio_dt = datetime.combine(fecha_inicio, time.min) termino_dt = datetime.combine(fecha_termino + timedelta(days=1), time.min) control_qr_usado = ControlQr.objects.filter( qr_id=OuterRef("qr_id"), fecha__gte=inicio_dt, fecha__lt=termino_dt, puesto_control__index=1, ) ticket_ya_certificado = TicketsCertificado.objects.filter( ticket_id=OuterRef("id") ) transacciones = ( Transaccion.objects .filter( cliente=cliente, obra=obra, activo=1, ) .annotate( usada=Exists(control_qr_usado), ya_certificada=Exists(ticket_ya_certificado), ) .filter( usada=True, ya_certificada=False, ) .select_related( "cliente", "obra", "camion", "qr", "estado_pago", "pesaje", ) .order_by("fecha") ) resumen = transacciones.aggregate( total_transacciones=Count("id"), total_m3=Coalesce( Sum("cantidad", filter=Q(unidad__iexact="m3")), Value(0.0), output_field=FloatField() ), total_toneladas=Coalesce( Sum("cantidad", filter=Q(unidad__iexact="toneladas")), Value(0.0), output_field=FloatField() ), ) total_transacciones = resumen["total_transacciones"] or 0 total_m3 = resumen["total_m3"] or 0 total_toneladas = resumen["total_toneladas"] or 0 if total_transacciones == 0 and (total_m3 + total_toneladas) <= 0: return Utiles.responder_500("No existen tickets utilizados para los filtros seleccionados") if tipo == 2: inicio_dt = datetime.combine(fecha_inicio, time.min) termino_dt = datetime.combine(fecha_termino + timedelta(days=1), time.min) control_qr_usado = ControlQrPresencial.objects.filter( qr_id=OuterRef("qr_id"), fecha__gte=inicio_dt, fecha__lt=termino_dt, puesto_control__index=1, ) ticket_ya_certificado = TicketsCertificadoPresencial.objects.filter( ticket_id=OuterRef("id") ) filtros = { "cliente": cliente, "activo": 1, } if obra_id: filtros["obra__icontains"] = obra_id transacciones = ( TicketPresencial.objects .filter(**filtros) .annotate( usada=Exists(control_qr_usado), ya_certificada=Exists(ticket_ya_certificado), ) .filter( usada=True, ya_certificada=False, ) .select_related( "cliente", "camion", "qr", "estado", "pesaje", ) .order_by("fecha_creacion") ) resumen = transacciones.aggregate( total_transacciones=Count("id"), total_m3=Coalesce( Sum("cantidad", filter=Q(unidad__iexact="m3")), Value(0.0), output_field=FloatField() ), total_toneladas=Coalesce( Sum("cantidad", filter=Q(unidad__iexact="toneladas")), Value(0.0), output_field=FloatField() ), ) total_transacciones = resumen["total_transacciones"] or 0 total_m3 = resumen["total_m3"] or 0 total_toneladas = resumen["total_toneladas"] or 0 if total_transacciones == 0 and (total_m3 + total_toneladas) <= 0: return Utiles.responder_500("No existen tickets utilizados para los filtros seleccionados") datos = { "cliente": cliente.nombre, "rut": cliente.rut, "obra": obra_nombre_respuesta, "m3": total_m3, "toneladsa": total_toneladas } return Utiles.responder_200("Existen tickets disponibles para generar un certificado", datos) except Obra.DoesNotExist: return Utiles.responder_500("No existe obra con el id indicado") except Cliente.DoesNotExist: return Utiles.responder_500("No existe cliente con el id indicado") except ClientePresencial.DoesNotExist: return Utiles.responder_500("No existe cliente con el id indicado") except Exception: traceback.print_exc() return Utiles.responder_500("Error interno") @api_view(["GET"]) def consultar_certificados_disponibles_cliente(request): try: validacion = validaciones_basicas_despachador(validaciones=[1, 2, 3, 4], perfiles=[1000, 999], request=request) if validacion != "ok": return Utiles.responder_500(validacion) obra_id = request.GET.get("obra", "").strip() fecha_inicio = request.GET.get("fecha_inicio", "").strip() fecha_termino = request.GET.get("fecha_termino", "").strip() token = request.GET.get("token", "").strip() if not token: return Utiles.responder_500("No existe usuario con el token indicado") despachador = buscar_despachador_activo(token) if not despachador: return Utiles.responder_500("No existe usuario con el token indicado") cliente = despachador.cliente if bool(fecha_inicio) != bool(fecha_termino): return Utiles.responder_500( "Debe enviar fecha inicio y fecha término juntas" ) if fecha_inicio and fecha_termino: try: fecha_inicio = datetime.strptime( fecha_inicio, "%Y-%m-%d" ).date() fecha_termino = datetime.strptime( fecha_termino, "%Y-%m-%d" ).date() except ValueError: return Utiles.responder_500( "Formato de fecha inválido. Use YYYY-MM-DD" ) if fecha_inicio > fecha_termino: return Utiles.responder_500( "La fecha de inicio no puede ser mayor a la fecha término" ) if (fecha_termino - fecha_inicio).days > 366: return Utiles.responder_500( "El rango de fechas no puede ser superior a un año" ) else: fecha_termino = date.today() fecha_inicio = fecha_termino - timedelta(days=366) filtros = { "cliente": cliente, "activo": 1, } if obra_id: obra = Obra.objects.get(id=obra_id) filtros["obra"] = obra datos = None certificados = ( Certificado.objects .filter( **filtros, fecha_desde__lte=fecha_termino, fecha_hasta__gte=fecha_inicio, ) .order_by("fecha_desde") ) datos = [ formatear_certificado_credito(c) for c in certificados ] # Si hay certificados con los datos indicados, se envian los certificados if datos: return Utiles.responder_500(datos) # Si no hay certificados con los datos indicados, se revisa que existan tickets para generar un certificado con esos datos inicio_dt = datetime.combine(fecha_inicio, time.min) termino_dt = datetime.combine(fecha_termino + timedelta(days=1), time.min) control_qr_usado = ControlQr.objects.filter( qr_id=OuterRef("qr_id"), fecha__gte=inicio_dt, fecha__lt=termino_dt, puesto_control__index=1, ) ticket_ya_certificado = TicketsCertificado.objects.filter( ticket_id=OuterRef("id") ) transacciones = ( Transaccion.objects .filter( cliente=cliente, obra=obra, activo=1, ) .annotate( usada=Exists(control_qr_usado), ya_certificada=Exists(ticket_ya_certificado), ) .filter( usada=True, ya_certificada=False, ) .select_related( "cliente", "obra", "camion", "qr", "estado_pago", "pesaje", ) .order_by("fecha") ) resumen = transacciones.aggregate( total_transacciones=Count("id"), total_m3=Coalesce( Sum("cantidad", filter=Q(unidad__iexact="m3")), Value(0.0), output_field=FloatField() ), total_toneladas=Coalesce( Sum("cantidad", filter=Q(unidad__iexact="toneladas")), Value(0.0), output_field=FloatField() ), ) total_transacciones = resumen["total_transacciones"] or 0 total_m3 = resumen["total_m3"] or 0 total_toneladas = resumen["total_toneladas"] or 0 if total_transacciones == 0 and (total_m3 + total_toneladas) <= 0: return Utiles.responder_500("No existen tickets utilizados para los filtros seleccionados") datos = { "cliente": cliente.nombre, "rut": cliente.rut, "obra": obra.nombre, "m3": total_m3, "toneladsa": total_toneladas } return Utiles.responder_200("Existen tickets disponibles para generar un certificado", datos) except Obra.DoesNotExist: return Utiles.responder_500("No existe obra con el id indicado") except Cliente.DoesNotExist: return Utiles.responder_500("No existe cliente con el id indicado") except ClientePresencial.DoesNotExist: return Utiles.responder_500("No existe cliente con el id indicado") except Exception: traceback.print_exc() return Utiles.responder_500("Error interno") @api_view(["GET"]) def ver_certificados(request): try: validacion = validaciones_basicas(validaciones=[1, 2, 3, 4], perfiles=[1, 2, 3], request=request) if validacion != "ok": return Utiles.responder_500(validacion) texto = request.GET.get("texto") tipo = request.GET.get("tipo") if not tipo: return Utiles.responder_500("Falta tipo") try: tipo = int(tipo) except ValueError: return Utiles.responder_500("El valor de tipo debe ser 1 o 2") if tipo != 1 and tipo != 2: return Utiles.responder_500("El valor de tipo debe ser 1 o 2") if texto: if tipo == 1: filtros = ( Q(cliente__rut__icontains=texto) | Q(cliente__nombre__icontains=texto) | Q(obra__nombre__icontains=texto) ) if tipo == 2: filtros = ( Q(cliente__rut__icontains=texto) | Q(cliente__nombre__icontains=texto) | Q(obra__icontains=texto) ) if str(texto).isdigit(): filtros |= Q(id=int(texto)) if tipo == 1: certificados = ( Certificado.objects .filter(filtros) .order_by("-fecha_solicitud", "-id") ) if tipo == 2: certificados = ( CertificadoPresencial.objects .filter(filtros) .order_by("-fecha_solicitud", "-id") ) else: if tipo == 1: certificados = ( Certificado.objects .all() .order_by("-fecha_solicitud", "-id") ) if tipo == 2: certificados = ( CertificadoPresencial.objects .all() .order_by("-fecha_solicitud", "-id") ) # Paginación page = request.GET.get('page', 1) # Página solicitada, por defecto es 1 paginator = Paginator(certificados, 20) # 20 elementos por página try: paginated_certificados = paginator.page(page) except PageNotAnInteger: paginated_certificados = paginator.page(1) except EmptyPage: paginated_certificados = paginator.page(paginator.num_pages) # Datos de paginación total_items = paginator.count total_pages = paginator.num_pages current_page = paginated_certificados.number start_index = paginated_certificados.start_index() end_index = paginated_certificados.end_index() # Preparar la respuesta con los datos paginados if tipo == 1: datos = [formatear_certificado_credito(certificado) for certificado in paginated_certificados] if tipo == 2: datos = [formatear_certificado_presencial(certificado) for certificado in paginated_certificados] # Respuesta final incluyendo la información de paginación respuesta = { "mensaje": "Listado facturas", "datos": datos, "paginacion": { "pagina_actual": current_page, "total_paginas": total_pages, "total_elementos": total_items, "elementos_pagina_actual": f"{start_index} al {end_index}" } } mantener_personal_activo(request.GET.get("token")) return Utiles.responder_200("Listado de facturas", respuesta) except Exception: traceback.print_exc() return Utiles.responder_500("Error interno") @api_view(["GET"]) def ver_certificados_cliente(request): try: validacion = validaciones_basicas_despachador( validaciones=[1, 2, 3, 4], perfiles=[1000, 999], request=request ) if validacion != "ok": return Utiles.responder_500(validacion) token = request.GET.get("token") if not token: return Utiles.responder_500("Debe enviar el token de usuario") despachador = buscar_despachador_activo(token) if not despachador: return Utiles.responder_500("No existe usuario activo con el token indicado") cliente = getattr(despachador, "cliente", None) if not cliente: return Utiles.responder_500("El usuario no tiene cliente asociado") fecha_inicio = request.GET.get("fecha_inicio") or None fecha_termino = request.GET.get("fecha_termino") or None if bool(fecha_inicio) != bool(fecha_termino): return Utiles.responder_500( "Para filtrar por fecha, debe enviar ambas fechas" ) estado = request.GET.get("estado") if estado is not None: try: estado = int(estado) except ValueError: return Utiles.responder_500("Estado inválido") if estado not in [-1, 0, 1]: return Utiles.responder_500("Estado inválido") certificados = Certificado.objects.filter( cliente=cliente ) if fecha_inicio and fecha_termino: fecha_inicio_parseada = parse_date(fecha_inicio) fecha_termino_parseada = parse_date(fecha_termino) if not fecha_inicio_parseada or not fecha_termino_parseada: return Utiles.responder_500( "Las fechas deben tener formato válido YYYY-MM-DD" ) if fecha_inicio_parseada > fecha_termino_parseada: return Utiles.responder_500( "La fecha de inicio no puede ser mayor a la fecha de término" ) certificados = certificados.filter( fecha_solicitud__date__range=[ fecha_inicio_parseada, fecha_termino_parseada ] ) if estado is not None: if estado == 1 or estado == 0: certificados = certificados.filter( aprobado=estado ) if estado == -1: certificados = certificados.filter( activo=0 ) certificados = certificados.order_by( "-fecha_solicitud", "-id" ) # Paginación page = request.GET.get("page", 1) paginator = Paginator(certificados, 20) try: paginated_certificados = paginator.page(page) except PageNotAnInteger: paginated_certificados = paginator.page(1) except EmptyPage: paginated_certificados = paginator.page(paginator.num_pages) total_items = paginator.count total_pages = paginator.num_pages current_page = paginated_certificados.number start_index = paginated_certificados.start_index() end_index = paginated_certificados.end_index() datos = [ formatear_certificado_credito(certificado) for certificado in paginated_certificados ] respuesta = { "mensaje": "Listado facturas", "datos": datos, "paginacion": { "pagina_actual": current_page, "total_paginas": total_pages, "total_elementos": total_items, "elementos_pagina_actual": f"{start_index} al {end_index}" } } mantener_personal_activo(token) return Utiles.responder_200("Listado de facturas", respuesta) except Exception: traceback.print_exc() return Utiles.responder_500("Error interno") @api_view(["POST"]) def anular_certificado(request): try: validacion = validaciones_basicas(validaciones=[1, 2, 3, 4], perfiles=[1, 2], request=request) if validacion != "ok": return Utiles.responder_500(validacion) cert_id = request.POST.get("certificado") tipo = request.POST.get("tipo") liberar = request.POST.get("liberar") accion = request.POST.get("accion") # 1 = anular, 2 = rechazar if not cert_id or not tipo or not liberar: return Utiles.responder_500("Debe enviar el certificado, tipo y liberar") try: accion = int(accion) except ValueError: return Utiles.responder_500("El valor de acción debe ser un número") if accion != 1 and accion != 2: return Utiles.responder_500("Valor de acción incorrecto") try: cert_id = int(cert_id) except Value: return Utiles.responder_500("El id de certificado debe ser un número") try: liberar = int(liberar) except ValueError: return Utiles.responder_500("El valor de liberar debe ser un número") if liberar != 1 and liberar != 0: return Utiles.responder_500("El valor de liberar debe ser un número") try: tipo = int(tipo) except ValueError: return Utiles.responder_500("El valor de tipo debe ser un número") cert = None if tipo == 1: cert = Certificado.objects.filter(id=cert_id).first() if tipo == 2: cert = CertificadoPresencial.objects.filter(id=cert_id).first() if not cert: return Utiles.responder_500("No existe certificado con el id indicado") if accion == 2: cert.aprobado = -1 cert.activo = 0 cert.save() if liberar == 1: if tipo == 1: TicketsCertificado.objects.filter(certificado=cert).delete() if tipo == 2: TicketsCertificadoPresencial.objects.filter(certificado=cert).delete() if accion == 1: return Utiles.responder_200("Se ha anulado el certificado de forma correcta") if accion == 2: return Utiles.responder_200("Se ha rechazado el certificado de forma correcta") except Exception: traceback.print_exc() return Utiles.responder_500("Error interno") def formatear_certificado_credito(c: Certificado): datos = { "id": c.id, "cliente": c.cliente.nombre, "cliente_rut": c.cliente.rut, "obra": c.obra.nombre, "fecha_desde": c.fecha_desde, "fecha_hasta": c.fecha_hasta, "total_m3": c.total_m3, "total_toneladas": c.total_toneladas, "fecha_solicitud": c.fecha_solicitud, "aprobado": c.aprobado, "aprobado_por": c.aprobado_por.nombre if c and c.aprobado_por and c.aprobado_por.nombre else None, "fecha_aprobacion": c.fecha_aprobacion, "activo": c.activo } return datos def formatear_certificado_presencial(c: CertificadoPresencial): datos = { "id": c.id, "cliente": c.cliente.nombre, "cliente_rut": c.cliente.rut, "obra": c.obra if c and c.obra else None, "fecha_desde": c.fecha_desde, "fecha_hasta": c.fecha_hasta, "total_m3": c.total_m3, "total_toneladas": c.total_toneladas, "fecha_solicitud": c.fecha_solicitud, "aprobado": c.aprobado, "aprobado_por": c.aprobado_por.nombre if c and c.aprobado_por and c.aprobado_por.nombre else None, "fecha_aprobacion": c.fecha_aprobacion, "activo": c.activo } return datos def generar_token_validacion_certificado(fecha, codigo_validacion, tipo): try: fecha_base = fecha.strftime("%Y%m%d%H%M%S") if tipo == 1: payload = f"CC:{fecha_base}:{codigo_validacion}" else: payload = f"CP:{fecha_base}:{codigo_validacion}" return hmac.new( settings.SECRET_KEY.encode("utf-8"), payload.encode("utf-8"), hashlib.sha512 ).hexdigest() except: traceback.print_exc() def buscar_personal_activo(token): personal = Personal.objects.filter(token=token, estado=1).first() return personal def buscar_despachador_activo(token): despachador = Despachador.objects.filter(token=token, estado=1).first() return despachador def validar_campos_requeridos(request, campos): # Valida que los campos requeridos estén presentes y no sean nulos o vacíos en la solicitud (POST o GET) for campo in campos: valor_post = request.POST.get(campo, "").strip() valor_get = request.GET.get(campo, "").strip() if not valor_post and not valor_get: return f"Falta {campo}" return "ok"