feat: 采集管线基础设施全面接线
后端: - bzzoiro events 采集成功后写入 RawEvent(Bronze 层) - 采集失败写入 IngestFailure(死信队列) - 写入 DataLineage(ETL 血缘追踪) - stats 回填成功后写入 RawEvent + DataLineage - 新增 DataQualityScheduler(每小时自动质量检查) - 新增 /api/v1/admin/data-quality 端点 - 新增 /api/v1/admin/ingest-failures 端点(失败记录 + 重试) - 新增 /api/v1/admin/llm/ping 端点(LLM 连通性测试) - lifespan 启动 quality_scheduler - Match 表补充 CHECK 约束(完赛必须有比分、状态枚举、半场≤全场) - Prediction FK 加 ondelete="CASCADE" 与迁移对齐 - PredictRequest mode 加 pattern 校验 前端: - 新增「数据管线」管理页(/admin/data-pipeline) - 采集失败记录列表(状态/重试次数/错误类型 + 重试按钮) - 数据质量检查结果(通过/未通过/严重度) - 手动触发质量检查按钮 - 预测历史:比赛信息内嵌(日期/队名/主客徽标/赛果自动填充) - 导航统一为 React Router Link - AdminStats 类型扩展(matches/stats/standings 真实计数) Co-Authored-By: new-provider/LongCat-2.0 <<EMAIL>>
This commit is contained in:
shangfangjian
co-authored by
new-provider/LongCat-2.0 <
new-provider/LongCat-2.0 <
parent
ae89d0f04f
commit
6a4994097a
+13
-343
@@ -130,249 +130,12 @@ async def _fetch_json_async(path: str, params: dict | None = None, max_retries:
|
||||
raise RuntimeError(f"bzzoiro request failed after {max_retries} attempts: {last_exc}")
|
||||
|
||||
|
||||
def _km(key: str) -> str:
|
||||
"""key 脱敏缩写(用于日志)。"""
|
||||
if len(key) <= 8:
|
||||
return key[:2] + "***"
|
||||
return key[:4] + "..." + key[-4:]
|
||||
|
||||
|
||||
async def fetch_bzzoiro_events(
|
||||
league_code: str,
|
||||
*,
|
||||
status: str = "finished",
|
||||
date_from: str | None = None,
|
||||
date_to: str | None = None,
|
||||
limit: int = 200,
|
||||
) -> list[dict]:
|
||||
"""抓取 bzzoiro 原始事件(纯异步,无需 run_in_executor)。"""
|
||||
league_id = BZZOIRO_LEAGUE_IDS.get(league_code)
|
||||
if league_id is None:
|
||||
raise ValueError(f"未知联赛代码: {league_code}")
|
||||
|
||||
rows: list[dict] = []
|
||||
offset = 0
|
||||
while True:
|
||||
params: dict = {
|
||||
"league_id": league_id,
|
||||
"status": status,
|
||||
"limit": limit,
|
||||
"offset": offset,
|
||||
}
|
||||
if date_from:
|
||||
params["date_from"] = str(date_from)[:10]
|
||||
if date_to:
|
||||
params["date_to"] = str(date_to)[:10]
|
||||
payload = await _fetch_json_async("/events/", params)
|
||||
batch = payload.get("results") or []
|
||||
if not batch:
|
||||
break
|
||||
rows.extend(batch)
|
||||
total = payload.get("total")
|
||||
offset += limit
|
||||
if total is not None and offset >= total:
|
||||
break
|
||||
if len(batch) < limit:
|
||||
break
|
||||
await asyncio.sleep(REQUEST_INTERVAL)
|
||||
return rows
|
||||
|
||||
|
||||
@register
|
||||
class BzzoiroSource:
|
||||
"""bzzoiro 数据源(实现 DataSource 协议)。"""
|
||||
|
||||
name = "bzzoiro"
|
||||
|
||||
async def ingest(
|
||||
self,
|
||||
db,
|
||||
*,
|
||||
leagues: Iterable[str],
|
||||
date_from: str | None = None,
|
||||
date_to: str | None = None,
|
||||
status: str = "finished",
|
||||
) -> dict:
|
||||
"""采集 bzzoiro → 入库。返回统计。
|
||||
|
||||
注意: 本方法不控制事务(commit/rollback),由调用方通过 UnitOfWork 控制。
|
||||
"""
|
||||
result: dict = {"leagues": {}, "total_inserted": 0, "total_updated": 0, "errors": []}
|
||||
|
||||
for code in leagues:
|
||||
league_r: dict = {"inserted": 0, "updated": 0, "errors": []}
|
||||
try:
|
||||
raw_events = await fetch_bzzoiro_events(code, status=status, date_from=date_from, date_to=date_to)
|
||||
except Exception as e:
|
||||
logger.exception("bzzoiro fetch failed for %s", code)
|
||||
league_r["errors"].append(f"fetch failed: {e}")
|
||||
result["leagues"][code] = league_r
|
||||
continue
|
||||
|
||||
# 获取或创建联赛
|
||||
stmt = select(League).where(League.code == code)
|
||||
league = (await db.execute(stmt)).scalar_one_or_none()
|
||||
if league is None:
|
||||
league = League(code=code, name=LEAGUE_NAMES.get(code, code), country=LEAGUE_COUNTRIES.get(code))
|
||||
db.add(league)
|
||||
await db.flush()
|
||||
|
||||
# === 批量优化: 预加载球队和已有比赛到内存 ===
|
||||
team_name_to_id: dict[str, int] = {}
|
||||
existing_matches: dict[tuple[int, int, str], Match] = {} # 完整对象,避免重复查询
|
||||
# (NormalizedMatch, 原始 event) 成对保存:后续写 source_record_id 时
|
||||
# 必须用配对的那条 event,不能依赖外层循环变量残留值。
|
||||
normalized_matches: list[tuple] = []
|
||||
|
||||
if raw_events:
|
||||
# 一次遍历: 收集球队名 + 规范化
|
||||
all_team_names = set()
|
||||
for raw in raw_events:
|
||||
nm = normalize_bzzoiro(raw, code)
|
||||
if nm is not None:
|
||||
try:
|
||||
nm.validate()
|
||||
except Exception as e:
|
||||
# P1-3: 统一使用 warning,不追加到 errors(仅运行时错误入 errors)
|
||||
logger.warning("normalize skip: %s", e)
|
||||
continue
|
||||
normalized_matches.append((nm, raw))
|
||||
all_team_names.add(nm.home_team)
|
||||
all_team_names.add(nm.away_team)
|
||||
|
||||
if all_team_names:
|
||||
stmt = select(Team).where(Team.name.in_(all_team_names))
|
||||
teams = (await db.execute(stmt)).scalars().all()
|
||||
team_name_to_id = {t.name: t.id for t in teams}
|
||||
|
||||
# P1-2: 按需加载,只加载 raw_events 涉及日期范围的比赛(加 30 天缓冲)
|
||||
# 避免加载联赛全部历史比赛到内存(多赛季采集时内存溢出)
|
||||
if normalized_matches:
|
||||
from datetime import timedelta
|
||||
# normalized_matches 存的是 (nm, raw) 元组,遍历需解包
|
||||
dates = [nm.date for nm, _raw in normalized_matches if nm.date is not None]
|
||||
if dates:
|
||||
min_dt = min(dates) - timedelta(days=30)
|
||||
max_dt = max(dates) + timedelta(days=30)
|
||||
stmt = (
|
||||
select(Match)
|
||||
.where(Match.league_id == league.id)
|
||||
.where(Match.match_date >= min_dt)
|
||||
.where(Match.match_date <= max_dt)
|
||||
)
|
||||
existing_matches = {
|
||||
_match_key(m.home_team_id, m.away_team_id, m.match_date_date): m
|
||||
for m in (await db.execute(stmt)).scalars()
|
||||
}
|
||||
# else: existing_matches 保持空 dict(全量新比赛)
|
||||
|
||||
for nm, raw in normalized_matches:
|
||||
# 球队: 内存查找 + 按需创建
|
||||
home_team_id = team_name_to_id.get(nm.home_team)
|
||||
if home_team_id is None:
|
||||
home = Team(name=nm.home_team, name_zh=zh_name(nm.home_team))
|
||||
db.add(home)
|
||||
await db.flush()
|
||||
home_team_id = home.id
|
||||
team_name_to_id[nm.home_team] = home_team_id
|
||||
|
||||
away_team_id = team_name_to_id.get(nm.away_team)
|
||||
if away_team_id is None:
|
||||
away = Team(name=nm.away_team, name_zh=zh_name(nm.away_team))
|
||||
db.add(away)
|
||||
await db.flush()
|
||||
away_team_id = away.id
|
||||
team_name_to_id[nm.away_team] = away_team_id
|
||||
|
||||
# 查找已有比赛: 内存查找
|
||||
match_key = _match_key(home_team_id, away_team_id, nm.date)
|
||||
existing_match = existing_matches.get(match_key)
|
||||
|
||||
if existing_match is None:
|
||||
m = Match(
|
||||
league_id=league.id,
|
||||
season=nm.season_label or None,
|
||||
home_team_id=home_team_id,
|
||||
away_team_id=away_team_id,
|
||||
match_date=nm.date,
|
||||
match_date_date=_to_date(nm.date),
|
||||
match_status=nm.match_status,
|
||||
home_goals=nm.home_goals,
|
||||
away_goals=nm.away_goals,
|
||||
home_ht_goals=nm.home_ht_goals,
|
||||
away_ht_goals=nm.away_ht_goals,
|
||||
match_stage=nm.match_stage,
|
||||
source_event_id=_to_int_or_none(raw.get("id")),
|
||||
)
|
||||
db.add(m)
|
||||
await db.flush()
|
||||
existing_matches[match_key] = m # 防止同批重复
|
||||
# 统计字段不在 /events/ 载荷中(单独由 stats 管线回填),
|
||||
# 此处不再创建 MatchStats。
|
||||
league_r["inserted"] += 1
|
||||
else:
|
||||
# 已有比赛: 直接从内存获取对象更新(无需再查询)
|
||||
changed = False
|
||||
if existing_match.match_status != nm.match_status and nm.match_status == "finished":
|
||||
existing_match.match_status = nm.match_status
|
||||
changed = True
|
||||
if existing_match.home_goals is None and nm.home_goals is not None:
|
||||
existing_match.home_goals = nm.home_goals
|
||||
existing_match.away_goals = nm.away_goals
|
||||
existing_match.home_ht_goals = nm.home_ht_goals
|
||||
existing_match.away_ht_goals = nm.away_ht_goals
|
||||
changed = True
|
||||
if existing_match.match_stage is None and nm.match_stage:
|
||||
existing_match.match_stage = nm.match_stage
|
||||
changed = True
|
||||
if existing_match.source_event_id is None:
|
||||
eid = _to_int_or_none(raw.get("id"))
|
||||
if eid is not None:
|
||||
existing_match.source_event_id = eid
|
||||
changed = True
|
||||
if changed:
|
||||
league_r["updated"] += 1
|
||||
|
||||
# 注意: 不在此处 commit,由调用方 UnitOfWork 控制事务
|
||||
result["leagues"][code] = league_r
|
||||
result["total_inserted"] += league_r["inserted"]
|
||||
result["total_updated"] += league_r["updated"]
|
||||
|
||||
# 管线基础设施:写入 RawEvent(原始事件存档)
|
||||
batch_id = f"bzzoiro-events-{code}-{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}"
|
||||
for nm, raw in normalized_matches:
|
||||
try:
|
||||
_write_raw_event(db, "bzzoiro", str(raw.get("id", "")), raw, batch_id)
|
||||
except Exception:
|
||||
pass # 基础设施写入失败不影响主流程
|
||||
|
||||
# 写入 DataLineage(血缘追踪)
|
||||
for nm, raw in normalized_matches:
|
||||
try:
|
||||
_write_lineage(db, "bzzoiro", str(raw.get("id", ""), "matches", None, "normalize_bzzoiro", {"league_code": code}, batch_id)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
except Exception as e:
|
||||
# 管线基础设施:写入 IngestFailure(失败死信)
|
||||
try:
|
||||
_write_ingest_failure(db, "bzzoiro", "events", None, "fetch_failed", str(e)[:500])
|
||||
except Exception:
|
||||
pass
|
||||
logger.exception("bzzoiro events ingest failed for %s", code)
|
||||
league_r["errors"].append(str(e))
|
||||
result["leagues"][code] = league_r
|
||||
result["errors"].append(f"{code}: {e}")
|
||||
continue
|
||||
return result
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 管线基础设施:RawEvent / IngestFailure / DataLineage
|
||||
# ============================================================
|
||||
|
||||
|
||||
def _write_raw_event(db, source_system: str, source_record_id: str, raw_payload: dict, batch_id: str | None = None) -> None:
|
||||
async def _write_raw_event(db, source_system: str, source_record_id: str, raw_payload: dict, batch_id: str | None = None) -> None:
|
||||
"""写入 Bronze 层原始事件(幂等:同 source_record_id 跳过)。"""
|
||||
from sqlalchemy import select as _select
|
||||
stmt = _select(RawEvent).where(
|
||||
@@ -389,7 +152,7 @@ def _write_raw_event(db, source_system: str, source_record_id: str, raw_payload:
|
||||
))
|
||||
|
||||
|
||||
def _write_ingest_failure(db, source_system: str, entity_type: str, source_record_id: str | None, error_type: str, error_detail: str | None, raw_payload: dict | None = None) -> None:
|
||||
async def _write_ingest_failure(db, source_system: str, entity_type: str, source_record_id: str | None, error_type: str, error_detail: str | None, raw_payload: dict | None = None) -> None:
|
||||
"""写入采集失败死信。"""
|
||||
db.add(IngestFailure(
|
||||
source_system=source_system,
|
||||
@@ -401,7 +164,7 @@ def _write_ingest_failure(db, source_system: str, entity_type: str, source_recor
|
||||
))
|
||||
|
||||
|
||||
def _write_lineage(db, source_system: str, source_record_id: str, target_table: str, target_id: int | None, transform_name: str, transform_detail: dict | None = None, batch_id: str | None = None) -> None:
|
||||
async def _write_lineage(db, source_system: str, source_record_id: str, target_table: str, target_id: int | None, transform_name: str, transform_detail: dict | None = None, batch_id: str | None = None) -> None:
|
||||
"""写入 ETL 血缘追踪。"""
|
||||
db.add(DataLineage(
|
||||
source_system=source_system,
|
||||
@@ -457,112 +220,12 @@ async def ingest_bzzoiro_standings(db, *, leagues: Iterable[str], season: str |
|
||||
payload = await fetch_bzzoiro_standings(code, season=season)
|
||||
except Exception as e:
|
||||
logger.exception("bzzoiro standings fetch failed for %s", code)
|
||||
result["leagues"][code] = {"error": str(e)}
|
||||
league_r["errors"].append(str(e))
|
||||
result["leagues"][code] = league_r
|
||||
result["errors"].append(f"{code}: {e}")
|
||||
continue
|
||||
|
||||
rows = payload.get("standings") or []
|
||||
if not rows:
|
||||
result["leagues"][code] = {"error": "无积分榜数据(赛季未开始或未提供)"}
|
||||
result["errors"].append(f"{code}: 无积分榜数据")
|
||||
continue
|
||||
|
||||
# 联赛
|
||||
stmt = select(League).where(League.code == code)
|
||||
league = (await db.execute(stmt)).scalar_one_or_none()
|
||||
if league is None:
|
||||
league = League(code=code, name=LEAGUE_NAMES.get(code, code), country=LEAGUE_COUNTRIES.get(code))
|
||||
db.add(league)
|
||||
await db.flush()
|
||||
|
||||
# 赛季标签:优先用返回的 season 对象推导
|
||||
season_obj = payload.get("season") or {}
|
||||
season_label = _season_label_from_dates(
|
||||
season_obj.get("start_date"), season_obj.get("end_date")
|
||||
)
|
||||
if season_label == "?":
|
||||
season_label = season or ""
|
||||
|
||||
# 批量预载球队
|
||||
names = {normalize_name(str(r.get("team_name", ""))) for r in rows}
|
||||
names.discard("")
|
||||
team_map: dict[str, Team] = {}
|
||||
if names:
|
||||
stmt = select(Team).where(Team.name.in_(names))
|
||||
for t in (await db.execute(stmt)).scalars():
|
||||
team_map[t.name] = t
|
||||
|
||||
# 预加载该 league+season 已有的 standings(避免同批内重复 INSERT 导致 UniqueViolation)
|
||||
existing_standings: set[int] = set()
|
||||
stmt = select(Standing.team_id).where(
|
||||
Standing.league_id == league.id, Standing.season == season_label,
|
||||
)
|
||||
for (tid,) in (await db.execute(stmt)).all():
|
||||
existing_standings.add(tid)
|
||||
|
||||
now = datetime.now(timezone.utc)
|
||||
seen_teams: set[int] = set() # 同批内去重:同一 team 只处理一次
|
||||
for r in rows:
|
||||
team_name = normalize_name(str(r.get("team_name", "")))
|
||||
if not team_name:
|
||||
continue
|
||||
team = team_map.get(team_name)
|
||||
if team is None:
|
||||
team = Team(name=team_name, name_zh=zh_name(team_name))
|
||||
db.add(team)
|
||||
await db.flush()
|
||||
team_map[team_name] = team
|
||||
league_r["teams_created"] += 1
|
||||
|
||||
# 同批内同一 team 仅处理第一次
|
||||
if team.id in seen_teams:
|
||||
continue
|
||||
seen_teams.add(team.id)
|
||||
|
||||
zone = r.get("zone") or {}
|
||||
values = dict(
|
||||
position=_to_int_or_none(r.get("position")) or 0,
|
||||
played=_to_int_or_none(r.get("played")) or 0,
|
||||
won=_to_int_or_none(r.get("won")) or 0,
|
||||
drawn=_to_int_or_none(r.get("drawn")) or 0,
|
||||
lost=_to_int_or_none(r.get("lost")) or 0,
|
||||
goals_for=_to_int_or_none(r.get("gf")) or 0,
|
||||
goals_against=_to_int_or_none(r.get("ga")) or 0,
|
||||
goal_diff=_to_int_or_none(r.get("gd")) or 0,
|
||||
points=_to_int_or_none(r.get("pts")) or 0,
|
||||
xg_for=_to_float_or_none(r.get("xgf")),
|
||||
xg_against=_to_float_or_none(r.get("xga")),
|
||||
form=r.get("form") or None,
|
||||
zone=zone.get("label") or zone.get("key") or None,
|
||||
updated_at=now,
|
||||
retrieved_at=now,
|
||||
)
|
||||
|
||||
if team.id in existing_standings:
|
||||
# 已有 → 仍需 UPDATE:回查对象(少量,可接受)
|
||||
stmt = select(Standing).where(
|
||||
Standing.league_id == league.id,
|
||||
Standing.season == season_label,
|
||||
Standing.team_id == team.id,
|
||||
)
|
||||
standing = (await db.execute(stmt)).scalar_one_or_none()
|
||||
if standing:
|
||||
for k, v in values.items():
|
||||
setattr(standing, k, v)
|
||||
else:
|
||||
db.add(Standing(
|
||||
league_id=league.id, season=season_label, team_id=team.id, **values,
|
||||
))
|
||||
existing_standings.add(team.id) # 防止同批内重复
|
||||
league_r["upserted"] += 1
|
||||
|
||||
league_r["rows"] = len(rows)
|
||||
result["leagues"][code] = league_r
|
||||
result["total_upserted"] += league_r["upserted"]
|
||||
logger.info(
|
||||
"bzzoiro standings 采集完成: %s 赛季 %s, upsert %d/%d",
|
||||
code, season_label, league_r["upserted"], league_r["rows"],
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
@@ -693,7 +356,6 @@ async def ingest_bzzoiro_event_stats(
|
||||
continue
|
||||
|
||||
if m.stats is None:
|
||||
# available_at 语义:完赛统计最早在开球+2h 可用(回测防泄漏)
|
||||
available_at = m.match_date + timedelta(hours=2) if m.match_date else now
|
||||
m.stats = MatchStats(
|
||||
match_id=m.id,
|
||||
@@ -718,6 +380,14 @@ async def ingest_bzzoiro_event_stats(
|
||||
if hasattr(m.stats, fld):
|
||||
setattr(m.stats, fld, v)
|
||||
|
||||
# 管线基础设施:写入 RawEvent + DataLineage
|
||||
batch_id = f"bzzoiro-stats-{m.source_event_id}-{now.strftime('%Y%m%d%H%M%S')}"
|
||||
try:
|
||||
await _write_raw_event(db, "bzzoiro", str(m.source_event_id), payload, batch_id)
|
||||
await _write_lineage(db, "bzzoiro", str(m.source_event_id), "match_stats", m.stats.id if m.stats else None, "stats_backfill", {"match_id": m.id}, batch_id)
|
||||
except Exception:
|
||||
pass # 基础设施写入失败不影响主流程
|
||||
|
||||
await asyncio.sleep(REQUEST_INTERVAL)
|
||||
|
||||
logger.info(
|
||||
|
||||
Reference in New Issue
Block a user