shangfangjian/Profeto
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: 以 bzzoiro 为唯一数据源的全面重构

Browse Source 
数据源统一为 bzzoiro,移除 Understat 与 injuries:
- 删除 src/data/understat.py / injuries.py 及相关测试
- 删除 injuries 模型与表;扩展 match_stats(xG 之外增加 big_chances/fouls)
- 新增 standings 表(联赛积分榜:位置/积分/xG/走势/分区)
- matches 表增加 source_event_id 血缘列,支撑统计回填

采集管线(bzzoiro 三条管线):
- events:赛程/比分(/events/),记录 source_event_id
- standings:积分榜快照(/leagues/{id}/standings/)
- stats:已完赛比赛详细统计回填(/events/{id}/stats/)

预测增强:
- standings_slice 替代 injuries_slice;积分榜专家替代阵容完整性专家
- AGENT_META runtime_config 同步更新

管理后台:
- ingest 路由重写为单一 bzzoiro 入口 + task 参数(events/standings/stats/all)
- 新增 /admin/data-completeness 数据完整性分析 API
- 数据源状态页简化为 bzzoiro 单源

前端:
- 采集页重构为任务驱动(比赛/积分榜/统计回填/全量)
- 新增「数据完整性」可视化页(覆盖率矩阵/字段完整率/健康摘要)
- 新增主站积分榜页(/standings)与比赛详情完整统计面板
- agent 名称同步更新(injuries→standings)

迁移 0015_bzzoiro_single_source 已在容器内验证通过,后端测试全部通过。

