shangfangjian/Profeto
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request 'fix: 死信表真正接线 + 前端 HTTP 收敛与状态修正' (#9) from fix-deadletter-frontend-cleanup into main

Browse Source 
Reviewed-on: #9
This commit was merged in pull request #9.
This commit is contained in:
shangfangjian
2026-09-21 18:39:08 +08:00
parent b5f7e682f0 b6a251407d
commit f6c0145c32
8 changed files with 297 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frontend/src/admin/api.ts
+20 -87
View File
@@ -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<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)
}
/** Admin 请求可覆盖项(与 lib/http RequestOptions 对齐的子集) */
type ApiOpts = {
timeoutMs?: number
/** 改密接口的 401 表示「当前密码错误」,非会话过期,置 true 跳过登出广播 */
skipAuthHandling?: boolean
}
export const api = {
get: <T>(path: string) => request<T>(path),
post: <T>(path: string, body?: unknown, opts?: { timeoutMs?: number; skipAuthHandling?: boolean }) =>
request<T>(path, { method: 'POST', body: body ? JSON.stringify(body) : undefined, ...opts }),
put: <T>(path: string, body?: unknown, opts?: { timeoutMs?: number; skipAuthHandling?: boolean }) =>
request<T>(path, { method: 'PUT', body: body ? JSON.stringify(body) : undefined, ...opts }),
delete: <T>(path: string) => request<T>(path, { method: 'DELETE' }),
get: <T>(path: string) => http.get<T>(path),
post: <T>(path: string, body?: unknown, opts?: ApiOpts) =>
http.post<T>(path, body, opts),
put: <T>(path: string, body?: unknown, opts?: ApiOpts) =>
http.put<T>(path, body, opts),
delete: <T>(path: string) => http.delete<T>(path),
}
// ── 认证 ────────────────────────────────────────────────────────
frontend/src/admin/dal.ts
+5 -7
View File
@@ -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<DashboardStats> {
// 并行获取各端点数据
// 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<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
+2 -1
View File
@@ -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
+4 -1
View File
@@ -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
+6 -1
View File
@@ -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
+47 -1
View File
@@ -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
+4 -2
View File
@@ -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"
tests/test_ingest_deadletter.py
+209
View File
@@ -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"]