consultoria-as/Trivy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Initial project structure for WebTriviasMulti

Browse Source 
- Backend: FastAPI + Python-SocketIO + SQLAlchemy
  - Models for categories, questions, game sessions, events
  - AI services for answer validation and question generation (Claude)
  - Room management with Redis
  - Game logic with stealing mechanics
  - Admin API for question management

- Frontend: React + Vite + TypeScript + Tailwind
  - 5 visual themes (DRRR, Retro, Minimal, RGB, Anime 90s)
  - Real-time game with Socket.IO
  - Achievement system
  - Replay functionality
  - Sound effects per theme

- Docker Compose for deployment
- Design documentation

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
consultoria-as
2026-01-26 07:50:48 +00:00
commit 43021b9c3c
57 changed files with 5446 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
backend/app/services/__init__.py Normal file
View File

@@ -0,0 +1,6 @@
from app.services.ai_validator import AIValidator
from app.services.ai_generator import AIGenerator
from app.services.game_manager import GameManager
from app.services.room_manager import RoomManager
__all__ = ["AIValidator", "AIGenerator", "GameManager", "RoomManager"]

97
backend/app/services/ai_generator.py Normal file
View File

@@ -0,0 +1,97 @@
import json
from anthropic import Anthropic
from app.config import get_settings
settings = get_settings()
class AIGenerator:
def __init__(self):
self.client = Anthropic(api_key=settings.anthropic_api_key)
async def generate_questions(
self,
category_name: str,
difficulty: int,
count: int = 5
) -> list[dict]:
"""
Generate trivia questions using Claude AI.
Args:
category_name: Name of the category (e.g., "Nintendo", "Anime")
difficulty: 1-5 (1=very easy, 5=very hard)
count: Number of questions to generate
Returns:
list[dict]: List of question objects
"""
difficulty_descriptions = {
1: "muy fácil - conocimiento básico que la mayoría conoce",
2: "fácil - conocimiento común entre fans casuales",
3: "medio - requiere ser fan de la categoría",
4: "difícil - conocimiento profundo del tema",
5: "muy difícil - solo expertos conocerían esto"
}
prompt = f"""Genera {count} preguntas de trivia para la categoría "{category_name}".
Dificultad: {difficulty} ({difficulty_descriptions.get(difficulty, 'medio')})
Requisitos:
- Las preguntas deben ser verificables y precisas
- Evitar ambigüedades
- Las respuestas deben ser específicas y concisas
- Incluir variaciones comunes de la respuesta
- Para gaming: referencias a juegos, personajes, mecánicas, fechas de lanzamiento
- Para anime: personajes, series, estudios, seiyuus
- Para música: artistas, canciones, álbumes, letras famosas
- Para películas: actores, directores, frases icónicas, premios
- Para libros: autores, obras, personajes literarios
- Para historia-cultura: eventos, fechas, personajes históricos, arte
Formato JSON (array de objetos):
[
{{
"question": "texto de la pregunta",
"correct_answer": "respuesta principal",
"alt_answers": ["variación1", "variación2"],
"fun_fact": "dato curioso opcional sobre la respuesta"
}}
]
Responde SOLO con el JSON, sin texto adicional."""
try:
message = self.client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=2000,
messages=[
{"role": "user", "content": prompt}
]
)
response_text = message.content[0].text.strip()
# Parse JSON response
questions = json.loads(response_text)
# Add metadata to each question
for q in questions:
q["difficulty"] = difficulty
q["points"] = settings.default_points.get(difficulty, 300)
q["time_seconds"] = settings.default_times.get(difficulty, 25)
return questions
except json.JSONDecodeError as e:
print(f"Error parsing AI response: {e}")
print(f"Response was: {response_text}")
return []
except Exception as e:
print(f"Error generating questions: {e}")
return []
# Singleton instance
ai_generator = AIGenerator()

80
backend/app/services/ai_validator.py Normal file
View File

