166 lines
5.7 KiB
Python
166 lines
5.7 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 sqlalchemy.orm import selectinload
|
||
|
||
from ..database import get_db
|
||
from ..models import Match, UserPoints, UserPrediction
|
||
from ..scoring import is_excluded, score_prediction
|
||
from ..schemas import (
|
||
AILeaderboardOut,
|
||
AIStandingOut,
|
||
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),
|
||
league: str = Query("", description="wc | kbo | mlb — 빈값이면 전체 합산"),
|
||
db: AsyncSession = Depends(get_db),
|
||
) -> LeaderboardOut:
|
||
if league:
|
||
# 리그별 랭킹 — 채점된 픽(points)을 리그 경기로 한정해 이메일별 재집계.
|
||
picks = (
|
||
await db.execute(
|
||
select(UserPrediction, Match)
|
||
.join(Match, Match.match_id == UserPrediction.match_id)
|
||
.where(
|
||
Match.league == league,
|
||
Match.result_outcome.is_not(None),
|
||
UserPrediction.points.is_not(None),
|
||
UserPrediction.email.is_not(None),
|
||
)
|
||
)
|
||
).all()
|
||
agg: dict[str, list] = {} # email → [pts, exact, played, first_ts]
|
||
for pk, m in picks:
|
||
row = agg.setdefault(pk.email, [0, 0, 0, None])
|
||
row[0] += pk.points or 0
|
||
row[1] += 1 if (
|
||
pk.score_a == m.result_score_a and pk.score_b == m.result_score_b
|
||
) else 0
|
||
row[2] += 1
|
||
ranked_rows = sorted(
|
||
((e, v) for e, v in agg.items() if not is_excluded(e)),
|
||
key=lambda kv: (-kv[1][0], -kv[1][1], -kv[1][2]),
|
||
)
|
||
scored_matches = (
|
||
await db.execute(
|
||
select(func.count()).select_from(Match).where(
|
||
Match.league == league, Match.result_outcome.is_not(None)
|
||
)
|
||
)
|
||
).scalar() or 0
|
||
standings = [
|
||
StandingOut(
|
||
rank=i + 1, emailMasked=_mask(e),
|
||
totalPoints=v[0], exactCount=v[1], matchesPlayed=v[2],
|
||
)
|
||
for i, (e, v) in enumerate(ranked_rows[:limit])
|
||
]
|
||
return LeaderboardOut(standings=standings, scoredMatches=scored_matches)
|
||
|
||
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)
|
||
|
||
|
||
@router.get("/ai", response_model=AILeaderboardOut)
|
||
async def ai_leaderboard(
|
||
league: str = Query("", description="wc | kbo | mlb — 빈값이면 전체"),
|
||
db: AsyncSession = Depends(get_db),
|
||
) -> AILeaderboardOut:
|
||
"""AI 모델 누적 랭킹 — 종료된 경기의 AI 예측을 유저와 동일한 배점으로 채점·합산.
|
||
|
||
별도 누적 테이블 없이 매 조회 시 종료 경기 전수 재계산 → 결과/배점 변동에
|
||
항상 일관. (모델 3개 × 경기 수라 비용 무시 가능.)
|
||
"""
|
||
q = (
|
||
select(Match)
|
||
.where(Match.result_outcome.is_not(None))
|
||
.options(selectinload(Match.predictions))
|
||
)
|
||
if league:
|
||
q = q.where(Match.league == league)
|
||
matches = (await db.execute(q)).scalars().all()
|
||
|
||
# model → [points, exact_count, matches_played]
|
||
agg: dict[str, list[int]] = {}
|
||
for m in matches:
|
||
bb = m.league in ("kbo", "mlb")
|
||
for p in m.predictions:
|
||
pts = score_prediction(
|
||
p.score_a, p.score_b, m.result_score_a, m.result_score_b, baseball=bb
|
||
)
|
||
row = agg.setdefault(p.model, [0, 0, 0])
|
||
row[0] += pts
|
||
row[1] += 1 if (p.score_a == m.result_score_a and p.score_b == m.result_score_b) else 0
|
||
row[2] += 1
|
||
|
||
order = {"GPT": 0, "Claude": 1, "Gemini": 2}
|
||
ranked = sorted(
|
||
agg.items(),
|
||
key=lambda kv: (-kv[1][0], -kv[1][1], -kv[1][2], order.get(kv[0], 9)),
|
||
)
|
||
|
||
standings = [
|
||
AIStandingOut(
|
||
rank=i + 1,
|
||
model=model, # type: ignore[arg-type]
|
||
totalPoints=pts,
|
||
exactCount=exact,
|
||
matchesPlayed=played,
|
||
)
|
||
for i, (model, (pts, exact, played)) in enumerate(ranked)
|
||
]
|
||
return AILeaderboardOut(standings=standings, scoredMatches=len(matches))
|