WorkBuddy
4514ef4e92
- repositories.py 新增 MatchRepository.find_by_league_and_date_range / find_finished_with_stats(含 stats 预加载),TeamRepository.get_or_create 支持 name_zh - bzzoiro events/standings/stats 三条管线的联赛、球队、比赛查找与批量预载 全部改走 Repository;事务提交仍由调用方 UnitOfWork 控制 - Standing/RawEvent/Lineage 管线内私有读写保留在本模块(不强行 Repository 化) - 模块 docstring 纠正为准确表述,移除「已全面 Repository 化」的误导性声明
794 lines
32 KiB
Python
794 lines
32 KiB
Python
"""Bzzoiro 数据源:抓取 + 入库(单一数据源)。
|
|
|
|
三条管线:
|
|
1. events — 比赛日程/比分(/events/),含 source_event_id 血缘
|
|
2. standings— 联赛积分榜快照(/leagues/{id}/standings/)
|
|
3. stats — 已完赛比赛详细统计回填(/events/{id}/stats/)
|
|
|
|
D4(工程债): Team/League/Match 的查找/创建经 Repository 层(src/db/repositories.py),
|
|
本模块不直接控制事务(commit/rollback 由调用方 UnitOfWork 控制,这里只 flush)。
|
|
Standing/RawEvent/Lineage 等管线内私有读写仍在本模块内实现,不强行 Repository 化。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import random
|
|
from collections.abc import Iterable
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import select
|
|
|
|
import httpx
|
|
|
|
from src.core.runtime_config import get_runtime_value
|
|
from src.core.http_client import get_client
|
|
from src.data.config import BZZOIRO_LEAGUE_IDS, LEAGUE_COUNTRIES, LEAGUE_NAMES, REQUEST_INTERVAL
|
|
from src.data.key_ring import _mask, get_key_ring
|
|
from src.data.normalize import normalize_bzzoiro
|
|
from src.data.team_names_zh import zh_name
|
|
from src.data.sources import register
|
|
from src.db.models import Match, MatchStats, Standing, Team, RawEvent, IngestFailure, DataLineage
|
|
from src.db.repositories import LeagueRepository, MatchRepository, TeamRepository
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _to_date(value):
|
|
"""把 datetime / date / str 统一成 `date`。"""
|
|
if value is None:
|
|
return None
|
|
if hasattr(value, "date") and callable(value.date):
|
|
return value.date()
|
|
return value
|
|
|
|
|
|
def _to_int_or_none(value) -> int | None:
|
|
"""宽松转 int(用于上游 ID 解析,失败返回 None 不抛错)。"""
|
|
if value is None:
|
|
return None
|
|
try:
|
|
return int(str(value).strip())
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _match_key(home_team_id: int, away_team_id: int, match_date) -> tuple[int, int, str]:
|
|
"""比赛去重键:(主队, 客队, 天级日期 ISO 字符串)。
|
|
|
|
统一在这里构造,避免"预加载时用 str(date)、写入时用 isoformat()"这类
|
|
隐式格式依赖 —— 两者当前恰好相等,但一旦有人改动其一就会静默失配,
|
|
导致所有比赛被判为不存在而重复插入。
|
|
"""
|
|
d = _to_date(match_date)
|
|
return (home_team_id, away_team_id, d.isoformat() if d is not None else "")
|
|
|
|
|
|
async def _fetch_json_async(path: str, params: dict | None = None, max_retries: int = 3) -> dict | list:
|
|
"""异步 HTTP(bzzoiro 使用 httpx,不再阻塞事件循环线程池)。
|
|
|
|
多 key 轮换:遇到 429 自动切换到下一个 key;全部 key 冷却时等待最早恢复。
|
|
"""
|
|
base = (await get_runtime_value("BZZOIRO_BASE")).rstrip("/")
|
|
raw_keys = await get_runtime_value("BZZOIRO_KEY")
|
|
ring = get_key_ring(base, raw_keys)
|
|
|
|
url = f"{base}/{path.lstrip('/')}"
|
|
key = ring.get()
|
|
if not key:
|
|
raise RuntimeError("BZZOIRO_KEY 未设置")
|
|
|
|
last_exc: Exception | None = None
|
|
for attempt in range(max_retries):
|
|
headers = {
|
|
"Authorization": f"Token {key}",
|
|
"Accept": "application/json",
|
|
}
|
|
try:
|
|
client = get_client()
|
|
# 整请求兜底: httpx 无 total 超时,用 wait_for 防「滴水式」限速挂死
|
|
resp = await asyncio.wait_for(
|
|
client.get(
|
|
url, headers=headers, params=params,
|
|
timeout=httpx.Timeout(connect=10.0, read=30.0, write=10.0, pool=10.0),
|
|
),
|
|
timeout=60.0,
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
except Exception as e:
|
|
last_exc = e
|
|
status = getattr(getattr(e, "response", None), "status_code", None)
|
|
if status == 429:
|
|
# 限流:标记当前 key 冷却,切换到下一个
|
|
new_key = ring.report_rate_limited(key)
|
|
if new_key and new_key != key:
|
|
logger.info("bzzoiro 429 → 切换 key: %s → %s,立即重试", _mask(key), _mask(new_key))
|
|
key = new_key
|
|
continue # 立即重试,不等待
|
|
# 单 key 或全部冷却:等待最早恢复的 key
|
|
wait = ring.wait_if_all_blocked()
|
|
if wait > 0:
|
|
logger.warning("bzzoiro 全部 key 冷却,等待 %.1fs 后重试", wait)
|
|
await asyncio.sleep(min(wait, 30.0))
|
|
else:
|
|
delay = min(2 ** attempt, 16) + random.uniform(0, 1)
|
|
logger.warning("bzzoiro 429, retry %d in %.1fs", attempt + 1, delay)
|
|
await asyncio.sleep(delay)
|
|
key = ring.get() or key
|
|
continue
|
|
if 500 <= (status or 0) < 600:
|
|
delay = min(2 ** attempt, 16) + random.uniform(0, 1)
|
|
logger.warning("bzzoiro %d, retry %d in %.1fs", status, attempt + 1, delay)
|
|
await asyncio.sleep(delay)
|
|
continue
|
|
# 网络错误(连接失败/超时)也退避重试
|
|
if isinstance(e, (TimeoutError, ConnectionError, OSError)):
|
|
delay = min(2 ** attempt, 16) + random.uniform(0, 1)
|
|
logger.warning("bzzoiro network error, retry %d in %.1fs: %s", attempt + 1, delay, e)
|
|
await asyncio.sleep(delay)
|
|
continue
|
|
raise
|
|
raise RuntimeError(f"bzzoiro request failed after {max_retries} attempts: {last_exc}")
|
|
|
|
|
|
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}")
|
|
await _safe_write_ingest_failure(
|
|
db,
|
|
entity_type="events",
|
|
source_record_id=None,
|
|
error=e,
|
|
raw_payload={"league": code, "status": status, "date_from": date_from, "date_to": date_to},
|
|
)
|
|
result["leagues"][code] = league_r
|
|
continue
|
|
|
|
# D4: 联赛查找/创建经 LeagueRepository(事务仍由调用方 UoW 提交)
|
|
league = await LeagueRepository(db).get_or_create(
|
|
code, LEAGUE_NAMES.get(code, code), LEAGUE_COUNTRIES.get(code)
|
|
)
|
|
team_r = TeamRepository(db)
|
|
match_r = MatchRepository(db)
|
|
|
|
# === 批量优化: 预加载球队和已有比赛到内存 ===
|
|
team_name_to_id: dict[str, int] = {}
|
|
existing_matches: dict[tuple[int, int, str], Match] = {} # 完整对象,避免重复查询
|
|
# (NormalizedMatch, 原始 event) 成对保存:后续写 source_event_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:
|
|
team_name_to_id = {
|
|
name: t.id
|
|
for name, t in (await team_r.get_all_by_names(list(all_team_names))).items()
|
|
}
|
|
|
|
# P1-2: 按需加载,只加载 raw_events 涉及日期范围的比赛(加 30 天缓冲)
|
|
# 避免加载联赛全部历史比赛到内存(多赛季采集时内存溢出)
|
|
if normalized_matches:
|
|
# 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)
|
|
matches_in_range = await match_r.find_by_league_and_date_range(
|
|
league.id, min_dt, max_dt
|
|
)
|
|
existing_matches = {
|
|
_match_key(m.home_team_id, m.away_team_id, m.match_date_date): m
|
|
for m in matches_in_range
|
|
}
|
|
# else: existing_matches 保持空 dict(全量新比赛)
|
|
|
|
# D1: Bronze 层批次信息(每联赛每批次一个 batch_id;seen 防同批重复写入)
|
|
now = datetime.now(timezone.utc)
|
|
bronze_batch_id = f"bzzoiro-events-{code}-{now:%Y%m%d%H%M%S}"
|
|
bronze_written: set[str] = set()
|
|
|
|
for nm, raw in normalized_matches:
|
|
# D1: RawEvent 幂等键(上游 id 或合成键),插入/变更更新共用
|
|
record_id = _events_record_id(code, nm, raw)
|
|
|
|
# 球队: 内存查找 + 按需创建(D4: 经 TeamRepository)
|
|
home_team_id = team_name_to_id.get(nm.home_team)
|
|
if home_team_id is None:
|
|
home = await team_r.get_or_create(nm.home_team, name_zh=zh_name(nm.home_team))
|
|
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 = await team_r.get_or_create(nm.away_team, name_zh=zh_name(nm.away_team))
|
|
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
|
|
# D1: 成功插入 → 补写 Bronze 层(原始载荷 + 血缘)
|
|
if record_id not in bronze_written:
|
|
bronze_written.add(record_id)
|
|
await _write_events_bronze(
|
|
db,
|
|
source_record_id=record_id,
|
|
raw_payload=raw,
|
|
target_match_id=m.id,
|
|
league_code=code,
|
|
match_status=nm.match_status,
|
|
batch_id=bronze_batch_id,
|
|
)
|
|
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
|
|
# D1: 变更更新 → 补写血缘(RawEvent 幂等键不变,重复采集自动跳过)
|
|
if record_id not in bronze_written:
|
|
bronze_written.add(record_id)
|
|
await _write_events_bronze(
|
|
db,
|
|
source_record_id=record_id,
|
|
raw_payload=raw,
|
|
target_match_id=existing_match.id,
|
|
league_code=code,
|
|
match_status=nm.match_status,
|
|
batch_id=bronze_batch_id,
|
|
)
|
|
|
|
# 注意: 不在此处 commit,由调用方 UnitOfWork 控制事务
|
|
result["leagues"][code] = league_r
|
|
result["total_inserted"] += league_r["inserted"]
|
|
result["total_updated"] += league_r["updated"]
|
|
return result
|
|
|
|
|
|
# ============================================================
|
|
# 管线基础设施:RawEvent / IngestFailure / DataLineage
|
|
# ============================================================
|
|
|
|
|
|
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(
|
|
RawEvent.source_system == source_system,
|
|
RawEvent.source_record_id == source_record_id,
|
|
)
|
|
existing = (await db.execute(stmt)).scalar_one_or_none()
|
|
if existing is None:
|
|
db.add(RawEvent(
|
|
source_system=source_system,
|
|
source_record_id=source_record_id,
|
|
raw_payload=raw_payload,
|
|
ingest_batch_id=batch_id,
|
|
))
|
|
|
|
|
|
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,
|
|
entity_type=entity_type,
|
|
source_record_id=source_record_id,
|
|
error_type=error_type,
|
|
error_detail=error_detail,
|
|
raw_payload=raw_payload,
|
|
))
|
|
|
|
|
|
async def _safe_write_ingest_failure(
|
|
db,
|
|
*,
|
|
entity_type: str,
|
|
source_record_id: str | None,
|
|
error: Exception,
|
|
raw_payload: dict | None = None,
|
|
) -> None:
|
|
"""抓取失败时尽力写入死信表(失败不影响主流程)。
|
|
|
|
死信是「可观测性」基础设施,与 RawEvent/Lineage 同级:写入失败只记
|
|
warning,绝不能让原始抓取错误之外的新异常打断采集循环。
|
|
"""
|
|
try:
|
|
await _write_ingest_failure(
|
|
db, "bzzoiro", entity_type, source_record_id,
|
|
"fetch_error", str(error), raw_payload,
|
|
)
|
|
except Exception:
|
|
logger.warning(
|
|
"写入 ingest_failures 死信失败(entity=%s, record=%s): %s",
|
|
entity_type, source_record_id, error, exc_info=True,
|
|
)
|
|
|
|
|
|
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,
|
|
source_record_id=source_record_id,
|
|
target_table=target_table,
|
|
target_id=target_id,
|
|
transform_name=transform_name,
|
|
transform_detail=transform_detail,
|
|
batch_id=batch_id,
|
|
))
|
|
|
|
|
|
def _events_record_id(league_code: str, nm, raw: dict) -> str:
|
|
"""events 载荷的 RawEvent 幂等键。
|
|
|
|
优先用上游 event id;缺失时用 (league:home:away:date) 合成稳定键 ——
|
|
取 normalize 后的队名与天级日期(与 _match_key 同口径),不依赖 DB 自增 id,
|
|
保证同一来源比赛重复采集时命中同一条 RawEvent,不产生重复原始载荷。
|
|
"""
|
|
eid = _to_int_or_none(raw.get("id"))
|
|
if eid is not None:
|
|
return str(eid)
|
|
d = _to_date(nm.date)
|
|
date_part = d.isoformat() if d is not None else "na"
|
|
return f"{league_code}:{nm.home_team}:{nm.away_team}:{date_part}"
|
|
|
|
|
|
async def _write_events_bronze(
|
|
db,
|
|
*,
|
|
source_record_id: str,
|
|
raw_payload: dict,
|
|
target_match_id: int | None,
|
|
league_code: str,
|
|
match_status: str | None,
|
|
batch_id: str,
|
|
) -> None:
|
|
"""events 成功插入/更新单场比赛后的 Bronze 层补写:RawEvent(幂等) + DataLineage。
|
|
|
|
D1(工程债):此前只有 stats 回填写 RawEvent/Lineage,events 管线作为比赛
|
|
主数据的唯一入口反而不留溯源记录。幂等性由 _write_raw_event 的
|
|
source_record_id 查重保证;best-effort:基础设施写入失败只记 warning,
|
|
绝不拖垮采集主流程(与 _safe_write_ingest_failure 同级约束)。
|
|
"""
|
|
try:
|
|
await _write_raw_event(db, "bzzoiro", source_record_id, raw_payload, batch_id)
|
|
await _write_lineage(
|
|
db, "bzzoiro", source_record_id,
|
|
"matches", target_match_id, "events_ingest",
|
|
{"league": league_code, "match_status": match_status},
|
|
batch_id,
|
|
)
|
|
except Exception:
|
|
logger.warning(
|
|
"events Bronze 写入失败(record=%s, match=%s),不影响采集主流程",
|
|
source_record_id, target_match_id, exc_info=True,
|
|
)
|
|
|
|
|
|
# ============================================================
|
|
# 积分榜管线:/leagues/{id}/standings/ → standings 表
|
|
# ============================================================
|
|
|
|
async def fetch_bzzoiro_standings(league_code: str, season: str | None = None) -> dict:
|
|
"""抓取联赛积分榜(纯抓取,不入库)。season 为 None 时取当前赛季。"""
|
|
league_id = BZZOIRO_LEAGUE_IDS.get(league_code)
|
|
if league_id is None:
|
|
raise ValueError(f"未知联赛代码: {league_code}")
|
|
params: dict = {}
|
|
if season:
|
|
params["season"] = season
|
|
return await _fetch_json_async(f"/leagues/{league_id}/standings/", params)
|
|
|
|
|
|
def _season_label_from_dates(start_date, end_date) -> str:
|
|
"""从赛季起止日期推导赛季标签(与 derive_season_label 语义一致)。"""
|
|
try:
|
|
if isinstance(start_date, str):
|
|
start = datetime.fromisoformat(start_date[:10])
|
|
else:
|
|
start = start_date
|
|
y = start.year
|
|
return f"{y}-{y + 1}" if start.month >= 8 else f"{y - 1}-{y}"
|
|
except (TypeError, ValueError):
|
|
return "?"
|
|
|
|
|
|
async def ingest_bzzoiro_standings(db, *, leagues: Iterable[str], season: str | None = None) -> dict:
|
|
"""采集积分榜 → upsert standings 表。
|
|
|
|
season 为 None 时采集当前赛季(bzzoiro 默认返回 is_current 赛季)。
|
|
球队名与 events 管线使用同一 normalize 规则,保证 Team 匹配。
|
|
"""
|
|
from src.data.team_names import normalize as normalize_name
|
|
|
|
result: dict = {"leagues": {}, "total_upserted": 0, "errors": []}
|
|
for code in leagues:
|
|
league_r: dict = {"upserted": 0, "teams_created": 0, "rows": 0, "errors": []}
|
|
try:
|
|
payload = await fetch_bzzoiro_standings(code, season=season)
|
|
except Exception as e:
|
|
logger.exception("bzzoiro standings fetch failed for %s", code)
|
|
league_r["errors"].append(str(e))
|
|
await _safe_write_ingest_failure(
|
|
db,
|
|
entity_type="standings",
|
|
source_record_id=None,
|
|
error=e,
|
|
raw_payload={"league": code, "season": season},
|
|
)
|
|
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
|
|
|
|
# 联赛(get-or-create,D4: 经 LeagueRepository)
|
|
league = await LeagueRepository(db).get_or_create(
|
|
code, LEAGUE_NAMES.get(code, code), LEAGUE_COUNTRIES.get(code)
|
|
)
|
|
team_r = TeamRepository(db)
|
|
|
|
# 赛季标签:优先用返回的 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 ""
|
|
|
|
# 批量预载球队(与 events 管线使用同一 normalize 规则,保证 Team 匹配)
|
|
names = {normalize_name(str(r.get("team_name", ""))) for r in rows}
|
|
names.discard("")
|
|
team_map: dict[str, Team] = await team_r.get_all_by_names(list(names))
|
|
|
|
now = datetime.now(timezone.utc)
|
|
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 = await team_r.get_or_create(team_name, name_zh=zh_name(team_name))
|
|
team_map[team_name] = team
|
|
league_r["teams_created"] += 1
|
|
|
|
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,
|
|
)
|
|
|
|
# 同一联赛同一赛季只保留最新快照:按 (league, season, team) upsert
|
|
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 is None:
|
|
standing = Standing(
|
|
league_id=league.id, season=season_label, team_id=team.id, **values
|
|
)
|
|
db.add(standing)
|
|
else:
|
|
for k, v in values.items():
|
|
setattr(standing, k, v)
|
|
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
|
|
|
|
|
|
# ============================================================
|
|
# 统计回填管线:/events/{id}/stats/ → match_stats 表
|
|
# ============================================================
|
|
|
|
# bzzoiro stats 字段 → MatchStats 字段映射(stats.home / stats.away 下)
|
|
_STATS_FIELD_MAP = {
|
|
"xg": ("home_xg", "away_xg"), # 回退 expected_goals
|
|
"ball_possession": ("home_possession", None), # 只取主队值,客队=100-home
|
|
"total_shots": ("home_shots", "away_shots"),
|
|
"shots_on_target": ("home_shots_on_target", "away_shots_on_target"),
|
|
"corner_kicks": ("home_corners", "away_corners"),
|
|
"yellow_cards": ("home_yellow_cards", "away_yellow_cards"),
|
|
"red_cards": ("home_red_cards", "away_red_cards"),
|
|
"big_chances": ("home_big_chances", "away_big_chances"),
|
|
"fouls": ("home_fouls", "away_fouls"),
|
|
}
|
|
|
|
|
|
def _pick(d: dict, *keys):
|
|
"""按优先级取第一个非空字段值。"""
|
|
for k in keys:
|
|
v = d.get(k)
|
|
if v is not None:
|
|
return v
|
|
return None
|
|
|
|
|
|
def _stats_from_payload(payload: dict) -> dict:
|
|
"""把 /events/{id}/stats/ 响应映射成 MatchStats 字段 dict。
|
|
|
|
响应结构: {"event_id": ..., "stats": {"home": {...}, "away": {...}}}
|
|
"""
|
|
stats = (payload or {}).get("stats") or {}
|
|
home = stats.get("home") or {}
|
|
away = stats.get("away") or {}
|
|
out: dict = {}
|
|
|
|
xg_h = _pick(home, "xg", "expected_goals")
|
|
xg_a = _pick(away, "xg", "expected_goals")
|
|
if xg_h is not None:
|
|
out["home_xg"] = _to_float_or_none(xg_h)
|
|
if xg_a is not None:
|
|
out["away_xg"] = _to_float_or_none(xg_a)
|
|
|
|
poss = home.get("ball_possession")
|
|
if poss is not None:
|
|
p = _to_float_or_none(poss)
|
|
if p is not None:
|
|
out["home_possession"] = p
|
|
|
|
for src, (h_fld, a_fld) in _STATS_FIELD_MAP.items():
|
|
if src in ("xg", "ball_possession"):
|
|
continue # 已处理
|
|
hv = home.get(src)
|
|
av = away.get(src)
|
|
if hv is not None and h_fld:
|
|
out[h_fld] = _to_int_or_none(hv)
|
|
if av is not None and a_fld:
|
|
out[a_fld] = _to_int_or_none(av)
|
|
return out
|
|
|
|
|
|
def _to_float_or_none(value) -> float | None:
|
|
if value is None:
|
|
return None
|
|
try:
|
|
return float(str(value).strip())
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
async def ingest_bzzoiro_event_stats(
|
|
db,
|
|
*,
|
|
leagues: Iterable[str],
|
|
limit: int = 100,
|
|
only_missing: bool = True,
|
|
) -> dict:
|
|
"""回填已完赛比赛的详细统计(逐场调 /events/{id}/stats/)。
|
|
|
|
筛选条件: match_status=finished 且 source_event_id 非空。
|
|
only_missing=True 时跳过已有统计的比赛(增量);False 则全量刷新。
|
|
limit 控制单次最多处理的比赛数(上游限速 1.2s/请求,大批量需分次触发)。
|
|
"""
|
|
result: dict = {"fetched": 0, "created": 0, "updated": 0, "skipped": 0, "errors": []}
|
|
|
|
league_ids = [BZZOIRO_LEAGUE_IDS[c] for c in leagues if c in BZZOIRO_LEAGUE_IDS]
|
|
if not league_ids:
|
|
result["errors"].append("无有效联赛代码")
|
|
return result
|
|
|
|
# D4: 候选比赛查询经 MatchRepository(含 stats 预加载,筛选/排序/limit 语义不变)
|
|
matches = await MatchRepository(db).find_finished_with_stats(
|
|
league_ids, limit=limit * 3 if only_missing else limit
|
|
)
|
|
|
|
now = datetime.now(timezone.utc)
|
|
processed = 0
|
|
for m in matches:
|
|
if processed >= limit:
|
|
break
|
|
if only_missing and m.stats is not None and m.stats.home_shots is not None:
|
|
result["skipped"] += 1
|
|
continue
|
|
processed += 1
|
|
try:
|
|
payload = await _fetch_json_async(f"/events/{m.source_event_id}/stats/")
|
|
except Exception as e:
|
|
logger.warning("stats fetch failed match=%s event=%s: %s", m.id, m.source_event_id, e)
|
|
result["errors"].append(f"match {m.id}: {e}")
|
|
await _safe_write_ingest_failure(
|
|
db,
|
|
entity_type="match_stats",
|
|
source_record_id=str(m.source_event_id),
|
|
error=e,
|
|
raw_payload={"match_id": m.id},
|
|
)
|
|
await asyncio.sleep(REQUEST_INTERVAL)
|
|
continue
|
|
|
|
result["fetched"] += 1
|
|
fields = _stats_from_payload(payload)
|
|
if not fields:
|
|
result["skipped"] += 1
|
|
await asyncio.sleep(REQUEST_INTERVAL)
|
|
continue
|
|
|
|
if m.stats is None:
|
|
available_at = m.match_date + timedelta(hours=2) if m.match_date else now
|
|
m.stats = MatchStats(
|
|
match_id=m.id,
|
|
source="bzzoiro",
|
|
source_record_id=str(m.source_event_id),
|
|
retrieved_at=now,
|
|
available_at=available_at,
|
|
)
|
|
db.add(m.stats)
|
|
result["created"] += 1
|
|
else:
|
|
result["updated"] += 1
|
|
if m.stats.source is None:
|
|
m.stats.source = "bzzoiro"
|
|
m.stats.source_record_id = str(m.source_event_id)
|
|
if m.stats.retrieved_at is None:
|
|
m.stats.retrieved_at = now
|
|
if m.stats.available_at is None and m.match_date:
|
|
m.stats.available_at = m.match_date + timedelta(hours=2)
|
|
|
|
for fld, v in fields.items():
|
|
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(
|
|
"bzzoiro stats 回填完成: 抓取 %d, 新建 %d, 更新 %d, 跳过 %d, 错误 %d",
|
|
result["fetched"], result["created"], result["updated"],
|
|
result["skipped"], len(result["errors"]),
|
|
)
|
|
return result
|