diff --git a/alembic/versions/0020_team_aliases.py b/alembic/versions/0020_team_aliases.py new file mode 100644 index 0000000..c6e8d57 --- /dev/null +++ b/alembic/versions/0020_team_aliases.py @@ -0,0 +1,32 @@ +"""球队别名表 team_aliases + +Revision ID: 0020_team_aliases +Revises: 0019_ingest_jobs +Create Date: 2026-09-22 +""" + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + +revision: str = '0020_team_aliases' +down_revision: Union[str, None] = '0019_ingest_jobs' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + 'team_aliases', + sa.Column('alias_normalized', sa.String(120), primary_key=True), + sa.Column('team_id', sa.Integer, sa.ForeignKey('teams.id', ondelete='CASCADE'), nullable=False), + sa.Column('original_alias', sa.String(120), nullable=False), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now()), + ) + op.create_index('ix_team_aliases_team_id', 'team_aliases', ['team_id']) + + +def downgrade() -> None: + op.drop_index('ix_team_aliases_team_id', table_name='team_aliases') + op.drop_table('team_aliases') diff --git a/docs/05-data.md b/docs/05-data.md index e1525dd..76a7d90 100644 --- a/docs/05-data.md +++ b/docs/05-data.md @@ -60,7 +60,7 @@ ### 队名归一化 -`src/data/team_names.py` 维护 `NORMALIZE_MAP`(如 `Man City` → `Manchester City`),未命中映射的队名原样返回。 +`src/data.team_names.py` 维护 `NORMALIZE_MAP`(如 `Man City` → `Manchester City`),未命中映射的队名原样返回。 归一前先做 Unicode NFKD 去重音。 **唯一键是归一后英文名**:`teams.name` 带 `UNIQUE` 约束,所有入库路径均经 `TeamRepository.get_or_create` 收敛归一化 @@ -70,6 +70,32 @@ >(如 `"Man City"` → `"Manchester City"`,但 `"man city"` 原样保留)。上游 bzzoiro 返回的队名首字母大写, >实际命中无问题;若新增数据源返回全小写/全大写队名,需先 `title()` 再归一,否则会绕过映射产生重复 Team。 +#### 别名机制(`team_aliases`) + +归一仍可能遗漏历史重复队(如 `"Bayern Munich"` 与 `"Bayern München"` 经 NFKD 后相同则命中, +但 `"Man United"` vs `"Manchester United"` 若漏映射)。`team_aliases` 表提供**显式别名→teams.id** 映射: + +| 列 | 说明 | +|----|------| +| `alias_normalized` | PK,`normalize(别名)` 后的稳定幂等键 | +| `team_id` | FK → `teams.id`(ON DELETE CASCADE) | +| `original_alias` | 原始写法(保留供参考) | + +**定位三步链**(`get_or_create`):`normalize(name)` → 查 `teams.name` → 查 `team_aliases`(以 `normalize(name)` 为 PK)→ 都没有才 insert 新 Team。别名命中即复用已有 Team,避免产生重复。 + +**添加别名**(不自动合并历史重复队): + +- **Admin 接口**(推荐):`POST /api/v1/admin/teams/aliases {"alias": "Man United", "team_id": 42}`(require_admin,幂等) +- **直接 SQL**: + ```sql + INSERT INTO team_aliases(alias_normalized, team_id, original_alias) + VALUES ('man united', 42, 'Man United') + ON CONFLICT (alias_normalized) DO UPDATE SET team_id = EXCLUDED.team_id, original_alias = EXCLUDED.original_alias; + ``` + +> ⚠️ **别名不自动合并**:发现历史重复队 A/B 后,需人工确认归一目标(如保留 B),再为 A 的归一名添加别名指向 B。 +> 合并前请确认 A 的 `matches`/`standings` 引用是否需要迁移(可先 `SELECT COUNT(*) FROM matches WHERE home_team_id = A.id OR away_team_id = A.id` 评估)。 + **改名 / 合并流程**(人工): 当发现两个 `teams` 行实际是同一球队(如 `Manchester City` 与 `Man City` 因历史数据大小写差异各占一行): diff --git a/src/api/routes/admin_settings.py b/src/api/routes/admin_settings.py index 62ac05e..fc6b6ab 100644 --- a/src/api/routes/admin_settings.py +++ b/src/api/routes/admin_settings.py @@ -15,11 +15,15 @@ from fastapi import APIRouter from src.api.routes.admin_config import router as admin_config_router from src.api.routes.admin_datasources import router as admin_datasources_router +from src.api.routes.admin_ingest_jobs import router as admin_ingest_jobs_router from src.api.routes.admin_llm import router as admin_llm_router from src.api.routes.admin_quality import router as admin_quality_router +from src.api.routes.admin_teams import router as admin_teams_router router = APIRouter() router.include_router(admin_datasources_router) router.include_router(admin_config_router) +router.include_router(admin_ingest_jobs_router) router.include_router(admin_llm_router) router.include_router(admin_quality_router) +router.include_router(admin_teams_router) diff --git a/src/api/routes/admin_teams.py b/src/api/routes/admin_teams.py new file mode 100644 index 0000000..afa5b16 --- /dev/null +++ b/src/api/routes/admin_teams.py @@ -0,0 +1,60 @@ +"""后台管理:球队别名管理(只读列表 + 添加别名)。 + +归一名(teams.name)是球队唯一键;别名(team_aliases)是同一球队的不同写法 +(大小写/译名/缩写)到归一后 teams.id 的映射。入库时 normalize(name) 依次查 +teams.name 与 team_aliases,命中即复用,避免重复 Team。 + +不自动合并历史重复队;需显式添加别名(或先 SQL/再经由此接口)。 +""" +from __future__ import annotations + +import logging + +from fastapi import APIRouter, Depends, HTTPException +from sqlalchemy import desc, select + +from src.api.deps import require_admin +from src.api.schemas import TeamAliasIn, TeamAliasOut +from src.db.base import AsyncSession, get_db_read +from src.db.models import Team, TeamAlias +from src.db.repositories import TeamRepository + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/api/v1/admin", tags=["admin"], dependencies=[Depends(require_admin)]) + + +@router.get("/teams/aliases", response_model=list[TeamAliasOut]) +async def list_team_aliases(db: AsyncSession = Depends(get_db_read)): + """列出所有球队别名(最新在前)。""" + rows = (await db.execute(select(TeamAlias).order_by(desc(TeamAlias.created_at)).limit(200))).scalars().all() + return [ + TeamAliasOut( + alias_normalized=r.alias_normalized, + team_id=r.team_id, + original_alias=r.original_alias, + ) + for r in rows + ] + + +@router.post("/teams/aliases", response_model=TeamAliasOut, status_code=201) +async def add_team_alias(req: TeamAliasIn, db: AsyncSession = Depends(get_db_read)): + """为已有 Team 添加别名(幂等:重复添加会更新指向)。 + + 不自动合并历史重复队。若需合并 A→B:先为 A 的归一名添加别名指向 B, + 再人工确认 A 是否仍有独立引用。 + """ + # 校验目标 Team 存在 + team = await db.get(Team, req.team_id) + if team is None: + raise HTTPException(404, f"目标 Team 不存在: id={req.team_id}") + + repo = TeamRepository(db) + row = await repo.add_alias(req.alias, req.team_id) + logger.info("添加 Team 别名: %s -> team_id=%s", req.alias, req.team_id) + return TeamAliasOut( + alias_normalized=row.alias_normalized, + team_id=row.team_id, + original_alias=row.original_alias, + ) diff --git a/src/api/schemas.py b/src/api/schemas.py index 69b4122..b124e28 100644 --- a/src/api/schemas.py +++ b/src/api/schemas.py @@ -117,6 +117,19 @@ class IngestBzzoiroRequest(BaseModel): season: str | None = Field(None, description="standings 赛季,如 '2026-2027';空 = 当前赛季") +class TeamAliasIn(BaseModel): + """POST /api/v1/admin/teams/aliases 请求体:为已有 Team 添加别名。""" + + alias: str = Field(..., min_length=1, max_length=120, description="球队别名(原始写法)") + team_id: int = Field(..., gt=0, description="归一后的目标 teams.id") + + +class TeamAliasOut(BaseModel): + alias_normalized: str + team_id: int + original_alias: str + + class IngestResponse(BaseModel): leagues: dict total_inserted: int @@ -124,6 +137,28 @@ class IngestResponse(BaseModel): errors: list[str] = [] +class IngestBzzoiroResponse(BaseModel): + """POST /api/v1/ingest/bzzoiro 响应:兼容原 message 字段,新增 job_id 供轮询。""" + + ok: bool = True + job_id: str = Field(..., description="采集任务 ID(GET /api/v1/admin/ingest/jobs/{job_id} 轮询)") + message: str = "" + + +class IngestJobOut(BaseModel): + """采集任务状态详情。""" + + id: str + task: str + params: dict + status: str # pending | running | success | failed + result: dict | None = None + error: str | None = None + created_at: datetime | None = None + started_at: datetime | None = None + finished_at: datetime | None = None + + class ScheduleIn(BaseModel): id: str = Field(..., description="任务唯一标识,如 'daily-events'") task: str = Field(..., description="events / standings / stats / all") diff --git a/src/db/models.py b/src/db/models.py index 351d09b..a62f546 100644 --- a/src/db/models.py +++ b/src/db/models.py @@ -56,6 +56,22 @@ class Team(Base): away_matches: Mapped[list["Match"]] = relationship(foreign_keys="Match.away_team_id", back_populates="away_team") +class TeamAlias(Base): + """球队别名:同一球队的不同写法(大小写/译名/缩写)映射到归一后的 teams.id。 + + 入库流程(get_or_create):normalize(name) → 查 teams.name → 查 team_aliases + → 都没有再 insert 新 Team。别名不自动合并历史重复队,需显式添加。 + alias_normalized 为 normalize(别名)后的稳定幂等键,用作 PK 避免重复插入。 + """ + __tablename__ = "team_aliases" + + # normalize(别名)后的值,稳定幂等,用作主键 + alias_normalized: Mapped[str] = mapped_column(String(120), primary_key=True) + team_id: Mapped[int] = mapped_column(ForeignKey("teams.id", ondelete="CASCADE"), nullable=False) + original_alias: Mapped[str] = mapped_column(String(120), nullable=False) # 原始写法(保留供参考) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow) + + class Match(Base): __tablename__ = "matches" @@ -324,6 +340,35 @@ class RawEvent(Base): ) +class IngestJob(Base): + """采集任务状态:跟踪每次触后台采集任务的执行进度与结果。 + + POST /api/v1/ingest/bzzoiro 触发时写入(pending→running→success/failed), + 前端 Collection 页据此轮询到终态,替代此前"30 秒后盲标完成"的模拟。 + 分批 get_uow / BzzoiroSource / IngestFailure / Bronze/Lineage 均不受影响 + (本表仅作状态追踪,不介入采集事务)。 + """ + __tablename__ = "ingest_jobs" + + id: Mapped[str] = mapped_column(String(36), primary_key=True) # uuid4 + task: Mapped[str] = mapped_column(String(20), nullable=False) + params: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict) + status: Mapped[str] = mapped_column(String(20), nullable=False, server_default="pending") + result: Mapped[dict | None] = mapped_column(JSONB) + error: Mapped[str | None] = mapped_column(Text) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + + __table_args__ = ( + Index("ix_ingest_job_status_created", "status", "created_at"), + CheckConstraint( + "status IN ('pending', 'running', 'success', 'failed')", + name="ck_ingest_job_status", + ), + ) + + class IngestFailure(Base): """采集失败死信:记录失败原因、重试次数与下次重试时间。 diff --git a/src/db/repositories.py b/src/db/repositories.py index 1768380..cbc7ff4 100644 --- a/src/db/repositories.py +++ b/src/db/repositories.py @@ -114,24 +114,59 @@ class TeamRepository: async def get_or_create(self, name: str, *, name_zh: str | None = None) -> Team: """按名获取球队,不存在则创建(name_zh 供 bzzoiro 管线写中文名)。 - 归一化咽喉:所有入库 Team.name 必须经过 team_names.normalize, - 此处统一收敛,避免各调用点散落归一化逻辑导致重复 Team。 + 归一化咽喉 + 别名查找,三步定位: + 1) normalize(name) → 查 teams.name + 2) 查 team_aliases(以 normalize(name) 为幂等键)→ 复用已映射的 teams.id + 3) 都没有 → insert 新 Team(归一名) 创建新 Team 时 info 打出原始名与归一后的规范名,便于排查重名。 + 不自动合并历史重复队;需显式添加别名。 """ from src.data.team_names import normalize as normalize_name + from src.db.models import TeamAlias normalized = normalize_name(name) or name.strip() + + # 1) 归一名直查 teams team = await self.get_by_name(normalized) - if team is None: - logger.info( - "创建新 Team: %s -> %s", - name, normalized, - ) - team = Team(name=normalized, name_zh=name_zh) - self._session.add(team) - await self._session.flush() + if team is not None: + return team + + # 2) 别名查找:normalize(别名) 作为幂等键,命中即复用已有 Team + alias = await self._session.get(TeamAlias, normalized) + if alias is not None: + team = await self._session.get(Team, alias.team_id) + if team is not None: + logger.info("Team 别名命中: %s -> %s(已有 id=%s)", name, normalized, team.id) + return team + + # 3) 新建 Team(归一名) + 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 + async def add_alias(self, alias: str, team_id: int) -> TeamAlias: + """为已有 Team 添加别名。 + + 幂等:以 normalize(alias) 为 PK,重复添加同一别名会 upsert。 + 不自动合并历史重复队,仅建立别名映射。 + """ + from src.data.team_names import normalize as normalize_name + from src.db.models import TeamAlias + + normalized = normalize_name(alias) or alias.strip() + existing = await self._session.get(TeamAlias, normalized) + if existing is not None: + existing.team_id = team_id # 允许重新指向 + existing.original_alias = alias + await self._session.flush() + return existing + row = TeamAlias(alias_normalized=normalized, team_id=team_id, original_alias=alias) + self._session.add(row) + await self._session.flush() + return row + async def get_all_by_names(self, names: list[str]) -> dict[str, Team]: """批量获取球队,返回 name → Team 映射。""" if not names: diff --git a/tests/test_events_bronze.py b/tests/test_events_bronze.py index 0d62990..32e875d 100644 --- a/tests/test_events_bronze.py +++ b/tests/test_events_bronze.py @@ -20,7 +20,7 @@ from datetime import date, datetime, timezone import pytest import src.data.bzzoiro as bz -from src.db.models import DataLineage, League, Match, RawEvent, Team +from src.db.models import DataLineage, League, Match, RawEvent, Team, TeamAlias def _event(eid=1001, status="finished", home="Arsenal", away="Chelsea", hs=2, as_=1): @@ -71,11 +71,20 @@ class _FakeDB: League: list(leagues), RawEvent: list(raw_events), } - self._next_id = 0 + self._teams_by_id: dict[int, Team] = {t.id: t for t in teams if getattr(t, "id", None)} + self._aliases: dict[str, TeamAlias] = {} + self._next_id = max((t.id for t in teams if getattr(t, "id", None)), default=0) def add(self, obj): self.added.append(obj) + async def get(self, cls, key): + if cls is Team: + return self._teams_by_id.get(key) + if cls is TeamAlias: + return self._aliases.get(key) + return None + async def execute(self, stmt): entities = set() for d in (stmt.column_descriptions or []): @@ -90,6 +99,8 @@ class _FakeDB: if getattr(obj, "id", None) is None: self._next_id += 1 obj.id = self._next_id + if isinstance(obj, Team) and obj.id is not None: + self._teams_by_id[obj.id] = obj @pytest.fixture(autouse=True) diff --git a/tests/test_review_required_fixes.py b/tests/test_review_required_fixes.py index a80bdcb..bd031be 100644 --- a/tests/test_review_required_fixes.py +++ b/tests/test_review_required_fixes.py @@ -17,6 +17,7 @@ import re import pytest +from src.db.models import Team, TeamAlias from src.data.key_ring import _mask from src.llm import backtest as bt_mod from src.llm.agents import orchestrator as orch_mod @@ -105,6 +106,15 @@ class _FakeDb: self.added: list = [] self.flush_count = 0 self._next_id = 1000 + self._teams_by_id: dict[int, Team] = {} + self._aliases: dict[str, TeamAlias] = {} + + async def get(self, cls, key): + if cls is Team: + return self._teams_by_id.get(key) + if cls is TeamAlias: + return self._aliases.get(key) + return None async def execute(self, _stmt): if self._results: @@ -120,6 +130,10 @@ class _FakeDb: if getattr(obj, "id", None) is None: self._next_id += 1 obj.id = self._next_id + if isinstance(obj, Team) and getattr(obj, "id", None) is not None: + self._teams_by_id[obj.id] = obj + if isinstance(obj, TeamAlias): + self._aliases[obj.alias_normalized] = obj async def test_r2_standings_actually_upserts(monkeypatch): diff --git a/tests/test_standings_bronze.py b/tests/test_standings_bronze.py index 8635a30..ce5d399 100644 --- a/tests/test_standings_bronze.py +++ b/tests/test_standings_bronze.py @@ -18,7 +18,7 @@ from __future__ import annotations import pytest import src.data.bzzoiro as bz -from src.db.models import DataLineage, IngestFailure, League, RawEvent, Standing, Team +from src.db.models import DataLineage, IngestFailure, League, RawEvent, Standing, Team, TeamAlias def _payload(): @@ -78,11 +78,21 @@ class _FakeDB: Standing: list(standings), RawEvent: list(raw_events), } - self._next_id = 0 + # session.get 查找表(Team/TeamAlias) + self._teams_by_id: dict[int, Team] = {t.id: t for t in teams if getattr(t, "id", None)} + self._aliases: dict[str, TeamAlias] = {} + self._next_id = max((t.id for t in teams if getattr(t, "id", None)), default=0) def add(self, obj): self.added.append(obj) + async def get(self, cls, key): + if cls is Team: + return self._teams_by_id.get(key) + if cls is TeamAlias: + return self._aliases.get(key) + return None + async def execute(self, stmt): entities = set() for d in (stmt.column_descriptions or []): @@ -110,6 +120,9 @@ class _FakeDB: if getattr(obj, "id", None) is None: self._next_id += 1 obj.id = self._next_id + # 同步 session.get 可查到新建 Team + if isinstance(obj, Team) and obj.id is not None: + self._teams_by_id[obj.id] = obj def _preset_league():