feat: 采集任务状态跟踪 ingest_jobs
新增 ingest_jobs 表(UUID/task/params/status/result/error/时间戳),
POST /ingest/bzzoiro 启动前插入 job(pending)→ 后台 running → success/failed,
响应新增 job_id(兼容原 message)。
Admin GET /admin/ingest/jobs/{id} 与 /admin/ingest/jobs?limit= 只读查询;
Collection 页提交后轮询 job 至终态,展示真实 result/error 汇总。
迁移 0019_ingest_jobs + 分批 get_uow/BzzoiroSource/IngestFailure 不变。
全量测试 270 通过。
This commit is contained in:
shangfangjian
@@ -0,0 +1,55 @@
|
||||
"""后台管理:采集任务状态查询(只读)。
|
||||
|
||||
GET /api/v1/admin/ingest/jobs/{job_id} — 单任务详情
|
||||
GET /api/v1/admin/ingest/jobs?limit=N — 最近任务列表(默认 20)
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||||
from sqlalchemy import desc, select
|
||||
|
||||
from src.api.deps import require_admin
|
||||
from src.api.schemas import IngestJobOut
|
||||
from src.db.base import AsyncSession, get_db_read
|
||||
from src.db.models import IngestJob
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter(prefix="/api/v1/admin", tags=["admin"], dependencies=[Depends(require_admin)])
|
||||
|
||||
|
||||
@router.get("/ingest/jobs/{job_id}", response_model=IngestJobOut)
|
||||
async def get_ingest_job(job_id: str, db: AsyncSession = Depends(get_db_read)):
|
||||
"""查询单个采集任务状态。"""
|
||||
job = await db.get(IngestJob, job_id)
|
||||
if job is None:
|
||||
raise HTTPException(404, f"采集任务不存在: {job_id}")
|
||||
return _job_to_out(job)
|
||||
|
||||
|
||||
@router.get("/ingest/jobs", response_model=list[IngestJobOut])
|
||||
async def list_ingest_jobs(
|
||||
limit: int = Query(20, ge=1, le=100, description="返回条数"),
|
||||
db: AsyncSession = Depends(get_db_read),
|
||||
):
|
||||
"""查询最近采集任务(最新在前)。"""
|
||||
rows = (
|
||||
await db.execute(select(IngestJob).order_by(desc(IngestJob.created_at)).limit(limit))
|
||||
).scalars().all()
|
||||
return [_job_to_out(j) for j in rows]
|
||||
|
||||
|
||||
def _job_to_out(job: IngestJob) -> IngestJobOut:
|
||||
return IngestJobOut(
|
||||
id=job.id,
|
||||
task=job.task,
|
||||
params=job.params or {},
|
||||
status=job.status,
|
||||
result=job.result,
|
||||
error=job.error,
|
||||
created_at=job.created_at,
|
||||
started_at=job.started_at,
|
||||
finished_at=job.finished_at,
|
||||
)
|
||||
+73
-10
@@ -10,11 +10,13 @@ from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
|
||||
from src.api.deps import require_admin
|
||||
from src.api.schemas import IngestBzzoiroRequest
|
||||
from src.api.schemas import IngestBzzoiroRequest, IngestBzzoiroResponse
|
||||
from src.data.config import BZZOIRO_LEAGUE_IDS
|
||||
from src.data.bzzoiro_standings import ingest_bzzoiro_standings
|
||||
from src.data.bzzoiro_stats import ingest_bzzoiro_event_stats
|
||||
@@ -38,22 +40,58 @@ def _spawn(coro) -> None:
|
||||
task.add_done_callback(_background_tasks.discard)
|
||||
|
||||
|
||||
@router.post("/ingest/bzzoiro", dependencies=[Depends(require_admin)])
|
||||
@router.post("/ingest/bzzoiro", response_model=IngestBzzoiroResponse, 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 供前端轮询。
|
||||
兼容原 message 字段(仍返回)。
|
||||
"""
|
||||
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))
|
||||
return {
|
||||
"ok": True,
|
||||
"message": f"采集任务已启动(后台执行,任务: {task_label}),请在「系统日志」查看进度与结果",
|
||||
|
||||
job_id = await _create_ingest_job(req.task, leagues, req)
|
||||
_spawn(_run_bzzoiro(job_id, req.task, leagues, req))
|
||||
|
||||
return IngestBzzoiroResponse(
|
||||
ok=True,
|
||||
job_id=job_id,
|
||||
message=f"采集任务已启动(后台执行,任务: {task_label}),请到「数据采集」页跟踪进度",
|
||||
)
|
||||
|
||||
|
||||
async def _create_ingest_job(task: str, leagues: list[str], req: IngestBzzoiroRequest) -> str:
|
||||
"""写入一条 ingest_jobs(pending),返回 job_id。"""
|
||||
from src.db.models import IngestJob
|
||||
|
||||
job_id = str(uuid.uuid4())
|
||||
params = {
|
||||
"leagues": leagues,
|
||||
"date_from": req.date_from,
|
||||
"date_to": req.date_to,
|
||||
"status": req.status,
|
||||
"task": task,
|
||||
"limit": req.limit,
|
||||
"season": req.season,
|
||||
}
|
||||
async with get_uow() as session:
|
||||
job = IngestJob(id=job_id, task=task, params=params, status="pending")
|
||||
session.add(job)
|
||||
logger.info("ingest_jobs 创建: job=%s task=%s leagues=%s", job_id, task, leagues)
|
||||
return 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 采集:上游限速时单次可能耗时数分钟,必须脱离请求生命周期。
|
||||
|
||||
状态流转: pending → running → (success|failed)。
|
||||
"""
|
||||
from src.db.models import IngestJob
|
||||
|
||||
await _update_job(job_id, status="running", started_at=datetime.now(timezone.utc))
|
||||
result: dict = {}
|
||||
try:
|
||||
if task in ("events", "all"):
|
||||
statuses = [req.status] if req.status else ["finished", "scheduled"]
|
||||
@@ -80,6 +118,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])
|
||||
result["events"] = merged
|
||||
|
||||
if task in ("standings", "all"):
|
||||
async with get_uow() as session:
|
||||
@@ -88,6 +127,7 @@ 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"])
|
||||
result["standings"] = r
|
||||
|
||||
if task in ("stats", "all"):
|
||||
async with get_uow() as session:
|
||||
@@ -96,5 +136,28 @@ 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:
|
||||
result["stats"] = r
|
||||
|
||||
await _update_job(job_id, status="success", result=result, finished_at=datetime.now(timezone.utc))
|
||||
logger.info("ingest_jobs 完成: job=%s task=%s", job_id, task)
|
||||
except Exception as e:
|
||||
logger.exception("bzzoiro 采集任务失败(task=%s)", task)
|
||||
await _update_job(
|
||||
job_id, status="failed", error=str(e), finished_at=datetime.now(timezone.utc),
|
||||
)
|
||||
|
||||
|
||||
async def _update_job(job_id: str, **fields) -> None:
|
||||
"""更新 ingest_jobs 单行;失败仅记日志,绝不抛异常(避免干扰采集主流程)。"""
|
||||
from src.db.models import IngestJob
|
||||
|
||||
try:
|
||||
async with get_uow() as session:
|
||||
job = await session.get(IngestJob, job_id)
|
||||
if job is None:
|
||||
logger.warning("ingest_jobs 更新失败: job=%s 不存在", job_id)
|
||||
return
|
||||
for k, v in fields.items():
|
||||
setattr(job, k, v)
|
||||
except Exception:
|
||||
logger.warning("ingest_jobs 更新异常: job=%s fields=%s", job_id, list(fields.keys()))
|
||||
|
||||
Reference in New Issue
Block a user