feat: SSE progress streaming & cookie management UI

.gitignore

@@ -177,3 +177,6 @@ outputs/
 # Executable files
 *.exe
 .b2t/
+
+# Bilibili cookie credentials (contains sensitive session data)
+cookies.txt

src/b2t/cli.py

@@ -4,6 +4,7 @@
 import shutil
 import sys
 from pathlib import Path
+from typing import TYPE_CHECKING
 
 import typer
 
@@ -19,6 +20,9 @@
 from b2t.tasks import TaskService
 from b2t.user_config import AppConfig
 
+if TYPE_CHECKING:
+    from b2t.sse import SSEManager
+
 
 def create_app(language: str = DEFAULT_LANGUAGE) -> typer.Typer:
     app = typer.Typer(
@@ -309,14 +313,20 @@ def _run_server(*, host: str, port: int, provider: str | None, model: str | None
         )
         raise typer.Exit(code=1) from exc
 
+    from b2t.sse import SSEManager
     from b2t.web import create_app
 
     settings, config = _load_runtime(workspace=workspace, provider=provider, model=model)
-    service = _build_task_service(settings=settings, config=config, provider=provider, model=model)
+    sse = SSEManager()
+    service = _build_task_service(
+        settings=settings, config=config, provider=provider, model=model, sse_manager=sse,
+    )
     app_instance = create_app(
         task_service=service,
         library=service.library,
         database=service.database,
+        settings=settings,
+        sse_manager=sse,
         default_provider=provider or config.default_provider,
         default_model=model or config.default_model,
         language=config.language,
@@ -341,6 +351,7 @@ def _build_task_service(
     config: AppConfig,
     provider: str | None = None,
     model: str | None = None,
+    sse_manager: "SSEManager | None" = None,
 ) -> TaskService:
     database = AppDatabase(settings)
     library = WorkspaceLibrary(settings, database)
@@ -353,6 +364,7 @@ def _build_task_service(
             provider=selected_provider or provider or config.default_provider,
             model=selected_model or model or config.default_model,
         ),
+        sse_manager=sse_manager,
     )
     service.ensure_indexed()
     return service

src/b2t/i18n.py

@@ -181,6 +181,20 @@
         "web_result_video": "视频",
         "web_result_text": "文本内容",
 
+        # ── Cookie Settings ────────────────────────────────
+        "web_cookie_title": "B站 Cookie 设置",
+        "web_cookie_loading": "检查中...",
+        "web_cookie_hint": "粘贴或上传 Bilibili 的 Cookie（Netscape 格式），用于下载需要登录的视频。可在浏览器中安装 \"Get cookies.txt LOCALLY\" 扩展导出。",
+        "web_cookie_configured": "已配置",
+        "web_cookie_not_configured": "未配置",
+        "web_cookie_upload": "上传文件",
+        "web_cookie_save": "保存",
+        "web_cookie_delete": "删除",
+        "web_cookie_saved": "Cookie 已保存",
+        "web_cookie_save_failed": "保存失败",
+        "web_cookie_deleted": "Cookie 已删除",
+        "web_cookie_empty": "Cookie 内容不能为空",
+
         # ── Progress ─────────────────────────────────────────
         "progress_stage_queued": "已排队",
         "progress_stage_preparing": "准备中",
@@ -371,6 +385,20 @@
         "web_result_video": "Video",
         "web_result_text": "Transcript Text",
 
+        # ── Cookie Settings ────────────────────────────────
+        "web_cookie_title": "Bilibili Cookie Settings",
+        "web_cookie_loading": "Checking...",
+        "web_cookie_hint": "Paste or upload your Bilibili cookie (Netscape format) to download videos that require login. Use a browser extension like \"Get cookies.txt LOCALLY\" to export.",
+        "web_cookie_configured": "Configured",
+        "web_cookie_not_configured": "Not configured",
+        "web_cookie_upload": "Upload file",
+        "web_cookie_save": "Save",
+        "web_cookie_delete": "Delete",
+        "web_cookie_saved": "Cookie saved",
+        "web_cookie_save_failed": "Save failed",
+        "web_cookie_deleted": "Cookie deleted",
+        "web_cookie_empty": "Cookie content cannot be empty",
+
         # ── Progress ─────────────────────────────────────────
         "progress_stage_queued": "Queued",
         "progress_stage_preparing": "Preparing",

src/b2t/sse.py

@@ -0,0 +1,140 @@
+from __future__ import annotations
+
+import asyncio
+import json
+from collections import defaultdict
+from dataclasses import asdict
+from threading import Lock
+from typing import Any, AsyncIterator
+
+from b2t.models import ProgressSnapshot
+
+
+_TERMINAL_STATUSES = frozenset({"completed", "failed", "cancelled"})
+
+
+class SSEManager:
+    """Bridge between the ThreadPoolExecutor-based task worker and FastAPI's
+    async event loop so that progress updates can be streamed via SSE."""
+
+    def __init__(self) -> None:
+        self._subscribers: dict[str, list[asyncio.Queue[dict[str, Any] | None]]] = defaultdict(list)
+        self._lock = Lock()
+        self._loop: asyncio.AbstractEventLoop | None = None
+
+    # -- public API (called from async endpoints) -----------------------
+
+    def bind_loop(self, loop: asyncio.AbstractEventLoop | None = None) -> None:
+        """Bind the running asyncio event loop.  Must be called once at
+        application startup (from inside an async context)."""
+        self._loop = loop or asyncio.get_running_loop()
+
+    def subscribe(self, task_id: str) -> asyncio.Queue[dict[str, Any] | None]:
+        queue: asyncio.Queue[dict[str, Any] | None] = asyncio.Queue()
+        with self._lock:
+            self._subscribers[task_id].append(queue)
+        return queue
+
+    def unsubscribe(self, task_id: str, queue: asyncio.Queue[dict[str, Any] | None]) -> None:
+        with self._lock:
+            queues = self._subscribers.get(task_id, [])
+            try:
+                queues.remove(queue)
+            except ValueError:
+                pass
+            if not queues:
+                self._subscribers.pop(task_id, None)
+
+    async def event_stream(
+        self,
+        task_ids: list[str],
+        *,
+        history: list[dict[str, Any]] | None = None,
+    ) -> AsyncIterator[str]:
+        """Yield SSE-formatted strings for one or more task ids.
+
+        *history* is a list of already-serialised events that should be
+        replayed to a newly-connecting client so it can catch up.
+        """
+        queues: dict[str, asyncio.Queue[dict[str, Any] | None]] = {}
+        for tid in task_ids:
+            queues[tid] = self.subscribe(tid)
+
+        try:
+            # Replay historical events first.
+            if history:
+                for event in history:
+                    yield _format_sse("progress", event)
+
+            # Stream live events until every watched task reaches a terminal
+            # state.
+            finished = {tid: False for tid in task_ids}
+            while not all(finished.values()):
+                # Wait for the next event from *any* subscribed queue.
+                done_queues = {tid: q for tid, q in queues.items() if not finished[tid]}
+                if not done_queues:
+                    break
+
+                # Use asyncio.wait on all queue-get tasks.
+                tasks = {asyncio.ensure_future(q.get()): tid for tid, q in done_queues.items()}
+                done, _ = await asyncio.wait(tasks.keys(), return_when=asyncio.FIRST_COMPLETED)
+
+                for completed_task in done:
+                    tid = tasks[completed_task]
+                    try:
+                        payload = completed_task.result()
+                    except Exception:
+                        continue
+
+                    if payload is None:
+                        # Sentinel — queue closed.
+                        finished[tid] = True
+                        continue
+
+                    yield _format_sse("progress", payload)
+
+                    if payload.get("status") in _TERMINAL_STATUSES:
+                        finished[tid] = True
+        finally:
+            for tid, q in queues.items():
+                self.unsubscribe(tid, q)
+
+    # -- public API (called from worker threads) -----------------------
+
+    def publish(self, snapshot: ProgressSnapshot) -> None:
+        """Push a progress snapshot to all SSE subscribers.
+
+        Safe to call from any thread — it schedules the queue put on the
+        bound asyncio event loop.
+        """
+        payload = asdict(snapshot)
+        with self._lock:
+            queues = list(self._subscribers.get(snapshot.task_id, []))
+
+        loop = self._loop
+        if loop is None or loop.is_closed():
+            # No running loop yet — silently drop.
+            return
+
+        for queue in queues:
+            asyncio.run_coroutine_threadsafe(queue.put(payload), loop)
+
+    def notify_terminal(self, task_id: str) -> None:
+        """Send a sentinel ``None`` to all subscribers so their generators
+        can exit cleanly."""
+        with self._lock:
+            queues = list(self._subscribers.get(task_id, []))
+
+        loop = self._loop
+        if loop is None or loop.is_closed():
+            return
+
+        for queue in queues:
+            asyncio.run_coroutine_threadsafe(queue.put(None), loop)
+
+
+# -- helpers ---------------------------------------------------------------
+
+
+def _format_sse(event: str, data: dict[str, Any]) -> str:
+    return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"

src/b2t/tasks.py

@@ -2,23 +2,34 @@
 
 from concurrent.futures import Future, ThreadPoolExecutor
 from threading import Lock
-from typing import Callable
+from typing import TYPE_CHECKING, Callable
 
 from b2t.database import AppDatabase
 from b2t.library import WorkspaceLibrary
 from b2t.models import TaskRecord
 from b2t.pipeline import B2TPipeline
 from b2t.progress import ProgressCallback, ProgressReporter
 
+if TYPE_CHECKING:
+    from b2t.sse import SSEManager
+
 
 PipelineFactory = Callable[[str, str], B2TPipeline]
 
 
 class TaskService:
-    def __init__(self, *, database: AppDatabase, library: WorkspaceLibrary, pipeline_factory: PipelineFactory) -> None:
+    def __init__(
+        self,
+        *,
+        database: AppDatabase,
+        library: WorkspaceLibrary,
+        pipeline_factory: PipelineFactory,
+        sse_manager: "SSEManager | None" = None,
+    ) -> None:
         self.database = database
         self.library = library
         self.pipeline_factory = pipeline_factory
+        self.sse_manager = sse_manager
         self.executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="b2t-task")
         self._listeners: dict[str, list[ProgressCallback]] = {}
         self._futures: dict[str, Future[object]] = {}
