diff --git a/frontend/src/admin/api.ts b/frontend/src/admin/api.ts index 9ab4274..07d37ab 100644 --- a/frontend/src/admin/api.ts +++ b/frontend/src/admin/api.ts @@ -1,102 +1,35 @@ /** - * Admin 后台管理系统 - 统一 API 客户端 + * Admin 后台管理系统 - 统一 API 客户端(门面) * * 鉴权:通过 POST /api/v1/auth/login 用密码换取 HttpOnly Cookie 会话, * 同源请求自动携带 Cookie,无需手动管理密钥。 * 收到 401 时广播 `profeto:unauthorized` 事件,由 AdminLayout 切回登录页。 + * + * 实现已收敛到共享层 lib/http.ts(超时/错误解析/401 广播只此一份), + * 本文件仅保留 Admin 侧的门面签名与认证接口,供既有页面按原路径导入。 */ +import { http, ApiError, UNAUTHORIZED_EVENT } from '../lib/http' + +/** Admin 侧兼容导出:错误类型与会话失效事件名的规范来源在 lib/http */ +export { ApiError, UNAUTHORIZED_EVENT } + const API_BASE = '/api/v1' -const TIMEOUT_MS = 30_000 -/** 会话失效事件名,AdminLayout 监听后弹出登录页 */ -export const UNAUTHORIZED_EVENT = 'profeto:unauthorized' - -export class ApiError extends Error { - constructor( - message: string, - public status: number, - public data?: unknown, - ) { - super(message) - this.name = 'ApiError' - } -} - -async function request( - path: string, - options: RequestInit & { timeoutMs?: number; skipAuthHandling?: boolean } = {}, -): Promise { - // 修复: 正确拼接 API_BASE - const url = path.startsWith('http') - ? path - : path.startsWith('/') - ? path // 已经是绝对路径(如 /health) - : `${API_BASE}${path}` - - const { timeoutMs = TIMEOUT_MS, skipAuthHandling, ...fetchOptions } = options - const controller = new AbortController() - const timer = setTimeout(() => controller.abort(), timeoutMs) - - try { - const res = await fetch(url, { - ...fetchOptions, - signal: controller.signal, - headers: { - 'Content-Type': 'application/json', - ...fetchOptions.headers, - }, - }) - - if (!res.ok) { - // 先读 text 再尝试 JSON 解析:Response body 流只能读一次, - // 若先调 res.json() 失败(如返回 HTML 错误页),再调 res.text() 会抛 "body stream already read"。 - const rawText = await res.text() - let detail: unknown = rawText - try { - detail = JSON.parse(rawText) - } catch { - // 非 JSON(如 HTML 错误页),保留原始文本 - } - let message = - detail && typeof detail === 'object' && detail !== null && 'detail' in detail - ? String((detail as { detail: unknown }).detail) - : `HTTP ${res.status}: ${res.statusText}` - // 401 仅在没有显式跳过时广播未登录事件(改密接口的 401 表示当前密码错误,非会话过期) - if (res.status === 401 && !skipAuthHandling) { - message += '\n登录已过期,请重新登录。' - window.dispatchEvent(new CustomEvent(UNAUTHORIZED_EVENT)) - } - throw new ApiError(message, res.status, detail) - } - - // 修复: 正确判断 204 No Content - if (res.status === 204) { - return undefined as T - } - - return res.json() as Promise - } catch (err) { - if (err instanceof ApiError) throw err - if (err instanceof DOMException && err.name === 'AbortError') { - throw new ApiError('请求超时,请稍后重试', 0) - } - throw new ApiError( - err instanceof Error ? err.message : '网络错误,请检查连接', - 0, - ) - } finally { - clearTimeout(timer) - } +/** Admin 请求可覆盖项(与 lib/http RequestOptions 对齐的子集) */ +type ApiOpts = { + timeoutMs?: number + /** 改密接口的 401 表示「当前密码错误」,非会话过期,置 true 跳过登出广播 */ + skipAuthHandling?: boolean } export const api = { - get: (path: string) => request(path), - post: (path: string, body?: unknown, opts?: { timeoutMs?: number; skipAuthHandling?: boolean }) => - request(path, { method: 'POST', body: body ? JSON.stringify(body) : undefined, ...opts }), - put: (path: string, body?: unknown, opts?: { timeoutMs?: number; skipAuthHandling?: boolean }) => - request(path, { method: 'PUT', body: body ? JSON.stringify(body) : undefined, ...opts }), - delete: (path: string) => request(path, { method: 'DELETE' }), + get: (path: string) => http.get(path), + post: (path: string, body?: unknown, opts?: ApiOpts) => + http.post(path, body, opts), + put: (path: string, body?: unknown, opts?: ApiOpts) => + http.put(path, body, opts), + delete: (path: string) => http.delete(path), } // ── 认证 ──────────────────────────────────────────────────────── diff --git a/frontend/src/admin/dal.ts b/frontend/src/admin/dal.ts index 8aa3e9e..e81075f 100644 --- a/frontend/src/admin/dal.ts +++ b/frontend/src/admin/dal.ts @@ -32,23 +32,21 @@ import type { * 从多个端点聚合仪表盘数据。 * 后端暂无专用仪表盘端点,这里组合 health + 各列表端点。 * - * P2-8 修复: 使用 items.length 替代不存在的 total 字段, - * 并扩大 limit 以获得更有参考价值的数量。 + * 注: 比赛真实总量请用 fetchAdminStats()(GET /admin/stats, + * 后端 COUNT(*) 精确计数)。此处曾用 matches items.length 近似, + * 已随仪表盘切换真实计数而移除,防止误用 100 上限的假总量。 */ export async function fetchDashboard(): Promise { // 并行获取各端点数据 - // matches 返回 {items, next_cursor, has_more}, predictions 返回数组 - const [leagues, matches, predictions, health] = await Promise.allSettled([ + // predictions 返回数组 + const [leagues, predictions, health] = await Promise.allSettled([ api.get(`${API_BASE}/leagues`), - api.get<{ items: Match[]; has_more: boolean }>(`${API_BASE}/matches?limit=100`), api.get(`${API_BASE}/predictions?limit=100`), api.get<{ status: string }>('/health'), ]) return { leagues: leagues.status === 'fulfilled' ? leagues.value : [], - // P2-8: matches 无 total 字段,用 items.length 近似(上限 100) - total_matches: matches.status === 'fulfilled' ? (matches.value as any)?.items?.length ?? 0 : 0, // predictions 直接返回数组 total_predictions: predictions.status === 'fulfilled' ? (predictions.value as any)?.length ?? 0 : 0, health: health.status === 'fulfilled' ? (health.value as any).status : 'unknown', diff --git a/frontend/src/admin/types.ts b/frontend/src/admin/types.ts index 580d08e..89a5a88 100644 --- a/frontend/src/admin/types.ts +++ b/frontend/src/admin/types.ts @@ -17,7 +17,8 @@ export interface HealthStatus { export interface DashboardStats { leagues: League[] - total_matches: number + // 比赛总量已移除: 用 fetchAdminStats()(/admin/stats)的精确 COUNT, + // 不要再用列表 items.length 近似(上限 100 会严重失真) total_predictions: number health: string db_tables: { name: string; row_count: number; size_mb: number; last_updated: string | null }[] diff --git a/frontend/src/lib/http.ts b/frontend/src/lib/http.ts index 3e57e18..ddfabb3 100644 --- a/frontend/src/lib/http.ts +++ b/frontend/src/lib/http.ts @@ -7,9 +7,12 @@ * - 401 自动广播(Admin 场景) * - JSON/HTML 容错解析 * - 请求竞态防护(可选 signal) + * + * 注: Admin 侧的 admin/api.ts 是本模块的薄门面,不再有第二套实现。 */ -import { UNAUTHORIZED_EVENT } from '../admin/api' +/** 会话失效事件名,AdminLayout 监听后弹出登录页 */ +export const UNAUTHORIZED_EVENT = 'profeto:unauthorized' const API_BASE = '/api/v1' const DEFAULT_TIMEOUT = 30_000 diff --git a/frontend/src/pages/Matches.tsx b/frontend/src/pages/Matches.tsx index bc24a4a..4f102ef 100644 --- a/frontend/src/pages/Matches.tsx +++ b/frontend/src/pages/Matches.tsx @@ -82,7 +82,12 @@ const CN_NUM = ['一', '二', '三', '四', '五', '六', '七', '八'] const STATUS_META: Record = { finished: { label: '已完赛', cls: 'text-ink-400' }, scheduled: { label: '未开赛', cls: 'text-ink-600' }, - live: { label: '进行中', cls: 'text-press font-medium' }, + // 键与 normalize.py 的 VALID_STATUS 对齐: 库里存的是 in_play(上游 live 被归一化),不存在 'live' 状态 + in_play: { label: '进行中', cls: 'text-press font-medium' }, + paused: { label: '暂停', cls: 'text-press font-medium' }, + postponed: { label: '延期', cls: 'text-ink-400' }, + cancelled: { label: '取消', cls: 'text-ink-400' }, + suspended: { label: '中止', cls: 'text-ink-400' }, } /** 1x2 → 中文标签 */ diff --git a/src/data/bzzoiro.py b/src/data/bzzoiro.py index 0278828..29db35d 100644 --- a/src/data/bzzoiro.py +++ b/src/data/bzzoiro.py @@ -200,6 +200,13 @@ class BzzoiroSource: # 单联赛抓取失败隔离:记录错误后继续其余联赛,不拖垮整批 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 @@ -367,6 +374,31 @@ async def _write_ingest_failure(db, source_system: str, entity_type: str, source )) +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( @@ -418,12 +450,19 @@ async def ingest_bzzoiro_standings(db, *, leagues: Iterable[str], season: str | result: dict = {"leagues": {}, "total_upserted": 0, "errors": []} for code in leagues: - league_r: dict = {"upserted": 0, "teams_created": 0, "rows": 0} + 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 @@ -634,6 +673,13 @@ async def ingest_bzzoiro_event_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 diff --git a/src/db/models.py b/src/db/models.py index bafff7a..b242245 100644 --- a/src/db/models.py +++ b/src/db/models.py @@ -325,8 +325,10 @@ class RawEvent(Base): class IngestFailure(Base): """采集失败死信:记录失败原因、重试次数与下次重试时间。 - ⚠️ 预留未启用:当前 bzzoiro 管线不写入此表。 - 未来接线计划:采集失败时写入,支持按 next_retry_at 自动重试。 + bzzoiro 三条管线(events / standings / stats)抓取失败时经由 + bzzoiro._safe_write_ingest_failure 写入本表(尽力而为,写入失败 + 不影响采集主流程)。admin 可通过 /admin/schedules/ingest-failures + 查看与重试。 """ __tablename__ = "ingest_failures" diff --git a/tests/test_ingest_deadletter.py b/tests/test_ingest_deadletter.py new file mode 100644 index 0000000..91aef6b --- /dev/null +++ b/tests/test_ingest_deadletter.py @@ -0,0 +1,209 @@ +"""死信接线回归测试: bzzoiro 三条管线抓取失败必须写入 IngestFailure。 + +背景: IngestFailure 死信表此前「预留未启用」——events / standings / stats +抓取失败只打日志 + errors 列表,admin 的 /ingest-failures 列表与 retry +端点形同虚设。本测试守护三条管线的失败写入路径: + 1. events 整联赛抓取失败 → entity_type="events" + 2. standings 整联赛抓取失败 → entity_type="standings" + 3. stats 单场统计抓取失败 → entity_type="match_stats"(带 source_record_id) + +范式: 假 db(记录 add 调用) + monkeypatch 抓取函数,不依赖真实数据库 —— +与 test_review_required_fixes.py R2 相同。失败写入是「尽力而为」: +写入器自身抛错只记日志,不得拖垮采集主流程(最后一个测试守护)。 +""" +from __future__ import annotations + +from types import SimpleNamespace + +import pytest + +import src.data.bzzoiro as bz +from src.db.models import IngestFailure + + +class _FakeResult: + """支持 .scalars().all() / .scalar_one_or_none() / .scalar() 的最小假结果集。""" + + def __init__(self, items): + self._items = items + + def scalars(self): + return self + + def all(self): + return self._items + + def scalar_one_or_none(self): + return None + + def scalar(self): + return None + + +class _FakeDB: + """只记录 add() 的假会话 —— 失败路径不触发真实查询。""" + + def __init__(self, items=None): + self.added = [] + self._items = items or [] + + def add(self, obj): + self.added.append(obj) + + async def execute(self, stmt): + return _FakeResult(self._items) + + async def flush(self): + pass + + +@pytest.fixture(autouse=True) +def _no_request_interval(monkeypatch): + """失败路径会 await asyncio.sleep(REQUEST_INTERVAL),置 0 加速测试。""" + monkeypatch.setattr(bz, "REQUEST_INTERVAL", 0) + + +def _failures(db: _FakeDB) -> list[IngestFailure]: + return [o for o in db.added if isinstance(o, IngestFailure)] + + +# ============================================================ +# 1. events: 整联赛抓取失败 +# ============================================================ + + +class TestEventsFetchFailureDeadLetter: + async def test_writes_deadletter_with_league_context(self, monkeypatch): + async def _boom(*args, **kwargs): + raise RuntimeError("network down") + + monkeypatch.setattr(bz, "fetch_bzzoiro_events", _boom) + db = _FakeDB() + + result = await bz.BzzoiroSource().ingest(db, leagues=["E0"]) + + rows = _failures(db) + assert len(rows) == 1 + row = rows[0] + assert row.source_system == "bzzoiro" + assert row.entity_type == "events" + assert row.error_type == "fetch_error" + assert "network down" in row.error_detail + # 上下文足够管理员定位:联赛代码必须随行 + assert row.raw_payload is not None + assert row.raw_payload.get("league") == "E0" + # 主流程不受影响:错误仍记录在 result 中 + assert result["leagues"]["E0"]["errors"] + assert result["total_inserted"] == 0 + + async def test_other_leagues_continue_after_failure(self, monkeypatch): + async def _boom(league_code, **kwargs): + if league_code == "E0": + raise RuntimeError("boom") + return [] + + monkeypatch.setattr(bz, "fetch_bzzoiro_events", _boom) + db = _FakeDB() + + await bz.BzzoiroSource().ingest(db, leagues=["E0", "SP1"]) + + rows = _failures(db) + assert len(rows) == 1 + assert rows[0].raw_payload.get("league") == "E0" + + +# ============================================================ +# 2. standings: 整联赛抓取失败 +# ============================================================ + + +class TestStandingsFetchFailureDeadLetter: + async def test_writes_deadletter_with_season_context(self, monkeypatch): + async def _boom(league_code, season=None): + raise RuntimeError("upstream 500") + + monkeypatch.setattr(bz, "fetch_bzzoiro_standings", _boom) + db = _FakeDB() + + result = await bz.ingest_bzzoiro_standings(db, leagues=["SP1"], season="2025-2026") + + rows = _failures(db) + assert len(rows) == 1 + row = rows[0] + assert row.entity_type == "standings" + assert row.error_type == "fetch_error" + assert "upstream 500" in row.error_detail + assert row.raw_payload == {"league": "SP1", "season": "2025-2026"} + assert result["errors"] + + +# ============================================================ +# 3. stats: 单场统计抓取失败 +# ============================================================ + + +class TestStatsFetchFailureDeadLetter: + def _match(self) -> SimpleNamespace: + return SimpleNamespace( + id=42, + source_event_id=777, + stats=None, + match_date=None, + league_id=1, + match_status="finished", + ) + + async def test_writes_deadletter_with_source_record_id(self, monkeypatch): + async def _boom(path, params=None, max_retries=3): + raise TimeoutError("read timeout") + + monkeypatch.setattr(bz, "_fetch_json_async", _boom) + db = _FakeDB(items=[self._match()]) + + result = await bz.ingest_bzzoiro_event_stats(db, leagues=["E0"], limit=1) + + rows = _failures(db) + assert len(rows) == 1 + row = rows[0] + assert row.entity_type == "match_stats" + # source_record_id 必须是上游 event id,retry 端点据此定位 + assert row.source_record_id == "777" + assert "timeout" in (row.error_detail or "").lower() + assert row.raw_payload == {"match_id": 42} + assert result["errors"] + + async def test_success_path_does_not_write_deadletter(self, monkeypatch): + async def _ok(path, params=None, max_retries=3): + return {"stats": {"home": {"total_shots": 10}, "away": {"total_shots": 5}}} + + monkeypatch.setattr(bz, "_fetch_json_async", _ok) + db = _FakeDB(items=[self._match()]) + + await bz.ingest_bzzoiro_event_stats(db, leagues=["E0"], limit=1) + + assert _failures(db) == [] + + +# ============================================================ +# 4. 死信写入自身失败: 尽力而为,不拖垮主流程 +# ============================================================ + + +class TestDeadLetterWriteIsBestEffort: + async def test_db_add_failure_does_not_break_ingest(self, monkeypatch): + async def _boom(*args, **kwargs): + raise RuntimeError("network down") + + monkeypatch.setattr(bz, "fetch_bzzoiro_events", _boom) + + class _BrokenDB(_FakeDB): + def add(self, obj): + if isinstance(obj, IngestFailure): + raise RuntimeError("session closed") + super().add(obj) + + db = _BrokenDB() + + # 不应抛异常:死信写不进去只记 warning + result = await bz.BzzoiroSource().ingest(db, leagues=["E0"]) + assert result["leagues"]["E0"]["errors"]