"""bzzoiro events 管线:比赛日程/比分抓取(/events/)与入库(matches 表)。 从 bzzoiro.py 拆出。比赛主数据唯一入口;Team/League/Match 查找/创建经 Repository 层,事务由调用方 UnitOfWork 控制(分批事务约定不变)。 可替换协作者(抓取函数 / Bronze 写入助手 / REQUEST_INTERVAL)在运行期 经聚合门面 src.data.bzzoiro 解析 —— 与拆分前的单文件 monkeypatch 语义一致。 """ from __future__ import annotations import asyncio import logging from collections.abc import Iterable 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 from src.db.repositories import LeagueRepository, MatchRepository, TeamRepository logger = logging.getLogger(__name__) 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)。""" from src.data import bzzoiro as bz 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 bz._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(bz.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 控制。 """ from src.data import bzzoiro as bz 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 bz.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}") await _safe_write_ingest_failure( db, entity_type="events", source_record_id=None, error=e, raw_payload={"league": code, "status": status, "date_from": date_from, "date_to": date_to}, ) result["leagues"][code] = league_r continue # 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] = {} 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: continue normalized_matches.append((nm, raw)) all_team_names.add(nm.home_team) all_team_names.add(nm.away_team) if all_team_names: 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 天缓冲) # 避免加载联赛全部历史比赛到内存(多赛季采集时内存溢出) 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) 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 matches_in_range } # else: existing_matches 保持空 dict(全量新比赛) # D1: Bronze 层批次信息(每联赛每批次一个 batch_id;seen 防同批重复写入) now = datetime.now(timezone.utc) bronze_batch_id = f"bzzoiro-events-{code}-{now:%Y%m%d%H%M%S}" bronze_written: set[str] = set() for nm, raw in normalized_matches: # 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 = 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 = 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 # 查找已有比赛:优先按 upstream event_id 定位(命中即唯一), # 否则回退自然键(联赛+主客+天级日期)内存查找。 # source_event_id 上有 partial unique 索引保障 upstream 唯一。 eid = _to_int_or_none(raw.get("id")) existing_match = None if eid is not None: existing_match = await match_r.find_by_source_event_id(eid) if existing_match is None: 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, score_status=nm.score_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 # D1: 成功插入 → 补写 Bronze 层(原始载荷 + 血缘) if record_id not in bronze_written: bronze_written.add(record_id) await _write_events_bronze( db, source_record_id=record_id, raw_payload=raw, target_match_id=m.id, league_code=code, match_status=nm.match_status, batch_id=bronze_batch_id, ) 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 # 比分由缺变有 → 标记 known existing_match.score_status = "known" changed = True elif ( nm.match_status == "finished" and nm.home_goals is None and existing_match.score_status == "unknown" ): # 确认完赛仍缺分 → 标记 missing(不伪造 0:0) existing_match.score_status = "missing" 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 # D1: 变更更新 → 补写血缘(RawEvent 幂等键不变,重复采集自动跳过) if record_id not in bronze_written: bronze_written.add(record_id) await _write_events_bronze( db, source_record_id=record_id, raw_payload=raw, target_match_id=existing_match.id, league_code=code, match_status=nm.match_status, batch_id=bronze_batch_id, ) # 注意: 不在此处 commit,由调用方 UnitOfWork 控制事务 result["leagues"][code] = league_r result["total_inserted"] += league_r["inserted"] result["total_updated"] += league_r["updated"] return result def _events_record_id(league_code: str, nm, raw: dict) -> str: """events 载荷的 RawEvent 幂等键。 优先用上游 event id;缺失时用 (league:home:away:date) 合成稳定键 —— 取 normalize 后的队名与天级日期(与 _match_key 同口径),不依赖 DB 自增 id, 保证同一来源比赛重复采集时命中同一条 RawEvent,不产生重复原始载荷。 """ eid = _to_int_or_none(raw.get("id")) if eid is not None: return str(eid) d = _to_date(nm.date) date_part = d.isoformat() if d is not None else "na" return f"{league_code}:{nm.home_team}:{nm.away_team}:{date_part}" async def _write_events_bronze( db, *, source_record_id: str, raw_payload: dict, target_match_id: int | None, league_code: str, match_status: str | None, batch_id: str, ) -> None: """events 成功插入/更新单场比赛后的 Bronze 层补写:RawEvent(幂等) + DataLineage。 D1(工程债):此前只有 stats 回填写 RawEvent/Lineage,events 管线作为比赛 主数据的唯一入口反而不留溯源记录。幂等性由 _write_raw_event 的 source_record_id 查重保证;best-effort:基础设施写入失败只记 warning, 绝不拖垮采集主流程(与 _safe_write_ingest_failure 同级约束)。 """ try: 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}, batch_id, ) except Exception: logger.warning( "events Bronze 写入失败(record=%s, match=%s),不影响采集主流程", source_record_id, target_match_id, exc_info=True, )