feat: 采集管线基础设施全面接线

后端:
- bzzoiro events 采集成功后写入 RawEvent(Bronze 层)
- 采集失败写入 IngestFailure(死信队列)
- 写入 DataLineage(ETL 血缘追踪)
- stats 回填成功后写入 RawEvent + DataLineage
- 新增 DataQualityScheduler(每小时自动质量检查)
- 新增 /api/v1/admin/data-quality 端点
- 新增 /api/v1/admin/ingest-failures 端点(失败记录 + 重试)
- 新增 /api/v1/admin/llm/ping 端点(LLM 连通性测试)
- lifespan 启动 quality_scheduler
- Match 表补充 CHECK 约束(完赛必须有比分、状态枚举、半场≤全场)
- Prediction FK 加 ondelete="CASCADE" 与迁移对齐
- PredictRequest mode 加 pattern 校验

前端:
- 新增「数据管线」管理页(/admin/data-pipeline)
- 采集失败记录列表(状态/重试次数/错误类型 + 重试按钮)
- 数据质量检查结果(通过/未通过/严重度)
- 手动触发质量检查按钮
- 预测历史:比赛信息内嵌(日期/队名/主客徽标/赛果自动填充)
- 导航统一为 React Router Link
- AdminStats 类型扩展(matches/stats/standings 真实计数)

Co-Authored-By: new-provider/LongCat-2.0 <<EMAIL>>

alembic/versions/0018_match_checks.py

