54 lines
2.0 KiB
Python
54 lines
2.0 KiB
Python
"""리그 순위표 API — 워커가 캐싱한 standings:{league} 를 팀 정보와 합쳐 서빙.
|
||
|
||
KBO: 단일 테이블(10팀, 순위순). MLB: 디비전(AL/NL × 동·중·서) 6그룹.
|
||
캐시가 아직 없으면 빈 groups 를 반환한다(프론트는 안내 문구 표시).
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from fastapi import APIRouter, Depends, Query
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from ..database import get_db
|
||
from ..models import DataCache
|
||
from ..teams_baseball import team_info
|
||
|
||
router = APIRouter(prefix="/api/standings", tags=["standings"])
|
||
|
||
# MLB 디비전 표시 순서 (AL 동→중→서, NL 동→중→서)
|
||
_MLB_DIV_ORDER = ["ALE", "ALC", "ALW", "NLE", "NLC", "NLW"]
|
||
|
||
|
||
@router.get("")
|
||
async def get_standings(
|
||
league: str = Query(..., description="kbo | mlb"),
|
||
db: AsyncSession = Depends(get_db),
|
||
) -> dict:
|
||
if league not in ("kbo", "mlb"):
|
||
return {"league": league, "updatedAt": None, "groups": []}
|
||
row = await db.get(DataCache, f"standings:{league}")
|
||
table: dict = row.payload if row else {}
|
||
rows = [{**team_info(league, code), **st} for code, st in table.items()]
|
||
|
||
if league == "kbo":
|
||
rows.sort(key=lambda r: r.get("rank") or 99)
|
||
groups = [{"key": None, "rows": rows}] if rows else []
|
||
else:
|
||
by_div: dict[str, list] = {}
|
||
for r in rows:
|
||
by_div.setdefault(r.get("div") or "", []).append(r)
|
||
for lst in by_div.values():
|
||
lst.sort(key=lambda r: r.get("rank") or 99)
|
||
groups = [
|
||
{"key": d, "rows": by_div[d]} for d in _MLB_DIV_ORDER if d in by_div
|
||
]
|
||
if not groups and rows:
|
||
# 캐시가 div 주입 이전 버전이면 전체 승률순 단일 그룹으로 폴백
|
||
rows.sort(key=lambda r: -float(r.get("wra") or 0))
|
||
groups = [{"key": None, "rows": rows}]
|
||
|
||
return {
|
||
"league": league,
|
||
"updatedAt": row.fetched_at.isoformat() if row and row.fetched_at else None,
|
||
"groups": groups,
|
||
}
|