# retoor <retoor@molodetz.nl>
from __future__ import annotations
from devplacepy.cache import TTLCache
from devplacepy.config import QUIZ_SCOREBOARD_CACHE_SECONDS, QUIZ_SCOREBOARD_LIMIT
from devplacepy.database import _in_clause, db, get_users_by_uids
_scoreboard_cache = TTLCache(ttl=QUIZ_SCOREBOARD_CACHE_SECONDS, max_size=8)
BEST_ATTEMPTS_SQL = """
SELECT a.user_uid AS user_uid,
a.quiz_uid AS quiz_uid,
MAX(a.score_points) AS best_points,
MAX(a.score_percent) AS best_percent,
MAX(q.total_points) AS quiz_points
FROM quiz_attempts a
JOIN quizzes q ON q.uid = a.quiz_uid
WHERE a.deleted_at IS NULL
AND a.status = 'completed'
AND q.deleted_at IS NULL
AND q.status = 'published'
GROUP BY a.user_uid, a.quiz_uid
"""
def _totals() -> list[dict]:
if "quiz_attempts" not in db.tables or "quizzes" not in db.tables:
return []
rows = db.query(
f"SELECT user_uid, "
f"SUM(best_points) AS total_points, "
f"COUNT(*) AS quizzes_completed, "
f"AVG(best_percent) AS avg_percent, "
f"SUM(CASE WHEN best_percent >= 100 THEN 1 ELSE 0 END) AS perfect_count "
f"FROM ({BEST_ATTEMPTS_SQL}) best "
f"GROUP BY user_uid"
)
entries = [
{
"user_uid": row["user_uid"],
"total_points": round(float(row.get("total_points") or 0.0), 2),
"quizzes_completed": int(row.get("quizzes_completed") or 0),
"avg_percent": round(float(row.get("avg_percent") or 0.0), 2),
"perfect_count": int(row.get("perfect_count") or 0),
}
for row in rows
]
entries.sort(
key=lambda entry: (
-entry["total_points"],
-entry["quizzes_completed"],
entry["user_uid"],
)
)
for rank, entry in enumerate(entries, start=1):
entry["rank"] = rank
return entries
def _ranked() -> list[dict]:
cached = _scoreboard_cache.get("ranked")
if cached is not None:
return cached
entries = _totals()
_scoreboard_cache.set("ranked", entries)
return entries
def scoreboard(limit: int = QUIZ_SCOREBOARD_LIMIT) -> list[dict]:
entries = _ranked()[: max(1, int(limit or QUIZ_SCOREBOARD_LIMIT))]
users = get_users_by_uids([entry["user_uid"] for entry in entries])
return [
{**entry, "user": users.get(entry["user_uid"])}
for entry in entries
if users.get(entry["user_uid"])
]
def standing_for(user_uid: str) -> dict | None:
if not user_uid:
return None
for entry in _ranked():
if entry["user_uid"] == user_uid:
users = get_users_by_uids([user_uid])
return {**entry, "user": users.get(user_uid)}
return None
def clear_cache() -> None:
_scoreboard_cache.pop("ranked")
def completed_quiz_uids(user_uid: str) -> list[str]:
if not user_uid or "quiz_attempts" not in db.tables:
return []
rows = db.query(
"SELECT DISTINCT quiz_uid FROM quiz_attempts "
"WHERE user_uid = :user AND status = 'completed' AND deleted_at IS NULL",
user=user_uid,
)
return [row["quiz_uid"] for row in rows if row.get("quiz_uid")]
def attempt_states_for(user_uid: str, quiz_uids: list[str]) -> dict[str, dict]:
uids = [uid for uid in (quiz_uids or []) if uid]
if not user_uid or not uids or "quiz_attempts" not in db.tables:
return {}
placeholders, params = _in_clause(uids)
params["user"] = user_uid
rows = db.query(
f"SELECT quiz_uid, uid, status, score_points, score_percent, completed_at "
f"FROM quiz_attempts WHERE user_uid = :user AND deleted_at IS NULL "
f"AND quiz_uid IN ({placeholders}) ORDER BY created_at",
**params,
)
states: dict[str, dict] = {}
for row in rows:
quiz_uid = row["quiz_uid"]
state = states.setdefault(
quiz_uid,
{
"state": "todo",
"best_percent": 0.0,
"best_points": 0.0,
"attempt_uid": "",
"completed_at": "",
},
)
status = row.get("status") or ""
if status == "in_progress":
state["state"] = "in_progress"
state["attempt_uid"] = row["uid"]
elif status == "completed":
percent = float(row.get("score_percent") or 0.0)
if state["state"] != "in_progress":
state["state"] = "done"
if percent >= state["best_percent"] or not state["completed_at"]:
state["best_percent"] = percent
state["best_points"] = round(float(row.get("score_points") or 0.0), 2)
state["completed_at"] = row.get("completed_at") or ""
if state["state"] == "done":
state["attempt_uid"] = row["uid"]
return states
def quiz_leaderboard(quiz_uid: str, limit: int = 25) -> list[dict]:
if not quiz_uid or "quiz_attempts" not in db.tables:
return []
rows = list(
db.query(
"SELECT uid, user_uid, score_points, score_percent, passed, completed_at "
"FROM quiz_attempts WHERE quiz_uid = :quiz AND status = 'completed' "
"AND deleted_at IS NULL ORDER BY score_percent DESC, completed_at DESC "
"LIMIT :limit",
quiz=quiz_uid,
limit=max(1, int(limit or 25)),
)
)
users = get_users_by_uids([row["user_uid"] for row in rows])
entries = []
for rank, row in enumerate(rows, start=1):
user = users.get(row["user_uid"])
if not user:
continue
entries.append(
{
"rank": rank,
"user": user,
"score_points": round(float(row.get("score_points") or 0.0), 2),
"score_percent": round(float(row.get("score_percent") or 0.0), 2),
"passed": bool(int(row.get("passed") or 0)),
"completed_at": row.get("completed_at") or "",
}
)
return entries
def _count(sql: str, **params) -> int:
for row in db.query(sql, **params):
return int(row.get("total") or 0)
return 0
def progress_for(user_uid: str) -> dict:
standing = standing_for(user_uid) or {}
completed = int(standing.get("quizzes_completed") or 0)
published = 0
if "quizzes" in db.tables:
published = _count(
"SELECT COUNT(*) AS total FROM quizzes "
"WHERE status = 'published' AND deleted_at IS NULL"
)
return {
"completed": completed,
"todo": max(0, published - completed),
"avg_percent": standing.get("avg_percent", 0.0),
"total_points": standing.get("total_points", 0.0),
"rank": standing.get("rank", 0),
"perfect_count": standing.get("perfect_count", 0),
}
View Git Blame Copy Permalink