refactor: 完成了阶段二的修复

src/gateway.py

@@ -6,16 +6,24 @@ from __future__ import annotations
 import asyncio
 import contextlib
 import html
+import json
 import logging
 import os
 import shutil
 import uuid
+from collections import defaultdict
 from pathlib import Path
 from typing import Any
 
 import httpx
 from fastapi import Depends, FastAPI, File, Form, HTTPException, Request, UploadFile
-from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse, Response
+from fastapi.responses import (
+    FileResponse,
+    HTMLResponse,
+    RedirectResponse,
+    Response,
+    StreamingResponse,
+)
 from pdf2zh_next import BasicSettings
 from pdf2zh_next import OpenAISettings
 from pdf2zh_next import PDFSettings
@@ -53,6 +61,7 @@ _job_queue: asyncio.Queue[str] = asyncio.Queue()
 _worker_task: asyncio.Task[None] | None = None
 _running_tasks: dict[str, asyncio.Task[None]] = {}
 _active_job_by_user: dict[str, str] = {}
+_job_subscribers: dict[str, set[asyncio.Queue[dict[str, Any]]]] = defaultdict(set)
 
 
 def _build_settings_for_job(row: sqlite3.Row) -> SettingsModel:
@@ -82,16 +91,21 @@ async def _run_single_job(job_id: str) -> None:
     row = jobs.get_job_row(job_id)
     if row is None:
         return
-    if row["status"] != "queued":
+    if row["status"] != jobs.STATUS_QUEUED:
         return
     if row["cancel_requested"]:
-        _update_job(job_id, status="cancelled", message="Cancelled before start")
+        await _transition_and_notify(
+            job_id,
+            "cancel_before_start",
+            message="Cancelled before start",
+            finished_at=storage.now_iso(),
+        )
         return
 
     username = row["username"]
