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>
137 lines
3.9 KiB
Python
137 lines
3.9 KiB
Python
from typing import Optional, List, Dict
|
||
from datetime import date
|
||
from sqlalchemy import select, and_
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.models.question import Question
|
||
from app.models.category import Category
|
||
|
||
|
||
class QuestionService:
|
||
async def get_daily_questions(
|
||
self,
|
||
db: AsyncSession,
|
||
target_date: Optional[date] = None
|
||
) -> Dict[int, List[dict]]:
|
||
"""
|
||
Obtiene preguntas del día organizadas por categoría.
|
||
|
||
Args:
|
||
db: Sesión de BD
|
||
target_date: Fecha (default: hoy)
|
||
|
||
Returns:
|
||
Dict[category_id, List[question_dict]] para el tablero
|
||
"""
|
||
if target_date is None:
|
||
target_date = date.today()
|
||
|
||
# Buscar preguntas aprobadas para la fecha
|
||
query = select(Question).where(
|
||
and_(
|
||
Question.date_active == target_date,
|
||
Question.status == "approved"
|
||
)
|
||
).order_by(Question.category_id, Question.difficulty)
|
||
|
||
result = await db.execute(query)
|
||
questions = result.scalars().all()
|
||
|
||
# Organizar por categoría
|
||
board: Dict[int, List[dict]] = {}
|
||
for q in questions:
|
||
if q.category_id not in board:
|
||
board[q.category_id] = []
|
||
board[q.category_id].append({
|
||
"id": q.id,
|
||
"category_id": q.category_id,
|
||
"question_text": q.question_text,
|
||
"correct_answer": q.correct_answer,
|
||
"alt_answers": q.alt_answers or [],
|
||
"difficulty": q.difficulty,
|
||
"points": q.points,
|
||
"time_seconds": q.time_seconds,
|
||
"fun_fact": q.fun_fact,
|
||
"answered": False,
|
||
"selected": False
|
||
})
|
||
|
||
return board
|
||
|
||
async def get_board_for_game(
|
||
self,
|
||
db: AsyncSession,
|
||
target_date: Optional[date] = None
|
||
) -> Dict[str, List[dict]]:
|
||
"""
|
||
Genera el tablero 8×5 para el juego.
|
||
Si no hay suficientes preguntas, retorna lo disponible.
|
||
|
||
Returns:
|
||
Dict con category_id como string (para JSON) -> lista de preguntas
|
||
"""
|
||
board = await self.get_daily_questions(db, target_date)
|
||
|
||
# Convertir keys a string para JSON
|
||
return {str(k): v for k, v in board.items()}
|
||
|
||
async def get_question_by_id(
|
||
self,
|
||
db: AsyncSession,
|
||
question_id: int
|
||
) -> Optional[Question]:
|
||
"""Obtiene una pregunta por ID."""
|
||
result = await db.execute(
|
||
select(Question).where(Question.id == question_id)
|
||
)
|
||
return result.scalar_one_or_none()
|
||
|
||
async def mark_question_used(
|
||
self,
|
||
db: AsyncSession,
|
||
question_id: int
|
||
) -> bool:
|
||
"""Marca una pregunta como usada."""
|
||
question = await self.get_question_by_id(db, question_id)
|
||
if question:
|
||
question.status = "used"
|
||
await db.commit()
|
||
return True
|
||
return False
|
||
|
||
async def get_categories_with_questions(
|
||
self,
|
||
db: AsyncSession,
|
||
target_date: Optional[date] = None
|
||
) -> List[dict]:
|
||
"""
|
||
Obtiene categorías que tienen preguntas para la fecha.
|
||
Útil para mostrar solo categorías disponibles.
|
||
"""
|
||
board = await self.get_daily_questions(db, target_date)
|
||
|
||
# Obtener info de categorías
|
||
cat_ids = list(board.keys())
|
||
if not cat_ids:
|
||
return []
|
||
|
||
result = await db.execute(
|
||
select(Category).where(Category.id.in_(cat_ids))
|
||
)
|
||
categories = result.scalars().all()
|
||
|
||
return [
|
||
{
|
||
"id": c.id,
|
||
"name": c.name,
|
||
"icon": c.icon,
|
||
"color": c.color,
|
||
"question_count": len(board.get(c.id, []))
|
||
}
|
||
for c in categories
|
||
]
|
||
|
||
|
||
# Singleton
|
||
question_service = QuestionService()
|