| """数据访问层的基础原语：比赛结果、match→dict、赛季过滤子句。 |
| |
| 被 repo 各操作复用；单独成文件，使 repo.py 控制在 200–400 行内。 |
| """ |
| from __future__ import annotations |
|
|
| from typing import Any, Optional |
|
|
| from sqlalchemy import extract |
|
|
| from .models import Match |
|
|
|
|
| def _result(our: int, opp: int) -> str: |
| if our > opp: |
| return "W" |
| if our < opp: |
| return "L" |
| return "D" |
|
|
|
|
| def _match_dict(m: Match) -> dict[str, Any]: |
| return { |
| "id": m.id, |
| "date": m.match_date.isoformat() if m.match_date else "", |
| "opponent": m.opponent, |
| "home_away": m.home_away, |
| "our": m.our_score, |
| "opp": m.opp_score, |
| "result": _result(m.our_score, m.opp_score), |
| "score_str": f"{m.our_score} : {m.opp_score}", |
| "note": m.note or "", |
| "mvp_locked": bool(getattr(m, "mvp_locked", False)), |
| "mvp_user_id": getattr(m, "mvp_user_id", None), |
| "mvp2_user_id": getattr(m, "mvp2_user_id", None), |
| "dmvp_user_id": getattr(m, "dmvp_user_id", None), |
| } |
|
|
|
|
| def _season_clause(season: Optional[int]): |
| """赛季过滤子句（按 match_date 年份）；season 为 None 时返回 None 表示不过滤。""" |
| return extract("year", Match.match_date) == season if season else None |
|
|
|
|
| def _season_match_ids(s, season: Optional[int]) -> list[int]: |
| q = s.query(Match.id) |
| cl = _season_clause(season) |
| if cl is not None: |
| q = q.filter(cl) |
| return [r[0] for r in q.all()] |
|
|
