541 lines
22 KiB
Python
541 lines
22 KiB
Python
"""야구 부가 데이터 — 프리뷰(선발투수·상대전적)·리그 순위·라이브 필드 뷰.
|
||
|
||
KBO: 네이버 스포츠 비공식 API (프리뷰·순위·문자중계 relay)
|
||
MLB: 공식 Stats API (probablePitcher·standings·feed/live)
|
||
|
||
수집(refresh_*)은 워커가 매일, 조회(get_extras)는 라우터가 캐시만 읽음.
|
||
라이브(fetch_live)는 요청 시 프록시 + 짧은 TTL 메모리 캐시.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import time
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
from ..config import settings
|
||
from ..domain import ensure_aware, now_utc
|
||
from ..models import DataCache, Match
|
||
from ..teams_baseball import KBO_SHORT_TO_CODE, MLB_ID_TO_CODE, MLB_TEAMS
|
||
from .baseball_sync import match_seq
|
||
|
||
log = logging.getLogger("triplepick.baseball")
|
||
|
||
KST = timezone(timedelta(hours=9))
|
||
UA = {
|
||
"User-Agent": (
|
||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
||
"(KHTML, like Gecko) Chrome/126.0 Safari/537.36"
|
||
)
|
||
}
|
||
|
||
|
||
def naver_game_id(m: Match, game_no: str = "0") -> str:
|
||
kst = ensure_aware(m.kickoff_at).astimezone(KST)
|
||
return f"{kst.strftime('%Y%m%d')}{m.team_a_code}{m.team_b_code}{game_no}{kst.year}"
|
||
|
||
|
||
def naver_game_id_candidates(m: Match) -> list[str]:
|
||
"""더블헤더 대응 gameId 후보 — 끝번호 0(단일)/1(DH 1차전)/2(DH 2차전).
|
||
seq1 은 단일을 먼저 시도하고 DH 1차전으로 폴백, seq2 는 2차전 고정."""
|
||
nos = ["2"] if match_seq(m.match_id) == 2 else ["0", "1"]
|
||
return [naver_game_id(m, n) for n in nos]
|
||
|
||
|
||
async def _naver_get_first(client, m: Match, suffix: str) -> dict | None:
|
||
"""gameId 후보를 순서대로 시도해 첫 성공 응답을 반환."""
|
||
for gid in naver_game_id_candidates(m):
|
||
try:
|
||
res = await _naver_get(client, f"/schedule/games/{gid}/{suffix}")
|
||
except Exception: # noqa: BLE001 — 후보 불일치(404 등)는 다음 후보로
|
||
continue
|
||
if res:
|
||
return res
|
||
return None
|
||
|
||
|
||
async def _naver_get(client, path: str) -> dict | None:
|
||
r = await client.get(settings.naver_api_base + path, headers=UA)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
return data.get("result") if data.get("success") else None
|
||
|
||
|
||
async def _upsert(db, key: str, payload: dict) -> None:
|
||
row = await db.get(DataCache, key)
|
||
if row:
|
||
row.payload = payload
|
||
row.fetched_at = now_utc()
|
||
else:
|
||
db.add(DataCache(key=key, payload=payload, fetched_at=now_utc()))
|
||
|
||
|
||
# ── 프리뷰 (선발투수·시즌 상대전적) ────────────────────────────
|
||
def _kbo_starter(raw: dict | None) -> dict | None:
|
||
if not raw:
|
||
return None
|
||
info = raw.get("playerInfo") or {}
|
||
season = raw.get("currentSeasonStats") or {}
|
||
vs = raw.get("currentSeasonStatsOnOpponents") or {}
|
||
out = {
|
||
"name": info.get("name", ""),
|
||
"hitType": info.get("hitType", ""),
|
||
"era": season.get("era"),
|
||
"w": season.get("w"), "l": season.get("l"),
|
||
"vsEra": vs.get("era"),
|
||
}
|
||
return out if out["name"] else None
|
||
|
||
|
||
async def _refresh_previews_kbo(db, matches: list[Match]) -> int:
|
||
import httpx
|
||
|
||
n = 0
|
||
async with httpx.AsyncClient(timeout=15) as client:
|
||
for m in matches:
|
||
try:
|
||
res = await _naver_get_first(client, m, "preview")
|
||
p = (res or {}).get("previewData") or {}
|
||
except Exception as e: # noqa: BLE001
|
||
log.warning("kbo preview 실패 %s: %s", m.match_id, e)
|
||
continue
|
||
vs = p.get("seasonVsResult") or {}
|
||
payload = {
|
||
"starterA": _kbo_starter(p.get("awayStarter")),
|
||
"starterB": _kbo_starter(p.get("homeStarter")),
|
||
"seasonVs": {
|
||
"aWin": vs.get("aw"), "draw": vs.get("ad") or 0, "bWin": vs.get("hw"),
|
||
} if vs else None,
|
||
}
|
||
if payload["starterA"] or payload["starterB"] or payload["seasonVs"]:
|
||
await _upsert(db, f"preview:{m.match_id}", payload)
|
||
n += 1
|
||
return n
|
||
|
||
|
||
_HAND_KO = {"L": "좌", "R": "우", "S": "양"}
|
||
|
||
|
||
def _mlb_starter(p: dict, stats: dict[int, dict]) -> dict | None:
|
||
if not p.get("fullName"):
|
||
return None
|
||
out: dict = {"name": p["fullName"]}
|
||
out.update(stats.get(p.get("id"), {}))
|
||
return out
|
||
|
||
|
||
async def _mlb_season_vs(c, m: Match, year: int) -> dict | None:
|
||
"""시즌 정규 상대전적 — 두 팀 간 완료 경기 승수 집계 (팀쌍당 1콜)."""
|
||
aid = (MLB_TEAMS.get(m.team_a_code) or {}).get("mlb_id")
|
||
bid = (MLB_TEAMS.get(m.team_b_code) or {}).get("mlb_id")
|
||
if not aid or not bid:
|
||
return None
|
||
r = await c.get(
|
||
f"{settings.mlb_api_base}/v1/schedule?sportId=1&season={year}&gameType=R"
|
||
f"&teamId={bid}&opponentId={aid}"
|
||
f"&startDate={year}-03-01&endDate={datetime.now(KST).date().isoformat()}"
|
||
)
|
||
r.raise_for_status()
|
||
wins = {aid: 0, bid: 0}
|
||
for day in r.json().get("dates") or []:
|
||
for g in day.get("games") or []:
|
||
if (g.get("status") or {}).get("abstractGameState") != "Final":
|
||
continue
|
||
for side in ("away", "home"):
|
||
t = g["teams"][side]
|
||
tid = (t.get("team") or {}).get("id")
|
||
if t.get("isWinner") and tid in wins:
|
||
wins[tid] += 1
|
||
if wins[aid] + wins[bid] == 0:
|
||
return None
|
||
return {"aWin": wins[aid], "draw": 0, "bWin": wins[bid]}
|
||
|
||
|
||
async def _refresh_previews_mlb(db, matches: list[Match]) -> int:
|
||
"""MLB 예고 선발(시즌 ERA·승패·투타 포함)·시즌 상대전적 — 공식 Stats API.
|
||
|
||
호출량: 일정 1콜 + 선발 스탯 일괄 1콜 + 상대전적 팀쌍당 1콜.
|
||
"""
|
||
import httpx
|
||
|
||
if not matches:
|
||
return 0
|
||
dates = sorted({ensure_aware(m.kickoff_at).astimezone(KST).date() for m in matches})
|
||
year = datetime.now(KST).year
|
||
# statsapi 의 start/endDate 는 미국 날짜 — KST 새벽~오전 경기는 미국 전날이라
|
||
# 시작일을 하루 앞당겨야 누락되지 않는다.
|
||
url = (
|
||
f"{settings.mlb_api_base}/v1/schedule?sportId=1"
|
||
f"&startDate={(dates[0] - timedelta(days=1)).isoformat()}"
|
||
f"&endDate={dates[-1].isoformat()}"
|
||
"&hydrate=probablePitcher"
|
||
)
|
||
async with httpx.AsyncClient(timeout=20) as c:
|
||
r = await c.get(url)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
|
||
# (dateKst, away, home) → (원정 선발 raw, 홈 선발 raw)
|
||
starters: dict[tuple, tuple[dict, dict]] = {}
|
||
for day in data.get("dates") or []:
|
||
for g in day.get("games") or []:
|
||
a = MLB_ID_TO_CODE.get((g["teams"]["away"]["team"] or {}).get("id"))
|
||
b = MLB_ID_TO_CODE.get((g["teams"]["home"]["team"] or {}).get("id"))
|
||
gd = g.get("gameDate")
|
||
if not a or not b or not gd:
|
||
continue
|
||
d = (
|
||
datetime.fromisoformat(gd.replace("Z", "+00:00"))
|
||
.astimezone(KST).strftime("%Y%m%d")
|
||
)
|
||
pa = g["teams"]["away"].get("probablePitcher") or {}
|
||
pb = g["teams"]["home"].get("probablePitcher") or {}
|
||
starters[(d, a, b)] = (pa, pb)
|
||
|
||
# 선발 시즌 스탯 — people 일괄 조회 1콜 (ERA·승패·투타)
|
||
pids = sorted({
|
||
p["id"] for pair in starters.values() for p in pair if p.get("id")
|
||
})
|
||
pstats: dict[int, dict] = {}
|
||
if pids:
|
||
try:
|
||
r2 = await c.get(
|
||
f"{settings.mlb_api_base}/v1/people"
|
||
f"?personIds={','.join(map(str, pids))}"
|
||
f"&hydrate=stats(group=[pitching],type=[season],season={year})"
|
||
)
|
||
r2.raise_for_status()
|
||
for p in r2.json().get("people") or []:
|
||
splits = (p.get("stats") or [{}])[0].get("splits") or []
|
||
s = splits[0].get("stat", {}) if splits else {}
|
||
hand = _HAND_KO.get((p.get("pitchHand") or {}).get("code"))
|
||
bat = _HAND_KO.get((p.get("batSide") or {}).get("code"))
|
||
pstats[p["id"]] = {
|
||
"hitType": f"{hand}투{bat}타" if hand and bat else None,
|
||
"era": s.get("era"),
|
||
"w": s.get("wins"), "l": s.get("losses"),
|
||
}
|
||
except Exception as e: # noqa: BLE001 — 스탯 실패 시 이름만 표시
|
||
log.warning("mlb 선발 스탯 실패: %s", e)
|
||
|
||
n = 0
|
||
vs_cache: dict[tuple, dict | None] = {}
|
||
for m in matches:
|
||
d = ensure_aware(m.kickoff_at).astimezone(KST).strftime("%Y%m%d")
|
||
pa, pb = starters.get((d, m.team_a_code, m.team_b_code), ({}, {}))
|
||
pair = (m.team_a_code, m.team_b_code)
|
||
if pair not in vs_cache:
|
||
try:
|
||
vs_cache[pair] = await _mlb_season_vs(c, m, year)
|
||
except Exception as e: # noqa: BLE001
|
||
log.warning("mlb 상대전적 실패 %s: %s", m.match_id, e)
|
||
vs_cache[pair] = None
|
||
sa = _mlb_starter(pa, pstats)
|
||
sb = _mlb_starter(pb, pstats)
|
||
if sa or sb or vs_cache[pair]:
|
||
await _upsert(db, f"preview:{m.match_id}", {
|
||
"starterA": sa,
|
||
"starterB": sb,
|
||
"seasonVs": vs_cache[pair],
|
||
})
|
||
n += 1
|
||
return n
|
||
|
||
|
||
# ── 리그 순위 ──────────────────────────────────────────────────
|
||
async def _refresh_standings_kbo(db) -> bool:
|
||
import httpx
|
||
|
||
year = datetime.now(KST).year
|
||
try:
|
||
async with httpx.AsyncClient(timeout=15) as client:
|
||
res = await _naver_get(client, f"/stats/categories/kbo/seasons/{year}/teams")
|
||
except Exception as e: # noqa: BLE001
|
||
log.warning("kbo standings 실패: %s", e)
|
||
return False
|
||
table: dict[str, dict] = {}
|
||
for r in (res or {}).get("seasonTeamStats") or []:
|
||
code = KBO_SHORT_TO_CODE.get(r.get("teamShortName", ""))
|
||
if code:
|
||
table[code] = {
|
||
"rank": r.get("ranking"),
|
||
"w": r.get("winGameCount"), "d": r.get("drawnGameCount"),
|
||
"l": r.get("loseGameCount"), "wra": r.get("wra"),
|
||
"gb": r.get("gameBehind"), "last5": r.get("lastFiveGames"),
|
||
"avg": r.get("offenseHra"), "era": r.get("defenseEra"),
|
||
}
|
||
if not table:
|
||
return False
|
||
await _upsert(db, "standings:kbo", table)
|
||
return True
|
||
|
||
|
||
# statsapi division.id → 순위표 그룹 키 (AL/NL × 동·중·서)
|
||
_MLB_DIV = {201: "ALE", 202: "ALC", 200: "ALW", 204: "NLE", 205: "NLC", 203: "NLW"}
|
||
|
||
|
||
async def _refresh_standings_mlb(db) -> bool:
|
||
import httpx
|
||
|
||
year = datetime.now(KST).year
|
||
table: dict[str, dict] = {}
|
||
try:
|
||
async with httpx.AsyncClient(timeout=15) as c:
|
||
for lid in (103, 104): # AL, NL
|
||
r = await c.get(
|
||
f"{settings.mlb_api_base}/v1/standings?leagueId={lid}&season={year}"
|
||
)
|
||
r.raise_for_status()
|
||
for rec_div in r.json().get("records") or []:
|
||
div = _MLB_DIV.get((rec_div.get("division") or {}).get("id"))
|
||
for t in rec_div.get("teamRecords") or []:
|
||
code = MLB_ID_TO_CODE.get((t.get("team") or {}).get("id"))
|
||
if code:
|
||
table[code] = {
|
||
"div": div,
|
||
"rank": int(t.get("divisionRank") or 0) or None,
|
||
"w": t.get("wins"), "d": 0, "l": t.get("losses"),
|
||
"wra": t.get("winningPercentage"),
|
||
"gb": t.get("gamesBack"),
|
||
"last5": None, "avg": None, "era": None,
|
||
}
|
||
except Exception as e: # noqa: BLE001
|
||
log.warning("mlb standings 실패: %s", e)
|
||
return False
|
||
if not table:
|
||
return False
|
||
await _upsert(db, "standings:mlb", table)
|
||
return True
|
||
|
||
|
||
async def refresh_baseball_details(db, league: str, matches: list[Match]) -> None:
|
||
"""임박(48h 내) 미종료 경기 프리뷰 + 리그 순위 캐시 갱신."""
|
||
horizon = now_utc() + timedelta(hours=48)
|
||
targets = [
|
||
m for m in matches
|
||
if m.result_outcome is None
|
||
and m.status != "cancelled"
|
||
and ensure_aware(m.kickoff_at) <= horizon
|
||
]
|
||
if league == "kbo":
|
||
n = await _refresh_previews_kbo(db, targets)
|
||
await _refresh_standings_kbo(db)
|
||
elif league == "mlb":
|
||
n = await _refresh_previews_mlb(db, targets)
|
||
await _refresh_standings_mlb(db)
|
||
else:
|
||
return
|
||
await db.commit()
|
||
log.info("baseball details(%s): 프리뷰 %d경기 캐싱", league, n)
|
||
|
||
|
||
# ── extras 조회 (라우터 — 캐시만) ──────────────────────────────
|
||
async def get_extras(db, matches: list[Match]) -> dict[str, dict]:
|
||
standings_cache: dict[str, dict] = {}
|
||
out: dict[str, dict] = {}
|
||
for m in matches:
|
||
if m.league not in ("kbo", "mlb"):
|
||
continue
|
||
if m.league not in standings_cache:
|
||
row = await db.get(DataCache, f"standings:{m.league}")
|
||
standings_cache[m.league] = row.payload if row else {}
|
||
extras: dict = {}
|
||
prev = await db.get(DataCache, f"preview:{m.match_id}")
|
||
if prev:
|
||
extras.update(prev.payload)
|
||
st = standings_cache[m.league]
|
||
st_a, st_b = st.get(m.team_a_code), st.get(m.team_b_code)
|
||
if st_a or st_b:
|
||
extras["standings"] = {"a": st_a, "b": st_b}
|
||
if extras:
|
||
out[m.match_id] = extras
|
||
return out
|
||
|
||
|
||
# ── 라이브 필드 뷰 ─────────────────────────────────────────────
|
||
_LIVE_TTL_SEC = 15.0
|
||
_live_cache: dict[str, tuple[float, dict]] = {}
|
||
|
||
FIELD_POSITIONS = (
|
||
"포수", "1루수", "2루수", "3루수", "유격수", "좌익수", "중견수", "우익수",
|
||
)
|
||
# MLB 포지션 약어 → 한글 (필드 좌표 키와 통일)
|
||
MLB_POS = {
|
||
"C": "포수", "1B": "1루수", "2B": "2루수", "3B": "3루수", "SS": "유격수",
|
||
"LF": "좌익수", "CF": "중견수", "RF": "우익수",
|
||
}
|
||
|
||
|
||
def _current_slots(batters: list[dict]) -> list[dict]:
|
||
by_order: dict[int, dict] = {}
|
||
for b in batters or []:
|
||
o = b.get("batOrder")
|
||
if o is None:
|
||
continue
|
||
cur = by_order.get(o)
|
||
if cur is None or (b.get("seqno") or 0) > (cur.get("seqno") or 0):
|
||
by_order[o] = b
|
||
return [by_order[o] for o in sorted(by_order)]
|
||
|
||
|
||
def _batter_out(b: dict) -> dict:
|
||
return {
|
||
"order": b.get("batOrder"),
|
||
"name": b.get("name", ""),
|
||
"pos": b.get("posName", ""),
|
||
"avg": b.get("seasonHra"),
|
||
"sub": (b.get("seqno") or 1) > 1,
|
||
}
|
||
|
||
|
||
def _transform_naver_relay(t: dict) -> dict:
|
||
gs = t.get("currentGameState") or {}
|
||
home_batting = str(t.get("homeOrAway")) == "1"
|
||
home_lu = t.get("homeLineup") or {}
|
||
away_lu = t.get("awayLineup") or {}
|
||
offense_lu = home_lu if home_batting else away_lu
|
||
defense_lu = away_lu if home_batting else home_lu
|
||
|
||
offense = _current_slots(offense_lu.get("batter"))
|
||
defense = _current_slots(defense_lu.get("batter"))
|
||
pitchers = defense_lu.get("pitcher") or []
|
||
pitcher = max(pitchers, key=lambda p: p.get("seqno") or 0) if pitchers else {}
|
||
batter_code = str(gs.get("batter") or "")
|
||
batter = next((b for b in offense if str(b.get("pcode")) == batter_code), None)
|
||
|
||
return {
|
||
"available": True,
|
||
"inn": t.get("inn"),
|
||
"half": "B" if home_batting else "T",
|
||
"score": {"away": gs.get("awayScore"), "home": gs.get("homeScore")},
|
||
"bso": {"b": gs.get("ball"), "s": gs.get("strike"), "o": gs.get("out")},
|
||
"bases": [
|
||
str(gs.get(k) or "0") != "0" for k in ("base1", "base2", "base3")
|
||
],
|
||
"batter": _batter_out(batter) if batter else None,
|
||
"pitcher": {
|
||
"name": pitcher.get("name", ""),
|
||
"ballCount": pitcher.get("ballCount"),
|
||
} if pitcher else None,
|
||
"vsRecord": t.get("pitcherVsBatterCareerStats") or "",
|
||
"defense": [
|
||
{"pos": b.get("posName"), "name": b.get("name", "")}
|
||
for b in defense if b.get("posName") in FIELD_POSITIONS
|
||
],
|
||
"offenseLineup": [_batter_out(b) for b in offense],
|
||
}
|
||
|
||
|
||
def _transform_mlb_feed(feed: dict) -> dict:
|
||
ld = feed.get("liveData") or {}
|
||
ls = ld.get("linescore") or {}
|
||
box = (ld.get("boxscore") or {}).get("teams") or {}
|
||
half_top = (ls.get("inningHalf") or "").lower() == "top"
|
||
offense_side, defense_side = ("away", "home") if half_top else ("home", "away")
|
||
off = ls.get("offense") or {}
|
||
defn = ls.get("defense") or {}
|
||
|
||
def _players(side: str) -> dict:
|
||
return (box.get(side) or {}).get("players") or {}
|
||
|
||
# 수비 배치: boxscore players 의 position + 현재 출장(battingOrder 존재)
|
||
defense = []
|
||
for p in _players(defense_side).values():
|
||
pos = MLB_POS.get(((p.get("position") or {}).get("abbreviation") or ""))
|
||
name = ((p.get("person") or {}).get("fullName")) or ""
|
||
if pos and name and p.get("gameStatus", {}).get("isCurrentBatter") is not None:
|
||
defense.append({"pos": pos, "name": name})
|
||
# 같은 포지션 중복(교체) — 마지막 것만
|
||
dedup: dict[str, dict] = {f["pos"]: f for f in defense}
|
||
|
||
order_raw = (box.get(offense_side) or {}).get("battingOrder") or []
|
||
id_to_player = _players(offense_side)
|
||
lineup = []
|
||
for i, pid in enumerate(order_raw[:9]):
|
||
p = id_to_player.get(f"ID{pid}") or {}
|
||
lineup.append({
|
||
"order": i + 1,
|
||
"name": ((p.get("person") or {}).get("fullName")) or "",
|
||
"pos": MLB_POS.get(((p.get("position") or {}).get("abbreviation") or ""), ""),
|
||
"avg": None,
|
||
"sub": False,
|
||
})
|
||
|
||
batter_name = ((off.get("batter") or {}).get("fullName")) or ""
|
||
batter = next((b for b in lineup if b["name"] == batter_name), None)
|
||
pitcher = (defn.get("pitcher") or {}).get("fullName") or \
|
||
(off.get("pitcher") or {}).get("fullName") or ""
|
||
|
||
return {
|
||
"available": bool(ls.get("currentInning")),
|
||
"inn": ls.get("currentInning"),
|
||
"half": "T" if half_top else "B",
|
||
"score": {
|
||
"away": ((ls.get("teams") or {}).get("away") or {}).get("runs"),
|
||
"home": ((ls.get("teams") or {}).get("home") or {}).get("runs"),
|
||
},
|
||
"bso": {"b": ls.get("balls"), "s": ls.get("strikes"), "o": ls.get("outs")},
|
||
"bases": [bool(off.get("first")), bool(off.get("second")), bool(off.get("third"))],
|
||
"batter": batter or ({"order": None, "name": batter_name} if batter_name else None),
|
||
"pitcher": {"name": pitcher, "ballCount": None} if pitcher else None,
|
||
"vsRecord": "",
|
||
"defense": list(dedup.values()),
|
||
"offenseLineup": lineup,
|
||
}
|
||
|
||
|
||
async def fetch_live(m: Match) -> dict:
|
||
"""라이브 필드 뷰 페이로드 (리그별 소스). 미게시면 available=False."""
|
||
import httpx
|
||
|
||
cached = _live_cache.get(m.match_id)
|
||
if cached and time.monotonic() - cached[0] < _LIVE_TTL_SEC:
|
||
return cached[1]
|
||
payload: dict = {"available": False}
|
||
try:
|
||
if m.league == "kbo":
|
||
async with httpx.AsyncClient(timeout=10) as client:
|
||
res = await _naver_get_first(client, m, "relay")
|
||
t = (res or {}).get("textRelayData")
|
||
if t:
|
||
payload = _transform_naver_relay(t)
|
||
elif m.league == "mlb":
|
||
# gamePk 를 일정에서 재조회 — MLB 일정 API 의 date 는 미국 날짜라
|
||
# KST 기준 하루 전~당일 범위로 조회 후 dateKst 로 정확히 매칭.
|
||
kst = ensure_aware(m.kickoff_at).astimezone(KST)
|
||
date_kst = kst.strftime("%Y%m%d")
|
||
start = (kst.date() - timedelta(days=1)).isoformat()
|
||
async with httpx.AsyncClient(timeout=15) as c:
|
||
r = await c.get(
|
||
f"{settings.mlb_api_base}/v1/schedule?sportId=1"
|
||
f"&startDate={start}&endDate={kst.date().isoformat()}"
|
||
)
|
||
r.raise_for_status()
|
||
# 더블헤더 대응: 같은 날짜·팀쌍 경기를 시작시각순으로 모아
|
||
# match_id 의 차수(seq)에 해당하는 경기를 고른다.
|
||
cands: list[tuple[str, int]] = []
|
||
for day in r.json().get("dates") or []:
|
||
for g in day.get("games") or []:
|
||
a = MLB_ID_TO_CODE.get((g["teams"]["away"]["team"] or {}).get("id"))
|
||
b = MLB_ID_TO_CODE.get((g["teams"]["home"]["team"] or {}).get("id"))
|
||
gd = g.get("gameDate")
|
||
if not a or not b or not gd:
|
||
continue
|
||
g_kst = (
|
||
datetime.fromisoformat(gd.replace("Z", "+00:00"))
|
||
.astimezone(KST).strftime("%Y%m%d")
|
||
)
|
||
if a == m.team_a_code and b == m.team_b_code and g_kst == date_kst:
|
||
cands.append((gd, g.get("gamePk")))
|
||
cands.sort()
|
||
idx = match_seq(m.match_id) - 1
|
||
pk = cands[idx][1] if idx < len(cands) else None
|
||
if pk:
|
||
r2 = await c.get(f"{settings.mlb_api_base}/v1.1/game/{pk}/feed/live")
|
||
r2.raise_for_status()
|
||
payload = _transform_mlb_feed(r2.json())
|
||
except Exception as e: # noqa: BLE001
|
||
log.warning("live 실패 %s: %s", m.match_id, e)
|
||
payload = {"available": False}
|
||
_live_cache[m.match_id] = (time.monotonic(), payload)
|
||
return payload
|