""" gamification/engine.py ======================= Motor central de gamificación. Este es el ÚNICO lugar donde se otorgan puntos, fichas e insignias. Las vistas y consumers NUNCA modifican gamificación directamente; siempre llaman a GamificationEngine. Diseño: - Stateless: no guarda estado interno, opera sobre la BD - Transaccional: cada operación es atómica con select_for_update - Asíncrono-safe: los métodos que emiten a WebSocket se marcan con _notify_* y pueden envolverse en async context si se necesita Flujo típico de una respuesta correcta: 1. sessions/views.py recibe POST /respuestas/ 2. Valida y guarda la Respuesta 3. Llama engine.procesar_respuesta(respuesta) 4. engine calcula puntos → agregar_puntos() → ficha si corresponde 5. engine evalúa insignias → desbloquear si cumple criterio 6. engine retorna dict con puntos, fichas_nuevas, insignias 7. La vista notifica al alumno por WebSocket con ese dict """ import logging from datetime import timedelta, date from typing import NamedTuple from django.db import transaction from django.db.models import F from django.conf import settings logger = logging.getLogger('eduplay') # ── Resultado estructurado que retorna el engine ── class ResultadoProcesamiento(NamedTuple): puntos_ganados: int # Puntos de esta respuesta fichas_nuevas: int # Fichas obtenidas (0 o más) es_correcta: bool insignias_nuevas: list # Lista de dicts con datos de insignias desbloqueadas nueva_racha: int # Racha actual tras esta sesión class GamificationEngine: """ Motor de gamificación reutilizable. Se instancia una vez por request o se usa como clase estática. """ # Puntos según velocidad de respuesta (se aplica solo a correctas) PUNTOS_RAPIDO = 100 # Respondió en el primer 30% del tiempo PUNTOS_MEDIO = 75 # Respondió en el 30–70% del tiempo PUNTOS_LENTO = 50 # Respondió en el 70–100% del tiempo # Puntos por participar en reflexiones (confuso/minuto) sin importar contenido PUNTOS_REFLEXION = 10 # ────────────────────────────────────────────────────────────── # PUNTO DE ENTRADA PRINCIPAL # ────────────────────────────────────────────────────────────── @classmethod @transaction.atomic def procesar_respuesta(cls, respuesta) -> ResultadoProcesamiento: """ Procesa una respuesta recién guardada: 1. Calcula puntos según velocidad y corrección 2. Actualiza el acumulado de Puntos del alumno 3. Evalúa y desbloquea insignias si corresponde 4. Retorna el resultado para notificar al alumno Args: respuesta: instancia de sessions.models.Respuesta Returns: ResultadoProcesamiento con todo lo ganado en esta respuesta """ from gamification.models import Puntos pregunta_activa = respuesta.pregunta_activa sesion = pregunta_activa.sesion alumno = respuesta.alumno seccion = sesion.seccion # ── 1. Calcular puntos ── puntos = cls._calcular_puntos(respuesta) # ── 2. Guardar puntos en Respuesta ── respuesta.puntos = puntos respuesta.save(update_fields=['puntos']) # ── 3. Actualizar acumulado con bloqueo de fila ── puntos_obj, _ = Puntos.objects.select_for_update().get_or_create( alumno=alumno, seccion=seccion, temporada=sesion.temporada, defaults={'total_puntos': 0, 'total_fichas': 0} ) fichas_nuevas = puntos_obj.agregar_puntos(puntos) # ── 4. Evaluar y desbloquear insignias ── insignias_nuevas = cls._evaluar_insignias( alumno, seccion, puntos_obj, respuesta ) logger.info( f'Gamificación: alumno={alumno.id} sesion={sesion.id} ' f'pts={puntos} fichas={fichas_nuevas} insignias={len(insignias_nuevas)}' ) return ResultadoProcesamiento( puntos_ganados=puntos, fichas_nuevas=fichas_nuevas, es_correcta=respuesta.es_correcta, insignias_nuevas=insignias_nuevas, nueva_racha=puntos_obj.racha_sesiones, ) # ────────────────────────────────────────────────────────────── # CÁLCULO DE PUNTOS # ────────────────────────────────────────────────────────────── @classmethod def _calcular_puntos(cls, respuesta) -> int: """ Calcula los puntos de una respuesta según: - Tipo de pregunta (reflexiones = puntos fijos) - Corrección (incorrecta = 0 puntos) - Velocidad de respuesta (proporcional al tiempo total) El tiempo de respuesta se calcula en el servidor (respuesta.tiempo_ms), no puede ser manipulado por el cliente. """ pregunta = respuesta.pregunta_activa.pregunta # Reflexiones no tienen respuesta correcta/incorrecta → puntos fijos if pregunta.tipo in ('confuso', 'minuto'): return cls.PUNTOS_REFLEXION if respuesta.texto_respuesta.strip() else 0 # Respuesta incorrecta → 0 puntos if not respuesta.es_correcta: return 0 # Calcular porcentaje de tiempo usado tiempo_total_ms = pregunta.tiempo_seg * 1000 tiempo_usado_ms = respuesta.tiempo_ms if tiempo_total_ms <= 0: return cls.PUNTOS_LENTO ratio = tiempo_usado_ms / tiempo_total_ms # Velocidad rápida: primer 30% del tiempo disponible if ratio <= 0.30: return cls.PUNTOS_RAPIDO # Velocidad media: entre 30% y 70% if ratio <= 0.70: return cls.PUNTOS_MEDIO # Velocidad lenta: más del 70% return cls.PUNTOS_LENTO # ────────────────────────────────────────────────────────────── # EVALUACIÓN DE INSIGNIAS # ────────────────────────────────────────────────────────────── @classmethod def _evaluar_insignias(cls, alumno, seccion, puntos_obj, respuesta) -> list: """ Evalúa TODOS los criterios de insignia aplicables tras una respuesta. Solo desbloquea las que el alumno no tiene aún. Returns: Lista de dicts con datos de cada insignia nueva """ from gamification.models import Insignia, InsigniaAlumno from sessions.models import Respuesta as RespuestaModel # Insignias ya obtenidas en esta sección (para no evaluar de nuevo) ya_tiene = set(InsigniaAlumno.objects.filter( alumno=alumno, seccion=seccion ).values_list('insignia__codigo', flat=True)) nuevas = [] # ── Criterios a evaluar ── criterios = cls._build_criterios(alumno, seccion, puntos_obj, respuesta) for codigo_insignia, cumple in criterios.items(): if codigo_insignia in ya_tiene or not cumple: continue try: insignia = Insignia.objects.get(codigo=codigo_insignia) InsigniaAlumno.objects.create( alumno=alumno, insignia=insignia, seccion=seccion ) nuevas.append({ 'codigo': insignia.codigo, 'emoji': insignia.emoji, 'nombre': insignia.nombre, 'descripcion': insignia.descripcion, 'nivel': insignia.nivel, }) logger.info(f'Insignia desbloqueada: {codigo_insignia} alumno={alumno.id}') except Insignia.DoesNotExist: logger.warning(f'Insignia no encontrada en fixtures: {codigo_insignia}') return nuevas @classmethod def _build_criterios(cls, alumno, seccion, puntos_obj, respuesta) -> dict: """ Construye un dict {codigo_insignia: bool} con todos los criterios. True = el alumno cumple el criterio ahora. Separado de _evaluar_insignias para facilitar tests unitarios sin tocar la BD de insignias. """ from sessions.models import Respuesta as RespuestaModel # Estadísticas del alumno en esta sección total_respuestas = RespuestaModel.objects.filter( alumno=alumno, pregunta_activa__sesion__seccion=seccion ).count() total_correctas = RespuestaModel.objects.filter( alumno=alumno, pregunta_activa__sesion__seccion=seccion, es_correcta=True ).count() es_correcta = respuesta.es_correcta or False tiempo_ms = respuesta.tiempo_ms tipo_pregunta = respuesta.pregunta_activa.pregunta.tipo return { # ── Participación ── 'PRIMERA_RESPUESTA': total_respuestas == 1, 'PARTICIPANTE_10': total_respuestas >= 10, 'PARTICIPANTE_50': total_respuestas >= 50, 'PARTICIPANTE_100': total_respuestas >= 100, 'PRIMERA_REFLEXION': tipo_pregunta in ('confuso', 'minuto') and total_respuestas >= 1, # ── Velocidad ── 'RAYO': es_correcta and tiempo_ms < 3000, # < 3 segundos 'VELOZ_5': es_correcta and tiempo_ms < 5000, 'VELOCISTA_SERIAL': total_correctas >= 5 and tiempo_ms < 5000, # ── Conocimiento ── 'PRIMERA_CORRECTA': total_correctas == 1, 'CORRECTAS_10': total_correctas >= 10, 'CORRECTAS_25': total_correctas >= 25, 'CORRECTAS_50': total_correctas >= 50, 'CORRECTAS_100': total_correctas >= 100, 'PERFECTO': cls._sesion_perfecta(alumno, respuesta.pregunta_activa.sesion), # ── Rachas ── 'RACHA_3': puntos_obj.racha_sesiones >= 3, 'RACHA_5': puntos_obj.racha_sesiones >= 5, 'RACHA_10': puntos_obj.racha_sesiones >= 10, # ── Fichas ── 'PRIMERA_FICHA': puntos_obj.total_fichas >= 1, 'RICO_5_FICHAS': puntos_obj.total_fichas >= 5, # ── Puntos acumulados ── 'MIL_PUNTOS': puntos_obj.total_puntos >= 1000, 'CINCO_MIL': puntos_obj.total_puntos >= 5000, 'DIEZ_MIL': puntos_obj.total_puntos >= 10000, } @classmethod def _sesion_perfecta(cls, alumno, sesion) -> bool: """ True si el alumno respondió TODAS las preguntas correctamente en la sesión. Solo se evalúa si la sesión tiene al menos 3 preguntas. """ from sessions.models import Respuesta as RespuestaModel total_preguntas = sesion.preguntas_activas.filter(estado='cerrada').count() if total_preguntas < 3: return False correctas = RespuestaModel.objects.filter( alumno=alumno, pregunta_activa__sesion=sesion, es_correcta=True, ).count() return correctas == total_preguntas # ────────────────────────────────────────────────────────────── # ACTUALIZACIÓN DE RACHA # ────────────────────────────────────────────────────────────── @classmethod @transaction.atomic def actualizar_racha(cls, alumno, seccion, sesion_fecha: date): """ Actualiza la racha de sesiones consecutivas del alumno. Se llama al CERRAR una sesión, no al responder preguntas. Una racha se rompe si el alumno falta a una sesión (hay más de 1 día entre la última sesión y esta). """ from gamification.models import Puntos puntos_obj, _ = Puntos.objects.select_for_update().get_or_create( alumno=alumno, seccion=seccion, defaults={'total_puntos': 0, 'total_fichas': 0} ) ultima = puntos_obj.ultima_sesion_fecha if ultima is None: # Primera sesión → racha 1 puntos_obj.racha_sesiones = 1 elif (sesion_fecha - ultima).days == 1: # Sesión consecutiva → racha +1 puntos_obj.racha_sesiones += 1 elif sesion_fecha == ultima: # Misma fecha (varias sesiones el mismo día) → no cambia racha pass else: # Gap → racha reinicia puntos_obj.racha_sesiones = 1 puntos_obj.ultima_sesion_fecha = sesion_fecha puntos_obj.save(update_fields=['racha_sesiones', 'ultima_sesion_fecha']) # ────────────────────────────────────────────────────────────── # GESTIÓN MANUAL DE FICHAS (docente) # ────────────────────────────────────────────────────────────── @classmethod @transaction.atomic def ajustar_fichas(cls, alumno, seccion, delta: int, motivo: str, docente) -> int: """ Ajuste manual de fichas por el docente (+/–). Registrado en el log de auditoría. Args: delta: entero positivo (sumar) o negativo (restar) motivo: descripción del ajuste docente: usuario docente que realiza el ajuste Returns: Nuevo total de fichas del alumno """ from gamification.models import Puntos puntos_obj = Puntos.objects.select_for_update().get( alumno=alumno, seccion=seccion ) nuevo_total = puntos_obj.total_fichas + delta if nuevo_total < 0: raise ValueError( f'No se pueden restar {abs(delta)} fichas. ' f'El alumno solo tiene {puntos_obj.total_fichas}.' ) puntos_obj.total_fichas = nuevo_total puntos_obj.save(update_fields=['total_fichas']) logger.info( f'Ajuste fichas: alumno={alumno.id} delta={delta:+d} ' f'total={nuevo_total} docente={docente.id} motivo={motivo[:80]}' ) return nuevo_total