#!/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))