shangfangjian
ff0045ad93
P1-1 [context_builder] build_context 共享 session,切片函数传 db 参数,
回测 20 场并发连接需求从 100+ 降至每场 1 个
P1-2 [bzzoiro] 预加载改为按 raw_events 日期范围 ±30 天按需加载
P1-3 [understat] 批量查询球队 + 比赛,从 1140 次往返降到 3 次
P1-4 [injuries] 批量幂等检查 + 分批 flush,IntegrityError 逐条回退
P1-5 [predict] 删除 threading.Lock,dict 操作原子无需同步锁
P1-6 [models] 添加 (match_id, provider, model) 唯一约束 + 迁移
P2-1 [unit_of_work] get_uow 返回类型改为 AsyncIterator[AsyncSession]
P2-2 [normalize] _parse_date 失败时记录 warning 避免静默丢数据
P2-3 [repositories] find_by_teams_and_date 改用 match_date_date 等值匹配
P2-4 [migration] 幽灵列 cutoff_at 已在 0006 迁移删除(已有)
P2-5 [migration] injuries 约束命名对齐 ORM,UniqueConstraint → 唯一索引
286 lines
9.5 KiB
Python
286 lines
9.5 KiB
Python
"""伤停数据采集器(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 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"
|
|
|
|
# P2-3: 缓存目录改用系统临时目录,避免源码树内写入
|
|
_CACHE_DIR = Path(tempfile.gettempdir()) / "profeto_injuries"
|
|
|
|
|
|
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 控制。
|
|
|
|
P1-4: 批量幂等检查,避免逐条查询的竞态条件(并发采集时 IntegrityError)。
|
|
"""
|
|
from sqlalchemy import select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
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}
|
|
|
|
# P1-4: 收集所有待插入记录的键,批量查询已存在的记录
|
|
# 避免逐条查询 + 插入的竞态条件(两个并发请求同时通过检查 → IntegrityError)
|
|
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)
|
|
|
|
# 解析日期
|
|
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
|
|
|
|
# 强制 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,
|
|
})
|
|
except Exception as e:
|
|
result["errors"].append(f"parse error: {e}")
|
|
|
|
# P1-4: 批量查询已存在的记录(1 次 DB 往返)
|
|
existing_keys: set[tuple] = set()
|
|
if pending_records:
|
|
# 构造查询条件:所有 (player_id, fixture_id, injury_type) 组合
|
|
# 使用 OR 条件批量查询
|
|
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}
|
|
|
|
# P1-4: 批量插入(跳过已存在的)
|
|
for rec in pending_records:
|
|
key = (rec["player_id"], rec["fixture_id"], rec["injury_type"])
|
|
if key in existing_keys:
|
|
continue
|
|
|
|
injury = Injury(**rec)
|
|
db.add(injury)
|
|
result["inserted"] += 1
|
|
|
|
# 每 50 条 flush 一次,减少内存压力,同时捕获 IntegrityError
|
|
if result["inserted"] % 50 == 0:
|
|
try:
|
|
await db.flush()
|
|
except IntegrityError:
|
|
# P1-4: 并发采集时可能仍有竞态,回退到逐条插入
|
|
await db.rollback()
|
|
logger.warning("injuries batch IntegrityError, falling back to per-record insert")
|
|
return await _ingest_injuries_fallback(db, pending_records, result)
|
|
|
|
# 最终 flush
|
|
try:
|
|
await db.flush()
|
|
except IntegrityError:
|
|
await db.rollback()
|
|
logger.warning("injuries final flush IntegrityError, falling back to per-record insert")
|
|
return await _ingest_injuries_fallback(db, pending_records, result)
|
|
|
|
# 注意: 不在此处 commit,由调用方 UnitOfWork 控制事务
|
|
logger.info("injuries: fetched %d, inserted %d for %s", result["count"], result["inserted"], date)
|
|
return result
|
|
|
|
|
|
async def _ingest_injuries_fallback(db, pending_records: list[dict], result: dict) -> dict:
|
|
"""P1-4: 逐条插入回退,捕获每条 IntegrityError 避免整批回滚。"""
|
|
from sqlalchemy.exc import IntegrityError
|
|
from src.db.models import Injury
|
|
|
|
inserted = 0
|
|
for rec in pending_records:
|
|
injury = Injury(**rec)
|
|
db.add(injury)
|
|
try:
|
|
await db.flush()
|
|
inserted += 1
|
|
except IntegrityError:
|
|
await db.rollback()
|
|
# 已存在或其他冲突,跳过
|
|
continue
|
|
|
|
result["inserted"] = inserted
|
|
logger.info("injuries fallback: inserted %d records", inserted)
|
|
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: 数据截止时间(用于回测防泄漏)
|
|
|
|
Returns:
|
|
伤停记录列表
|
|
"""
|
|
from sqlalchemy import select
|
|
from src.db.models import Injury
|
|
|
|
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:
|
|
if hasattr(as_of, "date") and callable(as_of.date):
|
|
as_of = as_of.date()
|
|
stmt = stmt.where(Injury.retrieved_at <= as_of)
|
|
|
|
result = await db.execute(stmt)
|
|
return list(result.scalars().all())
|