From 66f08447987f16a80114b43361adac27057024c0 Mon Sep 17 00:00:00 2001 From: shangfangjian Date: Mon, 21 Sep 2026 23:28:46 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20standings=20Bronze=20=E8=A1=80=E7=BC=98?= =?UTF-8?q?=E8=A1=A5=E9=BD=90=20+=20=E9=98=9F=E5=90=8D=E5=BD=92=E4=B8=80?= =?UTF-8?q?=20MVP?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit standings 成功 upsert 后写入 RawEvent(source_record_id=standings:{league}:{season}) + DataLineage(target_table=standings),与 events/stats 管线对称。 队名归一化收敛到 TeamRepository.get_or_create 唯一咽喉点, 创建新 Team 时 info 日志打出原始名与归一后名; Admin 新增 GET /api/v1/admin/team-name-duplicates 只读接口, 启发式列出近似重名候选(大小写变体/子串/前缀碰撞),不做自动合并。 --- docs/05-data.md | 30 +++ src/api/routes/admin_quality.py | 343 ++++++++++++++++++++++++++++ src/db/repositories.py | 22 +- tests/test_review_required_fixes.py | 7 +- tests/test_standings_bronze.py | 288 +++++++++++++++++++++++ 5 files changed, 685 insertions(+), 5 deletions(-) create mode 100644 src/api/routes/admin_quality.py create mode 100644 tests/test_standings_bronze.py diff --git a/docs/05-data.md b/docs/05-data.md index e648bf0..e1525dd 100644 --- a/docs/05-data.md +++ b/docs/05-data.md @@ -63,6 +63,36 @@ `src/data/team_names.py` 维护 `NORMALIZE_MAP`(如 `Man City` → `Manchester City`),未命中映射的队名原样返回。 归一前先做 Unicode NFKD 去重音。 +**唯一键是归一后英文名**:`teams.name` 带 `UNIQUE` 约束,所有入库路径均经 `TeamRepository.get_or_create` 收敛归一化 +(events / standings 管线在调用前归一,仓库层再做一次幂等归一作为兜底)。创建新 Team 时打 `info` 日志记录「原始名 → 归一后名」。 + +> ⚠️ **`normalize` 当前大小写敏感**:仅当入参大小写与 `NORMALIZE_MAP` 键完全匹配时才触发映射 +>(如 `"Man City"` → `"Manchester City"`,但 `"man city"` 原样保留)。上游 bzzoiro 返回的队名首字母大写, +>实际命中无问题;若新增数据源返回全小写/全大写队名,需先 `title()` 再归一,否则会绕过映射产生重复 Team。 + +**改名 / 合并流程**(人工): + +当发现两个 `teams` 行实际是同一球队(如 `Manchester City` 与 `Man City` 因历史数据大小写差异各占一行): + +1. 确定**保留行**(通常选归一后规范名、且被更多 Match 引用的那行)。 +2. 将被删行的所有引用指向保留行(`UPDATE matches SET home_team_id = 保留id WHERE home_team_id = 删行id`,客场同理; + `standings` / `match_stats` 按 `team_id` 同理)。 +3. 删掉多余行:`DELETE FROM teams WHERE id = 删行id`。 + +> 此过程引入外键约束风险,务必在事务中执行并先 `BEGIN; ... ` 验证行数后再 `COMMIT`。 +> 暂不做自动合并(避免误合相似名),仅通过下方 Admin 接口列出「近似重名」候选,由人工判定。 + +## Admin:近似重名候选 + +`GET /api/v1/admin/team-name-duplicates` 只读列出启发式相似候选(大小写差异、子串包含、前缀碰撞),不做自动合并。 +典型用途:定期巡检,发现候选后走上方人工 SQL 合并。启发式规则: + +- **大小写变体**:`lower(name)` 相同但 `name` 不同(如 `Arsenal FC` / `arsenal fc`)。 +- **子串包含**:A 是 B 的子串且长度 ≥ 5(如 `Manchester` / `Manchester City`)。 +- **前缀碰撞**:前 8 个字符相同的两队。 + +命中任一规则即列为候选,按相似度分组返回。 + ## 数据库 Schema 12 张表:核心业务表 5 张见下方 DDL,其余 7 张(积分榜/配置/调度/治理)见后文表格。 diff --git a/src/api/routes/admin_quality.py b/src/api/routes/admin_quality.py new file mode 100644 index 0000000..d2abfdc --- /dev/null +++ b/src/api/routes/admin_quality.py @@ -0,0 +1,343 @@ +"""后台管理:管理区统计、数据完整性分析、数据质量检查。 + +所有接口需管理员鉴权(require_admin)。路由前缀 /api/v1/admin。 +""" +from __future__ import annotations + +import logging +from datetime import datetime, timedelta, timezone + +from fastapi import APIRouter, Depends +from sqlalchemy import func, select + +from src.api.deps import require_admin +from src.db.base import AsyncSession, get_db_read +from src.db.models import DataQualityCheck, IngestFailure, League, Match, MatchStats, Prediction, Standing + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/api/v1/admin", tags=["admin"], dependencies=[Depends(require_admin)]) + + +@router.get("/stats") +async def admin_stats(db: AsyncSession = Depends(get_db_read)): + """管理区统计(只读):预测次数 + 比赛覆盖。轻量聚合,无 LLM 调用。""" + day_ago = datetime.now(timezone.utc) - timedelta(days=1) + week_ago = datetime.now(timezone.utc) - timedelta(days=7) + r = ( + await db.execute( + select( + func.count().label("total"), + func.count().filter(Prediction.created_at >= day_ago).label("last_24h"), + func.count().filter(Prediction.created_at >= week_ago).label("last_7d"), + ) + ) + ).one() + # F3 修复: 补充真实比赛计数(非 limit=100 近似) + match_cnt = (await db.execute(select(func.count()).select_from(Match))).scalar() or 0 + finished_cnt = (await db.execute(select(func.count()).where(Match.match_status == "finished"))).scalar() or 0 + stats_cnt = (await db.execute(select(func.count()).select_from(MatchStats))).scalar() or 0 + standings_cnt = (await db.execute(select(func.count()).select_from(Standing))).scalar() or 0 + return { + "predictions": {"total": r.total, "last_24h": r.last_24h, "last_7d": r.last_7d}, + "matches": {"total": match_cnt, "finished": finished_cnt}, + "stats": {"total": stats_cnt}, + "standings": {"total": standings_cnt}, + } + + +# ── 数据完整性分析(可视化数据源) ──────────────────────────────── + + +@router.get("/data-completeness") +async def data_completeness(db: AsyncSession = Depends(get_db_read)): + """按联赛统计数据完整性:比赛覆盖、字段覆盖、积分榜覆盖。 + + 前端「数据完整性」页据此渲染,回答三个问题: + 1. 数据是否齐全(各联赛比赛/统计/积分榜量级) + 2. 字段是否齐全(每张统计表各字段非空率) + 3. 覆盖是否新鲜(最近一场/最近一次采集) + """ + from src.data.config import BZZOIRO_LEAGUE_IDS, LEAGUE_NAMES, LEAGUE_COUNTRIES + + out_leagues: list[dict] = [] + for code, bzz_id in BZZOIRO_LEAGUE_IDS.items(): + # 比赛覆盖 + m = ( + await db.execute( + select( + func.count().label("total"), + func.count().filter(Match.match_status == "finished").label("finished"), + func.count().filter(Match.match_status == "scheduled").label("scheduled"), + func.count().filter(Match.source_event_id.is_not(None)).label("with_source_id"), + func.max(Match.match_date).label("latest_match"), + func.min(Match.match_date).label("earliest_match"), + ) + .select_from(Match) + .join(League, League.id == Match.league_id) + .where(League.code == code) + ) + ).one() + # 统计字段覆盖(联表 matches) + s = ( + await db.execute( + select( + func.count().label("rows"), + func.count(MatchStats.home_xg).label("xg"), + func.count(MatchStats.home_shots).label("shots"), + func.count(MatchStats.home_possession).label("possession"), + func.count(MatchStats.home_corners).label("corners"), + func.count(MatchStats.home_fouls).label("fouls"), + func.count(MatchStats.home_big_chances).label("big_chances"), + func.count(MatchStats.home_yellow_cards).label("cards"), + ) + .select_from(MatchStats) + .join(Match, Match.id == MatchStats.match_id) + .join(League, League.id == Match.league_id) + .where(League.code == code) + ) + ).one() + # 积分榜覆盖 + st = ( + await db.execute( + select( + func.count().label("rows"), + func.max(Standing.retrieved_at).label("latest_retrieved"), + ) + .select_from(Standing) + .join(League, League.id == Standing.league_id) + .where(League.code == code) + ) + ).one() + + stats_rows = s.rows or 0 + pct = lambda n: round(n / stats_rows * 100, 1) if stats_rows else 0.0 # noqa: E731 + out_leagues.append( + { + "code": code, + "name": LEAGUE_NAMES.get(code, code), + "country": LEAGUE_COUNTRIES.get(code), + "matches": { + "total": m.total or 0, + "finished": m.finished or 0, + "scheduled": m.scheduled or 0, + "with_source_id": m.with_source_id or 0, + "earliest_match": m.earliest_match.isoformat() if m.earliest_match else None, + "latest_match": m.latest_match.isoformat() if m.latest_match else None, + }, + "stats": { + "rows": stats_rows, + "fields": { + "xg": {"count": s.xg or 0, "pct": pct(s.xg or 0)}, + "shots": {"count": s.shots or 0, "pct": pct(s.shots or 0)}, + "possession": {"count": s.possession or 0, "pct": pct(s.possession or 0)}, + "corners": {"count": s.corners or 0, "pct": pct(s.corners or 0)}, + "fouls": {"count": s.fouls or 0, "pct": pct(s.fouls or 0)}, + "big_chances": {"count": s.big_chances or 0, "pct": pct(s.big_chances or 0)}, + "cards": {"count": s.cards or 0, "pct": pct(s.cards or 0)}, + }, + }, + "standings": { + "rows": st.rows or 0, + "latest_retrieved": st.latest_retrieved.isoformat() if st.latest_retrieved else None, + }, + } + ) + + # 整体健康信号 + total_finished = sum(l["matches"]["finished"] for l in out_leagues) + total_stats = sum(l["stats"]["rows"] for l in out_leagues) + stats_coverage = round(total_stats / total_finished * 100, 1) if total_finished else 0.0 + issues: list[str] = [] + for l in out_leagues: + if l["matches"]["finished"] == 0: + issues.append(f"{l['name']}: 无已完赛比赛,请先运行「比赛数据」采集") + elif l["stats"]["rows"] == 0: + issues.append(f"{l['name']}: 已完赛 {l['matches']['finished']} 场但无统计回填,请运行「统计回填」采集") + elif stats_coverage < 80: + issues.append(f"{l['name']}: 统计覆盖率仅 {stats_coverage}%,建议增量回填") + if l["standings"]["rows"] == 0: + issues.append(f"{l['name']}: 无积分榜数据,请运行「积分榜」采集") + if not issues: + issues.append("各联赛数据完整度良好") + + return { + "generated_at": datetime.now(timezone.utc).isoformat(), + "leagues": out_leagues, + "totals": { + "finished_matches": total_finished, + "stats_rows": total_stats, + "stats_coverage_pct": stats_coverage, + }, + "issues": issues, + } + + +# ── 数据质量检查 API ──────────────────────────────────────────── + + +@router.get("/data-quality") +async def data_quality_checks(db: AsyncSession = Depends(get_db_read)): + """数据质量检查结果(只读)。""" + # 最近的失败记录 + failures = ( + await db.execute( + select(IngestFailure) + .where(IngestFailure.status.in_(["pending", "retrying"])) + .order_by(IngestFailure.created_at.desc()) + .limit(20) + ) + ).scalars().all() + + # 最近的质量检查 + checks = ( + await db.execute( + select(DataQualityCheck) + .order_by(DataQualityCheck.checked_at.desc()) + .limit(20) + ) + ).scalars().all() + + return { + "failures": [ + { + "id": f.id, + "source": f.source_system, + "entity_type": f.entity_type, + "source_record_id": f.source_record_id, + "error_type": f.error_type, + "error_detail": f.error_detail, + "retry_count": f.retry_count, + "status": f.status, + "created_at": f.created_at.isoformat() if f.created_at else None, + } + for f in failures + ], + "checks": [ + { + "id": c.id, + "check_name": c.check_name, + "entity_type": c.entity_type, + "passed": c.passed, + "severity": c.severity, + "detail": c.detail, + "checked_at": c.checked_at.isoformat() if c.checked_at else None, + } + for c in checks + ], + } + + +@router.post("/data-quality/run") +async def run_data_quality_check(db: AsyncSession = Depends(get_db_read)): + """手动触发一次数据质量检查。""" + checks = [] + + # 检查1: 已完赛但无统计的比赛 + finished_no_stats = ( + await db.execute( + select(func.count()) + .select_from(Match) + .outerjoin(MatchStats, Match.id == MatchStats.match_id) + .where(Match.match_status == "finished") + .where(MatchStats.id.is_(None)) + ) + ).scalar() or 0 + + checks.append(DataQualityCheck( + check_name="finished_without_stats", + entity_type="match", + actual_value=float(finished_no_stats), + passed=finished_no_stats == 0, + severity="warning" if finished_no_stats > 0 else "info", + detail={"message": f"{finished_no_stats} 场已完赛比赛缺少统计数据"}, + )) + + # 检查2: 积分榜缺失的联赛 + leagues_without_standings = ( + await db.execute( + select(func.count()) + .select_from(League) + .outerjoin(Standing, League.id == Standing.league_id) + .where(Standing.id.is_(None)) + ) + ).scalar() or 0 + + checks.append(DataQualityCheck( + check_name="league_without_standings", + entity_type="league", + actual_value=float(leagues_without_standings), + passed=leagues_without_standings == 0, + severity="warning" if leagues_without_standings > 0 else "info", + detail={"message": f"{leagues_without_standings} 个联赛缺少积分榜"}, + )) + + for c in checks: + db.add(c) + await db.commit() + + return {"ok": True, "checks": [{"name": c.check_name, "passed": c.passed} for c in checks]} + + +# ── 近似重名候选(只读,启发式,不做自动合并) ────────────────────── + + +@router.get("/team-name-duplicates") +async def team_name_duplicates(db: AsyncSession = Depends(get_db_read)): + """只读列出近似重名候选(大小写变体/子串包含/前缀碰撞)。 + + 启发式规则(命中任一即列为候选): + - 大小写变体: lower(name) 相同但 name 不同 + - 子串包含: A 是 B 的子串且 len(A) ≥ 5 + - 前缀碰撞: 前 8 字符相同(忽略大小写) + + 仅作排查参考,合并需走人工 SQL(见 docs/05-data.md)。 + """ + teams = (await db.execute(select(Team.id, Team.name))).all() + by_lower: dict[str, list[dict]] = {} + for t in teams: + key = (t.name or "").lower() + by_lower.setdefault(key, []).append({"id": t.id, "name": t.name}) + + groups: list[dict] = [] + + # 规则1: 大小写变体(lower 相同但原名不同) + for key, members in by_lower.items(): + if len(members) > 1: + groups.append({ + "rule": "case_variant", + "key": key, + "members": members, + }) + + # 规则2 & 3: 子串包含 / 前缀碰撞(仅在 lower 名不同的组间比较) + distinct = [m for members in by_lower.values() for m in members] + seen_pairs: set[tuple[int, int]] = set() + for i, a in enumerate(distinct): + na = (a["name"] or "").lower() + for b in distinct[i + 1:]: + nb = (b["name"] or "").lower() + if na == nb: + continue # 已被规则1覆盖 + pair = (min(a["id"], b["id"]), max(a["id"], b["id"])) + if pair in seen_pairs: + continue + hit = None + if len(na) >= 5 and na in nb: + hit = "substring" + elif len(nb) >= 5 and nb in na: + hit = "substring" + elif len(na) >= 8 and len(nb) >= 8 and na[:8] == nb[:8]: + hit = "prefix" + if hit: + seen_pairs.add(pair) + groups.append({ + "rule": hit, + "members": [a, b], + }) + + return { + "count": len(groups), + "hint": "命中任一启发式仅表示'可疑',合并前请人工确认是否同一球队", + "groups": groups, + } diff --git a/src/db/repositories.py b/src/db/repositories.py index 76f9b94..1768380 100644 --- a/src/db/repositories.py +++ b/src/db/repositories.py @@ -5,6 +5,10 @@ Repository 只负责查询,不负责事务提交。 """ from __future__ import annotations +import logging + +logger = logging.getLogger(__name__) + from datetime import datetime from sqlalchemy import select from sqlalchemy.orm import selectinload @@ -108,10 +112,22 @@ class TeamRepository: return (await self._session.execute(stmt)).scalar_one_or_none() async def get_or_create(self, name: str, *, name_zh: str | None = None) -> Team: - """按名获取球队,不存在则创建(name_zh 供 bzzoiro 管线写中文名)。""" - team = await self.get_by_name(name) + """按名获取球队,不存在则创建(name_zh 供 bzzoiro 管线写中文名)。 + + 归一化咽喉:所有入库 Team.name 必须经过 team_names.normalize, + 此处统一收敛,避免各调用点散落归一化逻辑导致重复 Team。 + 创建新 Team 时 info 打出原始名与归一后的规范名,便于排查重名。 + """ + from src.data.team_names import normalize as normalize_name + + normalized = normalize_name(name) or name.strip() + team = await self.get_by_name(normalized) if team is None: - team = Team(name=name, name_zh=name_zh) + logger.info( + "创建新 Team: %s -> %s", + name, normalized, + ) + team = Team(name=normalized, name_zh=name_zh) self._session.add(team) await self._session.flush() return team diff --git a/tests/test_review_required_fixes.py b/tests/test_review_required_fixes.py index e070e45..a80bdcb 100644 --- a/tests/test_review_required_fixes.py +++ b/tests/test_review_required_fixes.py @@ -125,7 +125,7 @@ class _FakeDb: async def test_r2_standings_actually_upserts(monkeypatch): """行为测试: 喂一份积分榜 payload,断言真的构造了 Standing 且计数 > 0。""" import src.data.bzzoiro as bz - from src.db.models import League, Standing, Team + from src.db.models import DataLineage, League, RawEvent, Standing, Team payload = { "season": {"start_date": "2025-08-01", "end_date": "2026-05-31"}, @@ -166,7 +166,10 @@ async def test_r2_standings_actually_upserts(monkeypatch): standings = [o for o in db.added if isinstance(o, Standing)] assert len(standings) == 2, "应真的构造 Standing 行" - assert all(isinstance(o, (Standing, Team)) for o in db.added) + # standings 采集接线 Bronze 后(RawEvent + DataLineage),add 的对象类型白名单随之放宽 + assert all( + isinstance(o, (Standing, Team, RawEvent, DataLineage)) for o in db.added + ) first = standings[0] assert first.league_id == 42 diff --git a/tests/test_standings_bronze.py b/tests/test_standings_bronze.py new file mode 100644 index 0000000..8635a30 --- /dev/null +++ b/tests/test_standings_bronze.py @@ -0,0 +1,288 @@ +"""standings 成功路径 Bronze 层回归测试(RawEvent + DataLineage)。 + +背景: events/stats 管线成功后均已补写 Bronze 层,唯独 standings 采集 +成功后既不留原始载荷,也不留血缘 —— 三条管线的溯源链条在积分榜一环 +缺失。本测试守护(与 test_events_bronze.py 对称): + 1. 联赛成功 upsert → RawEvent(幂等键=standings:{league}:{season}) + + Lineage(target_table="standings", transform_name="standings_ingest") + 2. 更新已有快照(非插入)同样写 Bronze —— 积分榜是快照,刷新即采集 + 3. RawEvent 幂等: 同 source_record_id 已存在则跳过,血缘照写 + 4. 基础设施写入失败 → 只 warning,不拖垮采集主流程 + 5. 抓取失败路径继续走 _safe_write_ingest_failure,且不写 Bronze + +范式: 假 db(按查询实体分发预置数据 + 记录 add,flush 分配自增 id) ++ monkeypatch 抓取函数,不依赖真实数据库。 +""" +from __future__ import annotations + +import pytest + +import src.data.bzzoiro as bz +from src.db.models import DataLineage, IngestFailure, League, RawEvent, Standing, Team + + +def _payload(): + """构造一份最小合法的 bzzoiro /leagues/{id}/standings/ 原始载荷。""" + return { + "season": {"start_date": "2025-08-01", "end_date": "2026-05-31"}, + "standings": [ + { + "position": 1, "team_name": "Arsenal FC", + "played": 10, "won": 8, "drawn": 1, "lost": 1, + "gf": 22, "ga": 8, "gd": 14, "pts": 25, + "zone": {"key": "champions_league", "label": "Champions League"}, + }, + { + "position": 2, "team_name": "Chelsea FC", + "played": 10, "won": 6, "drawn": 2, "lost": 2, + "gf": 18, "ga": 12, "gd": 6, "pts": 20, + }, + ], + } + + +def _patch_fetch(monkeypatch, payload): + async def _fetch(league_code, season=None): + return payload + + monkeypatch.setattr(bz, "fetch_bzzoiro_standings", _fetch) + + +class _FakeResult: + """支持 .scalars().all() / .scalar_one_or_none() 的最小假结果集。""" + + def __init__(self, items): + self._items = list(items) + + def scalars(self): + return self + + def __iter__(self): + return iter(self._items) + + def all(self): + return self._items + + def scalar_one_or_none(self): + return self._items[0] if self._items else None + + +class _FakeDB: + """按查询实体分发预置数据;记录 add();flush 为无 id 对象分配自增主键。""" + + def __init__(self, leagues=(), teams=(), standings=(), raw_events=()): + self.added = [] + self._by_entity = { + League: list(leagues), + Team: list(teams), + Standing: list(standings), + RawEvent: list(raw_events), + } + self._next_id = 0 + + def add(self, obj): + self.added.append(obj) + + async def execute(self, stmt): + entities = set() + for d in (stmt.column_descriptions or []): + entities.add(d.get("entity") or d.get("type")) + for entity, items in self._by_entity.items(): + if entity in entities: + return _FakeResult(self._filter(entity, items, stmt)) + return _FakeResult([]) + + @staticmethod + def _filter(entity, items, stmt): + """RawEvent 查询按 source_record_id 过滤 —— 幂等测试需区分不同键。""" + if entity is RawEvent: + try: + params = stmt.compile().params + except Exception: + return items + rid = next((v for k, v in params.items() if "source_record_id" in k), None) + if rid is not None: + return [i for i in items if i.source_record_id == rid] + return items + + async def flush(self): + for obj in self.added: + if getattr(obj, "id", None) is None: + self._next_id += 1 + obj.id = self._next_id + + +def _preset_league(): + lg = League(code="EPL", name="Premier League", country="England") + lg.id = 42 + return lg + + +def _raw_events(db): + return [o for o in db.added if isinstance(o, RawEvent)] + + +def _lineages(db): + return [o for o in db.added if isinstance(o, DataLineage)] + + +# ============================================================ +# 1. 成功 upsert → RawEvent + DataLineage +# ============================================================ + + +class TestStandingsBronzeOnUpsert: + async def test_upsert_writes_raw_event_and_lineage(self, monkeypatch): + _patch_fetch(monkeypatch, _payload()) + db = _FakeDB(leagues=[_preset_league()]) + + result = await bz.ingest_bzzoiro_standings(db, leagues=["EPL"]) + + assert result["errors"] == [] + assert result["total_upserted"] == 2 + + raws = _raw_events(db) + assert len(raws) == 1 + raw = raws[0] + assert raw.source_system == "bzzoiro" + # 幂等键: 联赛 + 实际入库的赛季标签(由载荷日期推导,与 Standing.season 同口径) + assert raw.source_record_id == "standings:EPL:2025-2026" + assert raw.ingest_batch_id.startswith("bzzoiro-standings-EPL-") + # 整份原始载荷完整保留 + assert raw.raw_payload["standings"][0]["team_name"] == "Arsenal FC" + + lineages = _lineages(db) + assert len(lineages) == 1 + lin = lineages[0] + assert lin.source_system == "bzzoiro" + assert lin.source_record_id == "standings:EPL:2025-2026" + assert lin.target_table == "standings" + assert lin.target_id == 42 # 联赛 id + assert lin.transform_name == "standings_ingest" + assert lin.transform_detail == { + "league": "EPL", "season": "2025-2026", "rows_upserted": 2, + } + # RawEvent 与 Lineage 同批次,便于按批追溯 + assert lin.batch_id == raw.ingest_batch_id + + async def test_updated_snapshot_also_writes_bronze(self, monkeypatch): + """已有快照就地更新(非插入)同样是成功采集,必须留 Bronze 记录。""" + payload = _payload() + payload["standings"] = payload["standings"][:1] # 单队,便于命中同一行 + _patch_fetch(monkeypatch, payload) + + team = Team(name="Arsenal FC", name_zh="阿森纳") + team.id = 7 + existing = Standing(league_id=42, season="2025-2026", team_id=7, position=9) + existing.points = 1 + db = _FakeDB(leagues=[_preset_league()], teams=[team], standings=[existing]) + + result = await bz.ingest_bzzoiro_standings(db, leagues=["EPL"]) + + assert result["total_upserted"] == 1 + assert result["leagues"]["EPL"]["teams_created"] == 0 + # 快照刷新也要留痕: RawEvent(幂等) + 血缘 + assert len(_raw_events(db)) == 1 + lineages = _lineages(db) + assert len(lineages) == 1 + assert lineages[0].transform_name == "standings_ingest" + assert lineages[0].transform_detail["rows_upserted"] == 1 + + async def test_empty_upsert_writes_no_bronze(self, monkeypatch): + """载荷有行但全部队名为空 → 没有任何 upsert,不应产生 RawEvent/Lineage。""" + payload = {"standings": [{"position": 1, "team_name": ""}]} + _patch_fetch(monkeypatch, payload) + db = _FakeDB(leagues=[_preset_league()]) + + result = await bz.ingest_bzzoiro_standings(db, leagues=["EPL"]) + + assert result["total_upserted"] == 0 + assert _raw_events(db) == [] + assert _lineages(db) == [] + + +# ============================================================ +# 2. RawEvent 幂等: 同 source_record_id 跳过 +# ============================================================ + + +class TestStandingsRawEventIdempotent: + async def test_existing_raw_event_is_skipped(self, monkeypatch): + existing = RawEvent( + source_system="bzzoiro", + source_record_id="standings:EPL:2025-2026", + raw_payload={"old": True}, + ) + _patch_fetch(monkeypatch, _payload()) + db = _FakeDB(leagues=[_preset_league()], raw_events=[existing]) + + await bz.ingest_bzzoiro_standings(db, leagues=["EPL"]) + + new_raws = [r for r in _raw_events(db) if r is not existing] + assert new_raws == [] + assert len(_lineages(db)) == 1 # 血缘仍然记录本次采集 + + async def test_different_season_writes_new_raw_event(self, monkeypatch): + """幂等键含赛季: 同联赛不同赛季各留一条 RawEvent。""" + payload = _payload() + payload["season"] = {"start_date": "2024-08-01", "end_date": "2025-05-31"} + existing = RawEvent( + source_system="bzzoiro", + source_record_id="standings:EPL:2025-2026", + raw_payload={"old": True}, + ) + _patch_fetch(monkeypatch, payload) + db = _FakeDB(leagues=[_preset_league()], raw_events=[existing]) + + await bz.ingest_bzzoiro_standings(db, leagues=["EPL"]) + + new_raws = [r for r in _raw_events(db) if r is not existing] + assert len(new_raws) == 1 + assert new_raws[0].source_record_id == "standings:EPL:2024-2025" + + +# ============================================================ +# 3. 基础设施写入失败: 尽力而为,不拖垮主流程 +# ============================================================ + + +class TestStandingsBronzeIsBestEffort: + async def test_bronze_write_failure_does_not_break_ingest(self, monkeypatch): + async def _boom(*args, **kwargs): + raise RuntimeError("infra down") + + monkeypatch.setattr(bz, "_write_raw_event", _boom) + monkeypatch.setattr(bz, "_write_lineage", _boom) + _patch_fetch(monkeypatch, _payload()) + db = _FakeDB(leagues=[_preset_league()]) + + # 不应抛异常:Bronze 写不进去只记 warning + result = await bz.ingest_bzzoiro_standings(db, leagues=["EPL"]) + + assert result["total_upserted"] == 2 + assert [o for o in db.added if isinstance(o, Standing)] + + +# ============================================================ +# 4. 抓取失败: 继续写死信,且不写 Bronze +# ============================================================ + + +class TestStandingsFailurePathKeepsDeadLetter: + async def test_fetch_failure_writes_deadletter_and_no_bronze(self, monkeypatch): + async def _boom(league_code, season=None): + raise RuntimeError("upstream 500") + + monkeypatch.setattr(bz, "fetch_bzzoiro_standings", _boom) + db = _FakeDB() + + result = await bz.ingest_bzzoiro_standings(db, leagues=["SP1"], season="2025-2026") + + assert result["errors"] + failures = [o for o in db.added if isinstance(o, IngestFailure)] + assert len(failures) == 1 + assert failures[0].entity_type == "standings" + assert failures[0].error_type == "fetch_error" + # 失败路径绝不写 Bronze(没有任何成功 upsert) + assert _raw_events(db) == [] + assert _lineages(db) == []