From b5f41c3dda8770d4d5925f80bbaa166fbba14bfa Mon Sep 17 00:00:00 2001 From: admin <572701190@qq.com> Date: Wed, 27 May 2026 12:59:15 +0800 Subject: [PATCH] Improve AI batch progress controls --- 患者首页处理/数据可视化网页端/app/main.py | 169 +++++++++++++++++- .../数据可视化网页端/app/static/app.js | 83 +++++++-- 2 files changed, 229 insertions(+), 23 deletions(-) diff --git a/患者首页处理/数据可视化网页端/app/main.py b/患者首页处理/数据可视化网页端/app/main.py index 616dfbc..e91c272 100644 --- 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") diff --git a/患者首页处理/数据可视化网页端/app/static/app.js b/患者首页处理/数据可视化网页端/app/static/app.js index 0bdbe56..6bdf2df 100644 --- 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();