From 9ccda4bf51c5ae534f8212736bb74e11ff838d98 Mon Sep 17 00:00:00 2001
From: shangfangjian <shangfangjian@bilidili.cn>
Date: Mon, 21 Sep 2026 17:10:17 +0800
Subject: [PATCH] =?UTF-8?q?fix(critical):=20=E6=81=A2=E5=A4=8D=20bzzoiro?=
 =?UTF-8?q?=20events=20=E9=87=87=E9=9B=86=E7=AE=A1=E7=BA=BF,=E6=B6=88?=
 =?UTF-8?q?=E9=99=A4=E6=95=B0=E6=8D=AE=E6=BA=90=E6=B3=A8=E5=86=8C=E8=A1=A8?=
 =?UTF-8?q?=E9=9D=99=E9=BB=98=E5=A4=B1=E6=95=88?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

C2 数据丢失修复。重构提交 6a49940 删除了 src/data/bzzoiro.py 中的
fetch_bzzoiro_events 与 BzzoiroSource 后未做替换,同时 sources.py 用
`try/except Exception: pass` 吞掉了 ImportError,导致:

  - _SOURCES 注册表恒为空
  - get_source("bzzoiro") 恒抛 ValueError("未知数据源: bzzoiro")
  - src/api/routes/ingest.py 与 schedules.py 运行时全线失效,
    且日志中无任何导入错误痕迹

修复内容:
1. 从 f05dc1a 恢复并移植 events 管线:
   - fetch_bzzoiro_events():分页抓取 /events/(纯异步,复用现有 _fetch_json_async)
   - @register class BzzoiroSource + ingest():保留原签名
     (db, *, leagues, date_from, date_to, status),保持「不 commit,
     事务由调用方 UnitOfWork 控制」的契约,并保留单联赛抓取失败的
     错误隔离(记录 error 后 continue)
   - 复用现有 _to_date / _to_int_or_none / _match_key,未重复定义
   - 血缘字段继续使用与 nm 配对的 raw(避免 P0-3 回归)
2. sources.py:导入失败改为 logger.exception 记录,新增 _loaded 标记
   保证 _load_sources() 至多执行一次;失败时不置位以便后续重试。
   get_source() 对真正未知的名字仍抛 ValueError。

验证:
   - 新增 tests/test_bzzoiro_source_registry.py(8 项):注册表非空、
     get_source/list_sources 可用、ingest 签名契约、不再静默吞异常
   - test_regressions.py:3 failed -> 2 failed
   - 全量:14 failed -> 13 failed,通过数 190 -> 199

注意事项:未改动任何 caller(routes/ingest.py、routes/schedules.py 原样
可用,证明签名保持正确)。
---
 src/data/bzzoiro.py                   | 293 +++++++++++++++++++++++++-
 src/data/sources.py                   |  38 +++-
 tests/test_bzzoiro_source_registry.py |  92 ++++++++
 3 files changed, 410 insertions(+), 13 deletions(-)
 create mode 100644 tests/test_bzzoiro_source_registry.py

diff --git a/src/data/bzzoiro.py b/src/data/bzzoiro.py
index 4fb345e..0278828 100644
--- a/src/data/bzzoiro.py
+++ b/src/data/bzzoiro.py
@@ -23,7 +23,7 @@ import httpx
 from src.core.runtime_config import get_runtime_value
 from src.core.http_client import get_client
 from src.data.config import BZZOIRO_LEAGUE_IDS, LEAGUE_COUNTRIES, LEAGUE_NAMES, REQUEST_INTERVAL
-from src.data.key_ring import get_key_ring
+from src.data.key_ring import _mask, get_key_ring
 from src.data.normalize import normalize_bzzoiro
 from src.data.team_names_zh import zh_name
 from src.data.sources import register
