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 监听后弹出登录页 */
+/** Admin 请求可覆盖项(与 lib/http RequestOptions 对齐的子集) */
-export const UNAUTHORIZED_EVENT = 'profeto:unauthorized'
+type ApiOpts = {
-
+  timeoutMs?: number
-export class ApiError extends Error {
+  /** 改密接口的 401 表示「当前密码错误」,非会话过期,置 true 跳过登出广播 */
-  constructor(
+  skipAuthHandling?: boolean
-    message: string,
-    public status: number,
-    public data?: unknown,
-  ) {
-    super(message)
-    this.name = 'ApiError'
-  }
-}
-
-async function request<T>(
-  path: string,
-  options: RequestInit & { timeoutMs?: number; skipAuthHandling?: boolean } = {},
-): Promise<T> {
-  // 修复: 正确拼接 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<T>
-  } 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)
-  }
 }
 
 export const api = {
-  get: <T>(path: string) => request<T>(path),
+  get: <T>(path: string) => http.get<T>(path),
-  post: <T>(path: string, body?: unknown, opts?: { timeoutMs?: number; skipAuthHandling?: boolean }) =>
+  post: <T>(path: string, body?: unknown, opts?: ApiOpts) =>
-    request<T>(path, { method: 'POST', body: body ? JSON.stringify(body) : undefined, ...opts }),
+    http.post<T>(path, body, opts),
-  put: <T>(path: string, body?: unknown, opts?: { timeoutMs?: number; skipAuthHandling?: boolean }) =>
+  put: <T>(path: string, body?: unknown, opts?: ApiOpts) =>
-    request<T>(path, { method: 'PUT', body: body ? JSON.stringify(body) : undefined, ...opts }),
+    http.put<T>(path, body, opts),
-  delete: <T>(path: string) => request<T>(path, { method: 'DELETE' }),
+  delete: <T>(path: string) => http.delete<T>(path),
 }
 
 // ── 认证 ────────────────────────────────────────────────────────

frontend/src/admin/dal.ts

@@ -32,23 +32,21 @@ import type {
  * 从多个端点聚合仪表盘数据。
  * 后端暂无专用仪表盘端点,这里组合 health + 各列表端点。
  *
- * P2-8 修复: 使用 items.length 替代不存在的 total 字段,
+ * 注: 比赛真实总量请用 fetchAdminStats()(GET /admin/stats,
- * 并扩大 limit 以获得更有参考价值的数量。
+ * 后端 COUNT(*) 精确计数)。此处曾用 matches items.length 近似,
+ * 已随仪表盘切换真实计数而移除,防止误用 100 上限的假总量。
  */
 export async function fetchDashboard(): Promise<DashboardStats> {
   // 并行获取各端点数据
-  // matches 返回 {items, next_cursor, has_more}, predictions 返回数组
+  // predictions 返回数组
-  const [leagues, matches, predictions, health] = await Promise.allSettled([
+  const [leagues, predictions, health] = await Promise.allSettled([
     api.get<League[]>(`${API_BASE}/leagues`),
-    api.get<{ items: Match[]; has_more: boolean }>(`${API_BASE}/matches?limit=100`),
     api.get<Prediction[]>(`${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',

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 }[]

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

frontend/src/pages/Matches.tsx

@@ -82,7 +82,12 @@ const CN_NUM = ['一', '二', '三', '四', '五', '六', '七', '八']
 const STATUS_META: Record<string, { label: string; cls: string }> = {
   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 → 中文标签 */

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
 

src/db/models.py

@@ -325,8 +325,10 @@ class RawEvent(Base):
 class IngestFailure(Base):
     """采集失败死信:记录失败原因、重试次数与下次重试时间。
 
-    ⚠️ 预留未启用:当前 bzzoiro 管线不写入此表。
+    bzzoiro 三条管线(events / standings / stats)抓取失败时经由
-    未来接线计划:采集失败时写入,支持按 next_retry_at 自动重试。
+    bzzoiro._safe_write_ingest_failure 写入本表(尽力而为,写入失败
+    不影响采集主流程)。admin 可通过 /admin/schedules/ingest-failures
+    查看与重试。
     """
     __tablename__ = "ingest_failures"
 

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"]