@@ -88,6 +99,10 @@ def _run_transcription(self, task_id: str, source: str, provider: str, model: st
 
     def _handle_progress(self, snapshot) -> None:  # type: ignore[no-untyped-def]
         self.database.record_progress(snapshot)
+        if self.sse_manager is not None:
+            self.sse_manager.publish(snapshot)
+            if snapshot.status in {"completed", "failed", "cancelled"}:
+                self.sse_manager.notify_terminal(snapshot.task_id)
         with self._lock:
             callbacks = list(self._listeners.get(snapshot.task_id, []))
         for callback in callbacks:

src/b2t/templates/batch.html

@@ -8,7 +8,7 @@
 </head>
 <body class="bg-gray-50 text-gray-900">
   <main class="max-w-5xl mx-auto px-6 py-8 space-y-6">
-    <a href="/" class="text-sm text-indigo-600">← {{ t("web_back_home") }}</a>
+    <a href="/" class="text-sm text-indigo-600">&larr; {{ t("web_back_home") }}</a>
     <section class="rounded-xl border border-gray-200 bg-white p-6 space-y-4">
       <div class="flex items-center justify-between gap-4">
         <h1 class="text-2xl font-semibold">{{ t("web_batch_submitted", count=tasks|length) }}</h1>
@@ -49,31 +49,43 @@ <h1 class="text-2xl font-semibold">{{ t("web_batch_submitted", count=tasks|lengt
 
   <script>
     const rows = Array.from(document.querySelectorAll("[data-task-id]"));
+    const summaryEl = document.getElementById("batch-summary");
+    const finishedSet = new Set();
+    const totalRows = rows.length;
 
-    async function pollRow(row) {
-      const taskId = row.dataset.taskId;
-      const response = await fetch(`/api/tasks/${taskId}/progress`);
-      const data = await response.json();
-      const percent = Math.max(0, Math.min(100, Math.round((data.progress_percent || 0) * 100)));
-      row.querySelector(".task-status").textContent = `${data.status} · ${data.current_stage || ""}`;
-      row.querySelector(".task-bar").style.width = `${percent}%`;
-      if (data.status === "completed" && data.video_id) {
-        row.querySelector(".task-status").innerHTML = `<a class="text-indigo-600" href="/videos/${data.video_id}">${data.status}</a>`;
-      }
-      return data.status;
+    function updateSummary() {
+      summaryEl.textContent = `${finishedSet.size}/${totalRows}`;
     }
 
-    async function pollBatch() {
-      const statuses = await Promise.all(rows.map(pollRow));
-      const done = statuses.filter((status) => ["completed", "failed", "cancelled"].includes(status)).length;
-      document.getElementById("batch-summary").textContent = `${done}/${rows.length}`;
-      if (done < rows.length) {
-        setTimeout(pollBatch, 1000);
+    function applyProgress(data) {
+      const row = document.querySelector(`[data-task-id="${data.task_id}"]`);
+      if (!row) return;
+
+      const percent = Math.max(0, Math.min(100, Math.round((data.percent || data.progress_percent || 0) * 100)));
+      const stage = data.stage || data.current_stage || "";
+      row.querySelector(".task-status").textContent = `${data.status} \u00b7 ${stage}`;
+      row.querySelector(".task-bar").style.width = `${percent}%`;
+
+      if (data.status === "completed" && (data.video_id || (data.detail && data.detail.video_id))) {
+        const vid = data.video_id || data.detail.video_id;
+        row.querySelector(".task-status").innerHTML = `<a class="text-indigo-600" href="/videos/${vid}">${data.status}</a>`;
+        finishedSet.add(data.task_id);
+      } else if (data.status === "failed" || data.status === "cancelled") {
+        if (data.status === "failed") row.querySelector(".task-status").classList.add("text-red-600");
+        finishedSet.add(data.task_id);
       }
+      updateSummary();
     }
 
-    if (rows.length > 0) {
-      pollBatch();
+    if (totalRows > 0) {
+      const ids = rows.map((r) => r.dataset.taskId).join(",");
+      const es = new EventSource(`/api/tasks/stream?ids=${ids}`);
+      es.addEventListener("progress", (e) => {
+        applyProgress(JSON.parse(e.data));
+      });
+      es.addEventListener("error", () => {
+        // EventSource auto-reconnects; history replay covers missed events.
+      });
     }
   </script>
 </body>