src/data/bzzoiro.py

@@ -5,7 +5,9 @@
   2. standings— 联赛积分榜快照(/leagues/{id}/standings/)
   3. stats    — 已完赛比赛详细统计回填(/events/{id}/stats/)
 
-使用 Repository 模式进行数据访问,不直接控制事务(由调用方 UnitOfWork 控制)。
+D4(工程债): Team/League/Match 的查找/创建经 Repository 层(src/db/repositories.py),
+本模块不直接控制事务(commit/rollback 由调用方 UnitOfWork 控制,这里只 flush)。
+Standing/RawEvent/Lineage 等管线内私有读写仍在本模块内实现,不强行 Repository 化。
 """
 from __future__ import annotations
 
@@ -16,7 +18,6 @@ from collections.abc import Iterable
 from datetime import datetime, timedelta, timezone
 
 from sqlalchemy import select
-from sqlalchemy.orm import selectinload
 
 import httpx
 
@@ -27,7 +28,8 @@ 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
-from src.db.models import League, Match, MatchStats, Standing, Team, RawEvent, IngestFailure, DataLineage
+from src.db.models import Match, MatchStats, Standing, Team, RawEvent, IngestFailure, DataLineage
+from src.db.repositories import LeagueRepository, MatchRepository, TeamRepository
 
 logger = logging.getLogger(__name__)
 
@@ -210,13 +212,12 @@ class BzzoiroSource:
                 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()
+            # D4: 联赛查找/创建经 LeagueRepository(事务仍由调用方 UoW 提交)
+            league = await LeagueRepository(db).get_or_create(
+                code, LEAGUE_NAMES.get(code, code), LEAGUE_COUNTRIES.get(code)
+            )
+            team_r = TeamRepository(db)
+            match_r = MatchRepository(db)
 
             # === 批量优化: 预加载球队和已有比赛到内存 ===
             team_name_to_id: dict[str, int] = {}
@@ -242,9 +243,10 @@ class BzzoiroSource:
                         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}
+                    team_name_to_id = {
+                        name: t.id
+                        for name, t in (await team_r.get_all_by_names(list(all_team_names))).items()
+                    }
 
                 # P1-2: 按需加载,只加载 raw_events 涉及日期范围的比赛(加 30 天缓冲)
                 # 避免加载联赛全部历史比赛到内存(多赛季采集时内存溢出)
@@ -254,15 +256,12 @@ class BzzoiroSource:
                     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)
+                        matches_in_range = await match_r.find_by_league_and_date_range(
+                            league.id, min_dt, 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()
+                            for m in matches_in_range
                         }
                 # else: existing_matches 保持空 dict(全量新比赛)
 
@@ -275,20 +274,16 @@ class BzzoiroSource:
                 # D1: RawEvent 幂等键(上游 id 或合成键),插入/变更更新共用
                 record_id = _events_record_id(code, nm, raw)
 
-                # 球队: 内存查找 + 按需创建
+                # 球队: 内存查找 + 按需创建(D4: 经 TeamRepository)
                 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 = await team_r.get_or_create(nm.home_team, name_zh=zh_name(nm.home_team))
                     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 = await team_r.get_or_create(nm.away_team, name_zh=zh_name(nm.away_team))
                     away_team_id = away.id
                     team_name_to_id[nm.away_team] = away_team_id
 
@@ -552,13 +547,11 @@ async def ingest_bzzoiro_standings(db, *, leagues: Iterable[str], season: str |
             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()
+        # 联赛(get-or-create,D4: 经 LeagueRepository)
+        league = await LeagueRepository(db).get_or_create(
+            code, LEAGUE_NAMES.get(code, code), LEAGUE_COUNTRIES.get(code)
+        )
+        team_r = TeamRepository(db)
 
         # 赛季标签:优先用返回的 season 对象推导
         season_obj = payload.get("season") or {}
@@ -571,11 +564,7 @@ async def ingest_bzzoiro_standings(db, *, leagues: Iterable[str], season: str |
         # 批量预载球队(与 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
+        team_map: dict[str, Team] = await team_r.get_all_by_names(list(names))
 
         now = datetime.now(timezone.utc)
         for r in rows:
@@ -584,9 +573,7 @@ async def ingest_bzzoiro_standings(db, *, leagues: Iterable[str], season: str |
                 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 = await team_r.get_or_create(team_name, name_zh=zh_name(team_name))
                 team_map[team_name] = team
                 league_r["teams_created"] += 1
 
@@ -727,16 +714,10 @@ async def ingest_bzzoiro_event_stats(
         result["errors"].append("无有效联赛代码")
         return result
 
-    stmt = (
-        select(Match)
-        .options(selectinload(Match.stats))
-        .where(Match.match_status == "finished")
-        .where(Match.source_event_id.is_not(None))
-        .where(Match.league_id.in_(league_ids))
-        .order_by(Match.match_date.desc())
-        .limit(limit * 3 if only_missing else limit)
+    # D4: 候选比赛查询经 MatchRepository(含 stats 预加载,筛选/排序/limit 语义不变)
+    matches = await MatchRepository(db).find_finished_with_stats(
+        league_ids, limit=limit * 3 if only_missing else limit
     )
-    matches = (await db.execute(stmt)).scalars().all()
 
     now = datetime.now(timezone.utc)
     processed = 0

src/db/repositories.py

@@ -64,6 +64,34 @@ class MatchRepository:
         )
         return (await self._session.execute(stmt)).scalar_one_or_none()
 
+    async def find_by_league_and_date_range(
+        self, league_id: int, start, end
+    ) -> list[Match]:
+        """批量预加载某联赛日期范围内的比赛(ingest 管线内存去重用)。"""
+        stmt = (
+            select(Match)
+            .where(Match.league_id == league_id)
+            .where(Match.match_date >= start)
+            .where(Match.match_date <= end)
+        )
+        return (await self._session.execute(stmt)).scalars().all()
+
+    async def find_finished_with_stats(self, league_ids: list[int], *, limit: int) -> list[Match]:
+        """已完赛且有上游 event id 的比赛(按日期倒序),供统计回填逐场拉取。
+
+        预加载 stats:调用方需读取 existing.stats 判断是否跳过。
+        """
+        stmt = (
+            select(Match)
+            .options(selectinload(Match.stats))
+            .where(Match.match_status == "finished")
+            .where(Match.source_event_id.is_not(None))
+            .where(Match.league_id.in_(league_ids))
+            .order_by(Match.match_date.desc())
+            .limit(limit)
+        )
+        return (await self._session.execute(stmt)).scalars().all()
+
     async def add(self, match: Match) -> None:
         self._session.add(match)
         await self._session.flush()
@@ -79,11 +107,11 @@ class TeamRepository:
         stmt = select(Team).where(Team.name == name)
         return (await self._session.execute(stmt)).scalar_one_or_none()
 
-    async def get_or_create(self, name: str) -> Team:
-        """按名获取球队,不存在则创建。"""
+    async def get_or_create(self, name: str, *, name_zh: str | None = None) -> Team:
+        """按名获取球队,不存在则创建(name_zh 供 bzzoiro 管线写中文名)。"""
         team = await self.get_by_name(name)
         if team is None:
-            team = Team(name=name)
+            team = Team(name=name, name_zh=name_zh)
             self._session.add(team)
             await self._session.flush()
         return team