From 317a5e338a7a911be20500944aa1882d353d83c5 Mon Sep 17 00:00:00 2001 From: shangfangjian Date: Mon, 21 Sep 2026 23:27:48 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20bzzoiro.py=20=E6=8C=89=E7=AE=A1?= =?UTF-8?q?=E7=BA=BF=E6=8B=86=E5=88=86=E4=B8=BA=205=20=E4=B8=AA=E6=A8=A1?= =?UTF-8?q?=E5=9D=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 单文件 852 行按职责拆分,保持 BzzoiroSource 与 get_source("bzzoiro") 行为不变: - bzzoiro_common HTTP 抓取(多 key 轮换) + 字段转换原语 - bzzoiro_events fetch_bzzoiro_events + BzzoiroSource.ingest + Bronze 补写 - bzzoiro_standings standings 管线 - bzzoiro_stats stats 回填 - pipeline_write RawEvent/IngestFailure/DataLineage 写入助手 子模块运行期经聚合门面 src.data.bzzoiro 解析可替换协作者, 单文件时代的 bz.* monkeypatch 语义完全保留。 路由 import 已指向新模块(ingest.py / schedules.py)。 --- src/api/routes/ingest.py | 3 +- src/api/routes/schedules.py | 3 +- src/data/bzzoiro.py | 852 +++------------------------------- src/data/bzzoiro_common.py | 123 +++++ src/data/bzzoiro_events.py | 310 +++++++++++++ src/data/bzzoiro_standings.py | 215 +++++++++ src/data/bzzoiro_stats.py | 180 +++++++ src/data/pipeline_write.py | 80 ++++ tests/test_available_at.py | 8 +- tests/test_regressions.py | 6 +- 10 files changed, 984 insertions(+), 796 deletions(-) create mode 100644 src/data/bzzoiro_common.py create mode 100644 src/data/bzzoiro_events.py create mode 100644 src/data/bzzoiro_standings.py create mode 100644 src/data/bzzoiro_stats.py create mode 100644 src/data/pipeline_write.py diff --git a/src/api/routes/ingest.py b/src/api/routes/ingest.py index 0799979..41ef124 100644 --- a/src/api/routes/ingest.py +++ b/src/api/routes/ingest.py @@ -16,7 +16,8 @@ from fastapi import APIRouter, Depends, HTTPException from src.api.deps import require_admin from src.api.schemas import IngestBzzoiroRequest from src.data.config import BZZOIRO_LEAGUE_IDS -from src.data.bzzoiro import ingest_bzzoiro_event_stats, ingest_bzzoiro_standings +from src.data.bzzoiro_standings import ingest_bzzoiro_standings +from src.data.bzzoiro_stats import ingest_bzzoiro_event_stats from src.data.sources import get_source from src.db.unit_of_work import get_uow diff --git a/src/api/routes/schedules.py b/src/api/routes/schedules.py index 75f1992..faf4acb 100644 --- a/src/api/routes/schedules.py +++ b/src/api/routes/schedules.py @@ -10,7 +10,8 @@ from sqlalchemy import select, delete from src.api.deps import require_admin from src.api.schemas import ScheduleIn, ScheduleUpdate, ScheduleOut from src.core.scheduler import scheduler -from src.data.bzzoiro import ingest_bzzoiro_event_stats, ingest_bzzoiro_standings +from src.data.bzzoiro_standings import ingest_bzzoiro_standings +from src.data.bzzoiro_stats import ingest_bzzoiro_event_stats from src.data.sources import get_source from src.data.config import BZZOIRO_LEAGUE_IDS from src.db.base import AsyncSession, get_db_read diff --git a/src/data/bzzoiro.py b/src/data/bzzoiro.py index 725aa7a..556f500 100644 --- a/src/data/bzzoiro.py +++ b/src/data/bzzoiro.py @@ -1,793 +1,71 @@ -"""Bzzoiro 数据源:抓取 + 入库(单一数据源)。 +"""Bzzoiro 数据源:抓取 + 入库(单一数据源)—— 聚合门面。 -三条管线: - 1. events — 比赛日程/比分(/events/),含 source_event_id 血缘 - 2. standings— 联赛积分榜快照(/leagues/{id}/standings/) - 3. stats — 已完赛比赛详细统计回填(/events/{id}/stats/) +实现按管线拆分(单文件 → 多模块),本模块只做再导出,保持两个不变量: + 1. sources._load_sources() 仍从本模块导入 BzzoiroSource(注册表入口不变); + 2. 测试与脚本对 `bz.<名称>` 的 monkeypatch 语义不变 —— 子模块在运行期 + 经本门面解析可替换协作者(抓取函数 / Bronze 写入助手 / REQUEST_INTERVAL), + 与拆分前的单文件行为一致。 + +三条管线(各自模块): + 1. events — 比赛日程/比分(/events/),含 source_event_id 血缘 → bzzoiro_events.py + 2. standings— 联赛积分榜快照(/leagues/{id}/standings/) → bzzoiro_standings.py + 3. stats — 已完赛比赛详细统计回填(/events/{id}/stats/) → bzzoiro_stats.py + +共享基础:HTTP 抓取(多 key 轮换)与字段转换 → bzzoiro_common.py; +Bronze 基础设施(RawEvent/IngestFailure/DataLineage)→ pipeline_write.py。 D4(工程债): Team/League/Match 的查找/创建经 Repository 层(src/db/repositories.py), -本模块不直接控制事务(commit/rollback 由调用方 UnitOfWork 控制,这里只 flush)。 -Standing/RawEvent/Lineage 等管线内私有读写仍在本模块内实现,不强行 Repository 化。 +各管线不直接控制事务(commit/rollback 由调用方 UnitOfWork 控制,只 flush)。 +Standing/RawEvent/Lineage 等管线内私有读写仍不强行 Repository 化。 """ from __future__ import annotations -import asyncio -import logging -import random -from collections.abc import Iterable -from datetime import datetime, timedelta, timezone - -from sqlalchemy import select - -import httpx - -from src.core.runtime_config import get_runtime_value -from src.core.http_client import get_client -from src.data.config import BZZOIRO_LEAGUE_IDS, LEAGUE_COUNTRIES, LEAGUE_NAMES, REQUEST_INTERVAL -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 Match, MatchStats, Standing, Team, RawEvent, IngestFailure, DataLineage -from src.db.repositories import LeagueRepository, MatchRepository, TeamRepository - -logger = logging.getLogger(__name__) - - -def _to_date(value): - """把 datetime / date / str 统一成 `date`。""" - if value is None: - return None - if hasattr(value, "date") and callable(value.date): - return value.date() - return value - - -def _to_int_or_none(value) -> int | None: - """宽松转 int(用于上游 ID 解析,失败返回 None 不抛错)。""" - if value is None: - return None - try: - return int(str(value).strip()) - except (TypeError, ValueError): - return None - - -def _match_key(home_team_id: int, away_team_id: int, match_date) -> tuple[int, int, str]: - """比赛去重键:(主队, 客队, 天级日期 ISO 字符串)。 - - 统一在这里构造,避免"预加载时用 str(date)、写入时用 isoformat()"这类 - 隐式格式依赖 —— 两者当前恰好相等,但一旦有人改动其一就会静默失配, - 导致所有比赛被判为不存在而重复插入。 - """ - d = _to_date(match_date) - return (home_team_id, away_team_id, d.isoformat() if d is not None else "") - - -async def _fetch_json_async(path: str, params: dict | None = None, max_retries: int = 3) -> dict | list: - """异步 HTTP(bzzoiro 使用 httpx,不再阻塞事件循环线程池)。 - - 多 key 轮换:遇到 429 自动切换到下一个 key;全部 key 冷却时等待最早恢复。 - """ - base = (await get_runtime_value("BZZOIRO_BASE")).rstrip("/") - raw_keys = await get_runtime_value("BZZOIRO_KEY") - ring = get_key_ring(base, raw_keys) - - url = f"{base}/{path.lstrip('/')}" - key = ring.get() - if not key: - raise RuntimeError("BZZOIRO_KEY 未设置") - - last_exc: Exception | None = None - for attempt in range(max_retries): - headers = { - "Authorization": f"Token {key}", - "Accept": "application/json", - } - try: - client = get_client() - # 整请求兜底: httpx 无 total 超时,用 wait_for 防「滴水式」限速挂死 - resp = await asyncio.wait_for( - client.get( - url, headers=headers, params=params, - timeout=httpx.Timeout(connect=10.0, read=30.0, write=10.0, pool=10.0), - ), - timeout=60.0, - ) - resp.raise_for_status() - return resp.json() - except Exception as e: - last_exc = e - status = getattr(getattr(e, "response", None), "status_code", None) - if status == 429: - # 限流:标记当前 key 冷却,切换到下一个 - new_key = ring.report_rate_limited(key) - if new_key and new_key != key: - logger.info("bzzoiro 429 → 切换 key: %s → %s,立即重试", _mask(key), _mask(new_key)) - key = new_key - continue # 立即重试,不等待 - # 单 key 或全部冷却:等待最早恢复的 key - wait = ring.wait_if_all_blocked() - if wait > 0: - logger.warning("bzzoiro 全部 key 冷却,等待 %.1fs 后重试", wait) - await asyncio.sleep(min(wait, 30.0)) - else: - delay = min(2 ** attempt, 16) + random.uniform(0, 1) - logger.warning("bzzoiro 429, retry %d in %.1fs", attempt + 1, delay) - await asyncio.sleep(delay) - key = ring.get() or key - continue - if 500 <= (status or 0) < 600: - delay = min(2 ** attempt, 16) + random.uniform(0, 1) - logger.warning("bzzoiro %d, retry %d in %.1fs", status, attempt + 1, delay) - await asyncio.sleep(delay) - continue - # 网络错误(连接失败/超时)也退避重试 - if isinstance(e, (TimeoutError, ConnectionError, OSError)): - delay = min(2 ** attempt, 16) + random.uniform(0, 1) - logger.warning("bzzoiro network error, retry %d in %.1fs: %s", attempt + 1, delay, e) - await asyncio.sleep(delay) - continue - raise - raise RuntimeError(f"bzzoiro request failed after {max_retries} attempts: {last_exc}") - - -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)。""" - 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 _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(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 控制。 - """ - 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 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 as e: - # P1-3: 统一使用 warning,不追加到 errors(仅运行时错误入 errors) - logger.warning("normalize skip: %s", e) - 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 - - # 查找已有比赛: 内存查找 - 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, - 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 - 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 - - -# ============================================================ -# 管线基础设施:RawEvent / IngestFailure / DataLineage -# ============================================================ - - -async def _write_raw_event(db, source_system: str, source_record_id: str, raw_payload: dict, batch_id: str | None = None) -> None: - """写入 Bronze 层原始事件(幂等:同 source_record_id 跳过)。""" - from sqlalchemy import select as _select - stmt = _select(RawEvent).where( - RawEvent.source_system == source_system, - RawEvent.source_record_id == source_record_id, - ) - existing = (await db.execute(stmt)).scalar_one_or_none() - if existing is None: - db.add(RawEvent( - source_system=source_system, - source_record_id=source_record_id, - raw_payload=raw_payload, - ingest_batch_id=batch_id, - )) - - -async def _write_ingest_failure(db, source_system: str, entity_type: str, source_record_id: str | None, error_type: str, error_detail: str | None, raw_payload: dict | None = None) -> None: - """写入采集失败死信。""" - db.add(IngestFailure( - source_system=source_system, - entity_type=entity_type, - source_record_id=source_record_id, - error_type=error_type, - error_detail=error_detail, - raw_payload=raw_payload, - )) - - -async def _safe_write_ingest_failure( - db, - *, - entity_type: str, - source_record_id: str | None, - error: Exception, - raw_payload: dict | None = None, -) -> None: - """抓取失败时尽力写入死信表(失败不影响主流程)。 - - 死信是「可观测性」基础设施,与 RawEvent/Lineage 同级:写入失败只记 - warning,绝不能让原始抓取错误之外的新异常打断采集循环。 - """ - try: - await _write_ingest_failure( - db, "bzzoiro", entity_type, source_record_id, - "fetch_error", str(error), raw_payload, - ) - except Exception: - logger.warning( - "写入 ingest_failures 死信失败(entity=%s, record=%s): %s", - entity_type, source_record_id, error, exc_info=True, - ) - - -async def _write_lineage(db, source_system: str, source_record_id: str, target_table: str, target_id: int | None, transform_name: str, transform_detail: dict | None = None, batch_id: str | None = None) -> None: - """写入 ETL 血缘追踪。""" - db.add(DataLineage( - source_system=source_system, - source_record_id=source_record_id, - target_table=target_table, - target_id=target_id, - transform_name=transform_name, - transform_detail=transform_detail, - batch_id=batch_id, - )) - - -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, - ) - - -# ============================================================ -# 积分榜管线:/leagues/{id}/standings/ → standings 表 -# ============================================================ - -async def fetch_bzzoiro_standings(league_code: str, season: str | None = None) -> dict: - """抓取联赛积分榜(纯抓取,不入库)。season 为 None 时取当前赛季。""" - league_id = BZZOIRO_LEAGUE_IDS.get(league_code) - if league_id is None: - raise ValueError(f"未知联赛代码: {league_code}") - params: dict = {} - if season: - params["season"] = season - return await _fetch_json_async(f"/leagues/{league_id}/standings/", params) - - -def _season_label_from_dates(start_date, end_date) -> str: - """从赛季起止日期推导赛季标签(与 derive_season_label 语义一致)。""" - try: - if isinstance(start_date, str): - start = datetime.fromisoformat(start_date[:10]) - else: - start = start_date - y = start.year - return f"{y}-{y + 1}" if start.month >= 8 else f"{y - 1}-{y}" - except (TypeError, ValueError): - return "?" - - -async def ingest_bzzoiro_standings(db, *, leagues: Iterable[str], season: str | None = None) -> dict: - """采集积分榜 → upsert standings 表。 - - season 为 None 时采集当前赛季(bzzoiro 默认返回 is_current 赛季)。 - 球队名与 events 管线使用同一 normalize 规则,保证 Team 匹配。 - """ - from src.data.team_names import normalize as normalize_name - - result: dict = {"leagues": {}, "total_upserted": 0, "errors": []} - for code in leagues: - league_r: dict = {"upserted": 0, "teams_created": 0, "rows": 0, "errors": []} - try: - payload = await fetch_bzzoiro_standings(code, season=season) - except Exception as e: - logger.exception("bzzoiro standings fetch failed for %s", code) - league_r["errors"].append(str(e)) - await _safe_write_ingest_failure( - db, - entity_type="standings", - source_record_id=None, - error=e, - raw_payload={"league": code, "season": season}, - ) - result["leagues"][code] = league_r - result["errors"].append(f"{code}: {e}") - continue - - rows = payload.get("standings") or [] - if not rows: - result["leagues"][code] = {"error": "无积分榜数据(赛季未开始或未提供)"} - result["errors"].append(f"{code}: 无积分榜数据") - continue - - # 联赛(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 {} - season_label = _season_label_from_dates( - season_obj.get("start_date"), season_obj.get("end_date") - ) - if season_label == "?": - season_label = season or "" - - # 批量预载球队(与 events 管线使用同一 normalize 规则,保证 Team 匹配) - names = {normalize_name(str(r.get("team_name", ""))) for r in rows} - names.discard("") - team_map: dict[str, Team] = await team_r.get_all_by_names(list(names)) - - now = datetime.now(timezone.utc) - for r in rows: - team_name = normalize_name(str(r.get("team_name", ""))) - if not team_name: - continue - team = team_map.get(team_name) - if team is None: - 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 - - zone = r.get("zone") or {} - values = dict( - position=_to_int_or_none(r.get("position")) or 0, - played=_to_int_or_none(r.get("played")) or 0, - won=_to_int_or_none(r.get("won")) or 0, - drawn=_to_int_or_none(r.get("drawn")) or 0, - lost=_to_int_or_none(r.get("lost")) or 0, - goals_for=_to_int_or_none(r.get("gf")) or 0, - goals_against=_to_int_or_none(r.get("ga")) or 0, - goal_diff=_to_int_or_none(r.get("gd")) or 0, - points=_to_int_or_none(r.get("pts")) or 0, - xg_for=_to_float_or_none(r.get("xgf")), - xg_against=_to_float_or_none(r.get("xga")), - form=r.get("form") or None, - zone=zone.get("label") or zone.get("key") or None, - updated_at=now, - retrieved_at=now, - ) - - # 同一联赛同一赛季只保留最新快照:按 (league, season, team) upsert - stmt = select(Standing).where( - Standing.league_id == league.id, - Standing.season == season_label, - Standing.team_id == team.id, - ) - standing = (await db.execute(stmt)).scalar_one_or_none() - if standing is None: - standing = Standing( - league_id=league.id, season=season_label, team_id=team.id, **values - ) - db.add(standing) - else: - for k, v in values.items(): - setattr(standing, k, v) - league_r["upserted"] += 1 - - league_r["rows"] = len(rows) - result["leagues"][code] = league_r - result["total_upserted"] += league_r["upserted"] - logger.info( - "bzzoiro standings 采集完成: %s 赛季 %s, upsert %d/%d", - code, season_label, league_r["upserted"], league_r["rows"], - ) - return result - - -# ============================================================ -# 统计回填管线:/events/{id}/stats/ → match_stats 表 -# ============================================================ - -# bzzoiro stats 字段 → MatchStats 字段映射(stats.home / stats.away 下) -_STATS_FIELD_MAP = { - "xg": ("home_xg", "away_xg"), # 回退 expected_goals - "ball_possession": ("home_possession", None), # 只取主队值,客队=100-home - "total_shots": ("home_shots", "away_shots"), - "shots_on_target": ("home_shots_on_target", "away_shots_on_target"), - "corner_kicks": ("home_corners", "away_corners"), - "yellow_cards": ("home_yellow_cards", "away_yellow_cards"), - "red_cards": ("home_red_cards", "away_red_cards"), - "big_chances": ("home_big_chances", "away_big_chances"), - "fouls": ("home_fouls", "away_fouls"), -} - - -def _pick(d: dict, *keys): - """按优先级取第一个非空字段值。""" - for k in keys: - v = d.get(k) - if v is not None: - return v - return None - - -def _stats_from_payload(payload: dict) -> dict: - """把 /events/{id}/stats/ 响应映射成 MatchStats 字段 dict。 - - 响应结构: {"event_id": ..., "stats": {"home": {...}, "away": {...}}} - """ - stats = (payload or {}).get("stats") or {} - home = stats.get("home") or {} - away = stats.get("away") or {} - out: dict = {} - - xg_h = _pick(home, "xg", "expected_goals") - xg_a = _pick(away, "xg", "expected_goals") - if xg_h is not None: - out["home_xg"] = _to_float_or_none(xg_h) - if xg_a is not None: - out["away_xg"] = _to_float_or_none(xg_a) - - poss = home.get("ball_possession") - if poss is not None: - p = _to_float_or_none(poss) - if p is not None: - out["home_possession"] = p - - for src, (h_fld, a_fld) in _STATS_FIELD_MAP.items(): - if src in ("xg", "ball_possession"): - continue # 已处理 - hv = home.get(src) - av = away.get(src) - if hv is not None and h_fld: - out[h_fld] = _to_int_or_none(hv) - if av is not None and a_fld: - out[a_fld] = _to_int_or_none(av) - return out - - -def _to_float_or_none(value) -> float | None: - if value is None: - return None - try: - return float(str(value).strip()) - except (TypeError, ValueError): - return None - - -async def ingest_bzzoiro_event_stats( - db, - *, - leagues: Iterable[str], - limit: int = 100, - only_missing: bool = True, -) -> dict: - """回填已完赛比赛的详细统计(逐场调 /events/{id}/stats/)。 - - 筛选条件: match_status=finished 且 source_event_id 非空。 - only_missing=True 时跳过已有统计的比赛(增量);False 则全量刷新。 - limit 控制单次最多处理的比赛数(上游限速 1.2s/请求,大批量需分次触发)。 - """ - result: dict = {"fetched": 0, "created": 0, "updated": 0, "skipped": 0, "errors": []} - - league_ids = [BZZOIRO_LEAGUE_IDS[c] for c in leagues if c in BZZOIRO_LEAGUE_IDS] - if not league_ids: - result["errors"].append("无有效联赛代码") - return result - - # D4: 候选比赛查询经 MatchRepository(含 stats 预加载,筛选/排序/limit 语义不变) - matches = await MatchRepository(db).find_finished_with_stats( - league_ids, limit=limit * 3 if only_missing else limit - ) - - now = datetime.now(timezone.utc) - processed = 0 - for m in matches: - if processed >= limit: - break - if only_missing and m.stats is not None and m.stats.home_shots is not None: - result["skipped"] += 1 - continue - processed += 1 - try: - payload = await _fetch_json_async(f"/events/{m.source_event_id}/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 _safe_write_ingest_failure( - db, - entity_type="match_stats", - source_record_id=str(m.source_event_id), - error=e, - raw_payload={"match_id": m.id}, - ) - await asyncio.sleep(REQUEST_INTERVAL) - continue - - result["fetched"] += 1 - fields = _stats_from_payload(payload) - if not fields: - result["skipped"] += 1 - await asyncio.sleep(REQUEST_INTERVAL) - continue - - if m.stats is None: - available_at = m.match_date + timedelta(hours=2) if m.match_date else now - m.stats = MatchStats( - match_id=m.id, - source="bzzoiro", - source_record_id=str(m.source_event_id), - retrieved_at=now, - available_at=available_at, - ) - db.add(m.stats) - result["created"] += 1 - else: - result["updated"] += 1 - if m.stats.source is None: - m.stats.source = "bzzoiro" - m.stats.source_record_id = str(m.source_event_id) - if m.stats.retrieved_at is None: - m.stats.retrieved_at = now - if m.stats.available_at is None and m.match_date: - m.stats.available_at = m.match_date + timedelta(hours=2) - - for fld, v in fields.items(): - if hasattr(m.stats, fld): - setattr(m.stats, fld, v) - - # 管线基础设施:写入 RawEvent + DataLineage - batch_id = f"bzzoiro-stats-{m.source_event_id}-{now.strftime('%Y%m%d%H%M%S')}" - try: - 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 # 基础设施写入失败不影响主流程 - - await asyncio.sleep(REQUEST_INTERVAL) - - logger.info( - "bzzoiro stats 回填完成: 抓取 %d, 新建 %d, 更新 %d, 跳过 %d, 错误 %d", - result["fetched"], result["created"], result["updated"], - result["skipped"], len(result["errors"]), - ) - return result +# ── 配置常量(原文件即从 config 再导出,维持 bz.REQUEST_INTERVAL 等引用) ── +from src.data.config import ( # noqa: F401 + BZZOIRO_LEAGUE_IDS, + LEAGUE_COUNTRIES, + LEAGUE_NAMES, + REQUEST_INTERVAL, +) +from src.data.key_ring import _mask # noqa: F401 (R1 测试引用 bz._mask) +from src.data.normalize import normalize_bzzoiro # noqa: F401 + +# ── 共享原语:HTTP 抓取 + 宽松字段转换 ── +from src.data.bzzoiro_common import ( # noqa: F401 + _fetch_json_async, + _match_key, + _to_date, + _to_float_or_none, + _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, + _events_record_id, + _write_events_bronze, + fetch_bzzoiro_events, +) + +# ── standings 管线 ── +from src.data.bzzoiro_standings import ( # noqa: F401 + _season_label_from_dates, + _write_standings_bronze, + fetch_bzzoiro_standings, + ingest_bzzoiro_standings, +) + +# ── stats 回填管线 ── +from src.data.bzzoiro_stats import ( # noqa: F401 + _pick, + _stats_from_payload, + ingest_bzzoiro_event_stats, +) diff --git a/src/data/bzzoiro_common.py b/src/data/bzzoiro_common.py new file mode 100644 index 0000000..f91b149 --- /dev/null +++ b/src/data/bzzoiro_common.py @@ -0,0 +1,123 @@ +"""bzzoiro 管线共享原语:HTTP 抓取(多 key 轮换)与宽松字段转换。 + +从 bzzoiro.py 拆出(单文件 → 多模块):仅放无业务语义的共享基础, +三条管线(events/standings/stats)与聚合门面见 bzzoiro.py。 +""" +from __future__ import annotations + +import asyncio +import logging +import random + +import httpx + +from src.core.http_client import get_client +from src.core.runtime_config import get_runtime_value +from src.data.key_ring import _mask, get_key_ring + +logger = logging.getLogger(__name__) + + +def _to_date(value): + """把 datetime / date / str 统一成 `date`。""" + if value is None: + return None + if hasattr(value, "date") and callable(value.date): + return value.date() + return value + + +def _to_int_or_none(value) -> int | None: + """宽松转 int(用于上游 ID 解析,失败返回 None 不抛错)。""" + if value is None: + return None + try: + return int(str(value).strip()) + except (TypeError, ValueError): + return None + + +def _to_float_or_none(value) -> float | None: + try: + return float(str(value).strip()) + except (TypeError, ValueError): + return None + + +def _match_key(home_team_id: int, away_team_id: int, match_date) -> tuple[int, int, str]: + """比赛去重键:(主队, 客队, 天级日期 ISO 字符串)。 + + 统一在这里构造,避免"预加载时用 str(date)、写入时用 isoformat()"这类 + 隐式格式依赖 —— 两者当前恰好相等,但一旦有人改动其一就会静默失配, + 导致所有比赛被判为不存在而重复插入。 + """ + d = _to_date(match_date) + return (home_team_id, away_team_id, d.isoformat() if d is not None else "") + + +async def _fetch_json_async(path: str, params: dict | None = None, max_retries: int = 3) -> dict | list: + """异步 HTTP(bzzoiro 使用 httpx,不再阻塞事件循环线程池)。 + + 多 key 轮换:遇到 429 自动切换到下一个 key;全部 key 冷却时等待最早恢复。 + """ + base = (await get_runtime_value("BZZOIRO_BASE")).rstrip("/") + raw_keys = await get_runtime_value("BZZOIRO_KEY") + ring = get_key_ring(base, raw_keys) + + url = f"{base}/{path.lstrip('/')}" + key = ring.get() + if not key: + raise RuntimeError("BZZOIRO_KEY 未设置") + + last_exc: Exception | None = None + for attempt in range(max_retries): + headers = { + "Authorization": f"Token {key}", + "Accept": "application/json", + } + try: + client = get_client() + # 整请求兜底: httpx 无 total 超时,用 wait_for 防「滴水式」限速挂死 + resp = await asyncio.wait_for( + client.get( + url, headers=headers, params=params, + timeout=httpx.Timeout(connect=10.0, read=30.0, write=10.0, pool=10.0), + ), + timeout=60.0, + ) + resp.raise_for_status() + return resp.json() + except Exception as e: + last_exc = e + status = getattr(getattr(e, "response", None), "status_code", None) + if status == 429: + # 限流:标记当前 key 冷却,切换到下一个 + new_key = ring.report_rate_limited(key) + if new_key and new_key != key: + logger.info("bzzoiro 429 → 切换 key: %s → %s,立即重试", _mask(key), _mask(new_key)) + key = new_key + continue # 立即重试,不等待 + # 单 key 或全部冷却:等待最早恢复的 key + wait = ring.wait_if_all_blocked() + if wait > 0: + logger.warning("bzzoiro 全部 key 冷却,等待 %.1fs 后重试", wait) + await asyncio.sleep(min(wait, 30.0)) + else: + delay = min(2 ** attempt, 16) + random.uniform(0, 1) + logger.warning("bzzoiro 429, retry %d in %.1fs", attempt + 1, delay) + await asyncio.sleep(delay) + key = ring.get() or key + continue + if 500 <= (status or 0) < 600: + delay = min(2 ** attempt, 16) + random.uniform(0, 1) + logger.warning("bzzoiro %d, retry %d in %.1fs", status, attempt + 1, delay) + await asyncio.sleep(delay) + continue + # 网络错误(连接失败/超时)也退避重试 + if isinstance(e, (TimeoutError, ConnectionError, OSError)): + delay = min(2 ** attempt, 16) + random.uniform(0, 1) + logger.warning("bzzoiro network error, retry %d in %.1fs: %s", attempt + 1, delay, e) + await asyncio.sleep(delay) + continue + raise + raise RuntimeError(f"bzzoiro request failed after {max_retries} attempts: {last_exc}") diff --git a/src/data/bzzoiro_events.py b/src/data/bzzoiro_events.py new file mode 100644 index 0000000..92231b7 --- /dev/null +++ b/src/data/bzzoiro_events.py @@ -0,0 +1,310 @@ +"""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.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 bz._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 + + # 查找已有比赛: 内存查找 + 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, + 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 + 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 同级约束)。 + """ + 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( + 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, + ) diff --git a/src/data/bzzoiro_standings.py b/src/data/bzzoiro_standings.py new file mode 100644 index 0000000..46f5edf --- /dev/null +++ b/src/data/bzzoiro_standings.py @@ -0,0 +1,215 @@ +"""bzzoiro standings 管线:联赛积分榜快照(/leagues/{id}/standings/)→ standings 表。 + +从 bzzoiro.py 拆出。同一联赛同一赛季只保留最新快照(按 (league, season, team) +upsert);球队名与 events 管线使用同一 normalize 规则,保证 Team 匹配。 + +可替换协作者(抓取函数 / Bronze 写入助手)在运行期经聚合门面 +src.data.bzzoiro 解析 —— 与拆分前的单文件 monkeypatch 语义一致。 +""" +from __future__ import annotations + +import logging +from collections.abc import Iterable +from datetime import datetime, timezone + +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.team_names_zh import zh_name +from src.db.models import Standing, Team +from src.db.repositories import LeagueRepository, TeamRepository + +logger = logging.getLogger(__name__) + + +async def fetch_bzzoiro_standings(league_code: str, season: str | None = None) -> dict: + """抓取联赛积分榜(纯抓取,不入库)。season 为 None 时取当前赛季。""" + 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}") + params: dict = {} + if season: + params["season"] = season + return await bz._fetch_json_async(f"/leagues/{league_id}/standings/", params) + + +def _season_label_from_dates(start_date, end_date) -> str: + """从赛季起止日期推导赛季标签(与 derive_season_label 语义一致)。""" + try: + if isinstance(start_date, str): + start = datetime.fromisoformat(start_date[:10]) + else: + start = start_date + if start is None: + return "?" + y = start.year + return f"{y}-{y + 1}" if start.month >= 8 else f"{y - 1}-{y}" + except (TypeError, ValueError): + return "?" + + +async def ingest_bzzoiro_standings(db, *, leagues: Iterable[str], season: str | None = None) -> dict: + """采集积分榜 → upsert standings 表。 + + 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 + + result: dict = {"leagues": {}, "total_upserted": 0, "errors": []} + for code in leagues: + league_r: dict = {"upserted": 0, "teams_created": 0, "rows": 0, "errors": []} + try: + payload = await bz.fetch_bzzoiro_standings(code, season=season) + 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( + db, + entity_type="standings", + source_record_id=None, + error=e, + raw_payload={"league": code, "season": season}, + ) + result["leagues"][code] = league_r + result["errors"].append(f"{code}: {e}") + continue + + rows = payload.get("standings") or [] + if not rows: + result["leagues"][code] = {"error": "无积分榜数据(赛季未开始或未提供)"} + result["errors"].append(f"{code}: 无积分榜数据") + continue + + # 联赛(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 {} + season_label = _season_label_from_dates( + season_obj.get("start_date"), season_obj.get("end_date") + ) + if season_label == "?": + season_label = season or "" + + # 批量预载球队(与 events 管线使用同一 normalize 规则,保证 Team 匹配) + names = {normalize_name(str(r.get("team_name", ""))) for r in rows} + names.discard("") + team_map: dict[str, Team] = await team_r.get_all_by_names(list(names)) + + now = datetime.now(timezone.utc) + for r in rows: + team_name = normalize_name(str(r.get("team_name", ""))) + if not team_name: + continue + team = team_map.get(team_name) + if team is None: + 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 + + zone = r.get("zone") or {} + values = dict( + position=_to_int_or_none(r.get("position")) or 0, + played=_to_int_or_none(r.get("played")) or 0, + won=_to_int_or_none(r.get("won")) or 0, + drawn=_to_int_or_none(r.get("drawn")) or 0, + lost=_to_int_or_none(r.get("lost")) or 0, + goals_for=_to_int_or_none(r.get("gf")) or 0, + goals_against=_to_int_or_none(r.get("ga")) or 0, + goal_diff=_to_int_or_none(r.get("gd")) or 0, + points=_to_int_or_none(r.get("pts")) or 0, + xg_for=_to_float_or_none(r.get("xgf")), + xg_against=_to_float_or_none(r.get("xga")), + form=r.get("form") or None, + zone=zone.get("label") or zone.get("key") or None, + updated_at=now, + retrieved_at=now, + ) + + # 同一联赛同一赛季只保留最新快照:按 (league, season, team) upsert + stmt = select(Standing).where( + Standing.league_id == league.id, + Standing.season == season_label, + Standing.team_id == team.id, + ) + standing = (await db.execute(stmt)).scalar_one_or_none() + if standing is None: + standing = Standing( + league_id=league.id, season=season_label, team_id=team.id, **values + ) + db.add(standing) + else: + for k, v in values.items(): + setattr(standing, k, v) + league_r["upserted"] += 1 + + league_r["rows"] = len(rows) + + # D1(对称 events/stats 管线): 联赛成功 upsert → 补写 Bronze 层。 + # 幂等键 standings:{league}:{season}:积分榜是联赛级快照,一次成功 + # 采集写一条 RawEvent(整份原始载荷)+ 一条血缘。season 用实际入库的 + # 标签(由载荷推导,与 Standing.season 同口径),不依赖调用方传参, + # 保证不同调用方(season=None 或显式传参)对同一赛季命中同一条 RawEvent。 + if league_r["upserted"] > 0: + bronze_batch_id = f"bzzoiro-standings-{code}-{now:%Y%m%d%H%M%S}" + await _write_standings_bronze( + db, + source_record_id=f"standings:{code}:{season_label}", + raw_payload=payload, + league_id=league.id, + league_code=code, + season_label=season_label, + rows_upserted=league_r["upserted"], + batch_id=bronze_batch_id, + ) + + result["leagues"][code] = league_r + result["total_upserted"] += league_r["upserted"] + logger.info( + "bzzoiro standings 采集完成: %s 赛季 %s, upsert %d/%d", + code, season_label, league_r["upserted"], league_r["rows"], + ) + return result + + +async def _write_standings_bronze( + db, + *, + source_record_id: str, + raw_payload: dict, + league_id: int | None, + league_code: str, + season_label: str, + rows_upserted: int, + batch_id: str, +) -> None: + """standings 成功 upsert 一个联赛后的 Bronze 层补写:RawEvent(幂等) + DataLineage。 + + 与 _write_events_bronze 同级约束:幂等性由 _write_raw_event 的 + source_record_id 查重保证(积分榜是联赛级快照,同联赛同赛季重复采集 + 命中同一条 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( + db, "bzzoiro", source_record_id, + "standings", league_id, "standings_ingest", + {"league": league_code, "season": season_label, "rows_upserted": rows_upserted}, + batch_id, + ) + except Exception: + logger.warning( + "standings Bronze 写入失败(record=%s, league=%s),不影响采集主流程", + source_record_id, league_code, exc_info=True, + ) diff --git a/src/data/bzzoiro_stats.py b/src/data/bzzoiro_stats.py new file mode 100644 index 0000000..134e69f --- /dev/null +++ b/src/data/bzzoiro_stats.py @@ -0,0 +1,180 @@ +"""bzzoiro stats 回填管线:已完赛比赛详细统计(/events/{id}/stats/)→ match_stats 表。 + +从 bzzoiro.py 拆出。上游限速(REQUEST_INTERVAL 秒/请求),大批量回填需分次触发; +只 add/flush 不 commit,事务由调用方 UnitOfWork 控制。 + +可替换协作者(_fetch_json_async / 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 _to_float_or_none, _to_int_or_none +from src.data.config import BZZOIRO_LEAGUE_IDS +from src.db.models import MatchStats +from src.db.repositories import MatchRepository + +logger = logging.getLogger(__name__) + + +# bzzoiro stats 字段 → MatchStats 字段映射(stats.home / stats.away 下) +_STATS_FIELD_MAP = { + "xg": ("home_xg", "away_xg"), # 回退 expected_goals + "ball_possession": ("home_possession", None), # 只取主队值,客队=100-home + "total_shots": ("home_shots", "away_shots"), + "shots_on_target": ("home_shots_on_target", "away_shots_on_target"), + "corner_kicks": ("home_corners", "away_corners"), + "yellow_cards": ("home_yellow_cards", "away_yellow_cards"), + "red_cards": ("home_red_cards", "away_red_cards"), + "big_chances": ("home_big_chances", "away_big_chances"), + "fouls": ("home_fouls", "away_fouls"), +} + + +def _pick(d: dict, *keys): + """按优先级取第一个非空字段值。""" + for k in keys: + v = d.get(k) + if v is not None: + return v + return None + + +def _stats_from_payload(payload: dict) -> dict: + """把 /events/{id}/stats/ 响应映射成 MatchStats 字段 dict。 + + 响应结构: {"event_id": ..., "stats": {"home": {...}, "away": {...}}} + """ + stats = (payload or {}).get("stats") or {} + home = stats.get("home") or {} + away = stats.get("away") or {} + out: dict = {} + + xg_h = _pick(home, "xg", "expected_goals") + xg_a = _pick(away, "xg", "expected_goals") + if xg_h is not None: + out["home_xg"] = _to_float_or_none(xg_h) + if xg_a is not None: + out["away_xg"] = _to_float_or_none(xg_a) + + poss = home.get("ball_possession") + if poss is not None: + p = _to_float_or_none(poss) + if p is not None: + out["home_possession"] = p + + for src, (h_fld, a_fld) in _STATS_FIELD_MAP.items(): + if src in ("xg", "ball_possession"): + continue # 已处理 + hv = home.get(src) + av = away.get(src) + if hv is not None and h_fld: + out[h_fld] = _to_int_or_none(hv) + if av is not None and a_fld: + out[a_fld] = _to_int_or_none(av) + return out + + +async def ingest_bzzoiro_event_stats( + db, + *, + leagues: Iterable[str], + limit: int = 100, + only_missing: bool = True, +) -> dict: + """回填已完赛比赛的详细统计(逐场调 /events/{id}/stats/)。 + + 筛选条件: match_status=finished 且 source_event_id 非空。 + only_missing=True 时跳过已有统计的比赛(增量);False 则全量刷新。 + limit 控制单次最多处理的比赛数(上游限速 1.2s/请求,大批量需分次触发)。 + """ + from src.data import bzzoiro as bz + + result: dict = {"fetched": 0, "created": 0, "updated": 0, "skipped": 0, "errors": []} + + league_ids = [BZZOIRO_LEAGUE_IDS[c] for c in leagues if c in BZZOIRO_LEAGUE_IDS] + if not league_ids: + result["errors"].append("无有效联赛代码") + return result + + # D4: 候选比赛查询经 MatchRepository(含 stats 预加载,筛选/排序/limit 语义不变) + matches = await MatchRepository(db).find_finished_with_stats( + league_ids, limit=limit * 3 if only_missing else limit + ) + + now = datetime.now(timezone.utc) + processed = 0 + for m in matches: + if processed >= limit: + break + if only_missing and m.stats is not None and m.stats.home_shots is not None: + result["skipped"] += 1 + continue + processed += 1 + try: + payload = await bz._fetch_json_async(f"/events/{m.source_event_id}/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( + db, + entity_type="match_stats", + source_record_id=str(m.source_event_id), + error=e, + raw_payload={"match_id": m.id}, + ) + await asyncio.sleep(bz.REQUEST_INTERVAL) + continue + + result["fetched"] += 1 + fields = _stats_from_payload(payload) + if not fields: + result["skipped"] += 1 + await asyncio.sleep(bz.REQUEST_INTERVAL) + continue + + if m.stats is None: + available_at = m.match_date + timedelta(hours=2) if m.match_date else now + m.stats = MatchStats( + match_id=m.id, + source="bzzoiro", + source_record_id=str(m.source_event_id), + retrieved_at=now, + available_at=available_at, + ) + db.add(m.stats) + result["created"] += 1 + else: + result["updated"] += 1 + if m.stats.source is None: + m.stats.source = "bzzoiro" + m.stats.source_record_id = str(m.source_event_id) + if m.stats.retrieved_at is None: + m.stats.retrieved_at = now + if m.stats.available_at is None and m.match_date: + m.stats.available_at = m.match_date + timedelta(hours=2) + + for fld, v in fields.items(): + if hasattr(m.stats, fld): + setattr(m.stats, fld, v) + + # 管线基础设施:写入 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) + except Exception: + pass # 基础设施写入失败不影响主流程 + + await asyncio.sleep(bz.REQUEST_INTERVAL) + + logger.info( + "bzzoiro stats 回填完成: 抓取 %d, 新建 %d, 更新 %d, 跳过 %d, 错误 %d", + result["fetched"], result["created"], result["updated"], + result["skipped"], len(result["errors"]), + ) + return result diff --git a/src/data/pipeline_write.py b/src/data/pipeline_write.py new file mode 100644 index 0000000..745f66e --- /dev/null +++ b/src/data/pipeline_write.py @@ -0,0 +1,80 @@ +"""管线基础设施写入助手:RawEvent(Bronze 原始载荷)/ IngestFailure(死信)/ DataLineage(血缘)。 + +从 bzzoiro.py 拆出。约定(与拆分前一致): + - 只 add 不 commit —— 事务由调用方 UnitOfWork 控制,分批事务约定不变; + - 死信与 Bronze 写入同为 best-effort:失败只记 warning,绝不拖垮采集主流程。 +""" +from __future__ import annotations + +import logging + +from src.db.models import DataLineage, IngestFailure, RawEvent + +logger = logging.getLogger(__name__) + + +async def _write_raw_event(db, source_system: str, source_record_id: str, raw_payload: dict, batch_id: str | None = None) -> None: + """写入 Bronze 层原始事件(幂等:同 source_record_id 跳过)。""" + from sqlalchemy import select as _select + stmt = _select(RawEvent).where( + RawEvent.source_system == source_system, + RawEvent.source_record_id == source_record_id, + ) + existing = (await db.execute(stmt)).scalar_one_or_none() + if existing is None: + db.add(RawEvent( + source_system=source_system, + source_record_id=source_record_id, + raw_payload=raw_payload, + ingest_batch_id=batch_id, + )) + + +async def _write_ingest_failure(db, source_system: str, entity_type: str, source_record_id: str | None, error_type: str, error_detail: str | None, raw_payload: dict | None = None) -> None: + """写入采集失败死信。""" + db.add(IngestFailure( + source_system=source_system, + entity_type=entity_type, + source_record_id=source_record_id, + error_type=error_type, + error_detail=error_detail, + raw_payload=raw_payload, + )) + + +async def _safe_write_ingest_failure( + db, + *, + entity_type: str, + source_record_id: str | None, + error: Exception, + raw_payload: dict | None = None, +) -> None: + """抓取失败时尽力写入死信表(失败不影响主流程)。 + + 死信是「可观测性」基础设施,与 RawEvent/Lineage 同级:写入失败只记 + warning,绝不能让原始抓取错误之外的新异常打断采集循环。 + """ + try: + await _write_ingest_failure( + db, "bzzoiro", entity_type, source_record_id, + "fetch_error", str(error), raw_payload, + ) + except Exception: + logger.warning( + "写入 ingest_failures 死信失败(entity=%s, record=%s): %s", + entity_type, source_record_id, error, exc_info=True, + ) + + +async def _write_lineage(db, source_system: str, source_record_id: str, target_table: str, target_id: int | None, transform_name: str, transform_detail: dict | None = None, batch_id: str | None = None) -> None: + """写入 ETL 血缘追踪。""" + db.add(DataLineage( + source_system=source_system, + source_record_id=source_record_id, + target_table=target_table, + target_id=target_id, + transform_name=transform_name, + transform_detail=transform_detail, + batch_id=batch_id, + )) diff --git a/tests/test_available_at.py b/tests/test_available_at.py index 89d6ebd..a5adffb 100644 --- a/tests/test_available_at.py +++ b/tests/test_available_at.py @@ -79,18 +79,18 @@ class TestWriteBufferStrategy: def test_bzzoirot_new_match_available_at_is_two_hours_after_kickoff(self): """bzzoiro 新建比赛(stats 回填创建 MatchStats)时 available_at 应为开球 + 2 小时。""" import inspect - from src.data import bzzoiro + from src.data import bzzoiro_stats - source = inspect.getsource(bzzoiro) + source = inspect.getsource(bzzoiro_stats) assert 'timedelta(hours=2)' in source, \ "bzzoiro 应使用 match_date + timedelta(hours=2) 作为 available_at" def test_bzzoirot_multiple_writes_use_two_hour_buffer(self): """bzzoiro 多处写入(创建/更新)都应使用 2 小时缓冲。""" import inspect - from src.data import bzzoiro + from src.data import bzzoiro_stats - source = inspect.getsource(bzzoiro) + source = inspect.getsource(bzzoiro_stats) count = source.count('timedelta(hours=2)') assert count >= 2, f"期望至少 2 处 timedelta(hours=2),实际 {count} 处" diff --git a/tests/test_regressions.py b/tests/test_regressions.py index 4ff3601..1bc1311 100644 --- a/tests/test_regressions.py +++ b/tests/test_regressions.py @@ -175,7 +175,7 @@ class TestBzzoiroLineage: return bad def test_normalized_matches_carries_raw(self): - src = _read("data/bzzoiro.py") + src = _read("data/bzzoiro_events.py") # 规范化结果必须与原始 event 成对保存 assert "normalized_matches.append((nm, raw))" in src, ( "normalized_matches 未携带 (nm, raw) 元组 —— raw 变量泄漏会回归 (P0-3)" @@ -193,7 +193,7 @@ class TestBzzoiroLineage: 它不是 `raw.get(` 同一行,但同样正确。非法写法(回归)是直接 `existing_match.source_event_id = orphan_var`。 """ - src = _read("data/bzzoiro.py") + src = _read("data/bzzoiro_events.py") seg = self._consume_loop_body(src) bad = self._bad_assignments(seg) assert len(bad) == 0, ( @@ -206,7 +206,7 @@ class TestBzzoiroLineage: 下游 `_backfill_stats` 里合法地在 ORM 对象上访问 `m.source_event_id` (与配对 raw 无关)。若 seg 越界,test_no_orphan_raw_use 会误报。 """ - src = _read("data/bzzoiro.py") + src = _read("data/bzzoiro_events.py") seg = self._consume_loop_body(src) assert "m.source_event_id" not in seg, ( "循环体截取越界,扫到了下游 stats 管线 —— 会误报 P0-3"