diff --git a/alembic/versions/0019_ingest_jobs.py b/alembic/versions/0019_ingest_jobs.py new file mode 100644 index 0000000..210e170 --- /dev/null +++ b/alembic/versions/0019_ingest_jobs.py @@ -0,0 +1,42 @@ +"""采集任务状态表 ingest_jobs + +Revision ID: 0019_ingest_jobs +Revises: 0018_match_checks +Create Date: 2026-09-22 +""" + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + +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', sa.JSON(), nullable=False, server_default='{}'), + sa.Column('status', sa.String(20), nullable=False, server_default='pending'), + sa.Column('result', sa.JSON(), nullable=True), + sa.Column('error', sa.Text(), nullable=True), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.Column('started_at', sa.DateTime(timezone=True), nullable=True), + sa.Column('finished_at', sa.DateTime(timezone=True), nullable=True), + ) + op.create_index('ix_ingest_job_status_created', 'ingest_jobs', ['status', 'created_at']) + op.create_check_constraint( + 'ck_ingest_job_status', 'ingest_jobs', + "status IN ('pending', 'running', 'success', 'failed')", + ) + + +def downgrade() -> None: + op.drop_constraint('ck_ingest_job_status', 'ingest_jobs', type_='check') + op.drop_index('ix_ingest_job_status_created', table_name='ingest_jobs') + op.drop_table('ingest_jobs') diff --git a/frontend/src/admin/dal.ts b/frontend/src/admin/dal.ts index e81075f..4aa998a 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, @@ -362,6 +363,20 @@ export function fetchIngestStatus(): Promise<{ sources: IngestSourceStatus[] }> return api.get<{ sources: IngestSourceStatus[] }>(`${API_BASE}/admin/ingest/status`) } +/** + * 采集任务状态轮询(单任务) + */ +export function fetchIngestJob(jobId: string): Promise { + return api.get(`${API_BASE}/admin/ingest/jobs/${jobId}`) +} + +/** + * 最近采集任务列表(最新在前) + */ +export function fetchIngestJobs(limit = 20): Promise { + return api.get(`${API_BASE}/admin/ingest/jobs?limit=${limit}`) +} + /** * 比赛详情(含最近预测摘要) */ diff --git a/frontend/src/admin/pages/Collection.tsx b/frontend/src/admin/pages/Collection.tsx index 4591947..6eafd29 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 type { IngestSourceStatus } from '../types' -import type { CollectionRequest, League } from '../types' +import { triggerCollection, fetchLeagues, fetchIngestJob } from '../dal' +import type { IngestJob, League } from '../types' +import type { CollectionRequest } from '../types' import { Card, CardBody, CardHeader, Badge, SectionHeader, Alert, Spinner } from '../components' const TASKS = [ @@ -23,7 +23,9 @@ const TASKS = [ { value: 'all', label: '全量采集', desc: '依次采集比赛 + 积分榜 + 统计回填', icon: '⏵⏵' }, ] as const -type TaskStatus = 'idle' | 'running' | 'done' | 'error' +type TaskUIStatus = 'idle' | 'running' | 'done' | 'error' + +const TERMINAL_STATUSES: ReadonlySet = new Set(['success', 'failed']) export default function CollectionPage() { const [leagues, setLeagues] = useState([]) @@ -49,13 +51,13 @@ export default function CollectionPage() { const [limit, setLimit] = useState(100) const [loading, setLoading] = useState(false) const [error, setError] = useState(null) - const [result, setResult] = useState<{ title: string; detail: string } | null>(null) - // 任务进度反馈 - const [taskStatus, setTaskStatus] = useState('idle') + // 任务进度反馈:跟踪真实 ingest_job 状态 + const [taskStatus, setTaskStatus] = useState('idle') + const [jobId, setJobId] = useState(null) + const [jobInfo, setJobInfo] = useState(null) const [taskStartedAt, setTaskStartedAt] = useState(null) const pollRef = useRef | null>(null) - const [ingestSnap, setIngestSnap] = useState(null) const loadLeagues = useCallback(async () => { const lg = await fetchLeagues() @@ -64,30 +66,65 @@ export default function CollectionPage() { useEffect(() => { loadLeagues() }, [loadLeagues]) - // 轮询采集状态(任务启动后) - const startPolling = useCallback(() => { - if (pollRef.current) clearInterval(pollRef.current) - pollRef.current = setInterval(async () => { - try { - const { sources } = await fetchIngestStatus() - const bz = sources.find(s => s.name === 'bzzoiro') - if (bz) setIngestSnap(bz) - } catch { /* ignore */ } - }, 5_000) - }, []) - + // 轮询采集 job 直到终态(success/failed) const stopPolling = useCallback(() => { if (pollRef.current) { clearInterval(pollRef.current); pollRef.current = null } }, []) useEffect(() => () => stopPolling(), [stopPolling]) + const startJobPolling = useCallback((id: string) => { + stopPolling() + const tick = async () => { + try { + const job = await fetchIngestJob(id) + setJobInfo(job) + if (TERMINAL_STATUSES.has(job.status)) { + setTaskStatus(job.status === 'success' ? 'done' : 'error') + stopPolling() + } + } catch { /* 单次轮询失败不影响后续 */ } + } + tick() + pollRef.current = setInterval(tick, 3_000) + }, [stopPolling]) + const isEventsTask = task === 'events' || task === 'all' + // 友好汇总 job.result + const jobSummary = (j: IngestJob | null): { title: string; detail: string } | null => { + if (!j) return null + if (j.status === 'failed') { + return { title: '采集失败', detail: j.error || '采集任务异常终止,请到「系统日志」查看详细堆栈。' } + } + if (j.status !== 'success') return null + const r = j.result as Record | null + if (!r) return { title: '采集完成', detail: '任务成功(无汇总数据)。' } + const ev = r.events as Record | undefined + const evTotal = ev ? (ev.total_inserted as number ?? 0) + (ev.total_updated as number ?? 0) : 0 + const st = r.standings as Record | undefined + const stTotal = st ? (st.total_upserted as number ?? 0) : 0 + const stats = r.stats as Record | undefined + const statsTotal = stats ? (stats.created as number ?? 0) + (stats.updated as number ?? 0) : 0 + const evErr = (ev?.errors as string[] | undefined)?.length ?? 0 + const stErr = (st?.errors as string[] | undefined)?.length ?? 0 + const statsErr = (stats?.errors as string[] | undefined)?.length ?? 0 + const totalErr = evErr + stErr + statsErr + const parts: string[] = [] + if (ev) parts.push(`比赛 +${evTotal}`) + if (st) parts.push(`积分榜 +${stTotal}`) + if (stats) parts.push(`统计 +${statsTotal}`) + const detail = parts.length + ? `共更新: ${parts.join(' / ')}${totalErr ? `,错误 ${totalErr} 条(见日志)` : ''}` + : '任务成功' + return { title: '采集完成', detail } + } + async function handleSubmit(e: React.FormEvent) { e.preventDefault() setError(null) - setResult(null) + setJobInfo(null) + setJobId(null) setLoading(true) setTaskStatus('running') setTaskStartedAt(Date.now()) @@ -103,18 +140,15 @@ 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 res = await triggerCollection(body) + const id: string | undefined = res?.job_id + if (id) { + setJobId(id) + startJobPolling(id) + } else { + // 后端未返回 job_id(旧版兼容):退化为原逻辑 + setTimeout(() => { setTaskStatus('done'); }, 30_000) + } } catch (err: unknown) { setTaskStatus('error') setError(err instanceof Error ? err.message : '采集触发失败') @@ -125,6 +159,7 @@ export default function CollectionPage() { } const elapsed = taskStartedAt ? Math.round((Date.now() - taskStartedAt) / 1000) : 0 + const summary = jobInfo ? jobSummary(jobInfo) : null return (
@@ -248,12 +283,11 @@ export default function CollectionPage() { {/* 消息提示 */} {error && setError(null)} />} - {result && ( + {summary && ( setResult(null)} + kind={jobInfo?.status === 'success' ? 'ok' : 'error'} + title={summary.title} + message={summary.detail} /> )} @@ -278,10 +312,10 @@ export default function CollectionPage() {
- 任务执行中,已运行 {elapsed}s… + 任务执行中{jobId ? `(job ${jobId.slice(0, 8)}…)` : ''},已运行 {elapsed}s…

- 后台异步执行,关闭页面不影响结果。可稍后查看「系统日志」确认完成。 + 后台异步执行,关闭页面不影响结果。每 3 秒自动轮询进度。

)} @@ -289,23 +323,25 @@ export default function CollectionPage() {
- 任务已提交,后台执行中(可能尚未完成) + 采集完成{jobId ? `(job ${jobId.slice(0, 8)}…)` : ''}
-

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

+ {summary &&

{summary.detail}

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

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

- )} - {ingestSnap?.last_success_at && ( -
-

- bzzoiro 最近一次采集: {new Date(ingestSnap.last_success_at).toLocaleString('zh-CN', { hour12: false })} -

+
+

采集失败{jobId ? `(job ${jobId.slice(0, 8)}…)` : ''}

+ {jobInfo?.error && ( +

{jobInfo.error.slice(0, 200)}

+ )}
)} + {jobInfo?.created_at && ( +

+ 创建于 {new Date(jobInfo.created_at).toLocaleString('zh-CN', { hour12: false })} + {jobInfo.finished_at && ` · 完成于 ${new Date(jobInfo.finished_at).toLocaleString('zh-CN', { hour12: false })}`} +

+ )} diff --git a/frontend/src/admin/types.ts b/frontend/src/admin/types.ts index 89a5a88..eda3723 100644 --- a/frontend/src/admin/types.ts +++ b/frontend/src/admin/types.ts @@ -283,6 +283,20 @@ export interface IngestSourceStatus { last_failure: IngestLastFailure | null } +// ── 采集任务状态 ────────────────────────────────────────────── + +export interface IngestJob { + id: string + task: string + 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 MatchRecentPrediction { diff --git a/src/api/routes/admin_ingest_jobs.py b/src/api/routes/admin_ingest_jobs.py new file mode 100644 index 0000000..5366443 --- /dev/null +++ b/src/api/routes/admin_ingest_jobs.py @@ -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, + ) diff --git a/src/api/routes/ingest.py b/src/api/routes/ingest.py index 41ef124..b3efc3f 100644 --- a/src/api/routes/ingest.py +++ b/src/api/routes/ingest.py @@ -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()))