consultoria-as
1e1daf94f6
- Script generate_daily_questions.py: genera 5 preguntas por categoría/dificultad - Usa Claude API para generar preguntas en español - Cron job configurado para medianoche (0 0 * * *) - 14 categorías × 5 dificultades × 5 preguntas = 350 preguntas/día - Evita duplicados verificando preguntas existentes fix: rotación de jugadores en robo fallido/pasado Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
257 lines
8.5 KiB
Python
Executable File
257 lines
8.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Script para generar preguntas diarias automáticamente usando Claude API.
|
|
Ejecutar con cron a medianoche para generar preguntas del día siguiente.
|
|
"""
|
|
|
|
import asyncio
|
|
import sys
|
|
import os
|
|
from datetime import date, timedelta
|
|
from typing import List, Dict
|
|
import json
|
|
|
|
# Add backend to path
|
|
sys.path.insert(0, '/root/Trivy/backend')
|
|
|
|
import anthropic
|
|
from sqlalchemy import select
|
|
from app.models.base import get_async_session
|
|
from app.models.category import Category
|
|
from app.models.question import Question
|
|
|
|
# Configuration
|
|
QUESTIONS_PER_DIFFICULTY = 5 # 5 preguntas por cada dificultad
|
|
DIFFICULTIES = [1, 2, 3, 4, 5]
|
|
POINTS_MAP = {1: 100, 2: 200, 3: 300, 4: 400, 5: 500}
|
|
TIME_MAP = {1: 30, 2: 30, 3: 25, 4: 20, 5: 15}
|
|
|
|
# Get API key from environment
|
|
ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")
|
|
|
|
|
|
def get_difficulty_description(difficulty: int) -> str:
|
|
"""Get description for difficulty level."""
|
|
descriptions = {
|
|
1: "muy fácil, conocimiento básico que casi todos saben",
|
|
2: "fácil, conocimiento común para fans casuales",
|
|
3: "moderada, requiere conocimiento intermedio del tema",
|
|
4: "difícil, requiere conocimiento profundo del tema",
|
|
5: "muy difícil, solo expertos o super fans sabrían la respuesta"
|
|
}
|
|
return descriptions.get(difficulty, "moderada")
|
|
|
|
|
|
async def generate_questions_for_category(
|
|
client: anthropic.Anthropic,
|
|
category: Dict,
|
|
difficulty: int,
|
|
target_date: date,
|
|
existing_questions: List[str]
|
|
) -> List[Dict]:
|
|
"""Generate questions for a specific category and difficulty using Claude."""
|
|
|
|
difficulty_desc = get_difficulty_description(difficulty)
|
|
points = POINTS_MAP[difficulty]
|
|
time_seconds = TIME_MAP[difficulty]
|
|
|
|
# Build prompt
|
|
prompt = f"""Genera exactamente {QUESTIONS_PER_DIFFICULTY} preguntas de trivia sobre "{category['name']}" con dificultad {difficulty} ({difficulty_desc}).
|
|
|
|
REGLAS IMPORTANTES:
|
|
1. Las preguntas deben ser en español
|
|
2. Las respuestas deben ser cortas (1-4 palabras idealmente)
|
|
3. Incluye respuestas alternativas válidas cuando aplique
|
|
4. NO repitas estas preguntas existentes: {json.dumps(existing_questions[:20], ensure_ascii=False) if existing_questions else "ninguna"}
|
|
5. Cada pregunta debe tener un dato curioso relacionado
|
|
6. Las preguntas deben ser verificables y tener una respuesta objetiva correcta
|
|
|
|
Responde SOLO con un JSON array válido con esta estructura exacta:
|
|
[
|
|
{{
|
|
"question_text": "¿Pregunta aquí?",
|
|
"correct_answer": "Respuesta correcta",
|
|
"alt_answers": ["alternativa1", "alternativa2"],
|
|
"fun_fact": "Dato curioso relacionado con la pregunta"
|
|
}}
|
|
]
|
|
|
|
Genera exactamente {QUESTIONS_PER_DIFFICULTY} preguntas diferentes y variadas sobre {category['name']}."""
|
|
|
|
try:
|
|
response = client.messages.create(
|
|
model="claude-sonnet-4-20250514",
|
|
max_tokens=2000,
|
|
messages=[{"role": "user", "content": prompt}]
|
|
)
|
|
|
|
# Extract JSON from response
|
|
response_text = response.content[0].text.strip()
|
|
|
|
# Try to find JSON array in response
|
|
start_idx = response_text.find('[')
|
|
end_idx = response_text.rfind(']') + 1
|
|
|
|
if start_idx == -1 or end_idx == 0:
|
|
print(f" ERROR: No se encontró JSON válido para {category['name']} dificultad {difficulty}")
|
|
return []
|
|
|
|
json_str = response_text[start_idx:end_idx]
|
|
questions_data = json.loads(json_str)
|
|
|
|
# Format questions for database
|
|
formatted_questions = []
|
|
for q in questions_data:
|
|
formatted_questions.append({
|
|
"category_id": category['id'],
|
|
"question_text": q["question_text"],
|
|
"correct_answer": q["correct_answer"],
|
|
"alt_answers": q.get("alt_answers", []),
|
|
"difficulty": difficulty,
|
|
"points": points,
|
|
"time_seconds": time_seconds,
|
|
"date_active": target_date,
|
|
"status": "approved",
|
|
"fun_fact": q.get("fun_fact", "")
|
|
})
|
|
|
|
return formatted_questions
|
|
|
|
except json.JSONDecodeError as e:
|
|
print(f" ERROR JSON para {category['name']} dificultad {difficulty}: {e}")
|
|
return []
|
|
except Exception as e:
|
|
print(f" ERROR generando para {category['name']} dificultad {difficulty}: {e}")
|
|
return []
|
|
|
|
|
|
async def get_existing_questions(db, category_id: int) -> List[str]:
|
|
"""Get existing question texts to avoid duplicates."""
|
|
result = await db.execute(
|
|
select(Question.question_text).where(Question.category_id == category_id)
|
|
)
|
|
return [row[0] for row in result.fetchall()]
|
|
|
|
|
|
async def generate_daily_questions(target_date: date = None):
|
|
"""Main function to generate all daily questions."""
|
|
|
|
if not ANTHROPIC_API_KEY:
|
|
print("ERROR: ANTHROPIC_API_KEY no está configurada")
|
|
sys.exit(1)
|
|
|
|
if target_date is None:
|
|
# Generate for tomorrow by default
|
|
target_date = date.today() + timedelta(days=1)
|
|
|
|
print(f"=== Generando preguntas para {target_date} ===")
|
|
|
|
client = anthropic.Anthropic(api_key=ANTHROPIC_API_KEY)
|
|
AsyncSessionLocal = get_async_session()
|
|
|
|
async with AsyncSessionLocal() as db:
|
|
# Get all categories
|
|
result = await db.execute(select(Category))
|
|
categories = result.scalars().all()
|
|
|
|
if not categories:
|
|
print("ERROR: No hay categorías en la base de datos")
|
|
return
|
|
|
|
print(f"Categorías encontradas: {len(categories)}")
|
|
|
|
total_generated = 0
|
|
|
|
for category in categories:
|
|
cat_dict = {"id": category.id, "name": category.name}
|
|
print(f"\n📁 Categoría: {category.name}")
|
|
|
|
# Get existing questions to avoid duplicates
|
|
existing = await get_existing_questions(db, category.id)
|
|
|
|
for difficulty in DIFFICULTIES:
|
|
print(f" Dificultad {difficulty}...", end=" ", flush=True)
|
|
|
|
questions = await generate_questions_for_category(
|
|
client, cat_dict, difficulty, target_date, existing
|
|
)
|
|
|
|
if questions:
|
|
# Insert into database
|
|
for q_data in questions:
|
|
question = Question(**q_data)
|
|
db.add(question)
|
|
existing.append(q_data["question_text"])
|
|
|
|
print(f"✓ {len(questions)} preguntas")
|
|
total_generated += len(questions)
|
|
else:
|
|
print("✗ Error")
|
|
|
|
# Small delay to avoid rate limiting
|
|
await asyncio.sleep(1)
|
|
|
|
await db.commit()
|
|
|
|
print(f"\n=== COMPLETADO ===")
|
|
print(f"Total de preguntas generadas: {total_generated}")
|
|
print(f"Fecha activa: {target_date}")
|
|
|
|
|
|
async def check_existing_questions(target_date: date = None):
|
|
"""Check if questions already exist for target date."""
|
|
if target_date is None:
|
|
target_date = date.today() + timedelta(days=1)
|
|
|
|
AsyncSessionLocal = get_async_session()
|
|
async with AsyncSessionLocal() as db:
|
|
result = await db.execute(
|
|
select(Question).where(Question.date_active == target_date)
|
|
)
|
|
existing = result.scalars().all()
|
|
return len(existing)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description="Generar preguntas diarias para Trivy")
|
|
parser.add_argument(
|
|
"--date",
|
|
type=str,
|
|
help="Fecha objetivo (YYYY-MM-DD). Default: mañana"
|
|
)
|
|
parser.add_argument(
|
|
"--force",
|
|
action="store_true",
|
|
help="Generar aunque ya existan preguntas para esa fecha"
|
|
)
|
|
parser.add_argument(
|
|
"--check",
|
|
action="store_true",
|
|
help="Solo verificar si existen preguntas"
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
target_date = None
|
|
if args.date:
|
|
target_date = date.fromisoformat(args.date)
|
|
|
|
if args.check:
|
|
count = asyncio.run(check_existing_questions(target_date))
|
|
check_date = target_date or (date.today() + timedelta(days=1))
|
|
print(f"Preguntas para {check_date}: {count}")
|
|
sys.exit(0)
|
|
|
|
# Check if questions already exist
|
|
if not args.force:
|
|
count = asyncio.run(check_existing_questions(target_date))
|
|
if count > 0:
|
|
check_date = target_date or (date.today() + timedelta(days=1))
|
|
print(f"Ya existen {count} preguntas para {check_date}")
|
|
print("Usa --force para regenerar")
|
|
sys.exit(0)
|
|
|
|
asyncio.run(generate_daily_questions(target_date))
|