consultoria-as
27ac4cb0cf
Achievement System:
- Add Achievement model with condition types (streak, steal, specialist, etc.)
- Add AchievementManager service for tracking and awarding achievements
- Add Pydantic schemas for achievements (AchievementResponse, PlayerStats, etc.)
- Seed 18 achievements from design doc
- Add GET /api/game/achievements endpoint
Replay System:
- Add ReplayManager service for saving/loading game replays
- Add GET /api/replay/{code} and /api/replay/session/{id} endpoints
- Format replays for frontend consumption
Phase 2 tasks completed:
- F2.1: Achievement model and migration
- F2.2: Pydantic schemas
- F2.3: AchievementManager service
- F2.4: ReplayManager service
- F2.5: API endpoints
- F2.6: Seed 18 achievements data
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
246 lines
8.6 KiB
Python
246 lines
8.6 KiB
Python
from typing import Optional, List
|
|
from datetime import datetime
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.achievement import Achievement
|
|
from app.schemas.achievement import PlayerStats, AchievementUnlock, AchievementResponse
|
|
|
|
|
|
class AchievementManager:
|
|
"""Servicio para detectar y otorgar logros a jugadores durante las partidas."""
|
|
|
|
async def get_all_achievements(self, db: AsyncSession) -> List[Achievement]:
|
|
"""Obtiene todos los logros disponibles de la base de datos."""
|
|
result = await db.execute(select(Achievement))
|
|
return list(result.scalars().all())
|
|
|
|
async def check_achievements(
|
|
self,
|
|
db: AsyncSession,
|
|
player_stats: PlayerStats,
|
|
game_context: dict
|
|
) -> List[AchievementUnlock]:
|
|
"""
|
|
Verifica que logros ha desbloqueado el jugador.
|
|
|
|
Args:
|
|
db: Sesion de base de datos
|
|
player_stats: Estadisticas actuales del jugador
|
|
game_context: {
|
|
'won': bool,
|
|
'team_score': int,
|
|
'opponent_score': int,
|
|
'max_deficit_overcome': int, # para comeback
|
|
'categories_swept': List[int], # IDs de categorias con 5/5
|
|
'no_mistakes': bool, # para perfect_game
|
|
}
|
|
|
|
Returns:
|
|
Lista de logros desbloqueados
|
|
"""
|
|
achievements = await self.get_all_achievements(db)
|
|
unlocked: List[AchievementUnlock] = []
|
|
now = datetime.utcnow()
|
|
|
|
for achievement in achievements:
|
|
is_unlocked = self._evaluate_condition(
|
|
achievement=achievement,
|
|
stats=player_stats,
|
|
game_context=game_context
|
|
)
|
|
|
|
if is_unlocked:
|
|
achievement_response = AchievementResponse(
|
|
id=achievement.id,
|
|
name=achievement.name,
|
|
description=achievement.description,
|
|
icon=achievement.icon,
|
|
condition_type=achievement.condition_type,
|
|
condition_value=achievement.condition_value,
|
|
category_id=achievement.category_id,
|
|
created_at=achievement.created_at
|
|
)
|
|
|
|
unlocked.append(AchievementUnlock(
|
|
player_name=player_stats.player_name,
|
|
achievement=achievement_response,
|
|
unlocked_at=now
|
|
))
|
|
|
|
return unlocked
|
|
|
|
def _evaluate_condition(
|
|
self,
|
|
achievement: Achievement,
|
|
stats: PlayerStats,
|
|
game_context: dict
|
|
) -> bool:
|
|
"""Evalua si se cumple la condicion de un logro."""
|
|
condition_type = achievement.condition_type
|
|
condition_value = achievement.condition_value
|
|
|
|
if condition_type == "first_win":
|
|
return game_context.get("won", False)
|
|
|
|
elif condition_type == "streak":
|
|
return self._check_streak(stats, condition_value)
|
|
|
|
elif condition_type == "steal_success":
|
|
return self._check_steal_success(stats, condition_value)
|
|
|
|
elif condition_type == "category_specialist":
|
|
if achievement.category_id is not None:
|
|
return self._check_category_specialist(
|
|
stats, achievement.category_id, condition_value
|
|
)
|
|
return False
|
|
|
|
elif condition_type == "perfect_game":
|
|
return (
|
|
game_context.get("won", False) and
|
|
game_context.get("no_mistakes", False)
|
|
)
|
|
|
|
elif condition_type == "fast_answer":
|
|
return self._check_fast_answer(stats, condition_value)
|
|
|
|
elif condition_type == "comeback":
|
|
return (
|
|
game_context.get("won", False) and
|
|
game_context.get("max_deficit_overcome", 0) >= condition_value
|
|
)
|
|
|
|
elif condition_type == "category_sweep":
|
|
categories_swept = game_context.get("categories_swept", [])
|
|
return len(categories_swept) >= condition_value
|
|
|
|
elif condition_type == "high_stakes":
|
|
return self._check_high_stakes(stats, condition_value)
|
|
|
|
return False
|
|
|
|
def _check_streak(self, stats: PlayerStats, required: int) -> bool:
|
|
"""Verifica racha de respuestas correctas."""
|
|
return stats.current_streak >= required
|
|
|
|
def _check_steal_success(self, stats: PlayerStats, required: int) -> bool:
|
|
"""Verifica robos exitosos."""
|
|
return stats.successful_steals >= required
|
|
|
|
def _check_category_specialist(
|
|
self,
|
|
stats: PlayerStats,
|
|
category_id: int,
|
|
required: int
|
|
) -> bool:
|
|
"""Verifica especialista de categoria."""
|
|
return stats.category_correct.get(category_id, 0) >= required
|
|
|
|
def _check_fast_answer(self, stats: PlayerStats, max_seconds: int) -> bool:
|
|
"""Verifica respuesta rapida."""
|
|
if stats.fastest_answer_seconds is None:
|
|
return False
|
|
return stats.fastest_answer_seconds <= max_seconds
|
|
|
|
def _check_high_stakes(self, stats: PlayerStats, required: int) -> bool:
|
|
"""Verifica preguntas de 500 pts correctas."""
|
|
return stats.questions_500_correct >= required
|
|
|
|
def update_stats_on_answer(
|
|
self,
|
|
stats: PlayerStats,
|
|
was_correct: bool,
|
|
was_steal: bool,
|
|
category_id: int,
|
|
points: int,
|
|
answer_time_seconds: float
|
|
) -> PlayerStats:
|
|
"""
|
|
Actualiza estadisticas despues de una respuesta.
|
|
|
|
Args:
|
|
stats: Estadisticas actuales del jugador
|
|
was_correct: Si la respuesta fue correcta
|
|
was_steal: Si fue un intento de robo
|
|
category_id: ID de la categoria de la pregunta
|
|
points: Puntos de la pregunta
|
|
answer_time_seconds: Tiempo de respuesta en segundos
|
|
|
|
Returns:
|
|
PlayerStats actualizado
|
|
"""
|
|
# Crear copia mutable de las estadisticas
|
|
updated_category_correct = dict(stats.category_correct)
|
|
|
|
if was_correct:
|
|
# Actualizar racha
|
|
new_streak = stats.current_streak + 1
|
|
new_total_correct = stats.total_correct + 1
|
|
|
|
# Actualizar correctas por categoria
|
|
updated_category_correct[category_id] = (
|
|
updated_category_correct.get(category_id, 0) + 1
|
|
)
|
|
|
|
# Actualizar tiempo mas rapido
|
|
new_fastest = stats.fastest_answer_seconds
|
|
if new_fastest is None or answer_time_seconds < new_fastest:
|
|
new_fastest = answer_time_seconds
|
|
|
|
# Actualizar preguntas de 500 pts
|
|
new_500_correct = stats.questions_500_correct
|
|
if points == 500:
|
|
new_500_correct += 1
|
|
|
|
# Actualizar robos exitosos
|
|
new_successful_steals = stats.successful_steals
|
|
new_total_steals = stats.total_steals
|
|
if was_steal:
|
|
new_successful_steals += 1
|
|
new_total_steals += 1
|
|
|
|
return PlayerStats(
|
|
player_name=stats.player_name,
|
|
current_streak=new_streak,
|
|
total_correct=new_total_correct,
|
|
total_steals=new_total_steals,
|
|
successful_steals=new_successful_steals,
|
|
category_correct=updated_category_correct,
|
|
fastest_answer_seconds=new_fastest,
|
|
questions_500_correct=new_500_correct
|
|
)
|
|
else:
|
|
# Respuesta incorrecta - resetear racha
|
|
new_total_steals = stats.total_steals
|
|
if was_steal:
|
|
new_total_steals += 1
|
|
|
|
return PlayerStats(
|
|
player_name=stats.player_name,
|
|
current_streak=0,
|
|
total_correct=stats.total_correct,
|
|
total_steals=new_total_steals,
|
|
successful_steals=stats.successful_steals,
|
|
category_correct=updated_category_correct,
|
|
fastest_answer_seconds=stats.fastest_answer_seconds,
|
|
questions_500_correct=stats.questions_500_correct
|
|
)
|
|
|
|
def create_initial_stats(self, player_name: str) -> PlayerStats:
|
|
"""Crea estadisticas iniciales para un jugador."""
|
|
return PlayerStats(
|
|
player_name=player_name,
|
|
current_streak=0,
|
|
total_correct=0,
|
|
total_steals=0,
|
|
successful_steals=0,
|
|
category_correct={},
|
|
fastest_answer_seconds=None,
|
|
questions_500_correct=0
|
|
)
|
|
|
|
|
|
# Singleton instance
|
|
achievement_manager = AchievementManager()
|