diff --git a/alembic/versions/0014_predictions_agent_weights.py b/alembic/versions/0014_predictions_agent_weights.py new file mode 100644 index 0000000..c0afef6 --- /dev/null +++ b/alembic/versions/0014_predictions_agent_weights.py @@ -0,0 +1,39 @@ +"""为 predictions 表增加 agent_weights 列 + +Revision ID: 0014_predictions_agent_weights +Revises: 0013_predictions_unique_constraint_mode_run_type +Create Date: 2026-09-20 + +背景: + multi-agent 终裁的 agent_weights 原本只在 raw_response JSON 中, + 无独立列,评估不便。本迁移增加 agent_weights JSONB 列,可空,旧行保持 NULL。 +""" + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + +# revision identifiers, used by Alembic. +revision: str = '0014_predictions_agent_weights' +down_revision: Union[str, None] = '0013_predictions_unique_constraint_mode_run_type' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # 新增 agent_weights 列(JSONB, 可空, 旧行保持 NULL) + op.add_column( + "predictions", + sa.Column( + "agent_weights", + sa.dialects.postgresql.JSONB(), + nullable=True, + comment="multi-agent 终裁各专家权重, 格式: {expert_name: weight}", + ), + ) + + +def downgrade() -> None: + # 删除 agent_weights 列 + op.drop_column("predictions", "agent_weights") diff --git a/src/api/deps.py b/src/api/deps.py index 7e462ba..88da424 100644 --- a/src/api/deps.py +++ b/src/api/deps.py @@ -76,6 +76,17 @@ async def require_admin( 用法: `@router.get("/leagues", dependencies=[Depends(require_admin)])` """ if not await auth_configured(): + # 生产环境 fail-closed:未配置鉴权则拒绝,不放行 + if settings.REQUIRE_ADMIN_AUTH or settings.APP_ENV == "production": + logger.error( + "生产环境管理接口未配置鉴权(REQUIRE_ADMIN_AUTH=True 或 APP_ENV=production)," + "拒绝访问。请设置 ADMIN_PASSWORD 或 ADMIN_API_KEY。" + ) + raise HTTPException( + status_code=503, + detail="服务未配置管理鉴权,请联系管理员", + ) + # 开发环境 fail-open + warning logger.warning( "管理员密码 / ADMIN_API_KEY 均未设置,管理接口当前【无鉴权】。" "生产环境请至少设置其中一项。" @@ -98,6 +109,31 @@ async def require_admin( raise HTTPException(status_code=401, detail="未登录或凭证无效") +# ── 客户端 IP 提取(防 X-Forwarded-For 伪造) ── + +def get_client_ip(request: Request) -> str: + """获取客户端真实 IP,防 X-Forwarded-For 伪造。 + + 规则: + - TRUST_PROXY_HEADERS=False(默认):只用 request.client.host, + 忽略 X-Forwarded-For,防止客户端伪造。 + - TRUST_PROXY_HEADERS=True:解析 X-Forwarded-For 第一个 IP, + 适用于 Nginx 等可信反代后方。 + + 部署建议: + - 公网必须设 TRUST_PROXY_HEADERS=True,并在 Nginx 配置: + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + - Nginx 层也建议做限流(limit_req),作为第二道防线。 + """ + if settings.TRUST_PROXY_HEADERS: + # 信任反代:X-Forwarded-For 可能包含多个 IP(代理链),取第一个 + forwarded = request.headers.get("X-Forwarded-For") + if forwarded: + return forwarded.split(",")[0].strip() + # 默认或无反代头:直接用连接层 IP + return request.client.host if request.client else "unknown" + + # ── 简易内存限流(按 IP,无外部依赖) ── class _RateLimiter: @@ -139,13 +175,10 @@ _predict_limiter = _RateLimiter(max_requests=10, window_seconds=60) async def rate_limit_predict(request: Request) -> None: """POST /api/v1/predict 限流依赖。 - 基于客户端 IP(考虑 X-Forwarded-For),超过 10 次/分钟返回 429。 + 基于客户端 IP,超过 10 次/分钟返回 429。 + IP 提取逻辑:优先用 get_client_ip(防伪造)。 """ - # 获取客户端 IP(支持反向代理) - client_ip = request.headers.get("X-Forwarded-For", request.client.host if request.client else "unknown") - # X-Forwarded-For 可能包含多个 IP(代理链),取第一个 - if "," in client_ip: - client_ip = client_ip.split(",")[0].strip() + client_ip = get_client_ip(request) if not _predict_limiter.is_allowed(client_ip): logger.warning("rate limit exceeded for %s", client_ip) diff --git a/src/api/routes/auth.py b/src/api/routes/auth.py index a906ab2..62b1ddc 100644 --- a/src/api/routes/auth.py +++ b/src/api/routes/auth.py @@ -54,7 +54,9 @@ class PasswordChangeIn(BaseModel): def _client_ip(request: Request) -> str: - return request.client.host if request.client else "unknown" + """获取客户端 IP,与限流共用同一套逻辑(防伪造)。""" + from src.api.deps import get_client_ip + return get_client_ip(request) def _is_locked(ip: str) -> bool: diff --git a/src/core/config.py b/src/core/config.py index 76cfaba..8df9a69 100644 --- a/src/core/config.py +++ b/src/core/config.py @@ -11,6 +11,9 @@ class Settings(BaseSettings): # --- app --- APP_ENV: str = "development" LOG_LEVEL: str = "INFO" + # 生产环境强制要求管理鉴权配置,即使 APP_ENV=production 也生效。 + # True 时若 auth_configured() 为 False 则拒绝(503),development 保持 fail-open。 + REQUIRE_ADMIN_AUTH: bool = False # --- database --- DATABASE_URL: str = "postgresql+asyncpg://football:football@localhost:5432/football" @@ -30,6 +33,12 @@ class Settings(BaseSettings): BZZOIRO_BASE: str = "https://sports.bzzoiro.com/api/v2" API_FOOTBALL_KEY: str = "" + # --- 代理头信任 --- + # 为 True 时才解析 X-Forwarded-For,否则只用 request.client.host。 + # 公网部署时应设为 True,并确保仅 Nginx 等可信反代能访问 API, + # 且 Nginx 层已覆盖真实 IP(X-Real-IP / proxy_protocol)。 + TRUST_PROXY_HEADERS: bool = False + # --- CORS --- CORS_ORIGINS: str = "http://localhost:5173,http://localhost:3000" CORS_METHODS: str = "GET,POST,PUT,DELETE,OPTIONS" diff --git a/src/data/bzzoiro.py b/src/data/bzzoiro.py index ef4a8b0..cb7f827 100644 --- a/src/data/bzzoiro.py +++ b/src/data/bzzoiro.py @@ -266,8 +266,12 @@ class BzzoiroSource: db.add(m) await db.flush() existing_matches[match_key] = m # 防止同批重复 - if nm.home_xg is not None or nm.away_xg is not None: + # 存在任一统计字段即可创建 MatchStats(不再强制要求 xG) + if any(getattr(nm, f) is not None for f in ['home_xg', 'away_xg', 'home_shots', 'away_shots', 'home_shots_on_target', 'away_shots_on_target', 'home_corners', 'away_corners', 'home_possession', 'home_yellow_cards', 'away_yellow_cards', 'home_red_cards', 'away_red_cards']): now = datetime.now(timezone.utc) + # available_at 语义:统计「可被使用」的最早时间,至少不早于比赛结束 + # 局限:用 match_date(开球时间)近似,实际完赛时间约为 +2 小时 + available_at = nm.date if nm.date else now stats = MatchStats( match_id=m.id, home_xg=nm.home_xg, @@ -286,7 +290,7 @@ class BzzoiroSource: source="bzzoiro", source_record_id=str(raw.get("id", "")), retrieved_at=now, - available_at=now, + available_at=available_at, ) db.add(stats) league_r["inserted"] += 1 @@ -305,14 +309,23 @@ class BzzoiroSource: if existing_match.match_stage is None and nm.match_stage: existing_match.match_stage = nm.match_stage changed = True - if existing_match.stats is None and (nm.home_xg is not None or nm.away_xg is not None): + # 存在任一统计字段即可创建 MatchStats(不再强制要求 xG) + if existing_match.stats is None and ( + nm.home_xg is not None or nm.away_xg is not None + or nm.home_shots is not None or nm.away_shots is not None + or nm.home_corners is not None or nm.away_corners is not None + or nm.home_possession is not None + ): now = datetime.now(timezone.utc) + # available_at 语义:统计「可被使用」的最早时间,至少不早于比赛结束 + # 局限:用 match_date(开球时间)近似,实际完赛时间约为 +2 小时 + available_at = nm.date if nm.date else now existing_match.stats = MatchStats( match_id=existing_match.id, source="bzzoiro", source_record_id=str(raw.get("id", "")), retrieved_at=now, - available_at=now, + available_at=available_at, ) db.add(existing_match.stats) await db.flush() diff --git a/src/data/injuries.py b/src/data/injuries.py index 4201344..29e0a35 100644 --- a/src/data/injuries.py +++ b/src/data/injuries.py @@ -12,11 +12,25 @@ import tempfile import time from datetime import datetime, timezone from pathlib import Path +from dataclasses import dataclass from typing import Any import httpx from src.core.runtime_config import get_runtime_value + + +@dataclass +class InjuryQueryResult: + """伤停查询结果(区分「查询成功但为空」与「查询失败/源未配置」)。""" + + records: list["Injury"] + query_status: str # "success" | "source_not_configured" | "query_error" + + @property + def has_data(self) -> bool: + """成功查询(即使结果为空)视为有明确名单,has_data=True。""" + return self.query_status == "success" from src.core.http_client import get_client logger = logging.getLogger(__name__) @@ -213,42 +227,59 @@ async def ingest_injuries(db, *, date: str | None = None) -> dict: rows = (await db.execute(stmt)).all() existing_keys = {(r[0], r[1], r[2]) for r in rows} - # Fix 1: 使用 SAVEPOINT(begin_nested)避免整批回滚丢数据 - # 每个 batch 使用独立的 savepoint,失败时只回滚该 batch + # Fix 1: 使用 begin_nested(SAVEPOINT)隔离每批 flush + # IntegrityError 时只回滚到 savepoint,不影响其它已成功批次 BATCH_SIZE = 50 + batch: list[Injury] = [] + + async def _flush_batch(): + """使用 savepoint flush 一批记录;失败只回滚本批。""" + if not batch: + return + async with db.begin_nested(): + for obj in batch: + db.add(obj) + await db.flush() + batch.clear() + for i, rec in enumerate(pending_records): key = (rec["player_id"], rec["fixture_id"], rec["injury_type"]) if key in existing_keys: continue - injury = Injury(**rec) - db.add(injury) + batch.append(Injury(**rec)) result["inserted"] += 1 - # 每 BATCH_SIZE 条 flush 一次,使用 SAVEPOINT 隔离 - if result["inserted"] % BATCH_SIZE == 0: + # 每 BATCH_SIZE 条 flush 一次 + if len(batch) >= BATCH_SIZE: try: - await db.flush() + await _flush_batch() except IntegrityError: - # 只回滚到上一个 savepoint,不影响已提交的数据 - await db.rollback() - logger.warning("injuries batch IntegrityError at record %d, continuing", i + 1) - # 从当前位置继续处理剩余记录 + logger.warning( + "injuries batch IntegrityError at record %d, " + "rolled back to savepoint, continuing", + i + 1, + ) + # begin_nested 已回滚到 savepoint,清空 batch 继续 + batch.clear() continue # 最终 flush(剩余不足一批的记录) try: - await db.flush() + await _flush_batch() except IntegrityError: - await db.rollback() - logger.warning("injuries final flush IntegrityError, some records may be lost") + logger.warning( + "injuries final flush IntegrityError, " + "rolled back to savepoint, some records may be lost", + ) + batch.clear() # 注意: 不在此处 commit,由调用方 UnitOfWork 控制事务 logger.info("injuries: fetched %d, inserted %d for %s", result["count"], result["inserted"], date) return result -async def get_injuries_for_match(db, team_id: int, match_date, as_of=None) -> list[Injury]: +async def get_injuries_for_match(db, team_id: int, match_date, as_of=None) -> InjuryQueryResult: """查询某场比赛前某队的伤停名单(比赛日仍缺阵的)。 Args: @@ -258,31 +289,46 @@ async def get_injuries_for_match(db, team_id: int, match_date, as_of=None) -> li as_of: 数据截止时间(用于回测防泄漏) Returns: - 伤停记录列表 + InjuryQueryResult:包含查询记录与状态 + - query_status="success": 查询成功(即使结果也为空) + - query_status="source_not_configured": API_FOOTBALL_KEY 未配置 + - query_status="query_error": 查询异常 + + 语义区分: + - 成功查询 + 空结果 → has_data=True(明确知道「无人伤停」) + - 源未配置 / 查询失败 → has_data=False(无法判断,跳过 LLM) """ from sqlalchemy import select, func from src.db.models import Injury - # Fix 3: 统一用 timezone-aware datetime 比较,禁止 date() 截断 - # retrieved_at 是 timestamptz,as_of 也应该是 datetime - # 比较时统一转为 date 避免时间部分导致当天数据不可见 - if hasattr(match_date, "date") and callable(match_date.date): - match_date = match_date.date() + # 检查 API 是否配置(只读配置,不发网络) + api_key = await get_runtime_value("API_FOOTBALL_KEY") + if not api_key: + logger.debug("API_FOOTBALL_KEY 未配置,跳过伤停查询 team=%s", team_id) + return InjuryQueryResult(records=[], query_status="source_not_configured") - stmt = ( - select(Injury) - .where(Injury.team_id == team_id) - .where(Injury.injury_date <= match_date) - .where( - (Injury.return_date.is_(None)) | (Injury.return_date >= match_date) + try: + # Fix 3: 统一用 timezone-aware datetime 比较,禁止 date() 截断 + if hasattr(match_date, "date") and callable(match_date.date): + match_date = match_date.date() + + stmt = ( + select(Injury) + .where(Injury.team_id == team_id) + .where(Injury.injury_date <= match_date) + .where( + (Injury.return_date.is_(None)) | (Injury.return_date >= match_date) + ) ) - ) - if as_of is not None: - # Fix 3: 统一用 date 比较,避免 timestamptz vs date 的时区问题 - if hasattr(as_of, "date") and callable(as_of.date): - as_of = as_of.date() - # 使用 func.date() 将 timestamptz 转为 date,确保当天白天采到的数据对当晚比赛可见 - stmt = stmt.where(func.date(Injury.retrieved_at) <= as_of) + if as_of is not None: + # Fix 3: 统一用 date 比较,避免 timestamptz vs date 的时区问题 + if hasattr(as_of, "date") and callable(as_of.date): + as_of = as_of.date() + stmt = stmt.where(func.date(Injury.retrieved_at) <= as_of) - result = await db.execute(stmt) - return list(result.scalars().all()) + result = await db.execute(stmt) + records = list(result.scalars().all()) + return InjuryQueryResult(records=records, query_status="success") + except Exception as e: + logger.exception("伤停查询异常 team=%s: %s", team_id, e) + return InjuryQueryResult(records=[], query_status="query_error") diff --git a/src/data/normalize.py b/src/data/normalize.py index 87a0b94..b3bc8d1 100644 --- a/src/data/normalize.py +++ b/src/data/normalize.py @@ -177,6 +177,29 @@ def normalize_bzzoiro(raw: dict, league_type: str) -> NormalizedMatch | None: m.match_stage = _rn_name elif _rn: m.match_stage = f"第 {_rn} 轮" + + # 统计字段映射(API 字段名 → NormalizedMatch) + # API 可能提供的字段:home_shots/away_shots, shots_on_target, corners, possession, xg, cards + # API 没有的字段保持 None,不伪造 + m.home_shots = _to_int(raw.get("home_shots", raw.get("shots_home"))) + m.away_shots = _to_int(raw.get("away_shots", raw.get("shots_away"))) + m.home_shots_on_target = _to_int(raw.get("home_shots_on_target", raw.get("sot_home"))) + m.away_shots_on_target = _to_int(raw.get("away_shots_on_target", raw.get("sot_away"))) + m.home_corners = _to_int(raw.get("home_corners", raw.get("corners_home"))) + m.away_corners = _to_int(raw.get("away_corners", raw.get("corners_away"))) + # 控球率:API 通常只给 home 值,away = 100 - home + possession_home = _to_float(raw.get("home_possession", raw.get("possession"))) + if possession_home is not None: + m.home_possession = possession_home + # xG + m.home_xg = _to_float(raw.get("home_xg", raw.get("xg_home", raw.get("expected_goals_home")))) + m.away_xg = _to_float(raw.get("away_xg", raw.get("xg_away", raw.get("expected_goals_away")))) + # 牌 + m.home_yellow_cards = _to_int(raw.get("home_yellow_cards", raw.get("yellow_cards_home"))) + m.away_yellow_cards = _to_int(raw.get("away_yellow_cards", raw.get("yellow_cards_away"))) + m.home_red_cards = _to_int(raw.get("home_red_cards", raw.get("red_cards_home"))) + m.away_red_cards = _to_int(raw.get("away_red_cards", raw.get("red_cards_away"))) + if m.match_status == "finished" and m.home_goals is None: m.match_status = "scheduled" return m diff --git a/src/data/team_names.py b/src/data/team_names.py index 04d452c..4b3b503 100644 --- a/src/data/team_names.py +++ b/src/data/team_names.py @@ -1,6 +1,9 @@ """队名归一化:各源队名 → 统一规范名。 迁移自旧项目 app/data/team_names.py。 + +NFKD 归一化会剥离变音符号(ü→u),导致「Bayern München」与「Bayern Munich」 +映射到不同规范名。修复:双重查找(原始名 + NFKD 名) + 补齐常见变体键。 """ from __future__ import annotations @@ -61,10 +64,15 @@ NORMALIZE_MAP = { "Barcelona": "Barcelona", # ---- 德甲 ---- "Bayern Munich": "Bayern München", + "Bayern Munchen": "Bayern München", # NFKD stripped variant + "Bayern München": "Bayern München", # canonical with umlaut (direct hit) "FC Koln": "FC Köln", + "FC Köln": "FC Köln", "RB Leipzig": "RB Leipzig", "Borussia Dortmund": "Borussia Dortmund", "Borussia M'gladbach": "Borussia Mönchengladbach", + "Borussia Monchengladbach": "Borussia Mönchengladbach", # NFKD stripped + "Borussia Mönchengladbach": "Borussia Mönchengladbach", # canonical "Bayer Leverkusen": "Bayer Leverkusen", "Eintracht Frankfurt": "Eintracht Frankfurt", "VfB Stuttgart": "VfB Stuttgart", @@ -121,6 +129,9 @@ NORMALIZE_MAP = { "Le Havre": "Le Havre AC", "Lorient": "FC Lorient", "Saint-Etienne": "AS Saint-Étienne", + "Saint-Etienne": "AS Saint-Étienne", # NFKD stripped (ê → e) + "AS Saint-Étienne": "AS Saint-Étienne", # canonical prefix + "AS Saint-Etienne": "AS Saint-Étienne", # NFKD stripped with prefix "Angers": "Angers SCO", "Auxerre": "AJ Auxerre", "Leganes": "Leganés", @@ -128,10 +139,25 @@ NORMALIZE_MAP = { def normalize(name: str) -> str: + """队名归一化:变音符号变体 → 统一规范名。 + + 双重查找策略: + 1. 原始名(带变音)直接查表 + 2. NFKD 去变音后再查表 + + 确保「Bayern München」「Bayern Munchen」「Bayern Munich」 + 都映射到同一规范名「Bayern München」。 + """ if not name: return "" - # unicode 归一(重音) - n = unicodedata.normalize("NFKD", name) + + # 1. 原始名直接查表(保留变音变体) + stripped = name.strip() + if stripped in NORMALIZE_MAP: + return NORMALIZE_MAP[stripped] + + # 2. NFKD 去变音后查表 + n = unicodedata.normalize("NFKD", stripped) n = "".join(c for c in n if not unicodedata.combining(c)) n = n.strip() return NORMALIZE_MAP.get(n, n) diff --git a/src/data/understat.py b/src/data/understat.py index 8eba7c8..99b45a9 100644 --- a/src/data/understat.py +++ b/src/data/understat.py @@ -199,12 +199,16 @@ class UnderstatSource: # 回填 xG if existing.stats is None and (nm.home_xg is not None or nm.away_xg is not None): now = datetime.now(timezone.utc) + # available_at 语义:统计「可被使用」的最早时间,至少不早于比赛结束 + # 局限:用 match_date(开球时间)近似,实际完赛时间约为 +2 小时 + match_date = existing.match_date if existing.match_date else now + available_at = match_date existing.stats = MatchStats( match_id=existing.id, source="understat", source_record_id=str(raw.get("id", "")), retrieved_at=now, - available_at=now, + available_at=available_at, ) db.add(existing.stats) await db.flush() diff --git a/src/db/models.py b/src/db/models.py index caf8c91..dc6bd77 100644 --- a/src/db/models.py +++ b/src/db/models.py @@ -205,6 +205,8 @@ class Prediction(Base): # multi-agent 模式: 各专家报告 mode: Mapped[str] = mapped_column(String(20), nullable=False, default="single") agent_outputs: Mapped[dict | None] = mapped_column(JSONB) + # Fix: agent_weights 独立持久化到列(原本只在 raw_response 中) + agent_weights: Mapped[dict | None] = mapped_column(JSONB) # 预测状态: success / failed / degraded status: Mapped[str] = mapped_column(String(20), nullable=False, default="success") # Fix: run_type 区分实盘(live)与回测(backtest),避免回测覆盖实盘预测 diff --git a/src/llm/agents/orchestrator.py b/src/llm/agents/orchestrator.py index df3ee6c..7b2bdcb 100644 --- a/src/llm/agents/orchestrator.py +++ b/src/llm/agents/orchestrator.py @@ -243,11 +243,35 @@ async def predict_match_multi( # 2. 并行专家(各自独立配置,使用统一 cutoff) reports = await run_specialists(header, version=version, before=cutoff) - # 3. 终裁 - aggregator_provider = await _agent_provider("aggregator", tier="aggregator") - final, agg_prompt_tokens, agg_completion_tokens = await run_aggregator( - header, reports, provider=aggregator_provider, version=version - ) + # 2.5 统计有效专家报告数量 + ok_reports = [r for r in reports if r.status == "ok"] + has_valid_data = len(ok_reports) > 0 + + # 3. 终裁(仅当有有效专家报告时执行) + if has_valid_data: + aggregator_provider = await _agent_provider("aggregator", tier="aggregator") + final, agg_prompt_tokens, agg_completion_tokens = await run_aggregator( + header, reports, provider=aggregator_provider, version=version + ) + else: + # 所有专家无数据/均失败:跳过终裁,标记 degraded + logger.warning( + "match %s: 所有 %d 位专家均无有效数据,跳过终裁,标记 degraded", + match_id, len(reports), + ) + aggregator_provider = await _agent_provider("aggregator", tier="aggregator") + final = { + "pred_home_goals": None, + "pred_away_goals": None, + "alt_pred_home_goals": None, + "alt_pred_away_goals": None, + "pred_1x2": None, + "subjective_confidence": None, + "reasoning": f"所有 {len(reports)} 位专家均无有效数据或预测失败(状态: {','.join(r.status for r in reports)})", + "agent_weights": {}, + } + agg_prompt_tokens = 0 + agg_completion_tokens = 0 latency_ms = int((time.perf_counter() - start) * 1000) @@ -262,15 +286,23 @@ async def predict_match_multi( if m is None: raise ValueError(f"match {match_id} not found") - # 严格校验终裁输出 + # 根据是否有有效数据决定校验策略 from src.llm.validation import validate_agent_weights, validate_prediction_output - try: - validated = validate_prediction_output(final) - except Exception as e: - raise RuntimeError(f"终裁输出校验失败: {e}") - # agent_weights 同样必须过校验(旧实现直接取 raw 值落库,未做任何检查) - agent_weights = validate_agent_weights(final.get("agent_weights")) + if has_valid_data: + # 有有效报告:严格校验终裁输出 + try: + validated = validate_prediction_output(final) + except Exception as e: + raise RuntimeError(f"终裁输出校验失败: {e}") + agent_weights = validate_agent_weights(final.get("agent_weights")) + pred_status = "success" + else: + # 无有效报告:跳过严格校验,直接构造降级结果 + validated = None # type: ignore + agent_weights = {} + pred_status = "degraded" + pred = await _upsert_prediction( session, match_id=match_id, @@ -283,16 +315,17 @@ async def predict_match_multi( "prompt_tokens": sum(r.prompt_tokens or 0 for r in reports) + agg_prompt_tokens, "completion_tokens": sum(r.completion_tokens or 0 for r in reports) + agg_completion_tokens, "latency_ms": latency_ms, - "pred_home_goals": validated.pred_home_goals, - "pred_away_goals": validated.pred_away_goals, - "alt_pred_home_goals": validated.alt_pred_home_goals, - "alt_pred_away_goals": validated.alt_pred_away_goals, - "pred_1x2": validated.pred_1x2, - "subjective_confidence": validated.subjective_confidence, - "reasoning": validated.reasoning, + "pred_home_goals": validated.pred_home_goals if validated else None, + "pred_away_goals": validated.pred_away_goals if validated else None, + "alt_pred_home_goals": validated.alt_pred_home_goals if validated else None, + "alt_pred_away_goals": validated.alt_pred_away_goals if validated else None, + "pred_1x2": validated.pred_1x2 if validated else None, + "subjective_confidence": validated.subjective_confidence if validated else None, + "reasoning": validated.reasoning if validated else final.get("reasoning", ""), "raw_response": final, "agent_outputs": [r.to_dict() for r in reports], - "status": "success", + "agent_weights": agent_weights, + "status": pred_status, "match_kickoff_at": match_kickoff_at, "prediction_cutoff_at": prediction_cutoff_at, "prediction_created_at": now, diff --git a/src/llm/context_builder.py b/src/llm/context_builder.py index 6a53aae..76139ba 100644 --- a/src/llm/context_builder.py +++ b/src/llm/context_builder.py @@ -40,11 +40,23 @@ def _outcome(home_goals: int, away_goals: int, side: str) -> str: def _is_stats_available(stats, before) -> bool: - """检查统计数据在 cutoff 时间是否已可用。""" + """检查统计数据在 cutoff 时间是否已可用。 + + available_at 语义:该条统计「对外可被使用」的最早时间, + 至少不得早于比赛结束。用于回测防泄漏。 + + 规则: + - before is None(实盘):available_at 为 None 时允许(兼容旧数据) + - before is not None(回测):available_at 为 None 视为不可用(保守) + - available_at > cutoff:不可用(数据在 cutoff 之后才生成) + """ if before is None: + # 实盘模式:无时间信息时允许(兼容旧数据) return True + # 回测模式(cutoff 不为 None): + # available_at 为 None → 无法确认是否在 cutoff 前可用,保守视为不可用 if stats.available_at is None: - return True # 无时间信息时保守处理:允许使用 + return False return stats.available_at <= before @@ -330,34 +342,62 @@ async def injuries_slice(header: MatchHeader, *, before=None, db: AsyncSession | before=cutoff: 只使用 cutoff 之前已采集的伤停数据,防回测泄漏。 db: 可选共享 session(见模块 docstring)。 + + 语义区分: + - 查询成功 + 空结果 → has_data=True(明确知道「无人伤停」) + - 源未配置 / 查询失败 → has_data=False(无法判断,跳过 LLM) """ - from src.data.injuries import get_injuries_for_match + from src.data.injuries import get_injuries_for_match, InjuryQueryResult cutoff = before or header.match_dt if db is not None: - home_injuries = await get_injuries_for_match(db, header.home_team_id, cutoff, as_of=cutoff) - away_injuries = await get_injuries_for_match(db, header.away_team_id, cutoff, as_of=cutoff) + home_result = await get_injuries_for_match(db, header.home_team_id, cutoff, as_of=cutoff) + away_result = await get_injuries_for_match(db, header.away_team_id, cutoff, as_of=cutoff) else: async with AsyncSessionLocal() as new_db: - home_injuries = await get_injuries_for_match(new_db, header.home_team_id, cutoff, as_of=cutoff) - away_injuries = await get_injuries_for_match(new_db, header.away_team_id, cutoff, as_of=cutoff) + home_result = await get_injuries_for_match(new_db, header.home_team_id, cutoff, as_of=cutoff) + away_result = await get_injuries_for_match(new_db, header.away_team_id, cutoff, as_of=cutoff) + + # 判断是否有有效查询结果 + # 两队都成功查询(即使为空) → has_data=True + # 任一查询失败或源未配置 → has_data=False + both_succeeded = ( + home_result.query_status == "success" + and away_result.query_status == "success" + ) + any_configured = ( + home_result.query_status != "source_not_configured" + or away_result.query_status != "source_not_configured" + ) lines = ["── 阵容完整性 ──"] n_records = 0 - for label, injuries in (("主队", home_injuries), ("客队", away_injuries)): - if injuries: - n_records += len(injuries) - lines.append(f" {label}伤停({len(injuries)}人):") - for inj in injuries[:8]: # 最多显示 8 条 + + for label, result in (("主队", home_result), ("客队", away_result)): + if result.query_status == "source_not_configured": + lines.append(f" {label}: 伤停源未配置") + elif result.query_status == "query_error": + lines.append(f" {label}: 查询异常") + elif result.records: + n_records += len(result.records) + lines.append(f" {label}伤停({len(result.records)}人):") + for inj in result.records[:8]: reason = inj.reason or inj.injury_type or "未知" lines.append(f" - {inj.player_name}: {reason}") - if len(injuries) > 8: - lines.append(f" ...及其他 {len(injuries) - 8} 人") + if len(result.records) > 8: + lines.append(f" ...及其他 {len(result.records) - 8} 人") else: - lines.append(f" {label}: 无伤停数据") + # 查询成功但无人伤停 + lines.append(f" {label}: 当前无伤停记录") - if n_records == 0: - return SliceResult(text="── 阵容完整性 ──\n 无数据", has_data=False, n_records=0) + # 决定 has_data: + # - 两队都成功查询(即使为空) → True(明确知道名单) + # - 源未配置且无数据 → False + has_data = both_succeeded or (any_configured and n_records > 0) + + if not has_data: + # 保留详细状态文案(伤停源未配置/查询异常),而非通用「无数据」 + return SliceResult(text="\n".join(lines), has_data=False, n_records=0) return SliceResult(text="\n".join(lines), has_data=True, n_records=n_records) diff --git a/tests/test_agent_weights_persist.py b/tests/test_agent_weights_persist.py new file mode 100644 index 0000000..6c4f501 --- /dev/null +++ b/tests/test_agent_weights_persist.py @@ -0,0 +1,141 @@ +"""回归测试: predictions 表 agent_weights 独立持久化。 + +验证: + 1. ORM 模型有 agent_weights 列(JSONB, 可空) + 2. 迁移文件存在且可逆 + 3. orchestrator 写入 agent_weights +""" +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest + +from src.db.models import Prediction + + +class TestAgentWeightsColumn: + """验证 predictions 表有 agent_weights 列。""" + + def test_orm_has_agent_weights_column(self): + """ORM 模型应包含 agent_weights 列。""" + cols = {c.name: c for c in Prediction.__table__.columns} + assert "agent_weights" in cols, "predictions 表应有 agent_weights 列" + + def test_agent_weights_is_jsonb(self): + """agent_weights 应为 JSONB 类型。""" + col = Prediction.__table__.columns["agent_weights"] + # JSONB 类型检查 + assert "JSON" in str(col.type).upper() or "JSONB" in str(col.type).upper() + + def test_agent_weights_nullable(self): + """agent_weights 应可空(旧行保持 NULL)。""" + col = Prediction.__table__.columns["agent_weights"] + assert col.nullable is True, "agent_weights 应可空" + + +class TestMigration: + """验证迁移文件存在且内容正确。""" + + def test_migration_exists(self): + import os + + path = "/.octop/workspaces/CA7PFH/Profeto/alembic/versions/0014_predictions_agent_weights.py" + assert os.path.exists(path) + + def test_migration_content(self): + path = "/.octop/workspaces/CA7PFH/Profeto/alembic/versions/0014_predictions_agent_weights.py" + content = open(path).read() + + assert "agent_weights" in content + assert "upgrade" in content + assert "downgrade" in content + assert "downgrade" in content and "drop_column" in content + assert "nullable=True" in content + + +class TestOrchestratorWritesAgentWeights: + """验证 orchestrator 写入 agent_weights。""" + + @pytest.mark.asyncio + async def test_orchestrator_writes_agent_weights_to_upsert(self): + """orchestrator 应将 agent_weights 传入 _upsert_prediction。""" + from src.llm.agents import orchestrator as orch_mod + from src.llm.agents.base import AgentReport + from src.llm.context_builder import MatchHeader + + header = MatchHeader( + match_id=999, home_name="A", away_name="B", + league_name="X", season=None, match_date="?", + match_dt=None, stage=None, + home_team_id=1, away_team_id=2, league_id=1, + ) + + # 构造有效专家报告(至少 1 个 ok) + reports = [ + AgentReport(agent="form", status="ok", analysis="good"), + AgentReport(agent="stats", status="error", analysis="failed"), + AgentReport(agent="home_away", status="ok", analysis="good"), + AgentReport(agent="injuries", status="no_data", analysis="无数据"), + AgentReport(agent="h2h", status="error", analysis="failed"), + ] + + captured_values = {} + + async def mock_specialists(h, *, version, before): + return reports + + async def mock_provider(aid, **kw): + return MagicMock(model="test-model") + + async def mock_header(mid, db=None): + return header + + async def mock_aggregator(header, reports, *, provider, version): + return { + "pred_home_goals": 2, + "pred_away_goals": 1, + "pred_1x2": "1", + "subjective_confidence": 0.7, + "reasoning": "test", + "agent_weights": {"form": 0.3, "home_away": 0.5, "stats": 0.2}, + }, 100, 50 + + async def mock_upsert(session, **kw): + captured_values.update(kw.get("values", {})) + p = MagicMock() + p.id = 1 + p.provider = "test" + p.model = "test" + p.prompt_version = "v1" + p.pred_home_goals = 2 + p.pred_away_goals = 1 + p.pred_1x2 = "1" + p.subjective_confidence = 0.7 + p.reasoning = "test" + p.agent_outputs = [] + p.agent_weights = kw["values"].get("agent_weights") + return p + + class FakeUow: + async def __aenter__(self): + return self + async def __aexit__(self, *a): + pass + async def get(self, cls, id): + return MagicMock() + + with patch.object(orch_mod, "run_specialists", mock_specialists), \ + patch.object(orch_mod, "_agent_provider", mock_provider), \ + patch.object(orch_mod, "load_match_header", mock_header), \ + patch.object(orch_mod, "_upsert_prediction", mock_upsert), \ + patch.object(orch_mod, "run_aggregator", mock_aggregator), \ + patch.object(orch_mod, "get_uow", FakeUow): + + result = await orch_mod.predict_match_multi(999) + + # 断言 agent_weights 被写入 + assert "agent_weights" in captured_values, "agent_weights 应传入 _upsert_prediction" + assert captured_values["agent_weights"] is not None, "agent_weights 不应为 None" + assert "form" in captured_values["agent_weights"], "agent_weights 应包含专家权重" + print(f"PASS: agent_weights = {captured_values['agent_weights']}") diff --git a/tests/test_available_at.py b/tests/test_available_at.py new file mode 100644 index 0000000..df6285e --- /dev/null +++ b/tests/test_available_at.py @@ -0,0 +1,107 @@ +"""回归测试: match_stats.available_at 回测防泄漏语义。 + +验证: + 1. _is_stats_available: available_at is None + cutoff 不为 None → 不可用 + 2. _is_stats_available: available_at is None + cutoff is None(实盘) → 可用(兼容旧数据) + 3. _is_stats_available: available_at > cutoff → 不可用 + 4. _is_stats_available: available_at <= cutoff → 可用 +""" +from __future__ import annotations + +from datetime import datetime, timezone, timedelta +from unittest.mock import MagicMock + +import pytest + +from src.llm.context_builder import _is_stats_available + + +def _make_stats(available_at): + s = MagicMock() + s.available_at = available_at + return s + + +class TestIsStatsAvailable: + """_is_stats_available 回测防泄漏语义。""" + + def test_none_before_allows_none_available_at(self): + """实盘模式(before=None): available_at 为 None 时允许(兼容旧数据)。""" + stats = _make_stats(available_at=None) + assert _is_stats_available(stats, before=None) is True + + def test_cutoff_with_none_available_at_is_unavailable(self): + """回测模式(before=cutoff): available_at 为 None → 不可用(保守)。 + + 这是核心修复:防止无血缘时间的后期回填数据进入回测。 + """ + stats = _make_stats(available_at=None) + cutoff = datetime(2026, 1, 13, 20, 0, tzinfo=timezone.utc) + assert _is_stats_available(stats, before=cutoff) is False + + def test_available_at_after_cutoff_is_unavailable(self): + """available_at 在 cutoff 之后 → 不可用。""" + cutoff = datetime(2026, 1, 13, 20, 0, tzinfo=timezone.utc) + stats = _make_stats(available_at=cutoff + timedelta(hours=1)) + assert _is_stats_available(stats, before=cutoff) is False + + def test_available_at_before_cutoff_is_available(self): + """available_at 在 cutoff 之前 → 可用。""" + cutoff = datetime(2026, 1, 13, 20, 0, tzinfo=timezone.utc) + stats = _make_stats(available_at=cutoff - timedelta(hours=1)) + assert _is_stats_available(stats, before=cutoff) is True + + def test_available_at_equals_cutoff_is_available(self): + """available_at == cutoff → 可用(边界包含)。""" + cutoff = datetime(2026, 1, 13, 20, 0, tzinfo=timezone.utc) + stats = _make_stats(available_at=cutoff) + assert _is_stats_available(stats, before=cutoff) is True + + def test_real_world_scenario_backtest_avoids_future_data(self): + """真实场景:回测时,赛后才生成的统计数据不应出现。 + + 比赛:2026-01-15 20:00 + cutoff(回测):2026-01-14 20:00(赛前 1 天) + stats 在赛后才生成(available_at=2026-01-15 22:00) + → 不可用 + """ + cutoff = datetime(2026, 1, 14, 20, 0, tzinfo=timezone.utc) + stats = _make_stats(available_at=datetime(2026, 1, 15, 22, 0, tzinfo=timezone.utc)) + assert _is_stats_available(stats, before=cutoff) is False + + +class TestBzzoirotAvailableAt: + """验证 bzzoiro.py 写入 available_at 使用 match_date 而非 now。""" + + def test_bzzoiro_sets_available_at_from_match_date(self): + """bzzoiro.py 应在创建 stats 时使用 nm.date 作为 available_at。""" + import inspect + from src.data import bzzoiro + + source = inspect.getsource(bzzoiro) + # 验证:存在 available_at = nm.date 的逻辑 + assert 'available_at = nm.date if nm.date else now' in source, \ + "bzzoiro.py 应使用 nm.date 作为 available_at" + + def test_bzzoiro_existing_match_uses_match_date(self): + """bzzoiro.py 更新已有比赛时也应用 nm.date。""" + import inspect + from src.data import bzzoiro + + source = inspect.getsource(bzzoiro) + # 验证两处都更新 + count = source.count('available_at = nm.date if nm.date else now') + assert count == 2, f"期望 2 处使用 nm.date,实际 {count} 处" + + +class TestUnderstatAvailableAt: + """验证 understat.py 写入 available_at 使用 match_date。""" + + def test_understat_sets_available_at_from_match_date(self): + """understat.py 应使用 existing.match_date 作为 available_at。""" + import inspect + from src.data import understat + + source = inspect.getsource(understat) + assert 'available_at = match_date' in source, \ + "understat.py 应使用 match_date 作为 available_at" diff --git a/tests/test_bzzoirot_stats.py b/tests/test_bzzoirot_stats.py new file mode 100644 index 0000000..93f50b6 --- /dev/null +++ b/tests/test_bzzoirot_stats.py @@ -0,0 +1,206 @@ +"""回归测试: bzzoiro 采集链路正确映射射门/控球/角球/xG。 + +验证: + 1. normalize_bzzoiro 正确映射统计字段 + 2. 入库条件不再强制要求 xG(任一统计字段即可) + 3. API 没有的字段保持 None,不伪造 +""" +from __future__ import annotations + +from datetime import datetime, timezone +from unittest.mock import MagicMock + +import pytest + +from src.data.normalize import NormalizedMatch, normalize_bzzoiro + + +class TestNormalizeBzzoirotStats: + """normalize_bzzoiro 应正确映射统计字段。""" + + def test_maps_shots(self): + """API 提供 shots 字段时应正确映射。""" + raw = { + "event_date": "2026-01-15T15:00:00Z", + "status": "finished", + "home_team": "Man City", + "away_team": "Man United", + "home_score": 2, + "away_score": 1, + "home_shots": 15, + "away_shots": 8, + "home_shots_on_target": 6, + "away_shots_on_target": 3, + } + m = normalize_bzzoiro(raw, "E0") + assert m is not None + assert m.home_shots == 15 + assert m.away_shots == 8 + assert m.home_shots_on_target == 6 + assert m.away_shots_on_target == 3 + + def test_maps_corners(self): + """API 提供 corners 字段时应正确映射。""" + raw = { + "event_date": "2026-01-15T15:00:00Z", + "status": "finished", + "home_team": "Liverpool", + "away_team": "Chelsea", + "home_score": 1, + "away_score": 1, + "home_corners": 7, + "away_corners": 4, + } + m = normalize_bzzoiro(raw, "E0") + assert m is not None + assert m.home_corners == 7 + assert m.away_corners == 4 + + def test_maps_possession(self): + """API 提供 possession 字段时应正确映射。""" + raw = { + "event_date": "2026-01-15T15:00:00Z", + "status": "finished", + "home_team": "Barcelona", + "away_team": "Real Madrid", + "home_score": 2, + "away_score": 0, + "home_possession": 62.5, + } + m = normalize_bzzoiro(raw, "SP1") + assert m is not None + assert m.home_possession == 62.5 + + def test_maps_xg(self): + """API 提供 xG 字段时应正确映射。""" + raw = { + "event_date": "2026-01-15T15:00:00Z", + "status": "finished", + "home_team": "Bayern", + "away_team": "Dortmund", + "home_score": 3, + "away_score": 1, + "home_xg": 2.5, + "away_xg": 0.8, + } + m = normalize_bzzoiro(raw, "D1") + assert m is not None + assert m.home_xg == 2.5 + assert m.away_xg == 0.8 + + def test_maps_cards(self): + """API 提供 cards 字段时应正确映射。""" + raw = { + "event_date": "2026-01-15T15:00:00Z", + "status": "finished", + "home_team": "Arsenal", + "away_team": "Tottenham", + "home_score": 1, + "away_score": 0, + "home_yellow_cards": 2, + "away_yellow_cards": 3, + "home_red_cards": 0, + "away_red_cards": 1, + } + m = normalize_bzzoiro(raw, "E0") + assert m is not None + assert m.home_yellow_cards == 2 + assert m.away_yellow_cards == 3 + assert m.home_red_cards == 0 + assert m.away_red_cards == 1 + + def test_missing_stats_stay_none(self): + """API 没有统计字段时应保持 None,不伪造。""" + raw = { + "event_date": "2026-01-15T15:00:00Z", + "status": "finished", + "home_team": "A", + "away_team": "B", + "home_score": 1, + "away_score": 0, + } + m = normalize_bzzoiro(raw, "E0") + assert m is not None + # 无比分数据时不应有统计字段 + assert m.home_shots is None + assert m.away_shots is None + assert m.home_xg is None + assert m.away_xg is None + assert m.home_possession is None + assert m.home_corners is None + + def test_alternative_field_names(self): + """API 使用替代字段名时也应正确映射。""" + raw = { + "event_date": "2026-01-15T15:00:00Z", + "status": "finished", + "home_team": "A", + "away_team": "B", + "home_score": 1, + "away_score": 0, + "shots_home": 12, + "shots_away": 6, + "sot_home": 5, + "sot_away": 2, + "corners_home": 8, + "corners_away": 3, + "xg_home": 1.8, + "xg_away": 0.5, + } + m = normalize_bzzoiro(raw, "E0") + assert m is not None + assert m.home_shots == 12 + assert m.away_shots == 6 + assert m.home_shots_on_target == 5 + assert m.away_shots_on_target == 2 + assert m.home_corners == 8 + assert m.away_corners == 3 + assert m.home_xg == 1.8 + assert m.away_xg == 0.5 + + +class TestIngestionCondition: + """入库条件应不再强制要求 xG。""" + + def test_any_stat_field_triggers_stats_creation(self): + """任一统计字段存在即可触发 MatchStats 创建。""" + # 模拟 normalize 后的结果 + nm = NormalizedMatch( + league_type="E0", + date=datetime(2026, 1, 15, 15, 0, tzinfo=timezone.utc), + home_team="A", + away_team="B", + match_status="finished", + home_goals=1, + away_goals=0, + # 只有 shots,无 xG + home_shots=10, + away_shots=5, + ) + # 验证:任一统计字段存在 + has_stats = ( + nm.home_xg is not None or nm.away_xg is not None + or nm.home_shots is not None or nm.away_shots is not None + or nm.home_corners is not None or nm.away_corners is not None + or nm.home_possession is not None + ) + assert has_stats is True, "有 shots 时应视为有统计数据" + + def test_no_stats_means_no_match_stats(self): + """没有任何统计字段时不应创建 MatchStats。""" + nm = NormalizedMatch( + league_type="E0", + date=datetime(2026, 1, 15, 15, 0, tzinfo=timezone.utc), + home_team="A", + away_team="B", + match_status="finished", + home_goals=1, + away_goals=0, + ) + has_stats = ( + nm.home_xg is not None or nm.away_xg is not None + or nm.home_shots is not None or nm.away_shots is not None + or nm.home_corners is not None or nm.away_corners is not None + or nm.home_possession is not None + ) + assert has_stats is False, "无统计字段时不应创建 MatchStats" diff --git a/tests/test_injuries_integrity_rollback.py b/tests/test_injuries_integrity_rollback.py new file mode 100644 index 0000000..8e0bedf --- /dev/null +++ b/tests/test_injuries_integrity_rollback.py @@ -0,0 +1,172 @@ +"""回归测试: injuries IntegrityError 处理不再整批回滚。 + +模拟场景:连续插入多条伤停记录,中间一批触发 IntegrityError, +断言其它批次记录不会丢失。 +""" +from __future__ import annotations + +import asyncio +from unittest.mock import MagicMock, AsyncMock, patch + +import pytest + +from src.data import injuries as inj_mod + + +class FakeNestedCtx: + """模拟 SQLAlchemy begin_nested() 上下文。 + + __enter__:标记进入 savepoint + __exit__:如果有异常,模拟 ROLLBACK TO SAVEPOINT(不清空已 flush 的对象) + """ + + def __init__(self, session): + self.session = session + self.rolled_back = False + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + if exc_type is not None: + # ROLLBACK TO SAVEPOINT — 不清空 session 中已存在的对象 + self.rolled_back = True + return True # suppress exception + + +class FakeSession: + """模拟 AsyncSession,记录 flush 调用和 begin_nested 使用。""" + + def __init__(self, fail_on_flush_indices: set[int] | None = None): + self.flush_count = 0 + self.nested_count = 0 + self.flushed_records: list[dict] = [] + self.added_records: list[dict] = [] + self.fail_on = fail_on_flush_indices or set() + + async def execute(self, stmt): + class Result: + def all(self_inner): + return [] + return Result() + + async def get(self, cls, id): + return None + + def add(self, obj): + self.added_records.append(obj) + + async def flush(self): + self.flush_count += 1 + if self.flush_count in self.fail_on: + from sqlalchemy.exc import IntegrityError + raise IntegrityError("mock duplicate", None, None) + + @property + def _nested_ctx(self): + return FakeNestedCtx(self) + + def begin_nested(self): + self.nested_count += 1 + return self._nested_ctx + + +@pytest.mark.asyncio +async def test_integrity_error_does_not_lose_other_batches(): + """核心测试:一批触发 IntegrityError,其它批次记录不丢失。 + + 场景:3 批记录,第 2 批 flush 时 IntegrityError。 + 断言:第 1 批和第 3 批的记录仍存在于 flushed_records 中。 + """ + session = FakeSession(fail_on_flush_indices={2}) # 第 2 次 flush 失败 + + # 构造 3 批记录,每批 2 条(BATCH_SIZE 用 2 方便测试) + pending = [ + {"player_id": i, "player_name": f"Player{i}", "team_id": 1, + "fixture_id": 100 + i, "injury_type": "Hamstring", + "reason": "strain", "injury_date": None, "return_date": None} + for i in range(6) + ] + + # 临时覆盖 BATCH_SIZE + original_batch_size = 50 + try: + inj_mod.ingest_injuries.__globals__['__dict__'] # no-op + + # 手动模拟 ingest_injuries 的核心逻辑 + batch = [] + flushed_ids = [] + errors = [] + + async def _flush_batch(): + if not batch: + return + async with session.begin_nested(): + for obj in batch: + session.add(obj) + await session.flush() + flushed_ids.extend([r["player_id"] for r in batch]) + batch.clear() + + for rec in pending: + batch.append(rec) + if len(batch) >= 2: # BATCH_SIZE = 2 + try: + await _flush_batch() + except Exception: + batch.clear() + continue + + # 最终 flush + try: + await _flush_batch() + except Exception: + batch.clear() + + except Exception: + pass + + # 断言:flush 成功的记录是第 1 批(id=0,1)和第 3 批(id=4,5) + # 第 2 批(id=2,3)因 IntegrityError 被 savepoint 回滚 + # 关键:第 1 批和第 3 批的记录必须仍在 flushed_ids 中 + assert 0 in flushed_ids, "第 1 批记录 0 不应丢失" + assert 1 in flushed_ids, "第 1 批记录 1 不应丢失" + assert 4 in flushed_ids or 5 in flushed_ids, "第 3 批记录不应丢失" + # 第 2 批(flush 失败的)不应在 flushed_ids 中 + assert 2 not in flushed_ids, "第 2 批应被回滚" + assert 3 not in flushed_ids, "第 2 批应被回滚" + print("PASS: IntegrityError 只回滚失败批次,其它批次不丢失") + + +@pytest.mark.asyncio +async def test_begin_nested_is_used(): + """验证 begin_nested() 被调用(而非全事务 rollback)。""" + session = FakeSession() + + batch = [{"player_id": i, "player_name": f"P{i}", "team_id": 1, + "fixture_id": 100 + i, "injury_type": "None", + "reason": None, "injury_date": None, "return_date": None} + for i in range(3)] + + async def _flush_batch(): + if not batch: + return + async with session.begin_nested(): + for obj in batch: + session.add(obj) + await session.flush() + batch.clear() + + try: + await _flush_batch() + except Exception: + pass + + # 验证 begin_nested 被调用(说明使用了 savepoint) + assert session.nested_count >= 1, "应使用 begin_nested(SAVEPOINT)" + print(f"PASS: begin_nested 被调用 {session.nested_count} 次") + + +if __name__ == "__main__": + asyncio.run(test_integrity_error_does_not_lose_other_batches()) + asyncio.run(test_begin_nested_is_used()) diff --git a/tests/test_injuries_slice_semantic.py b/tests/test_injuries_slice_semantic.py new file mode 100644 index 0000000..c0dc182 --- /dev/null +++ b/tests/test_injuries_slice_semantic.py @@ -0,0 +1,110 @@ +"""回归测试: 伤停切片区分「查询成功但无人伤停」与「无数据/未接入」。 + +验证: + 1. 查询成功 + 空结果 → has_data=True + 2. 源未配置 → has_data=False + 3. 查询异常 → has_data=False + 4. 查询成功 + 有数据 → has_data=True +""" +from __future__ import annotations + +from datetime import datetime, timezone +from unittest.mock import MagicMock, AsyncMock, patch + +import pytest + +from src.data.injuries import InjuryQueryResult, get_injuries_for_match +from src.llm.context_builder import MatchHeader, injuries_slice + + +def _make_header(): + return MatchHeader( + match_id=999, home_name="A", away_name="B", + league_name="X", season=None, match_date="?", + match_date=None, stage=None, + home_team_id=1, away_team_id=2, league_id=1, + ) + + +class TestInjuryQueryResult: + """InjuryQueryResult 基础属性。""" + + def test_has_data_success(self): + result = InjuryQueryResult(records=[], query_status="success") + assert result.has_data is True + + def test_has_data_source_not_configured(self): + result = InjuryQueryResult(records=[], query_status="source_not_configured") + assert result.has_data is False + + def test_has_data_query_error(self): + result = InjuryQueryResult(records=[], query_status="query_error") + assert result.has_data is False + + +class TestInjuriesSliceEmptyVsNotConfigured: + """injuries_slice 应区分「查询成功但为空」与「无数据/未接入」。""" + + @pytest.mark.asyncio + async def test_empty_result_has_data_true(self): + """查询成功 + 空结果 → has_data=True,文案显示「当前无伤停记录」。""" + header = _make_header() + + # Mock get_injuries_for_match 返回成功但空的结果 + async def mock_query(db, team_id, match_date, as_of=None): + return InjuryQueryResult(records=[], query_status="success") + + with patch("src.llm.context_builder.get_injuries_for_match", mock_query): + result = await injuries_slice(header, before=None) + + assert result.has_data is True, "查询成功+空结果应 has_data=True" + assert "当前无伤停记录" in result.text, "文案应表明无伤停" + + @pytest.mark.asyncio + async def test_source_not_configured_has_data_false(self): + """源未配置 → has_data=False。""" + header = _make_header() + + async def mock_query(db, team_id, match_date, as_of=None): + return InjuryQueryResult(records=[], query_status="source_not_configured") + + with patch("src.llm.context_builder.get_injuries_for_match", mock_query): + result = await injuries_slice(header, before=None) + + assert result.has_data is False, "源未配置应 has_data=False" + assert "伤停源未配置" in result.text + + @pytest.mark.asyncio + async def test_query_error_has_data_false(self): + """查询异常 → has_data=False。""" + header = _make_header() + + async def mock_query(db, team_id, match_date, as_of=None): + return InjuryQueryResult(records=[], query_status="query_error") + + with patch("src.llm.context_builder.get_injuries_for_match", mock_query): + result = await injuries_slice(header, before=None) + + assert result.has_data is False, "查询异常应 has_data=False" + assert "查询异常" in result.text + + @pytest.mark.asyncio + async def test_with_records_has_data_true(self): + """查询成功 + 有数据 → has_data=True。""" + header = _make_header() + + mock_inj = MagicMock() + mock_inj.reason = "Hamstring" + mock_inj.injury_type = None + mock_inj.player_name = "Player A" + + async def mock_query(db, team_id, match_date, as_of=None): + if team_id == 1: + return InjuryQueryResult(records=[mock_inj], query_status="success") + return InjuryQueryResult(records=[], query_status="success") + + with patch("src.llm.context_builder.get_injuries_for_match", mock_query): + result = await injuries_slice(header, before=None) + + assert result.has_data is True + assert "Player A" in result.text diff --git a/tests/test_ip_spoofing.py b/tests/test_ip_spoofing.py new file mode 100644 index 0000000..2684aee --- /dev/null +++ b/tests/test_ip_spoofing.py @@ -0,0 +1,88 @@ +"""回归测试: 限流与登录防爆破在不可信 X-Forwarded-For 下的 IP 伪造问题。 + +验证: + 1. TRUST_PROXY_HEADERS=False(默认)时忽略伪造的 X-Forwarded-For + 2. TRUST_PROXY_HEADERS=True 时解析 X-Forwarded-For + 3. 限流与登录共用同一套 IP 提取逻辑 +""" +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest + +from src.api.deps import get_client_ip +from src.core.config import Settings + + +class TestGetClientIP: + """get_client_ip 防伪造逻辑。""" + + def _make_request(self, client_host: str | None, xff: str | None = None): + req = MagicMock() + req.client = MagicMock(host=client_host) if client_host else None + req.headers = {} + if xff is not None: + req.headers["X-Forwarded-For"] = xff + return req + + @patch("src.api.deps.settings", Settings(TRUST_PROXY_HEADERS=False)) + def test_untrusted_proxy_ignores_xff(self): + """TRUST_PROXY_HEADERS=False 时忽略伪造的 X-Forwarded-For。""" + # 客户端伪造 X-Forwarded-For,但 TRUST_PROXY_HEADERS=False + req = self._make_request("1.2.3.4", xff="10.0.0.1, 192.168.1.1") + ip = get_client_ip(req) + assert ip == "1.2.3.4", f"应使用连接层 IP,实际 {ip}" + + @patch("src.api.deps.settings", Settings(TRUST_PROXY_HEADERS=True)) + def test_trusted_proxy_parses_xff(self): + """TRUST_PROXY_HEADERS=True 时解析 X-Forwarded-For 第一个 IP。""" + req = self._make_request("127.0.0.1", xff="10.0.0.1, 192.168.1.1") + ip = get_client_ip(req) + assert ip == "10.0.0.1", f"应使用 XFF 第一个 IP,实际 {ip}" + + @patch("src.api.deps.settings", Settings(TRUST_PROXY_HEADERS=True)) + def test_trusted_proxy_without_xff(self): + """TRUST_PROXY_HEADERS=True 但无 XFF 头时回退到 client.host。""" + req = self._make_request("1.2.3.4", xff=None) + ip = get_client_ip(req) + assert ip == "1.2.3.4", f"应回退到连接层 IP,实际 {ip}" + + @patch("src.api.deps.settings", Settings(TRUST_PROXY_HEADERS=False)) + def test_untrusted_proxy_no_client(self): + """TRUST_PROXY_HEADERS=False 且无 client 时返回 unknown。""" + req = self._make_request(None, xff="10.0.0.1") + ip = get_client_ip(req) + assert ip == "unknown", f"应返回 unknown,实际 {ip}" + + @patch("src.api.deps.settings", Settings(TRUST_PROXY_HEADERS=True)) + def test_trusted_proxy_single_ip(self): + """TRUST_PROXY_HEADERS=True 且 XFF 只有一个 IP。""" + req = self._make_request("127.0.0.1", xff="10.0.0.1") + ip = get_client_ip(req) + assert ip == "10.0.0.1", f"应返回 10.0.0.1,实际 {ip}" + + +class TestRateLimitIPSpoofing: + """验证限流使用 get_client_ip 防伪造。""" + + @patch("src.api.deps.settings", Settings(TRUST_PROXY_HEADERS=False)) + def test_rate_limit_ignores_spoofed_xff(self): + """限流在 TRUST_PROXY_HEADERS=False 时不受 XFF 伪造影响。""" + from src.api.deps import _RateLimiter, get_client_ip + + limiter = _RateLimiter(max_requests=10, window_seconds=60) + + # 模拟不同伪造 XFF,但真实 IP 相同 + def make_request(spoofed_xff): + req = MagicMock() + req.client = MagicMock(host="1.2.3.4") + req.headers = {"X-Forwarded-For": spoofed_xff} + return req + + # 伪造不同 XFF,但真实 IP 都是 1.2.3.4 + for i in range(10): + req = make_request(f"10.0.0.{i}") + ip = get_client_ip(req) + assert ip == "1.2.3.4", f"迭代 {i}: 应返回 1.2.3.4,实际 {ip}" + assert limiter.is_allowed(ip), f"迭代 {i}: 应允许" diff --git a/tests/test_multi_agent_degraded.py b/tests/test_multi_agent_degraded.py new file mode 100644 index 0000000..cd8c87a --- /dev/null +++ b/tests/test_multi_agent_degraded.py @@ -0,0 +1,246 @@ +"""回归测试: multi-agent 全专家失败/无数据时 status=degraded。 + +验证: + 1. 5 个专家全 no_data/error → status="degraded",不调终裁 + 2. 至少 1 个专家 ok → status="success",正常走终裁 +""" +from __future__ import annotations + +from unittest.mock import MagicMock, AsyncMock, patch + +import pytest + +from src.llm.agents.base import AgentReport +from src.llm.agents import orchestrator as orch_mod + + +def _make_header(): + from src.llm.context_builder import MatchHeader + return MatchHeader( + match_id=999, home_name="A", away_name="B", + league_name="X", season=None, match_date="?", + match_dt=None, stage=None, + home_team_id=1, away_team_id=2, league_id=1, + ) + + +def _all_error_reports(): + """5 个专家全 error.""" + return [ + AgentReport(agent="form", status="error", analysis="slice failed"), + AgentReport(agent="stats", status="error", analysis="slice failed"), + AgentReport(agent="home_away", status="error", analysis="slice failed"), + AgentReport(agent="injuries", status="error", analysis="slice failed"), + AgentReport(agent="h2h", status="error", analysis="slice failed"), + ] + + +def _all_no_data_reports(): + """5 个专家全 no_data.""" + return [ + AgentReport(agent="form", status="no_data", analysis="无数据"), + AgentReport(agent="stats", status="no_data", analysis="无数据"), + AgentReport(agent="home_away", status="no_data", analysis="无数据"), + AgentReport(agent="injuries", status="no_data", analysis="无数据"), + AgentReport(agent="h2h", status="no_data", analysis="无数据"), + ] + + +def _mixed_reports(): + """1 个 ok,4 个 error.""" + return [ + AgentReport(agent="form", status="ok", analysis="good"), + AgentReport(agent="stats", status="error", analysis="failed"), + AgentReport(agent="home_away", status="no_data", analysis="无数据"), + AgentReport(agent="injuries", status="error", analysis="failed"), + AgentReport(agent="h2h", status="no_data", analysis="无数据"), + ] + + +class TestAllExpertsFailed: + """全部专家失败/无数据时,status 应为 degraded。""" + + @pytest.mark.asyncio + async def test_all_error_reports_yields_degraded(self): + """5 个专家全 error → status=degraded,不调终裁。""" + header = _make_header() + + # Mock run_specialists 返回全 error + async def mock_run_specialists(header, *, version, before): + return _all_error_reports() + + # Mock _agent_provider + async def mock_agent_provider(agent_id, *, tier): + return MagicMock(model="test-model") + + # Mock load_match_header + async def mock_load_header(mid, db=None): + return header + + # Mock _upsert_prediction — 捕获写入的 status + captured_status = {} + + async def mock_upsert(session, **kw): + captured_status.update(kw.get("values", {})) + mock_pred = MagicMock() + mock_pred.id = 1 + mock_pred.provider = "test" + mock_pred.model = "test" + mock_pred.prompt_version = "v1" + mock_pred.pred_home_goals = None + mock_pred.pred_away_goals = None + mock_pred.alt_pred_home_goals = None + mock_pred.alt_pred_away_goals = None + mock_pred.pred_1x2 = None + mock_pred.subjective_confidence = None + mock_pred.reasoning = kw["values"].get("reasoning") + mock_pred.agent_outputs = [] + return mock_pred + + # Mock get_uow + class FakeUow: + async def __aenter__(self): + return self + async def __aexit__(self, *a): + pass + async def get(self, cls, id): + return MagicMock() # match exists + + with patch.object(orch_mod, "run_specialists", mock_run_specialists), \ + patch.object(orch_mod, "_agent_provider", mock_agent_provider), \ + patch.object(orch_mod, "load_match_header", mock_load_header), \ + patch.object(orch_mod, "_upsert_prediction", mock_upsert), \ + patch.object(orch_mod, "get_uow", FakeUow): + + result = await orch_mod.predict_match_multi(999) + + # 断言:status 是 degraded,不是 success + assert captured_status.get("status") == "degraded", \ + f"期望 status=degraded,实际 {captured_status.get('status')}" + # 断言:reasoning 包含说明 + assert "专家" in captured_status.get("reasoning", ""), \ + f"reasoning 应说明原因,实际 {captured_status.get('reasoning')}" + # 断言:pred_* 全为 None + assert captured_status.get("pred_home_goals") is None + assert captured_status.get("pred_1x2") is None + print(f"PASS: 全 error → status={captured_status.get('status')}") + print(f" reasoning={captured_status.get('reasoning')[:50]}...") + + @pytest.mark.asyncio + async def test_all_no_data_reports_yields_degraded(self): + """5 个专家全 no_data → status=degraded,不调终裁。""" + header = _make_header() + + async def mock_run_specialists(header, *, version, before): + return _all_no_data_reports() + + async def mock_agent_provider(agent_id, *, tier): + return MagicMock(model="test-model") + + async def mock_load_header(mid, db=None): + return header + + captured_status = {} + + async def mock_upsert(session, **kw): + captured_status.update(kw.get("values", {})) + mock_pred = MagicMock() + mock_pred.id = 1 + mock_pred.provider = "test" + mock_pred.model = "test" + mock_pred.prompt_version = "v1" + mock_pred.pred_home_goals = None + mock_pred.pred_away_goals = None + mock_pred.alt_pred_home_goals = None + mock_pred.alt_pred_away_goals = None + mock_pred.pred_1x2 = None + mock_pred.subjective_confidence = None + mock_pred.reasoning = kw["values"].get("reasoning") + mock_pred.agent_outputs = [] + return mock_pred + + class FakeUow: + async def __aenter__(self): + return self + async def __aexit__(self, *a): + pass + async def get(self, cls, id): + return MagicMock() + + with patch.object(orch_mod, "run_specialists", mock_run_specialists), \ + patch.object(orch_mod, "_agent_provider", mock_agent_provider), \ + patch.object(orch_mod, "load_match_header", mock_load_header), \ + patch.object(orch_mod, "_upsert_prediction", mock_upsert), \ + patch.object(orch_mod, "get_uow", FakeUow): + + result = await orch_mod.predict_match_multi(999) + + assert captured_status.get("status") == "degraded", \ + f"期望 status=degraded,实际 {captured_status.get('status')}" + print(f"PASS: 全 no_data → status={captured_status.get('status')}") + + +class TestPartialExpertsOk: + """部分专家 ok 时,status 仍可为 success。""" + + @pytest.mark.asyncio + async def test_one_ok_report_allows_success(self): + """1 个 ok + 4 个 error → status=success(走终裁)。""" + header = _make_header() + + async def mock_run_specialists(header, *, version, before): + return _mixed_reports() + + async def mock_agent_provider(agent_id, *, tier): + return MagicMock(model="test-model") + + async def mock_load_header(mid, db=None): + return header + + captured_status = {} + + async def mock_aggregator(header, reports, *, provider, version): + # 终裁返回合法 JSON + return { + "pred_home_goals": 2, + "pred_away_goals": 1, + "pred_1x2": "1", + "subjective_confidence": 0.7, + "reasoning": "test prediction", + "agent_weights": {"form": 0.5}, + }, 100, 50 + + async def mock_upsert(session, **kw): + captured_status.update(kw.get("values", {})) + mock_pred = MagicMock() + mock_pred.id = 1 + mock_pred.provider = "test" + mock_pred.model = "test" + mock_pred.prompt_version = "v1" + mock_pred.pred_home_goals = 2 + mock_pred.pred_away_goals = 1 + mock_pred.pred_1x2 = "1" + mock_pred.subjective_confidence = 0.7 + mock_pred.reasoning = "test" + return mock_pred + + class FakeUow: + async def __aenter__(self): + return self + async def __aexit__(self, *a): + pass + async def get(self, cls, id): + return MagicMock() + + with patch.object(orch_mod, "run_specialists", mock_run_specialists), \ + patch.object(orch_mod, "_agent_provider", mock_agent_provider), \ + patch.object(orch_mod, "load_match_header", mock_load_header), \ + patch.object(orch_mod, "_upsert_prediction", mock_upsert), \ + patch.object(orch_mod, "run_aggregator", mock_aggregator), \ + patch.object(orch_mod, "get_uow", FakeUow): + + result = await orch_mod.predict_match_multi(999) + + assert captured_status.get("status") == "success", \ + f"期望 status=success,实际 {captured_status.get('status')}" + print(f"PASS: 1 ok + 4 error → status={captured_status.get('status')}") diff --git a/tests/test_require_admin_fail_closed.py b/tests/test_require_admin_fail_closed.py new file mode 100644 index 0000000..be5ede1 --- /dev/null +++ b/tests/test_require_admin_fail_closed.py @@ -0,0 +1,82 @@ +"""回归测试: 生产环境管理接口鉴权 fail-closed。 + +验证: + 1. REQUIRE_ADMIN_AUTH=True + 未配置 → 拒绝(503) + 2. APP_ENV=production + 未配置 → 拒绝(503) + 3. development + 未配置 → 放行(fail-open + warning) + 4. 已配置密码 → 正常验证路径不受影响 +""" +from __future__ import annotations + +from unittest.mock import AsyncMock, patch + +import pytest +from fastapi import HTTPException + +from src.api.deps import require_admin +from src.core.config import Settings + + +class TestRequireAdminFailClosed: + """生产环境 fail-closed 逻辑。""" + + @pytest.mark.asyncio + async def test_require_admin_auth_true_rejects_when_unconfigured(self): + """REQUIRE_ADMIN_AUTH=True + 未配置 → 503 拒绝。""" + mock_request = AsyncMock() + mock_request.cookies = {} + mock_request.headers = {} + + with patch("src.api.deps.settings", Settings(REQUIRE_ADMIN_AUTH=True, APP_ENV="development")), \ + patch("src.api.deps.auth_configured", AsyncMock(return_value=False)): + + with pytest.raises(HTTPException) as exc_info: + await require_admin(mock_request) + + assert exc_info.value.status_code == 503 + assert "未配置" in exc_info.value.detail or "鉴权" in exc_info.value.detail + + @pytest.mark.asyncio + async def test_production_env_rejects_when_unconfigured(self): + """APP_ENV=production + 未配置 → 503 拒绝。""" + mock_request = AsyncMock() + mock_request.cookies = {} + mock_request.headers = {} + + with patch("src.api.deps.settings", Settings(REQUIRE_ADMIN_AUTH=False, APP_ENV="production")), \ + patch("src.api.deps.auth_configured", AsyncMock(return_value=False)): + + with pytest.raises(HTTPException) as exc_info: + await require_admin(mock_request) + + assert exc_info.value.status_code == 503 + + @pytest.mark.asyncio + async def test_development_env_allows_when_unconfigured(self): + """development + 未配置 → fail-open 放行。""" + mock_request = AsyncMock() + mock_request.cookies = {} + mock_request.headers = {} + + with patch("src.api.deps.settings", Settings(REQUIRE_ADMIN_AUTH=False, APP_ENV="development")), \ + patch("src.api.deps.auth_configured", AsyncMock(return_value=False)): + + # 不应抛异常 + await require_admin(mock_request) + + @pytest.mark.asyncio + async def test_configured_password_works_normally(self): + """已配置密码 → 正常验证路径(401 未登录,而非 503)。""" + mock_request = AsyncMock() + mock_request.cookies = {} # 无 cookie + mock_request.headers = {} # 无 API Key + + with patch("src.api.deps.settings", Settings(REQUIRE_ADMIN_AUTH=True, APP_ENV="production")), \ + patch("src.api.deps.auth_configured", AsyncMock(return_value=True)), \ + patch("src.api.deps.get_session_secret", AsyncMock(return_value=b"secret")): + + with pytest.raises(HTTPException) as exc_info: + await require_admin(mock_request) + + # 已配置 → 401(未登录),不是 503(未配置) + assert exc_info.value.status_code == 401 diff --git a/tests/test_team_names_normalize.py b/tests/test_team_names_normalize.py new file mode 100644 index 0000000..a058bbf --- /dev/null +++ b/tests/test_team_names_normalize.py @@ -0,0 +1,71 @@ +"""回归测试: team_names.normalize NFKD 去变音导致同一俱乐部映射成两个队名。 + +验证: + 1. München / Munich / Munchen → 同一规范名 + 2. Atlético / Atletico → 同一规范名 + 3. Köln / Koln → 同一规范名 + 4. Mönchengladbach / Monchengladbach / M'gladbach → 同一规范名 +""" +from __future__ import annotations + +from src.data.team_names import normalize + + +class TestNormalizeVariants: + """同一俱乐部的不同变音写法应映射到同一规范名。""" + + def test_bayern_munich_variants(self): + """Bayern München / Munich / Munchen → 同一规范名。""" + canonical = normalize("Bayern München") + assert canonical == "Bayern München" + assert normalize("Bayern Munich") == canonical + assert normalize("Bayern Munchen") == canonical + + def test_atletico_madrid_variants(self): + """Atlético Madrid / Atletico Madrid → 同一规范名。""" + canonical = normalize("Atlético Madrid") + assert canonical == "Atlético Madrid" + assert normalize("Atletico Madrid") == canonical + + def test_koln_variants(self): + """FC Köln / FC Koln → 同一规范名。""" + canonical = normalize("FC Köln") + assert canonical == "FC Köln" + assert normalize("FC Koln") == canonical + + def test_monchengladbach_variants(self): + """Mönchengladbach / Monchengladbach / M'gladbach → 同一规范名。""" + canonical = normalize("Borussia Mönchengladbach") + assert canonical == "Borussia Mönchengladbach" + assert normalize("Borussia Monchengladbach") == canonical + assert normalize("Borussia M'gladbach") == canonical + + def test_leganes_variants(self): + """Leganés / Leganes → 同一规范名。""" + canonical = normalize("Leganés") + assert canonical == "Leganés" + assert normalize("Leganes") == canonical + + def test_alaves_variants(self): + """Deportivo Alavés / Alaves → 同一规范名。""" + canonical = normalize("Deportivo Alavés") + assert canonical == "Deportivo Alavés" + assert normalize("Deportivo Alaves") == canonical + + def test_saint_etienne_variants(self): + """AS Saint-Étienne / Saint-Etienne → 同一规范名。""" + canonical = normalize("AS Saint-Étienne") + assert canonical == "AS Saint-Étienne" + assert normalize("Saint-Etienne") == canonical + + def test_empty_and_none(self): + """空字符串应返回空。""" + assert normalize("") == "" + assert normalize(None) == "" + + def test_original_name_takes_priority(self): + """原始名(带变音)应优先于 NFKD 去变音名。""" + # "Bayern München" 在 map 中有键,应直接命中 + # 而不是先 NFKD 成 "Bayern Munchen" 再查 + result = normalize("Bayern München") + assert result == "Bayern München"