306 lines
13 KiB
Python
306 lines
13 KiB
Python
"""축구 데이터 연동 (API-Football 무료 티어) — 예측 프롬프트용 수집·캐싱·조립.
|
|
|
|
설계 요약
|
|
- 수집(refresh): 워커가 다가오는 경기의 팀/H2H 를 무료 한도(100/일·10/분) 안에서
|
|
점진 갱신해 FootballCache(DB)에 저장. 호출 간 간격(throttle)으로 분당 제한 회피.
|
|
- 조립(build_data_block): 예측 생성 시 캐시만 읽어 프롬프트 블록을 만든다(외부 호출 X).
|
|
2026 실제 결과는 우리 DB(matches)에서 직접 계산해 보완(무료 API 가 2026 시즌 차단).
|
|
- FOOTBALL_API_KEY 미설정이면 전부 no-op → 기존(이름만) 예측으로 자연 폴백.
|
|
|
|
무료 티어 제약 우회
|
|
- season=2026, last=N 파라미터는 막힘 → season(2024,2023) 조회로 폼/H2H 확보.
|
|
- FIFA 랭킹은 API 에 없음 → 아래 FIFA_RANK(수동 관리, 외부 무료 소스 기반)로 보완.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from datetime import timedelta
|
|
|
|
import httpx
|
|
from sqlalchemy import or_, select
|
|
|
|
from ..config import settings
|
|
from ..domain import ensure_aware, now_utc
|
|
from ..models import FootballCache, Match
|
|
|
|
log = logging.getLogger("triplepick.football")
|
|
|
|
# 무료 접근 가능한 시즌(최신 우선). 폼 표본 확보용.
|
|
_SEASONS = (2024, 2023)
|
|
|
|
# FIFA 랭킹(수동 관리) — API-Football 에 없어 외부 무료 소스값을 코드코어로 둔다.
|
|
# 없는 팀은 블록에서 랭킹 줄을 생략(폴백). 갱신은 가끔 수동.
|
|
# FIFA 세계랭킹 — 공식 발표 기준(2026-06-11). API-Football 엔 없어 외부값을 코드에 둔다.
|
|
# 갱신: FIFA 공식 랭킹 새로 나오면 숫자만 교체.
|
|
FIFA_RANK: dict[str, int] = {
|
|
"MEX": 13, "RSA": 61, "KOR": 22, "CZE": 43, # A
|
|
"SUI": 19, "BIH": 63, "CAN": 32, "QAT": 49, # B
|
|
"SCO": 38, "MAR": 7, "BRA": 6, "HAI": 84, # C
|
|
"USA": 15, "AUS": 23, "TUR": 26, "PAR": 42, # D
|
|
"GER": 9, "CIV": 29, "ECU": 28, "CUW": 82, # E
|
|
"NED": 8, "SWE": 35, "TUN": 56, "JPN": 18, # F
|
|
"BEL": 10, "EGY": 30, "IRN": 20, "NZL": 85, # G
|
|
"ESP": 2, "CPV": 67, "KSA": 60, "URU": 17, # H
|
|
"FRA": 3, "SEN": 16, "IRQ": 57, "NOR": 31, # I
|
|
"ARG": 1, "ALG": 27, "AUT": 24, "JOR": 64, # J
|
|
"POR": 5, "COD": 45, "UZB": 50, "COL": 14, # K
|
|
"ENG": 4, "CRO": 11, "GHA": 73, "PAN": 34, # L
|
|
}
|
|
|
|
# 팀코드→API-Football 팀ID 직접 매핑(이름검색 오인 방지 — 청소년/여자/표기차).
|
|
# 48개 본선 팀 전체. 2026 본선 확정명단 기준 시니어 대표팀 ID.
|
|
TEAM_ID_OVERRIDE: dict[str, int] = {
|
|
"MEX": 16, "RSA": 1531, "KOR": 17, "CZE": 770, # A
|
|
"SUI": 15, "BIH": 1113, "CAN": 5529, "QAT": 1569, # B
|
|
"SCO": 1108, "MAR": 31, "BRA": 6, "HAI": 2386, # C
|
|
"USA": 2384, "AUS": 20, "TUR": 777, "PAR": 2380, # D
|
|
"GER": 25, "CIV": 1501, "ECU": 2382, "CUW": 5530, # E
|
|
"NED": 1118, "SWE": 5, "TUN": 28, "JPN": 12, # F
|
|
"BEL": 1, "EGY": 32, "IRN": 22, "NZL": 4673, # G
|
|
"ESP": 9, "CPV": 1533, "KSA": 23, "URU": 7, # H
|
|
"FRA": 2, "SEN": 13, "IRQ": 1567, "NOR": 1090, # I
|
|
"ARG": 26, "ALG": 1532, "AUT": 775, "JOR": 1548, # J
|
|
"POR": 27, "COD": 1508, "UZB": 1568, "COL": 8, # K
|
|
"ENG": 10, "CRO": 3, "GHA": 1504, "PAN": 11, # L
|
|
}
|
|
|
|
|
|
def _is_senior(name: str) -> bool:
|
|
"""청소년/여자/올림픽 팀 제외(시니어 대표팀만)."""
|
|
n = name.upper()
|
|
if any(b in n for b in ("U23", "U21", "U20", "U19", "U17", "U-23", "WOMEN", "OLYMPIC")):
|
|
return False
|
|
return not n.endswith(" W")
|
|
|
|
|
|
def enabled() -> bool:
|
|
return bool(settings.football_api_key)
|
|
|
|
|
|
# ── 저수준: API 호출 (throttle 포함) ───────────────────────────
|
|
async def _get(client: httpx.AsyncClient, path: str, **params) -> dict:
|
|
r = await client.get(
|
|
settings.football_api_base + path,
|
|
headers={"x-apisports-key": settings.football_api_key},
|
|
params=params or None,
|
|
timeout=25,
|
|
)
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
# API-Football 은 한도초과·레이트·파라미터 오류를 HTTP 200 + errors 로 준다.
|
|
# 빈 응답을 정상으로 오해해 빈 데이터를 캐싱하지 않도록 여기서 예외 발생.
|
|
errs = data.get("errors")
|
|
if errs: # 빈 리스트([])는 정상, 채워진 dict 면 오류
|
|
raise RuntimeError(f"API-Football error: {errs}")
|
|
await asyncio.sleep(settings.football_call_interval_sec) # 10/분 제한 회피
|
|
return data
|
|
|
|
|
|
# ── 캐시 upsert ────────────────────────────────────────────────
|
|
async def _upsert(db, key: str, payload: dict) -> None:
|
|
row = await db.get(FootballCache, key)
|
|
if row:
|
|
row.payload = payload
|
|
row.fetched_at = now_utc()
|
|
else:
|
|
db.add(FootballCache(key=key, payload=payload, fetched_at=now_utc()))
|
|
|
|
|
|
async def _team_id(db, client, code: str, name: str) -> int | None:
|
|
"""팀코드→API팀ID. 1)직접매핑 2)캐시 3)이름검색(시니어 대표팀 필터)."""
|
|
if code in TEAM_ID_OVERRIDE:
|
|
return TEAM_ID_OVERRIDE[code]
|
|
cached = await db.get(FootballCache, f"teamid:{code}")
|
|
if cached and cached.payload.get("id"):
|
|
return cached.payload["id"]
|
|
query = name.split(" (")[0].strip() # "Korea Republic (...)" → "Korea Republic"
|
|
data = await _get(client, "/teams", search=query)
|
|
nats = [
|
|
x["team"] for x in (data.get("response") or [])
|
|
if x["team"].get("national") and _is_senior(x["team"]["name"])
|
|
]
|
|
if not nats:
|
|
return None
|
|
tid = nats[0]["id"]
|
|
await _upsert(db, f"teamid:{code}", {"id": tid, "name": nats[0]["name"]})
|
|
return tid
|
|
|
|
|
|
# ── 팀 베이스라인 수집 → 압축 payload ──────────────────────────
|
|
async def _fetch_team(db, client, code: str, name: str) -> bool:
|
|
tid = await _team_id(db, client, code, name)
|
|
if not tid:
|
|
log.warning("football: team id 미확인 (%s/%s)", code, name)
|
|
return False
|
|
|
|
sq = await _get(client, "/players/squads", team=tid)
|
|
players = (sq.get("response") or [{}])[0].get("players", []) if sq.get("response") else []
|
|
|
|
fixtures = []
|
|
for season in _SEASONS:
|
|
d = await _get(client, "/fixtures", team=tid, season=season)
|
|
fixtures += d.get("response", [])
|
|
|
|
results = []
|
|
for f in fixtures:
|
|
g = f["goals"]
|
|
home, away = f["teams"]["home"], f["teams"]["away"]
|
|
is_home = home["id"] == tid
|
|
gf = g["home"] if is_home else g["away"]
|
|
ga = g["away"] if is_home else g["home"]
|
|
if gf is None or ga is None:
|
|
continue
|
|
r = "W" if gf > ga else "L" if gf < ga else "D"
|
|
results.append({"date": f["fixture"]["date"][:10], "r": r, "gf": gf, "ga": ga})
|
|
results.sort(key=lambda x: x["date"])
|
|
|
|
n = max(len(results), 1)
|
|
w = sum(1 for x in results if x["r"] == "W")
|
|
d_ = sum(1 for x in results if x["r"] == "D")
|
|
l = sum(1 for x in results if x["r"] == "L")
|
|
payload = {
|
|
"team_id": tid,
|
|
"form": " ".join(x["r"] for x in results[-8:]),
|
|
"w": w, "d": d_, "l": l,
|
|
"gf_avg": round(sum(x["gf"] for x in results) / n, 2),
|
|
"ga_avg": round(sum(x["ga"] for x in results) / n, 2),
|
|
"clean_sheets": sum(1 for x in results if x["ga"] == 0),
|
|
"stars": [p["name"] for p in players if p.get("position") == "Attacker"][:5],
|
|
"squad_n": len(players),
|
|
}
|
|
await _upsert(db, f"team:{code}", payload)
|
|
return True
|
|
|
|
|
|
async def _fetch_h2h(db, client, code_a, id_a, code_b, id_b) -> None:
|
|
d = await _get(client, "/fixtures/headtohead", h2h=f"{id_a}-{id_b}")
|
|
rows = []
|
|
for f in d.get("response", []):
|
|
g = f["goals"]
|
|
if g["home"] is None:
|
|
continue
|
|
rows.append({
|
|
"date": f["fixture"]["date"][:10],
|
|
"home": f["teams"]["home"]["name"], "hg": g["home"],
|
|
"ag": g["away"], "away": f["teams"]["away"]["name"],
|
|
})
|
|
await _upsert(db, f"h2h:{code_a}-{code_b}", {"rows": rows[-5:]})
|
|
|
|
|
|
# ── 수집 잡 (워커 호출) ─────────────────────────────────────────
|
|
async def refresh(db, matches: list[Match], limit: int | None = None) -> int:
|
|
"""주어진 경기들의 팀/H2H 캐시를 stale 우선으로 갱신. 갱신한 팀 수 반환."""
|
|
if not enabled():
|
|
return 0
|
|
limit = limit if limit is not None else settings.football_refresh_batch
|
|
cutoff = now_utc() - timedelta(hours=settings.football_cache_ttl_hours)
|
|
refreshed = 0
|
|
async with httpx.AsyncClient() as client:
|
|
seen: set[str] = set()
|
|
for m in matches:
|
|
for code, name in ((m.team_a_code, m.team_a_name), (m.team_b_code, m.team_b_name)):
|
|
if code in seen or refreshed >= limit:
|
|
continue
|
|
seen.add(code)
|
|
row = await db.get(FootballCache, f"team:{code}")
|
|
if row and ensure_aware(row.fetched_at) > cutoff:
|
|
continue # 신선 — skip
|
|
try:
|
|
if await _fetch_team(db, client, code, name):
|
|
refreshed += 1
|
|
except Exception as e: # noqa: BLE001
|
|
log.warning("football: 팀 %s 수집 실패 %s", code, e)
|
|
# H2H (양 팀 캐시가 있을 때만)
|
|
for m in matches:
|
|
h = await db.get(FootballCache, f"h2h:{m.team_a_code}-{m.team_b_code}")
|
|
if h and ensure_aware(h.fetched_at) > cutoff:
|
|
continue
|
|
ta = await db.get(FootballCache, f"team:{m.team_a_code}")
|
|
tb = await db.get(FootballCache, f"team:{m.team_b_code}")
|
|
if not (ta and tb):
|
|
continue
|
|
try:
|
|
await _fetch_h2h(db, client, m.team_a_code, ta.payload["team_id"],
|
|
m.team_b_code, tb.payload["team_id"])
|
|
except Exception as e: # noqa: BLE001
|
|
log.warning("football: H2H %s 수집 실패 %s", m.match_id, e)
|
|
await db.commit()
|
|
if refreshed:
|
|
log.info("football: 팀 %d개 캐시 갱신", refreshed)
|
|
return refreshed
|
|
|
|
|
|
# ── 라이브 WC 결과 (우리 DB) ───────────────────────────────────
|
|
async def _live_wc(db, code: str, before) -> list[str]:
|
|
rows = (await db.execute(
|
|
select(Match)
|
|
.where(
|
|
or_(Match.team_a_code == code, Match.team_b_code == code),
|
|
Match.result_outcome.isnot(None),
|
|
Match.kickoff_at < before,
|
|
)
|
|
.order_by(Match.kickoff_at)
|
|
)).scalars().all()
|
|
out = []
|
|
for m in rows:
|
|
is_a = m.team_a_code == code
|
|
gf = m.result_score_a if is_a else m.result_score_b
|
|
ga = m.result_score_b if is_a else m.result_score_a
|
|
opp = m.team_b_short if is_a else m.team_a_short
|
|
if gf is None or ga is None:
|
|
continue
|
|
r = "W" if gf > ga else "L" if gf < ga else "D"
|
|
out.append(f"{r} {gf}-{ga} vs {opp}")
|
|
return out
|
|
|
|
|
|
def _h2h_line(row, default_code: str) -> str:
|
|
if not row or not row.payload.get("rows"):
|
|
return "no recent meetings"
|
|
rows = row.payload["rows"][-3:]
|
|
return "; ".join(
|
|
f'{r["home"]} {r["hg"]}-{r["ag"]} {r["away"]} ({r["date"][:4]})' for r in rows
|
|
)
|
|
|
|
|
|
async def _team_lines(db, code, name, row, kickoff) -> str:
|
|
rank = FIFA_RANK.get(code)
|
|
head = f"[{name}]" + (f" FIFA #{rank}" if rank else "")
|
|
if not row:
|
|
return f"{head}\n (no external data — use general knowledge)"
|
|
p = row.payload
|
|
live = await _live_wc(db, code, kickoff)
|
|
wc = "; ".join(live) if live else "(none yet)"
|
|
return (
|
|
f"{head}\n"
|
|
f" Form(last8): {p.get('form','')} | {p.get('w',0)}-{p.get('d',0)}-{p.get('l',0)} "
|
|
f"(W-D-L), avg {p.get('gf_avg','?')}-{p.get('ga_avg','?')}, "
|
|
f"clean sheets {p.get('clean_sheets',0)}\n"
|
|
f" WC2026 so far: {wc}\n"
|
|
f" Key attackers: {', '.join(p.get('stars', []))}"
|
|
)
|
|
|
|
|
|
async def build_data_block(db, match: Match) -> str | None:
|
|
"""경기 예측 프롬프트에 끼울 데이터블록. 캐시만 읽음(외부 호출 X).
|
|
양 팀 모두 데이터가 없으면 None → 기존(이름만) 동작 폴백."""
|
|
if not enabled():
|
|
return None
|
|
ta = await db.get(FootballCache, f"team:{match.team_a_code}")
|
|
tb = await db.get(FootballCache, f"team:{match.team_b_code}")
|
|
# 라이브 WC 결과는 캐시 없어도 의미 있으므로, 캐시가 둘 다 없을 때만 폴백
|
|
has_live = bool(await _live_wc(db, match.team_a_code, match.kickoff_at)) or \
|
|
bool(await _live_wc(db, match.team_b_code, match.kickoff_at))
|
|
if not ta and not tb and not has_live:
|
|
return None
|
|
h2h = (await db.get(FootballCache, f"h2h:{match.team_a_code}-{match.team_b_code}")) or \
|
|
(await db.get(FootballCache, f"h2h:{match.team_b_code}-{match.team_a_code}"))
|
|
return "\n".join([
|
|
"=== MATCH DATA (factual, weigh heavily over priors) ===",
|
|
await _team_lines(db, match.team_a_code, match.team_a_name, ta, match.kickoff_at),
|
|
await _team_lines(db, match.team_b_code, match.team_b_name, tb, match.kickoff_at),
|
|
f"[Head-to-head] {_h2h_line(h2h, match.team_a_code)}",
|
|
"===",
|
|
])
|