consultoria-as
0141153653
Timer System: - Add TimerManager service with asyncio for server-side timers - Support steal time reduction (50% time) - Automatic timer cancellation on answer Question Loading: - Add QuestionService to load daily questions from PostgreSQL - Generate 8×5 board (categories × difficulties) - Filter by date_active and approved status Database Integration: - Create GameSession in PostgreSQL when game starts - Update scores during game and on finish - Store db_session_id in Redis for cross-reference Replay Integration: - Save all game events: question_selected, answer_submitted, steal_attempted, steal_passed, game_finished - Generate unique replay code on game finish Achievement Integration: - Initialize PlayerStats in Redis when joining room - Update stats on every answer (streak, category, speed, etc.) - Check achievements on game finish for all players Game Finish: - Automatic finish when all questions answered - Manual finish by host - Emit game_finished with winner, scores, replay_code, achievements Phase 3 tasks completed: - F3.1: Timer manager with asyncio - F3.2: Question service for board loading - F3.3: GameSession PostgreSQL integration - F3.4: Replay event saving - F3.5: Achievement stats tracking - F3.6: Complete game finish flow Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
87 lines
2.8 KiB
Python
87 lines
2.8 KiB
Python
import asyncio
|
|
from typing import Optional, Dict, Callable, Awaitable
|
|
from datetime import datetime, timedelta
|
|
from app.config import get_settings
|
|
|
|
settings = get_settings()
|
|
|
|
|
|
class TimerManager:
|
|
def __init__(self):
|
|
self.active_timers: Dict[str, asyncio.Task] = {} # room_code -> task
|
|
self.timer_end_times: Dict[str, datetime] = {} # room_code -> end_time
|
|
|
|
async def start_timer(
|
|
self,
|
|
room_code: str,
|
|
seconds: int,
|
|
on_expire: Callable[[str], Awaitable[None]],
|
|
is_steal: bool = False
|
|
) -> datetime:
|
|
"""
|
|
Inicia un timer para una sala.
|
|
|
|
Args:
|
|
room_code: Codigo de la sala
|
|
seconds: Segundos base (se reduce si is_steal)
|
|
on_expire: Callback async cuando expire el timer
|
|
is_steal: Si es intento de robo (tiempo reducido)
|
|
|
|
Returns:
|
|
datetime cuando expira el timer
|
|
"""
|
|
# Cancelar timer anterior si existe
|
|
await self.cancel_timer(room_code)
|
|
|
|
# Calcular tiempo real
|
|
if is_steal:
|
|
seconds = int(seconds * settings.steal_time_multiplier)
|
|
|
|
end_time = datetime.utcnow() + timedelta(seconds=seconds)
|
|
self.timer_end_times[room_code] = end_time
|
|
|
|
# Crear task
|
|
async def timer_task():
|
|
try:
|
|
await asyncio.sleep(seconds)
|
|
# Timer expiro
|
|
if room_code in self.active_timers:
|
|
del self.active_timers[room_code]
|
|
del self.timer_end_times[room_code]
|
|
await on_expire(room_code)
|
|
except asyncio.CancelledError:
|
|
# Timer fue cancelado (respuesta recibida a tiempo)
|
|
pass
|
|
|
|
self.active_timers[room_code] = asyncio.create_task(timer_task())
|
|
return end_time
|
|
|
|
async def cancel_timer(self, room_code: str) -> bool:
|
|
"""Cancela el timer de una sala."""
|
|
if room_code in self.active_timers:
|
|
self.active_timers[room_code].cancel()
|
|
try:
|
|
await self.active_timers[room_code]
|
|
except asyncio.CancelledError:
|
|
pass
|
|
del self.active_timers[room_code]
|
|
if room_code in self.timer_end_times:
|
|
del self.timer_end_times[room_code]
|
|
return True
|
|
return False
|
|
|
|
def get_remaining_time(self, room_code: str) -> Optional[float]:
|
|
"""Obtiene segundos restantes del timer."""
|
|
if room_code in self.timer_end_times:
|
|
remaining = (self.timer_end_times[room_code] - datetime.utcnow()).total_seconds()
|
|
return max(0, remaining)
|
|
return None
|
|
|
|
def is_timer_active(self, room_code: str) -> bool:
|
|
"""Verifica si hay un timer activo para la sala."""
|
|
return room_code in self.active_timers
|
|
|
|
|
|
# Singleton
|
|
timer_manager = TimerManager()
|