"""Understat xG 数据源。 迁移自旧项目 app/data/sources/understat.py,改成 async。 使用 Repository 模式进行数据访问,不直接控制事务。 """ from __future__ import annotations import asyncio import json import logging import random import re from datetime import datetime, timezone from sqlalchemy import func, select from src.core.http_client import get_client from src.data.config import FDCO_TO_UNDERSTAT, LEAGUE_NAMES from src.data.normalize import normalize_understat from src.data.sources import register from src.db.models import League, Match, MatchStats, Team logger = logging.getLogger(__name__) UNDERSTAT_BASE = "https://understat.com/getLeagueData/{league}/{season}" async def fetch_understat(league_code: str, season: int) -> list[dict]: """抓取 understat 单赛季 xG 数据。 Args: league_code: fdco 风格代码,如 'E0' season: 赛季起始年,如 2025 表示 2025-2026 赛季 Returns: 比赛数组,每项含 datetime/h/a/xG """ understat_league = FDCO_TO_UNDERSTAT.get(league_code) if understat_league is None: raise ValueError(f"未知联赛代码: {league_code}") url = UNDERSTAT_BASE.format(league=understat_league, season=season) headers = { "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36", "X-Requested-With": "XMLHttpRequest", "Referer": f"https://understat.com/league/{understat_league}/{season}", } # 重试:网络错误 / 5xx / 429 last_exc: Exception | None = None for attempt in range(3): try: client = get_client() resp = await client.get(url, headers=headers, 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("understat fetch failed, retry %d in %.1fs: %s", attempt + 1, delay, e) await asyncio.sleep(delay) else: raise RuntimeError(f"understat fetch failed: {last_exc}") # understat 返回 JS 对象,需要提取 JSON text = resp.text # 匹配 var datesData = JSON.parse('...'); match = re.search(r"var\s+datesData\s*=\s*JSON\.parse\('([^']+)'\)", text) if not match: logger.warning("understat 响应格式不符: %s...", text[:200]) return [] decoded = match.group(1).encode().decode("unicode_escape") data = json.loads(decoded) return data @register class UnderstatSource: """understat xG 数据源(实现 DataSource 协议)。""" name = "understat" async def ingest(self, db, *, league: str, season: int) -> dict: """采集 understat xG → 回填到现有 Match。只回填 xG 字段,不创建新 Match。 注意: 本方法不控制事务(commit/rollback),由调用方通过 UnitOfWork 控制。 """ from src.db.repositories import LeagueRepository, MatchRepository, TeamRepository result = {"updated": 0, "skipped": 0, "unmatched": 0, "errors": []} try: raw_matches = await fetch_understat(league, season) except Exception as e: logger.exception("understat fetch failed for %s %s", league, season) result["errors"].append(f"fetch failed: {e}") return result # 使用 Repository league_repo = LeagueRepository(db) team_repo = TeamRepository(db) match_repo = MatchRepository(db) # 查联赛 league_obj = await league_repo.get_by_code(league) if league_obj is None: result["errors"].append(f"league {league} not found in DB") return result for raw in raw_matches: if not raw.get("isResult"): continue try: nm = normalize_understat(raw, league) if nm is None: result["skipped"] += 1 continue except Exception as e: result["errors"].append(f"normalize: {e}") continue # 匹配已有 Match(天级) - 使用 Repository home_team = await team_repo.get_by_name(nm.home_team) away_team = await team_repo.get_by_name(nm.away_team) if home_team is None or away_team is None: result["unmatched"] += 1 continue existing = await match_repo.find_by_teams_and_date( league_obj.id, home_team.id, away_team.id, nm.date ) if existing is None: result["unmatched"] += 1 continue # 回填 xG if existing.stats is None and (nm.home_xg is not None or nm.away_xg is not None): now = datetime.now(timezone.utc) existing.stats = MatchStats( match_id=existing.id, source="understat", source_event_id=str(raw.get("id", "")), retrieved_at=now, available_at=now, ) db.add(existing.stats) await db.flush() if existing.stats is not None: if existing.stats.home_xg is None and nm.home_xg is not None: existing.stats.home_xg = nm.home_xg result["updated"] += 1 if existing.stats.away_xg is None and nm.away_xg is not None: existing.stats.away_xg = nm.away_xg # 注意: 不在此处 commit,由调用方 UnitOfWork 控制事务 return result