141 lines
5.8 KiB
Python
141 lines
5.8 KiB
Python
"""야구 일정·결과 수집 — KBO(네이버 비공식) + MLB(공식 Stats API).
|
|
|
|
반환 표준 레코드 (두 리그 공통):
|
|
{league, teamA(원정), teamB(홈), dateKst 'YYYYMMDD', kickoffKst ISO(+09:00),
|
|
venue, cancelled, [scoreA, scoreB]}
|
|
야구는 같은 두 팀이 시즌 내 반복 대결 → 경기 식별에 반드시 dateKst 포함.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from ..config import settings
|
|
from ..teams_baseball import KBO_TEAMS, MLB_ID_TO_CODE
|
|
|
|
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"
|
|
)
|
|
}
|
|
|
|
|
|
# ── KBO (네이버) ───────────────────────────────────────────────
|
|
async def _fetch_kbo(days_back: int, days_ahead: int) -> list[dict]:
|
|
import httpx
|
|
|
|
today = datetime.now(KST).date()
|
|
url = (
|
|
f"{settings.naver_api_base}/schedule/games"
|
|
"?fields=basic,stadium,statusNum"
|
|
"&upperCategoryId=kbaseball&categoryId=kbo"
|
|
f"&fromDate={(today - timedelta(days=days_back)).isoformat()}"
|
|
f"&toDate={(today + timedelta(days=days_ahead)).isoformat()}&size=500"
|
|
)
|
|
async with httpx.AsyncClient(timeout=20) as c:
|
|
r = await c.get(url, headers=UA)
|
|
r.raise_for_status()
|
|
games = (r.json().get("result") or {}).get("games") or []
|
|
|
|
out: list[dict] = []
|
|
for g in games:
|
|
a = (g.get("awayTeamCode") or "").upper()
|
|
b = (g.get("homeTeamCode") or "").upper()
|
|
if a not in KBO_TEAMS or b not in KBO_TEAMS:
|
|
continue # 올스타/시범 등 제외
|
|
dt = g.get("gameDateTime") # KST, 오프셋 없음
|
|
if not dt:
|
|
continue
|
|
kickoff = datetime.fromisoformat(dt).replace(tzinfo=KST)
|
|
rec = {
|
|
"league": "kbo",
|
|
"teamA": a, "teamB": b,
|
|
"dateKst": kickoff.strftime("%Y%m%d"),
|
|
"kickoffKst": kickoff.isoformat(),
|
|
"venue": g.get("stadium", ""),
|
|
"cancelled": bool(g.get("cancel")),
|
|
}
|
|
if g.get("statusCode") == "RESULT" and not rec["cancelled"]:
|
|
sa, sb = g.get("awayTeamScore"), g.get("homeTeamScore")
|
|
if sa is not None and sb is not None:
|
|
rec["scoreA"], rec["scoreB"] = int(sa), int(sb)
|
|
out.append(rec)
|
|
return out
|
|
|
|
|
|
# ── MLB (공식 Stats API) ───────────────────────────────────────
|
|
async def _fetch_mlb(days_back: int, days_ahead: int) -> list[dict]:
|
|
import httpx
|
|
|
|
today = datetime.now(KST).date()
|
|
url = (
|
|
f"{settings.mlb_api_base}/v1/schedule?sportId=1"
|
|
f"&startDate={(today - timedelta(days=days_back)).isoformat()}"
|
|
f"&endDate={(today + timedelta(days=days_ahead)).isoformat()}"
|
|
)
|
|
async with httpx.AsyncClient(timeout=20) as c:
|
|
r = await c.get(url)
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
|
|
out: list[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"))
|
|
if not a or not b:
|
|
continue # 올스타전 등 제외
|
|
gd = g.get("gameDate") # ISO UTC
|
|
if not gd:
|
|
continue
|
|
kickoff = (
|
|
datetime.fromisoformat(gd.replace("Z", "+00:00")).astimezone(KST)
|
|
)
|
|
state = (g.get("status") or {}).get("detailedState", "")
|
|
rec = {
|
|
"league": "mlb",
|
|
"teamA": a, "teamB": b,
|
|
"dateKst": kickoff.strftime("%Y%m%d"),
|
|
"kickoffKst": kickoff.isoformat(),
|
|
"venue": (g.get("venue") or {}).get("name", ""),
|
|
"cancelled": state in ("Postponed", "Cancelled", "Suspended"),
|
|
"gamePk": g.get("gamePk"), # MLB 라이브 피드 키
|
|
}
|
|
if state == "Final" and not rec["cancelled"]:
|
|
sa = g["teams"]["away"].get("score")
|
|
sb = g["teams"]["home"].get("score")
|
|
if sa is not None and sb is not None:
|
|
rec["scoreA"], rec["scoreB"] = int(sa), int(sb)
|
|
out.append(rec)
|
|
return out
|
|
|
|
|
|
# ── 공개 API ────────────────────────────────────────────────────
|
|
async def fetch_baseball_schedule(league: str) -> list[dict]:
|
|
"""리그 일정+결과 수집 (취소 포함). 실패 시 빈 리스트 — 기존 일정 유지."""
|
|
try:
|
|
if league == "kbo":
|
|
records = await _fetch_kbo(settings.result_recheck_days, settings.baseball_days_ahead)
|
|
elif league == "mlb":
|
|
records = await _fetch_mlb(settings.result_recheck_days, settings.baseball_days_ahead)
|
|
else:
|
|
return []
|
|
# 더블헤더 차수 부여 — sync·정산이 같은 seq 로 경기를 식별한다.
|
|
from .baseball_sync import assign_seq
|
|
|
|
assign_seq(records)
|
|
log.info("baseball schedule(%s): %d경기 수집", league, len(records))
|
|
return records
|
|
except Exception as e: # noqa: BLE001
|
|
log.error("baseball schedule(%s) 실패: %s — 기존 일정 유지", league, e)
|
|
return []
|
|
|
|
|
|
async def fetch_baseball_results(league: str) -> list[dict]:
|
|
"""확정 스코어만 → [{league, teamA, teamB, dateKst, scoreA, scoreB}]."""
|
|
return [r for r in await fetch_baseball_schedule(league) if "scoreA" in r]
|