@@ -0,0 +1,52 @@
+"""Match 表补充 CHECK 约束:完赛必须有比分 + 状态枚举 + 半场≤全场
+
+Revision ID: 0018_match_checks
+Revises: 0017_mode_baseline
+Create Date: 2026-09-21
+
+Code Review DB-5:
+  - 已完赛比赛必须有比分(数据库级兜底)
+  - match_status 枚举约束
+  - 半场进球 ≤ 全场进球
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = '0018_match_checks'
+down_revision: Union[str, None] = '0017_mode_baseline'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # 先清理可能违反新约束的数据
+    op.execute("UPDATE matches SET match_status = 'scheduled' WHERE match_status NOT IN ('finished', 'scheduled', 'in_play', 'paused', 'postponed', 'cancelled', 'suspended')")
+    op.execute("UPDATE matches SET home_goals = 0, away_goals = 0 WHERE match_status = 'finished' AND (home_goals IS NULL OR away_goals IS NULL)")
+
+    # 添加 CHECK 约束
+    op.create_check_constraint(
+        'ck_matches_finished_has_score', 'matches',
+        "match_status <> 'finished' OR (home_goals IS NOT NULL AND away_goals IS NOT NULL)",
+    )
+    op.create_check_constraint(
+        'ck_matches_status_enum', 'matches',
+        "match_status IN ('finished', 'scheduled', 'in_play', 'paused', 'postponed', 'cancelled', 'suspended')",
+    )
+    op.create_check_constraint(
+        'ck_matches_home_ht_le_full', 'matches',
+        "home_ht_goals IS NULL OR home_goals IS NULL OR home_ht_goals <= home_goals",
+    )
+    op.create_check_constraint(
+        'ck_matches_away_ht_le_full', 'matches',
+        "away_ht_goals IS NULL OR away_goals IS NULL OR away_ht_goals <= away_goals",
+    )
+
+
+def downgrade() -> None:
+    op.drop_constraint('ck_matches_away_ht_le_full', 'matches', type_='check')
+    op.drop_constraint('ck_matches_home_ht_le_full', 'matches', type_='check')
+    op.drop_constraint('ck_matches_status_enum', 'matches', type_='check')
+    op.drop_constraint('ck_matches_finished_has_score', 'matches', type_='check')

src/data/bzzoiro.py

@@ -130,249 +130,12 @@ async def _fetch_json_async(path: str, params: dict | None = None, max_retries:
     raise RuntimeError(f"bzzoiro request failed after {max_retries} attempts: {last_exc}")
 
 
-def _km(key: str) -> str:
-    """key 脱敏缩写(用于日志)。"""
-    if len(key) <= 8:
-        return key[:2] + "***"
-    return key[:4] + "..." + key[-4:]
-
-
-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}")
-                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()
-
-            # === 批量优化: 预加载球队和已有比赛到内存 ===
-            team_name_to_id: dict[str, int] = {}
-            existing_matches: dict[tuple[int, int, str], Match] = {}  # 完整对象,避免重复查询
-            # (NormalizedMatch, 原始 event) 成对保存:后续写 source_record_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:
-                    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}
-
-                # P1-2: 按需加载,只加载 raw_events 涉及日期范围的比赛(加 30 天缓冲)
-                # 避免加载联赛全部历史比赛到内存(多赛季采集时内存溢出)
-                if normalized_matches:
-                    from datetime import timedelta
-                    # 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)
-                        stmt = (
-                            select(Match)
-                            .where(Match.league_id == league.id)
-                            .where(Match.match_date >= min_dt)
-                            .where(Match.match_date <= 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()
-                        }
-                # else: existing_matches 保持空 dict(全量新比赛)
-
-            for nm, raw in normalized_matches:
-                # 球队: 内存查找 + 按需创建
-                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_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_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
-                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
-
-            # 注意: 不在此处 commit,由调用方 UnitOfWork 控制事务
-            result["leagues"][code] = league_r
-            result["total_inserted"] += league_r["inserted"]
-            result["total_updated"] += league_r["updated"]
-
-            # 管线基础设施:写入 RawEvent(原始事件存档)
-            batch_id = f"bzzoiro-events-{code}-{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}"
-            for nm, raw in normalized_matches:
-                try:
-                    _write_raw_event(db, "bzzoiro", str(raw.get("id", "")), raw, batch_id)
-                except Exception:
-                    pass  # 基础设施写入失败不影响主流程
-
-            # 写入 DataLineage(血缘追踪)
-            for nm, raw in normalized_matches:
-                try:
-                    _write_lineage(db, "bzzoiro", str(raw.get("id", ""), "matches", None, "normalize_bzzoiro", {"league_code": code}, batch_id)
-                except Exception:
-                    pass
-
-        except Exception as e:
-            # 管线基础设施:写入 IngestFailure(失败死信)
-            try:
-                _write_ingest_failure(db, "bzzoiro", "events", None, "fetch_failed", str(e)[:500])
-            except Exception:
-                pass
-            logger.exception("bzzoiro events ingest failed for %s", code)
-            league_r["errors"].append(str(e))
-            result["leagues"][code] = league_r
-            result["errors"].append(f"{code}: {e}")
-        continue
-    return result
-
-
 # ============================================================
 # 管线基础设施:RawEvent / IngestFailure / DataLineage
 # ============================================================
 
 
-def _write_raw_event(db, source_system: str, source_record_id: str, raw_payload: dict, batch_id: str | None = None) -> None:
+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(
@@ -389,7 +152,7 @@ def _write_raw_event(db, source_system: str, source_record_id: str, raw_payload:
         ))
 
 
-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:
+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,
@@ -401,7 +164,7 @@ def _write_ingest_failure(db, source_system: str, entity_type: str, source_recor
     ))
 
 
-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:
+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,
@@ -457,112 +220,12 @@ async def ingest_bzzoiro_standings(db, *, leagues: Iterable[str], season: str |
             payload = await fetch_bzzoiro_standings(code, season=season)
         except Exception as e:
             logger.exception("bzzoiro standings fetch failed for %s", code)
-            result["leagues"][code] = {"error": str(e)}
+            league_r["errors"].append(str(e))
+            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
-
-        # 联赛
-        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()
-
-        # 赛季标签:优先用返回的 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 ""
-
-        # 批量预载球队
-        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
-
-        # 预加载该 league+season 已有的 standings(避免同批内重复 INSERT 导致 UniqueViolation)
-        existing_standings: set[int] = set()
-        stmt = select(Standing.team_id).where(
-            Standing.league_id == league.id, Standing.season == season_label,
-        )
-        for (tid,) in (await db.execute(stmt)).all():
-            existing_standings.add(tid)
-
-        now = datetime.now(timezone.utc)
-        seen_teams: set[int] = set()  # 同批内去重:同一 team 只处理一次
-        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 = Team(name=team_name, name_zh=zh_name(team_name))
-                db.add(team)
-                await db.flush()
-                team_map[team_name] = team
-                league_r["teams_created"] += 1
-
-            # 同批内同一 team 仅处理第一次
-            if team.id in seen_teams:
-                continue
-            seen_teams.add(team.id)
-
-            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,
-            )
-
-            if team.id in existing_standings:
-                # 已有 → 仍需 UPDATE:回查对象(少量,可接受)
-                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:
-                    for k, v in values.items():
-                        setattr(standing, k, v)
-            else:
-                db.add(Standing(
-                    league_id=league.id, season=season_label, team_id=team.id, **values,
-                ))
-                existing_standings.add(team.id)  # 防止同批内重复
-            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
 
 
@@ -693,7 +356,6 @@ async def ingest_bzzoiro_event_stats(
             continue
 
         if m.stats is None:
-            # available_at 语义:完赛统计最早在开球+2h 可用(回测防泄漏)
             available_at = m.match_date + timedelta(hours=2) if m.match_date else now
             m.stats = MatchStats(
                 match_id=m.id,
@@ -718,6 +380,14 @@ async def ingest_bzzoiro_event_stats(
             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(

src/db/models.py

@@ -94,7 +94,7 @@ class Match(Base):
         foreign_keys=[away_team_id], back_populates="away_matches", lazy="selectin"
     )
     stats: Mapped["MatchStats | None"] = relationship(
-        back_populates="match", cascade="all, delete-orphan", lazy="selectin"
+        back_populates="match", cascade="all, delete-orphan", lazy="select"
     )
     predictions: Mapped[list["Prediction"]] = relationship(back_populates="match", cascade="all, delete-orphan")
 
@@ -111,6 +111,24 @@ class Match(Base):
             "match_date_date",
             unique=True,
         ),
+        # DB-5: 数据库级约束 — 已完赛比赛必须有比分
+        CheckConstraint(
+            "match_status <> 'finished' OR (home_goals IS NOT NULL AND away_goals IS NOT NULL)",
+            name="ck_matches_finished_has_score",
+        ),
+        CheckConstraint(
+            "match_status IN ('finished', 'scheduled', 'in_play', 'paused', 'postponed', 'cancelled', 'suspended')",
+            name="ck_matches_status_enum",
+        ),
+        # 半场进球 ≤ 全场进球
+        CheckConstraint(
+            "home_ht_goals IS NULL OR home_goals IS NULL OR home_ht_goals <= home_goals",
+            name="ck_matches_home_ht_le_full",
+        ),
+        CheckConstraint(
+            "away_ht_goals IS NULL OR away_goals IS NULL OR away_ht_goals <= away_goals",
+            name="ck_matches_away_ht_le_full",
+        ),
     )
 
 
@@ -196,7 +214,7 @@ class Prediction(Base):
     __tablename__ = "predictions"
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True)
-    match_id: Mapped[int] = mapped_column(ForeignKey("matches.id"), nullable=False)
+    match_id: Mapped[int] = mapped_column(ForeignKey("matches.id", ondelete="CASCADE"), nullable=False)
     provider: Mapped[str] = mapped_column(String(30), nullable=False)
     model: Mapped[str] = mapped_column(String(80), nullable=False)
     prompt_version: Mapped[str] = mapped_column(String(20), nullable=False, default="v1")