255 lines
8.9 KiB
Python
255 lines
8.9 KiB
Python
"""경기 일정 외부 수집 — 매일 워커가 호출(크롤링).
|
|
|
|
소스(SCHEDULE_SOURCE):
|
|
openfootball raw GitHub JSON (키 불필요, 기본). 2026 Group A = 본 서비스 6경기.
|
|
football-data football-data.org /v4/competitions/WC/matches (FOOTBALL_DATA_TOKEN 필요)
|
|
fallback 외부 호출 없이 빈 결과 → 기존(시드) 일정 유지
|
|
|
|
반환 표준 레코드:
|
|
{teamA: code, teamB: code, kickoffKst: ISO(+09:00), venue: str}
|
|
team 코드로 매핑되지 않는 경기(다른 그룹/팀)는 제외.
|
|
시각은 절대 시점(UTC)으로 환산 후 KST(+09:00) 고정 오프셋으로 표기.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import re
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from ..config import settings
|
|
from ..schedule_data import code_for
|
|
from ..teams_data import code_for_tla
|
|
|
|
log = logging.getLogger("triplepick.schedule")
|
|
|
|
KST = timezone(timedelta(hours=9))
|
|
# "13:00 UTC-6" / "20:00 UTC+2" 형태 파싱
|
|
_TIME_RE = re.compile(r"(\d{1,2}):(\d{2})\s*UTC\s*([+-]\d{1,2})")
|
|
|
|
|
|
def _to_kst_iso(date_str: str, time_str: str) -> str | None:
|
|
m = _TIME_RE.search(time_str or "")
|
|
if not m:
|
|
return None
|
|
hh, mm, off = int(m.group(1)), int(m.group(2)), int(m.group(3))
|
|
try:
|
|
y, mo, d = (int(x) for x in date_str.split("-"))
|
|
except ValueError:
|
|
return None
|
|
local = datetime(y, mo, d, hh, mm, tzinfo=timezone(timedelta(hours=off)))
|
|
return local.astimezone(KST).isoformat()
|
|
|
|
|
|
async def _fetch_openfootball() -> list[dict]:
|
|
import httpx
|
|
|
|
async with httpx.AsyncClient(timeout=20) as c:
|
|
r = await c.get(settings.schedule_url)
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
|
|
out: list[dict] = []
|
|
for m in data.get("matches", []):
|
|
if (m.get("group") or "").strip() != settings.schedule_group:
|
|
continue
|
|
a, b = code_for(m.get("team1", "")), code_for(m.get("team2", ""))
|
|
if not a or not b:
|
|
continue
|
|
kickoff = _to_kst_iso(m.get("date", ""), m.get("time", ""))
|
|
if not kickoff:
|
|
continue
|
|
out.append({"teamA": a, "teamB": b, "kickoffKst": kickoff, "venue": m.get("ground", "")})
|
|
return out
|
|
|
|
|
|
async def _fetch_football_data() -> list[dict]:
|
|
if not settings.football_data_token:
|
|
log.warning("schedule: football-data 토큰 미설정 — 수집 생략")
|
|
return []
|
|
import httpx
|
|
|
|
url = (
|
|
f"https://api.football-data.org/v4/competitions/"
|
|
f"{settings.football_data_competition}/matches"
|
|
)
|
|
async with httpx.AsyncClient(timeout=20) as c:
|
|
r = await c.get(url, headers={"X-Auth-Token": settings.football_data_token})
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
|
|
out: list[dict] = []
|
|
for m in data.get("matches", []):
|
|
if (m.get("group") or "").replace("GROUP_", "Group ") != settings.schedule_group:
|
|
continue
|
|
a = code_for((m.get("homeTeam") or {}).get("name", "") or "")
|
|
b = code_for((m.get("awayTeam") or {}).get("name", "") or "")
|
|
if not a or not b:
|
|
continue
|
|
utc = m.get("utcDate") # ISO Z
|
|
if not utc:
|
|
continue
|
|
kickoff = (
|
|
datetime.fromisoformat(utc.replace("Z", "+00:00")).astimezone(KST).isoformat()
|
|
)
|
|
venue = m.get("venue") or ""
|
|
out.append({"teamA": a, "teamB": b, "kickoffKst": kickoff, "venue": venue})
|
|
return out
|
|
|
|
|
|
async def _fetch_all_groups_football_data() -> list[dict]:
|
|
"""football-data 에서 전체 조별리그(12개조 72경기) 수집.
|
|
|
|
반환: [{teamA, teamB, group('A'..'L'), kickoffKst, venue}] (tla→코드)
|
|
"""
|
|
if not settings.football_data_token:
|
|
log.warning("schedule(all): football-data 토큰 미설정 — 수집 생략")
|
|
return []
|
|
import httpx
|
|
|
|
url = (
|
|
f"https://api.football-data.org/v4/competitions/"
|
|
f"{settings.football_data_competition}/matches"
|
|
)
|
|
async with httpx.AsyncClient(timeout=20) as c:
|
|
r = await c.get(url, headers={"X-Auth-Token": settings.football_data_token})
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
|
|
out: list[dict] = []
|
|
for m in data.get("matches", []):
|
|
if m.get("stage") != "GROUP_STAGE":
|
|
continue
|
|
g = (m.get("group") or "").replace("GROUP_", "") # 'A'..'L'
|
|
ht = (m.get("homeTeam") or {}).get("tla")
|
|
at = (m.get("awayTeam") or {}).get("tla")
|
|
if not ht or not at:
|
|
continue
|
|
utc = m.get("utcDate")
|
|
if not utc:
|
|
continue
|
|
kickoff = (
|
|
datetime.fromisoformat(utc.replace("Z", "+00:00")).astimezone(KST).isoformat()
|
|
)
|
|
out.append(
|
|
{
|
|
"teamA": code_for_tla(ht),
|
|
"teamB": code_for_tla(at),
|
|
"group": g,
|
|
"kickoffKst": kickoff,
|
|
"venue": m.get("venue") or "",
|
|
}
|
|
)
|
|
return out
|
|
|
|
|
|
async def fetch_schedule() -> list[dict]:
|
|
"""일정 수집. 전체 조 모드면 football-data 전 조별리그, 아니면 단일 조."""
|
|
if settings.schedule_all_groups:
|
|
try:
|
|
records = await _fetch_all_groups_football_data()
|
|
log.info("schedule: 전체 조별리그 %d경기 수집", len(records))
|
|
if records:
|
|
return records
|
|
except Exception as e: # noqa: BLE001
|
|
log.error("schedule(all) 실패: %s — 단일 조 폴백", e)
|
|
|
|
src = settings.schedule_source.lower()
|
|
try:
|
|
if src == "openfootball":
|
|
records = await _fetch_openfootball()
|
|
elif src == "football-data":
|
|
records = await _fetch_football_data()
|
|
else:
|
|
return []
|
|
log.info("schedule: %s 에서 %d경기 수집", src, len(records))
|
|
return records
|
|
except Exception as e: # noqa: BLE001
|
|
log.error("schedule fetch 실패(%s): %s — 기존 일정 유지", src, e)
|
|
return []
|
|
|
|
|
|
# ── 경기 결과(스코어) 수집 — 자동 정산용 ─────────────────────
|
|
def _extract_ft(m: dict) -> tuple[int, int] | None:
|
|
"""openfootball 매치에서 정규시간 스코어 추출(다양한 표기 포괄)."""
|
|
sc = m.get("score")
|
|
if isinstance(sc, dict):
|
|
ft = sc.get("ft")
|
|
if isinstance(ft, (list, tuple)) and len(ft) == 2 and ft[0] is not None:
|
|
return int(ft[0]), int(ft[1])
|
|
if m.get("score1") is not None and m.get("score2") is not None:
|
|
return int(m["score1"]), int(m["score2"])
|
|
return None
|
|
|
|
|
|
async def _results_openfootball() -> list[dict]:
|
|
import httpx
|
|
|
|
async with httpx.AsyncClient(timeout=20) as c:
|
|
r = await c.get(settings.schedule_url)
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
|
|
out: list[dict] = []
|
|
for m in data.get("matches", []):
|
|
if (m.get("group") or "").strip() != settings.schedule_group:
|
|
continue
|
|
a, b = code_for(m.get("team1", "")), code_for(m.get("team2", ""))
|
|
if not a or not b:
|
|
continue
|
|
ft = _extract_ft(m)
|
|
if ft is None: # 아직 미진행/미집계
|
|
continue
|
|
out.append({"teamA": a, "teamB": b, "scoreA": ft[0], "scoreB": ft[1]})
|
|
return out
|
|
|
|
|
|
async def _results_football_data() -> list[dict]:
|
|
if not settings.football_data_token:
|
|
return []
|
|
import httpx
|
|
|
|
url = (
|
|
f"https://api.football-data.org/v4/competitions/"
|
|
f"{settings.football_data_competition}/matches"
|
|
)
|
|
async with httpx.AsyncClient(timeout=20) as c:
|
|
r = await c.get(url, headers={"X-Auth-Token": settings.football_data_token})
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
|
|
out: list[dict] = []
|
|
for m in data.get("matches", []):
|
|
if m.get("status") != "FINISHED":
|
|
continue
|
|
a = code_for((m.get("homeTeam") or {}).get("name", "") or "")
|
|
b = code_for((m.get("awayTeam") or {}).get("name", "") or "")
|
|
if not a or not b:
|
|
continue
|
|
ft = ((m.get("score") or {}).get("fullTime") or {})
|
|
if ft.get("home") is None or ft.get("away") is None:
|
|
continue
|
|
out.append({"teamA": a, "teamB": b, "scoreA": int(ft["home"]), "scoreB": int(ft["away"])})
|
|
return out
|
|
|
|
|
|
async def fetch_results() -> list[dict]:
|
|
"""확정된 경기 스코어 수집 → [{teamA, teamB, scoreA, scoreB}]. 실패 시 빈 리스트.
|
|
|
|
소스는 result_source(없으면 schedule_source). openfootball 은 결과 미게시라
|
|
자동 종료에는 football-data 를 권장.
|
|
"""
|
|
src = settings.effective_result_source
|
|
try:
|
|
if src == "openfootball":
|
|
out = await _results_openfootball()
|
|
elif src == "football-data":
|
|
out = await _results_football_data()
|
|
else:
|
|
return []
|
|
if out:
|
|
log.info("results: %s 에서 %d경기 스코어 수집", src, len(out))
|
|
return out
|
|
except Exception as e: # noqa: BLE001
|
|
log.error("results fetch 실패(%s): %s", src, e)
|
|
return []
|