from django.db import transaction from django.http import HttpResponse from django.shortcuts import get_object_or_404 from django.utils import timezone from rest_framework import viewsets, status from rest_framework.decorators import api_view, permission_classes, action, throttle_classes from rest_framework.parsers import MultiPartParser, FormParser, JSONParser from rest_framework.permissions import AllowAny from rest_framework.response import Response from rest_framework.throttling import AnonRateThrottle from .models import ( Facultad, Carrera, Gestion, Asignatura, Persona, Grupo, EncuestaPlantilla, Pregunta, Token, Respuesta, ) from .serializers import ( FacultadSerializer, CarreraSerializer, GestionSerializer, AsignaturaSerializer, PersonaSerializer, GrupoSerializer, EncuestaPlantillaSerializer, PreguntaSerializer, TokenSerializer, RespuestaSerializer, ) from .permissions import EsPersonalAdministrativo from .services.excel_loader import cargar_excel_grupos, ExcelValidationError from .services.qr_generator import generar_qr_para_grupo, generar_zip_de_grupo, QRGenerationError from .services.reporte_pdf import generar_pdf_resultado_grupo # ============================================================================= # Throttling para endpoints publicos (mitigar abuso/spam de respuestas) # ============================================================================= class EncuestaLecturaThrottle(AnonRateThrottle): scope = "encuesta_lectura" class EncuestaEnvioThrottle(AnonRateThrottle): scope = "encuesta_envio" # ============================================================================= # Navegacion jerarquica: Facultad -> Carrera -> Asignatura -> Grupo # ============================================================================= class FacultadViewSet(viewsets.ModelViewSet): queryset = Facultad.objects.all() serializer_class = FacultadSerializer permission_classes = [EsPersonalAdministrativo] parser_classes = [MultiPartParser, FormParser, JSONParser] class CarreraViewSet(viewsets.ModelViewSet): serializer_class = CarreraSerializer permission_classes = [EsPersonalAdministrativo] parser_classes = [MultiPartParser, FormParser, JSONParser] def get_queryset(self): qs = Carrera.objects.select_related("facultad").all() facultad_id = self.request.query_params.get("facultad") if facultad_id: qs = qs.filter(facultad_id=facultad_id) return qs @action(detail=True, methods=["post"], url_path="carga-excel") def carga_excel(self, request, pk=None): carrera = self.get_object() gestion_id = request.data.get("gestion") archivo = request.FILES.get("archivo") if not gestion_id: return Response({"detail": "Debe indicar la gestion (campo 'gestion')."}, status=status.HTTP_400_BAD_REQUEST) if not archivo: return Response({"detail": "Debe adjuntar el archivo Excel en el campo 'archivo'."}, status=status.HTTP_400_BAD_REQUEST) gestion = get_object_or_404(Gestion, id=gestion_id) try: resumen = cargar_excel_grupos(archivo, carrera, gestion) except ExcelValidationError as e: return Response({"detail": str(e)}, status=status.HTTP_400_BAD_REQUEST) return Response(resumen, status=status.HTTP_201_CREATED) class GestionViewSet(viewsets.ModelViewSet): queryset = Gestion.objects.all() serializer_class = GestionSerializer permission_classes = [EsPersonalAdministrativo] class AsignaturaViewSet(viewsets.ModelViewSet): serializer_class = AsignaturaSerializer permission_classes = [EsPersonalAdministrativo] def get_queryset(self): qs = Asignatura.objects.select_related("carrera").all() carrera_id = self.request.query_params.get("carrera") if carrera_id: qs = qs.filter(carrera_id=carrera_id) return qs class PersonaViewSet(viewsets.ModelViewSet): serializer_class = PersonaSerializer permission_classes = [EsPersonalAdministrativo] def get_queryset(self): qs = Persona.objects.all() rol = self.request.query_params.get("rol") if rol: qs = qs.filter(rol=rol) return qs class EncuestaPlantillaViewSet(viewsets.ModelViewSet): queryset = EncuestaPlantilla.objects.prefetch_related("preguntas").all() serializer_class = EncuestaPlantillaSerializer permission_classes = [EsPersonalAdministrativo] class PreguntaViewSet(viewsets.ModelViewSet): queryset = Pregunta.objects.all() serializer_class = PreguntaSerializer permission_classes = [EsPersonalAdministrativo] class GrupoViewSet(viewsets.ModelViewSet): serializer_class = GrupoSerializer permission_classes = [EsPersonalAdministrativo] def get_queryset(self): qs = Grupo.objects.select_related( "asignatura", "asignatura__carrera", "gestion", "docente", "director" ).all() carrera_id = self.request.query_params.get("carrera") asignatura_id = self.request.query_params.get("asignatura") gestion_id = self.request.query_params.get("gestion") if carrera_id: qs = qs.filter(asignatura__carrera_id=carrera_id) if asignatura_id: qs = qs.filter(asignatura_id=asignatura_id) if gestion_id: qs = qs.filter(gestion_id=gestion_id) return qs @action(detail=True, methods=["post"], url_path="generar-qr") def generar_qr(self, request, pk=None): grupo = self.get_object() regenerar = str(request.data.get("regenerar", "false")).lower() == "true" try: resultado = generar_qr_para_grupo(grupo, regenerar=regenerar) except QRGenerationError as e: return Response({"detail": str(e)}, status=status.HTTP_400_BAD_REQUEST) return Response({ "grupo_id": resultado["grupo_id"], "total_qr": resultado["total_qr"], "mensaje": "QR generados correctamente. Use /descargar-qr/ para obtener el ZIP.", }) @action(detail=True, methods=["get"], url_path="descargar-qr") def descargar_qr(self, request, pk=None): grupo = self.get_object() if not grupo.tokens.exists(): return Response({"detail": "Este grupo aun no tiene QR generados."}, status=status.HTTP_404_NOT_FOUND) zip_buf, nombre_archivo = generar_zip_de_grupo(grupo) response = HttpResponse(zip_buf.read(), content_type="application/zip") response["Content-Disposition"] = f'attachment; filename="{nombre_archivo}"' return response @action(detail=True, methods=["get"], url_path="resultados") def resultados(self, request, pk=None): grupo = self.get_object() return Response(_calcular_resultados_grupo(grupo)) @action(detail=True, methods=["get"], url_path="resultados/descargar-pdf") def resultados_pdf(self, request, pk=None): grupo = self.get_object() resultado = _calcular_resultados_grupo(grupo) pdf_buf = generar_pdf_resultado_grupo(grupo, resultado) response = HttpResponse(pdf_buf.read(), content_type="application/pdf") nombre = f"resultado_{grupo.asignatura.nombre}_{grupo.codigo}.pdf".replace(" ", "_") response["Content-Disposition"] = f'attachment; filename="{nombre}"' return response def _calcular_resultados_grupo(grupo): """Calcula promedios agregados por tipo de encuesta para un solo Grupo.""" respuestas = Respuesta.objects.filter(token__grupo=grupo).select_related("plantilla") resultado = {"grupo_id": grupo.id, "por_tipo": {}} for tipo, _ in EncuestaPlantilla.TIPOS: respuestas_tipo = respuestas.filter(plantilla__tipo=tipo) total_respuestas = respuestas_tipo.count() preguntas_ids_likert = set() for r in respuestas_tipo: for pid_str, valor in r.respuestas.items(): if isinstance(valor, (int, float)): preguntas_ids_likert.add(pid_str) detalle_preguntas = [] suma_promedios = 0 cantidad_preguntas_likert = 0 for pid_str in preguntas_ids_likert: valores = [ r.respuestas[pid_str] for r in respuestas_tipo if pid_str in r.respuestas and isinstance(r.respuestas[pid_str], (int, float)) ] if valores: promedio = sum(valores) / len(valores) try: texto_pregunta = Pregunta.objects.get(id=int(pid_str)).texto except (Pregunta.DoesNotExist, ValueError): texto_pregunta = f"Pregunta {pid_str}" detalle_preguntas.append({ "pregunta_id": pid_str, "texto": texto_pregunta, "promedio": round(promedio, 2), "n": len(valores), }) suma_promedios += promedio cantidad_preguntas_likert += 1 detalle_preguntas.sort(key=lambda p: p["pregunta_id"]) promedio_general = ( round(suma_promedios / cantidad_preguntas_likert, 2) if cantidad_preguntas_likert else None ) resultado["por_tipo"][tipo] = { "total_respuestas": total_respuestas, "promedio_general": promedio_general, "detalle_preguntas": detalle_preguntas, } tokens_grupo = Token.objects.filter(grupo=grupo) resultado["participacion"] = { "tokens_generados": tokens_grupo.count(), "tokens_usados": tokens_grupo.filter(estado=Token.ESTADO_USADO).count(), } return resultado # ============================================================================= # Endpoints PUBLICOS (sin autenticacion) para el flujo de escaneo del QR # ============================================================================= @api_view(["GET"]) @permission_classes([AllowAny]) @throttle_classes([EncuestaLecturaThrottle]) def obtener_encuesta_publica(request, token_valor): token = get_object_or_404(Token, token=token_valor) if token.estado != Token.ESTADO_NO_USADO: return Response( {"detail": "Este codigo QR ya fue utilizado o ha expirado.", "estado": token.estado}, status=status.HTTP_410_GONE, ) # Expiracion automatica por fecha: si la gestion ya cerro, el token deja de servir. if timezone.now().date() > token.grupo.gestion.fecha_fin: token.estado = Token.ESTADO_EXPIRADO token.save(update_fields=["estado"]) return Response( {"detail": "El periodo de evaluacion para este grupo ya ha finalizado."}, status=status.HTTP_410_GONE, ) plantilla = token.plantilla return Response({ "titulo": plantilla.titulo, "tipo": plantilla.tipo, "version": plantilla.version, "docente": token.grupo.docente.nombre, "asignatura": token.grupo.asignatura.nombre, "grupo": token.grupo.codigo, "preguntas": PreguntaSerializer(plantilla.preguntas.all(), many=True).data, }) @api_view(["POST"]) @permission_classes([AllowAny]) @throttle_classes([EncuestaEnvioThrottle]) @transaction.atomic def enviar_respuesta_publica(request, token_valor): # select_for_update bloquea la fila hasta el commit, evitando doble envio # por dos peticiones casi simultaneas con el mismo token. token = get_object_or_404(Token.objects.select_for_update(), token=token_valor) if token.estado != Token.ESTADO_NO_USADO: return Response( {"detail": "Este codigo QR ya fue utilizado o ha expirado."}, status=status.HTTP_410_GONE, ) if timezone.now().date() > token.grupo.gestion.fecha_fin: token.estado = Token.ESTADO_EXPIRADO token.save(update_fields=["estado"]) return Response( {"detail": "El periodo de evaluacion para este grupo ya ha finalizado."}, status=status.HTTP_410_GONE, ) respuestas_data = request.data.get("respuestas") if not isinstance(respuestas_data, dict): return Response({"detail": "El campo 'respuestas' debe ser un objeto {pregunta_id: valor}."}, status=status.HTTP_400_BAD_REQUEST) preguntas_requeridas = token.plantilla.preguntas.filter(requerida=True).values_list("id", flat=True) faltantes = [str(pid) for pid in preguntas_requeridas if str(pid) not in respuestas_data] if faltantes: return Response({"detail": f"Faltan respuestas obligatorias: {faltantes}"}, status=status.HTTP_400_BAD_REQUEST) # Validacion de rango para respuestas tipo likert (defensa en profundidad, # aunque el frontend ya restringe los valores a 1-5). preguntas_likert_ids = set( token.plantilla.preguntas.filter(escala=Pregunta.ESCALA_LIKERT).values_list("id", flat=True) ) for pid_str, valor in respuestas_data.items(): try: pid = int(pid_str) except ValueError: continue if pid in preguntas_likert_ids: if not isinstance(valor, (int, float)) or not (1 <= valor <= 5): return Response({"detail": f"Valor invalido para la pregunta {pid_str}."}, status=status.HTTP_400_BAD_REQUEST) token.estado = Token.ESTADO_USADO token.usado_en = timezone.now() token.save(update_fields=["estado", "usado_en"]) Respuesta.objects.create(token=token, plantilla=token.plantilla, respuestas=respuestas_data) return Response({"detail": "Respuesta registrada correctamente."}, status=status.HTTP_201_CREATED)