refactor: 管线写入依赖解耦——pipeline_write 直接 import

bzzoiro_events / bzzoiro_standings / bzzoiro_stats 直接 import
src.data.pipeline_write(_write_raw_event/_write_lineage/_safe_write_ingest_failure),
删除 bzzoiro.py 门面中的 pipeline_write 转发胶水。

保留 fetch_*/_fetch_json_async/REQUEST_INTERVAL 经 bz. 门面调用(测试 monkeypatch 入口);
测试 best-effort 改为 patch 管线模块自身命名空间(from-import 绑定语义)。
函数语义与「失败不拖垮主流程」不变;source_record_id/transform_name 约定不变。

全量测试 270 通过。

src/data/bzzoiro.py

@@ -1,10 +1,11 @@
 """Bzzoiro 数据源:抓取 + 入库(单一数据源)—— 聚合门面。
 
-实现按管线拆分(单文件 → 多模块),本模块只做再导出,保持两个不变量:
+实现按管线拆分(单文件 → 多模块),本模块只做再导出,保持不变量:
   1. sources._load_sources() 仍从本模块导入 BzzoiroSource(注册表入口不变);
-  2. 测试与脚本对 `bz.<名称>` 的 monkeypatch 语义不变 —— 子模块在运行期
+  2. 测试与脚本对 `bz.<名称>` 的 monkeypatch 语义不变 —— 抓取函数 / REQUEST_INTERVAL
-     经本门面解析可替换协作者(抓取函数 / Bronze 写入助手 / REQUEST_INTERVAL),
+     仍经本门面解析可替换;Bronze 写入助手(_write_raw_event/_write_lineage/
-     与拆分前的单文件行为一致。
+     _safe_write_ingest_failure)已改为管线模块直接 import pipeline_write,
+    测试需 patch `src.data.pipeline_write.*` 源模块。
 
 三条管线(各自模块):
   1. events   — 比赛日程/比分(/events/),含 source_event_id 血缘  → bzzoiro_events.py
@@ -12,7 +13,8 @@
   3. stats    — 已完赛比赛详细统计回填(/events/{id}/stats/)        → bzzoiro_stats.py
 
 共享基础:HTTP 抓取(多 key 轮换)与字段转换 → bzzoiro_common.py;
-Bronze 基础设施(RawEvent/IngestFailure/DataLineage)→ pipeline_write.py。
+Bronze 基础设施(RawEvent/IngestFailure/DataLineage)→ pipeline_write.py
+(各管线模块直接 import pipeline_write,不再经本门面转发)。
 
 D4(工程债): Team/League/Match 的查找/创建经 Repository 层(src/db/repositories.py),
 各管线不直接控制事务(commit/rollback 由调用方 UnitOfWork 控制,只 flush)。
@@ -31,6 +33,7 @@ from src.data.key_ring import _mask  # noqa: F401  (R1 测试引用 bz._mask)
 from src.data.normalize import normalize_bzzoiro  # noqa: F401
 
 # ── 共享原语:HTTP 抓取 + 宽松字段转换 ──
+# NOTE: 管线模块同时从 bzzoiro_common 直接 import _fetch_json_async 等(经本处也转发)。
 from src.data.bzzoiro_common import (  # noqa: F401
     _fetch_json_async,
     _match_key,
@@ -39,14 +42,6 @@ from src.data.bzzoiro_common import (  # noqa: F401
     _to_int_or_none,
 )
 
-# ── 管线基础设施:RawEvent / IngestFailure / DataLineage ──
-from src.data.pipeline_write import (  # noqa: F401
-    _safe_write_ingest_failure,
-    _write_ingest_failure,
-    _write_lineage,
-    _write_raw_event,
-)
-
 # ── events 管线:BzzoiroSource(注册表入口)+ 抓取/入库 ──
 from src.data.bzzoiro_events import (  # noqa: F401
     BzzoiroSource,

src/data/bzzoiro_events.py

@@ -16,6 +16,7 @@ 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
@@ -98,7 +99,7 @@ class BzzoiroSource:
                 # 单联赛抓取失败隔离:记录错误后继续其余联赛,不拖垮整批
                 logger.exception("bzzoiro fetch failed for %s", code)
                 league_r["errors"].append(f"fetch failed: {e}")
-                await bz._safe_write_ingest_failure(
+                await _safe_write_ingest_failure(
                     db,
                     entity_type="events",
                     source_record_id=None,
@@ -300,11 +301,9 @@ async def _write_events_bronze(
     source_record_id 查重保证;best-effort:基础设施写入失败只记 warning,
     绝不拖垮采集主流程(与 _safe_write_ingest_failure 同级约束)。
     """