Co-Authored-By: new-provider/LongCat-2.0 <<EMAIL>>
This commit is contained in:
shangfangjian
2026-09-20 19:13:07 +08:00
co-authored by new-provider/LongCat-2.0 <
parent ec8f36abb2
commit f05dc1ae15
41 changed files with 1603 additions and 1885 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/data/bzzoiro.py
+326 -64
View File
@@ -1,12 +1,15 @@
"""Bzzoiro 数据源:抓取 + 入库。
"""Bzzoiro 数据源:抓取 + 入库(单一数据源)。
迁移自旧项目 app/data/sources/bzzoiro/,改成 async + 简化入库。
使用 Repository 模式进行数据访问,不直接控制事务。
三条管线:
1. events — 比赛日程/比分(/events/),含 source_event_id 血缘
2. standings— 联赛积分榜快照(/leagues/{id}/standings/)
3. stats — 已完赛比赛详细统计回填(/events/{id}/stats/)
使用 Repository 模式进行数据访问,不直接控制事务(由调用方 UnitOfWork 控制)。
"""
from __future__ import annotations
import asyncio
import json as _json
import logging
import random
from collections.abc import Iterable
@@ -22,7 +25,7 @@ from src.data.config import BZZOIRO_LEAGUE_IDS, LEAGUE_COUNTRIES, LEAGUE_NAMES,
from src.data.normalize import normalize_bzzoiro
from src.data.team_names_zh import zh_name
from src.data.sources import register
from src.db.models import League, Match, MatchStats, Team
from src.db.models import League, Match, MatchStats, Standing, Team
logger = logging.getLogger(__name__)
@@ -36,6 +39,16 @@ def _to_date(value):
return value
def _to_int_or_none(value) -> int | None:
"""宽松转 int(用于上游 ID 解析,失败返回 None 不抛错)。"""
if value is None:
return None
try:
return int(str(value).strip())
except (TypeError, ValueError):
return None
def _match_key(home_team_id: int, away_team_id: int, match_date) -> tuple[int, int, str]:
"""比赛去重键:(主队, 客队, 天级日期 ISO 字符串)。
@@ -262,38 +275,13 @@ class BzzoiroSource:
home_ht_goals=nm.home_ht_goals,
away_ht_goals=nm.away_ht_goals,
match_stage=nm.match_stage,
source_event_id=_to_int_or_none(raw.get("id")),
)
db.add(m)
await db.flush()
existing_matches[match_key] = m # 防止同批重复
# 存在任一统计字段即可创建 MatchStats(不再强制要求 xG)
if any(getattr(nm, f) is not None for f in ['home_xg', 'away_xg', 'home_shots', 'away_shots', 'home_shots_on_target', 'away_shots_on_target', 'home_corners', 'away_corners', 'home_possession', 'home_yellow_cards', 'away_yellow_cards', 'home_red_cards', 'away_red_cards']):
now = datetime.now(timezone.utc)
# available_at 语义:统计「可被使用」的最早时间,至少不早于比赛结束
# 近似:开球 + 2 小时(实际完赛时间约为 +2 小时,非官方公布时间)
# 回测 cutoff 若贴着开球,不会把「完赛后才有的统计」误标为赛前可用
available_at = nm.date + timedelta(hours=2) if nm.date else now
stats = MatchStats(
match_id=m.id,
home_xg=nm.home_xg,
away_xg=nm.away_xg,
home_shots=nm.home_shots,
away_shots=nm.away_shots,
home_shots_on_target=nm.home_shots_on_target,
away_shots_on_target=nm.away_shots_on_target,
home_corners=nm.home_corners,
away_corners=nm.away_corners,
home_possession=nm.home_possession,
home_yellow_cards=nm.home_yellow_cards,
away_yellow_cards=nm.away_yellow_cards,
home_red_cards=nm.home_red_cards,
away_red_cards=nm.away_red_cards,
source="bzzoiro",
source_record_id=str(raw.get("id", "")),
retrieved_at=now,
available_at=available_at,
)
db.add(stats)
# 统计字段不在 /events/ 载荷中(单独由 stats 管线回填),
# 此处不再创建 MatchStats。
league_r["inserted"] += 1
else:
# 已有比赛: 直接从内存获取对象更新(无需再查询)
@@ -310,37 +298,11 @@ class BzzoiroSource:
if existing_match.match_stage is None and nm.match_stage:
existing_match.match_stage = nm.match_stage
changed = True
# 存在任一统计字段即可创建 MatchStats(不再强制要求 xG)
if existing_match.stats is None and (
nm.home_xg is not None or nm.away_xg is not None
or nm.home_shots is not None or nm.away_shots is not None
or nm.home_corners is not None or nm.away_corners is not None
or nm.home_possession is not None
):
now = datetime.now(timezone.utc)
# available_at 语义:统计「可被使用」的最早时间,至少不早于比赛结束
# 近似:开球 + 2 小时(实际完赛时间约为 +2 小时,非官方公布时间)
available_at = nm.date + timedelta(hours=2) if nm.date else now
existing_match.stats = MatchStats(
match_id=existing_match.id,
source="bzzoiro",
source_record_id=str(raw.get("id", "")),
retrieved_at=now,
available_at=available_at,
)
db.add(existing_match.stats)
await db.flush()
if existing_match.stats is not None:
for fld in ("home_xg", "away_xg", "home_shots", "away_shots",
"home_shots_on_target", "away_shots_on_target",
"home_corners", "away_corners", "home_possession",
"home_yellow_cards", "away_yellow_cards",
"home_red_cards", "away_red_cards"):
if getattr(existing_match.stats, fld, None) is None:
v = getattr(nm, fld, None)
if v is not None:
setattr(existing_match.stats, fld, v)
changed = True
if existing_match.source_event_id is None:
eid = _to_int_or_none(raw.get("id"))
if eid is not None:
existing_match.source_event_id = eid
changed = True
if changed:
league_r["updated"] += 1
@@ -349,3 +311,303 @@ class BzzoiroSource:
result["total_inserted"] += league_r["inserted"]
result["total_updated"] += league_r["updated"]
return result
# ============================================================
# 积分榜管线:/leagues/{id}/standings/ → standings 表
# ============================================================
async def fetch_bzzoiro_standings(league_code: str, season: str | None = None) -> dict:
"""抓取联赛积分榜(纯抓取,不入库)。season 为 None 时取当前赛季。"""
league_id = BZZOIRO_LEAGUE_IDS.get(league_code)
if league_id is None:
raise ValueError(f"未知联赛代码: {league_code}")
params: dict = {}
if season:
params["season"] = season
return await _fetch_json_async(f"/leagues/{league_id}/standings/", params)
def _season_label_from_dates(start_date, end_date) -> str:
"""从赛季起止日期推导赛季标签(与 derive_season_label 语义一致)。"""
try:
if isinstance(start_date, str):
start = datetime.fromisoformat(start_date[:10])
else:
start = start_date
y = start.year
return f"{y}-{y + 1}" if start.month >= 8 else f"{y - 1}-{y}"
except (TypeError, ValueError):
return "?"
async def ingest_bzzoiro_standings(db, *, leagues: Iterable[str], season: str | None = None) -> dict:
"""采集积分榜 → upsert standings 表。
season 为 None 时采集当前赛季(bzzoiro 默认返回 is_current 赛季)。
球队名与 events 管线使用同一 normalize 规则,保证 Team 匹配。
"""
from src.data.team_names import normalize as normalize_name
result: dict = {"leagues": {}, "total_upserted": 0, "errors": []}
for code in leagues:
league_r: dict = {"upserted": 0, "teams_created": 0, "rows": 0}
try:
payload = await fetch_bzzoiro_standings(code, season=season)
except Exception as e:
logger.exception("bzzoiro standings fetch failed for %s", code)
result["leagues"][code] = {"error": str(e)}
result["errors"].append(f"{code}: {e}")
continue
rows = payload.get("standings") or []
if not rows:
result["leagues"][code] = {"error": "无积分榜数据(赛季未开始或未提供)"}
result["errors"].append(f"{code}: 无积分榜数据")
continue
# 联赛
stmt = select(League).where(League.code == code)
league = (await db.execute(stmt)).scalar_one_or_none()
if league is None:
league = League(code=code, name=LEAGUE_NAMES.get(code, code), country=LEAGUE_COUNTRIES.get(code))
db.add(league)
await db.flush()
# 赛季标签:优先用返回的 season 对象推导
season_obj = payload.get("season") or {}
season_label = _season_label_from_dates(
season_obj.get("start_date"), season_obj.get("end_date")
)
if season_label == "?":
season_label = season or ""
# 批量预载球队
names = {normalize_name(str(r.get("team_name", ""))) for r in rows}
names.discard("")
team_map: dict[str, Team] = {}
if names:
stmt = select(Team).where(Team.name.in_(names))
for t in (await db.execute(stmt)).scalars():
team_map[t.name] = t
now = datetime.now(timezone.utc)
for r in rows:
team_name = normalize_name(str(r.get("team_name", "")))
if not team_name:
continue
team = team_map.get(team_name)
if team is None:
team = Team(name=team_name, name_zh=zh_name(team_name))
db.add(team)
await db.flush()
team_map[team_name] = team
league_r["teams_created"] += 1
zone = r.get("zone") or {}
values = dict(
position=_to_int_or_none(r.get("position")) or 0,
played=_to_int_or_none(r.get("played")) or 0,
won=_to_int_or_none(r.get("won")) or 0,
drawn=_to_int_or_none(r.get("drawn")) or 0,
lost=_to_int_or_none(r.get("lost")) or 0,
goals_for=_to_int_or_none(r.get("gf")) or 0,
goals_against=_to_int_or_none(r.get("ga")) or 0,
goal_diff=_to_int_or_none(r.get("gd")) or 0,
points=_to_int_or_none(r.get("pts")) or 0,
xg_for=_to_float_or_none(r.get("xgf")),
xg_against=_to_float_or_none(r.get("xga")),
form=r.get("form") or None,
zone=zone.get("label") or zone.get("key") or None,
updated_at=now,
retrieved_at=now,
)
stmt = select(Standing).where(
Standing.league_id == league.id,
Standing.season == season_label,
Standing.team_id == team.id,
)
standing = (await db.execute(stmt)).scalar_one_or_none()
if standing is None:
standing = Standing(
league_id=league.id, season=season_label, team_id=team.id, **values
)
db.add(standing)
else:
for k, v in values.items():
setattr(standing, k, v)
league_r["upserted"] += 1
league_r["rows"] = len(rows)
result["leagues"][code] = league_r
result["total_upserted"] += league_r["upserted"]
logger.info(
"bzzoiro standings 采集完成: %s 赛季 %s, upsert %d/%d",
code, season_label, league_r["upserted"], league_r["rows"],
)
return result
# ============================================================
# 统计回填管线:/events/{id}/stats/ → match_stats 表
# ============================================================
# bzzoiro stats 字段 → MatchStats 字段映射(stats.home / stats.away 下)
_STATS_FIELD_MAP = {
"xg": ("home_xg", "away_xg"), # 回退 expected_goals
"ball_possession": ("home_possession", None), # 只取主队值,客队=100-home
"total_shots": ("home_shots", "away_shots"),
"shots_on_target": ("home_shots_on_target", "away_shots_on_target"),
"corner_kicks": ("home_corners", "away_corners"),
"yellow_cards": ("home_yellow_cards", "away_yellow_cards"),
"red_cards": ("home_red_cards", "away_red_cards"),
"big_chances": ("home_big_chances", "away_big_chances"),
"fouls": ("home_fouls", "away_fouls"),
}
def _pick(d: dict, *keys):
"""按优先级取第一个非空字段值。"""
for k in keys:
v = d.get(k)
if v is not None:
return v
return None
def _stats_from_payload(payload: dict) -> dict:
"""把 /events/{id}/stats/ 响应映射成 MatchStats 字段 dict。
响应结构: {"event_id": ..., "stats": {"home": {...}, "away": {...}}}
"""
stats = (payload or {}).get("stats") or {}
home = stats.get("home") or {}
away = stats.get("away") or {}
out: dict = {}
xg_h = _pick(home, "xg", "expected_goals")
xg_a = _pick(away, "xg", "expected_goals")
if xg_h is not None:
out["home_xg"] = _to_float_or_none(xg_h)
if xg_a is not None:
out["away_xg"] = _to_float_or_none(xg_a)
poss = home.get("ball_possession")
if poss is not None:
p = _to_float_or_none(poss)
if p is not None:
out["home_possession"] = p
out["away_possession"] = round(100 - p, 1) if 0 <= p <= 100 else None
for src, (h_fld, a_fld) in _STATS_FIELD_MAP.items():
if src in ("xg", "ball_possession"):
continue # 已处理
hv = home.get(src)
av = away.get(src)
if hv is not None and h_fld:
out[h_fld] = _to_int_or_none(hv)
if av is not None and a_fld:
out[a_fld] = _to_int_or_none(av)
return out
def _to_float_or_none(value) -> float | None:
if value is None:
return None
try:
return float(str(value).strip())
except (TypeError, ValueError):
return None
async def ingest_bzzoiro_event_stats(
db,
*,
leagues: Iterable[str],
limit: int = 100,
only_missing: bool = True,
) -> dict:
"""回填已完赛比赛的详细统计(逐场调 /events/{id}/stats/)。
筛选条件: match_status=finished 且 source_event_id 非空。
only_missing=True 时跳过已有统计的比赛(增量);False 则全量刷新。
limit 控制单次最多处理的比赛数(上游限速 1.2s/请求,大批量需分次触发)。
"""
result: dict = {"fetched": 0, "created": 0, "updated": 0, "skipped": 0, "errors": []}
league_ids = [BZZOIRO_LEAGUE_IDS[c] for c in leagues if c in BZZOIRO_LEAGUE_IDS]
if not league_ids:
result["errors"].append("无有效联赛代码")
return result
stmt = (
select(Match)
.options(select(Match.stats))
.where(Match.match_status == "finished")
.where(Match.source_event_id.is_not(None))
.where(Match.league_id.in_(league_ids))
.order_by(Match.match_date.desc())
.limit(limit * 3 if only_missing else limit)
)
matches = (await db.execute(stmt)).scalars().all()
now = datetime.now(timezone.utc)
processed = 0
for m in matches:
if processed >= limit:
break
if only_missing and m.stats is not None and m.stats.home_shots is not None:
result["skipped"] += 1
continue
processed += 1
try:
payload = await _fetch_json_async(f"/events/{m.source_event_id}/stats/")
except Exception as e:
logger.warning("stats fetch failed match=%s event=%s: %s", m.id, m.source_event_id, e)
result["errors"].append(f"match {m.id}: {e}")
await asyncio.sleep(REQUEST_INTERVAL)
continue
result["fetched"] += 1
fields = _stats_from_payload(payload)
if not fields:
result["skipped"] += 1
await asyncio.sleep(REQUEST_INTERVAL)
continue
if m.stats is None:
# available_at 语义:完赛统计最早在开球+2h 可用(回测防泄漏)
available_at = m.match_date + timedelta(hours=2) if m.match_date else now
m.stats = MatchStats(
match_id=m.id,
source="bzzoiro",
source_record_id=str(m.source_event_id),
retrieved_at=now,
available_at=available_at,
)
db.add(m.stats)
result["created"] += 1
else:
result["updated"] += 1
if m.stats.source is None:
m.stats.source = "bzzoiro"
m.stats.source_record_id = str(m.source_event_id)
if m.stats.retrieved_at is None:
m.stats.retrieved_at = now
if m.stats.available_at is None and m.match_date:
m.stats.available_at = m.match_date + timedelta(hours=2)
for fld, v in fields.items():
# away_possession 为计算字段,模型无此列,跳过
if hasattr(m.stats, fld):
setattr(m.stats, fld, v)
await asyncio.sleep(REQUEST_INTERVAL)
logger.info(
"bzzoiro stats 回填完成: 抓取 %d, 新建 %d, 更新 %d, 跳过 %d, 错误 %d",
result["fetched"], result["created"], result["updated"],
result["skipped"], len(result["errors"]),
)
return result
src/data/config.py
+1 -10
View File
@@ -1,4 +1,4 @@
"""数据源配置常量(联赛映射)。"""
"""数据源配置常量(联赛映射)。数据源统一为 bzzoiro(单一数据源)。"""
from __future__ import annotations
# fdco 风格代码 → bzzoiro league_id
@@ -12,15 +12,6 @@ BZZOIRO_LEAGUE_IDS: dict[str, int] = {
"EL": 8, # Europa League
}
# fdco 代码 → understat 联赛代码
FDCO_TO_UNDERSTAT: dict[str, str] = {
"E0": "EPL",
"SP1": "La_liga",
"D1": "Bundesliga",
"I1": "Serie_A",
"F1": "Ligue_1",
}
# fdco 代码 → 显示名
LEAGUE_NAMES: dict[str, str] = {
"E0": "Premier League",
src/data/injuries.py
-347
View File
@@ -1,347 +0,0 @@
"""伤停数据采集器(api-football / api-sports.io)。
采集伤停数据并入库(injuries 表),供 injuries agent 使用。
"""
from __future__ import annotations
import asyncio
import json
import logging
import random
import tempfile
import time
from datetime import datetime, timezone
from pathlib import Path
from dataclasses import dataclass
from typing import Any
import httpx
from src.core.runtime_config import get_runtime_value
@dataclass
class InjuryQueryResult:
"""伤停查询结果(区分「查询成功但为空」与「查询失败/源未配置」)。"""
records: list["Injury"]
query_status: str # "success" | "source_not_configured" | "query_error"
@property
def has_data(self) -> bool:
"""成功查询(即使结果为空)视为有明确名单,has_data=True。"""
return self.query_status == "success"
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(tempfile.gettempdir()) / "profeto_injuries"
# Fix 5: 缓存 TTL 从 7 天改为 6 小时,同日再采不会命中旧数据
_CACHE_TTL_HOURS = 6
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 = await get_runtime_value("API_FOOTBALL_KEY")
if not api_key:
raise RuntimeError("API_FOOTBALL_KEY 未设置")
cache_dir = _CACHE_DIR
cache_dir.mkdir(parents=True, exist_ok=True)
# Fix 5: 缓存命中 (6 小时内有效)
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 < _CACHE_TTL_HOURS:
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 asyncio.wait_for(
client.get(
url, headers=headers, params=params,
timeout=httpx.Timeout(connect=10.0, read=30.0, write=10.0, pool=10.0),
),
timeout=60.0,
)
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 控制。
Fix 1: 使用 SAVEPOINT(begin_nested)避免整批回滚丢数据。
Fix 2: 正确解析并写入 return_date。
Fix 3: retrieved_at 比较统一用 timezone-aware datetime。
"""
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
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}
# 收集所有待插入记录(解析 + 校验)
pending_records: list[dict] = []
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)
# Fix 2: 解析日期(injury_date + return_date)
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
# 解析 return_date(如果数据源提供)
return_date = None
return_date_raw = player.get("return_date") or player.get("returnDate")
if return_date_raw:
try:
dt = datetime.fromisoformat(str(return_date_raw).replace("Z", "+00:00"))
return_date = dt.date()
except (ValueError, AttributeError):
pass
# 强制 int 转换,API 可能返回字符串
player_id = player.get("id")
try:
player_id = int(player_id) if player_id is not None else None
except (ValueError, TypeError):
player_id = None
fixture_id = fixture.get("id")
try:
fixture_id = int(fixture_id) if fixture_id is not None else None
except (ValueError, TypeError):
fixture_id = None
pending_records.append({
"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,
"return_date": return_date,
})
except Exception as e:
result["errors"].append(f"parse error: {e}")
# 批量查询已存在的记录(1 次 DB 往返)
existing_keys: set[tuple] = set()
if pending_records:
conditions = []
for rec in pending_records:
conditions.append(
(Injury.player_id == rec["player_id"])
& (Injury.fixture_id == rec["fixture_id"])
& (Injury.injury_type == rec["injury_type"])
)
if conditions:
from sqlalchemy import or_
stmt = select(Injury.player_id, Injury.fixture_id, Injury.injury_type).where(or_(*conditions))
rows = (await db.execute(stmt)).all()
existing_keys = {(r[0], r[1], r[2]) for r in rows}
# Fix 1: 使用 begin_nested(SAVEPOINT)隔离每批 flush
# IntegrityError 时只回滚到 savepoint,不影响其它已成功批次
BATCH_SIZE = 50
batch: list[Injury] = []
async def _flush_batch():
"""使用 savepoint flush 一批记录;失败只回滚本批。返回实际写入条数。"""
if not batch:
return 0
count = len(batch)
async with db.begin_nested():
for obj in batch:
db.add(obj)
await db.flush()
batch.clear()
return count
for i, rec in enumerate(pending_records):
key = (rec["player_id"], rec["fixture_id"], rec["injury_type"])
if key in existing_keys:
continue
batch.append(Injury(**rec))
# 每 BATCH_SIZE 条 flush 一次
if len(batch) >= BATCH_SIZE:
try:
result["inserted"] += await _flush_batch()
except IntegrityError:
logger.warning(
"injuries batch IntegrityError at record %d, "
"rolled back to savepoint, continuing",
i + 1,
)
# begin_nested 已回滚到 savepoint,清空 batch 继续
batch.clear()
continue
# 最终 flush(剩余不足一批的记录)
try:
result["inserted"] += await _flush_batch()
except IntegrityError:
logger.warning(
"injuries final flush IntegrityError, "
"rolled back to savepoint, some records may be lost",
)
batch.clear()
# 注意: 不在此处 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) -> InjuryQueryResult:
"""查询某场比赛前某队的伤停名单(比赛日仍缺阵的)。
Args:
db: 数据库 session
team_id: 球队 ID
match_date: 比赛日期
as_of: 数据截止时间(用于回测防泄漏)
Returns:
InjuryQueryResult:包含查询记录与状态
- query_status="success": 查询成功(即使结果也为空)
- query_status="source_not_configured": API_FOOTBALL_KEY 未配置
- query_status="query_error": 查询异常
- query_status="no_local_data": Key 已配置，但该队 injuries 表无任何历史记录
语义区分:
- success + 空结果 → has_data=True(明确知道「无人伤停」)
- no_local_data → has_data=False(本地尚未采集,需先 ingest)
- source_not_configured / query_error → has_data=False(无法判断)
"""
from sqlalchemy import select, func
from src.db.models import Injury
# 检查 API 是否配置(只读配置,不发网络)
api_key = await get_runtime_value("API_FOOTBALL_KEY")
if not api_key:
logger.debug("API_FOOTBALL_KEY 未配置,跳过伤停查询 team=%s", team_id)
return InjuryQueryResult(records=[], query_status="source_not_configured")
try:
# Fix 3: 统一用 timezone-aware datetime 比较,禁止 date() 截断
if hasattr(match_date, "date") and callable(match_date.date):
match_date = match_date.date()
stmt = (
select(Injury)
.where(Injury.team_id == team_id)
.where(Injury.injury_date <= match_date)
.where(
(Injury.return_date.is_(None)) | (Injury.return_date >= match_date)
)
)
if as_of is not None:
# Fix 3: 统一用 date 比较,避免 timestamptz vs date 的时区问题
if hasattr(as_of, "date") and callable(as_of.date):
as_of = as_of.date()
stmt = stmt.where(func.date(Injury.retrieved_at) <= as_of)
result = await db.execute(stmt)
records = list(result.scalars().all())
# 判定「无本地数据」:该队从未有伤停记录
# 规则:该 team_id 在 injuries 表中 count==0
if not records:
count_stmt = select(func.count()).where(Injury.team_id == team_id)
team_count = (await db.execute(count_stmt)).scalar_one() or 0
if team_count == 0:
logger.debug("API Key 已配置但本地无伤停数据 team=%s,标记 no_local_data", team_id)
return InjuryQueryResult(records=[], query_status="no_local_data")
return InjuryQueryResult(records=records, query_status="success")
except Exception as e:
logger.exception("伤停查询异常 team=%s: %s", team_id, e)
return InjuryQueryResult(records=[], query_status="query_error")
src/data/normalize.py
-32
View File
@@ -219,35 +219,3 @@ def normalize_bzzoiro(raw: dict, league_type: str) -> NormalizedMatch | None:
if m.match_status == "finished" and m.home_goals is None:
m.match_status = "scheduled"
return m
def normalize_understat(raw: dict, league_type: str) -> NormalizedMatch | None:
"""understat 单场 → NormalizedMatch(仅 xG)。"""
from src.data.team_names import normalize as normalize_name
dt_str = raw.get("datetime") or raw.get("date")
if not dt_str:
return None
dt = _parse_date(dt_str)
if dt is None:
return None
home_info = raw.get("h", {})
away_info = raw.get("a", {})
home_name = home_info.get("title", "") if isinstance(home_info, dict) else ""
away_name = away_info.get("title", "") if isinstance(away_info, dict) else ""
home = normalize_name(home_name)
away = normalize_name(away_name)
if not home or not away or home == away:
return None
home_xg = _to_float(raw["xG"].get("h")) if isinstance(raw.get("xG"), dict) else None
away_xg = _to_float(raw["xG"].get("a")) if isinstance(raw.get("xG"), dict) else None
return NormalizedMatch(
league_type=league_type,
date=dt,
home_team=home,
away_team=away,
match_status="finished",
season_label=derive_season_label(dt),
home_xg=home_xg,
away_xg=away_xg,
)
src/data/sources.py
+2 -2
View File
@@ -3,7 +3,8 @@
定义 DataSource 契约,并提供全局注册表供路由层分发。
每个比赛数据源实现该协议,注册后即可通过统一入口调度。
注: injuries 是球员级独立领域(写 Injury 表),不遵循此协议。
当前只有 bzzoiro 一个数据源(Understat / injuries 已移除),
保留协议与注册表是为了统一 ingest 调度入口的结构。
"""
from __future__ import annotations
@@ -58,7 +59,6 @@ def list_sources() -> list[str]:
def _load_sources() -> None:
"""延迟导入数据源触发 @register(避免循环导入)。"""
from src.data.bzzoiro import BzzoiroSource # noqa: F811
from src.data.understat import UnderstatSource # noqa: F811
# 保持向后兼容:模块加载时尝试加载(但不再强制)
src/data/understat.py
-223
View File
@@ -1,223 +0,0 @@
"""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
import httpx
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 asyncio.wait_for(
client.get(
url, headers=headers,
timeout=httpx.Timeout(connect=10.0, read=30.0, write=10.0, pool=10.0),
),
timeout=60.0,
)
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}")
# 优先按 JSON 响应解析(getLeagueData 接口返回 {teams, players, dates})
try:
data = resp.json()
except Exception:
data = None
if isinstance(data, dict) and isinstance(data.get("dates"), list):
return data["dates"]
# 兼容旧版联赛页面:内嵌 var datesData = JSON.parse('...')
text = resp.text
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)
# available_at 语义:统计「可被使用」的最早时间,至少不早于比赛结束
# 近似:开球 + 2 小时(实际完赛时间约为 +2 小时,非官方公布时间)
match_date = existing.match_date if existing.match_date else now
available_at = match_date + timedelta(hours=2)
existing.stats = MatchStats(
match_id=existing.id,
source="understat",
source_record_id=str(raw.get("id", "")),
retrieved_at=now,
available_at=available_at,
)
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