65 lines
2.0 KiB
Python
65 lines
2.0 KiB
Python
"""리더보드 API — 누적 포인트 1위 (docs/SCORING.md §3 랭킹 규칙).
|
|
|
|
user_points 테이블(채점 시 누적 갱신) 기준 랭킹.
|
|
1차 정렬: 누적 포인트. 동점: 정확스코어 횟수 → 적중률 → 참여수 → 최초도달.
|
|
주최측/임직원 배제(P1). 이메일은 마스킹하여 노출.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from ..database import get_db
|
|
from ..models import Match, UserPoints
|
|
from ..scoring import 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(UserPoints))
|
|
).scalars().all()
|
|
|
|
ranked = sorted(
|
|
(r for r in rows if not is_excluded(r.email)),
|
|
key=lambda r: (
|
|
-r.total_points,
|
|
-r.exact_count,
|
|
-(r.total_points / r.matches_played if r.matches_played else 0),
|
|
-r.matches_played,
|
|
r.first_scored_at.timestamp() if r.first_scored_at else 0,
|
|
),
|
|
)
|
|
|
|
scored_matches = (
|
|
await db.execute(
|
|
select(func.count())
|
|
.select_from(Match)
|
|
.where(Match.result_outcome.is_not(None))
|
|
)
|
|
).scalar() or 0
|
|
|
|
standings = [
|
|
StandingOut(
|
|
rank=i + 1,
|
|
emailMasked=_mask(r.email),
|
|
totalPoints=r.total_points,
|
|
exactCount=r.exact_count,
|
|
matchesPlayed=r.matches_played,
|
|
)
|
|
for i, r in enumerate(ranked[:limit])
|
|
]
|
|
return LeaderboardOut(standings=standings, scoredMatches=scored_matches)
|