@@ -101,7 +101,7 @@ async def _fetch_json_async(path: str, params: dict | None = None, max_retries:
                 # 限流:标记当前 key 冷却,切换到下一个
                 new_key = ring.report_rate_limited(key)
                 if new_key and new_key != key:
-                    logger.info("bzzoiro 429 → 切换 key: %s → %s,立即重试", _km(key), _km(new_key))
+                    logger.info("bzzoiro 429 → 切换 key: %s → %s,立即重试", _mask(key), _mask(new_key))
                     key = new_key
                     continue  # 立即重试,不等待
                 # 单 key 或全部冷却:等待最早恢复的 key
@@ -130,6 +130,209 @@ async def _fetch_json_async(path: str, params: dict | None = None, max_retries:
     raise RuntimeError(f"bzzoiro request failed after {max_retries} attempts: {last_exc}")
 
 
+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)。"""
+    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 _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(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 控制。
+        """
+        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 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}")
+                result["leagues"][code] = league_r
+                continue
+
+            # 获取或创建联赛
+            stmt = select(League).where(League.code == code)
+            league = (await db.execute(stmt)).scalar_one_or_none()
+            if league is None:
+                league = League(code=code, name=LEAGUE_NAMES.get(code, code), country=LEAGUE_COUNTRIES.get(code))
+                db.add(league)
+                await db.flush()
+
+            # === 批量优化: 预加载球队和已有比赛到内存 ===
+            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 as e:
+                            # P1-3: 统一使用 warning,不追加到 errors(仅运行时错误入 errors)
+                            logger.warning("normalize skip: %s", e)
+                            continue
+                        normalized_matches.append((nm, raw))
+                        all_team_names.add(nm.home_team)
+                        all_team_names.add(nm.away_team)
+
+                if all_team_names:
+                    stmt = select(Team).where(Team.name.in_(all_team_names))
+                    teams = (await db.execute(stmt)).scalars().all()
+                    team_name_to_id = {t.name: t.id for t in teams}
+
+                # 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)
+                        stmt = (
+                            select(Match)
+                            .where(Match.league_id == league.id)
+                            .where(Match.match_date >= min_dt)
+                            .where(Match.match_date <= max_dt)
+                        )
+                        existing_matches = {
+                            _match_key(m.home_team_id, m.away_team_id, m.match_date_date): m
+                            for m in (await db.execute(stmt)).scalars()
+                        }
+                # else: existing_matches 保持空 dict(全量新比赛)
+
+            for nm, raw in normalized_matches:
+                # 球队: 内存查找 + 按需创建
+                home_team_id = team_name_to_id.get(nm.home_team)
+                if home_team_id is None:
+                    home = Team(name=nm.home_team, name_zh=zh_name(nm.home_team))
+                    db.add(home)
+                    await db.flush()
+                    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 = Team(name=nm.away_team, name_zh=zh_name(nm.away_team))
+                    db.add(away)
+                    await db.flush()
+                    away_team_id = away.id
+                    team_name_to_id[nm.away_team] = away_team_id
+
+                # 查找已有比赛: 内存查找
+                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,
+                        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
+                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
+                        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
+
+            # 注意: 不在此处 commit,由调用方 UnitOfWork 控制事务
+            result["leagues"][code] = league_r
+            result["total_inserted"] += league_r["inserted"]
+            result["total_updated"] += league_r["updated"]
+        return result
+
+
 # ============================================================
 # 管线基础设施:RawEvent / IngestFailure / DataLineage
 # ============================================================
