diff --git a/alembic/versions/0018_match_checks.py b/alembic/versions/0018_match_checks.py new file mode 100644 index 0000000..c1bbc79 --- /dev/null +++ b/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') diff --git a/src/data/bzzoiro.py b/src/data/bzzoiro.py index 5942c64..4fb345e 100644 --- a/src/data/bzzoiro.py +++ b/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( diff --git a/src/db/models.py b/src/db/models.py index 4f0a72e..bafff7a 100644 --- a/src/db/models.py +++ b/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")