Merge pull request 'feat: 采集任务状态跟踪(ingest_jobs)——fire-and-forget 可观测性' (#12) from ingest-job-status into docs-public-api-fixes

Reviewed-on: #12

README.md

@@ -21,7 +21,7 @@
            │                             │
 ┌──────────▼──────────┐    ┌─────────────▼────────────┐
 │  PostgreSQL         │    │  LLM (OpenAI-compatible)  │
-│  12 张表            │    │  OpenAI / Deepseek /      │
+│  13 张表            │    │  OpenAI / Deepseek /      │
 │  leagues/teams/     │    │  Ollama / 任意网关        │
 │  matches/match_     │    └──────────────────────────┘
 │  stats/standings/   │

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')

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/

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": "<uuid>", "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 唯一来源而移除。
 

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` | 预留未启用 | 数据质量检查结果;规划中定时检查比赛/统计/积分榜完整性 |

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 测试

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<any> {
   return api.post(`${API_BASE}/ingest/bzzoiro`, body)
 }
 
+// 查询采集任务状态(ingest_jobs;提交采集返回 job_id 后轮询用)
+export async function fetchIngestJob(jobId: string): Promise<IngestJob> {
+  return api.get<IngestJob>(`${API_BASE}/admin/ingest/jobs/${jobId}`)
+}
+
 // ── 预测管理 ────────────────────────────────────────────────────
 
 export async function triggerPrediction(req: { match_id: number; mode?: string }): Promise<any> {

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<IngestJob['status'], string> = {
+  pending: '等待中',
+  running: '执行中',
+  success: '已完成',
+  failed: '失败',
+}
+
+/** job.result 摘要 → 多行文本(错误只显示条数,明细见「采集失败」页与系统日志) */
+function summarizeResult(result: Record<string, any> | 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<League[]>([])
   const [task, setTask] = useState<string>('events')
@@ -55,7 +81,10 @@ export default function CollectionPage() {
   const [taskStatus, setTaskStatus] = useState<TaskStatus>('idle')
   const [taskStartedAt, setTaskStartedAt] = useState<number | null>(null)
   const pollRef = useRef<ReturnType<typeof setInterval> | null>(null)
+  const jobPollRef = useRef<ReturnType<typeof setInterval> | null>(null)
   const [ingestSnap, setIngestSnap] = useState<IngestSourceStatus | null>(null)
+  // 采集任务状态(ingest_jobs):提交后有 job_id 即轮询,替代「30 秒盲等」
+  const [job, setJob] = useState<IngestJob | null>(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)
+      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()
-      // 30 秒后自动停止轮询并标记完成
         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() {
                 <div className="space-y-3">
                   <div className="flex items-center gap-2 text-xs text-ink-700">
                     <Spinner />
-                    <span>任务执行中,已运行 {elapsed}s…</span>
+                    <span>
+                      {job
+                        ? `${JOB_STATUS_LABEL[job.status]}(轮询中),已运行 ${elapsed}s…`
+                        : `任务执行中,已运行 ${elapsed}s…`}
+                    </span>
                   </div>
                   <p className="text-2xs text-ink-400">
                     后台异步执行,关闭页面不影响结果。可稍后查看「系统日志」确认完成。
@@ -289,15 +353,37 @@ export default function CollectionPage() {
                 <div className="space-y-2">
                   <div className="flex items-center gap-2 text-xs text-emerald-700">
                     <span className="inline-block h-2 w-2 rounded-full bg-emerald-500" />
-                    <span>任务已提交,后台执行中(可能尚未完成)</span>
+                    <span>
+                      {job
+                        ? `采集完成(${JOB_STATUS_LABEL[job.status]})`
+                        : '任务已提交,后台执行中(可能尚未完成)'}
+                    </span>
                   </div>
+                  {job?.result && summarizeResult(job.result).length > 0 && (
+                    <div className="rounded-md bg-ink-50 px-3 py-2">
+                      {summarizeResult(job.result).map((line, i) => (
+                        <p key={i} className="font-mono text-2xs leading-relaxed text-ink-600">{line}</p>
+                      ))}
+                    </div>
+                  )}
+                  {!job && (
                     <p className="text-2xs text-ink-400">
                       采集耗时取决于数据量。请到「系统日志」页查看最终结果。
                     </p>
+                  )}
                 </div>
               )}
               {taskStatus === 'error' && (
-                <p className="text-xs text-press">任务触发失败,请检查配置或网络。</p>
+                <div className="space-y-2">
+                  <p className="text-xs text-press">
+                    {job?.status === 'failed' ? '采集任务失败' : '任务触发失败,请检查配置或网络。'}
+                  </p>
+                  {job?.error && (
+                    <p className="break-all rounded-md bg-red-50 px-3 py-2 font-mono text-2xs text-press">
+                      {job.error}
+                    </p>
+                  )}
+                </div>
               )}
               {ingestSnap?.last_success_at && (
                 <div className="mt-3 border-t border-ink-100 pt-3">

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<string, unknown>
+  status: 'pending' | 'running' | 'success' | 'failed'
+  result: Record<string, any> | null
+  error: string | null
+  created_at: string | null
+  started_at: string | null
+  finished_at: string | null
+}
+
 // ── 评估 & 回测 ─────────────────────────────────────────────────
 
 export interface EvalCalibrationBucket {

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)

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)

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"),
+    )

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