-    from src.data import bzzoiro as bz
-
     try:
-        await bz._write_raw_event(db, "bzzoiro", source_record_id, raw_payload, batch_id)
+        await _write_raw_event(db, "bzzoiro", source_record_id, raw_payload, batch_id)
-        await bz._write_lineage(
+        await _write_lineage(
             db, "bzzoiro", source_record_id,
             "matches", target_match_id, "events_ingest",
             {"league": league_code, "match_status": match_status},

src/data/bzzoiro_standings.py

@@ -16,6 +16,7 @@ from sqlalchemy import select
 
 from src.data.bzzoiro_common import _to_float_or_none, _to_int_or_none
 from src.data.config import BZZOIRO_LEAGUE_IDS, LEAGUE_COUNTRIES, LEAGUE_NAMES
+from src.data.pipeline_write import _safe_write_ingest_failure, _write_lineage, _write_raw_event
 from src.data.team_names_zh import zh_name
 from src.db.models import Standing, Team
 from src.db.repositories import LeagueRepository, TeamRepository
@@ -33,6 +34,7 @@ async def fetch_bzzoiro_standings(league_code: str, season: str | None = None) -
     params: dict = {}
     if season:
         params["season"] = season
+
     return await bz._fetch_json_async(f"/leagues/{league_id}/standings/", params)
 
 
@@ -57,9 +59,10 @@ async def ingest_bzzoiro_standings(db, *, leagues: Iterable[str], season: str |
     season 为 None 时采集当前赛季(bzzoiro 默认返回 is_current 赛季)。
     球队名与 events 管线使用同一 normalize 规则,保证 Team 匹配。
     """
-    from src.data import bzzoiro as bz
     from src.data.team_names import normalize as normalize_name
 
+    from src.data import bzzoiro as bz
+
     result: dict = {"leagues": {}, "total_upserted": 0, "errors": []}
     for code in leagues:
         league_r: dict = {"upserted": 0, "teams_created": 0, "rows": 0, "errors": []}
@@ -68,7 +71,7 @@ async def ingest_bzzoiro_standings(db, *, leagues: Iterable[str], season: str |
         except Exception as e:
             logger.exception("bzzoiro standings fetch failed for %s", code)
             league_r["errors"].append(str(e))
-            await bz._safe_write_ingest_failure(
+            await _safe_write_ingest_failure(
                 db,
                 entity_type="standings",
                 source_record_id=None,
@@ -198,11 +201,9 @@ async def _write_standings_bronze(
     命中同一条 RawEvent);best-effort:基础设施写入失败只记 warning,
     绝不拖垮采集主流程。
     """
-    from src.data import bzzoiro as bz
-
     try:
-        await bz._write_raw_event(db, "bzzoiro", source_record_id, raw_payload, batch_id)
+        await _write_raw_event(db, "bzzoiro", source_record_id, raw_payload, batch_id)
-        await bz._write_lineage(
+        await _write_lineage(
             db, "bzzoiro", source_record_id,
             "standings", league_id, "standings_ingest",
             {"league": league_code, "season": season_label, "rows_upserted": rows_upserted},

src/data/bzzoiro_stats.py

@@ -15,6 +15,7 @@ from datetime import datetime, timedelta, timezone
 
 from src.data.bzzoiro_common import _to_float_or_none, _to_int_or_none
 from src.data.config import BZZOIRO_LEAGUE_IDS
+from src.data.pipeline_write import _safe_write_ingest_failure, _write_lineage, _write_raw_event
 from src.db.models import MatchStats
 from src.db.repositories import MatchRepository
 
@@ -90,7 +91,7 @@ async def ingest_bzzoiro_event_stats(
 
     筛选条件: match_status=finished 且 source_event_id 非空。
     only_missing=True 时跳过已有统计的比赛(增量);False 则全量刷新。
-    limit 控制单次最多处理的比赛数(上游限速 1.2s/请求,大批量需分次触发)。
+    limit 控制单次最多处理的比赛数(上游限速约 1.2s/请求,大批量需分次触发)。
     """
     from src.data import bzzoiro as bz
 
@@ -120,7 +121,7 @@ async def ingest_bzzoiro_event_stats(
         except Exception as e:
             logger.warning("stats fetch failed match=%s event=%s: %s", m.id, m.source_event_id, e)
             result["errors"].append(f"match {m.id}: {e}")
-            await bz._safe_write_ingest_failure(
+            await _safe_write_ingest_failure(
                 db,
                 entity_type="match_stats",
                 source_record_id=str(m.source_event_id),
@@ -165,8 +166,8 @@ async def ingest_bzzoiro_event_stats(
         # 管线基础设施:写入 RawEvent + DataLineage
         batch_id = f"bzzoiro-stats-{m.source_event_id}-{now.strftime('%Y%m%d%H%M%S')}"
         try:
-            await bz._write_raw_event(db, "bzzoiro", str(m.source_event_id), payload, batch_id)
+            await _write_raw_event(db, "bzzoiro", str(m.source_event_id), payload, batch_id)
-            await bz._write_lineage(db, "bzzoiro", str(m.source_event_id), "match_stats", m.stats.id if m.stats else None, "stats_backfill", {"match_id": m.id}, batch_id)
+            await _write_lineage(db, "bzzoiro", str(m.source_event_id), "match_stats", m.stats.id if m.stats else None, "stats_backfill", {"match_id": m.id}, batch_id)
         except Exception:
             pass  # 基础设施写入失败不影响主流程
 

tests/test_events_bronze.py

@@ -255,11 +255,14 @@ class TestEventsBronzeOnUpdate:
 
 class TestEventsBronzeIsBestEffort:
     async def test_bronze_write_failure_does_not_break_ingest(self, monkeypatch):
+        import src.data.bzzoiro_events as bz_events
+
         async def _boom(*args, **kwargs):
             raise RuntimeError("infra down")
 
-        monkeypatch.setattr(bz, "_write_raw_event", _boom)
+        # Bronze 写入助手直接 import 到 bzzoiro_events 命名空间,需 patch 该处
-        monkeypatch.setattr(bz, "_write_lineage", _boom)
+        monkeypatch.setattr(bz_events, "_write_raw_event", _boom)
+        monkeypatch.setattr(bz_events, "_write_lineage", _boom)
         _patch_fetch(monkeypatch, [_event()])
         db = _FakeDB()
 

tests/test_standings_bronze.py

@@ -261,11 +261,14 @@ class TestStandingsRawEventIdempotent:
 
 class TestStandingsBronzeIsBestEffort:
     async def test_bronze_write_failure_does_not_break_ingest(self, monkeypatch):
+        import src.data.bzzoiro_standings as bz_standings
+
         async def _boom(*args, **kwargs):
             raise RuntimeError("infra down")
 
-        monkeypatch.setattr(bz, "_write_raw_event", _boom)
+        # Bronze 写入助手直接 import 到 bzzoiro_standings 命名空间,需 patch 该处
-        monkeypatch.setattr(bz, "_write_lineage", _boom)
+        monkeypatch.setattr(bz_standings, "_write_raw_event", _boom)
+        monkeypatch.setattr(bz_standings, "_write_lineage", _boom)
         _patch_fetch(monkeypatch, _payload())
         db = _FakeDB(leagues=[_preset_league()])