@@ -226,6 +429,92 @@ async def ingest_bzzoiro_standings(db, *, leagues: Iterable[str], season: str |
             continue
 
         rows = payload.get("standings") or []
+        if not rows:
+            result["leagues"][code] = {"error": "无积分榜数据(赛季未开始或未提供)"}
+            result["errors"].append(f"{code}: 无积分榜数据")
+            continue
+
+        # 联赛(get-or-create)
+        stmt = select(League).where(League.code == code)
+        league = (await db.execute(stmt)).scalar_one_or_none()
+        if league is None:
+            league = League(code=code, name=LEAGUE_NAMES.get(code, code), country=LEAGUE_COUNTRIES.get(code))
+            db.add(league)
+            await db.flush()
+
+        # 赛季标签:优先用返回的 season 对象推导
+        season_obj = payload.get("season") or {}
+        season_label = _season_label_from_dates(
+            season_obj.get("start_date"), season_obj.get("end_date")
+        )
+        if season_label == "?":
+            season_label = season or ""
+
+        # 批量预载球队(与 events 管线使用同一 normalize 规则,保证 Team 匹配)
+        names = {normalize_name(str(r.get("team_name", ""))) for r in rows}
+        names.discard("")
+        team_map: dict[str, Team] = {}
+        if names:
+            stmt = select(Team).where(Team.name.in_(names))
+            for t in (await db.execute(stmt)).scalars():
+                team_map[t.name] = t
+
+        now = datetime.now(timezone.utc)
+        for r in rows:
+            team_name = normalize_name(str(r.get("team_name", "")))
+            if not team_name:
+                continue
+            team = team_map.get(team_name)
+            if team is None:
+                team = Team(name=team_name, name_zh=zh_name(team_name))
+                db.add(team)
+                await db.flush()
+                team_map[team_name] = team
+                league_r["teams_created"] += 1
+
+            zone = r.get("zone") or {}
+            values = dict(
+                position=_to_int_or_none(r.get("position")) or 0,
+                played=_to_int_or_none(r.get("played")) or 0,
+                won=_to_int_or_none(r.get("won")) or 0,
+                drawn=_to_int_or_none(r.get("drawn")) or 0,
+                lost=_to_int_or_none(r.get("lost")) or 0,
+                goals_for=_to_int_or_none(r.get("gf")) or 0,
+                goals_against=_to_int_or_none(r.get("ga")) or 0,
+                goal_diff=_to_int_or_none(r.get("gd")) or 0,
+                points=_to_int_or_none(r.get("pts")) or 0,
+                xg_for=_to_float_or_none(r.get("xgf")),
+                xg_against=_to_float_or_none(r.get("xga")),
+                form=r.get("form") or None,
+                zone=zone.get("label") or zone.get("key") or None,
+                updated_at=now,
+                retrieved_at=now,
+            )
+
+            # 同一联赛同一赛季只保留最新快照:按 (league, season, team) upsert
+            stmt = select(Standing).where(
+                Standing.league_id == league.id,
+                Standing.season == season_label,
+                Standing.team_id == team.id,
+            )
+            standing = (await db.execute(stmt)).scalar_one_or_none()
+            if standing is None:
+                standing = Standing(
+                    league_id=league.id, season=season_label, team_id=team.id, **values
+                )
+                db.add(standing)
+            else:
+                for k, v in values.items():
+                    setattr(standing, k, v)
+            league_r["upserted"] += 1
+
+        league_r["rows"] = len(rows)
+        result["leagues"][code] = league_r
+        result["total_upserted"] += league_r["upserted"]
+        logger.info(
+            "bzzoiro standings 采集完成: %s 赛季 %s, upsert %d/%d",
+            code, season_label, league_r["upserted"], league_r["rows"],
+        )
     return result
 
 
diff --git a/src/data/sources.py b/src/data/sources.py
index 6466323..f4f363a 100644
--- a/src/data/sources.py
+++ b/src/data/sources.py
@@ -8,10 +8,13 @@
 """
 from __future__ import annotations
 
+import logging
 from typing import Protocol
 
 from src.db.base import AsyncSession
 
+logger = logging.getLogger(__name__)
+
 
 class DataSource(Protocol):
     """比赛数据源契约:抓取 → 规范化 → 入库。"""
@@ -29,6 +32,9 @@ class DataSource(Protocol):
 # ── 注册表 ──
 _SOURCES: dict[str, DataSource] = {}
 
+# 延迟加载标记:保证 _load_sources() 最多执行一次,重复调用廉价
+_loaded = False
+
 
 def register(source):
     """装饰器:将数据源注册到全局注册表。
@@ -42,8 +48,7 @@ def register(source):
 
 def get_source(name: str) -> DataSource:
     """按名获取数据源。"""
-    if not _SOURCES:
-        _load_sources()
+    _load_sources()
     if name not in _SOURCES:
         raise ValueError(f"未知数据源: {name}")
     return _SOURCES[name]
@@ -51,18 +56,29 @@ def get_source(name: str) -> DataSource:
 
 def list_sources() -> list[str]:
     """列出所有已注册数据源名。"""
-    if not _SOURCES:
-        _load_sources()
+    _load_sources()
     return list(_SOURCES.keys())
 
 
 def _load_sources() -> None:
-    """延迟导入数据源触发 @register(避免循环导入)。"""
-    from src.data.bzzoiro import BzzoiroSource  # noqa: F811
+    """延迟导入数据源触发 @register(避免循环导入)。
+
+    导入失败必须留下痕迹:静默吞掉 ImportError 会让注册表恒为空,
+    导致 get_source() 对所有数据源都报「未知数据源」,把导入错误
+    伪装成「不存在的名字」—— 这类静默失效极难定位,故在此显式记日志。
+    """
+    global _loaded
+    if _loaded:
+        return
+    try:
+        from src.data.bzzoiro import BzzoiroSource  # noqa: F401
+    except Exception:
+        logger.exception(
+            "数据源模块导入失败,注册表将为空 —— get_source() 会对所有名字报「未知数据源」"
+        )
+        return  # 保持 _loaded=False,下次调用可重试
+    _loaded = True
 
 
-# 保持向后兼容:模块加载时尝试加载(但不再强制)
-try:
-    _load_sources()
-except Exception:
-    pass
+# 模块加载时预热(失败会记录日志,不再静默)
+_load_sources()
diff --git a/tests/test_bzzoiro_source_registry.py b/tests/test_bzzoiro_source_registry.py
new file mode 100644
index 0000000..e38aab2
--- /dev/null
+++ b/tests/test_bzzoiro_source_registry.py
@@ -0,0 +1,92 @@
+"""回归测试:锁定 bzzoiro events 管线被删除 + 注册表静默失效的缺陷不再复发。
+
+背景(真实数据丢失事故):
+  重构提交 6a49940 删除了 src/data/bzzoiro.py 中的 fetch_bzzoiro_events 与
+  BzzoiroSource,但 src/data/sources.py 的模块级预热把 ImportError 用
+  `try/except Exception: pass` 吞掉了。后果:
+    - _SOURCES 注册表恒为空
+    - get_source("bzzoiro") 恒抛 ValueError("未知数据源: bzzoiro")
+    - src/api/routes/ingest.py 与 schedules.py 在运行时全线失效,
+      且日志中看不到任何导入错误的痕迹 —— 这才是它长期漏网的原因。
+
+本测试用静态结构断言 + 真实导入来锁死这两点,不 mock 网络:
+  - 注册表必须非空(直接守卫「静默吞异常」回归)
+  - get_source("bzzoiro") 必须返回真实实例
+  - ingest() 的签名必须保持 caller 依赖的 keyword-only 参数
+"""
+from __future__ import annotations
+
+import inspect
+from pathlib import Path
+
+import pytest
+
+from src.data.bzzoiro import BzzoiroSource, fetch_bzzoiro_events
+from src.data.sources import get_source, list_sources
+
+SRC = Path(__file__).resolve().parent.parent / "src"
+
+
+class TestSourceRegistry:
+    """P0: 注册表必须真的装载到 bzzoiro,而不是静默为空。"""
+
+    def test_registry_is_not_empty(self):
+        """注册表恒非空 —— 直接守卫 `try/except: pass` 静默吞 ImportError。"""
+        assert list_sources(), (
+            "数据源注册表为空 —— _load_sources() 的导入失败了,"
+            "且(修复前)异常被静默吞掉。任何导入错误都必须被 logger.exception 记录。"
+        )
+
+    def test_get_source_returns_bzzoiro(self):
+        """get_source('bzzoiro') 必须返回实例,而不是抛 ValueError。"""
+        source = get_source("bzzoiro")
+        assert source.name == "bzzoiro"
+
+    def test_list_sources_contains_bzzoiro(self):
+        assert "bzzoiro" in list_sources()
+
+    def test_register_decorator_still_exports_the_class(self):
+        """@register 必须仍然返回原类(不能再被改成返回实例)。"""
+        assert isinstance(BzzoiroSource, type), "@register 不应把类替换成实例"
+
+
+class TestIngestContract:
+    """caller 依赖 ingest() 的签名,routes 传的就是这些参数。"""
+
+    def test_ingest_exists_and_is_async(self):
+        assert hasattr(BzzoiroSource, "ingest"), "BzzoiroSource.ingest 丢失(events 管线被删)"
+        assert inspect.iscoroutinefunction(BzzoiroSource.ingest), "ingest 必须是 async"
+
+    def test_ingest_accepts_caller_keyword_args(self):
+        """ingest.py:65 与 schedules.py:41 依赖的 keyword-only 参数必须齐全。"""
+        params = inspect.signature(BzzoiroSource.ingest).parameters
+        for name in ("leagues", "date_from", "date_to", "status"):
+            assert name in params, f"ingest() 缺少 keyword 参数 {name!r} —— caller 会 TypeError"
+        # leagues 必须 keyword-only(src/api/routes/ingest.py 用 leagues=[code] 传)
+        assert params["leagues"].kind is inspect.Parameter.KEYWORD_ONLY
+        # 默认值契约:schedules.py 只传 leagues + status
+        assert params["date_from"].default is None
+        assert params["date_to"].default is None
+        assert params["status"].default == "finished"
+
+    def test_fetch_bzzoiro_events_signature(self):
+        """抓取函数也必须存在,且 leagues 抓取走 keyword-only 的 status/日期。"""
+        assert callable(fetch_bzzoiro_events)
+        params = inspect.signature(fetch_bzzoiro_events).parameters
+        assert "league_code" in params
+        for name in ("status", "date_from", "date_to"):
+            assert name in params, f"fetch_bzzoiro_events() 缺少 {name!r}"
+
+
+class TestNoSilentImportSwallow:
+    """sources.py 不许再用 `except Exception: pass` 吞掉导入失败。"""
+
+    def test_load_sources_logs_failure(self):
+        src = (SRC / "data" / "sources.py").read_text(encoding="utf-8")
+        assert "logger" in src, "sources.py 缺少模块 logger,无法记录导入失败"
+        body = src[src.index("def _load_sources"):]
+        assert "logger.exception" in body, (
+            "_load_sources() 失败时未记日志 —— 静默吞异常会让注册表恒为空,"
+            "把 ImportError 伪装成「未知数据源」(P0 事故根因)"
+        )
+        assert "pass" not in body.split("def ")[0], "不得再用裸 pass 吞掉导入异常"