"""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, timedelta, timezone from sqlalchemy import select from sqlalchemy.orm import selectinload 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 def _match_key(home_team_id: int, away_team_id: int, match_date) -> tuple[int, int, str]: """比赛去重键:(主队, 客队, 天级日期 ISO 字符串)。 统一在这里构造,避免"预加载时用 str(date)、写入时用 isoformat()"这类 隐式格式依赖 —— 两者当前恰好相等,但一旦有人改动其一就会静默失配, 导致所有比赛被判为不存在而重复插入。 """ if hasattr(match_date, "date") and callable(match_date.date): match_date = match_date.date() return (home_team_id, away_team_id, match_date.isoformat() if match_date is not None else "") @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 控制。 P1-3: 批量查询优化,将单赛季 380 场 × 3 次 DB 往返降为 3 次查询。 """ from src.db.repositories import LeagueRepository, 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) # 查联赛 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 # === 批量优化: 一次规范化,收集球队名和日期 === normalized_matches: list = [] all_team_names: set[str] = set() 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 normalized_matches.append((nm, raw)) all_team_names.add(nm.home_team) all_team_names.add(nm.away_team) if not normalized_matches: return result # === 批量查询球队(1 次 DB 往返) === team_name_to_id = {} if all_team_names: teams = await team_repo.get_all_by_names(list(all_team_names)) team_name_to_id = {name: team.id for name, team in teams.items()} # === 批量查询已有比赛(1 次 DB 往返,按日期范围) === match_dict: dict[tuple, Match] = {} dates = [nm.date for nm, _ in normalized_matches if nm.date is not None] if dates: min_dt = min(dates) - timedelta(days=30) max_dt = max(dates) + timedelta(days=30) stmt = ( select(Match) .options(selectinload(Match.stats)) .where(Match.league_id == league_obj.id) .where(Match.match_date >= min_dt) .where(Match.match_date <= max_dt) ) for m in (await db.execute(stmt)).scalars(): key = _match_key(m.home_team_id, m.away_team_id, m.match_date_date) match_dict[key] = m # === 内存匹配 + 回填 xG === for nm, raw in normalized_matches: home_team_id = team_name_to_id.get(nm.home_team) away_team_id = team_name_to_id.get(nm.away_team) if home_team_id is None or away_team_id is None: result["unmatched"] += 1 continue match_key = _match_key(home_team_id, away_team_id, nm.date) existing = match_dict.get(match_key) 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