kobold-ai
816 lines
32 KiB
Python
Executable File
816 lines
32 KiB
Python
Executable File
import os
|
||
import uuid
|
||
import json
|
||
import random
|
||
import datetime
|
||
from fastapi import FastAPI, Request, Depends, HTTPException, status, Form
|
||
from fastapi.responses import HTMLResponse, RedirectResponse, JSONResponse
|
||
from fastapi.staticfiles import StaticFiles
|
||
from fastapi.templating import Jinja2Templates
|
||
from starlette.middleware.sessions import SessionMiddleware
|
||
from sqlalchemy.orm import Session
|
||
from dotenv import load_dotenv
|
||
|
||
from app.database import get_db
|
||
from app.models import Question, QuizSession, AdminUser, LeaderboardEntry
|
||
from app.matcher import is_answer_correct, levenshtein_distance, normalize_text
|
||
from app.auth import verify_password
|
||
from app.seed import seed_db
|
||
from app.schemas import GameCheckRequest, SuggestionCreateRequest, GameStartRequest, LeaderboardSubmitRequest
|
||
|
||
load_dotenv()
|
||
|
||
app = FastAPI(title="PotterTrivia – Das Harry-Potter-Wissensquiz")
|
||
|
||
SECRET_KEY = os.getenv("SECRET_KEY", "hogwarts_secret_default_key_2026")
|
||
app.add_middleware(SessionMiddleware, secret_key=SECRET_KEY, session_cookie="quiz_session")
|
||
|
||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
app.mount("/static", StaticFiles(directory=os.path.join(BASE_DIR, "static")), name="static")
|
||
templates = Jinja2Templates(directory=os.path.join(BASE_DIR, "templates"))
|
||
|
||
def format_question_for_client(q: Question, number: int, total: int, lang: str = "de") -> dict:
|
||
options_list = None
|
||
if lang == "en" and q.question_en:
|
||
q_text = q.question_en
|
||
opts_raw = q.options_en or q.options
|
||
else:
|
||
q_text = q.question
|
||
opts_raw = q.options
|
||
|
||
if q.question_type == "multiple_choice" and opts_raw:
|
||
try:
|
||
parsed = json.loads(opts_raw)
|
||
if isinstance(parsed, list):
|
||
options_list = list(parsed)
|
||
random.shuffle(options_list)
|
||
except Exception:
|
||
options_list = [opt.strip() for opt in opts_raw.split(",") if opt.strip()]
|
||
random.shuffle(options_list)
|
||
|
||
return {
|
||
"id": q.id,
|
||
"question_number": number,
|
||
"total_questions": total,
|
||
"category": q.category,
|
||
"question": q_text,
|
||
"question_type": q.question_type or "text",
|
||
"options": options_list,
|
||
"difficulty": q.difficulty or "medium"
|
||
}
|
||
|
||
def get_tier_evaluation(score: int, total: int = 10, lang: str = "de") -> dict:
|
||
if total <= 10 and score <= 10:
|
||
pct = (score / total) if total > 0 else 0
|
||
else:
|
||
max_possible = (total * 100) if total <= 10 else total
|
||
pct = (score / max_possible) if max_possible > 0 else 0
|
||
|
||
is_de = (lang != "en")
|
||
|
||
if pct >= 0.90:
|
||
grade = "O"
|
||
title = "🧙♂️ Großmeister der Zauberei" if is_de else "🧙♂️ Grandmaster of Wizardry"
|
||
subtitle = "Note: O (Ohnegleichen)" if is_de else "Grade: O (Outstanding)"
|
||
desc = (
|
||
"Unglaublich! Albus Dumbledore überlegt bereits, dir das Büro des Schulleiters zu überlassen. "
|
||
"Du kennst selbst die geheimsten Winkel von Hogwarts besser als der Fast Kopflose Nick. "
|
||
"Das Zaubereiministerium verneigt sich in Ehrfurcht!"
|
||
if is_de else
|
||
"Incredible! Albus Dumbledore is already considering handing you the headmaster's office. "
|
||
"You know the secret passages of Hogwarts better than Nearly Headless Nick. "
|
||
"The Ministry of Magic bows in deep awe!"
|
||
)
|
||
badge_class = "tier-outstanding"
|
||
elif pct >= 0.70:
|
||
grade = "E"
|
||
title = "⚡ Meister-Hexe / Zauberer" if is_de else "⚡ Master Witch / Wizard"
|
||
subtitle = "Note: E (Erwartungen übertroffen)" if is_de else "Grade: E (Exceeds Expectations)"
|
||
desc = (
|
||
"Hermine Granger nickt anerkennend und Professor McGonagall vergibt 50 Punkte für dein Haus! "
|
||
"Dein Zauberstab zuckt blitzschnell und dein Wissen lässt selbst Ravenclaws erblassen."
|
||
if is_de else
|
||
"Hermione Granger nods in deep approval and Professor McGonagall awards 50 points to your House! "
|
||
"Your wandwork is razor-sharp and your knowledge makes even Ravenclaws jealous."
|
||
)
|
||
badge_class = "tier-exceeds"
|
||
elif pct >= 0.50:
|
||
grade = "A"
|
||
title = "📜 Solider Hogwarts-Schüler" if is_de else "📜 Solid Hogwarts Student"
|
||
subtitle = "Note: A (Annehmbar)" if is_de else "Grade: A (Acceptable)"
|
||
desc = (
|
||
"Bestanden! Ein paar Kessel sind zwar im Zaubertränke-Unterricht explodiert und Peeves hat dich mit Kreide beworfen, "
|
||
"aber du hast dich tapfer durchgeschlagen. Eine solide magische Leistung!"
|
||
if is_de else
|
||
"Passed! A few cauldrons blew up during Potions class and Peeves threw chalk at your head, "
|
||
"but you bravely fought your way through. A solid magical performance!"
|
||
)
|
||
badge_class = "tier-acceptable"
|
||
elif pct >= 0.20:
|
||
grade = "M"
|
||
title = "🧹 Quaffel-Fänger in Not" if is_de else "🧹 Quaffle Catcher in Distress"
|
||
subtitle = "Note: M (Mies)" if is_de else "Grade: P (Poor)"
|
||
desc = (
|
||
"Au weia... Hat Ron Weasley dir seinen mit Tesafilm geklebten Zauberstab geliehen? "
|
||
"Deine Eule hat sich wohl im Nebel verflogen. Zeit für ein paar Extrastunden in der Bibliothek bei Madam Pince!"
|
||
if is_de else
|
||
"Yikes... Did Ron Weasley lend you his Spellotaped wand? "
|
||
"Your owl must have gotten lost in the fog. Time for extra study sessions in the library with Madam Pince!"
|
||
)
|
||
badge_class = "tier-poor"
|
||
else:
|
||
grade = "T"
|
||
title = "🧌 Bergtroll im Kerker" if is_de else "🧌 Mountain Troll in the Dungeon"
|
||
subtitle = "Note: T (Troll)" if is_de else "Grade: T (Troll)"
|
||
desc = (
|
||
"Alarm im Schloss! Du hast mit der Holzkeule auf die Fragen eingeschlagen. "
|
||
"Schnell weg, bevor Hausmeister Filch dich beim lebenslangen Nachsitzen in den Kerkern erwischt!"
|
||
if is_de else
|
||
"Troll in the dungeon! You smashed the questions with a wooden club. "
|
||
"Run quickly before caretaker Filch catches you for lifelong detention in the dungeons!"
|
||
)
|
||
badge_class = "tier-troll"
|
||
|
||
return {
|
||
"grade": grade,
|
||
"title": title,
|
||
"subtitle": subtitle,
|
||
"description": desc,
|
||
"badge_class": badge_class
|
||
}
|
||
|
||
@app.on_event("startup")
|
||
def startup_event():
|
||
seed_db()
|
||
|
||
# --- Public Endpoints ---
|
||
|
||
@app.get("/", response_class=HTMLResponse)
|
||
async def index(request: Request, db: Session = Depends(get_db)):
|
||
active_count = db.query(Question).filter(Question.status == "active").count()
|
||
return templates.TemplateResponse(
|
||
request=request,
|
||
name="index.html",
|
||
context={"question_count": active_count}
|
||
)
|
||
|
||
VALID_CATEGORIES = ["all", "books", "movies", "creatures", "spells", "characters", "hogwarts", "quidditch", "lore", "general"]
|
||
|
||
@app.post("/api/game/start")
|
||
async def start_game(payload: GameStartRequest = None, db: Session = Depends(get_db)):
|
||
lang = payload.lang if payload and payload.lang in ["de", "en"] else "de"
|
||
difficulty = payload.difficulty if payload and payload.difficulty in ["easy", "medium", "hard"] else "all"
|
||
category = payload.category if payload and payload.category in VALID_CATEGORIES else "all"
|
||
|
||
base_query = db.query(Question).filter(Question.status == "active")
|
||
if category != "all":
|
||
if category == "books":
|
||
cat_query = base_query.filter((Question.category == "books") | (Question.question.like("%Kröter%")))
|
||
else:
|
||
cat_query = base_query.filter(Question.category == category)
|
||
else:
|
||
cat_query = base_query
|
||
|
||
if difficulty != "all":
|
||
strict_pool = cat_query.filter(Question.difficulty == difficulty).all()
|
||
if len(strict_pool) >= 10 or category == "all":
|
||
pool = strict_pool
|
||
else:
|
||
# Keep all difficulty matches in this category and top up from the same category
|
||
cat_all = cat_query.all()
|
||
strict_ids = {q.id for q in strict_pool}
|
||
extra = [q for q in cat_all if q.id not in strict_ids]
|
||
random.shuffle(extra)
|
||
pool = strict_pool + extra[:max(0, 10 - len(strict_pool))]
|
||
else:
|
||
pool = cat_query.all()
|
||
|
||
if not pool:
|
||
pool = db.query(Question).filter(Question.status == "active").all()
|
||
if not pool:
|
||
raise HTTPException(status_code=400, detail="Keine aktiven Fragen vorhanden.")
|
||
|
||
recent_set = set(payload.recent_ids or []) if payload else set()
|
||
wrong_set = set(payload.wrong_ids or []) if payload else set()
|
||
|
||
# Priority pool 1: Questions user previously failed (learning pool)
|
||
wrong_pool = [q for q in pool if q.id in wrong_set]
|
||
# Priority pool 2: Fresh questions not seen recently and not in wrong pool
|
||
fresh_pool = [q for q in pool if q.id not in recent_set and q.id not in wrong_set]
|
||
# Priority pool 3: Recent questions fallback
|
||
recent_pool = [q for q in pool if q.id in recent_set and q.id not in wrong_set]
|
||
|
||
selected_questions = []
|
||
# Up to 3 questions from wrong_pool for spaced repetition
|
||
if wrong_pool:
|
||
k_wrong = min(3, len(wrong_pool))
|
||
selected_questions.extend(random.sample(wrong_pool, k_wrong))
|
||
|
||
# Fill remaining from fresh_pool
|
||
needed = min(10, len(pool)) - len(selected_questions)
|
||
if needed > 0 and fresh_pool:
|
||
k_fresh = min(needed, len(fresh_pool))
|
||
selected_questions.extend(random.sample(fresh_pool, k_fresh))
|
||
|
||
# If still needed, fill from recent_pool
|
||
needed = min(10, len(pool)) - len(selected_questions)
|
||
if needed > 0 and recent_pool:
|
||
k_recent = min(needed, len(recent_pool))
|
||
selected_questions.extend(random.sample(recent_pool, k_recent))
|
||
|
||
# Fallback to any remaining from pool
|
||
needed = min(10, len(pool)) - len(selected_questions)
|
||
if needed > 0:
|
||
remaining = [q for q in pool if q not in selected_questions]
|
||
if remaining:
|
||
selected_questions.extend(random.sample(remaining, min(needed, len(remaining))))
|
||
|
||
# Shuffle to mix learning and fresh questions
|
||
random.shuffle(selected_questions)
|
||
question_ids = [q.id for q in selected_questions]
|
||
session_id = str(uuid.uuid4())
|
||
|
||
new_session = QuizSession(
|
||
id=session_id,
|
||
score=0,
|
||
current_question_idx=0,
|
||
question_ids=json.dumps(question_ids),
|
||
attempts_current=0,
|
||
is_finished=False,
|
||
lang=lang,
|
||
difficulty=difficulty,
|
||
created_at=datetime.datetime.utcnow()
|
||
)
|
||
db.add(new_session)
|
||
db.commit()
|
||
|
||
first_q = selected_questions[0]
|
||
res = format_question_for_client(first_q, 1, len(question_ids), lang=lang)
|
||
res.update({
|
||
"session_id": session_id,
|
||
"score": 0,
|
||
"attempts_used": 0,
|
||
"max_attempts": 3,
|
||
"lang": lang,
|
||
"difficulty": difficulty
|
||
})
|
||
return res
|
||
|
||
@app.post("/api/game/check")
|
||
async def check_answer(payload: GameCheckRequest, db: Session = Depends(get_db)):
|
||
session = db.query(QuizSession).filter(QuizSession.id == payload.session_id).first()
|
||
if not session or session.is_finished:
|
||
raise HTTPException(status_code=400, detail="Ungültige oder beendete Spiel-Session.")
|
||
|
||
session_lang = session.lang or "de"
|
||
question_ids = json.loads(session.question_ids)
|
||
if session.current_question_idx >= len(question_ids):
|
||
session.is_finished = True
|
||
session.finished_at = datetime.datetime.utcnow()
|
||
if session.created_at:
|
||
session.duration_seconds = max(1, int((session.finished_at - session.created_at).total_seconds()))
|
||
db.commit()
|
||
return {
|
||
"correct": False,
|
||
"score": session.score,
|
||
"points_earned": 0,
|
||
"question_id": None,
|
||
"attempts_used": session.attempts_current,
|
||
"attempts_left": 0,
|
||
"is_finished": True,
|
||
"duration_seconds": session.duration_seconds,
|
||
"next_question": None
|
||
}
|
||
|
||
current_q_id = question_ids[session.current_question_idx]
|
||
current_q = db.query(Question).filter(Question.id == current_q_id).first()
|
||
if not current_q:
|
||
raise HTTPException(status_code=500, detail="Frage nicht in Datenbank gefunden.")
|
||
|
||
correct = is_answer_correct(payload.answer, current_q.canonical_answer, current_q.accepted_answers, current_q.answer_en)
|
||
|
||
if correct:
|
||
# Tiered points based on attempt
|
||
if session.attempts_current == 0:
|
||
points_earned = 100
|
||
elif session.attempts_current == 1:
|
||
points_earned = 70
|
||
else:
|
||
points_earned = 40
|
||
|
||
session.score += points_earned
|
||
session.current_question_idx += 1
|
||
session.attempts_current = 0
|
||
|
||
is_finished = session.current_question_idx >= len(question_ids)
|
||
if is_finished:
|
||
session.is_finished = True
|
||
session.finished_at = datetime.datetime.utcnow()
|
||
if session.created_at:
|
||
session.duration_seconds = max(1, int((session.finished_at - session.created_at).total_seconds()))
|
||
|
||
db.commit()
|
||
|
||
next_q_data = None
|
||
if not is_finished:
|
||
next_q = db.query(Question).filter(Question.id == question_ids[session.current_question_idx]).first()
|
||
if next_q:
|
||
next_q_data = format_question_for_client(next_q, session.current_question_idx + 1, len(question_ids), lang=session_lang)
|
||
|
||
return {
|
||
"correct": True,
|
||
"score": session.score,
|
||
"points_earned": points_earned,
|
||
"question_id": current_q_id,
|
||
"attempts_used": 0,
|
||
"attempts_left": 3,
|
||
"is_finished": is_finished,
|
||
"duration_seconds": session.duration_seconds,
|
||
"revealed_answer": None,
|
||
"next_question": next_q_data
|
||
}
|
||
else:
|
||
session.attempts_current += 1
|
||
attempts_left = max(0, 3 - session.attempts_current)
|
||
|
||
if attempts_left == 0:
|
||
if session_lang == "en" and current_q.answer_en:
|
||
revealed = current_q.answer_en
|
||
else:
|
||
revealed = current_q.canonical_answer
|
||
|
||
session.current_question_idx += 1
|
||
session.attempts_current = 0
|
||
|
||
is_finished = session.current_question_idx >= len(question_ids)
|
||
if is_finished:
|
||
session.is_finished = True
|
||
session.finished_at = datetime.datetime.utcnow()
|
||
if session.created_at:
|
||
session.duration_seconds = max(1, int((session.finished_at - session.created_at).total_seconds()))
|
||
|
||
db.commit()
|
||
|
||
next_q_data = None
|
||
if not is_finished:
|
||
next_q = db.query(Question).filter(Question.id == question_ids[session.current_question_idx]).first()
|
||
if next_q:
|
||
next_q_data = format_question_for_client(next_q, session.current_question_idx + 1, len(question_ids), lang=session_lang)
|
||
|
||
return {
|
||
"correct": False,
|
||
"score": session.score,
|
||
"points_earned": 0,
|
||
"question_id": current_q_id,
|
||
"attempts_used": 3,
|
||
"attempts_left": 0,
|
||
"is_finished": is_finished,
|
||
"duration_seconds": session.duration_seconds,
|
||
"revealed_answer": revealed,
|
||
"next_question": next_q_data
|
||
}
|
||
else:
|
||
db.commit()
|
||
return {
|
||
"correct": False,
|
||
"score": session.score,
|
||
"points_earned": 0,
|
||
"question_id": current_q_id,
|
||
"attempts_used": session.attempts_current,
|
||
"attempts_left": attempts_left,
|
||
"is_finished": False,
|
||
"revealed_answer": None,
|
||
"next_question": None
|
||
}
|
||
|
||
@app.get("/api/game/hint")
|
||
async def get_hint(session_id: str, db: Session = Depends(get_db)):
|
||
session = db.query(QuizSession).filter(QuizSession.id == session_id).first()
|
||
if not session or session.is_finished:
|
||
raise HTTPException(status_code=400, detail="Ungültige Session.")
|
||
|
||
session_lang = session.lang or "de"
|
||
question_ids = json.loads(session.question_ids)
|
||
if session.current_question_idx >= len(question_ids):
|
||
return {"hint": "No hint available." if session_lang == "en" else "Kein Hinweis verfügbar."}
|
||
|
||
current_q_id = question_ids[session.current_question_idx]
|
||
current_q = db.query(Question).filter(Question.id == current_q_id).first()
|
||
if not current_q:
|
||
return {"hint": "No hint available." if session_lang == "en" else "Kein Hinweis verfügbar."}
|
||
|
||
if session_lang == "en":
|
||
hint_text = current_q.hint_en if current_q.hint_en else (current_q.hint or "No hint available.")
|
||
else:
|
||
hint_text = current_q.hint or "Kein Tipp verfügbar."
|
||
|
||
return {"hint": hint_text}
|
||
|
||
@app.get("/finale", response_class=HTMLResponse)
|
||
async def finale(request: Request, score: int = 0, total: int = 10, time: int = 0, lang: str = "de", diff: str = "all"):
|
||
max_score = 1000 if (total <= 10 and score > 10) else (total * 100 if total <= 10 else total)
|
||
tier = get_tier_evaluation(score, total, lang)
|
||
if time >= 60:
|
||
minutes = time // 60
|
||
secs = time % 60
|
||
time_str = f"{minutes} Min. {secs} Sek." if lang != "en" else f"{minutes} min. {secs} sec."
|
||
else:
|
||
time_str = f"{time} Sek." if lang != "en" else f"{time} sec."
|
||
|
||
clean_diff = diff if diff in ["easy", "medium", "hard"] else "all"
|
||
|
||
return templates.TemplateResponse(
|
||
request=request,
|
||
name="finale.html",
|
||
context={
|
||
"score": score,
|
||
"total": total,
|
||
"max_score": max_score,
|
||
"time_seconds": time,
|
||
"time_formatted": time_str,
|
||
"lang": lang,
|
||
"difficulty": clean_diff,
|
||
"tier": tier
|
||
}
|
||
)
|
||
|
||
@app.get("/bestenliste", response_class=HTMLResponse)
|
||
async def bestenliste_page(request: Request, lang: str = "de"):
|
||
return templates.TemplateResponse(
|
||
request=request,
|
||
name="bestenliste.html",
|
||
context={"lang": lang}
|
||
)
|
||
|
||
@app.get("/impressum", response_class=HTMLResponse)
|
||
async def impressum_page(request: Request, lang: str = "de"):
|
||
return templates.TemplateResponse(
|
||
request=request,
|
||
name="impressum.html",
|
||
context={"lang": lang}
|
||
)
|
||
|
||
@app.get("/datenschutz", response_class=HTMLResponse)
|
||
async def datenschutz_page(request: Request, lang: str = "de"):
|
||
return templates.TemplateResponse(
|
||
request=request,
|
||
name="datenschutz.html",
|
||
context={"lang": lang}
|
||
)
|
||
|
||
@app.post("/api/leaderboard")
|
||
async def submit_leaderboard(payload: LeaderboardSubmitRequest, db: Session = Depends(get_db)):
|
||
nick = payload.nickname.strip()
|
||
if not nick:
|
||
wizard_adjectives = ["Mutiger", "Schlauer", "Weiser", "Flinke", "Listiger", "Fliegender"]
|
||
wizard_nouns = ["Greif", "Phönix", "Zauberer", "Drache", "Niffler", "Thestral"]
|
||
nick = f"{random.choice(wizard_adjectives)}_{random.choice(wizard_nouns)}_{random.randint(10, 99)}"
|
||
nick = nick[:30]
|
||
|
||
time_sec = max(1, payload.time_seconds)
|
||
diff = payload.difficulty if payload.difficulty in ["easy", "medium", "hard"] else "all"
|
||
|
||
entry = LeaderboardEntry(
|
||
nickname=nick,
|
||
score=payload.score,
|
||
total_questions=payload.total_questions or 10,
|
||
time_seconds=time_sec,
|
||
difficulty=diff,
|
||
lang=payload.lang or "de",
|
||
session_id=payload.session_id,
|
||
created_at=datetime.datetime.utcnow()
|
||
)
|
||
db.add(entry)
|
||
db.commit()
|
||
db.refresh(entry)
|
||
|
||
better_query = db.query(LeaderboardEntry)
|
||
if entry.difficulty and entry.difficulty != "all":
|
||
better_query = better_query.filter(LeaderboardEntry.difficulty == entry.difficulty)
|
||
|
||
better_count = better_query.filter(
|
||
(LeaderboardEntry.score > entry.score) |
|
||
((LeaderboardEntry.score == entry.score) & (LeaderboardEntry.time_seconds < entry.time_seconds)) |
|
||
((LeaderboardEntry.score == entry.score) & (LeaderboardEntry.time_seconds == entry.time_seconds) & (LeaderboardEntry.id < entry.id))
|
||
).count()
|
||
rank = better_count + 1
|
||
|
||
return {
|
||
"success": True,
|
||
"id": entry.id,
|
||
"entry_id": entry.id,
|
||
"rank": rank,
|
||
"nickname": entry.nickname,
|
||
"score": entry.score,
|
||
"total_questions": entry.total_questions,
|
||
"time_seconds": entry.time_seconds
|
||
}
|
||
|
||
@app.get("/api/leaderboard")
|
||
async def get_leaderboard(difficulty: str = None, limit: int = 20, db: Session = Depends(get_db)):
|
||
query = db.query(LeaderboardEntry)
|
||
if difficulty and difficulty != "all":
|
||
query = query.filter(LeaderboardEntry.difficulty == difficulty)
|
||
|
||
entries = query.order_by(
|
||
LeaderboardEntry.score.desc(),
|
||
LeaderboardEntry.time_seconds.asc(),
|
||
LeaderboardEntry.created_at.asc()
|
||
).limit(min(100, limit)).all()
|
||
|
||
result = []
|
||
for idx, e in enumerate(entries, 1):
|
||
m = e.time_seconds // 60
|
||
s = e.time_seconds % 60
|
||
time_str = f"{m:02d}:{s:02d}"
|
||
result.append({
|
||
"rank": idx,
|
||
"id": e.id,
|
||
"nickname": e.nickname,
|
||
"score": e.score,
|
||
"total_questions": e.total_questions,
|
||
"time_seconds": e.time_seconds,
|
||
"time_formatted": time_str,
|
||
"difficulty": e.difficulty,
|
||
"date": e.created_at.strftime("%d.%m.%Y") if e.created_at else ""
|
||
})
|
||
return result
|
||
|
||
@app.get("/vorschlag", response_class=HTMLResponse)
|
||
async def vorschlag_page(request: Request):
|
||
return templates.TemplateResponse(
|
||
request=request,
|
||
name="vorschlag.html",
|
||
context={}
|
||
)
|
||
|
||
@app.post("/api/suggestions")
|
||
async def submit_suggestion(payload: SuggestionCreateRequest, db: Session = Depends(get_db)):
|
||
if not payload.frage.strip() or not payload.antwort.strip():
|
||
raise HTTPException(status_code=400, detail="Frage und Antwort dürfen nicht leer sein.")
|
||
|
||
options_json = None
|
||
q_type = payload.question_type or "text"
|
||
if q_type == "multiple_choice":
|
||
opts = [payload.antwort.strip()]
|
||
if payload.falsche_antworten:
|
||
for opt in payload.falsche_antworten:
|
||
if opt and opt.strip():
|
||
opts.append(opt.strip())
|
||
options_json = json.dumps(opts, ensure_ascii=False)
|
||
|
||
chosen_cat = payload.category if payload.category in ["books", "movies", "hogwarts", "characters", "spells", "creatures", "quidditch", "lore", "general"] else "general"
|
||
diff = payload.difficulty if payload.difficulty in ["easy", "medium", "hard"] else "medium"
|
||
|
||
new_suggestion = Question(
|
||
category=chosen_cat,
|
||
question=payload.frage.strip(),
|
||
canonical_answer=payload.antwort.strip(),
|
||
hint=payload.tipp.strip() if payload.tipp else None,
|
||
question_type=q_type,
|
||
options=options_json,
|
||
difficulty=diff,
|
||
status="needs_review", # User Question Need Test / Review
|
||
submitted_by=payload.name.strip() if payload.name else "Gast",
|
||
sort_order=999
|
||
)
|
||
db.add(new_suggestion)
|
||
db.commit()
|
||
db.refresh(new_suggestion)
|
||
|
||
return {
|
||
"success": True,
|
||
"id": new_suggestion.id,
|
||
"message": "Vielen Dank! Dein Vorschlag wurde eingereicht und wird von uns geprüft, bevor er im Quiz erscheint."
|
||
}
|
||
|
||
# --- Admin Endpoints ---
|
||
|
||
def get_current_admin(request: Request) -> str:
|
||
admin = request.session.get("admin_user")
|
||
if not admin:
|
||
raise HTTPException(status_code=status.HTTP_303_SEE_OTHER, headers={"Location": "/admin/login"})
|
||
return admin
|
||
|
||
@app.get("/admin/login", response_class=HTMLResponse)
|
||
async def admin_login_page(request: Request):
|
||
if request.session.get("admin_user"):
|
||
return RedirectResponse(url="/admin", status_code=303)
|
||
return templates.TemplateResponse(
|
||
request=request,
|
||
name="admin_login.html",
|
||
context={"error": None}
|
||
)
|
||
|
||
@app.post("/admin/login")
|
||
async def admin_login(
|
||
request: Request,
|
||
username: str = Form(...),
|
||
password: str = Form(...),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
admin = db.query(AdminUser).filter(AdminUser.username == username.strip()).first()
|
||
if not admin or not verify_password(admin.password_hash, password):
|
||
return templates.TemplateResponse(
|
||
request=request,
|
||
name="admin_login.html",
|
||
context={"error": "Ungültiger Benutzername oder Zauberspruch (Passwort)."},
|
||
status_code=400
|
||
)
|
||
|
||
request.session["admin_user"] = admin.username
|
||
return RedirectResponse(url="/admin", status_code=303)
|
||
|
||
@app.get("/admin/logout")
|
||
async def admin_logout(request: Request):
|
||
request.session.clear()
|
||
return RedirectResponse(url="/admin/login", status_code=303)
|
||
|
||
@app.get("/admin", response_class=HTMLResponse)
|
||
async def admin_dashboard(request: Request, db: Session = Depends(get_db)):
|
||
if not request.session.get("admin_user"):
|
||
return RedirectResponse(url="/admin/login", status_code=303)
|
||
|
||
active_questions = db.query(Question).filter(Question.status == "active").order_by(Question.sort_order.asc(), Question.id.asc()).all()
|
||
pending_questions = db.query(Question).filter(Question.status == "needs_review").order_by(Question.created_at.desc()).all()
|
||
rejected_questions = db.query(Question).filter(Question.status == "rejected").order_by(Question.created_at.desc()).all()
|
||
|
||
return templates.TemplateResponse(
|
||
request=request,
|
||
name="admin.html",
|
||
context={
|
||
"admin_user": request.session["admin_user"],
|
||
"active_questions": active_questions,
|
||
"pending_questions": pending_questions,
|
||
"rejected_questions": rejected_questions
|
||
}
|
||
)
|
||
|
||
@app.post("/admin/questions/new")
|
||
async def admin_add_question(
|
||
request: Request,
|
||
question: str = Form(...),
|
||
canonical_answer: str = Form(...),
|
||
hint: str = Form(None),
|
||
category: str = Form("harry_potter"),
|
||
question_type: str = Form("text"),
|
||
difficulty: str = Form("medium"),
|
||
question_en: str = Form(None),
|
||
answer_en: str = Form(None),
|
||
hint_en: str = Form(None),
|
||
wrong_option_1: str = Form(None),
|
||
wrong_option_2: str = Form(None),
|
||
wrong_option_3: str = Form(None),
|
||
sort_order: int = Form(0),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
if not request.session.get("admin_user"):
|
||
return RedirectResponse(url="/admin/login", status_code=303)
|
||
|
||
options_json = None
|
||
if question_type == "multiple_choice":
|
||
opts = [canonical_answer.strip()]
|
||
for w in [wrong_option_1, wrong_option_2, wrong_option_3]:
|
||
if w and w.strip():
|
||
opts.append(w.strip())
|
||
options_json = json.dumps(opts, ensure_ascii=False)
|
||
|
||
diff = difficulty.strip() if difficulty and difficulty.strip() in ["easy", "medium", "hard"] else "medium"
|
||
|
||
new_q = Question(
|
||
category=category.strip(),
|
||
question=question.strip(),
|
||
canonical_answer=canonical_answer.strip(),
|
||
hint=hint.strip() if hint else None,
|
||
question_type=question_type.strip(),
|
||
options=options_json,
|
||
difficulty=diff,
|
||
question_en=question_en.strip() if question_en else None,
|
||
answer_en=answer_en.strip() if answer_en else None,
|
||
hint_en=hint_en.strip() if hint_en else None,
|
||
status="active",
|
||
sort_order=sort_order,
|
||
submitted_by=f"Admin ({request.session['admin_user']})"
|
||
)
|
||
db.add(new_q)
|
||
db.commit()
|
||
return RedirectResponse(url="/admin", status_code=303)
|
||
|
||
@app.post("/admin/questions/{qid}/approve")
|
||
async def admin_approve_question(qid: int, request: Request, db: Session = Depends(get_db)):
|
||
if not request.session.get("admin_user"):
|
||
return JSONResponse({"error": "Unauthorized"}, status_code=401)
|
||
|
||
q = db.query(Question).filter(Question.id == qid).first()
|
||
if not q:
|
||
raise HTTPException(status_code=404, detail="Frage nicht gefunden.")
|
||
|
||
# Determine highest sort_order
|
||
max_order = db.query(Question.sort_order).filter(Question.status == "active").order_by(Question.sort_order.desc()).first()
|
||
next_order = (max_order[0] + 1) if max_order and max_order[0] else 1
|
||
|
||
q.status = "active"
|
||
q.sort_order = next_order
|
||
q.reviewed_at = datetime.datetime.utcnow()
|
||
q.reviewed_by = request.session.get("admin_user")
|
||
db.commit()
|
||
return {"success": True, "message": f"Frage #{qid} wurde erfolgreich freigeschaltet!"}
|
||
|
||
@app.post("/admin/questions/{qid}/reject")
|
||
async def admin_reject_question(qid: int, request: Request, db: Session = Depends(get_db)):
|
||
if not request.session.get("admin_user"):
|
||
return JSONResponse({"error": "Unauthorized"}, status_code=401)
|
||
|
||
q = db.query(Question).filter(Question.id == qid).first()
|
||
if not q:
|
||
raise HTTPException(status_code=404, detail="Frage nicht gefunden.")
|
||
|
||
q.status = "rejected"
|
||
q.reviewed_at = datetime.datetime.utcnow()
|
||
q.reviewed_by = request.session.get("admin_user")
|
||
db.commit()
|
||
return {"success": True, "message": f"Frage #{qid} wurde abgelehnt."}
|
||
|
||
@app.post("/admin/questions/{qid}/edit")
|
||
async def admin_edit_question(
|
||
qid: int,
|
||
request: Request,
|
||
question: str = Form(...),
|
||
canonical_answer: str = Form(...),
|
||
hint: str = Form(None),
|
||
question_type: str = Form("text"),
|
||
difficulty: str = Form("medium"),
|
||
question_en: str = Form(None),
|
||
answer_en: str = Form(None),
|
||
hint_en: str = Form(None),
|
||
options_raw: str = Form(None),
|
||
status_val: str = Form("active"),
|
||
sort_order: int = Form(0),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
if not request.session.get("admin_user"):
|
||
return RedirectResponse(url="/admin/login", status_code=303)
|
||
|
||
q = db.query(Question).filter(Question.id == qid).first()
|
||
if not q:
|
||
raise HTTPException(status_code=404, detail="Frage nicht gefunden.")
|
||
|
||
q.question = question.strip()
|
||
q.canonical_answer = canonical_answer.strip()
|
||
q.hint = hint.strip() if hint else None
|
||
q.question_type = question_type.strip()
|
||
if difficulty and difficulty.strip() in ["easy", "medium", "hard"]:
|
||
q.difficulty = difficulty.strip()
|
||
q.question_en = question_en.strip() if question_en else None
|
||
q.answer_en = answer_en.strip() if answer_en else None
|
||
q.hint_en = hint_en.strip() if hint_en else None
|
||
|
||
if options_raw and options_raw.strip():
|
||
val = options_raw.strip()
|
||
if val.startswith("[") and val.endswith("]"):
|
||
q.options = val
|
||
else:
|
||
items = [x.strip() for x in val.split(",") if x.strip()]
|
||
q.options = json.dumps(items, ensure_ascii=False)
|
||
else:
|
||
q.options = None
|
||
|
||
q.status = status_val.strip()
|
||
q.sort_order = sort_order
|
||
db.commit()
|
||
return RedirectResponse(url="/admin", status_code=303)
|
||
|
||
@app.post("/admin/questions/{qid}/delete")
|
||
async def admin_delete_question(qid: int, request: Request, db: Session = Depends(get_db)):
|
||
if not request.session.get("admin_user"):
|
||
return JSONResponse({"error": "Unauthorized"}, status_code=401)
|
||
|
||
q = db.query(Question).filter(Question.id == qid).first()
|
||
if not q:
|
||
raise HTTPException(status_code=404, detail="Frage nicht gefunden.")
|
||
|
||
db.delete(q)
|
||
db.commit()
|
||
return {"success": True, "message": f"Frage #{qid} gelöscht."}
|
||
|
||
@app.post("/admin/questions/test-match")
|
||
async def admin_test_match(
|
||
request: Request,
|
||
test_input: str = Form(...),
|
||
canonical: str = Form(...)
|
||
):
|
||
if not request.session.get("admin_user"):
|
||
return JSONResponse({"error": "Unauthorized"}, status_code=401)
|
||
|
||
matched = is_answer_correct(test_input, canonical)
|
||
clean_in = normalize_text(test_input)
|
||
clean_target = normalize_text(canonical)
|
||
dist = levenshtein_distance(clean_in, clean_target)
|
||
|
||
return {
|
||
"matched": matched,
|
||
"input_clean": clean_in,
|
||
"target_clean": clean_target,
|
||
"levenshtein_distance": dist
|
||
}
|