diff --git a/src/data/bzzoiro.py b/src/data/bzzoiro.py index 29db35d..cfa86b0 100644 --- a/src/data/bzzoiro.py +++ b/src/data/bzzoiro.py @@ -266,7 +266,15 @@ class BzzoiroSource: } # 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) + # 球队: 内存查找 + 按需创建 home_team_id = team_name_to_id.get(nm.home_team) if home_team_id is None: @@ -310,6 +318,18 @@ class BzzoiroSource: # 统计字段不在 /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 @@ -332,6 +352,18 @@ class BzzoiroSource: 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 @@ -412,6 +444,53 @@ async def _write_lineage(db, source_system: str, source_record_id: str, target_t )) +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, + ) + + # ============================================================ # 积分榜管线:/leagues/{id}/standings/ → standings 表 # ============================================================ diff --git a/tests/test_events_bronze.py b/tests/test_events_bronze.py new file mode 100644 index 0000000..0d62990 --- /dev/null +++ b/tests/test_events_bronze.py @@ -0,0 +1,259 @@ +"""D1 工程债回归测试: events 成功路径必须写 Bronze 层(RawEvent + DataLineage)。 + +背景: stats 回填管线早已有 RawEvent/DataLineage 写入,但 events 管线(比赛 +主数据的唯一入口)成功插入/更新后既不留原始载荷,也不留血缘 —— 数据溯源 +链条在最关键的一环断掉。本测试守护: + 1. 插入新比赛 → RawEvent(幂等键=source_event_id 或合成键) + Lineage + (target_table="matches", transform_name="events_ingest") + 2. 变更更新(如补比分/状态) → 同样写血缘 + 3. 无变化跳过 → 不写(避免 lineage 刷屏) + 4. RawEvent 幂等: 同 source_record_id 已存在则跳过 + 5. 基础设施写入失败 → 只 warning,不拖垮采集主流程 + +范式: 假 db(按查询实体分发预置数据 + 记录 add,flush 分配自增 id) ++ monkeypatch 抓取函数,不依赖真实数据库。 +""" +from __future__ import annotations + +from datetime import date, datetime, timezone + +import pytest + +import src.data.bzzoiro as bz +from src.db.models import DataLineage, League, Match, RawEvent, Team + + +def _event(eid=1001, status="finished", home="Arsenal", away="Chelsea", hs=2, as_=1): + """构造一条最小合法的 bzzoiro /events/ 原始载荷。""" + raw = { + "event_date": "2026-09-20 15:00:00", + "status": status, + "home_team": home, + "away_team": away, + "home_score": hs, + "away_score": as_, + } + if eid is not None: + raw["id"] = eid + return raw + + +class _FakeResult: + """支持 .scalars().all() / .scalar_one_or_none() 的最小假结果集。""" + + def __init__(self, items): + self._items = list(items) + + def scalars(self): + return self + + def __iter__(self): + return iter(self._items) + + def all(self): + return self._items + + def scalar_one_or_none(self): + return self._items[0] if self._items else None + + def scalar(self): + return None + + +class _FakeDB: + """按查询实体分发预置数据;记录 add();flush 为无 id 对象分配自增主键。""" + + def __init__(self, matches=(), teams=(), leagues=(), raw_events=()): + self.added = [] + self._by_entity = { + Match: list(matches), + Team: list(teams), + League: list(leagues), + RawEvent: list(raw_events), + } + self._next_id = 0 + + def add(self, obj): + self.added.append(obj) + + async def execute(self, stmt): + entities = set() + for d in (stmt.column_descriptions or []): + entities.add(d.get("entity") or d.get("type")) + for entity, items in self._by_entity.items(): + if entity in entities: + return _FakeResult(items) + return _FakeResult([]) + + async def flush(self): + for obj in self.added: + if getattr(obj, "id", None) is None: + self._next_id += 1 + obj.id = self._next_id + + +@pytest.fixture(autouse=True) +def _no_request_interval(monkeypatch): + monkeypatch.setattr(bz, "REQUEST_INTERVAL", 0) + + +def _patch_fetch(monkeypatch, events): + async def _fetch(league_code, **kwargs): + return list(events) + + monkeypatch.setattr(bz, "fetch_bzzoiro_events", _fetch) + + +def _matches(db): + return [o for o in db.added if isinstance(o, Match)] + + +def _raw_events(db): + return [o for o in db.added if isinstance(o, RawEvent)] + + +def _lineages(db): + return [o for o in db.added if isinstance(o, DataLineage)] + + +# ============================================================ +# 1. 插入新比赛 → RawEvent + DataLineage +# ============================================================ + + +class TestEventsBronzeOnInsert: + async def test_insert_writes_raw_event_and_lineage(self, monkeypatch): + _patch_fetch(monkeypatch, [_event()]) + db = _FakeDB() + + result = await bz.BzzoiroSource().ingest(db, leagues=["E0"]) + + assert result["total_inserted"] == 1 + + raws = _raw_events(db) + assert len(raws) == 1 + raw = raws[0] + assert raw.source_system == "bzzoiro" + assert raw.source_record_id == "1001" # 有上游 id 时直接用 + assert raw.ingest_batch_id.startswith("bzzoiro-events-E0-") + assert raw.raw_payload["id"] == 1001 # 原始载荷完整保留 + + lineages = _lineages(db) + assert len(lineages) == 1 + lin = lineages[0] + assert lin.source_system == "bzzoiro" + assert lin.source_record_id == "1001" + assert lin.target_table == "matches" + assert lin.target_id == _matches(db)[0].id + assert lin.transform_name == "events_ingest" + # RawEvent 与 Lineage 同批次,便于按批追溯 + assert lin.batch_id == raw.ingest_batch_id + + async def test_missing_source_id_uses_synthetic_stable_key(self, monkeypatch): + """上游 id 缺失时,用 (league:home:away:date) 合成稳定幂等键。""" + _patch_fetch(monkeypatch, [_event(eid=None)]) + db = _FakeDB() + + result = await bz.BzzoiroSource().ingest(db, leagues=["E0"]) + + assert result["total_inserted"] == 1 + raws = _raw_events(db) + assert len(raws) == 1 + # 期望键基于 normalize 后的队名与天级日期 —— 与 _match_key 同口径, + # 不依赖 DB 自增 id,跨批次可复现 + nm = bz.normalize_bzzoiro(_event(eid=None), "E0") + expected = f"E0:{nm.home_team}:{nm.away_team}:{nm.date.date().isoformat()}" + assert raws[0].source_record_id == expected + + async def test_existing_raw_event_is_skipped(self, monkeypatch): + """RawEvent 幂等: 同 source_record_id 已存在则不再新增,但血缘照写。""" + existing = RawEvent( + source_system="bzzoiro", + source_record_id="1001", + raw_payload={"old": True}, + ) + _patch_fetch(monkeypatch, [_event()]) + db = _FakeDB(raw_events=[existing]) + + await bz.BzzoiroSource().ingest(db, leagues=["E0"]) + + new_raws = [r for r in _raw_events(db) if r is not existing] + assert new_raws == [] + assert len(_lineages(db)) == 1 # 血缘仍然记录本次采集 + + +# ============================================================ +# 2. 变更更新 → 写血缘;无变化 → 不写 +# ============================================================ + + +class TestEventsBronzeOnUpdate: + def _existing_match(self, **overrides): + m = Match( + league_id=1, + home_team_id=2, # 与本轮 Team 创建后 fake 自增 id 对齐(league=1, home=2, away=3) + away_team_id=3, + match_date=datetime(2026, 9, 20, 15, 0, tzinfo=timezone.utc), + match_date_date=date(2026, 9, 20), + match_status="scheduled", + source_event_id=1001, + ) + m.id = 42 + for k, v in overrides.items(): + setattr(m, k, v) + return m + + async def test_changed_update_writes_lineage(self, monkeypatch): + # 已有比赛处于 scheduled 且无比分;新载荷为 finished 2:1 → 触发变更更新 + db = _FakeDB(matches=[self._existing_match()]) + _patch_fetch(monkeypatch, [_event()]) + + result = await bz.BzzoiroSource().ingest(db, leagues=["E0"]) + + assert result["total_inserted"] == 0 + assert result["leagues"]["E0"]["updated"] == 1 + + lineages = _lineages(db) + assert len(lineages) == 1 + assert lineages[0].target_id == 42 + assert lineages[0].target_table == "matches" + assert lineages[0].transform_name == "events_ingest" + + async def test_unchanged_match_writes_nothing(self, monkeypatch): + # 已有比赛与新载荷完全一致 → 无变化,不应产生 RawEvent/Lineage + existing = self._existing_match( + match_status="finished", + home_goals=2, + away_goals=1, + ) + db = _FakeDB(matches=[existing]) + _patch_fetch(monkeypatch, [_event()]) + + result = await bz.BzzoiroSource().ingest(db, leagues=["E0"]) + + assert result["total_inserted"] == 0 + assert result["leagues"]["E0"]["updated"] == 0 + assert _raw_events(db) == [] + assert _lineages(db) == [] + + +# ============================================================ +# 3. 基础设施写入失败: 尽力而为,不拖垮主流程 +# ============================================================ + + +class TestEventsBronzeIsBestEffort: + async def test_bronze_write_failure_does_not_break_ingest(self, monkeypatch): + async def _boom(*args, **kwargs): + raise RuntimeError("infra down") + + monkeypatch.setattr(bz, "_write_raw_event", _boom) + monkeypatch.setattr(bz, "_write_lineage", _boom) + _patch_fetch(monkeypatch, [_event()]) + db = _FakeDB() + + # 不应抛异常:Bronze 写不进去只记 warning + result = await bz.BzzoiroSource().ingest(db, leagues=["E0"]) + + assert result["total_inserted"] == 1 + assert len(_matches(db)) == 1