diff --git a/患者首页处理/数据可视化网页端/app/main.py b/患者首页处理/数据可视化网页端/app/main.py index b318d39..facb90a 100644 --- a/患者首页处理/数据可视化网页端/app/main.py +++ b/患者首页处理/数据可视化网页端/app/main.py @@ -4,14 +4,12 @@ import json import os import base64 import hashlib -import tempfile import secrets import re import threading import time import urllib.error import urllib.request -import zipfile from datetime import date, datetime, timedelta from decimal import Decimal from pathlib import Path @@ -23,7 +21,7 @@ from zoneinfo import ZoneInfo import psycopg2 import psycopg2.extras from dotenv import load_dotenv -from fastapi import BackgroundTasks, FastAPI, HTTPException, Request, Response +from fastapi import FastAPI, HTTPException, Request, Response from fastapi.responses import FileResponse, JSONResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, Field @@ -965,12 +963,6 @@ def session_from_request(request: Request) -> dict[str, Any] | None: return SESSIONS.get(token) -def require_admin_user(request: Request) -> None: - user = getattr(request.state, "user", {}) or {} - if user.get("username") != admin_username(): - raise HTTPException(status_code=403, detail="只有 admin 可以使用导出功能") - - def page_permission_for_path(path: str, method: str) -> str | tuple[str, ...] | None: if path in {"/api/status", "/api/schema"}: return None @@ -1960,22 +1952,6 @@ def get_pdf_path(source_file: str) -> Path | None: return path if path.exists() and path.is_file() else None -def safe_download_name(value: Any, fallback: str = "未命名") -> str: - text = re.sub(r'[\\/:*?"<>|\r\n]+', "_", str(value or "").strip()) - text = re.sub(r"\s+", "_", text).strip("._ ") - return text[:80] or fallback - - -def record_pdf_download_name(record: dict[str, Any]) -> str: - source_file = Path(str(record.get("source_file") or "")).name - stem = Path(source_file).stem or f"record_{record.get('id', '')}" - suffix = Path(source_file).suffix or ".pdf" - patient = safe_download_name(record.get("patient_name"), "") - inpatient_no = safe_download_name(record.get("inpatient_no") or record.get("medical_record_no"), "") - parts = [part for part in [inpatient_no, patient, stem] if part] - return "_".join(parts) + suffix - - def parse_json_content(content: str) -> Any: text = content.strip() if text.startswith("```"): @@ -2369,8 +2345,9 @@ def build_ai_prompt(record: dict[str, Any], pdf_context: dict[str, Any], privacy 6. 如果手术表格中能看到“手术及操作编码”列但对应单元格为空,写“编码栏可见但为空白”,不要写“编码区域未在首页显示”;单元格本来为空且无需补录时不要因此判为 problem。 7. 手术操作名称可能因为换行被结构化解析截断;如果 PDF定位文本或局部截图中显示完整多行名称,请把完整名称放入 suggested_updates。 8. 门急诊诊断编码只能用于 outpatient_diagnosis_code;主要诊断请修正 discharge_diagnoses 中“诊断类别=主要诊断”的行,不要把门急诊诊断编码复制到 discharge_diagnoses[].疾病编码,除非出院诊断表格对应“疾病编码”单元格本身清楚显示该编码。 -9. remaining_issues 只写当前文档复核人应该特别注意的内容;不要写如何修改,不要重复 suggested_updates,不要写“无需补录/无需处理/首页原貌”这类已判定无问题的说明。 -10. 不要编造 PDF 中看不见的内容,不要输出置信度。 +9. PDF局部截图可能因遮挡、截屏边界或缩放只显示编码/文字前半段;如果 PDF 显示值是结构化字段值的前缀,且没有相反证据,应判为 match/ok,不要写“需确认完整编码”,也不要建议把结构化字段截短。 +10. remaining_issues 只写当前文档复核人应该特别注意的内容;不要写如何修改,不要重复 suggested_updates,不要写“无需补录/无需处理/首页原貌”这类已判定无问题的说明。 +11. 不要编造 PDF 中看不见的内容,不要输出置信度。 必须返回这个 JSON 结构: {{ @@ -2538,12 +2515,90 @@ def ai_needs_review_text(value: Any) -> bool: return bool(text) and not any(marker in text for marker in AI_NO_REVIEW_MARKERS) +def ai_compact_compare_value(value: Any) -> str: + text = str(value or "").strip() + text = text.replace(".", ".").replace("X", "X").replace("x", "x").replace("×", "x") + return re.sub(r"\s+", "", text).upper() + + +def ai_pdf_value_is_structured_prefix(pdf_value: Any, structured_value: Any) -> bool: + pdf_text = ai_compact_compare_value(pdf_value) + structured_text = ai_compact_compare_value(structured_value) + if len(pdf_text) < 3 or len(structured_text) <= len(pdf_text): + return False + return structured_text.startswith(pdf_text) + + +def ai_text_pdf_prefix_issue(text: Any) -> bool: + compact = re.sub(r"\s+", "", str(text or "")) + if not compact: + return False + patterns = [ + r"PDF(?:显示|中显示|可见|值)?(?:为|是|[::])?([A-Za-z0-9][A-Za-z0-9.+\-*/xX]*)[,,;;。、]?(?:结构化字段|结构化值|字段|系统字段)(?:为|是|[::])?([A-Za-z0-9][A-Za-z0-9.+\-*/xX]*)", + r"PDF[^,,;;。]*?([A-Za-z0-9][A-Za-z0-9.+\-*/xX]{2,})[^,,;;。]*?(?:结构化字段|结构化值|字段|系统字段)[^,,;;。]*?([A-Za-z0-9][A-Za-z0-9.+\-*/xX]{2,})", + ] + return any( + ai_pdf_value_is_structured_prefix(match.group(1), match.group(2)) + for pattern in patterns + for match in re.finditer(pattern, compact, flags=re.IGNORECASE) + ) + + +def ai_pdf_prefix_truncation_issue(value: Any) -> bool: + if isinstance(value, dict): + pdf_value = value.get("pdf_value") or value.get("pdf") or value.get("PDF值") or value.get("图片值") + structured_value = ( + value.get("structured_value") + or value.get("structured") + or value.get("field_value") + or value.get("结构化值") + or value.get("结构化字段") + ) + if ai_pdf_value_is_structured_prefix(pdf_value, structured_value): + return True + return ai_text_pdf_prefix_issue(value) + return ai_text_pdf_prefix_issue(value) + + +def ai_has_nonblank_suggested_updates(parsed: dict[str, Any]) -> bool: + suggested_updates = parsed.get("suggested_updates") + if not isinstance(suggested_updates, list): + return False + return any(isinstance(item, dict) and not blank_ai_value(suggested_update_value(item)) for item in suggested_updates) + + +def ai_problem_evidence(parsed: dict[str, Any]) -> list[dict[str, Any]]: + evidence = parsed.get("evidence") + if not isinstance(evidence, list): + return [] + problem_results = {"mismatch", "uncertain", "missing", "problem", "not_match", "not ok", "not_ok"} + return [ + item + for item in evidence + if isinstance(item, dict) and str(item.get("result") or "").strip().lower() in problem_results + ] + + +def ai_only_pdf_prefix_truncation(parsed: dict[str, Any]) -> bool: + if ai_has_nonblank_suggested_updates(parsed): + return False + issues = ai_remaining_issues(parsed) + if issues: + return all(ai_pdf_prefix_truncation_issue(issue) for issue in issues) + evidence = ai_problem_evidence(parsed) + if evidence: + return all(ai_pdf_prefix_truncation_issue(item) for item in evidence) + return ai_pdf_prefix_truncation_issue(parsed.get("summary")) or ai_pdf_prefix_truncation_issue(parsed.get("decision")) + + def normalize_ai_parsed(parsed: dict[str, Any]) -> dict[str, Any]: normalized = dict(parsed) remaining = normalized.get("remaining_issues") if isinstance(remaining, list): - normalized["remaining_issues"] = [item for item in remaining if ai_needs_review_text(item)] - elif remaining and ai_needs_review_text(remaining): + normalized["remaining_issues"] = [ + item for item in remaining if ai_needs_review_text(item) and not ai_pdf_prefix_truncation_issue(item) + ] + elif remaining and ai_needs_review_text(remaining) and not ai_pdf_prefix_truncation_issue(remaining): normalized["remaining_issues"] = [remaining] else: normalized["remaining_issues"] = [] @@ -2567,7 +2622,7 @@ def ai_has_confirmed_problem(parsed: dict[str, Any]) -> bool: if not isinstance(item, dict): continue result = str(item.get("result") or "").strip().lower() - if result in {"mismatch", "uncertain", "missing", "problem"}: + if result in {"mismatch", "uncertain", "missing", "problem"} and not ai_pdf_prefix_truncation_issue(item): return True if resolution in {"false_positive", "ok", "no_issue", "误报", "无问题"}: @@ -2596,6 +2651,8 @@ def ai_remaining_issues(parsed: dict[str, Any]) -> list[Any]: def ai_has_unresolved_problem(parsed: dict[str, Any]) -> bool: + if ai_only_pdf_prefix_truncation(parsed): + return False if ai_remaining_issues(parsed): return True @@ -2630,23 +2687,21 @@ def ai_classification(parsed: dict[str, Any]) -> str: resolution = str(parsed.get("issue_resolution") or "").strip().lower() ok_values = {"ok", "pass", "passed", "no_issue", "no issue", "无问题", "通过", "已通过"} problem_values = {"problem", "not_ok", "not ok", "confirm", "不ok", "不通过", "需确认", "待确认", "需复核"} + if ai_only_pdf_prefix_truncation(parsed): + return AI_OK_STATUS if classification in ok_values or decision in ok_values or resolution in {"false_positive", "ok", "no_issue", "误报", "无问题", "通过"}: return AI_OK_STATUS if classification in problem_values or decision in problem_values or resolution in {"confirmed_problem", "uncertain", "problem", "待确认", "已证实"}: + if not ai_remaining_issues(parsed) and ai_has_nonblank_suggested_updates(parsed): + return AI_OK_STATUS return AI_PROBLEM_STATUS if ai_remaining_issues(parsed) or ai_has_unresolved_problem(parsed): return AI_PROBLEM_STATUS suggested_updates = parsed.get("suggested_updates") if isinstance(suggested_updates, list) and suggested_updates: return AI_OK_STATUS - evidence = parsed.get("evidence") - if isinstance(evidence, list): - for item in evidence: - if not isinstance(item, dict): - continue - result = str(item.get("result") or "").strip().lower() - if result in {"mismatch", "uncertain", "missing", "problem"}: - return AI_PROBLEM_STATUS + if any(not ai_pdf_prefix_truncation_issue(item) for item in ai_problem_evidence(parsed)): + return AI_PROBLEM_STATUS return AI_OK_STATUS @@ -2774,12 +2829,16 @@ def ai_suggested_updates(result: dict[str, Any], before: dict[str, Any]) -> tupl compare_old = rows[index].get(key) if ai_outpatient_code_leak(path, value, before, compare_old): continue + if ai_pdf_value_is_structured_prefix(value, compare_old): + continue if comparable(compare_old) == comparable(value): continue rows[index][key] = value new_value = rows label = f"{FIELD_META[field]['label']}[{index + 1}].{key}" + if index is None and ai_pdf_value_is_structured_prefix(value, compare_old): + continue if comparable(compare_old) == comparable(value): continue working[field] = new_value @@ -3397,77 +3456,11 @@ def list_records(q: str = "", status_filter: str = "review_all", limit: int = 30 return {"records": rows, "limit": limit, "offset": offset, "has_more": has_more} -@app.get("/api/records/export") -def export_records(request: Request, background_tasks: BackgroundTasks, q: str = "", status_filter: str = "review_all"): - require_admin_user(request) - where_sql, params = record_filter_sql(q, status_filter) - max_export = 5000 - query = sql.SQL( - """ - SELECT - id, source_file, inpatient_no, medical_record_no, patient_name, review_status, - NULLIF(review_logs->-1->>'changed_at', '')::timestamp AS last_activity_at - FROM {table} - {where} - {order_by} - LIMIT %s - """ - ).format(table=table_identifier(), where=where_sql, order_by=RECORD_LIST_ORDER_SQL) - with connect() as conn, conn.cursor() as cur: - cur.execute(query, [*params, max_export]) - rows = [row_to_json(dict(row)) for row in cur.fetchall()] - - export_items: list[tuple[dict[str, Any], Path]] = [] - for row in rows: - pdf_path = get_pdf_path(row.get("source_file") or "") - if pdf_path: - export_items.append((row, pdf_path)) - if not export_items: - raise HTTPException(status_code=404, detail="当前筛选下没有可导出的 PDF") - - handle = tempfile.NamedTemporaryFile(prefix="frontpage_export_", suffix=".zip", delete=False) - zip_path = Path(handle.name) - handle.close() - used_names: set[str] = set() - with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED, allowZip64=True) as archive: - for index, (record, pdf_path) in enumerate(export_items, start=1): - name = record_pdf_download_name(record) - arcname = f"{index:04d}_{name}" - while arcname in used_names: - arcname = f"{index:04d}_{record.get('id')}_{name}" - used_names.add(arcname) - archive.write(pdf_path, arcname) - background_tasks.add_task(zip_path.unlink, missing_ok=True) - stamp = datetime.now(APP_TIMEZONE).strftime("%Y%m%d_%H%M%S") - return FileResponse( - zip_path, - media_type="application/zip", - filename=f"患者首页影像导出_{status_filter}_{stamp}.zip", - background=background_tasks, - headers={"Cache-Control": "no-store", "X-Content-Type-Options": "nosniff"}, - ) - - @app.get("/api/records/{record_id}") def get_record(record_id: int): return {"record": fetch_record(record_id)} -@app.get("/api/records/{record_id}/export") -def export_record(record_id: int, request: Request): - require_admin_user(request) - record = fetch_record(record_id) - pdf_path = get_pdf_path(record.get("source_file") or "") - if not pdf_path: - raise HTTPException(status_code=404, detail="PDF 文件不存在") - return FileResponse( - pdf_path, - media_type="application/pdf", - filename=record_pdf_download_name(record), - headers={"Cache-Control": "no-store", "X-Content-Type-Options": "nosniff"}, - ) - - @app.get("/api/records/{record_id}/ai-question-image/{log_index}/{module_index}") def ai_question_image(record_id: int, log_index: int, module_index: int): if log_index < 0 or module_index < 0: diff --git a/患者首页处理/数据可视化网页端/app/static/app.js b/患者首页处理/数据可视化网页端/app/static/app.js index c7c4269..513901e 100644 --- a/患者首页处理/数据可视化网页端/app/static/app.js +++ b/患者首页处理/数据可视化网页端/app/static/app.js @@ -190,10 +190,6 @@ function userPermissions() { return state.currentUser?.permissions || {}; } -function isAdminUser() { - return state.currentUser?.source === "env" || state.currentUser?.username === "admin"; -} - function canOpenPage(page) { const key = page === "auditHistory" ? "audit_history" : page; return userPermissions()[key] !== false; @@ -402,9 +398,6 @@ function renderRecords() { const active = record.id === state.selectedId ? " is-active" : ""; const manual = record.manual_corrected ? " · 人工已改" : ""; const activity = record.last_activity_at ? `最近 ${escapeHtml(formatTime(record.last_activity_at))}` : ""; - const exportButton = isAdminUser() && record.has_pdf - ? `` - : ""; return `