Improve AI batch progress controls

2026-05-27 12:59:15 +08:00
parent 5d96c1de6d
commit b5f41c3dda
2 changed files with 229 additions and 23 deletions
--- a/患者首页处理/数据可视化网页端/app/main.py
+++ b/患者首页处理/数据可视化网页端/app/main.py
@@ -232,6 +232,7 @@ WORKFLOW_LOCK = threading.Lock()
 WORKFLOW_READY = False
 AI_JOB_LOCK = threading.Lock()
 AI_REVIEW_JOB: dict[str, Any] = {
+    "kind": "ai_review",
    "running": False,
    "cancel_requested": False,
    "scope": "",
@@ -248,6 +249,19 @@ AI_REVIEW_JOB: dict[str, Any] = {
    "last_record_id": None,
    "privacy_mode": True,
 }
+BULK_JOB_LOCK = threading.Lock()
+BULK_APPROVE_JOB: dict[str, Any] = {
+    "kind": "approve_ai_passed",
+    "running": False,
+    "total": 0,
+    "processed": 0,
+    "updated": 0,
+    "failed": 0,
+    "message": "",
+    "started_at": "",
+    "finished_at": "",
+}
+APPROVE_BATCH_SIZE = 500
 DEFAULT_STATUS_CHECK_TIME = env("REVIEW_STATUS_CHECK_TIME", "03:00") or "03:00"
 DEFAULT_KIMI_API_BASE = env("MOONSHOT_API_BASE", env("KIMI_API_BASE", "https://api.moonshot.cn/v1")) or "https://api.moonshot.cn/v1"
 DEFAULT_KIMI_MODEL = env("KIMI_MODEL", "kimi-k2.6") or "kimi-k2.6"
@@ -677,7 +691,7 @@ def ai_action_mode_to_override(mode: str) -> dict[str, Any]:
 def ai_scope_allowed(mode: str, scope: str) -> bool:
    mode = normalize_kimi_ai_scope_mode(mode)
    if mode == "all":
-        return scope in {"current", "five", "fifty", "all", "ai_pending"}
+        return scope in {"current", "five", "fifty", "all", "ai_pending", "privacy_blocked"}
    if mode == "five":
        return scope in {"current", "five"}
    if mode == "current":
@@ -2716,8 +2730,8 @@ def apply_ai_review_with_retry(record_id: int, attempts: int = 3, kimi_override:


 def ai_target_ids(scope: str, record_id: int | None) -> list[int]:
-    if scope not in {"current", "five", "fifty", "all", "ai_pending"}:
-        raise HTTPException(status_code=400, detail="AI核验范围只能是 current/five/fifty/all/ai_pending")
+    if scope not in {"current", "five", "fifty", "all", "ai_pending", "privacy_blocked"}:
+        raise HTTPException(status_code=400, detail="AI核验范围只能是 current/five/fifty/all/ai_pending/privacy_blocked")
    if scope in {"current", "five", "fifty"} and not record_id:
        raise HTTPException(status_code=400, detail="缺少当前记录 ID")
    if scope == "current":
@@ -2728,6 +2742,43 @@ def ai_target_ids(scope: str, record_id: int | None) -> list[int]:
            cur.execute(query, (AI_PENDING_STATUS,))
            rows = cur.fetchall()
        return [int(row["id"]) for row in rows]
+    if scope == "privacy_blocked":
+        query = sql.SQL(
+            """
+            SELECT id
+            FROM {table}
+            WHERE review_status = %s
+              AND EXISTS (
+                SELECT 1
+                FROM jsonb_array_elements(COALESCE(review_logs, '[]'::jsonb)) AS log
+                WHERE log->>'changed_by' = 'AI'
+                  AND (
+                    log->'ai_result'->>'ai_question' ILIKE %s
+                    OR log->'ai_result'->'parsed'->>'summary' ILIKE %s
+                    OR (log->'ai_result'->'parsed'->'remaining_issues')::text ILIKE %s
+                  )
+                  AND (
+                    (log->'ai_result'->'parsed'->'remaining_issues')::text ILIKE %s
+                    OR (log->'ai_result'->'parsed'->'remaining_issues')::text ILIKE %s
+                  )
+              )
+            ORDER BY id
+            """
+        ).format(table=table_identifier())
+        with connect() as conn, conn.cursor() as cur:
+            cur.execute(
+                query,
+                (
+                    AI_PENDING_STATUS,
+                    "%隐私模式本地判定%",
+                    "%隐私模式未上传敏感区域%",
+                    "%隐私模式不上传%",
+                    "%基本信息%",
+                    "%地址联系人%",
+                ),
+            )
+            rows = cur.fetchall()
+        return [int(row["id"]) for row in rows]

    where = sql.SQL("review_status = 'needs_review'")
    params: list[Any] = []
@@ -2784,6 +2835,7 @@ def run_ai_review_job(scope: str, ids: list[int], kimi_override: dict[str, Any]
    privacy_mode = normalize_bool((kimi_override or {}).get("privacy_mode"), True)
    stop_event = threading.Event()
    update_ai_job(
+        kind="ai_review",
        running=True,
        cancel_requested=False,
        scope=scope,
@@ -2866,7 +2918,28 @@ def run_ai_review_job(scope: str, ids: list[int], kimi_override: dict[str, Any]
    )


-def mark_ai_no_issue_reviewed() -> int:
+def ai_no_issue_reviewed_ids() -> list[int]:
+    query = sql.SQL(
+        """
+        SELECT id
+        FROM {table}
+        WHERE review_status IN ('AI已处理-OK', 'AI复核-无问题')
+           OR (
+             review_status = 'auto_pass'
+             AND COALESCE(review_logs, '[]'::jsonb) @> jsonb_build_array(jsonb_build_object('changed_by', 'AI'))
+           )
+        ORDER BY id
+        """
+    ).format(table=table_identifier())
+    with connect() as conn, conn.cursor() as cur:
+        cur.execute(query)
+        rows = cur.fetchall()
+    return [int(row["id"]) for row in rows]
+
+
+def mark_ai_no_issue_reviewed_batch(record_ids: list[int]) -> int:
+    if not record_ids:
+        return 0
    changed_at = datetime.now().isoformat(timespec="seconds")
    note = f"批量确认历史AI已通过({datetime.now().strftime('%Y-%m-%d %H:%M:%S')}): 确认为已人工复核"
    query = sql.SQL(
@@ -2888,17 +2961,69 @@ def mark_ai_no_issue_reviewed() -> int:
                    )
                )
            )
-        WHERE review_status IN ('AI已处理-OK', 'AI复核-无问题')
+        WHERE id = ANY(%s)
+          AND (
+            review_status IN ('AI已处理-OK', 'AI复核-无问题')
+            OR (
+              review_status = 'auto_pass'
+              AND COALESCE(review_logs, '[]'::jsonb) @> jsonb_build_array(jsonb_build_object('changed_by', 'AI'))
+            )
+          )
        """
    ).format(table=table_identifier())
    with connect() as conn, conn.cursor() as cur:
-        cur.execute(query, (datetime.now().strftime("%Y%m%d%H%M%S%f"), changed_at, note))
+        cur.execute(query, (datetime.now().strftime("%Y%m%d%H%M%S%f"), changed_at, note, record_ids))
        count = cur.rowcount
        conn.commit()
-    refresh_status_snapshot(source="bulk")
    return int(count)


+def update_bulk_approve_job(**updates: Any) -> dict[str, Any]:
+    with BULK_JOB_LOCK:
+        BULK_APPROVE_JOB.update(updates)
+        return dict(BULK_APPROVE_JOB)
+
+
+def run_approve_ai_no_issue_job(record_ids: list[int]) -> None:
+    update_bulk_approve_job(
+        kind="approve_ai_passed",
+        running=True,
+        total=len(record_ids),
+        processed=0,
+        updated=0,
+        failed=0,
+        message="批量通过AI已通过中",
+        started_at=datetime.now().isoformat(timespec="seconds"),
+        finished_at="",
+    )
+    try:
+        updated = 0
+        for start in range(0, len(record_ids), APPROVE_BATCH_SIZE):
+            batch = record_ids[start : start + APPROVE_BATCH_SIZE]
+            updated += mark_ai_no_issue_reviewed_batch(batch)
+            update_bulk_approve_job(
+                processed=min(start + len(batch), len(record_ids)),
+                updated=updated,
+                message="批量通过AI已通过中",
+            )
+            time.sleep(0.05)
+        refresh_status_snapshot(source="bulk")
+        update_bulk_approve_job(
+            running=False,
+            processed=len(record_ids),
+            updated=updated,
+            message=f"批量通过完成，已更新 {updated} 条",
+            finished_at=datetime.now().isoformat(timespec="seconds"),
+        )
+    except Exception as exc:  # noqa: BLE001
+        update_bulk_approve_job(
+            running=False,
+            failed=1,
+            message=f"批量通过失败：{exc}",
+            finished_at=datetime.now().isoformat(timespec="seconds"),
+        )
+
+
 def submit_reviewed_records() -> int:
    changed_at = datetime.now().isoformat(timespec="seconds")
    note = f"一键提交已人工复核项目({datetime.now().strftime('%Y-%m-%d %H:%M:%S')}): 已从患者首页复核工作台隐藏"
@@ -3370,6 +3495,12 @@ def ai_review_status():
        return dict(AI_REVIEW_JOB)


+@app.get("/api/ai/review/approve-no-issue/status")
+def approve_ai_no_issue_status():
+    with BULK_JOB_LOCK:
+        return dict(BULK_APPROVE_JOB)
+
+
@app.post("/api/ai/review/cancel")
 def cancel_ai_review():
    with AI_JOB_LOCK:
@@ -3381,8 +3512,28 @@ def cancel_ai_review():

@app.post("/api/ai/review/approve-no-issue")
 def approve_ai_no_issue():
-    count = mark_ai_no_issue_reviewed()
-    return {"ok": True, "updated": count, "status_snapshot": load_local_settings().get("status_snapshot")}
+    with BULK_JOB_LOCK:
+        if BULK_APPROVE_JOB.get("running"):
+            raise HTTPException(status_code=409, detail="已有批量通过任务正在运行")
+    ids = ai_no_issue_reviewed_ids()
+    if not ids:
+        update_bulk_approve_job(
+            kind="approve_ai_passed",
+            running=False,
+            total=0,
+            processed=0,
+            updated=0,
+            failed=0,
+            message="当前没有AI已通过记录需要批量通过",
+            started_at=datetime.now().isoformat(timespec="seconds"),
+            finished_at=datetime.now().isoformat(timespec="seconds"),
+        )
+        return dict(BULK_APPROVE_JOB)
+    thread = threading.Thread(target=run_approve_ai_no_issue_job, args=(ids,), name="approve-ai-passed", daemon=True)
+    thread.start()
+    time.sleep(0.1)
+    with BULK_JOB_LOCK:
+        return dict(BULK_APPROVE_JOB)


@app.post("/api/ai/review")
--- a/患者首页处理/数据可视化网页端/app/static/app.js
+++ b/患者首页处理/数据可视化网页端/app/static/app.js
@@ -19,6 +19,8 @@ const state = {
  aiJob: null,
  aiJobOwned: false,
  aiPolling: null,
+  bulkJob: null,
+  bulkPolling: null,
  currentUser: null,
  authenticated: false,
  pdfZoom: Number(localStorage.getItem("frontPagePdfZoom")) || 110,
@@ -425,7 +427,7 @@ function renderFilterActions() {
  const aiPassedCount = Number(state.status?.ai_passed || 0);
  const shouldShow = ["review_all", "ai_passed"].includes(filter);
  button.classList.toggle("is-hidden", !shouldShow);
-  button.disabled = !shouldShow || aiPassedCount <= 0;
+  button.disabled = !shouldShow || aiPassedCount <= 0 || Boolean(state.bulkJob?.running);
  button.textContent = aiPassedCount > 0 ? `通过所有AI已通过（${aiPassedCount}）` : "通过所有AI已通过";
 }

@@ -1414,22 +1416,39 @@ function setAiButtonsDisabled(disabled) {
  });
 }

+function clearAiProgressPanel() {
+  const panel = $("aiProgressPanel");
+  if (!panel) return;
+  panel.classList.add("is-hidden");
+  $("aiProgressBar").style.width = "0%";
+  $("aiProgressText").textContent = "0/0";
+  $("aiProgressMeta").textContent = "等待开始";
+  $("aiCancelBtn").classList.remove("is-hidden");
+  $("aiCancelBtn").disabled = true;
+}
+
 function renderAiProgress(job = state.aiJob) {
  const panel = $("aiProgressPanel");
  if (!panel) return;
-  if (!job || (!job.running && !job.total && !job.message)) {
-    panel.classList.add("is-hidden");
+  if (!job || (!job.running && !job.total && !job.message) || (!job.running && job.cancel_requested)) {
+    clearAiProgressPanel();
    return;
  }
  const total = Number(job.total || 0);
  const processed = Number(job.processed || 0);
  const percent = total ? Math.min(100, Math.round((processed / total) * 100)) : 0;
+  const isBulkApprove = job.kind === "approve_ai_passed";
  panel.classList.remove("is-hidden");
-  $("aiProgressTitle").textContent = job.running ? "AI处理中" : displayText(job.message || "AI处理完成");
+  $("aiProgressTitle").textContent = isBulkApprove
+    ? (job.running ? "批量通过中" : displayText(job.message || "批量通过完成"))
+    : (job.running ? "AI处理中" : displayText(job.message || "AI处理完成"));
  $("aiProgressText").textContent = total ? `${processed}/${total}` : "--";
  $("aiProgressBar").style.width = `${percent}%`;
-  $("aiProgressMeta").textContent = `OK ${job.ok || 0} · 不OK ${job.pending || 0} · 失败 ${job.failed || 0} · 并发 ${job.concurrency || 1}`;
-  $("aiCancelBtn").disabled = !job.running;
+  $("aiProgressMeta").textContent = isBulkApprove
+    ? `已通过 ${job.updated || 0} · 剩余 ${Math.max(0, total - processed)} · 失败 ${job.failed || 0}`
+    : `OK ${job.ok || 0} · 不OK ${job.pending || 0} · 失败 ${job.failed || 0} · 并发 ${job.concurrency || 1}`;
+  $("aiCancelBtn").classList.toggle("is-hidden", isBulkApprove);
+  $("aiCancelBtn").disabled = isBulkApprove || !job.running;
 }

 async function cancelAiReview({ silent = false } = {}) {
@@ -1439,6 +1458,9 @@ async function cancelAiReview({ silent = false } = {}) {
  renderAiProgress(job);
  setAiButtonsDisabled(Boolean(job.running));
  if (!silent) showMessage(displayText(job.message || "AI核验正在中断..."));
+  if (job.running && !state.aiPolling) {
+    state.aiPolling = setTimeout(() => pollAiReviewJob().catch((error) => showMessage(error.message, "error")), 1200);
+  }
  return job;
 }

@@ -1448,8 +1470,8 @@ async function pollAiReviewJob() {
  renderAiProgress(job);
  const total = job.total || 0;
  const processed = job.processed || 0;
-  showMessage(`AI处理中：${processed}/${total}，并发 ${job.concurrency || 1}，OK ${job.ok || 0}，不OK ${job.pending || 0}，失败 ${job.failed || 0}`, "");
  if (job.running) {
+    showMessage(`AI处理中：${processed}/${total}，并发 ${job.concurrency || 1}，OK ${job.ok || 0}，不OK ${job.pending || 0}，失败 ${job.failed || 0}`, "");
    if (state.aiJobOwned && !job.cancel_requested && state.activePage !== "review") switchPage("review");
    setAiButtonsDisabled(true);
    state.aiPolling = setTimeout(() => pollAiReviewJob().catch((error) => showMessage(error.message, "error")), 2000);
@@ -1458,6 +1480,15 @@ async function pollAiReviewJob() {
  state.aiPolling = null;
  state.aiJobOwned = false;
  setAiButtonsDisabled(false);
+  if (job.cancel_requested) {
+    clearAiProgressPanel();
+    showMessage(displayText(job.message || "AI核验已中断"), "ok");
+    await loadStatus().catch(() => null);
+    await loadOverview().catch(() => null);
+    await loadRecords();
+    await refreshSelectedRecord();
+    return;
+  }
  showMessage(displayText(`${job.message || "AI处理完成"}：OK ${job.ok || 0}，不OK ${job.pending || 0}，失败 ${job.failed || 0}`), job.failed ? "error" : "ok");
  await loadStatus().catch(() => null);
  await loadOverview().catch(() => null);
@@ -1502,19 +1533,37 @@ async function approveAiNoIssueRecords() {
  button.disabled = true;
  showMessage("正在批量确认历史 AI 已通过记录...");
  try {
-    const data = await api("/api/ai/review/approve-no-issue", { method: "POST" });
-    state.status = data.status_snapshot || state.status;
-    showMessage(`已批量确认 ${data.updated || 0} 条历史 AI 已通过记录`, "ok");
-    await loadStatus().catch(() => null);
-    await loadOverview().catch(() => null);
-    await loadRecords();
+    const job = await api("/api/ai/review/approve-no-issue", { method: "POST" });
+    state.bulkJob = job;
+    renderAiProgress(job);
+    await pollApproveAiNoIssueJob();
  } catch (error) {
    showMessage(error.message, "error");
  } finally {
-    button.disabled = false;
+    renderFilterActions();
  }
 }

+async function pollApproveAiNoIssueJob() {
+  const job = await api("/api/ai/review/approve-no-issue/status");
+  state.bulkJob = job;
+  renderAiProgress(job);
+  const total = Number(job.total || 0);
+  const processed = Number(job.processed || 0);
+  if (job.running) {
+    showMessage(`批量通过AI已通过：${processed}/${total}，已更新 ${job.updated || 0} 条`, "");
+    renderFilterActions();
+    state.bulkPolling = setTimeout(() => pollApproveAiNoIssueJob().catch((error) => showMessage(error.message, "error")), 1200);
+    return;
+  }
+  state.bulkPolling = null;
+  showMessage(displayText(job.message || `批量通过完成，已更新 ${job.updated || 0} 条`), job.failed ? "error" : "ok");
+  await loadStatus().catch(() => null);
+  await loadOverview().catch(() => null);
+  await loadRecords();
+  renderFilterActions();
+}
+
 async function submitReviewedRecords() {
  if (!window.confirm("确认提交所有已人工复核项目？提交后这些项目将不再出现在复核工作台列表中。")) return;
  const button = $("submitReviewedBtn");
@@ -1857,9 +1906,15 @@ async function loadWorkspace() {
    state.aiJobOwned = false;
    renderAiProgress(job);
  }
+  const bulkJob = await api("/api/ai/review/approve-no-issue/status").catch(() => null);
+  if (bulkJob?.running) {
+    state.bulkJob = bulkJob;
+    renderAiProgress(bulkJob);
+  }
  const page = canOpenPage(state.activePage) ? state.activePage : firstAllowedPage();
  switchPage(job?.running && state.aiJobOwned && canOpenPage("review") ? "review" : page);
  if (job?.running) pollAiReviewJob().catch((error) => showMessage(error.message, "error"));
+  if (bulkJob?.running) pollApproveAiNoIssueJob().catch((error) => showMessage(error.message, "error"));
 }

 boot();