admin/HIS
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Ignore PDF prefix truncation AI issues

Browse Source
This commit is contained in:
Codex
2026-05-27 16:06:10 +08:00
parent e0574d16a3
commit 54ea338bd6
4 changed files with 101 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files

201
患者首页处理/数据可视化网页端/app/main.py
View File

@@ -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").replace("ｘ", "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: