fix(pipeline): 死信表真正接线 + 前端 HTTP 收敛与状态修正

- bzzoiro events/standings/stats 抓取失败写入 IngestFailure 死信(尽力而为, 写入失败不影响主流程);顺带修复 standings 失败路径 league_r 缺 errors 键的 KeyError —— 该路径此前从未跑通,一旦失败会顶掉原始异常 - admin/api.ts 收敛为 lib/http.ts 薄门面,消除第二套 HTTP 实现; UNAUTHORIZED_EVENT 定义移至共享层,断开 lib→admin 反向依赖 - STATUS_META 死键 live 改为 in_play(对齐 normalize.py 口径), 补 paused/postponed/cancelled/suspended;移除无人消费的 DashboardStats.total_matches(items.length 近似,上限 100) - 新增 tests/test_ingest_deadletter.py(6 例,变异验证判别力)
2026-09-21 18:36:43 +08:00
parent b5f7e682f0
commit b6a251407d
8 changed files with 297 additions and 100 deletions
@@ -0,0 +1,209 @@
+"""死信接线回归测试: bzzoiro 三条管线抓取失败必须写入 IngestFailure。
+
+背景: IngestFailure 死信表此前「预留未启用」——events / standings / stats
+抓取失败只打日志 + errors 列表,admin 的 /ingest-failures 列表与 retry
+端点形同虚设。本测试守护三条管线的失败写入路径:
+  1. events   整联赛抓取失败  → entity_type="events"
+  2. standings 整联赛抓取失败 → entity_type="standings"
+  3. stats    单场统计抓取失败 → entity_type="match_stats"(带 source_record_id)
+
+范式: 假 db(记录 add 调用) + monkeypatch 抓取函数,不依赖真实数据库 ——
+与 test_review_required_fixes.py R2 相同。失败写入是「尽力而为」:
+写入器自身抛错只记日志,不得拖垮采集主流程(最后一个测试守护)。
+"""
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+
+import src.data.bzzoiro as bz
+from src.db.models import IngestFailure
+
+
+class _FakeResult:
+    """支持 .scalars().all() / .scalar_one_or_none() / .scalar() 的最小假结果集。"""
+
+    def __init__(self, items):
+        self._items = items
+
+    def scalars(self):
+        return self
+
+    def all(self):
+        return self._items
+
+    def scalar_one_or_none(self):
+        return None
+
+    def scalar(self):
+        return None
+
+
+class _FakeDB:
+    """只记录 add() 的假会话 —— 失败路径不触发真实查询。"""
+
+    def __init__(self, items=None):
+        self.added = []
+        self._items = items or []
+
+    def add(self, obj):
+        self.added.append(obj)
+
+    async def execute(self, stmt):
+        return _FakeResult(self._items)
+
+    async def flush(self):
+        pass
+
+
+@pytest.fixture(autouse=True)
+def _no_request_interval(monkeypatch):
+    """失败路径会 await asyncio.sleep(REQUEST_INTERVAL),置 0 加速测试。"""
+    monkeypatch.setattr(bz, "REQUEST_INTERVAL", 0)
+
+
+def _failures(db: _FakeDB) -> list[IngestFailure]:
+    return [o for o in db.added if isinstance(o, IngestFailure)]
+
+
+# ============================================================
+# 1. events: 整联赛抓取失败
+# ============================================================
+
+
+class TestEventsFetchFailureDeadLetter:
+    async def test_writes_deadletter_with_league_context(self, monkeypatch):
+        async def _boom(*args, **kwargs):
+            raise RuntimeError("network down")
+
+        monkeypatch.setattr(bz, "fetch_bzzoiro_events", _boom)
+        db = _FakeDB()
+
+        result = await bz.BzzoiroSource().ingest(db, leagues=["E0"])
+
+        rows = _failures(db)
+        assert len(rows) == 1
+        row = rows[0]
+        assert row.source_system == "bzzoiro"
+        assert row.entity_type == "events"
+        assert row.error_type == "fetch_error"
+        assert "network down" in row.error_detail
+        # 上下文足够管理员定位:联赛代码必须随行
+        assert row.raw_payload is not None
+        assert row.raw_payload.get("league") == "E0"
+        # 主流程不受影响:错误仍记录在 result 中
+        assert result["leagues"]["E0"]["errors"]
+        assert result["total_inserted"] == 0
+
+    async def test_other_leagues_continue_after_failure(self, monkeypatch):
+        async def _boom(league_code, **kwargs):
+            if league_code == "E0":
+                raise RuntimeError("boom")
+            return []
+
+        monkeypatch.setattr(bz, "fetch_bzzoiro_events", _boom)
+        db = _FakeDB()
+
+        await bz.BzzoiroSource().ingest(db, leagues=["E0", "SP1"])
+
+        rows = _failures(db)
+        assert len(rows) == 1
+        assert rows[0].raw_payload.get("league") == "E0"
+
+
+# ============================================================
+# 2. standings: 整联赛抓取失败
+# ============================================================
+
+
+class TestStandingsFetchFailureDeadLetter:
+    async def test_writes_deadletter_with_season_context(self, monkeypatch):
+        async def _boom(league_code, season=None):
+            raise RuntimeError("upstream 500")
+
+        monkeypatch.setattr(bz, "fetch_bzzoiro_standings", _boom)
+        db = _FakeDB()
+
+        result = await bz.ingest_bzzoiro_standings(db, leagues=["SP1"], season="2025-2026")
+
+        rows = _failures(db)
+        assert len(rows) == 1
+        row = rows[0]
+        assert row.entity_type == "standings"
+        assert row.error_type == "fetch_error"
+        assert "upstream 500" in row.error_detail
+        assert row.raw_payload == {"league": "SP1", "season": "2025-2026"}
+        assert result["errors"]
+
+
+# ============================================================
+# 3. stats: 单场统计抓取失败
+# ============================================================
+
+
+class TestStatsFetchFailureDeadLetter:
+    def _match(self) -> SimpleNamespace:
+        return SimpleNamespace(
+            id=42,
+            source_event_id=777,
+            stats=None,
+            match_date=None,
+            league_id=1,
+            match_status="finished",
+        )
+
+    async def test_writes_deadletter_with_source_record_id(self, monkeypatch):
+        async def _boom(path, params=None, max_retries=3):
+            raise TimeoutError("read timeout")
+
+        monkeypatch.setattr(bz, "_fetch_json_async", _boom)
+        db = _FakeDB(items=[self._match()])
+
+        result = await bz.ingest_bzzoiro_event_stats(db, leagues=["E0"], limit=1)
+
+        rows = _failures(db)
+        assert len(rows) == 1
+        row = rows[0]
+        assert row.entity_type == "match_stats"
+        # source_record_id 必须是上游 event id,retry 端点据此定位
+        assert row.source_record_id == "777"
+        assert "timeout" in (row.error_detail or "").lower()
+        assert row.raw_payload == {"match_id": 42}
+        assert result["errors"]
+
+    async def test_success_path_does_not_write_deadletter(self, monkeypatch):
+        async def _ok(path, params=None, max_retries=3):
+            return {"stats": {"home": {"total_shots": 10}, "away": {"total_shots": 5}}}
+
+        monkeypatch.setattr(bz, "_fetch_json_async", _ok)
+        db = _FakeDB(items=[self._match()])
+
+        await bz.ingest_bzzoiro_event_stats(db, leagues=["E0"], limit=1)
+
+        assert _failures(db) == []
+
+
+# ============================================================
+# 4. 死信写入自身失败: 尽力而为,不拖垮主流程
+# ============================================================
+
+
+class TestDeadLetterWriteIsBestEffort:
+    async def test_db_add_failure_does_not_break_ingest(self, monkeypatch):
+        async def _boom(*args, **kwargs):
+            raise RuntimeError("network down")
+
+        monkeypatch.setattr(bz, "fetch_bzzoiro_events", _boom)
+
+        class _BrokenDB(_FakeDB):
+            def add(self, obj):
+                if isinstance(obj, IngestFailure):
+                    raise RuntimeError("session closed")
+                super().add(obj)
+
+        db = _BrokenDB()
+
+        # 不应抛异常:死信写不进去只记 warning
+        result = await bz.BzzoiroSource().ingest(db, leagues=["E0"])
+        assert result["leagues"]["E0"]["errors"]