shangfangjian
63caa6736c
新增 matches.score_status(known/missing/unknown): - 替换 ck_matches_finished_has_score 为 ck_matches_score_integrity: known → 必须有分; missing/unknown → goals 必须 NULL(不伪造 0:0) - normalize: 完赛缺分不再静默降级为 scheduled,改设 score_status=missing - events ingest: 创建/更新 Match 同步 score_status(比分由缺变 known / 确认缺分 missing) - slices(form/h2h/home_away)/backtest: 显式加 score_status='known' 过滤完赛样本 - 迁移 0022 回填现有数据(绝不 UPDATE goals=0) 测试 tests/test_p0_score_status.py(9/9):约束存在性/Match 构造/normalize 行为。 54 相关测试全绿。
328 lines
14 KiB
Python
328 lines
14 KiB
Python
"""bzzoiro events 管线:比赛日程/比分抓取(/events/)与入库(matches 表)。
|
|
|
|
从 bzzoiro.py 拆出。比赛主数据唯一入口;Team/League/Match 查找/创建经
|
|
Repository 层,事务由调用方 UnitOfWork 控制(分批事务约定不变)。
|
|
|
|
可替换协作者(抓取函数 / 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 _match_key, _to_date, _to_int_or_none
|
|
from src.data.config import BZZOIRO_LEAGUE_IDS, LEAGUE_COUNTRIES, LEAGUE_NAMES
|
|
from src.data.normalize import normalize_bzzoiro
|
|
from src.data.pipeline_write import _safe_write_ingest_failure, _write_lineage, _write_raw_event
|
|
from src.data.sources import register
|
|
from src.data.team_names_zh import zh_name
|
|
from src.db.models import Match
|
|
from src.db.repositories import LeagueRepository, MatchRepository, TeamRepository
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def fetch_bzzoiro_events(
|
|
league_code: str,
|
|
*,
|
|
status: str = "finished",
|
|
date_from: str | None = None,
|
|
date_to: str | None = None,
|
|
limit: int = 200,
|
|
) -> list[dict]:
|
|
"""抓取 bzzoiro 原始事件(纯异步,无需 run_in_executor)。"""
|
|
from src.data import bzzoiro as bz
|
|
|
|
league_id = BZZOIRO_LEAGUE_IDS.get(league_code)
|
|
if league_id is None:
|
|
raise ValueError(f"未知联赛代码: {league_code}")
|
|
|
|
rows: list[dict] = []
|
|
offset = 0
|
|
while True:
|
|
params: dict = {
|
|
"league_id": league_id,
|
|
"status": status,
|
|
"limit": limit,
|
|
"offset": offset,
|
|
}
|
|
if date_from:
|
|
params["date_from"] = str(date_from)[:10]
|
|
if date_to:
|
|
params["date_to"] = str(date_to)[:10]
|
|
payload = await bz._fetch_json_async("/events/", params)
|
|
batch = payload.get("results") or []
|
|
if not batch:
|
|
break
|
|
rows.extend(batch)
|
|
total = payload.get("total")
|
|
offset += limit
|
|
if total is not None and offset >= total:
|
|
break
|
|
if len(batch) < limit:
|
|
break
|
|
await asyncio.sleep(bz.REQUEST_INTERVAL)
|
|
return rows
|
|
|
|
|
|
@register
|
|
class BzzoiroSource:
|
|
"""bzzoiro 数据源(实现 DataSource 协议)。"""
|
|
|
|
name = "bzzoiro"
|
|
|
|
async def ingest(
|
|
self,
|
|
db,
|
|
*,
|
|
leagues: Iterable[str],
|
|
date_from: str | None = None,
|
|
date_to: str | None = None,
|
|
status: str = "finished",
|
|
) -> dict:
|
|
"""采集 bzzoiro → 入库。返回统计。
|
|
|
|
注意: 本方法不控制事务(commit/rollback),由调用方通过 UnitOfWork 控制。
|
|
"""
|
|
from src.data import bzzoiro as bz
|
|
|
|
result: dict = {"leagues": {}, "total_inserted": 0, "total_updated": 0, "errors": []}
|
|
|
|
for code in leagues:
|
|
league_r: dict = {"inserted": 0, "updated": 0, "errors": []}
|
|
try:
|
|
raw_events = await bz.fetch_bzzoiro_events(code, status=status, date_from=date_from, date_to=date_to)
|
|
except Exception as e:
|
|
# 单联赛抓取失败隔离:记录错误后继续其余联赛,不拖垮整批
|
|
logger.exception("bzzoiro fetch failed for %s", code)
|
|
league_r["errors"].append(f"fetch failed: {e}")
|
|
await _safe_write_ingest_failure(
|
|
db,
|
|
entity_type="events",
|
|
source_record_id=None,
|
|
error=e,
|
|
raw_payload={"league": code, "status": status, "date_from": date_from, "date_to": date_to},
|
|
)
|
|
result["leagues"][code] = league_r
|
|
continue
|
|
|
|
# D4: 联赛查找/创建经 LeagueRepository(事务仍由调用方 UoW 提交)
|
|
league = await LeagueRepository(db).get_or_create(
|
|
code, LEAGUE_NAMES.get(code, code), LEAGUE_COUNTRIES.get(code)
|
|
)
|
|
team_r = TeamRepository(db)
|
|
match_r = MatchRepository(db)
|
|
|
|
# === 批量优化: 预加载球队和已有比赛到内存 ===
|
|
team_name_to_id: dict[str, int] = {}
|
|
existing_matches: dict[tuple[int, int, str], Match] = {} # 完整对象,避免重复查询
|
|
# (NormalizedMatch, 原始 event) 成对保存:后续写 source_event_id 时
|
|
# 必须用配对的那条 event,不能依赖外层循环变量残留值。
|
|
normalized_matches: list[tuple] = []
|
|
|
|
if raw_events:
|
|
# 一次遍历: 收集球队名 + 规范化
|
|
all_team_names = set()
|
|
for raw in raw_events:
|
|
nm = normalize_bzzoiro(raw, code)
|
|
if nm is not None:
|
|
try:
|
|
nm.validate()
|
|
except Exception:
|
|
continue
|
|
normalized_matches.append((nm, raw))
|
|
all_team_names.add(nm.home_team)
|
|
all_team_names.add(nm.away_team)
|
|
|
|
if all_team_names:
|
|
team_name_to_id = {
|
|
name: t.id
|
|
for name, t in (await team_r.get_all_by_names(list(all_team_names))).items()
|
|
}
|
|
|
|
# P1-2: 按需加载,只加载 raw_events 涉及日期范围的比赛(加 30 天缓冲)
|
|
# 避免加载联赛全部历史比赛到内存(多赛季采集时内存溢出)
|
|
if normalized_matches:
|
|
# normalized_matches 存的是 (nm, raw) 元组,遍历需解包
|
|
dates = [nm.date for nm, _raw 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)
|
|
matches_in_range = await match_r.find_by_league_and_date_range(
|
|
league.id, min_dt, max_dt
|
|
)
|
|
existing_matches = {
|
|
_match_key(m.home_team_id, m.away_team_id, m.match_date_date): m
|
|
for m in matches_in_range
|
|
}
|
|
# else: existing_matches 保持空 dict(全量新比赛)
|
|
|
|
# D1: Bronze 层批次信息(每联赛每批次一个 batch_id;seen 防同批重复写入)
|
|
now = datetime.now(timezone.utc)
|
|
bronze_batch_id = f"bzzoiro-events-{code}-{now:%Y%m%d%H%M%S}"
|
|
bronze_written: set[str] = set()
|
|
|
|
for nm, raw in normalized_matches:
|
|
# D1: RawEvent 幂等键(上游 id 或合成键),插入/变更更新共用
|
|
record_id = _events_record_id(code, nm, raw)
|
|
|
|
# 球队: 内存查找 + 按需创建(D4: 经 TeamRepository)
|
|
home_team_id = team_name_to_id.get(nm.home_team)
|
|
if home_team_id is None:
|
|
home = await team_r.get_or_create(nm.home_team, name_zh=zh_name(nm.home_team))
|
|
home_team_id = home.id
|
|
team_name_to_id[nm.home_team] = home_team_id
|
|
|
|
away_team_id = team_name_to_id.get(nm.away_team)
|
|
if away_team_id is None:
|
|
away = await team_r.get_or_create(nm.away_team, name_zh=zh_name(nm.away_team))
|
|
away_team_id = away.id
|
|
team_name_to_id[nm.away_team] = away_team_id
|
|
|
|
# 查找已有比赛:优先按 upstream event_id 定位(命中即唯一),
|
|
# 否则回退自然键(联赛+主客+天级日期)内存查找。
|
|
# source_event_id 上有 partial unique 索引保障 upstream 唯一。
|
|
eid = _to_int_or_none(raw.get("id"))
|
|
existing_match = None
|
|
if eid is not None:
|
|
existing_match = await match_r.find_by_source_event_id(eid)
|
|
if existing_match is None:
|
|
match_key = _match_key(home_team_id, away_team_id, nm.date)
|
|
existing_match = existing_matches.get(match_key)
|
|
|
|
if existing_match is None:
|
|
m = Match(
|
|
league_id=league.id,
|
|
season=nm.season_label or None,
|
|
home_team_id=home_team_id,
|
|
away_team_id=away_team_id,
|
|
match_date=nm.date,
|
|
match_date_date=_to_date(nm.date),
|
|
match_status=nm.match_status,
|
|
score_status=nm.score_status,
|
|
home_goals=nm.home_goals,
|
|
away_goals=nm.away_goals,
|
|
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 # 防止同批重复
|
|
# 统计字段不在 /events/ 载荷中(单独由 stats 管线回填),
|
|
# 此处不再创建 MatchStats。
|
|
league_r["inserted"] += 1
|
|
# D1: 成功插入 → 补写 Bronze 层(原始载荷 + 血缘)
|
|
if record_id not in bronze_written:
|
|
bronze_written.add(record_id)
|
|
await _write_events_bronze(
|
|
db,
|
|
source_record_id=record_id,
|
|
raw_payload=raw,
|
|
target_match_id=m.id,
|
|
league_code=code,
|
|
match_status=nm.match_status,
|
|
batch_id=bronze_batch_id,
|
|
)
|
|
else:
|
|
# 已有比赛: 直接从内存获取对象更新(无需再查询)
|
|
changed = False
|
|
if existing_match.match_status != nm.match_status and nm.match_status == "finished":
|
|
existing_match.match_status = nm.match_status
|
|
changed = True
|
|
if existing_match.home_goals is None and nm.home_goals is not None:
|
|
existing_match.home_goals = nm.home_goals
|
|
existing_match.away_goals = nm.away_goals
|
|
existing_match.home_ht_goals = nm.home_ht_goals
|
|
existing_match.away_ht_goals = nm.away_ht_goals
|
|
# 比分由缺变有 → 标记 known
|
|
existing_match.score_status = "known"
|
|
changed = True
|
|
elif (
|
|
nm.match_status == "finished"
|
|
and nm.home_goals is None
|
|
and existing_match.score_status == "unknown"
|
|
):
|
|
# 确认完赛仍缺分 → 标记 missing(不伪造 0:0)
|
|
existing_match.score_status = "missing"
|
|
changed = True
|
|
if existing_match.match_stage is None and nm.match_stage:
|
|
existing_match.match_stage = nm.match_stage
|
|
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
|
|
# D1: 变更更新 → 补写血缘(RawEvent 幂等键不变,重复采集自动跳过)
|
|
if record_id not in bronze_written:
|
|
bronze_written.add(record_id)
|
|
await _write_events_bronze(
|
|
db,
|
|
source_record_id=record_id,
|
|
raw_payload=raw,
|
|
target_match_id=existing_match.id,
|
|
league_code=code,
|
|
match_status=nm.match_status,
|
|
batch_id=bronze_batch_id,
|
|
)
|
|
|
|
# 注意: 不在此处 commit,由调用方 UnitOfWork 控制事务
|
|
result["leagues"][code] = league_r
|
|
result["total_inserted"] += league_r["inserted"]
|
|
result["total_updated"] += league_r["updated"]
|
|
return result
|
|
|
|
|
|
def _events_record_id(league_code: str, nm, raw: dict) -> str:
|
|
"""events 载荷的 RawEvent 幂等键。
|
|
|
|
优先用上游 event id;缺失时用 (league:home:away:date) 合成稳定键 ——
|
|
取 normalize 后的队名与天级日期(与 _match_key 同口径),不依赖 DB 自增 id,
|
|
保证同一来源比赛重复采集时命中同一条 RawEvent,不产生重复原始载荷。
|
|
"""
|
|
eid = _to_int_or_none(raw.get("id"))
|
|
if eid is not None:
|
|
return str(eid)
|
|
d = _to_date(nm.date)
|
|
date_part = d.isoformat() if d is not None else "na"
|
|
return f"{league_code}:{nm.home_team}:{nm.away_team}:{date_part}"
|
|
|
|
|
|
async def _write_events_bronze(
|
|
db,
|
|
*,
|
|
source_record_id: str,
|
|
raw_payload: dict,
|
|
target_match_id: int | None,
|
|
league_code: str,
|
|
match_status: str | None,
|
|
batch_id: str,
|
|
) -> None:
|
|
"""events 成功插入/更新单场比赛后的 Bronze 层补写:RawEvent(幂等) + DataLineage。
|
|
|
|
D1(工程债):此前只有 stats 回填写 RawEvent/Lineage,events 管线作为比赛
|
|
主数据的唯一入口反而不留溯源记录。幂等性由 _write_raw_event 的
|
|
source_record_id 查重保证;best-effort:基础设施写入失败只记 warning,
|
|
绝不拖垮采集主流程(与 _safe_write_ingest_failure 同级约束)。
|
|
"""
|
|
try:
|
|
await _write_raw_event(db, "bzzoiro", source_record_id, raw_payload, batch_id)
|
|
await _write_lineage(
|
|
db, "bzzoiro", source_record_id,
|
|
"matches", target_match_id, "events_ingest",
|
|
{"league": league_code, "match_status": match_status},
|
|
batch_id,
|
|
)
|
|
except Exception:
|
|
logger.warning(
|
|
"events Bronze 写入失败(record=%s, match=%s),不影响采集主流程",
|
|
source_record_id, target_match_id, exc_info=True,
|
|
)
|