diff --git a/src/data/bzzoiro.py b/src/data/bzzoiro.py index cfa86b0..725aa7a 100644 --- a/src/data/bzzoiro.py +++ b/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 diff --git a/src/db/repositories.py b/src/db/repositories.py index 20b4a86..76f9b94 100644 --- a/src/db/repositories.py +++ b/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