81 lines
2.5 KiB
Python
81 lines
2.5 KiB
Python
"""리더보드 API — 누적 포인트 1위 (docs/SCORING.md §3 랭킹 규칙).
|
|
|
|
1차 정렬: 누적 포인트. 동점: 정확스코어 횟수 → 적중률 → 참여수 → 최초도달.
|
|
주최측/임직원 배제(P1). 이메일은 마스킹하여 노출.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from collections import defaultdict
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from ..database import get_db
|
|
from ..models import UserPrediction
|
|
from ..scoring import SCORE, is_excluded
|
|
from ..schemas import LeaderboardOut, StandingOut
|
|
|
|
router = APIRouter(prefix="/api/leaderboard", tags=["leaderboard"])
|
|
|
|
|
|
def _mask(email: str) -> str:
|
|
name, _, domain = email.partition("@")
|
|
head = name[:2] if len(name) >= 2 else name
|
|
return f"{head}{'*' * max(1, len(name) - 2)}@{domain}"
|
|
|
|
|
|
@router.get("", response_model=LeaderboardOut)
|
|
async def leaderboard(
|
|
limit: int = Query(50, ge=1, le=200), db: AsyncSession = Depends(get_db)
|
|
) -> LeaderboardOut:
|
|
rows = (
|
|
await db.execute(
|
|
select(UserPrediction).where(
|
|
UserPrediction.points.is_not(None),
|
|
UserPrediction.email.is_not(None),
|
|
)
|
|
)
|
|
).scalars().all()
|
|
|
|
agg: dict[str, dict] = defaultdict(
|
|
lambda: {"points": 0, "exact": 0, "played": 0, "first": None}
|
|
)
|
|
matches: set[str] = set()
|
|
for r in rows:
|
|
email = (r.email or "").strip().lower()
|
|
if not email or is_excluded(email):
|
|
continue
|
|
matches.add(r.match_id)
|
|
a = agg[email]
|
|
a["points"] += r.points or 0
|
|
a["played"] += 1
|
|
if (r.points or 0) >= SCORE["EXACT"]:
|
|
a["exact"] += 1
|
|
ts = r.scored_at or r.created_at
|
|
if a["first"] is None or (ts and ts < a["first"]):
|
|
a["first"] = ts
|
|
|
|
ranked = sorted(
|
|
agg.items(),
|
|
key=lambda kv: (
|
|
-kv[1]["points"],
|
|
-kv[1]["exact"],
|
|
-(kv[1]["points"] / kv[1]["played"] if kv[1]["played"] else 0),
|
|
-kv[1]["played"],
|
|
kv[1]["first"].timestamp() if kv[1]["first"] else 0,
|
|
),
|
|
)
|
|
|
|
standings = [
|
|
StandingOut(
|
|
rank=i + 1,
|
|
emailMasked=_mask(email),
|
|
totalPoints=v["points"],
|
|
exactCount=v["exact"],
|
|
matchesPlayed=v["played"],
|
|
)
|
|
for i, (email, v) in enumerate(ranked[:limit])
|
|
]
|
|
return LeaderboardOut(standings=standings, scoredMatches=len(matches))
|