refactor: bzzoiro.py 按管线拆分为 5 个模块
单文件 852 行按职责拆分,保持 BzzoiroSource 与 get_source("bzzoiro") 行为不变:
- bzzoiro_common HTTP 抓取(多 key 轮换) + 字段转换原语
- bzzoiro_events fetch_bzzoiro_events + BzzoiroSource.ingest + Bronze 补写
- bzzoiro_standings standings 管线
- bzzoiro_stats stats 回填
- pipeline_write RawEvent/IngestFailure/DataLineage 写入助手
子模块运行期经聚合门面 src.data.bzzoiro 解析可替换协作者,
单文件时代的 bz.* monkeypatch 语义完全保留。
路由 import 已指向新模块(ingest.py / schedules.py)。
This commit is contained in:
shangfangjian
@@ -0,0 +1,180 @@
|
||||
"""bzzoiro stats 回填管线:已完赛比赛详细统计(/events/{id}/stats/)→ match_stats 表。
|
||||
|
||||
从 bzzoiro.py 拆出。上游限速(REQUEST_INTERVAL 秒/请求),大批量回填需分次触发;
|
||||
只 add/flush 不 commit,事务由调用方 UnitOfWork 控制。
|
||||
|
||||
可替换协作者(_fetch_json_async / Bronze 写入助手 / REQUEST_INTERVAL)在运行期
|
||||
经聚合门面 src.data.bzzoiro 解析 —— 与拆分前的单文件 monkeypatch 语义一致。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from collections.abc import Iterable
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from src.data.bzzoiro_common import _to_float_or_none, _to_int_or_none
|
||||
from src.data.config import BZZOIRO_LEAGUE_IDS
|
||||
from src.db.models import MatchStats
|
||||
from src.db.repositories import MatchRepository
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# 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
|
||||
|
||||
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
|
||||
|
||||
|
||||
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/请求,大批量需分次触发)。
|
||||
"""
|
||||
from src.data import bzzoiro as bz
|
||||
|
||||
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
|
||||
|
||||
# D4: 候选比赛查询经 MatchRepository(含 stats 预加载,筛选/排序/limit 语义不变)
|
||||
matches = await MatchRepository(db).find_finished_with_stats(
|
||||
league_ids, limit=limit * 3 if only_missing else limit
|
||||
)
|
||||
|
||||
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 bz._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 bz._safe_write_ingest_failure(
|
||||
db,
|
||||
entity_type="match_stats",
|
||||
source_record_id=str(m.source_event_id),
|
||||
error=e,
|
||||
raw_payload={"match_id": m.id},
|
||||
)
|
||||
await asyncio.sleep(bz.REQUEST_INTERVAL)
|
||||
continue
|
||||
|
||||
result["fetched"] += 1
|
||||
fields = _stats_from_payload(payload)
|
||||
if not fields:
|
||||
result["skipped"] += 1
|
||||
await asyncio.sleep(bz.REQUEST_INTERVAL)
|
||||
continue
|
||||
|
||||
if m.stats is None:
|
||||
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():
|
||||
if hasattr(m.stats, fld):
|
||||
setattr(m.stats, fld, v)
|
||||
|
||||
# 管线基础设施:写入 RawEvent + DataLineage
|
||||
batch_id = f"bzzoiro-stats-{m.source_event_id}-{now.strftime('%Y%m%d%H%M%S')}"
|
||||
try:
|
||||
await bz._write_raw_event(db, "bzzoiro", str(m.source_event_id), payload, batch_id)
|
||||
await bz._write_lineage(db, "bzzoiro", str(m.source_event_id), "match_stats", m.stats.id if m.stats else None, "stats_backfill", {"match_id": m.id}, batch_id)
|
||||
except Exception:
|
||||
pass # 基础设施写入失败不影响主流程
|
||||
|
||||
await asyncio.sleep(bz.REQUEST_INTERVAL)
|
||||
|
||||
logger.info(
|
||||
"bzzoiro stats 回填完成: 抓取 %d, 新建 %d, 更新 %d, 跳过 %d, 错误 %d",
|
||||
result["fetched"], result["created"], result["updated"],
|
||||
result["skipped"], len(result["errors"]),
|
||||
)
|
||||
return result
|
||||
Reference in New Issue
Block a user