diff --git a/README.md b/README.md index f325e6a..b370729 100644 --- a/README.md +++ b/README.md @@ -21,7 +21,7 @@ │ │ ┌──────────▼──────────┐ ┌─────────────▼────────────┐ │ PostgreSQL │ │ LLM (OpenAI-compatible) │ -│ 12 张表 │ │ OpenAI / Deepseek / │ +│ 13 张表 │ │ OpenAI / Deepseek / │ │ leagues/teams/ │ │ Ollama / 任意网关 │ │ matches/match_ │ └──────────────────────────┘ │ stats/standings/ │ diff --git a/alembic/versions/0019_ingest_jobs.py b/alembic/versions/0019_ingest_jobs.py new file mode 100644 index 0000000..1851cf5 --- /dev/null +++ b/alembic/versions/0019_ingest_jobs.py @@ -0,0 +1,46 @@ +"""新增 ingest_jobs 表 + +Revision ID: 0019_ingest_jobs +Revises: 0018_match_checks +Create Date: 2026-09-21 + +采集任务状态跟踪:POST /ingest/bzzoiro 创建 pending job 后台执行, +解决 fire-and-forget 不可观测问题(admin 可查询任务级状态与结果摘要)。 +""" + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects.postgresql import JSONB + +# revision identifiers, used by Alembic. +revision: str = '0019_ingest_jobs' +down_revision: Union[str, None] = '0018_match_checks' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + 'ingest_jobs', + sa.Column('id', sa.String(36), primary_key=True), + sa.Column('task', sa.String(20), nullable=False), + sa.Column('params', JSONB, nullable=False), + sa.Column('status', sa.String(20), nullable=False, server_default='pending'), + sa.Column('result', JSONB), + sa.Column('error', sa.Text()), + sa.Column('created_at', sa.DateTime(timezone=True)), + sa.Column('started_at', sa.DateTime(timezone=True)), + sa.Column('finished_at', sa.DateTime(timezone=True)), + sa.CheckConstraint( + "status IN ('pending','running','success','failed')", + name='ck_ingest_jobs_status', + ), + ) + op.create_index('ix_ingest_jobs_created', 'ingest_jobs', ['created_at']) + + +def downgrade() -> None: + op.drop_index('ix_ingest_jobs_created', table_name='ingest_jobs') + op.drop_table('ingest_jobs') diff --git a/docs/01-architecture.md b/docs/01-architecture.md index 535ea5a..be09f06 100644 --- a/docs/01-architecture.md +++ b/docs/01-architecture.md @@ -20,7 +20,7 @@ │ │ │ └─ aggregator 终裁(强模型) │ │ │ │ └────────────┘ │ events / standings │ │ │ ┌──┴──────────────┴──┐ │ /stats 三条管线 │ │ -│ │ PostgreSQL (12 张表)│ └───────────────────┘ │ +│ │ PostgreSQL (13 张表)│ └───────────────────┘ │ │ └────────────────────┘ httpx → 外部 API │ └─────────────────────────────────────────────────────┘ ``` @@ -70,7 +70,7 @@ Profeto/ │ │ └── schedules.py # 定时任务 + 死信重试(router 级鉴权) │ ├── db/ │ │ ├── base.py # async engine + get_db/get_db_read -│ │ ├── models.py # 12 张表 ORM +│ │ ├── models.py # 13 张表 ORM │ │ ├── repositories.py # 仓储层 │ │ └── unit_of_work.py # 事务边界 │ ├── data/ diff --git a/docs/03-api.md b/docs/03-api.md index 4a932f5..ad98033 100644 --- a/docs/03-api.md +++ b/docs/03-api.md @@ -190,9 +190,27 @@ Base URL: `http://localhost:8000` · 交互式文档: `/docs`(Swagger)与 `/redo | `season` | 当前赛季 | standings 赛季,如 `"2026-2027"` | | `limit` | 100 | stats 回填单次最大比赛数(1–500) | -- 响应含每联赛 `inserted`/`updated`/`errors` 统计 +- 响应:`{"ok": true, "job_id": "", "message": "……"}`,`job_id` 用于查询任务状态 - `task=stats` 只补空字段、不创建比赛(xG/射门/控球等统计回填) +### `GET /api/v1/admin/ingest/jobs/{job_id}`(需管理员) + +查询一次采集任务的状态(`ingest_jobs` 表,任务级可观测性): + +```json +{"id": "…", "task": "standings", "params": {"leagues": ["E0"]}, + "status": "success", "result": {"total_upserted": 20, "errors": []}, + "error": null, "created_at": "…", "started_at": "…", "finished_at": "…"} +``` + +- `status` 取值:`pending`(已创建未开始)/ `running` / `success` / `failed` +- 管理端采集页提交后凭 `job_id` 轮询本端点直至终态 +- 与 `ingest_failures` 死信独立:死信记录单条管线抓取失败(行级),job 记录整次任务结果 + +### `GET /api/v1/admin/ingest/jobs?limit=20&status=`(需管理员) + +列出最近采集任务(最新在前),可按 `status` 过滤,`limit` 1–100。 + > 历史版本曾有独立的 understat(xG)与 injuries(伤停)采集端点, > 已随数据源收敛为 bzzoiro 唯一来源而移除。 diff --git a/docs/05-data.md b/docs/05-data.md index e648bf0..1d0e4d3 100644 --- a/docs/05-data.md +++ b/docs/05-data.md @@ -65,7 +65,7 @@ ## 数据库 Schema -12 张表:核心业务表 5 张见下方 DDL,其余 7 张(积分榜/配置/调度/治理)见后文表格。 +12 张业务/配置表 + 1 张任务状态表(`ingest_jobs`):核心业务表 5 张见下方 DDL,其余见后文表格。 ```sql -- 联赛 @@ -146,13 +146,14 @@ CREATE TABLE predictions ( ); ``` -其余 7 张表(DDL 略,详见 `src/db/models.py` 与 alembic 迁移): +其余 8 张表(DDL 略,详见 `src/db/models.py` 与 alembic 迁移): | 表 | 状态 | 用途 | |---|---|---| | `standings` | 已启用 | 联赛积分榜快照,按 `(league_id, season, team_id)` upsert,同联赛同赛季只保留最新快照;含排名/战绩/进失球/积分/分区(zone) | | `app_settings` | 已启用 | 后台运行时设置(如数据源 API Key),读取时优先于 `.env` 默认值 | | `schedules` | 已启用 | 定时采集任务配置(task/cron/leagues/enabled),供内置调度器执行 | +| `ingest_jobs` | 已启用 | 采集任务状态:POST ingest 创建 pending,后台流转 running→success/failed;result 存统计摘要,供 admin 轮询(任务级,与死信互补) | | `raw_events` | 预留未启用 | Bronze 层原始事件存档;规划中用于重放与审计 | | `ingest_failures` | 已启用 | 采集失败死信:bzzoiro 三条管线(events/standings/stats)抓取失败时写入,admin 后台可查看与重试 | | `data_quality_checks` | 预留未启用 | 数据质量检查结果;规划中定时检查比赛/统计/积分榜完整性 | diff --git a/docs/07-development.md b/docs/07-development.md index cc6dc24..755b1f8 100644 --- a/docs/07-development.md +++ b/docs/07-development.md @@ -129,7 +129,8 @@ Profeto/ │ ├── 0003_injuries.py │ ├── 0004_snapshot_and_constraints.py │ ├── 0005_prediction_status_and_stats_provenance.py -│ └── 0006_schema_model_drift_cleanup.py +│ ├── 0006_schema_model_drift_cleanup.py +│ └── …… (共 19 个迁移,最新 0019_ingest_jobs) ├── tests/ # 测试 │ ├── test_core.py # 核心逻辑测试 │ └── test_agents.py # 多 agent 测试 diff --git a/frontend/src/admin/dal.ts b/frontend/src/admin/dal.ts index e81075f..0f0d7f4 100644 --- a/frontend/src/admin/dal.ts +++ b/frontend/src/admin/dal.ts @@ -21,6 +21,7 @@ import type { LLMAgentConfig, LogEntry, IngestSourceStatus, + IngestJob, MatchDetailOut, MatchContextOut, AdminStats, @@ -71,6 +72,11 @@ export async function triggerCollection(req: CollectionRequest): Promise { return api.post(`${API_BASE}/ingest/bzzoiro`, body) } +// 查询采集任务状态(ingest_jobs;提交采集返回 job_id 后轮询用) +export async function fetchIngestJob(jobId: string): Promise { + return api.get(`${API_BASE}/admin/ingest/jobs/${jobId}`) +} + // ── 预测管理 ──────────────────────────────────────────────────── export async function triggerPrediction(req: { match_id: number; mode?: string }): Promise { diff --git a/frontend/src/admin/pages/Collection.tsx b/frontend/src/admin/pages/Collection.tsx index 4591947..74a18b9 100644 --- a/frontend/src/admin/pages/Collection.tsx +++ b/frontend/src/admin/pages/Collection.tsx @@ -11,9 +11,9 @@ */ import { useEffect, useState, useCallback, useRef } from 'react' -import { triggerCollection, fetchLeagues, fetchIngestStatus } from '../dal' +import { triggerCollection, fetchLeagues, fetchIngestStatus, fetchIngestJob } from '../dal' import type { IngestSourceStatus } from '../types' -import type { CollectionRequest, League } from '../types' +import type { CollectionRequest, League, IngestJob } from '../types' import { Card, CardBody, CardHeader, Badge, SectionHeader, Alert, Spinner } from '../components' const TASKS = [ @@ -25,6 +25,32 @@ const TASKS = [ type TaskStatus = 'idle' | 'running' | 'done' | 'error' +const JOB_STATUS_LABEL: Record = { + pending: '等待中', + running: '执行中', + success: '已完成', + failed: '失败', +} + +/** job.result 摘要 → 多行文本(错误只显示条数,明细见「采集失败」页与系统日志) */ +function summarizeResult(result: Record | null): string[] { + if (!result) return [] + return Object.entries(result).map(([k, v]) => { + if (v && typeof v === 'object') { + const bits: string[] = [] + for (const [kk, vv] of Object.entries(v)) { + if (kk === 'errors') { + if (Array.isArray(vv) && vv.length > 0) bits.push(`错误 ${vv.length} 条`) + } else if (vv !== null && vv !== undefined) { + bits.push(`${kk} ${vv}`) + } + } + return `${k}: ${bits.length ? bits.join(' · ') : '无变更'}` + } + return `${k}: ${String(v)}` + }) +} + export default function CollectionPage() { const [leagues, setLeagues] = useState([]) const [task, setTask] = useState('events') @@ -55,7 +81,10 @@ export default function CollectionPage() { const [taskStatus, setTaskStatus] = useState('idle') const [taskStartedAt, setTaskStartedAt] = useState(null) const pollRef = useRef | null>(null) + const jobPollRef = useRef | null>(null) const [ingestSnap, setIngestSnap] = useState(null) + // 采集任务状态(ingest_jobs):提交后有 job_id 即轮询,替代「30 秒盲等」 + const [job, setJob] = useState(null) const loadLeagues = useCallback(async () => { const lg = await fetchLeagues() @@ -80,7 +109,27 @@ export default function CollectionPage() { if (pollRef.current) { clearInterval(pollRef.current); pollRef.current = null } }, []) - useEffect(() => () => stopPolling(), [stopPolling]) + // job 轮询:3 秒一次,终态(success/failed)自动停止;10 分钟兜底防泄漏 + const stopJobPolling = useCallback(() => { + if (jobPollRef.current) { clearInterval(jobPollRef.current); jobPollRef.current = null } + }, []) + + const startJobPolling = useCallback((jobId: string) => { + stopJobPolling() + let ticks = 0 + jobPollRef.current = setInterval(async () => { + ticks += 1 + if (ticks > 200) { stopJobPolling(); setTaskStatus('done'); return } + try { + const j = await fetchIngestJob(jobId) + setJob(j) + if (j.status === 'success') { stopJobPolling(); setTaskStatus('done') } + else if (j.status === 'failed') { stopJobPolling(); setTaskStatus('error') } + } catch { /* 网络抖动忽略,下个周期重试 */ } + }, 3_000) + }, [stopJobPolling]) + + useEffect(() => () => { stopPolling(); stopJobPolling() }, [stopPolling, stopJobPolling]) const isEventsTask = task === 'events' || task === 'all' @@ -88,6 +137,7 @@ export default function CollectionPage() { e.preventDefault() setError(null) setResult(null) + setJob(null) setLoading(true) setTaskStatus('running') setTaskStartedAt(Date.now()) @@ -103,22 +153,32 @@ export default function CollectionPage() { date_from: isEventsTask ? dateFrom || undefined : undefined, date_to: isEventsTask ? dateTo || undefined : undefined, } - await triggerCollection(body) - setResult({ - title: '采集任务已启动', - detail: '正在后台执行(上游限速时可能需要数分钟)。完成结果与错误请到「系统日志」页查看(支持自动刷新)。', - }) - // 启动轮询,跟踪状态 - startPolling() - // 30 秒后自动停止轮询并标记完成 - setTimeout(() => { - setTaskStatus('done') - stopPolling() - }, 30_000) + const resp = await triggerCollection(body) + const jobId: string | undefined = resp?.job_id + if (jobId) { + // 有 job_id:轮询任务状态直至终态(3 秒/次) + setResult({ + title: '采集任务已启动', + detail: `任务 ID: ${jobId}。正在跟踪执行进度,完成后展示结果摘要。`, + }) + startJobPolling(jobId) + } else { + // 兜底:后端未返回 job_id(旧版本),退回「30 秒盲等 + 系统日志」 + setResult({ + title: '采集任务已启动', + detail: '正在后台执行(上游限速时可能需要数分钟)。完成结果与错误请到「系统日志」页查看(支持自动刷新)。', + }) + startPolling() + setTimeout(() => { + setTaskStatus('done') + stopPolling() + }, 30_000) + } } catch (err: unknown) { setTaskStatus('error') setError(err instanceof Error ? err.message : '采集触发失败') stopPolling() + stopJobPolling() } finally { setLoading(false) } @@ -278,7 +338,11 @@ export default function CollectionPage() {
- 任务执行中,已运行 {elapsed}s… + + {job + ? `${JOB_STATUS_LABEL[job.status]}(轮询中),已运行 ${elapsed}s…` + : `任务执行中,已运行 ${elapsed}s…`} +