@@ -0,0 +1,80 @@
import json
from anthropic import Anthropic
from app.config import get_settings
settings = get_settings()
class AIValidator:
def __init__(self):
self.client = Anthropic(api_key=settings.anthropic_api_key)
async def validate_answer(
self,
question: str,
correct_answer: str,
alt_answers: list[str],
player_answer: str
) -> dict:
"""
Validate if the player's answer is correct using Claude AI.
Returns:
dict: {"valid": bool, "reason": str}
"""
prompt = f"""Eres un validador de trivia. Determina si la respuesta del jugador
es correcta comparándola con la respuesta oficial.
Pregunta: {question}
Respuesta correcta: {correct_answer}
Respuestas alternativas válidas: {', '.join(alt_answers) if alt_answers else 'Ninguna'}
Respuesta del jugador: {player_answer}
Considera válido si:
- Es sinónimo o variación de la respuesta correcta
- Tiene errores menores de ortografía
- Usa abreviaciones comunes (ej: "BOTW" = "Breath of the Wild")
- Es conceptualmente equivalente
Responde SOLO con JSON: {{"valid": true/false, "reason": "breve explicación"}}"""
try:
message = self.client.messages.create(
model="claude-3-haiku-20240307",
max_tokens=150,
messages=[
{"role": "user", "content": prompt}
]
)
response_text = message.content[0].text.strip()
# Parse JSON response
result = json.loads(response_text)
return result
except json.JSONDecodeError:
# If JSON parsing fails, try to extract the result
response_lower = response_text.lower()
if "true" in response_lower:
return {"valid": True, "reason": "Respuesta validada por IA"}
return {"valid": False, "reason": "No se pudo validar la respuesta"}
except Exception as e:
print(f"Error validating answer: {e}")
# Fallback to exact match
player_lower = player_answer.lower().strip()
correct_lower = correct_answer.lower().strip()
if player_lower == correct_lower:
return {"valid": True, "reason": "Coincidencia exacta"}
for alt in alt_answers:
if player_lower == alt.lower().strip():
return {"valid": True, "reason": "Coincide con respuesta alternativa"}
return {"valid": False, "reason": "Respuesta incorrecta"}
# Singleton instance
ai_validator = AIValidator()

204
backend/app/services/game_manager.py Normal file
View File

@@ -0,0 +1,204 @@
from typing import Optional
from datetime import datetime, timedelta
from app.services.room_manager import room_manager
from app.services.ai_validator import ai_validator
from app.config import get_settings
settings = get_settings()
class GameManager:
async def start_game(self, room_code: str, board: dict) -> Optional[dict]:
"""
Start a game in a room.
Args:
room_code: The room code
board: Dict of category_id -> list of questions
Returns:
Updated room state
"""
room = await room_manager.get_room(room_code)
if not room:
return None
# Check minimum players
if not room["teams"]["A"] or not room["teams"]["B"]:
return None
# Set up game state
room["status"] = "playing"
room["current_team"] = "A"
room["current_player_index"] = {"A": 0, "B": 0}
room["board"] = board
room["scores"] = {"A": 0, "B": 0}
await room_manager.update_room(room_code, room)
return room
async def select_question(
self,
room_code: str,
question_id: int,
category_id: int
) -> Optional[dict]:
"""Select a question from the board."""
room = await room_manager.get_room(room_code)
if not room or room["status"] != "playing":
return None
# Mark question as current
room["current_question"] = question_id
room["can_steal"] = False
# Find and mark question on board
if str(category_id) in room["board"]:
for q in room["board"][str(category_id)]:
if q["id"] == question_id:
q["selected"] = True
break
await room_manager.update_room(room_code, room)
return room
async def submit_answer(
self,
room_code: str,
question: dict,
player_answer: str,
is_steal: bool = False
) -> dict:
"""
Submit an answer for validation.
Returns:
dict with validation result and updated game state
"""
room = await room_manager.get_room(room_code)
if not room:
return {"error": "Room not found"}
# Validate answer with AI
result = await ai_validator.validate_answer(
question=question["question_text"],
correct_answer=question["correct_answer"],
alt_answers=question.get("alt_answers", []),
player_answer=player_answer
)
is_correct = result.get("valid", False)
points = question["points"]
if is_correct:
# Award points
current_team = room["current_team"]
room["scores"][current_team] += points
# Mark question as answered
category_id = str(question["category_id"])
if category_id in room["board"]:
for q in room["board"][category_id]:
if q["id"] == question["id"]:
q["answered"] = True
break
# Winner chooses next
room["current_question"] = None
room["can_steal"] = False
# Advance player rotation
team_players = room["teams"][current_team]
room["current_player_index"][current_team] = (
room["current_player_index"][current_team] + 1
) % len(team_players)
else:
if is_steal:
# Failed steal - penalize
stealing_team = room["current_team"]
penalty = int(points * settings.steal_penalty_multiplier)
room["scores"][stealing_team] = max(
0, room["scores"][stealing_team] - penalty
)
# Mark question as answered (nobody gets it)
category_id = str(question["category_id"])
if category_id in room["board"]:
for q in room["board"][category_id]:
if q["id"] == question["id"]:
q["answered"] = True
break
# Original team chooses next
room["current_team"] = "B" if stealing_team == "A" else "A"
room["current_question"] = None
room["can_steal"] = False
else:
# Original team failed - enable steal
room["can_steal"] = True
# Switch to other team for potential steal
room["current_team"] = "B" if room["current_team"] == "A" else "A"
# Check if game is over (all questions answered)
all_answered = all(
q["answered"]
for questions in room["board"].values()
for q in questions
)
if all_answered:
room["status"] = "finished"
await room_manager.update_room(room_code, room)
return {
"valid": is_correct,
"reason": result.get("reason", ""),
"points_earned": points if is_correct else 0,
"room": room
}
async def pass_steal(self, room_code: str, question_id: int) -> Optional[dict]:
"""Pass on stealing opportunity."""
room = await room_manager.get_room(room_code)
if not room:
return None
# Mark question as answered
for category_id, questions in room["board"].items():
for q in questions:
if q["id"] == question_id:
q["answered"] = True
break
# Switch back to original team for next selection
room["current_team"] = "B" if room["current_team"] == "A" else "A"
room["current_question"] = None
room["can_steal"] = False
await room_manager.update_room(room_code, room)
return room
async def get_current_player(self, room: dict) -> Optional[dict]:
"""Get the current player who should answer."""
team = room["current_team"]
if not team:
return None
players = room["teams"][team]
if not players:
return None
index = room["current_player_index"][team]
return players[index % len(players)]
def calculate_timer_end(self, time_seconds: int, is_steal: bool = False) -> datetime:
"""Calculate when the timer should end."""
if is_steal:
time_seconds = int(time_seconds * settings.steal_time_multiplier)
return datetime.utcnow() + timedelta(seconds=time_seconds)
# Singleton instance
game_manager = GameManager()

