59 lines
1.8 KiB
Python
59 lines
1.8 KiB
Python
"""
|
||
Liderlik tablosu — adım (lifetime), level ve gold sıralamaları.
|
||
GET /leaderboard?by=steps|level|gold → top 20 + kendi sıran.
|
||
"""
|
||
from fastapi import APIRouter, Depends, HTTPException
|
||
from pydantic import BaseModel
|
||
from sqlalchemy import select, text
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from deps import get_current_user
|
||
from models import Character, User, get_db
|
||
|
||
router = APIRouter(prefix="/leaderboard", tags=["leaderboard"])
|
||
|
||
ORDER_COLS = {"steps": "step_lifetime", "level": "total_xp", "gold": "gold"}
|
||
|
||
|
||
class LeaderRow(BaseModel):
|
||
rank: int
|
||
name: str
|
||
level: int
|
||
value: int # sıralama kriterinin değeri
|
||
me: bool
|
||
|
||
|
||
class LeaderboardOut(BaseModel):
|
||
by: str
|
||
rows: list[LeaderRow]
|
||
my_rank: int
|
||
|
||
|
||
@router.get("", response_model=LeaderboardOut)
|
||
async def leaderboard(by: str = "steps",
|
||
user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)):
|
||
if by not in ORDER_COLS:
|
||
raise HTTPException(400, "by: steps | level | gold")
|
||
col = ORDER_COLS[by]
|
||
char = (await db.execute(select(Character).where(Character.user_id == user.id))).scalar_one_or_none()
|
||
if not char:
|
||
raise HTTPException(404, "Karakter yok")
|
||
|
||
rows = (await db.execute(text(f"""
|
||
SELECT rank, name, level, val, id FROM (
|
||
SELECT id, name, level, {col} AS val,
|
||
RANK() OVER (ORDER BY {col} DESC, id) AS rank
|
||
FROM characters
|
||
) t WHERE rank <= 20 OR id = :me
|
||
ORDER BY rank
|
||
"""), {"me": char.id})).all()
|
||
|
||
my_rank = 0
|
||
out: list[LeaderRow] = []
|
||
for r in rows:
|
||
me = r[4] == char.id
|
||
if me:
|
||
my_rank = int(r[0])
|
||
if int(r[0]) <= 20:
|
||
out.append(LeaderRow(rank=int(r[0]), name=r[1], level=int(r[2]), value=int(r[3]), me=me))
|
||
return LeaderboardOut(by=by, rows=out, my_rank=my_rank)
|