-    jobs.update_job(
+    await _transition_and_notify(
         job_id,
-        status="running",
+        "start",
         started_at=storage.now_iso(),
         message="Translation started",
         progress=0.0,
@@ -108,16 +122,17 @@ async def _run_single_job(job_id: str) -> None:
             if event_type in {"progress_start", "progress_update", "progress_end"}:
                 progress = float(event.get("overall_progress", 0.0))
                 stage = event.get("stage", "")
-                jobs.update_job(
+                await _transition_and_notify(
                     job_id,
+                    "progress",
                     progress=max(0.0, min(100.0, progress)),
                     message=f"{stage}" if stage else "Running",
                 )
             elif event_type == "error":
                 error_msg = str(event.get("error", "Unknown translation error"))
-                jobs.update_job(
+                await _transition_and_notify(
                     job_id,
-                    status="failed",
+                    "finish_error",
                     error=error_msg,
                     message="Translation failed",
                     finished_at=storage.now_iso(),
@@ -141,9 +156,9 @@ async def _run_single_job(job_id: str) -> None:
                         elif ".dual.pdf" in name and not dual_path:
                             dual_path = str(file)
 
-                jobs.update_job(
+                await _transition_and_notify(
                     job_id,
-                    status="succeeded",
+                    "finish_ok",
                     progress=100.0,
                     message="Translation finished",
                     finished_at=storage.now_iso(),
@@ -153,26 +168,26 @@ async def _run_single_job(job_id: str) -> None:
                 )
                 return
 
-        jobs.update_job(
+        await _transition_and_notify(
             job_id,
-            status="failed",
+            "finish_error",
             error="Translation stream ended unexpectedly",
             message="Translation failed",
             finished_at=storage.now_iso(),
-    )
+        )
     except asyncio.CancelledError:
-        jobs.update_job(
+        await _transition_and_notify(
             job_id,
-            status="cancelled",
+            "cancel_running",
             message="Cancelled by user",
             finished_at=storage.now_iso(),
         )
         raise
     except Exception as exc:  # noqa: BLE001
         logger.exception("Translation job failed: %s", job_id)
-        jobs.update_job(
+        await _transition_and_notify(
             job_id,
-            status="failed",
+            "finish_error",
             error=str(exc),
             message="Translation failed",
             finished_at=storage.now_iso(),
@@ -222,6 +237,79 @@ def _enqueue_pending_jobs() -> None:
         _job_queue.put_nowait(row["id"])
 
 
+async def _publish_job_event(job: dict[str, Any]) -> None:
+    """将任务更新推送给所有订阅该用户的 SSE 连接。"""
+    username = job.get("username")
+    if not username:
+        return
+
+    payload = {
+        "id": job["id"],
+        "username": username,
+        "status": job.get("status"),
+        "progress": job.get("progress"),
+        "message": job.get("message"),
+        "error": job.get("error"),
+        "updated_at": job.get("updated_at"),
+        "artifact_urls": job.get("artifact_urls") or {},
+        "model": job.get("model"),
+        "filename": job.get("filename"),
+        "created_at": job.get("created_at"),
+    }
+
+    queues = list(_job_subscribers.get(username, ()))
+    for q in queues:
+        try:
+            q.put_nowait(payload)
+        except asyncio.QueueFull:
+            # 简单策略：丢弃最旧一条再塞新事件，防止阻塞 worker
+            try:
+                _ = q.get_nowait()
+            except asyncio.QueueEmpty:
+                pass
+            try:
+                q.put_nowait(payload)
+            except asyncio.QueueFull:
+                logger.warning(
+                    "Dropping job event for user=%s job_id=%s due to full queue",
+                    username,
+                    job.get("id"),
+                )
+
+
+async def _transition_and_notify(
+    job_id: str,
+    event: str,
+    **fields: Any,
+) -> dict[str, Any] | None:
+    """状态机迁移并推送事件给订阅者。"""
+    job = jobs.transition_job(job_id, event, **fields)
+    if job is not None:
+        await _publish_job_event(job)
+    else:
+        logger.warning(
+            "Invalid job transition: job_id=%s event=%s", job_id, event
+        )
+    return job
+
+
+def _subscribe_user_jobs(username: str) -> asyncio.Queue[dict[str, Any]]:
+    """注册一个用户的 SSE 订阅队列。"""
+    q: asyncio.Queue[dict[str, Any]] = asyncio.Queue(maxsize=100)
+    _job_subscribers[username].add(q)
+    return q
+
+
+def _unsubscribe_user_jobs(username: str, queue: asyncio.Queue[dict[str, Any]]) -> None:
+    """取消用户的 SSE 订阅队列。"""
+    queues = _job_subscribers.get(username)
+    if not queues:
+        return
+    queues.discard(queue)
+    if not queues:
+        _job_subscribers.pop(username, None)
+
+
 def _login_page(error: str = "") -> str:
     """渲染登录页 HTML。"""
     tpl = load_template("login.html")
@@ -413,16 +501,21 @@ async def api_cancel_job(
         raise HTTPException(status_code=404, detail="Job not found")
 
     status = row["status"]
-    if status in {"succeeded", "failed", "cancelled"}:
+    if status in {
+        jobs.STATUS_SUCCEEDED,
+        jobs.STATUS_FAILED,
+        jobs.STATUS_CANCELLED,
+    }:
         return {"status": status, "message": "Job already finished"}
 
     jobs.update_job(job_id, cancel_requested=1, message="Cancel requested")
-    if status == "queued":
-        jobs.update_job(
+    if status == jobs.STATUS_QUEUED:
+        await _transition_and_notify(
             job_id,
-            status="cancelled",
+            "cancel_before_start",
             finished_at=storage.now_iso(),
             progress=0.0,
+            message="Job cancelled",
         )
         return {"status": "cancelled", "message": "Job cancelled"}
 
@@ -433,6 +526,36 @@ async def api_cancel_job(
     return {"status": "cancelling", "message": "Cancellation requested"}
 
 
+@app.get("/api/jobs/stream")
+async def api_jobs_stream(
+    request: Request,
+    username: str = Depends(auth._require_user),
+) -> StreamingResponse:
+    """任务状态 SSE 推送，仅推送当前用户的任务更新。"""
+
+    queue = _subscribe_user_jobs(username)
+
+    async def event_generator() -> Any:
+        try:
+            while True:
+                # 支持客户端主动断开
+                if await request.is_disconnected():
+                    break
+                payload = await queue.get()
+                yield f"data: {json.dumps(payload)}\n\n"
+        except asyncio.CancelledError:
+            logger.info("Job SSE connection cancelled for user=%s", username)
+            raise
+        finally:
+            _unsubscribe_user_jobs(username, queue)
+
+    return StreamingResponse(
+        event_generator(),
+        media_type="text/event-stream",
+        headers={"Cache-Control": "no-cache"},
+    )
+
+
 @app.get("/api/jobs/{job_id}/artifacts/{artifact_type}")
 async def api_download_artifact(
     job_id: str,

src/jobs.py

@@ -10,6 +10,24 @@ import sqlite3
 import storage
 
 
+# ── 任务状态与状态机 ────────────────────────────────────────────────────────────
+
+# 约定的任务状态枚举，避免在业务层随意写字符串
+STATUS_QUEUED = "queued"
+STATUS_RUNNING = "running"
+STATUS_SUCCEEDED = "succeeded"
+STATUS_FAILED = "failed"
+STATUS_CANCELLED = "cancelled"
+
+ALLOWED_STATUSES: set[str] = {
+    STATUS_QUEUED,
+    STATUS_RUNNING,
+    STATUS_SUCCEEDED,
+    STATUS_FAILED,
+    STATUS_CANCELLED,
+}
+
+
 def row_to_job_dict(row: sqlite3.Row) -> dict[str, Any]:
     """将任务行转换为对外暴露的字典结构。"""
     job = dict(row)
@@ -28,7 +46,11 @@ def row_to_job_dict(row: sqlite3.Row) -> dict[str, Any]:
 
 
 def update_job(job_id: str, **fields: Any) -> None:
-    """更新任务记录指定字段。"""
+    """更新任务记录指定字段。
+
+    注意：业务代码应该优先通过 transition_job 做状态机驱动更新，
+    直接调用本函数仅用于与状态无关的字段（例如 cancel_requested）。
+    """
     if not fields:
         return
     fields["updated_at"] = storage.now_iso()
@@ -133,3 +155,76 @@ def resolve_artifact_path(raw_path: str | None, output_dir: Path) -> Path | None
         return None
     return path
 
+
+def transition_job(job_id: str, event: str, **extra_fields: Any) -> dict[str, Any] | None:
+    """基于事件驱动的任务状态迁移。
+
+    这里只负责：
+      * 校验当前状态是否允许执行给定事件
+      * 决定目标状态（如果有）
+      * 写入数据库
+      * 返回更新后的任务字典（用于推送给前端）
+
+    状态枚举固定为 queued/running/succeeded/failed/cancelled，避免状态空间爆炸。
+    """
+    row = get_job_row(job_id)
+    if row is None:
+        return None
+
+    current_status = row["status"]
+    if current_status not in ALLOWED_STATUSES:
+        # 非法状态一律拒绝迁移，由调用方记录日志
+        return None
+
+    # 简单的事件 -> 允许来源状态集合、目标状态映射
+    # 对于 progress 这类事件，目标状态为 None，只更新进度等字段。
+    transitions: dict[str, dict[str, Any]] = {
+        "start": {
+            "from": {STATUS_QUEUED},
+            "to": STATUS_RUNNING,
+        },
+        "progress": {
+            "from": {STATUS_RUNNING},
+            "to": None,
+        },
+        "finish_ok": {
+            "from": {STATUS_RUNNING},
+            "to": STATUS_SUCCEEDED,
+        },
+        "finish_error": {
+            "from": {STATUS_QUEUED, STATUS_RUNNING},
+            "to": STATUS_FAILED,
+        },
+        "cancel_before_start": {
+            "from": {STATUS_QUEUED},
+            "to": STATUS_CANCELLED,
+        },
+        "cancel_running": {
+            "from": {STATUS_RUNNING},
+            "to": STATUS_CANCELLED,
+        },
+        # 预留重启失败事件，当前在 gateway 中直接 SQL 处理，不走这里
+        "restart_fail": {
+            "from": {STATUS_RUNNING},
+            "to": STATUS_FAILED,
+        },
+    }
+
+    cfg = transitions.get(event)
+    if cfg is None:
+        return None
+
+    if current_status not in cfg["from"]:
+        return None
+
+    fields: dict[str, Any] = dict(extra_fields)
+    target_status = cfg["to"]
+    if target_status is not None:
+        fields["status"] = target_status
+
+    update_job(job_id, **fields)
+    new_row = get_job_row(job_id)
+    if new_row is None:
+        return None
+    return row_to_job_dict(new_row)
+

src/web/static/dashboard.js

@@ -42,6 +42,9 @@ async function refreshBilling() {
   }
 }
 
+// 任务状态缓存：在前端维护一个简单的内存表，方便 SSE/轮询统一渲染
+const jobsState = new Map();
+
 function actionButtons(job) {
   const actions = [];
   if (job.status === "queued" || job.status === "running") {
@@ -78,12 +81,16 @@ function statusText(status) {
   return statusMap[status] || status;
 }
 
-async function refreshJobs() {
-  const data = await apiJson("/api/jobs?limit=50");
+function renderJobsFromState() {
   const body = document.getElementById("jobsBody");
   body.innerHTML = "";
 
-  for (const job of data.jobs) {
+  const jobs = Array.from(jobsState.values());
+  jobs.sort((a, b) =>
+    (b.created_at || "").localeCompare(a.created_at || ""),
+  );
+
+  for (const job of jobs) {
     const tr = document.createElement("tr");
     tr.innerHTML = `
       <td class="mono">${esc(job.id)}</td>
@@ -98,6 +105,21 @@ async function refreshJobs() {
   }
 }
 
+function upsertJob(jobPatch) {
+  const existing = jobsState.get(jobPatch.id) || {};
+  jobsState.set(jobPatch.id, { ...existing, ...jobPatch });
+  renderJobsFromState();
+}
+
+async function refreshJobs() {
+  const data = await apiJson("/api/jobs?limit=50");
+  jobsState.clear();
+  for (const job of data.jobs) {
+    jobsState.set(job.id, job);
+  }
+  renderJobsFromState();
+}
+
 async function cancelJob(jobId) {
   try {
     await apiJson(`/api/jobs/${jobId}/cancel`, { method: "POST" });
@@ -127,6 +149,55 @@ async function refreshAll() {
   await Promise.all([refreshJobs(), refreshBilling()]);
 }
 
+let jobEventSource = null;
+let pollingEnabled = true;
+const POLL_INTERVAL_MS = 10000;
+
+function setupJobStream() {
+  if (!("EventSource" in window)) {
+    console.warn("EventSource not supported, fallback to polling");
+    pollingEnabled = true;
+    return;
+  }
+
+  jobEventSource = new EventSource("/api/jobs/stream");
+
+  jobEventSource.onmessage = (event) => {
+    try {
+      const payload = JSON.parse(event.data);
+      if (!payload || !payload.id) {
+        return;
+      }
+      upsertJob(payload);
+    } catch (err) {
+      console.error("Failed to parse job SSE payload:", err);
+    }
+  };
+
+  jobEventSource.onerror = () => {
+    console.error("Job SSE error, switching back to polling");
+    if (jobEventSource) {
+      jobEventSource.close();
+      jobEventSource = null;
+    }
+    pollingEnabled = true;
+  };
+
+  pollingEnabled = false;
+}
+
 refreshAll();
-setInterval(refreshAll, 3000);
+setupJobStream();
 
+setInterval(async () => {
+  if (document.hidden) {
+    // 页面不可见时降低刷新频率：完全跳过本轮
+    return;
+  }
+
+  if (pollingEnabled) {
+    await refreshAll();
+  } else {
+    await refreshBilling();
+  }
+}, POLL_INTERVAL_MS);