"""경기 일정 외부 수집 — 매일 워커가 호출(크롤링). 소스(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 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_schedule() -> list[dict]: """소스에 따라 일정 수집. 실패 시 빈 리스트(기존 일정 유지).""" 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}]. 실패 시 빈 리스트.""" src = settings.schedule_source.lower() 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 []