"""伤停数据采集器(api-football / api-sports.io)。 采集伤停数据并入库(injuries 表),供 injuries agent 使用。 """ from __future__ import annotations import asyncio import json import logging import random import time from datetime import datetime, timezone from pathlib import Path from typing import Any import httpx from src.core.config import settings from src.core.http_client import get_client logger = logging.getLogger(__name__) API_BASE = "https://v3.football.api-sports.io" DEFAULT_HOST = "v3.football.api-sports.io" # 缓存目录 _CACHE_DIR = Path(__file__).resolve().parent.parent.parent / "data" / "injuries_cache" async def fetch_injuries(*, date: str | None = None, fixture_id: int | None = None, league_id: int | None = None) -> list[dict]: """采集伤停数据。 Args: date: 日期 (YYYY-MM-DD),返当天全部伤停 fixture_id: 指定比赛 ID league_id: 指定联赛 ID Returns: 伤停记录列表 """ api_key = settings.API_FOOTBALL_KEY if not api_key: raise RuntimeError("API_FOOTBALL_KEY 未设置") cache_dir = _CACHE_DIR cache_dir.mkdir(parents=True, exist_ok=True) # 缓存命中 (7 天内有效) cache_key = f"injuries_{date}_{fixture_id}_{league_id}.json" cache_file = cache_dir / cache_key if cache_file.exists(): age_hours = (time.time() - cache_file.stat().st_mtime) / 3600 if age_hours < 168: # 7 天 logger.debug("injuries cache hit: %s (%.1fh old)", cache_key, age_hours) with open(cache_file, encoding="utf-8") as f: return json.load(f) else: logger.debug("injuries cache expired: %s (%.1fh old)", cache_key, age_hours) headers = { "x-apisports-key": api_key, "x-rapidapi-host": DEFAULT_HOST, } params: dict[str, Any] = {} if date: params["date"] = date if fixture_id: params["fixture"] = fixture_id if league_id: params["league"] = league_id url = f"{API_BASE}/injuries" # 重试 last_exc: Exception | None = None for attempt in range(3): try: client = get_client() resp = await client.get(url, headers=headers, params=params, timeout=30) resp.raise_for_status() break except Exception as e: last_exc = e if attempt == 2: raise delay = min(2 ** attempt, 8) + random.uniform(0, 1) logger.warning("injuries fetch failed, retry %d in %.1fs: %s", attempt + 1, delay, e) await asyncio.sleep(delay) else: raise RuntimeError(f"injuries fetch failed: {last_exc}") data = resp.json() injuries = data.get("response", []) # 写缓存 with open(cache_file, "w", encoding="utf-8") as f: json.dump(injuries, ensure_ascii=False, default=str, fp=f) return injuries async def ingest_injuries(db, *, date: str | None = None) -> dict: """采集伤停数据并入库(injuries 表)。 注意: 本方法不控制事务(commit/rollback),由调用方通过 UnitOfWork 控制。 """ from sqlalchemy import select from src.data.team_names import normalize as normalize_name from src.db.models import Injury, Team result = {"count": 0, "inserted": 0, "errors": []} try: raw_injuries = await fetch_injuries(date=date) except Exception as e: logger.exception("injuries fetch failed") result["errors"].append(f"fetch failed: {e}") return result result["count"] = len(raw_injuries) # 预加载所有球队(用于按名匹配) teams = (await db.execute(select(Team))).scalars().all() team_by_name = {t.name: t.id for t in teams} for raw in raw_injuries: try: player = raw.get("player", {}) or {} team = raw.get("team", {}) or {} fixture = raw.get("fixture", {}) or {} player_name = player.get("name", "") team_name = normalize_name(team.get("name", "")) team_id = team_by_name.get(team_name) # 解析日期 fixture_date = fixture.get("date") injury_date = None if fixture_date: try: dt = datetime.fromisoformat(fixture_date.replace("Z", "+00:00")) injury_date = dt.date() except (ValueError, AttributeError): pass player_id = player.get("id") fixture_id = fixture.get("id") # 幂等: 已存在则跳过 existing = ( await db.execute( select(Injury).where( Injury.player_id == player_id, Injury.fixture_id == fixture_id, Injury.injury_type == player.get("type"), ) ) ).scalar_one_or_none() if existing is not None: continue injury = Injury( player_id=player_id, player_name=player_name, team_id=team_id, fixture_id=fixture_id, league_id=(raw.get("league") or {}).get("id"), injury_type=player.get("type"), reason=player.get("reason"), injury_date=injury_date, ) db.add(injury) result["inserted"] += 1 except Exception as e: result["errors"].append(f"parse error: {e}") # 注意: 不在此处 commit,由调用方 UnitOfWork 控制事务 logger.info("injuries: fetched %d, inserted %d for %s", result["count"], result["inserted"], date) return result async def get_injuries_for_match(db, team_id: int, match_date, as_of=None) -> list[Injury]: """查询某场比赛前某队的伤停名单(比赛日仍缺阵的)。 Args: db: 数据库 session team_id: 球队 ID match_date: 比赛日期 as_of: 截止时间(cutoff)。只返回 retrieved_at <= as_of 的记录。 用于回测时防止"未来采集的数据"泄漏到历史预测。 必须保持 timezone-aware datetime,不会截断为 date。 """ from sqlalchemy import or_, select from src.db.models import Injury # 只处理 match_date:去掉时间部分,仅比较日期 if hasattr(match_date, "date"): match_date = match_date.date() stmt = ( select(Injury) .where(Injury.team_id == team_id) .where(Injury.injury_date <= match_date) .where(or_(Injury.return_date.is_(None), Injury.return_date >= match_date)) ) # 回测防泄漏: 只使用 as_of 时间点之前已采集的数据 # 注意: as_of 保持 datetime,不截断为 date,避免错误排除同日合法数据 if as_of is not None: stmt = stmt.where(Injury.retrieved_at.is_not(None)) stmt = stmt.where(Injury.retrieved_at <= as_of) stmt = stmt.order_by(Injury.injury_date.desc()) result = await db.execute(stmt) return list(result.scalars().all())