From 03727bda00fcf5f160cf293514658e8b311be976 Mon Sep 17 00:00:00 2001 From: shangfangjian Date: Tue, 22 Sep 2026 01:15:45 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E7=AE=A1=E7=BA=BF=E5=86=99?= =?UTF-8?q?=E5=85=A5=E4=BE=9D=E8=B5=96=E8=A7=A3=E8=80=A6=E2=80=94=E2=80=94?= =?UTF-8?q?pipeline=5Fwrite=20=E7=9B=B4=E6=8E=A5=20import?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 21 ++++++++------------- src/data/bzzoiro_events.py | 9 ++++----- src/data/bzzoiro_standings.py | 13 +++++++------ src/data/bzzoiro_stats.py | 9 +++++---- tests/test_events_bronze.py | 7 +++++-- tests/test_standings_bronze.py | 7 +++++-- 6 files changed, 34 insertions(+), 32 deletions(-) diff --git a/src/data/bzzoiro.py b/src/data/bzzoiro.py index 556f500..1faba90 100644 --- a/src/data/bzzoiro.py +++ b/src/data/bzzoiro.py @@ -1,10 +1,11 @@ """Bzzoiro 数据源:抓取 + 入库(单一数据源)—— 聚合门面。 -实现按管线拆分(单文件 → 多模块),本模块只做再导出,保持两个不变量: +实现按管线拆分(单文件 → 多模块),本模块只做再导出,保持不变量: 1. sources._load_sources() 仍从本模块导入 BzzoiroSource(注册表入口不变); - 2. 测试与脚本对 `bz.<名称>` 的 monkeypatch 语义不变 —— 子模块在运行期 - 经本门面解析可替换协作者(抓取函数 / Bronze 写入助手 / REQUEST_INTERVAL), - 与拆分前的单文件行为一致。 + 2. 测试与脚本对 `bz.<名称>` 的 monkeypatch 语义不变 —— 抓取函数 / 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, diff --git a/src/data/bzzoiro_events.py b/src/data/bzzoiro_events.py index d2f0ca0..77270bb 100644 --- a/src/data/bzzoiro_events.py +++ b/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 bz._write_lineage( + 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}, diff --git a/src/data/bzzoiro_standings.py b/src/data/bzzoiro_standings.py index 46f5edf..0c82f95 100644 --- a/src/data/bzzoiro_standings.py +++ b/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 bz._write_lineage( + await _write_raw_event(db, "bzzoiro", source_record_id, raw_payload, batch_id) + await _write_lineage( db, "bzzoiro", source_record_id, "standings", league_id, "standings_ingest", {"league": league_code, "season": season_label, "rows_upserted": rows_upserted}, diff --git a/src/data/bzzoiro_stats.py b/src/data/bzzoiro_stats.py index 134e69f..444fa18 100644 --- a/src/data/bzzoiro_stats.py +++ b/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 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_raw_event(db, "bzzoiro", str(m.source_event_id), payload, 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 # 基础设施写入失败不影响主流程 diff --git a/tests/test_events_bronze.py b/tests/test_events_bronze.py index 32e875d..00d0392 100644 --- a/tests/test_events_bronze.py +++ b/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) - monkeypatch.setattr(bz, "_write_lineage", _boom) + # Bronze 写入助手直接 import 到 bzzoiro_events 命名空间,需 patch 该处 + monkeypatch.setattr(bz_events, "_write_raw_event", _boom) + monkeypatch.setattr(bz_events, "_write_lineage", _boom) _patch_fetch(monkeypatch, [_event()]) db = _FakeDB() diff --git a/tests/test_standings_bronze.py b/tests/test_standings_bronze.py index ce5d399..4d2b926 100644 --- a/tests/test_standings_bronze.py +++ b/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) - monkeypatch.setattr(bz, "_write_lineage", _boom) + # Bronze 写入助手直接 import 到 bzzoiro_standings 命名空间,需 patch 该处 + monkeypatch.setattr(bz_standings, "_write_raw_event", _boom) + monkeypatch.setattr(bz_standings, "_write_lineage", _boom) _patch_fetch(monkeypatch, _payload()) db = _FakeDB(leagues=[_preset_league()])