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 表
 # ============================================================

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