feat: show progress while preparing batch archives

apps/admin_console/services/batch_archive.py

@@ -0,0 +1,117 @@
+"""Background preparation and observable progress for batch report downloads."""
+
+import threading
+import time
+
+from apps.admin_console.services import report_export
+
+_lock = threading.Lock()
+_states: dict[str, dict] = {}
+
+
+def archive_path(batch_id: str):
+    return report_export.EXPORT_ROOT / "batches" / batch_id / "批次报告.zip"
+
+
+def _snapshot(state: dict) -> dict:
+    elapsed = max(0, (state.get("finished_at") or time.monotonic()) - state["started_at"])
+    completed, total = state["completed"], state["total"]
+    eta = (
+        max(0, round(elapsed / completed * (total - completed)))
+        if completed and total > completed
+        else None
+    )
+    return {
+        "status": state["status"],
+        "stage": state["stage"],
+        "completed": completed,
+        "total": total,
+        "elapsed_seconds": round(elapsed),
+        "eta_seconds": eta,
+        "error": state.get("error"),
+    }
+
+
+def status(batch_id: str, batch: dict) -> dict:
+    with _lock:
+        state = _states.get(batch_id)
+        if state:
+            return _snapshot(state)
+    if batch.get("report_ready") and archive_path(batch_id).is_file():
+        return {
+            "status": "ready",
+            "stage": "报告包已就绪",
+            "completed": 0,
+            "total": 0,
+            "elapsed_seconds": 0,
+            "eta_seconds": None,
+            "error": None,
+        }
+    return {
+        "status": "idle",
+        "stage": "尚未开始打包",
+        "completed": 0,
+        "total": 0,
+        "elapsed_seconds": 0,
+        "eta_seconds": None,
+        "error": None,
+    }
+
+
+def prepare(batch_id: str, batch: dict) -> dict:
+    with _lock:
+        current = _states.get(batch_id)
+        if current and current["status"] == "preparing":
+            return _snapshot(current)
+        if batch.get("report_ready") and archive_path(batch_id).is_file():
+            _states.pop(batch_id, None)
+            return status_from_cache()
+        state = {
+            "status": "preparing",
+            "stage": "等待打包",
+            "completed": 0,
+            "total": len(report_export.store.jobs(batch_id)),
+            "started_at": time.monotonic(),
+        }
+        _states[batch_id] = state
+
+    def update(completed: int, total: int, stage: str) -> None:
+        with _lock:
+            state.update(completed=completed, total=total, stage=stage)
+
+    def run() -> None:
+        try:
+            with report_export.EXPORT_LOCK:
+                latest = report_export.store.get("batch", batch_id)
+                if not (latest and latest.get("report_ready") and archive_path(batch_id).is_file()):
+                    report_export.export_batch(batch_id, progress=update)
+            with _lock:
+                state.update(status="ready", stage="报告包已就绪", finished_at=time.monotonic())
+        except Exception as exc:
+            with _lock:
+                state.update(
+                    status="failed",
+                    stage="打包失败",
+                    error=str(exc)[:500],
+                    finished_at=time.monotonic(),
+                )
+
+    threading.Thread(target=run, name=f"batch-archive-{batch_id[:8]}", daemon=True).start()
+    return status(batch_id, batch)
+
+
+def status_from_cache() -> dict:
+    return {
+        "status": "ready",
+        "stage": "报告包已就绪",
+        "completed": 0,
+        "total": 0,
+        "elapsed_seconds": 0,
+        "eta_seconds": None,
+        "error": None,
+    }
+
+
+def forget(batch_id: str) -> None:
+    with _lock:
+        _states.pop(batch_id, None)

apps/admin_console/services/report_export.py

@@ -1,7 +1,7 @@
 """Evidence-backed Word/Excel artifacts, independent of the language model runtime."""
 
 from datetime import datetime
-from typing import cast
+from typing import Callable, cast
 import json
 from pathlib import Path
 import re
@@ -695,7 +695,7 @@ def export_session_media(session_id: str, kind: str) -> Path:
         return target
 
 
-def export_batch(batch_id: str) -> Path:
+def export_batch(batch_id: str, progress: Callable[[int, int, str], None] | None = None) -> Path:
     with EXPORT_LOCK:
         batch = store.get("batch", identifier(batch_id))
         if not batch:
@@ -704,6 +704,8 @@ def export_batch(batch_id: str) -> Path:
         directory.mkdir(parents=True, exist_ok=True)
         formats = batch["formats"]
         records = [session_report(job["session_id"]) for job in store.jobs(batch_id)]
+        if progress:
+            progress(0, len(records), "正在生成批次汇总")
         rows = [("任务编号", "任务名称", "执行状态", "验证结论", "异常原因")]
         rows += [
             (r["id"], r["name"], STATUS.get(r["status"], r["status"]), r["verdict"], r["error"])
@@ -734,6 +736,8 @@ def export_batch(batch_id: str) -> Path:
             for path in outputs:
                 archive.write(path, path.name)
             for number, report in enumerate(records, 1):
+                if progress:
+                    progress(number - 1, len(records), f"正在打包任务 {number}/{len(records)}")
                 prefix = f"任务报告/{number:04d}_{archive_case_name(report)}"
                 for path in export_session(report["id"], formats):
                     archive.write(path, f"{prefix}/{path.name}")
@@ -741,5 +745,9 @@ def export_batch(batch_id: str) -> Path:
                     archive.write(source, f"{prefix}/步骤截图/{name}")
                 for source, name in session_videos(report):
                     archive.write(source, f"{prefix}/任务录屏/{name}")
+                if progress:
+                    progress(number, len(records), f"已打包任务 {number}/{len(records)}")
+        if progress:
+            progress(len(records), len(records), "正在完成压缩包")
         temporary.replace(target)
         return target