后台异步执行,关闭页面不影响结果。可稍后查看「系统日志」确认完成。 @@ -289,15 +353,37 @@ export default function CollectionPage() {

- 任务已提交,后台执行中(可能尚未完成) + + {job + ? `采集完成(${JOB_STATUS_LABEL[job.status]})` + : '任务已提交,后台执行中(可能尚未完成)'} +
-

- 采集耗时取决于数据量。请到「系统日志」页查看最终结果。 -

+ {job?.result && summarizeResult(job.result).length > 0 && ( +
+ {summarizeResult(job.result).map((line, i) => ( +

{line}

+ ))} +
+ )} + {!job && ( +

+ 采集耗时取决于数据量。请到「系统日志」页查看最终结果。 +

+ )}
)} {taskStatus === 'error' && ( -

任务触发失败,请检查配置或网络。

+
+

+ {job?.status === 'failed' ? '采集任务失败' : '任务触发失败,请检查配置或网络。'} +

+ {job?.error && ( +

+ {job.error} +

+ )} +
)} {ingestSnap?.last_success_at && (
diff --git a/frontend/src/admin/types.ts b/frontend/src/admin/types.ts index 89a5a88..fcbaeb9 100644 --- a/frontend/src/admin/types.ts +++ b/frontend/src/admin/types.ts @@ -103,6 +103,19 @@ export interface CollectionRequest { date_to?: string } +// 采集任务状态(ingest_jobs,提交采集后轮询) +export interface IngestJob { + id: string + task: 'events' | 'standings' | 'stats' | 'all' + params: Record + status: 'pending' | 'running' | 'success' | 'failed' + result: Record | null + error: string | null + created_at: string | null + started_at: string | null + finished_at: string | null +} + // ── 评估 & 回测 ───────────────────────────────────────────────── export interface EvalCalibrationBucket { diff --git a/src/api/routes/ingest.py b/src/api/routes/ingest.py index 0799979..81fb3c4 100644 --- a/src/api/routes/ingest.py +++ b/src/api/routes/ingest.py @@ -10,14 +10,17 @@ from __future__ import annotations import asyncio import logging +from datetime import datetime, timezone from fastapi import APIRouter, Depends, HTTPException +from sqlalchemy import select from src.api.deps import require_admin from src.api.schemas import IngestBzzoiroRequest from src.data.config import BZZOIRO_LEAGUE_IDS from src.data.bzzoiro import ingest_bzzoiro_event_stats, ingest_bzzoiro_standings from src.data.sources import get_source +from src.db.models import IngestJob from src.db.unit_of_work import get_uow logger = logging.getLogger(__name__) @@ -30,6 +33,29 @@ _background_tasks: set[asyncio.Task] = set() VALID_TASKS = {"events", "standings", "stats", "all"} +def _utcnow() -> datetime: + return datetime.now(timezone.utc) + + +async def _update_job(job_id: str, **fields) -> None: + """更新采集任务状态(尽力而为:失败只记日志,绝不拖垮采集主流程)。 + + job 记录是任务级可观测性,IngestFailure 死信是行级失败记录, + 两者独立工作 —— 本函数抛错不应影响采集结果,故整体吞异常。 + """ + try: + async with get_uow() as session: + stmt = select(IngestJob).where(IngestJob.id == job_id) + job = (await session.execute(stmt)).scalar_one_or_none() + if job is None: + logger.warning("ingest job %s 不存在,跳过状态更新(%s)", job_id, fields) + return + for k, v in fields.items(): + setattr(job, k, v) + except Exception: + logger.warning("ingest job %s 状态更新失败(不影响采集): %s", job_id, fields, exc_info=True) + + def _spawn(coro) -> None: """启动后台采集任务;异常已在任务内记录到系统日志。""" task = asyncio.create_task(coro) @@ -39,20 +65,50 @@ def _spawn(coro) -> None: @router.post("/ingest/bzzoiro", dependencies=[Depends(require_admin)]) async def ingest_bzzoiro_route(req: IngestBzzoiroRequest): - """触发 bzzoiro 采集(events / standings / stats / all)。""" + """触发 bzzoiro 采集(events / standings / stats / all)。 + + 启动后台任务前先创建 ingest_jobs 记录(pending),响应返回 job_id, + 供 admin 通过 GET /api/v1/admin/ingest/jobs/{job_id} 轮询状态。 + """ if req.task not in VALID_TASKS: raise HTTPException(status_code=422, detail=f"未知任务类型: {req.task}(可选: {', '.join(sorted(VALID_TASKS))})") leagues = req.leagues or list(BZZOIRO_LEAGUE_IDS.keys()) task_label = {"events": "比赛数据", "standings": "积分榜", "stats": "统计回填", "all": "全量(比赛+积分榜+统计)"}[req.task] - _spawn(_run_bzzoiro(req.task, leagues, req)) + + # 任务级可观测性:先落 pending 记录,后台 _run_bzzoiro 接管状态流转 + job = IngestJob( + task=req.task, + params={ + "leagues": leagues, + "date_from": req.date_from, + "date_to": req.date_to, + "status": req.status, + "limit": req.limit, + "season": req.season, + }, + status="pending", + ) + async with get_uow() as session: + session.add(job) + + _spawn(_run_bzzoiro(job.id, req.task, leagues, req)) return { "ok": True, - "message": f"采集任务已启动(后台执行,任务: {task_label}),请在「系统日志」查看进度与结果", + "job_id": job.id, + "message": f"采集任务已启动(后台执行,任务: {task_label}),可用 job_id 查询状态或在「系统日志」查看进度", } -async def _run_bzzoiro(task: str, leagues: list[str], req: IngestBzzoiroRequest) -> None: - """后台执行 bzzoiro 采集:上游限速时单次可能耗时数分钟,必须脱离请求生命周期。""" +async def _run_bzzoiro(job_id: str, task: str, leagues: list[str], req: IngestBzzoiroRequest) -> None: + """后台执行 bzzoiro 采集:上游限速时单次可能耗时数分钟,必须脱离请求生命周期。 + + 同时维护 ingest_jobs 状态(pending → running → success/failed): + - result 存各子任务统计摘要(errors 截断到前 10 条) + - 状态更新经 _update_job 尽力而为,失败不影响采集本身 + - 单条管线抓取失败的行级记录仍由 bzzoiro 死信逻辑(IngestFailure)负责 + """ + await _update_job(job_id, status="running", started_at=_utcnow()) + summary: dict = {} try: if task in ("events", "all"): statuses = [req.status] if req.status else ["finished", "scheduled"] @@ -79,6 +135,7 @@ async def _run_bzzoiro(task: str, leagues: list[str], req: IngestBzzoiroRequest) ) if merged["errors"]: logger.warning("bzzoiro 比赛采集错误 %d 条: %s", len(merged["errors"]), merged["errors"][:3]) + summary["events"] = {**merged, "errors": merged["errors"][:10]} if task in ("standings", "all"): async with get_uow() as session: @@ -87,6 +144,9 @@ async def _run_bzzoiro(task: str, leagues: list[str], req: IngestBzzoiroRequest) logger.warning("bzzoiro 积分榜采集部分失败: %s", r["errors"][:3]) else: logger.info("bzzoiro 积分榜采集完成: upsert %d 条", r["total_upserted"]) + summary["standings"] = {k: r.get(k) for k in ("total_upserted", "errors") if k in r} + if isinstance(summary["standings"].get("errors"), list): + summary["standings"]["errors"] = summary["standings"]["errors"][:10] if task in ("stats", "all"): async with get_uow() as session: @@ -95,5 +155,12 @@ async def _run_bzzoiro(task: str, leagues: list[str], req: IngestBzzoiroRequest) ) if r["errors"]: logger.warning("bzzoiro 统计回填错误 %d 条: %s", len(r["errors"]), r["errors"][:3]) - except Exception: + summary["stats"] = {k: r.get(k) for k in ("total_inserted", "total_updated", "errors") if k in r} + if isinstance(summary["stats"].get("errors"), list): + summary["stats"]["errors"] = summary["stats"]["errors"][:10] + except Exception as exc: logger.exception("bzzoiro 采集任务失败(task=%s)", task) + await _update_job(job_id, status="failed", finished_at=_utcnow(), error=repr(exc)) + return + + await _update_job(job_id, status="success", finished_at=_utcnow(), result=summary) diff --git a/src/api/routes/schedules.py b/src/api/routes/schedules.py index 75f1992..d1b42e5 100644 --- a/src/api/routes/schedules.py +++ b/src/api/routes/schedules.py @@ -182,3 +182,49 @@ async def retry_ingest_failure(failure_id: int, db: AsyncSession = Depends(get_d # 触发重试(简化版:仅标记状态,实际重试逻辑由调度器处理) return {"ok": True, "message": f"已标记重试 (第 {failure.retry_count} 次)"} + + +# ── 采集任务状态(ingest_jobs) ────────────────────────────────── + + +def _serialize_job(j) -> dict: + """ingest_jobs 行 → 响应 dict(时间统一 isoformat)。""" + return { + "id": j.id, + "task": j.task, + "params": j.params, + "status": j.status, + "result": j.result, + "error": j.error, + "created_at": j.created_at.isoformat() if j.created_at else None, + "started_at": j.started_at.isoformat() if j.started_at else None, + "finished_at": j.finished_at.isoformat() if j.finished_at else None, + } + + +@router.get("/ingest/jobs") +async def list_ingest_jobs( + limit: int = 20, + status: str | None = None, + db: AsyncSession = Depends(get_db_read), +): + """列出采集任务状态(最新在前),可按 status 过滤。""" + from src.db.models import IngestJob + + stmt = select(IngestJob).order_by(IngestJob.created_at.desc()).limit(max(1, min(limit, 100))) + if status: + stmt = stmt.where(IngestJob.status == status) + rows = (await db.execute(stmt)).scalars().all() + return [_serialize_job(j) for j in rows] + + +@router.get("/ingest/jobs/{job_id}") +async def get_ingest_job(job_id: str, db: AsyncSession = Depends(get_db_read)): + """查询单次采集任务的状态与结果(供前端提交后轮询)。""" + from src.db.models import IngestJob + + stmt = select(IngestJob).where(IngestJob.id == job_id) + job = (await db.execute(stmt)).scalar_one_or_none() + if job is None: + raise HTTPException(404, "采集任务不存在") + return _serialize_job(job) diff --git a/src/db/models.py b/src/db/models.py index 351d09b..95b9b66 100644 --- a/src/db/models.py +++ b/src/db/models.py @@ -1,9 +1,12 @@ -"""ORM 模型: leagues / teams / matches / match_stats / standings / predictions。 +"""ORM 模型(13 张表)。 -数据源统一为 bzzoiro(单一数据源),伤停(injuries)与 Understat 已移除。 +业务表: leagues / teams / matches / match_stats / standings / predictions +治理与配置表: raw_events / ingest_failures / data_quality_checks / data_lineage / +app_settings / schedules / ingest_jobs(完整说明见 docs/05-data.md)。 """ from __future__ import annotations +import uuid from datetime import date, datetime, timezone from sqlalchemy import ( @@ -404,3 +407,33 @@ class DataLineage(Base): Index("ix_lineage_target", "target_table", "target_id"), Index("ix_lineage_batch", "batch_id"), ) + + +class IngestJob(Base): + """采集任务状态跟踪:解决 POST /ingest/bzzoiro fire-and-forget 不可观测问题。 + + 生命周期: pending(路由创建) → running(后台开始) → success / failed(终态)。 + result 存各任务统计摘要(如 inserted/updated/errors 截断);error 存失败原因。 + 与 IngestFailure 死信相互独立:死信记录单条管线抓取失败(行级), + job 记录整次任务执行结果(任务级),两者可同时存在。 + job 状态更新是「尽力而为」:更新失败只记日志,不影响采集主流程。 + """ + __tablename__ = "ingest_jobs" + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4())) + task: Mapped[str] = mapped_column(String(20), nullable=False) # events / standings / stats / all + params: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict) # leagues/日期等请求参数 + status: Mapped[str] = mapped_column(String(20), nullable=False, default="pending") + result: Mapped[dict | None] = mapped_column(JSONB) # 成功时的统计摘要 + error: Mapped[str | None] = mapped_column(Text) # 失败原因(repr) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow) + started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + + __table_args__ = ( + CheckConstraint( + "status IN ('pending','running','success','failed')", + name="ck_ingest_jobs_status", + ), + Index("ix_ingest_jobs_created", "created_at"), + ) diff --git a/tests/test_ingest_jobs.py b/tests/test_ingest_jobs.py new file mode 100644 index 0000000..c26804d --- /dev/null +++ b/tests/test_ingest_jobs.py @@ -0,0 +1,313 @@ +"""采集任务状态(ingest_jobs)测试。 + +背景: POST /ingest/bzzoiro 此前 fire-and-forget —— 触发后只能翻系统日志, +无法程序化查询「这次采集跑到哪了/成没成」。本次改造: + 1. 路由创建 pending job → 响应返回 job_id + 2. 后台 _run_bzzoiro 维护 running → success / failed + result/error + 3. admin 端点 /admin/ingest/jobs/{job_id} 与列表可查询 + +守护点: + - job 更新是「尽力而为」: _update_job 自身失败被吞掉,不影响采集主流程 + - job(任务级)与 IngestFailure 死信(行级)相互独立,可同时存在 + (死信路径由 test_ingest_deadletter.py 守护,本文件不动 bzzoiro 内部) + +范式: 假 UoW(记录 add / 返回预设 job)+ monkeypatch source,不依赖真实数据库 +(与 test_ingest_deadletter.py / test_public_readonly_api.py 相同)。 +""" +from __future__ import annotations + +from contextlib import asynccontextmanager +from datetime import datetime, timezone + +from fastapi import FastAPI +from fastapi.testclient import TestClient + +import src.api.routes.ingest as ingest +import src.api.routes.schedules as schedules +from src.api.deps import require_admin +from src.api.schemas import IngestBzzoiroRequest +from src.db.base import get_db_read +from src.db.models import IngestJob + + +# ── 假基础设施 ────────────────────────────────────────────────── + + +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 self._items[0] if self._items else None + + def scalar(self): + return None + + +class _FakeSession: + """记录 add();execute 按预设队列依次返回(与真实 UoW 的单会话用法对齐)。""" + + def __init__(self, results=None): + self.added = [] + self._results = list(results or []) + + def add(self, obj): + self.added.append(obj) + + async def execute(self, stmt): + if self._results: + return self._results.pop(0) + return _FakeResult([]) + + +def _patch_uow(monkeypatch, session) -> None: + """把 ingest 模块的 get_uow 指向假会话。""" + + @asynccontextmanager + async def _fake(): + yield session + + monkeypatch.setattr(ingest, "get_uow", _fake) + + +def _ingest_app() -> TestClient: + app = FastAPI() + app.include_router(ingest.router) + app.dependency_overrides[require_admin] = lambda: None + return TestClient(app) + + +def _admin_app(fake_db) -> TestClient: + app = FastAPI() + app.include_router(schedules.router) + app.dependency_overrides[require_admin] = lambda: None + app.dependency_overrides[get_db_read] = lambda: fake_db + return TestClient(app) + + +def _job(jid: str = "job-1", status: str = "success", **kw) -> IngestJob: + return IngestJob( + id=jid, + task=kw.pop("task", "standings"), + params=kw.pop("params", {"leagues": ["E0"]}), + status=status, + result=kw.pop("result", {"total_upserted": 20}), + error=kw.pop("error", None), + created_at=kw.pop("created_at", datetime.now(timezone.utc)), + **kw, + ) + + +# ── 1. 路由:创建 job + 返回 job_id ───────────────────────────── + + +class TestRouteCreatesJob: + def test_post_returns_job_id_and_persists_pending_job(self, monkeypatch): + session = _FakeSession() + _patch_uow(monkeypatch, session) + + spawned: list = [] + monkeypatch.setattr(ingest, "_spawn", lambda coro: spawned.append(coro)) + + client = _ingest_app() + resp = client.post( + "/api/v1/ingest/bzzoiro", + json={"task": "standings", "leagues": ["E0", "SP1"], "season": "2026-2027"}, + ) + assert resp.status_code == 200 + body = resp.json() + assert body["ok"] is True + + jobs = [o for o in session.added if isinstance(o, IngestJob)] + assert len(jobs) == 1 + job = jobs[0] + assert body["job_id"] == job.id + assert job.status == "pending" + assert job.task == "standings" + # params 记录的是「实际将执行」的参数(含默认联赛展开) + assert job.params["leagues"] == ["E0", "SP1"] + assert job.params["season"] == "2026-2027" + + # 后台协程被捕获但未执行;显式关闭避免 un-awaited 告警 + assert len(spawned) == 1 + spawned[0].close() + + def test_post_invalid_task_422_and_no_job(self, monkeypatch): + session = _FakeSession() + _patch_uow(monkeypatch, session) + monkeypatch.setattr(ingest, "_spawn", lambda coro: coro.close()) + + client = _ingest_app() + resp = client.post("/api/v1/ingest/bzzoiro", json={"task": "bogus"}) + assert resp.status_code == 422 + assert not [o for o in session.added if isinstance(o, IngestJob)] + + +# ── 2. 后台执行:状态流转 ──────────────────────────────────────── + + +class TestRunBzzoiroJobLifecycle: + async def test_success_flow_updates_job_running_then_success(self, monkeypatch): + job = _job(jid="job-ok", status="pending", task="events") + session = _FakeSession(results=[_FakeResult([job])] * 10) + _patch_uow(monkeypatch, session) + + class _FakeSource: + async def ingest(self, session, **kw): + return {"inserted": 3, "updated": 1, "total_inserted": 3, + "total_updated": 1, "errors": ["e1", "e2"]} + + monkeypatch.setattr(ingest, "get_source", lambda name: _FakeSource()) + + req = IngestBzzoiroRequest(task="events", leagues=["E0"], status="finished") + await ingest._run_bzzoiro("job-ok", "events", ["E0"], req) + + assert job.status == "success" + assert job.started_at is not None + assert job.finished_at is not None + assert job.error is None + # result 含 events 子任务摘要,errors 截断到前 10 条 + assert job.result["events"]["total_inserted"] == 3 + assert job.result["events"]["errors"] == ["e1", "e2"] + + async def test_failure_flow_marks_failed_with_error(self, monkeypatch): + job = _job(jid="job-bad", status="running", task="standings", result=None) + session = _FakeSession(results=[_FakeResult([job])] * 10) + _patch_uow(monkeypatch, session) + + async def _boom(*args, **kwargs): + raise RuntimeError("network down") + + monkeypatch.setattr(ingest, "ingest_bzzoiro_standings", _boom) + + req = IngestBzzoiroRequest(task="standings", leagues=["E0"]) + await ingest._run_bzzoiro("job-bad", "standings", ["E0"], req) + + assert job.status == "failed" + assert "network down" in job.error + assert job.finished_at is not None + assert job.result is None + + async def test_all_task_collects_per_task_summaries(self, monkeypatch): + job = _job(jid="job-all", status="pending", task="all") + session = _FakeSession(results=[_FakeResult([job])] * 20) + _patch_uow(monkeypatch, session) + + class _FakeSource: + async def ingest(self, session, **kw): + return {"inserted": 1, "updated": 0, "total_inserted": 1, + "total_updated": 0, "errors": []} + + async def _standings(*args, **kwargs): + return {"total_upserted": 20, "errors": []} + + async def _stats(*args, **kwargs): + return {"total_inserted": 5, "total_updated": 2, "errors": []} + + monkeypatch.setattr(ingest, "get_source", lambda name: _FakeSource()) + monkeypatch.setattr(ingest, "ingest_bzzoiro_standings", _standings) + monkeypatch.setattr(ingest, "ingest_bzzoiro_event_stats", _stats) + + req = IngestBzzoiroRequest(task="all", leagues=["E0"]) + await ingest._run_bzzoiro("job-all", "all", ["E0"], req) + + assert job.status == "success" + assert set(job.result.keys()) == {"events", "standings", "stats"} + assert job.result["standings"]["total_upserted"] == 20 + assert job.result["stats"]["total_inserted"] == 5 + + async def test_update_job_failure_does_not_break_ingest(self, monkeypatch): + """_update_job 抛错必须被吞掉:job 可观测性失败 ≠ 采集失败。""" + session = _FakeSession() + _patch_uow(monkeypatch, session) + + class _FakeSource: + async def ingest(self, session, **kw): + return {"inserted": 1, "updated": 0, "total_inserted": 1, + "total_updated": 0, "errors": []} + + monkeypatch.setattr(ingest, "get_source", lambda name: _FakeSource()) + + # 让 execute 抛错(_update_job 内部会捕获) + async def _broken_execute(stmt): + raise RuntimeError("db gone") + + session.execute = _broken_execute + + req = IngestBzzoiroRequest(task="events", leagues=["E0"], status="finished") + # 不抛异常即通过;采集逻辑本身照常跑完 + await ingest._run_bzzoiro("job-x", "events", ["E0"], req) + + async def test_job_and_deadletter_are_independent_layers(self, monkeypatch): + """任务级(job)与行级(死信)互不干扰:source 内部返回 errors 时, + job 仍为 success(部分失败不算任务失败),死信由 bzzoiro 层另行记录。""" + job = _job(jid="job-part", status="pending", task="events") + session = _FakeSession(results=[_FakeResult([job])] * 10) + _patch_uow(monkeypatch, session) + + class _FakeSource: + async def ingest(self, session, **kw): + # 模拟 bzzoiro 管线:单条失败已写死信,汇总 errors 非空但返回正常 + return {"inserted": 9, "updated": 0, "total_inserted": 9, + "total_updated": 0, "errors": ["league F1 fetch failed"]} + + monkeypatch.setattr(ingest, "get_source", lambda name: _FakeSource()) + + req = IngestBzzoiroRequest(task="events", leagues=["E0"], status="finished") + await ingest._run_bzzoiro("job-part", "events", ["E0"], req) + + assert job.status == "success" + assert job.result["events"]["errors"] == ["league F1 fetch failed"] + + +# ── 3. admin 查询端点 ────────────────────────────────────────── + + +class TestAdminJobEndpoints: + def test_get_job_detail_200(self, monkeypatch): + job = _job(jid="abc-123", status="running") + client = _admin_app(_FakeSession(results=[_FakeResult([job])])) + + resp = client.get("/api/v1/admin/ingest/jobs/abc-123") + assert resp.status_code == 200 + body = resp.json() + assert body["id"] == "abc-123" + assert body["status"] == "running" + assert body["task"] == "standings" + assert body["params"] == {"leagues": ["E0"]} + assert body["result"] == {"total_upserted": 20} + assert body["error"] is None + assert body["created_at"] is not None + + def test_get_job_detail_404(self): + client = _admin_app(_FakeSession(results=[_FakeResult([])])) + resp = client.get("/api/v1/admin/ingest/jobs/missing") + assert resp.status_code == 404 + + def test_list_jobs_returns_serialized_rows(self): + j1 = _job(jid="j1", status="success") + j2 = _job(jid="j2", status="failed", task="events", + result=None, error="RuntimeError('x')") + client = _admin_app(_FakeSession(results=[_FakeResult([j1, j2])])) + + resp = client.get("/api/v1/admin/ingest/jobs?limit=10") + assert resp.status_code == 200 + rows = resp.json() + assert [r["id"] for r in rows] == ["j1", "j2"] + assert rows[1]["status"] == "failed" + assert rows[1]["error"] == "RuntimeError('x')" + + def test_job_routes_live_under_admin_router_with_require_admin(self): + """结构守护:job 端点必须挂在 /api/v1/admin 路由(路由级 require_admin)。""" + assert any(dep.dependency is require_admin for dep in schedules.router.dependencies) + paths = {getattr(r, "path", "") for r in schedules.router.routes} + assert "/api/v1/admin/ingest/jobs" in paths + assert "/api/v1/admin/ingest/jobs/{job_id}" in paths