173
backend/app/services/room_manager.py Normal file
View File

@@ -0,0 +1,173 @@
import json
import random
import string
from typing import Optional
import redis.asyncio as redis
from app.config import get_settings
settings = get_settings()
class RoomManager:
def __init__(self):
self.redis: Optional[redis.Redis] = None
async def connect(self):
if not self.redis:
self.redis = await redis.from_url(settings.redis_url)
async def disconnect(self):
if self.redis:
await self.redis.close()
def _generate_room_code(self) -> str:
"""Generate a 6-character room code."""
return ''.join(random.choices(string.ascii_uppercase + string.digits, k=6))
async def create_room(self, player_name: str, socket_id: str) -> dict:
"""Create a new game room."""
await self.connect()
# Generate unique room code
room_code = self._generate_room_code()
while await self.redis.exists(f"room:{room_code}"):
room_code = self._generate_room_code()
# Create room state
room_state = {
"code": room_code,
"status": "waiting",
"host": player_name,
"teams": {
"A": [],
"B": []
},
"current_team": None,
"current_player_index": {"A": 0, "B": 0},
"current_question": None,
"can_steal": False,
"scores": {"A": 0, "B": 0},
"questions_used": [],
"board": {}
}
# Save room state
await self.redis.setex(
f"room:{room_code}",
3600 * 3, # 3 hours TTL
json.dumps(room_state)
)
# Add player to room
await self.add_player(room_code, player_name, "A", socket_id)
return room_state
async def get_room(self, room_code: str) -> Optional[dict]:
"""Get room state by code."""
await self.connect()
data = await self.redis.get(f"room:{room_code}")
if data:
return json.loads(data)
return None
async def update_room(self, room_code: str, room_state: dict) -> bool:
"""Update room state."""
await self.connect()
await self.redis.setex(
f"room:{room_code}",
3600 * 3,
json.dumps(room_state)
)
return True
async def add_player(
self,
room_code: str,
player_name: str,
team: str,
socket_id: str
) -> Optional[dict]:
"""Add a player to a room."""
room = await self.get_room(room_code)
if not room:
return None
# Check if team is full
if len(room["teams"][team]) >= 4:
return None
# Check if name is taken
for t in ["A", "B"]:
for p in room["teams"][t]:
if p["name"].lower() == player_name.lower():
return None
# Add player
player = {
"name": player_name,
"team": team,
"position": len(room["teams"][team]),
"socket_id": socket_id
}
room["teams"][team].append(player)
# Save player mapping
await self.redis.setex(
f"player:{socket_id}",
3600 * 3,
json.dumps({"name": player_name, "room": room_code, "team": team})
)
await self.update_room(room_code, room)
return room
async def remove_player(self, socket_id: str) -> Optional[dict]:
"""Remove a player from their room."""
await self.connect()
# Get player info
player_data = await self.redis.get(f"player:{socket_id}")
if not player_data:
return None
player_info = json.loads(player_data)
room_code = player_info["room"]
team = player_info["team"]
# Get room
room = await self.get_room(room_code)
if not room:
return None
# Remove player from team
room["teams"][team] = [
p for p in room["teams"][team] if p["socket_id"] != socket_id
]
# Update positions
for i, p in enumerate(room["teams"][team]):
p["position"] = i
# Delete player mapping
await self.redis.delete(f"player:{socket_id}")
# If room is empty, delete it
if not room["teams"]["A"] and not room["teams"]["B"]:
await self.redis.delete(f"room:{room_code}")
return None
await self.update_room(room_code, room)
return room
async def get_player(self, socket_id: str) -> Optional[dict]:
"""Get player info by socket ID."""
await self.connect()
data = await self.redis.get(f"player:{socket_id}")
if data:
return json.loads(data)
return None
# Singleton instance
room_manager = RoomManager()