admin/HIS
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
e0574d16a324d63f76f19c2fa4dfa45a860ad108
HIS/患者首页处理/数据可视化网页端/app/main.py
Codex e0574d16a3 Sync primary diagnosis from discharge table
2026-05-27 15:42:49 +08:00

3889 lines
153 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
from __future__ import annotations
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
from queue import Empty, Queue
from typing import Any
from urllib.parse import urlparse
from zoneinfo import ZoneInfo
import psycopg2
import psycopg2.extras
from dotenv import load_dotenv
from fastapi import BackgroundTasks, FastAPI, HTTPException, Request, Response
from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, Field
from psycopg2 import sql
PROJECT_ROOT = Path(__file__).resolve().parents[2]
load_dotenv(PROJECT_ROOT / ".env")
APP_DIR = Path(__file__).resolve().parent
STATIC_DIR = APP_DIR / "static"
def env(name: str, default: str = "") -> str:
return os.getenv(name, default).strip()
DB_CONFIG = {
"host": env("PGHOST"),
"port": int(env("PGPORT", "5432")),
"dbname": env("PGDATABASE"),
"user": env("PGUSER"),
"password": env("PGPASSWORD"),
}
PGTABLE = env("PGTABLE")
PDF_DIR = Path(env("PDF_DIR", str(PROJECT_ROOT / "待处理-患者首页PDF"))).resolve()
SETTINGS_PATH = Path(env("REVIEW_SETTINGS_PATH", str(PROJECT_ROOT / "数据可视化网页端/review_settings.local.json"))).resolve()
APP_TIMEZONE = ZoneInfo(env("APP_TIMEZONE", "Asia/Shanghai") or "Asia/Shanghai")
DEFAULT_MAJOR_DEPARTMENT_OPTIONS = [
"肝胆外科及肝移植相关",
"普通外科及腹部外科",
"急诊医学科",
"重症医学科",
"骨科",
"泌尿外科",
"呼吸内科",
"耳鼻喉头颈外科",
"日间诊疗中心",
"乳腺外科",
"胸外科",
"肝病内科",
"感染科",
"肿瘤放疗科",
"消化内科",
"肿瘤外科",
"特需/涉外病房",
"老年外科",
]
def load_major_department_options() -> list[str]:
rule_path = PROJECT_ROOT / "数据处理工作区" / "01_配置规则" / "01_科室分类规则.json"
if rule_path.exists():
try:
data = json.loads(rule_path.read_text(encoding="utf-8"))
options = [str(group.get("大科室", "")).strip() for group in data.get("大科室列表", [])]
options = [option for option in options if option]
if options:
return options
except Exception: # noqa: BLE001
pass
return DEFAULT_MAJOR_DEPARTMENT_OPTIONS
MAJOR_DEPARTMENT_OPTIONS = load_major_department_options()
FIELD_GROUPS: list[dict[str, Any]] = [
{
"name": "基本信息",
"fields": [
("inpatient_no", "住院号", "text", None),
("medical_record_no", "病案号", "text", None),
("front_page_medical_record_no", "首页病案号", "text", None),
("patient_name", "姓名", "text", None),
("gender", "性别", "text", None),
("birth_date", "出生日期", "date", None),
("age", "年龄", "text", None),
("nationality", "国籍", "text", None),
("id_card_no", "身份证号", "text", None),
("payment_method", "医疗付费方式", "text", None),
("health_card_no", "健康卡号", "text", None),
("admission_count", "住院次数", "integer", None),
("occupation", "职业", "text", None),
("marital_status_code", "婚姻代码", "text", None),
("admission_path_code", "入院途径代码", "text", None),
("admission_time", "入院时间", "datetime", None),
("admission_dept", "入院科别", "text", None),
("admission_ward", "入院病房", "text", None),
("transfer_dept", "转科科别", "text", None),
("transfer_time", "转科时间", "text", None),
("discharge_time", "出院时间", "datetime", None),
("discharge_dept", "出院科别", "text", None),
("discharge_ward", "出院病房", "text", None),
("hospital_days", "实际住院天数", "integer", None),
("major_department", "大科室", "select", MAJOR_DEPARTMENT_OPTIONS),
],
},
{
"name": "地址联系人",
"fields": [
("current_address", "现住址", "text", None),
("current_address_phone", "现住址电话", "text", None),
("current_address_postcode", "现住址邮编", "text", None),
("household_address", "户口地址", "text", None),
("household_postcode", "户口地址邮编", "text", None),
("employer_address", "工作单位及地址", "text", None),
("employer_phone", "单位电话", "text", None),
("employer_postcode", "单位邮编", "text", None),
("contact_name", "联系人姓名", "text", None),
("contact_relationship", "联系人关系", "text", None),
("contact_address", "联系人地址", "text", None),
("contact_phone", "联系人电话", "text", None),
],
},
{
"name": "诊断表格",
"fields": [
("outpatient_diagnosis", "门急诊诊断", "text", None),
("outpatient_diagnosis_code", "门急诊诊断编码", "text", None),
("discharge_diagnoses", "出院诊断", "json", None),
("injury_poisoning_external_cause", "损伤中毒外部原因", "text", None),
("injury_poisoning_code", "损伤中毒疾病编码", "text", None),
("pathology_diagnosis", "病理诊断", "text", None),
("pathology_diagnosis_code", "病理诊断编码", "text", None),
("pathology_no", "病理号", "text", None),
],
},
{
"name": "手术表格",
"fields": [
("operations", "手术操作 JSON", "json", None),
],
},
{
"name": "离院费用",
"fields": [
("discharge_disposition_code", "离院方式代码", "text", None),
("receiving_org_name", "拟接收医疗机构名称", "text", None),
("readmission_plan_code", "出院31天内再住院计划代码", "text", None),
("readmission_plan_purpose", "再住院计划目的", "text", None),
("coma_before_days", "入院前昏迷天数", "integer", None),
("coma_before_hours", "入院前昏迷小时", "integer", None),
("coma_before_minutes", "入院前昏迷分钟", "integer", None),
("coma_after_days", "入院后昏迷天数", "integer", None),
("coma_after_hours", "入院后昏迷小时", "integer", None),
("coma_after_minutes", "入院后昏迷分钟", "integer", None),
("total_cost", "总费用", "numeric", None),
("self_pay_amount", "自付金额", "numeric", None),
("fee_details", "费用明细 JSON", "json", None),
],
},
]
FIELD_META: dict[str, dict[str, Any]] = {}
for group in FIELD_GROUPS:
for name, label, field_type, options in group["fields"]:
FIELD_META[name] = {"name": name, "label": label, "type": field_type, "options": options}
EDITABLE_FIELDS = set(FIELD_META)
JSON_FIELDS = {name for name, meta in FIELD_META.items() if meta["type"] == "json"}
JSON_DB_FIELDS = JSON_FIELDS | {"review_notes", "quality_notes", "auto_corrections"}
INTEGER_FIELDS = {name for name, meta in FIELD_META.items() if meta["type"] == "integer"}
NUMERIC_FIELDS = {name for name, meta in FIELD_META.items() if meta["type"] == "numeric"}
class UpdatePayload(BaseModel):
fields: dict[str, Any]
manual_note: str = ""
note_prefix: str = "人工复核"
class AuditPayload(BaseModel):
audit_status: str = "pending"
audit_notes: str = ""
ai_result: Any = None
class AuditClassifyPayload(BaseModel):
record_id: int
audit_source: str = "reviewed"
audit_status: str
audit_notes: str = ""
fields: dict[str, Any] = {}
class UserPayload(BaseModel):
username: str
password: str = ""
permissions: dict[str, bool] = {}
class UserUpdatePayload(BaseModel):
username: str = ""
password: str = ""
permissions: dict[str, bool] = {}
class PasswordPayload(BaseModel):
password: str = ""
class PermissionPayload(BaseModel):
permissions: dict[str, bool] = {}
class LoginPayload(BaseModel):
username: str = ""
password: str = ""
class SystemSettingsPayload(BaseModel):
status_check_time: str = ""
class KimiSettingsPayload(BaseModel):
enabled: bool = True
model: str = ""
api_base: str = ""
api_key: str = ""
concurrency: int = 3
thinking_enabled: bool = False
ai_scope_mode: str = "all"
ai_action_modes: dict[str, str] = Field(default_factory=dict)
ai_action_privacy_modes: dict[str, bool] = Field(default_factory=dict)
class AiReviewPayload(BaseModel):
scope: str = "current"
record_id: int | None = None
model: str = ""
thinking_enabled: bool | None = None
privacy_mode: bool | None = None
app = FastAPI(title="Patient Front Page Visual Review")
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
STATUS_CHECK_LOCK = threading.Lock()
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": "",
"total": 0,
"processed": 0,
"ok": 0,
"pending": 0,
"failed": 0,
"concurrency": 0,
"message": "",
"errors": [],
"started_at": "",
"finished_at": "",
"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"
AI_OK_STATUS = "auto_pass"
AI_PROBLEM_STATUS = "AI已处理-不OK"
AI_NO_ISSUE_STATUS = AI_OK_STATUS
AI_PENDING_STATUS = AI_PROBLEM_STATUS
LEGACY_AI_OK_STATUS = "AI复核-无问题"
LEGACY_AI_PENDING_STATUS = "AI复核-待确认"
AI_CONFIRMED_PROBLEM_KEYWORDS = (
"缺少",
"缺失",
"为空白",
"为空",
"空白",
"无编码",
"未填写",
"不清晰",
"不一致",
"错位",
"混乱",
"需人工",
"需要人工",
"待确认",
)
AI_CONFIRMED_PROBLEM_QUALIFIERS = ("确实", "证实", "属实", "仍", "依然", "存在", "需要", "需人工", "待确认")
AI_UNRESOLVED_PROBLEM_MARKERS = (
"需人工",
"需要人工",
"待确认",
"需确认",
"需补录",
"需补充",
"建议补录",
"建议补充",
"漏填",
"缺失需",
"空白需",
"截断需",
"需补全",
"不清晰",
"不一致",
)
AI_FORCE_PROBLEM_MARKERS = (
"需人工",
"需要人工",
"待确认",
"需确认",
"需补录",
"需补充",
"建议补录",
"建议补充",
"漏填",
"缺失需",
"空白需",
"截断需",
"需补全",
)
AI_FIXED_MARKERS = ("已修正", "已补齐", "已更正", "已改为", "已填入", "已补全")
AI_NO_REVIEW_MARKERS = ("无需", "无须", "不需", "不用", "无需补录", "无须补录", "无需处理", "原貌")
AI_STOP_ERROR_MARKERS = (
"exceeded_current_quota_error",
"insufficient_quota",
"consumption budget",
"billing details",
"quota",
"余额",
"额度",
)
AI_JOB_ERROR_LIMIT = 50
AI_SAFE_MODULE_NAMES = {"诊断表格", "手术表格", "离院费用"}
AI_REDACTED = "[已脱敏]"
AI_REDACT_PATTERNS = (
(re.compile(r"ZY\d{6,}", re.IGNORECASE), "[住院号]"),
(re.compile(r"\b\d{15,17}[\dXx]\b"), "[身份证号]"),
(re.compile(r"(?<![A-Za-z.])\b\d{7,14}\b(?![A-Za-z.])"), "[编号]"),
(re.compile(r"(身份证号|身份证|电话|手机号|联系电话|邮编)\s*[:：]?\s*[\dXx\- ]{5,}"), r"\1：[已脱敏]"),
(re.compile(r"(姓名|患者姓名|联系人姓名)\s*[:：]?\s*[\u4e00-\u9fff·]{2,6}"), r"\1：[已脱敏]"),
(re.compile(r"(现住址|户口地址|联系人地址|工作单位及地址|出生地|籍贯)\s*[:：]?\s*[^\n；;]{2,80}"), r"\1：[已脱敏]"),
)
AI_UPLOAD_FIELDS = [
"outpatient_diagnosis",
"outpatient_diagnosis_code",
"discharge_diagnoses",
"operations",
"pathology_diagnosis",
"pathology_diagnosis_code",
"total_cost",
"self_pay_amount",
"quality_status",
"quality_notes",
"review_notes",
]
PDF_MODULE_DEFINITIONS: list[dict[str, Any]] = [
{
"name": "基本信息",
"keywords": ["住院病案首页", "姓名", "性别", "出生日期", "入院时间", "出院时间", "住院号"],
"note_keywords": ["姓名", "性别", "出生", "入院", "出院", "住院号", "病案号", "年龄"],
"tail": 330,
},
{
"name": "地址联系人",
"keywords": ["现住址", "户口地址", "工作单位及地址", "联系人姓名", "联系人地址"],
"note_keywords": ["地址", "电话", "联系人", "户口", "单位", "邮编"],
"tail": 260,
},
{
"name": "诊断表格",
"keywords": ["门（急）诊诊断", "门急诊诊断", "出院诊断", "疾病编码", "入院病情", "其他诊断"],
"note_keywords": ["诊断", "疾病编码", "编码格式", "入院病情", "病理"],
"tail": 520,
},
{
"name": "手术表格",
"keywords": ["手术及操作编码", "手术及操作日期", "手术及操作名称", "手术级别", "麻醉方式", "术者"],
"note_keywords": ["手术", "操作", "麻醉", "切口", "术者"],
"tail": 520,
},
{
"name": "离院费用",
"keywords": ["离院方式", "出院31天内再住院计划", "住院费用", "总费用", "自付金额", "综合医疗服务类"],
"note_keywords": ["离院", "费用", "总费用", "自付", "金额", "再住院", "昏迷"],
"tail": 520,
},
]
AI_REVIEWABLE_STATUSES = ("needs_review", AI_PENDING_STATUS)
SUBMITTED_STATUS = "已提交"
@app.on_event("startup")
def start_status_scheduler() -> None:
threading.Thread(target=status_scheduler_loop, name="status-check-scheduler", daemon=True).start()
def table_identifier() -> sql.Composable:
if "." in PGTABLE:
schema, table = PGTABLE.split(".", 1)
return sql.Identifier(schema, table)
return sql.Identifier(PGTABLE)
def patient_lists_identifier() -> sql.Composable:
return sql.Identifier("Patient_Lists")
def patient_list_trigger_function_identifier(base_table: str) -> sql.Composable:
function_name = f"{base_table}_sync_patient_lists_trigger_fn"
if "." in PGTABLE:
schema = PGTABLE.split(".", 1)[0]
return sql.Identifier(schema, function_name)
return sql.Identifier(function_name)
def patient_dedupe_trigger_function_identifier(base_table: str) -> sql.Composable:
function_name = f"{base_table}_dedupe_inpatient_no_trigger_fn"
if "." in PGTABLE:
schema = PGTABLE.split(".", 1)[0]
return sql.Identifier(schema, function_name)
return sql.Identifier(function_name)
def related_table_identifier(suffix: str) -> sql.Composable:
if "." in PGTABLE:
schema, table = PGTABLE.split(".", 1)
return sql.Identifier(schema, f"{table}{suffix}")
return sql.Identifier(f"{PGTABLE}{suffix}")
def connect():
missing = [name for name, value in {
"PGHOST": DB_CONFIG["host"],
"PGDATABASE": DB_CONFIG["dbname"],
"PGUSER": DB_CONFIG["user"],
"PGPASSWORD": DB_CONFIG["password"],
"PGTABLE": PGTABLE,
}.items() if not value]
if missing:
raise RuntimeError("缺少 PostgreSQL 连接配置：" + "、".join(missing))
return psycopg2.connect(**DB_CONFIG, cursor_factory=psycopg2.extras.RealDictCursor)
def json_ready(value: Any) -> Any:
if isinstance(value, (date, datetime)):
return value.isoformat(sep=" ") if isinstance(value, datetime) else value.isoformat()
if isinstance(value, Decimal):
return str(value)
return value
def row_to_json(row: dict[str, Any]) -> dict[str, Any]:
return {key: json_ready(value) for key, value in row.items()}
def json_ready_deep(value: Any) -> Any:
if isinstance(value, dict):
return {key: json_ready_deep(item) for key, item in value.items()}
if isinstance(value, list):
return [json_ready_deep(item) for item in value]
return json_ready(value)
def comparable(value: Any) -> str:
return json.dumps(json_ready_deep(value), ensure_ascii=False, sort_keys=True, default=str)
PERMISSION_LABELS = {
"overview": "概览",
"review": "复核",
"audit": "抽查",
"audit_history": "抽查一览",
"settings": "设置",
}
DEFAULT_PERMISSIONS = {key: True for key in PERMISSION_LABELS}
SESSION_COOKIE = "frontpage_review_session"
SESSIONS: dict[str, dict[str, Any]] = {}
def password_hash(password: str) -> dict[str, str]:
salt = secrets.token_hex(12)
digest = hashlib.sha256((salt + password).encode("utf-8")).hexdigest()
return {"salt": salt, "password_hash": digest}
def verify_password(password: str, salt: str, digest: str) -> bool:
expected = hashlib.sha256((salt + password).encode("utf-8")).hexdigest()
return secrets.compare_digest(expected, digest or "")
def admin_username() -> str:
return env("REVIEW_ADMIN_USER", "admin") or "admin"
def admin_password() -> str:
return env("REVIEW_ADMIN_PASSWORD", "change-me") or "change-me"
def public_user(username: str, permissions: dict[str, bool], source: str) -> dict[str, Any]:
return {
"username": username,
"permissions": {**DEFAULT_PERMISSIONS, **(permissions or {})},
"source": source,
}
def load_local_settings() -> dict[str, Any]:
if not SETTINGS_PATH.exists():
return {
"users": [],
"permission_labels": PERMISSION_LABELS,
"system": default_system_settings(),
"kimi": default_kimi_settings(),
"status_snapshot": default_status_snapshot(),
}
try:
data = json.loads(SETTINGS_PATH.read_text(encoding="utf-8"))
except json.JSONDecodeError:
data = {
"users": [],
"permission_labels": PERMISSION_LABELS,
"system": default_system_settings(),
"kimi": default_kimi_settings(),
"status_snapshot": default_status_snapshot(),
}
data.setdefault("users", [])
data.setdefault("permission_labels", PERMISSION_LABELS)
data["system"] = normalize_system_settings(data.get("system") or {})
data["kimi"] = normalize_kimi_settings(data.get("kimi") or {})
data.setdefault("status_snapshot", default_status_snapshot())
return data
def save_local_settings(data: dict[str, Any]) -> None:
SETTINGS_PATH.parent.mkdir(parents=True, exist_ok=True)
temp = SETTINGS_PATH.with_suffix(SETTINGS_PATH.suffix + ".tmp")
temp.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
temp.replace(SETTINGS_PATH)
def public_settings() -> dict[str, Any]:
data = load_local_settings()
users = []
env_admin_seen = False
for user in data.get("users", []):
users.append(
{
"username": user.get("username", ""),
"permissions": {**DEFAULT_PERMISSIONS, **(user.get("permissions") or {})},
"source": "local",
"created_at": user.get("created_at", ""),
"updated_at": user.get("updated_at", ""),
"has_password": bool(user.get("password_hash")),
}
)
if user.get("username") == admin_username():
env_admin_seen = True
if not env_admin_seen:
users.insert(
0,
{
"username": admin_username(),
"permissions": dict(DEFAULT_PERMISSIONS),
"source": "env",
"created_at": "",
"updated_at": "",
"has_password": True,
},
)
return {
"users": users,
"permission_labels": PERMISSION_LABELS,
"system": normalize_system_settings(data.get("system") or {}),
"kimi": public_kimi_settings(data.get("kimi") or {}),
"status_snapshot": data.get("status_snapshot") or default_status_snapshot(),
}
def normalize_status_check_time(value: str) -> str:
value = (value or DEFAULT_STATUS_CHECK_TIME).strip()
match = re.fullmatch(r"([01]?\d|2[0-3]):([0-5]\d)", value)
if not match:
raise HTTPException(status_code=400, detail="状态检查时间必须是 HH:MM，例如 03:00")
return f"{int(match.group(1)):02d}:{match.group(2)}"
def kimi_api_key(kimi: dict[str, Any] | None = None) -> str:
local_key = str((kimi or {}).get("api_key") or "").strip()
if local_key:
return local_key
return env("MOONSHOT_API_KEY") or env("KIMI_API_KEY")
def normalize_kimi_concurrency(value: Any) -> int:
try:
concurrency = int(value)
except (TypeError, ValueError):
concurrency = int(env("KIMI_CONCURRENCY", "3") or 3)
return max(1, min(concurrency, 6))
def normalize_bool(value: Any, default: bool = False) -> bool:
if isinstance(value, bool):
return value
if value is None:
return default
return str(value).strip().lower() in {"1", "true", "yes", "on", "启用", "是"}
def normalize_kimi_ai_scope_mode(value: Any, default: str = "all") -> str:
aliases = {
"off": "off",
"none": "off",
"disabled": "off",
"关闭": "off",
"current": "current",
"single": "current",
"仅当前项": "current",
"five": "five",
"5": "five",
"current_five": "five",
"当前项和后5项": "five",
"all": "all",
"全部": "all",
}
mode = aliases.get(str(value or "").strip().lower(), "")
return mode or (default if default in {"off", "current", "five", "all"} else "all")
def normalize_ai_action_mode(value: Any, default: str = "default") -> str:
aliases = {
"off": "off",
"关闭": "off",
"disabled": "off",
"default": "default",
"follow": "default",
"跟随默认模型": "default",
"k25": "k25",
"kimi-k2.5": "k25",
"k25_thinking": "k25_thinking",
"kimi-k2.5-thinking": "k25_thinking",
"k26": "k26",
"kimi-k2.6": "k26",
"k26_thinking": "k26_thinking",
"kimi-k2.6-thinking": "k26_thinking",
}
mode = aliases.get(str(value or "").strip().lower(), "")
return mode or (default if default in {"off", "default", "k25", "k25_thinking", "k26", "k26_thinking"} else "default")
def default_ai_action_modes() -> dict[str, str]:
return {"current": "default", "five": "default", "all": "default"}
def normalize_ai_action_modes(value: Any) -> dict[str, str]:
defaults = default_ai_action_modes()
if not isinstance(value, dict):
return defaults
return {scope: normalize_ai_action_mode(value.get(scope), defaults[scope]) for scope in defaults}
def default_ai_action_privacy_modes() -> dict[str, bool]:
return {"current": True, "five": True, "all": True}
def normalize_ai_action_privacy_modes(value: Any) -> dict[str, bool]:
defaults = default_ai_action_privacy_modes()
if not isinstance(value, dict):
return defaults
return {scope: normalize_bool(value.get(scope), defaults[scope]) for scope in defaults}
def ai_action_mode_to_override(mode: str) -> dict[str, Any]:
normalized = normalize_ai_action_mode(mode)
if normalized == "k25":
return {"model": "kimi-k2.5", "thinking_enabled": False}
if normalized == "k25_thinking":
return {"model": "kimi-k2.5", "thinking_enabled": True}
if normalized == "k26":
return {"model": "kimi-k2.6", "thinking_enabled": False}
if normalized == "k26_thinking":
return {"model": "kimi-k2.6", "thinking_enabled": True}
return {}
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", "privacy_blocked"}
if mode == "five":
return scope in {"current", "five"}
if mode == "current":
return scope == "current"
return False
def default_kimi_settings() -> dict[str, Any]:
return {
"enabled": bool(kimi_api_key()),
"api_base": DEFAULT_KIMI_API_BASE,
"api_key": "",
"model": DEFAULT_KIMI_MODEL,
"concurrency": normalize_kimi_concurrency(env("KIMI_CONCURRENCY", "3")),
"thinking_enabled": normalize_bool(env("KIMI_THINKING_ENABLED", "false")),
"ai_scope_mode": normalize_kimi_ai_scope_mode(env("KIMI_AI_SCOPE_MODE", "all")),
"ai_action_modes": default_ai_action_modes(),
"ai_action_privacy_modes": default_ai_action_privacy_modes(),
}
def normalize_kimi_settings(kimi: dict[str, Any]) -> dict[str, Any]:
defaults = default_kimi_settings()
api_base = str(kimi.get("api_base") or defaults["api_base"]).strip().rstrip("/")
model = str(kimi.get("model") or defaults["model"]).strip()
return {
"enabled": bool(kimi.get("enabled", defaults["enabled"])),
"api_base": api_base or DEFAULT_KIMI_API_BASE,
"api_key": str(kimi.get("api_key") or defaults.get("api_key") or "").strip(),
"model": model or DEFAULT_KIMI_MODEL,
"concurrency": normalize_kimi_concurrency(kimi.get("concurrency", defaults["concurrency"])),
"thinking_enabled": normalize_bool(kimi.get("thinking_enabled"), defaults["thinking_enabled"]),
"ai_scope_mode": normalize_kimi_ai_scope_mode(kimi.get("ai_scope_mode"), defaults["ai_scope_mode"]),
"ai_action_modes": normalize_ai_action_modes(kimi.get("ai_action_modes") or defaults["ai_action_modes"]),
"ai_action_privacy_modes": normalize_ai_action_privacy_modes(
kimi.get("ai_action_privacy_modes") or defaults["ai_action_privacy_modes"]
),
}
def public_kimi_settings(kimi: dict[str, Any] | None = None) -> dict[str, Any]:
settings = normalize_kimi_settings(kimi or {})
local_key_configured = bool(settings.get("api_key"))
env_key_configured = bool(kimi_api_key())
settings.pop("api_key", None)
settings["api_key_configured"] = local_key_configured or env_key_configured
settings["api_key_source"] = "settings" if local_key_configured else "env" if env_key_configured else ""
settings["available"] = settings["enabled"] and settings["api_key_configured"]
return settings
def model_supports_thinking(model: str) -> bool:
return str(model or "").startswith("kimi-k2.")
def status_now() -> datetime:
return datetime.now(APP_TIMEZONE)
def default_system_settings() -> dict[str, Any]:
return {
"status_check_time": normalize_status_check_time(DEFAULT_STATUS_CHECK_TIME),
"last_status_check_date": "",
"last_status_checked_at": "",
}
def normalize_system_settings(system: dict[str, Any]) -> dict[str, Any]:
defaults = default_system_settings()
merged = {**defaults, **(system or {})}
merged["status_check_time"] = normalize_status_check_time(str(merged.get("status_check_time") or defaults["status_check_time"]))
return merged
def next_status_check_at(system: dict[str, Any] | None = None, now: datetime | None = None) -> str:
now = now or status_now()
system = normalize_system_settings(system or {})
hour, minute = [int(part) for part in system["status_check_time"].split(":")]
next_run = now.replace(hour=hour, minute=minute, second=0, microsecond=0)
if next_run <= now:
next_run += timedelta(days=1)
return next_run.isoformat(timespec="seconds")
def default_status_snapshot() -> dict[str, Any]:
system = default_system_settings()
return {
"database": "unchecked",
"host": DB_CONFIG["host"],
"port": DB_CONFIG["port"],
"database_name": DB_CONFIG["dbname"],
"table": PGTABLE,
"pdf_dir": str(PDF_DIR),
"pdf_count": None,
"total": None,
"workbench_total": None,
"review_needed": None,
"needs_review": None,
"auto_passed": None,
"ai_passed": None,
"ai_pending": None,
"reviewed": None,
"submitted": None,
"manual_corrected": None,
"audit_total": None,
"message": "尚未执行状态检查",
"checked_at": "",
"check_source": "",
"next_check_at": next_status_check_at(system),
}
def compute_status_snapshot(source: str = "manual") -> dict[str, Any]:
result: dict[str, Any] = {
"database": "offline",
"host": DB_CONFIG["host"],
"port": DB_CONFIG["port"],
"database_name": DB_CONFIG["dbname"],
"table": PGTABLE,
"pdf_dir": str(PDF_DIR),
"pdf_count": len(list(PDF_DIR.glob("*.pdf"))) if PDF_DIR.exists() else 0,
"checked_at": status_now().isoformat(timespec="seconds"),
"check_source": source,
}
try:
query = sql.SQL(
"""
SELECT
count(*) AS total,
count(*) FILTER (WHERE review_status IN ('needs_review', 'reviewed', 'AI已处理-OK', 'AI已处理-不OK', 'AI复核-无问题', 'AI复核-待确认') OR (
review_status = 'auto_pass'
AND COALESCE(review_logs, '[]'::jsonb) @> jsonb_build_array(jsonb_build_object('changed_by', 'AI'))
)) AS workbench_total,
count(*) FILTER (WHERE review_status IN ('needs_review', 'AI已处理-不OK', 'AI复核-待确认')) AS review_needed,
count(*) FILTER (WHERE review_status = 'needs_review') AS needs_review,
count(*) FILTER (WHERE review_status = 'auto_pass') AS auto_passed,
count(*) FILTER (WHERE (
review_status = 'auto_pass'
AND COALESCE(review_logs, '[]'::jsonb) @> jsonb_build_array(jsonb_build_object('changed_by', 'AI'))
) OR review_status IN ('AI已处理-OK', 'AI复核-无问题')) AS ai_passed,
count(*) FILTER (WHERE review_status IN ('AI已处理-不OK', 'AI复核-待确认')) AS ai_pending,
count(*) FILTER (WHERE review_status = 'reviewed') AS reviewed,
count(*) FILTER (WHERE review_status = '已提交') AS submitted,
count(*) FILTER (WHERE manual_corrected IS TRUE) AS manual_corrected
FROM {table}
"""
).format(table=table_identifier())
with connect() as conn, conn.cursor() as cur:
cur.execute(query)
row = cur.fetchone()
result.update(row_to_json(dict(row)))
result["audit_total"] = None
result["database"] = "online"
result["message"] = "连接正常"
except Exception as exc: # noqa: BLE001
result["message"] = str(exc)
return result
def refresh_status_snapshot(source: str = "manual") -> dict[str, Any]:
with STATUS_CHECK_LOCK:
snapshot = compute_status_snapshot(source=source)
data = load_local_settings()
system = normalize_system_settings(data.get("system") or {})
now = status_now()
if source == "scheduled":
system["last_status_check_date"] = now.date().isoformat()
system["last_status_checked_at"] = snapshot.get("checked_at", now.isoformat(timespec="seconds"))
snapshot["next_check_at"] = next_status_check_at(system, now)
data["system"] = system
data["status_snapshot"] = snapshot
save_local_settings(data)
return snapshot
def status_scheduler_loop() -> None:
while True:
try:
data = load_local_settings()
system = normalize_system_settings(data.get("system") or {})
now = status_now()
hour, minute = [int(part) for part in system["status_check_time"].split(":")]
due_time = now.replace(hour=hour, minute=minute, second=0, microsecond=0)
already_ran = system.get("last_status_check_date") == now.date().isoformat()
if now >= due_time and not already_ran:
refresh_status_snapshot(source="scheduled")
except Exception:
pass
time.sleep(60)
def clean_permissions(permissions: dict[str, bool]) -> dict[str, bool]:
return {key: bool(permissions.get(key, DEFAULT_PERMISSIONS[key])) for key in PERMISSION_LABELS}
def local_user_index(data: dict[str, Any], username: str) -> int | None:
for index, user in enumerate(data.get("users", [])):
if user.get("username") == username:
return index
return None
def validate_local_username(username: str, data: dict[str, Any], current_username: str = "") -> str:
username = username.strip()
if not username:
raise HTTPException(status_code=400, detail="用户名不能为空")
if username == admin_username() and username != current_username:
raise HTTPException(status_code=400, detail="不能覆盖环境变量管理员")
for user in data.get("users", []):
if user.get("username") == username and user.get("username") != current_username:
raise HTTPException(status_code=400, detail="用户已存在")
return username
def authenticate_user(username: str, password: str) -> dict[str, Any] | None:
username = username.strip()
if username == admin_username():
if secrets.compare_digest(password, admin_password()):
return public_user(username, DEFAULT_PERMISSIONS, "env")
return None
data = load_local_settings()
for user in data.get("users", []):
if user.get("username") != username:
continue
if not user.get("password_hash") or not user.get("salt"):
return None
if verify_password(password, user.get("salt", ""), user.get("password_hash", "")):
return public_user(username, clean_permissions(user.get("permissions") or {}), "local")
return None
return None
def session_from_request(request: Request) -> dict[str, Any] | None:
token = request.cookies.get(SESSION_COOKIE, "")
if not token:
return 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
if path.startswith("/api/settings"):
return "settings"
if path.startswith("/api/overview"):
return "overview"
if path.startswith("/api/audit/logs") and method == "GET":
return "audit_history"
if path.startswith("/api/audit"):
return "audit"
if path.startswith("/api/ai"):
return "review"
if path.startswith("/api/pdf/"):
return ("review", "audit")
if path == "/api/records":
return "review"
if path.startswith("/api/records/"):
return "review" if method != "GET" else ("review", "audit")
return None
def has_page_permission(user: dict[str, Any], requirement: str | tuple[str, ...] | None) -> bool:
if requirement is None:
return True
permissions = user.get("permissions") or {}
if isinstance(requirement, tuple):
return any(permissions.get(item, False) for item in requirement)
return bool(permissions.get(requirement, False))
@app.middleware("http")
async def auth_middleware(request: Request, call_next):
path = request.url.path
if path.startswith("/api/") and not path.startswith("/api/auth/"):
user = session_from_request(request)
if not user:
return JSONResponse({"detail": "请先登录"}, status_code=401)
requirement = page_permission_for_path(path, request.method)
if not has_page_permission(user, requirement):
return JSONResponse({"detail": "当前用户没有访问权限"}, status_code=403)
request.state.user = user
return await call_next(request)
def digits(value: Any, width: int) -> str:
text = re.sub(r"\D", "", str(value or ""))
return text[-width:].zfill(width) if text else ""
def source_file_inpatient_no(source_file: str) -> str:
match = re.match(r"^(ZY\d{12})", Path(source_file or "").stem, flags=re.IGNORECASE)
return match.group(1).upper() if match else ""
def source_file_admission_count(source_file: str) -> str:
match = re.match(r"^ZY(\d{2})\d{10}", Path(source_file or "").stem, flags=re.IGNORECASE)
return match.group(1) if match else ""
def source_file_medical_record_no(source_file: str) -> str:
match = re.match(r"^ZY\d{2}(\d{10})", Path(source_file or "").stem, flags=re.IGNORECASE)
return match.group(1) if match else ""
def build_inpatient_no_from_record(record: dict[str, Any]) -> str:
source_file = str(record.get("source_file") or "")
admission = digits(record.get("admission_count"), 2) or source_file_admission_count(source_file)
page_no = (
digits(record.get("front_page_medical_record_no"), 10)
or digits(record.get("medical_record_no"), 10)
or source_file_medical_record_no(source_file)
)
if admission and page_no:
return f"ZY{admission}{page_no}"
return source_file_inpatient_no(source_file)
def ensure_workflow_tables(force: bool = False) -> None:
global WORKFLOW_READY
if WORKFLOW_READY and not force:
return
with WORKFLOW_LOCK:
if WORKFLOW_READY and not force:
return
_ensure_workflow_tables_uncached()
WORKFLOW_READY = True
def _ensure_workflow_tables_uncached() -> None:
table = table_identifier()
old_review_logs = related_table_identifier("_review_logs")
old_audit_logs = related_table_identifier("_audit_logs")
schema = PGTABLE.split(".", 1)[0] if "." in PGTABLE else "public"
base_table = PGTABLE.split(".", 1)[-1]
old_review_regclass = f'{schema}."{base_table}_review_logs"'
old_audit_regclass = f'{schema}."{base_table}_audit_logs"'
with connect() as conn, conn.cursor() as cur:
cur.execute("SELECT pg_advisory_xact_lock(hashtext(%s))", (f"{PGTABLE}:workflow_storage",))
cur.execute(sql.SQL("ALTER TABLE {table} ADD COLUMN IF NOT EXISTS inpatient_no TEXT").format(table=table))
cur.execute(sql.SQL("ALTER TABLE {table} ADD COLUMN IF NOT EXISTS major_department TEXT").format(table=table))
cur.execute(
sql.SQL("ALTER TABLE {table} ADD COLUMN IF NOT EXISTS review_logs JSONB NOT NULL DEFAULT '[]'::jsonb").format(
table=table
)
)
cur.execute(
sql.SQL("ALTER TABLE {table} ADD COLUMN IF NOT EXISTS audit_logs JSONB NOT NULL DEFAULT '[]'::jsonb").format(
table=table
)
)
cur.execute(
sql.SQL("COMMENT ON COLUMN {table}.review_logs IS '人工复核修改记录，JSONB数组，已合并到患者首页主表'").format(
table=table
)
)
cur.execute(
sql.SQL("COMMENT ON COLUMN {table}.audit_logs IS '抽查归类记录，JSONB数组，已合并到患者首页主表'").format(
table=table
)
)
cur.execute(
sql.SQL("COMMENT ON COLUMN {table}.inpatient_no IS '患者号/住院号，作为首页与患者列表联动唯一键；不能为空，格式由患者目录核验端处理。'").format(
table=table
)
)
cur.execute(
sql.SQL("COMMENT ON COLUMN {table}.major_department IS '大科室分类，来源于01_科室分类规则.json。'").format(
table=table
)
)
cur.execute(
sql.SQL(
"""
UPDATE {table}
SET front_page_medical_record_no = RIGHT(LPAD(regexp_replace(front_page_medical_record_no, '\\D', '', 'g'), 10, '0'), 10)
WHERE front_page_medical_record_no IS NOT NULL
AND front_page_medical_record_no <> ''
AND front_page_medical_record_no !~ '^\\d{{10}}$'
AND regexp_replace(front_page_medical_record_no, '\\D', '', 'g') <> ''
"""
).format(table=table)
)
cur.execute(
sql.SQL(
"""
UPDATE {table}
SET inpatient_no =
'ZY'
|| COALESCE(
LPAD(admission_count::text, 2, '0'),
substring(source_file from '^ZY([0-9]{{2}})[0-9]{{10}}')
)
|| RIGHT(
LPAD(
COALESCE(
NULLIF(regexp_replace(COALESCE(front_page_medical_record_no, ''), '\\D', '', 'g'), ''),
NULLIF(regexp_replace(COALESCE(medical_record_no, ''), '\\D', '', 'g'), ''),
substring(source_file from '^ZY[0-9]{{2}}([0-9]{{10}})')
),
10,
'0'
),
10
)
WHERE (inpatient_no IS NULL OR BTRIM(inpatient_no) = '')
AND COALESCE(
LPAD(admission_count::text, 2, '0'),
substring(source_file from '^ZY([0-9]{{2}})[0-9]{{10}}')
) IS NOT NULL
AND COALESCE(
NULLIF(regexp_replace(COALESCE(front_page_medical_record_no, ''), '\\D', '', 'g'), ''),
NULLIF(regexp_replace(COALESCE(medical_record_no, ''), '\\D', '', 'g'), ''),
substring(source_file from '^ZY[0-9]{{2}}([0-9]{{10}})')
) IS NOT NULL
"""
).format(table=table)
)
cur.execute(
sql.SQL("ALTER TABLE {table} DROP CONSTRAINT IF EXISTS {constraint}").format(
table=table,
constraint=sql.Identifier(f"{base_table}_source_file_key"),
)
)
for constraint_name in {
f"ck_{base_table}_inpatient_no_format",
f"ck_{base_table.lower()}_inpatient_no_format",
f"ck_{base_table}_inpatient_no_required",
}:
cur.execute(
sql.SQL("ALTER TABLE {table} DROP CONSTRAINT IF EXISTS {constraint}").format(
table=table,
constraint=sql.Identifier(constraint_name),
)
)
cur.execute(
sql.SQL("DELETE FROM {table} WHERE NULLIF(BTRIM(inpatient_no), '') IS NULL").format(table=table)
)
cur.execute(
sql.SQL("UPDATE {table} SET inpatient_no = BTRIM(inpatient_no) WHERE inpatient_no <> BTRIM(inpatient_no)").format(
table=table
)
)
cur.execute(
sql.SQL(
"""
WITH ranked AS (
SELECT
id,
ROW_NUMBER() OVER (PARTITION BY BTRIM(inpatient_no) ORDER BY id DESC) AS duplicate_rank
FROM {table}
WHERE NULLIF(BTRIM(inpatient_no), '') IS NOT NULL
)
DELETE FROM {table} p
USING ranked
WHERE p.id = ranked.id
AND ranked.duplicate_rank > 1
"""
).format(table=table)
)
cur.execute(sql.SQL("ALTER TABLE {table} ALTER COLUMN inpatient_no SET NOT NULL").format(table=table))
cur.execute("SELECT 1 FROM pg_constraint WHERE conname = %s", (f"ck_{base_table}_inpatient_no_required",))
if not cur.fetchone():
cur.execute(
sql.SQL("ALTER TABLE {table} ADD CONSTRAINT {constraint} CHECK (NULLIF(BTRIM(inpatient_no), '') IS NOT NULL)").format(
table=table,
constraint=sql.Identifier(f"ck_{base_table}_inpatient_no_required"),
)
)
cur.execute(
sql.SQL("CREATE UNIQUE INDEX IF NOT EXISTS {index_name} ON {table}(inpatient_no)").format(
index_name=sql.Identifier(f"{base_table}_inpatient_no_uidx"),
table=table,
)
)
cur.execute("SELECT to_regclass(%s) AS table_oid", (old_review_regclass,))
if cur.fetchone()["table_oid"]:
cur.execute(
sql.SQL(
"""
WITH grouped AS (
SELECT
record_id,
jsonb_agg(
jsonb_build_object(
'id', id::text,
'record_id', record_id,
'source_file', source_file,
'changed_at', changed_at,
'changed_by', changed_by,
'manual_note', manual_note,
'changed_fields', changed_fields
)
ORDER BY changed_at DESC, id DESC
) AS logs
FROM {old_review_logs}
GROUP BY record_id
)
UPDATE {table} p
SET review_logs = COALESCE(p.review_logs, '[]'::jsonb) || grouped.logs
FROM grouped
WHERE p.id = grouped.record_id
"""
).format(table=table, old_review_logs=old_review_logs)
)
cur.execute(sql.SQL("DROP TABLE IF EXISTS {old_review_logs}").format(old_review_logs=old_review_logs))
cur.execute("SELECT to_regclass(%s) AS table_oid", (old_audit_regclass,))
if cur.fetchone()["table_oid"]:
cur.execute(
sql.SQL(
"""
WITH grouped AS (
SELECT
record_id,
jsonb_agg(
jsonb_build_object(
'id', id::text,
'record_id', record_id,
'source_file', source_file,
'audit_source', audit_source,
'audit_status', audit_status,
'audit_notes', audit_notes,
'ai_result', ai_result,
'snapshot', snapshot,
'created_at', created_at,
'updated_at', updated_at
)
ORDER BY updated_at DESC, id DESC
) AS logs
FROM {old_audit_logs}
GROUP BY record_id
)
UPDATE {table} p
SET audit_logs = COALESCE(p.audit_logs, '[]'::jsonb) || grouped.logs
FROM grouped
WHERE p.id = grouped.record_id
"""
).format(table=table, old_audit_logs=old_audit_logs)
)
cur.execute(sql.SQL("DROP TABLE IF EXISTS {old_audit_logs}").format(old_audit_logs=old_audit_logs))
ensure_patient_frontpage_dedupe_trigger(cur)
sync_patient_lists(cur)
ensure_patient_lists_trigger(cur)
conn.commit()
def sync_patient_lists(cur) -> None:
table = table_identifier()
list_table = patient_lists_identifier()
cur.execute(
sql.SQL(
"""
CREATE TABLE IF NOT EXISTS {list_table} (
record_id BIGSERIAL PRIMARY KEY,
batch_name TEXT NOT NULL DEFAULT 'Patient_FrontPages',
major_department TEXT NOT NULL DEFAULT '',
sub_department TEXT NOT NULL DEFAULT '',
source_folder TEXT NOT NULL DEFAULT 'Patient_FrontPages',
image_path TEXT NOT NULL DEFAULT '',
image_name TEXT NOT NULL DEFAULT '',
image_row_no INTEGER NOT NULL DEFAULT 0,
patient_name TEXT NOT NULL DEFAULT '',
gender TEXT,
age TEXT,
inpatient_no TEXT NOT NULL,
diagnosis TEXT,
admission_time TEXT,
last_write_time TEXT,
hospital_days INTEGER,
discharge_time TEXT,
postoperative_days TEXT,
review_status TEXT NOT NULL DEFAULT '首页自动关联',
review_notes TEXT,
manual_corrected BOOLEAN NOT NULL DEFAULT false,
imported_at TIMESTAMPTZ NOT NULL DEFAULT now()
)
"""
).format(list_table=list_table)
)
for name, column_type in [
("has_front_page", "BOOLEAN NOT NULL DEFAULT false"),
("front_page_id", "BIGINT"),
("front_page_source_file", "TEXT"),
]:
cur.execute(
sql.SQL("ALTER TABLE {list_table} ADD COLUMN IF NOT EXISTS {column} " + column_type).format(
list_table=list_table,
column=sql.Identifier(name),
)
)
for constraint_name in {
"ck_patient_lists_inpatient_no_format",
"ck_Patient_Lists_inpatient_no_format",
"ck_patient_lists_inpatient_no_required",
}:
cur.execute(
sql.SQL("ALTER TABLE {list_table} DROP CONSTRAINT IF EXISTS {constraint}").format(
list_table=list_table,
constraint=sql.Identifier(constraint_name),
)
)
cur.execute(
sql.SQL("DELETE FROM {list_table} WHERE NULLIF(BTRIM(inpatient_no), '') IS NULL").format(
list_table=list_table
)
)
cur.execute(
sql.SQL("UPDATE {list_table} SET inpatient_no = BTRIM(inpatient_no) WHERE inpatient_no <> BTRIM(inpatient_no)").format(
list_table=list_table
)
)
cur.execute(
sql.SQL(
"""
WITH ranked AS (
SELECT
record_id,
ROW_NUMBER() OVER (PARTITION BY BTRIM(inpatient_no) ORDER BY record_id DESC) AS duplicate_rank
FROM {list_table}
WHERE NULLIF(BTRIM(inpatient_no), '') IS NOT NULL
)
DELETE FROM {list_table} pl
USING ranked
WHERE pl.record_id = ranked.record_id
AND ranked.duplicate_rank > 1
"""
).format(list_table=list_table)
)
cur.execute(sql.SQL("ALTER TABLE {list_table} ALTER COLUMN inpatient_no SET NOT NULL").format(list_table=list_table))
cur.execute("SELECT 1 FROM pg_constraint WHERE conname = %s", ("ck_patient_lists_inpatient_no_required",))
if not cur.fetchone():
cur.execute(
sql.SQL(
"ALTER TABLE {list_table} ADD CONSTRAINT {constraint} CHECK (NULLIF(BTRIM(inpatient_no), '') IS NOT NULL)"
).format(
list_table=list_table,
constraint=sql.Identifier("ck_patient_lists_inpatient_no_required"),
)
)
cur.execute(sql.SQL("COMMENT ON COLUMN {list_table}.has_front_page IS '是否有患者首页：由Patient_FrontPages按住院号自动联动。'").format(list_table=list_table))
cur.execute(sql.SQL("COMMENT ON COLUMN {list_table}.front_page_id IS '关联的Patient_FrontPages.id。'").format(list_table=list_table))
cur.execute(sql.SQL("COMMENT ON COLUMN {list_table}.front_page_source_file IS '关联患者首页PDF文件名。'").format(list_table=list_table))
cur.execute(
sql.SQL("CREATE UNIQUE INDEX IF NOT EXISTS {index_name} ON {list_table}(inpatient_no)").format(
index_name=sql.Identifier("uq_patient_lists_inpatient_no"),
list_table=list_table,
)
)
cur.execute(
sql.SQL(
"""
WITH front_pages AS (
SELECT DISTINCT ON (BTRIM(inpatient_no))
id,
BTRIM(inpatient_no) AS inpatient_no,
source_file,
COALESCE(patient_name, '') AS patient_name,
gender,
age,
COALESCE(major_department, '') AS major_department,
COALESCE(discharge_dept, admission_dept, '') AS sub_department,
primary_diagnosis,
admission_time,
discharge_time,
hospital_days,
manual_corrected
FROM {table}
WHERE NULLIF(BTRIM(inpatient_no), '') IS NOT NULL
ORDER BY BTRIM(inpatient_no), id DESC
)
INSERT INTO {list_table} (
batch_name, major_department, sub_department, source_folder, image_path, image_name,
image_row_no, patient_name, gender, age, inpatient_no, diagnosis, admission_time,
hospital_days, discharge_time, review_status, review_notes, manual_corrected,
has_front_page, front_page_id, front_page_source_file, imported_at
)
SELECT
'Patient_FrontPages',
major_department,
sub_department,
'Patient_FrontPages',
source_file,
source_file,
0,
patient_name,
gender,
age,
inpatient_no,
primary_diagnosis,
to_char(admission_time, 'YYYY-MM-DD HH24:MI:SS'),
hospital_days,
to_char(discharge_time, 'YYYY-MM-DD HH24:MI:SS'),
'首页自动关联',
'由Patient_FrontPages按住院号自动关联',
manual_corrected,
true,
id,
source_file,
now()
FROM front_pages
ON CONFLICT (inpatient_no) DO UPDATE SET
has_front_page = true,
front_page_id = EXCLUDED.front_page_id,
front_page_source_file = EXCLUDED.front_page_source_file,
patient_name = COALESCE(NULLIF(EXCLUDED.patient_name, ''), {list_table}.patient_name),
gender = EXCLUDED.gender,
age = EXCLUDED.age,
major_department = EXCLUDED.major_department,
sub_department = EXCLUDED.sub_department,
manual_corrected = EXCLUDED.manual_corrected,
imported_at = now()
"""
).format(table=table, list_table=list_table)
)
cur.execute(
sql.SQL(
"""
UPDATE {list_table} AS pl
SET has_front_page = false,
front_page_id = NULL,
front_page_source_file = NULL,
imported_at = now()
WHERE has_front_page IS TRUE
AND NOT EXISTS (
SELECT 1 FROM {table} fp
WHERE BTRIM(fp.inpatient_no) = pl.inpatient_no
)
"""
).format(table=table, list_table=list_table)
)
def ensure_patient_frontpage_dedupe_trigger(cur) -> None:
table = table_identifier()
schema = PGTABLE.split(".", 1)[0] if "." in PGTABLE else "public"
base_table = PGTABLE.split(".", 1)[-1]
trigger_name = sql.Identifier(f"trg_{base_table}_dedupe_inpatient_no")
trigger_function = patient_dedupe_trigger_function_identifier(base_table)
cur.execute(
"""
SELECT column_name
FROM information_schema.columns
WHERE table_schema = %s
AND table_name = %s
AND column_name NOT IN ('id', 'inpatient_no', 'review_logs', 'audit_logs')
ORDER BY ordinal_position
""",
(schema, base_table),
)
update_columns = [row["column_name"] for row in cur.fetchall()]
update_assignments = sql.SQL(",\n ").join(
sql.SQL("{column} = NEW.{column}").format(column=sql.Identifier(column_name))
for column_name in update_columns
)
cur.execute(
sql.SQL(
"""
CREATE OR REPLACE FUNCTION {trigger_function}()
RETURNS trigger
LANGUAGE plpgsql
AS $trigger$
DECLARE
existing_id BIGINT;
BEGIN
NEW.inpatient_no := BTRIM(NEW.inpatient_no);
IF NULLIF(NEW.inpatient_no, '') IS NULL THEN
RETURN NEW;
END IF;
IF TG_OP = 'INSERT' THEN
SELECT id
INTO existing_id
FROM {table}
WHERE BTRIM(inpatient_no) = NEW.inpatient_no
ORDER BY id DESC
LIMIT 1;
IF existing_id IS NOT NULL THEN
DELETE FROM {table}
WHERE BTRIM(inpatient_no) = NEW.inpatient_no
AND id <> existing_id;
UPDATE {table}
SET {update_assignments}
WHERE id = existing_id;
RETURN NULL;
END IF;
END IF;
DELETE FROM {table}
WHERE BTRIM(inpatient_no) = NEW.inpatient_no
AND id <> NEW.id;
RETURN NEW;
END;
$trigger$;
"""
).format(trigger_function=trigger_function, table=table, update_assignments=update_assignments)
)
cur.execute(
sql.SQL("DROP TRIGGER IF EXISTS {trigger_name} ON {table}").format(
trigger_name=trigger_name,
table=table,
)
)
cur.execute(
sql.SQL(
"""
CREATE TRIGGER {trigger_name}
BEFORE INSERT OR UPDATE OF inpatient_no ON {table}
FOR EACH ROW EXECUTE FUNCTION {trigger_function}()
"""
).format(trigger_name=trigger_name, table=table, trigger_function=trigger_function)
)
def ensure_patient_lists_trigger(cur) -> None:
table = table_identifier()
list_table = patient_lists_identifier()
base_table = PGTABLE.split(".", 1)[-1]
trigger_name = sql.Identifier(f"trg_{base_table}_sync_patient_lists")
trigger_function = patient_list_trigger_function_identifier(base_table)
cur.execute(
sql.SQL(
"""
CREATE OR REPLACE FUNCTION {trigger_function}()
RETURNS trigger
LANGUAGE plpgsql
AS $trigger$
BEGIN
IF TG_OP = 'DELETE' THEN
IF NULLIF(BTRIM(OLD.inpatient_no), '') IS NOT NULL THEN
UPDATE {list_table} AS pl
SET has_front_page = false,
front_page_id = NULL,
front_page_source_file = NULL,
imported_at = now()
WHERE pl.inpatient_no = BTRIM(OLD.inpatient_no)
AND NOT EXISTS (
SELECT 1 FROM {table} fp
WHERE BTRIM(fp.inpatient_no) = BTRIM(OLD.inpatient_no)
);
END IF;
RETURN OLD;
END IF;
IF TG_OP = 'UPDATE'
AND NULLIF(BTRIM(OLD.inpatient_no), '') IS NOT NULL
AND BTRIM(OLD.inpatient_no) IS DISTINCT FROM BTRIM(NEW.inpatient_no) THEN
UPDATE {list_table} AS pl
SET has_front_page = false,
front_page_id = NULL,
front_page_source_file = NULL,
imported_at = now()
WHERE pl.inpatient_no = BTRIM(OLD.inpatient_no)
AND NOT EXISTS (
SELECT 1 FROM {table} fp
WHERE BTRIM(fp.inpatient_no) = BTRIM(OLD.inpatient_no)
);
END IF;
IF NULLIF(BTRIM(NEW.inpatient_no), '') IS NOT NULL THEN
INSERT INTO {list_table} (
batch_name, major_department, sub_department, source_folder, image_path, image_name,
image_row_no, patient_name, gender, age, inpatient_no, diagnosis, admission_time,
hospital_days, discharge_time, review_status, review_notes, manual_corrected,
has_front_page, front_page_id, front_page_source_file, imported_at
)
VALUES (
'Patient_FrontPages',
COALESCE(NEW.major_department, ''),
COALESCE(NEW.discharge_dept, NEW.admission_dept, ''),
'Patient_FrontPages',
COALESCE(NEW.source_file, ''),
COALESCE(NEW.source_file, ''),
0,
COALESCE(NEW.patient_name, ''),
NEW.gender,
NEW.age,
BTRIM(NEW.inpatient_no),
NEW.primary_diagnosis,
to_char(NEW.admission_time, 'YYYY-MM-DD HH24:MI:SS'),
NEW.hospital_days,
to_char(NEW.discharge_time, 'YYYY-MM-DD HH24:MI:SS'),
'首页自动关联',
'由Patient_FrontPages触发器按住院号自动关联',
COALESCE(NEW.manual_corrected, false),
true,
NEW.id,
NEW.source_file,
now()
)
ON CONFLICT (inpatient_no) DO UPDATE SET
has_front_page = true,
front_page_id = EXCLUDED.front_page_id,
front_page_source_file = EXCLUDED.front_page_source_file,
patient_name = COALESCE(NULLIF(EXCLUDED.patient_name, ''), {list_table}.patient_name),
gender = EXCLUDED.gender,
age = EXCLUDED.age,
major_department = EXCLUDED.major_department,
sub_department = EXCLUDED.sub_department,
manual_corrected = EXCLUDED.manual_corrected,
imported_at = now();
END IF;
RETURN NEW;
END;
$trigger$;
"""
).format(trigger_function=trigger_function, table=table, list_table=list_table)
)
cur.execute(
sql.SQL("DROP TRIGGER IF EXISTS {trigger_name} ON {table}").format(
trigger_name=trigger_name,
table=table,
)
)
cur.execute(
sql.SQL(
"""
CREATE TRIGGER {trigger_name}
AFTER INSERT OR UPDATE OR DELETE ON {table}
FOR EACH ROW EXECUTE FUNCTION {trigger_function}()
"""
).format(trigger_name=trigger_name, table=table, trigger_function=trigger_function)
)
def fetch_review_logs(record_id: int, limit: int = 30) -> list[dict[str, Any]]:
query = sql.SQL("SELECT review_logs FROM {table} WHERE id = %s").format(table=table_identifier())
with connect() as conn, conn.cursor() as cur:
cur.execute(query, (record_id,))
row = cur.fetchone()
if not row:
return []
logs = row.get("review_logs") or []
if not isinstance(logs, list):
return []
normalized = [row_to_json(dict(item)) for item in logs if isinstance(item, dict)]
normalized.sort(key=lambda item: (str(item.get("changed_at") or ""), str(item.get("id") or "")), reverse=True)
return normalized[:limit]
def fetch_audit_logs(limit: int = 100) -> list[dict[str, Any]]:
ensure_workflow_tables()
query = sql.SQL(
"""
SELECT
p.id AS record_id,
p.source_file,
p.inpatient_no,
p.medical_record_no,
p.patient_name,
p.primary_diagnosis,
p.review_status,
log_item.value AS log,
log_item.ordinality AS log_order
FROM {table} p
CROSS JOIN LATERAL jsonb_array_elements(COALESCE(p.audit_logs, '[]'::jsonb)) WITH ORDINALITY AS log_item(value, ordinality)
WHERE COALESCE(log_item.value->>'audit_status', '') <> 'pending'
ORDER BY COALESCE(log_item.value->>'updated_at', log_item.value->>'created_at', '') DESC, log_item.ordinality DESC
LIMIT %s
"""
).format(table=table_identifier())
with connect() as conn, conn.cursor() as cur:
cur.execute(query, (limit,))
rows = cur.fetchall()
logs: list[dict[str, Any]] = []
for row in rows:
base = row_to_json({key: value for key, value in dict(row).items() if key not in {"log", "log_order"}})
item = row.get("log") or {}
if isinstance(item, dict):
log = {**base, **row_to_json(dict(item))}
log.setdefault("record_id", base.get("record_id"))
log.setdefault("source_file", base.get("source_file"))
logs.append(log)
return logs
def insert_review_log(
cur,
record_id: int,
source_file: str,
changed_fields: list[dict[str, Any]],
manual_note: str,
changed_by: str = "web",
ai_result: Any = None,
) -> None:
changed_at = datetime.now().isoformat(timespec="seconds")
log = {
"id": datetime.now().strftime("%Y%m%d%H%M%S%f"),
"record_id": record_id,
"source_file": source_file,
"changed_at": changed_at,
"changed_by": changed_by,
"manual_note": manual_note,
"changed_fields": json_ready_deep(changed_fields),
}
if ai_result is not None:
log["ai_result"] = json_ready_deep(ai_result)
cur.execute(
sql.SQL(
"""
UPDATE {table}
SET review_logs = COALESCE(review_logs, '[]'::jsonb) || %s::jsonb
WHERE id = %s
"""
).format(table=table_identifier()),
(
json.dumps([log], ensure_ascii=False),
record_id,
),
)
def json_changed_fields(field: str, old_value: Any, new_value: Any) -> list[dict[str, Any]]:
label = FIELD_META[field]["label"]
if isinstance(old_value, list) or isinstance(new_value, list):
old_rows = old_value if isinstance(old_value, list) else []
new_rows = new_value if isinstance(new_value, list) else []
columns: list[str] = []
for row in [*old_rows, *new_rows]:
if isinstance(row, dict):
for key in row:
if key not in columns:
columns.append(str(key))
entries: list[dict[str, Any]] = []
for index in range(max(len(old_rows), len(new_rows))):
old_row = old_rows[index] if index < len(old_rows) and isinstance(old_rows[index], dict) else {}
new_row = new_rows[index] if index < len(new_rows) and isinstance(new_rows[index], dict) else {}
for column in columns:
old_cell = old_row.get(column, "")
new_cell = new_row.get(column, "")
if comparable(old_cell) == comparable(new_cell):
continue
entries.append(
{
"field": f"{field}[{index}].{column}",
"label": f"{label}[{index + 1}].{column}",
"old": json_ready_deep(old_cell),
"new": json_ready_deep(new_cell),
}
)
return entries
if isinstance(old_value, dict) or isinstance(new_value, dict):
old_dict = old_value if isinstance(old_value, dict) else {}
new_dict = new_value if isinstance(new_value, dict) else {}
keys = list(dict.fromkeys([*old_dict.keys(), *new_dict.keys()]))
return [
{
"field": f"{field}.{key}",
"label": f"{label}.{key}",
"old": json_ready_deep(old_dict.get(key, "")),
"new": json_ready_deep(new_dict.get(key, "")),
}
for key in keys
if comparable(old_dict.get(key, "")) != comparable(new_dict.get(key, ""))
]
return [
{
"field": field,
"label": label,
"old": json_ready_deep(old_value),
"new": json_ready_deep(new_value),
}
]
def main_diagnosis_row_index(rows: list[Any]) -> int:
for index, row in enumerate(rows):
if isinstance(row, dict) and str(row.get("诊断类别") or "").strip() == "主要诊断":
return index
return 0
def main_diagnosis_from_discharge_diagnoses(value: Any) -> dict[str, str]:
rows = value if isinstance(value, list) else []
if not rows:
return {"primary_diagnosis": "", "primary_diagnosis_code": "", "primary_admission_condition": ""}
index = main_diagnosis_row_index(rows)
row = rows[index] if index < len(rows) and isinstance(rows[index], dict) else {}
return {
"primary_diagnosis": str(row.get("出院诊断") or row.get("诊断名称") or "").strip(),
"primary_diagnosis_code": str(row.get("疾病编码") or row.get("诊断编码") or "").strip(),
"primary_admission_condition": str(row.get("入院病情") or "").strip(),
}
def sync_primary_diagnosis_updates(updates: dict[str, Any], before: dict[str, Any]) -> None:
if "discharge_diagnoses" not in updates:
return
derived = main_diagnosis_from_discharge_diagnoses(updates.get("discharge_diagnoses"))
for field, value in derived.items():
if comparable(before.get(field)) != comparable(value):
updates[field] = value
def build_record_updates(
record_id: int,
fields: dict[str, Any],
manual_note: str = "",
note_prefix: str = "人工复核",
force_reviewed: bool = True,
) -> tuple[dict[str, Any], list[dict[str, Any]], dict[str, Any]]:
before = fetch_record(record_id)
updates: dict[str, Any] = {}
for field, value in fields.items():
if field not in EDITABLE_FIELDS:
continue
updates[field] = parse_field_value(field, value)
for number_field in ["medical_record_no", "front_page_medical_record_no"]:
if updates.get(number_field):
normalized_number = digits(updates[number_field], 10)
if normalized_number:
updates[number_field] = normalized_number
if "inpatient_no" in updates and updates["inpatient_no"] is not None:
updates["inpatient_no"] = str(updates["inpatient_no"]).strip()
else:
preview = {**before, **updates}
derived_inpatient_no = build_inpatient_no_from_record(preview)
if derived_inpatient_no:
updates["inpatient_no"] = derived_inpatient_no
if "inpatient_no" in updates and not str(updates["inpatient_no"] or "").strip():
raise HTTPException(status_code=400, detail="患者号不能为空")
sync_primary_diagnosis_updates(updates, before)
manual_note = manual_note.strip()
if manual_note:
current = before.get("review_notes") or []
current_notes = current if isinstance(current, list) else [current]
current_notes.append(f"{note_prefix}({datetime.now().strftime('%Y-%m-%d %H:%M:%S')}): {manual_note}")
updates["review_notes"] = current_notes
changed_fields: list[dict[str, Any]] = []
for field, value in updates.items():
if field not in FIELD_META:
continue
old_value = before.get(field)
if comparable(old_value) == comparable(value):
continue
if field in JSON_FIELDS:
changed_fields.extend(json_changed_fields(field, old_value, value))
continue
changed_fields.append(
{
"field": field,
"label": FIELD_META[field]["label"],
"old": json_ready_deep(old_value),
"new": json_ready_deep(value),
}
)
should_mark_manual = force_reviewed or bool(changed_fields)
if force_reviewed and before.get("review_status") != "reviewed":
changed_fields.append(
{
"field": "review_status",
"label": "复核状态",
"old": json_ready_deep(before.get("review_status")),
"new": "reviewed",
}
)
if should_mark_manual and before.get("manual_corrected") is not True:
changed_fields.append(
{
"field": "manual_corrected",
"label": "人工修正",
"old": json_ready_deep(before.get("manual_corrected")),
"new": True,
}
)
return updates, changed_fields, before
def apply_record_updates(
cur,
record_id: int,
updates: dict[str, Any],
changed_fields: list[dict[str, Any]],
before: dict[str, Any],
manual_note: str = "",
force_reviewed: bool = True,
) -> None:
assignments = []
values: list[Any] = []
for field, value in updates.items():
assignments.append(sql.SQL("{} = %s").format(sql.Identifier(field)))
values.append(psycopg2.extras.Json(value, dumps=lambda obj: json.dumps(obj, ensure_ascii=False)) if field in JSON_DB_FIELDS else value)
if force_reviewed:
assignments.append(sql.SQL("review_status = 'reviewed'"))
assignments.append(sql.SQL("manual_corrected = TRUE"))
elif changed_fields:
assignments.append(sql.SQL("manual_corrected = TRUE"))
if before.get("review_status") == "auto_pass":
assignments.append(sql.SQL("review_status = 'reviewed'"))
if assignments:
query = sql.SQL("UPDATE {table} SET {assignments} WHERE id = %s").format(
table=table_identifier(),
assignments=sql.SQL(", ").join(assignments),
)
cur.execute(query, [*values, record_id])
if cur.rowcount != 1:
raise HTTPException(status_code=404, detail="记录不存在")
if changed_fields or manual_note.strip():
insert_review_log(cur, record_id, before.get("source_file", ""), changed_fields, manual_note.strip())
def safe_child(root: Path, child_name: str) -> Path:
if Path(child_name).name != child_name:
raise HTTPException(status_code=400, detail="非法文件名")
path = (root / child_name).resolve()
if root not in path.parents and path != root:
raise HTTPException(status_code=400, detail="非法路径")
return path
def get_pdf_path(source_file: str) -> Path | None:
path = safe_child(PDF_DIR, source_file)
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("```"):
text = text.strip("`")
if text.startswith("json"):
text = text[4:].strip()
try:
return json.loads(text)
except json.JSONDecodeError as exc:
start = text.find("{")
end = text.rfind("}")
if start >= 0 and end > start:
try:
return json.loads(text[start : end + 1])
except json.JSONDecodeError:
pass
raise ValueError(f"AI 返回 JSON 无法解析：{exc.msg}")
def render_pdf_page_png(pdf_path: Path, dpi: int = 96, page_index: int = 0, clip: Any | None = None) -> bytes:
try:
import fitz # type: ignore[import-not-found]
except ImportError as exc:
raise HTTPException(status_code=500, detail="Web 容器缺少 PyMuPDF，无法渲染 PDF 供 AI 核验") from exc
with fitz.open(str(pdf_path)) as document:
if document.page_count < 1:
raise HTTPException(status_code=400, detail="PDF 没有可渲染页面")
page = document.load_page(max(0, min(page_index, document.page_count - 1)))
matrix = fitz.Matrix(dpi / 72, dpi / 72)
pixmap = page.get_pixmap(matrix=matrix, alpha=False, clip=clip)
return pixmap.tobytes("png")
def render_pdf_page_data_url(pdf_path: Path, dpi: int = 96, page_index: int = 0, clip: Any | None = None) -> str:
encoded = base64.b64encode(render_pdf_page_png(pdf_path, dpi=dpi, page_index=page_index, clip=clip)).decode("utf-8")
return f"data:image/png;base64,{encoded}"
def normalize_pdf_text(text: str) -> str:
return re.sub(r"\s+", "", text or "")
def record_review_text(record: dict[str, Any]) -> str:
parts: list[str] = []
for key in ("review_notes", "quality_notes", "auto_corrections"):
value = record.get(key)
if isinstance(value, list):
parts.extend(json.dumps(item, ensure_ascii=False) if not isinstance(item, str) else item for item in value)
elif value:
parts.append(str(value))
return "；".join(parts)
def relevant_pdf_modules(record: dict[str, Any], privacy_mode: bool = True) -> list[dict[str, Any]]:
text = record_review_text(record)
normalized = normalize_pdf_text(text)
selected: list[dict[str, Any]] = []
for module in PDF_MODULE_DEFINITIONS:
if privacy_mode and module["name"] not in AI_SAFE_MODULE_NAMES:
continue
if any(normalize_pdf_text(keyword) in normalized for keyword in module["note_keywords"]):
selected.append(module)
if not selected:
default_names = {"诊断表格", "手术表格"} if privacy_mode else {module["name"] for module in PDF_MODULE_DEFINITIONS}
selected = [module for module in PDF_MODULE_DEFINITIONS if module["name"] in default_names]
return selected[: 4 if privacy_mode else 5]
def redact_text_for_ai(text: str) -> str:
redacted = str(text or "")
for pattern, replacement in AI_REDACT_PATTERNS:
redacted = pattern.sub(replacement, redacted)
return redacted
def redact_for_ai(value: Any) -> Any:
if isinstance(value, str):
return redact_text_for_ai(value)
if isinstance(value, list):
return [redact_for_ai(item) for item in value]
if isinstance(value, dict):
return {str(key): redact_for_ai(item) for key, item in value.items()}
return value
def upload_text_for_ai(text: str, privacy_mode: bool = True) -> str:
return redact_text_for_ai(text) if privacy_mode else str(text or "")
def upload_value_for_ai(value: Any, privacy_mode: bool = True) -> Any:
return redact_for_ai(value) if privacy_mode else json_ready_deep(value)
def privacy_excluded_review(record: dict[str, Any]) -> bool:
normalized = normalize_pdf_text(record_review_text(record))
excluded_modules = [module for module in PDF_MODULE_DEFINITIONS if module["name"] not in AI_SAFE_MODULE_NAMES]
has_excluded = any(
normalize_pdf_text(keyword) in normalized
for module in excluded_modules
for keyword in module["note_keywords"]
)
has_safe = any(
normalize_pdf_text(keyword) in normalized
for module in PDF_MODULE_DEFINITIONS
if module["name"] in AI_SAFE_MODULE_NAMES
for keyword in module["note_keywords"]
)
return has_excluded and not has_safe
def admission_path_crop_y(blocks: list[dict[str, Any]], page_index: int) -> float | None:
candidates: list[float] = []
for block in blocks:
if block["page_index"] != page_index:
continue
normalized = normalize_pdf_text(block["text"])
has_admission_path = "入院途径" in normalized
has_path_options = "急诊" in normalized and "门诊" in normalized and "其他医疗机构转入" in normalized
if has_admission_path or has_path_options:
candidates.append(float(block["rect"].y0))
return min(candidates) if candidates else None
def apply_ai_image_privacy_clip(page: Any, clip: Any, blocks: list[dict[str, Any]], page_index: int) -> Any | None:
if page_index != 0:
return clip
crop_y = admission_path_crop_y(blocks, page_index)
if crop_y is None:
return None
protected_y0 = max(float(clip.y0), crop_y - 16)
if float(clip.y1) - protected_y0 < 80:
return None
return type(clip)(clip.x0, protected_y0, clip.x1, clip.y1)
def fallback_privacy_page_context(
pdf_path: Path,
document: Any,
text_blocks: list[dict[str, Any]],
crop_blocks: list[dict[str, Any]],
) -> dict[str, Any] | None:
if document.page_count < 1:
return None
page_index = 0
page = document.load_page(page_index)
crop_y = admission_path_crop_y(crop_blocks, page_index)
if crop_y is None:
return None
clip = type(page.rect)(
max(page.rect.x0, page.rect.x0 + 18),
max(page.rect.y0, crop_y - 16),
min(page.rect.x1, page.rect.x1 - 18),
page.rect.y1,
)
if float(clip.y1) - float(clip.y0) < 120:
return None
return {
"name": "隐私裁剪首页",
"page": 1,
"bbox": [round(clip.x0, 1), round(clip.y0, 1), round(clip.x1, 1), round(clip.y1, 1)],
"text": redact_text_for_ai(clip_text_from_blocks(text_blocks, page_index, clip)),
"image_url": render_pdf_page_data_url(pdf_path, dpi=120, page_index=page_index, clip=clip),
}
def fallback_unrestricted_page_context(pdf_path: Path, document: Any, text_blocks: list[dict[str, Any]]) -> dict[str, Any] | None:
if document.page_count < 1:
return None
page_index = 0
page = document.load_page(page_index)
clip = page.rect
return {
"name": "整页首页",
"page": 1,
"bbox": [round(clip.x0, 1), round(clip.y0, 1), round(clip.x1, 1), round(clip.y1, 1)],
"text": clip_text_from_blocks(text_blocks, page_index, clip),
"image_url": render_pdf_page_data_url(pdf_path, dpi=110, page_index=page_index, clip=clip),
}
def clip_text_from_blocks(blocks: list[dict[str, Any]], page_index: int, clip: Any) -> str:
lines: list[str] = []
for block in blocks:
if block["page_index"] != page_index:
continue
rect = block["rect"]
if rect.intersects(clip):
text = " ".join(str(block["text"]).split())
if text:
lines.append(text)
joined = "\n".join(lines)
return joined[:2200]
def page_line_blocks(page: Any, page_index: int) -> list[dict[str, Any]]:
lines: list[dict[str, Any]] = []
page_dict = page.get_text("dict")
for block in page_dict.get("blocks", []):
for line in block.get("lines", []):
spans = line.get("spans") or []
text = "".join(str(span.get("text") or "") for span in spans).strip()
bbox = line.get("bbox")
if not text or not bbox:
continue
lines.append(
{
"page_index": page_index,
"rect": type(page.rect)(bbox),
"text": text,
"normalized": normalize_pdf_text(text),
}
)
return lines
def add_keyword_from_value(keywords: list[str], value: Any) -> None:
if value in {None, ""}:
return
text = str(value).strip()
if len(text) >= 2:
keywords.append(text)
for segment in re.split(r"[、,，;；\s]+", text):
if len(segment) >= 4:
keywords.append(segment)
compact = normalize_pdf_text(text)
if len(compact) >= 12:
keywords.extend([compact[:10], compact[-10:]])
def record_module_keywords(record: dict[str, Any], module_name: str) -> list[str]:
keywords: list[str] = []
if module_name == "诊断表格":
for field in ("outpatient_diagnosis", "outpatient_diagnosis_code", "primary_diagnosis", "primary_diagnosis_code", "pathology_diagnosis", "pathology_diagnosis_code"):
add_keyword_from_value(keywords, record.get(field))
diagnoses = record.get("discharge_diagnoses")
if isinstance(diagnoses, list):
for row in diagnoses[:8]:
if isinstance(row, dict):
for key in ("出院诊断", "疾病编码", "诊断名称", "诊断编码"):
add_keyword_from_value(keywords, row.get(key))
elif module_name == "手术表格":
operations = record.get("operations")
if isinstance(operations, list):
for row in operations[:6]:
if isinstance(row, dict):
for key in ("手术操作名称", "手术操作编码", "手术操作日期"):
add_keyword_from_value(keywords, row.get(key))
elif module_name == "地址联系人":
for field in ("current_address", "household_address", "employer_address", "contact_name", "contact_address", "contact_phone"):
add_keyword_from_value(keywords, record.get(field))
elif module_name == "基本信息":
for field in ("patient_name", "medical_record_no", "inpatient_no", "id_card_no", "admission_time", "discharge_time"):
add_keyword_from_value(keywords, record.get(field))
elif module_name == "离院费用":
for field in ("total_cost", "self_pay_amount", "discharge_disposition_code", "receiving_org_name"):
add_keyword_from_value(keywords, record.get(field))
return keywords
def extract_pdf_module_context(pdf_path: Path, record: dict[str, Any], privacy_mode: bool = True) -> dict[str, Any]:
try:
import fitz # type: ignore[import-not-found]
except ImportError as exc:
raise HTTPException(status_code=500, detail="Web 容器缺少 PyMuPDF，无法提取 PDF 文本") from exc
modules = relevant_pdf_modules(record, privacy_mode=privacy_mode)
privacy_skipped = privacy_mode and privacy_excluded_review(record)
contexts: list[dict[str, Any]] = []
all_blocks: list[dict[str, Any]] = []
all_lines: list[dict[str, Any]] = []
full_text_excerpt = ""
with fitz.open(str(pdf_path)) as document:
for page_index in range(document.page_count):
page = document.load_page(page_index)
all_lines.extend(page_line_blocks(page, page_index))
for block in page.get_text("blocks"):
if len(block) < 5:
continue
text = str(block[4] or "").strip()
if not text:
continue
all_blocks.append(
{
"page_index": page_index,
"rect": fitz.Rect(block[:4]),
"text": text,
"normalized": normalize_pdf_text(text),
}
)
for module in modules:
module_keywords = [*module["keywords"], *record_module_keywords(record, module["name"])]
keywords = [normalize_pdf_text(keyword) for keyword in module_keywords]
hits = [block for block in all_blocks if any(keyword and keyword in block["normalized"] for keyword in keywords)]
if not hits:
continue
page_counts: dict[int, int] = {}
for hit in hits:
page_counts[hit["page_index"]] = page_counts.get(hit["page_index"], 0) + 1
page_index = max(page_counts, key=page_counts.get)
page = document.load_page(page_index)
page_hits = [hit for hit in hits if hit["page_index"] == page_index]
min_y = min(hit["rect"].y0 for hit in page_hits)
max_y = max(hit["rect"].y1 for hit in page_hits)
tail = float(module.get("tail") or 420)
clip = fitz.Rect(
max(page.rect.x0, page.rect.x0 + 18),
max(page.rect.y0, min_y - 60),
min(page.rect.x1, page.rect.x1 - 18),
min(page.rect.y1, max(max_y + tail, min_y + 220)),
)
if privacy_mode:
clip = apply_ai_image_privacy_clip(page, clip, all_lines, page_index)
if clip is None:
continue
contexts.append(
{
"name": module["name"],
"page": page_index + 1,
"bbox": [round(clip.x0, 1), round(clip.y0, 1), round(clip.x1, 1), round(clip.y1, 1)],
"text": upload_text_for_ai(clip_text_from_blocks(all_blocks, page_index, clip), privacy_mode=privacy_mode),
"image_url": render_pdf_page_data_url(pdf_path, dpi=120, page_index=page_index, clip=clip),
}
)
if privacy_mode and not contexts and not privacy_skipped:
fallback_context = fallback_privacy_page_context(pdf_path, document, all_blocks, all_lines)
if fallback_context:
contexts.append(fallback_context)
if not privacy_mode:
full_text_excerpt = "\n".join(" ".join(str(block["text"]).split()) for block in all_blocks)[:9000]
if not contexts:
fallback_context = fallback_unrestricted_page_context(pdf_path, document, all_blocks)
if fallback_context:
contexts.append(fallback_context)
return {
"modules": contexts,
"full_text_excerpt": full_text_excerpt,
"privacy_skipped": privacy_skipped,
}
def ai_record_snapshot(record: dict[str, Any], privacy_mode: bool = True) -> dict[str, Any]:
if privacy_mode:
return {key: upload_value_for_ai(record.get(key), privacy_mode=True) for key in AI_UPLOAD_FIELDS}
snapshot: dict[str, Any] = {}
for group in FIELD_GROUPS:
for name, _label, _field_type, _options in group["fields"]:
snapshot[name] = upload_value_for_ai(record.get(name), privacy_mode=False)
for key in ("review_status", "quality_status", "quality_notes", "review_notes", "auto_corrections"):
snapshot[key] = upload_value_for_ai(record.get(key), privacy_mode=False)
return snapshot
def build_ai_prompt(record: dict[str, Any], pdf_context: dict[str, Any], privacy_mode: bool = True) -> str:
snapshot = json.dumps(ai_record_snapshot(record, privacy_mode=privacy_mode), ensure_ascii=False, indent=2)
context_for_prompt = {
"modules": [
{
"name": item["name"],
"page": item["page"],
"bbox": item["bbox"],
"pdf_text": upload_text_for_ai(item["text"], privacy_mode=privacy_mode),
}
for item in pdf_context.get("modules", [])
],
"full_text_excerpt": upload_text_for_ai(pdf_context.get("full_text_excerpt", ""), privacy_mode=privacy_mode),
"privacy_mode": "on" if privacy_mode else "off",
"excluded": (
"姓名、住院号、病案号、身份证、电话、地址、基本信息、地址联系人、整页截图、全文摘录均不上传"
if privacy_mode
else "未启用隐私模式；可上传基本信息、地址联系人、相关局部截图和PDF文本摘录用于核验"
),
}
context_json = json.dumps(context_for_prompt, ensure_ascii=False, indent=2)
mode_instruction = (
"当前为隐私模式：只上传诊断表格、手术表格、离院费用的脱敏文本和局部截图；不上传基本信息、地址联系人、整页截图或 PDF 全文摘录。"
if privacy_mode
else "当前未启用隐私模式：允许上传基本信息、地址联系人、诊断、手术、离院费用等相关 PDF 文本和局部截图；可对所有结构化字段提出修正。"
)
return f"""
请对这份住院病案首页做视觉核验，只返回 JSON，不要输出 Markdown。
任务目标：
1. {mode_instruction}
2. 将 PDF 文本、局部截图中的可见内容，与下面的结构化字段逐项对比。
3. 如果 PDF 能读出明确值且结构化字段缺失、截断或错填，请直接给出 suggested_updates；系统会先按你的建议修改，再归类。
4. 修改后如果 AI 认为没有必须复核的问题，classification 返回 "ok"；即使编码栏空白、某些字段为空，只要 PDF 证实首页本来如此且无需处理，也返回 "ok"。
5. 只有仍需要复核或纠正的问题，classification 返回 "problem"，并写入 remaining_issues。
6. 如果手术表格中能看到“手术及操作编码”列但对应单元格为空，写“编码栏可见但为空白”，不要写“编码区域未在首页显示”；单元格本来为空且无需补录时不要因此判为 problem。
7. 手术操作名称可能因为换行被结构化解析截断；如果 PDF定位文本或局部截图中显示完整多行名称，请把完整名称放入 suggested_updates。
8. 门急诊诊断编码只能用于 outpatient_diagnosis_code；主要诊断请修正 discharge_diagnoses 中“诊断类别=主要诊断”的行，不要把门急诊诊断编码复制到 discharge_diagnoses[].疾病编码，除非出院诊断表格对应“疾病编码”单元格本身清楚显示该编码。
9. remaining_issues 只写当前文档复核人应该特别注意的内容；不要写如何修改，不要重复 suggested_updates，不要写“无需补录/无需处理/首页原貌”这类已判定无问题的说明。
10. 不要编造 PDF 中看不见的内容，不要输出置信度。
必须返回这个 JSON 结构：
{{
"classification": "ok 或 problem",
"summary": "一句话结论，60字以内",
"method": "AI视觉核验：PDF文本定位+局部截图，对照复核定位和结构化字段",
"suggested_updates": [
{{"field": "字段名或路径，例如 discharge_diagnoses[0].疾病编码 或 operations[0].麻醉方式", "value": "PDF中应写入的值", "reason": "20字以内"}}
],
"remaining_issues": ["AI觉得有必要复核的内容，最多3条；无则返回空数组"],
"evidence": [
{{"target": "核验点", "pdf_value": "PDF图片值", "structured_value": "结构化值", "result": "match/mismatch/uncertain", "note": "30字以内"}}
]
}}
如果没有明确修正值，suggested_updates 返回 []。如果 suggested_updates 已经修正主要问题，remaining_issues 可以返回 [] 并把 classification 归为 "ok"。
结构化字段：
{snapshot}
PDF定位文本：
{context_json}
""".strip()
def local_privacy_ai_result(settings: dict[str, Any], pdf_context: dict[str, Any], reason: str) -> dict[str, Any]:
parsed = normalize_ai_parsed(
{
"classification": "problem",
"summary": "隐私模式未上传敏感区域，需人工复核",
"method": "本地隐私保护：未调用外部AI",
"suggested_updates": [],
"remaining_issues": [reason],
"evidence": [],
}
)
return {
"model": settings.get("model"),
"thinking_enabled": bool(settings.get("thinking_enabled")),
"privacy_mode": True,
"checked_at": datetime.now().isoformat(timespec="seconds"),
"ai_question": "隐私模式本地判定：未上传基本信息、地址联系人、整页截图或PDF全文摘录。",
"pdf_context": {"modules": [], "full_text_excerpt": ""},
"raw_response": "",
"parsed": parsed,
"usage": {},
}
def merge_kimi_override(kimi: dict[str, Any], override: dict[str, Any] | None = None) -> dict[str, Any]:
if not override:
return kimi
merged = dict(kimi)
if override.get("model"):
merged["model"] = str(override["model"]).strip()
if override.get("thinking_enabled") is not None:
merged["thinking_enabled"] = bool(override["thinking_enabled"])
if override.get("privacy_mode") is not None:
merged["privacy_mode"] = bool(override["privacy_mode"])
return merged
def call_kimi_ai_review(record: dict[str, Any], kimi_override: dict[str, Any] | None = None) -> dict[str, Any]:
local_kimi_settings = load_local_settings().get("kimi") or {}
local_kimi_settings = merge_kimi_override(local_kimi_settings, kimi_override)
settings = public_kimi_settings(local_kimi_settings)
api_key = kimi_api_key(local_kimi_settings)
privacy_mode = normalize_bool(local_kimi_settings.get("privacy_mode"), True)
if not settings["available"]:
raise HTTPException(status_code=400, detail="AI 核验未启用或未配置 API Key")
pdf_path = get_pdf_path(record.get("source_file") or "")
if not pdf_path:
raise HTTPException(status_code=404, detail="PDF 文件不存在，无法 AI 核验")
pdf_context = extract_pdf_module_context(pdf_path, record, privacy_mode=privacy_mode)
if privacy_mode and pdf_context.get("privacy_skipped"):
return local_privacy_ai_result(settings, pdf_context, "复核定位疑似基本信息或地址联系人；隐私模式不上传该区域，请人工复核。")
if privacy_mode and not pdf_context.get("modules"):
return local_privacy_ai_result(settings, pdf_context, "未定位到可脱敏上传的诊断/手术/费用局部区域；未上传整页截图，请人工复核。")
ai_question = build_ai_prompt(record, pdf_context, privacy_mode=privacy_mode)
content_parts: list[dict[str, Any]] = [{"type": "text", "text": ai_question}]
for item in pdf_context.get("modules", [])[: 4 if privacy_mode else 5]:
content_parts.append({"type": "text", "text": f"局部截图：{item['name']}，第 {item['page']} 页，bbox={item['bbox']}"})
content_parts.append({"type": "image_url", "image_url": {"url": item["image_url"]}})
supports_thinking = model_supports_thinking(settings["model"])
thinking_enabled = supports_thinking and bool(settings.get("thinking_enabled"))
payload = {
"model": settings["model"],
"temperature": 1.0 if thinking_enabled else 0.6,
"max_tokens": 16000 if thinking_enabled else 3200,
"response_format": {"type": "json_object"},
"messages": [
{"role": "system", "content": "你是严谨的病案首页视觉核验助手，只输出合法 JSON；字符串里的引号必须转义。"},
{
"role": "user",
"content": content_parts,
},
],
}
if supports_thinking:
payload["thinking"] = {"type": "enabled" if thinking_enabled else "disabled"}
request = urllib.request.Request(
f"{settings['api_base'].rstrip('/')}/chat/completions",
data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
},
method="POST",
)
try:
with urllib.request.urlopen(request, timeout=180 if thinking_enabled else 90) as response:
data = json.loads(response.read().decode("utf-8"))
except urllib.error.HTTPError as exc:
detail = exc.read().decode("utf-8", errors="replace")
raise HTTPException(status_code=502, detail=f"AI API 返回错误 {exc.code}: {detail}") from exc
except urllib.error.URLError as exc:
raise HTTPException(status_code=502, detail=f"AI API 调用失败：{exc}") from exc
message = data["choices"][0].get("message") or {}
content = message.get("content") or message.get("reasoning_content") or ""
parsed = parse_json_content(content)
if not isinstance(parsed, dict):
parsed = {"decision": "confirm", "summary": "AI 返回 JSON 不是对象", "raw_response": content}
parsed = normalize_ai_parsed(parsed)
return {
"model": settings["model"],
"thinking_enabled": thinking_enabled,
"privacy_mode": privacy_mode,
"checked_at": datetime.now().isoformat(timespec="seconds"),
"ai_question": ai_question,
"pdf_context": {
"modules": [
{key: item[key] for key in ("name", "page", "bbox", "text")}
for item in pdf_context.get("modules", [])
],
"full_text_excerpt": pdf_context.get("full_text_excerpt", ""),
},
"raw_response": content,
"parsed": parsed,
"usage": data.get("usage") or {},
}
def ai_bool(value: Any) -> bool:
if isinstance(value, bool):
return value
if isinstance(value, (int, float)):
return bool(value)
return str(value or "").strip().lower() in {"true", "yes", "1", "是", "有", "确认"}
def ai_join_text(value: Any) -> str:
if isinstance(value, str):
return value
if isinstance(value, list):
return " ".join(ai_join_text(item) for item in value)
if isinstance(value, dict):
return " ".join(ai_join_text(item) for item in value.values())
return str(value or "")
def ai_needs_review_text(value: Any) -> bool:
text = str(value or "").strip()
return bool(text) and not any(marker in text for marker in AI_NO_REVIEW_MARKERS)
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"] = [remaining]
else:
normalized["remaining_issues"] = []
return normalized
def ai_has_confirmed_problem(parsed: dict[str, Any]) -> bool:
resolution = str(parsed.get("issue_resolution") or "").strip().lower()
if ai_bool(parsed.get("confirmed_issue")):
return True
if resolution in {"confirmed_problem", "uncertain", "update_suggested", "problem", "待确认", "已证实"}:
return True
suggested_updates = parsed.get("suggested_updates")
if isinstance(suggested_updates, list) and suggested_updates:
return True
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 True
if resolution in {"false_positive", "ok", "no_issue", "误报", "无问题"}:
return False
text = ai_join_text(
{
"summary": parsed.get("summary"),
"evidence": parsed.get("evidence"),
}
)
if any(keyword in text for keyword in ("需人工", "需要人工", "待确认")):
return True
has_problem_word = any(keyword in text for keyword in AI_CONFIRMED_PROBLEM_KEYWORDS)
has_qualifier = any(keyword in text for keyword in AI_CONFIRMED_PROBLEM_QUALIFIERS)
return bool(has_problem_word and has_qualifier)
def ai_remaining_issues(parsed: dict[str, Any]) -> list[Any]:
issues = parsed.get("remaining_issues")
if isinstance(issues, list):
return [item for item in issues if str(item or "").strip()]
if issues:
return [issues]
return []
def ai_has_unresolved_problem(parsed: dict[str, Any]) -> bool:
if ai_remaining_issues(parsed):
return True
text = ai_join_text(
{
"summary": parsed.get("summary"),
"evidence": parsed.get("evidence"),
"decision": parsed.get("decision"),
}
)
compact = re.sub(r"\s+", "", text)
has_fixed_marker = any(marker in compact for marker in AI_FIXED_MARKERS)
has_force_problem_marker = any(marker in compact for marker in AI_FORCE_PROBLEM_MARKERS)
has_unresolved_marker = any(marker in compact for marker in AI_UNRESOLVED_PROBLEM_MARKERS)
if has_unresolved_marker:
if has_fixed_marker and not has_force_problem_marker:
return False
return True
if has_fixed_marker:
return False
unresolved_code_pattern = re.compile(
r"(主要诊断编码|手术及操作编码|手术操作编码|手术编码|疾病编码).{0,16}(空白|缺失|未填|未填写|漏填|为空)"
)
return bool(unresolved_code_pattern.search(compact))
def ai_classification(parsed: dict[str, Any]) -> str:
classification = str(parsed.get("classification") or parsed.get("category") or "").strip().lower()
decision = str(parsed.get("decision") or "").strip().lower()
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 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", "待确认", "已证实"}:
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
return AI_OK_STATUS
def ai_status_from_result(result: dict[str, Any]) -> str:
parsed = result.get("parsed") if isinstance(result.get("parsed"), dict) else {}
return ai_classification(parsed)
def ai_review_note(status: str, result: dict[str, Any]) -> str:
parsed = result.get("parsed") if isinstance(result.get("parsed"), dict) else {}
summary = str(parsed.get("summary") or "").strip()
verdict = "AI判断通过" if status == AI_OK_STATUS else "AI建议复核"
return f"AI视觉核验({datetime.now().strftime('%Y-%m-%d %H:%M:%S')}): {verdict}。{summary}".strip()
def blank_ai_value(value: Any) -> bool:
text = str(value or "").strip()
if not text:
return True
return any(marker in text for marker in ("空白", "未填写", "无内容", "可见但为空", "编码栏可见但为空", "null"))
def suggested_update_value(item: dict[str, Any]) -> Any:
for key in ("value", "new", "new_value", "pdf_value", "suggested_value"):
if key in item:
return item.get(key)
return None
def ai_outpatient_code_leak(path: tuple[str, int | None, str | None], value: Any, before: dict[str, Any], _current_value: Any) -> bool:
outpatient_code = str(before.get("outpatient_diagnosis_code") or "").strip()
if not outpatient_code or str(value or "").strip() != outpatient_code:
return False
field, _index, key = path
target_is_diagnosis_code = field == "primary_diagnosis_code" or (field == "discharge_diagnoses" and key == "疾病编码")
return bool(target_is_diagnosis_code)
PRIMARY_DIAGNOSIS_AI_TARGETS = {
"primary_diagnosis": "出院诊断",
"主要诊断": "出院诊断",
"主要诊断名称": "出院诊断",
"primary_diagnosis_code": "疾病编码",
"主要诊断编码": "疾病编码",
"primary_admission_condition": "入院病情",
"主要诊断入院病情": "入院病情",
}
def ai_update_path(field_text: str, item: dict[str, Any]) -> tuple[str, int | None, str | None] | None:
text = str(field_text or "").strip()
for keyword, column in sorted(PRIMARY_DIAGNOSIS_AI_TARGETS.items(), key=lambda entry: len(entry[0]), reverse=True):
if keyword in text:
return ("discharge_diagnoses", -1, column)
if text in EDITABLE_FIELDS:
return (text, None, None)
for name, meta in FIELD_META.items():
if text == meta["label"]:
return (name, None, None)
match = re.match(r"^(operations|discharge_diagnoses|fee_details)\[(\d+)\][.。．]?(.*)$", text)
if match:
return (match.group(1), int(match.group(2)), match.group(3).strip() or None)
row_index = item.get("row_index")
try:
index = int(row_index) if row_index not in {None, ""} else 0
except (TypeError, ValueError):
index = 0
operation_columns = {"手术操作编码", "手术操作日期", "手术级别", "手术操作名称", "术者", "I助", "II助", "切口愈合等级", "麻醉方式", "麻醉医师", "原始内容"}
diagnosis_columns = {"诊断类别", "出院诊断", "疾病编码", "入院病情"}
for column in operation_columns:
if column in text:
return ("operations", index, column)
for column in diagnosis_columns:
if column in text:
return ("discharge_diagnoses", index, column)
return None
def ai_suggested_updates(result: dict[str, Any], before: dict[str, Any]) -> tuple[dict[str, Any], list[dict[str, Any]]]:
parsed = result.get("parsed") if isinstance(result.get("parsed"), dict) else {}
items = parsed.get("suggested_updates")
if not isinstance(items, list):
return {}, []
updates: dict[str, Any] = {}
changed_fields: list[dict[str, Any]] = []
working = {key: json_ready_deep(before.get(key)) for key in EDITABLE_FIELDS}
for item in items:
if not isinstance(item, dict):
continue
value = suggested_update_value(item)
if blank_ai_value(value):
continue
path = ai_update_path(str(item.get("field") or item.get("target") or ""), item)
if not path:
continue
field, index, key = path
if field not in EDITABLE_FIELDS:
continue
old_value = working.get(field)
if index is None:
if ai_outpatient_code_leak(path, value, before, old_value):
continue
try:
new_value = parse_field_value(field, value)
except Exception:
continue
label = FIELD_META[field]["label"]
compare_old = old_value
else:
if field not in JSON_FIELDS or key is None:
continue
rows = old_value if isinstance(old_value, list) else []
rows = [dict(row) if isinstance(row, dict) else {} for row in rows]
if index == -1 and field == "discharge_diagnoses":
if not rows:
rows = [{"诊断类别": "主要诊断", "出院诊断": "", "疾病编码": "", "入院病情": ""}]
index = main_diagnosis_row_index(rows)
if index < 0 or index >= len(rows):
continue
compare_old = rows[index].get(key)
if ai_outpatient_code_leak(path, value, before, 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 comparable(compare_old) == comparable(value):
continue
working[field] = new_value
updates[field] = new_value
changed_fields.append(
{
"field": field if index is None else f"{field}[{index}].{key}",
"label": label,
"old": json_ready_deep(compare_old),
"new": json_ready_deep(value),
}
)
sync_primary_diagnosis_updates(updates, before)
return updates, changed_fields
def apply_ai_review(record_id: int, kimi_override: dict[str, Any] | None = None) -> dict[str, Any]:
before = fetch_record(record_id)
result = call_kimi_ai_review(before, kimi_override)
ai_updates, changed_fields = ai_suggested_updates(result, before)
new_status = ai_status_from_result(result)
note = ai_review_note(new_status, result)
assignments = [sql.SQL("review_status = %s")]
values: list[Any] = [new_status]
for field, value in ai_updates.items():
assignments.append(sql.SQL("{} = %s").format(sql.Identifier(field)))
values.append(psycopg2.extras.Json(value, dumps=lambda obj: json.dumps(obj, ensure_ascii=False)) if field in JSON_DB_FIELDS else value)
with connect() as conn, conn.cursor() as cur:
cur.execute(
sql.SQL("UPDATE {table} SET {assignments} WHERE id = %s").format(
table=table_identifier(),
assignments=sql.SQL(", ").join(assignments),
),
[*values, record_id],
)
insert_review_log(cur, record_id, before.get("source_file", ""), changed_fields, note, changed_by="AI", ai_result=result)
conn.commit()
return {"record_id": record_id, "status": new_status, "result": result}
def apply_ai_review_with_retry(record_id: int, attempts: int = 3, kimi_override: dict[str, Any] | None = None) -> dict[str, Any]:
last_exc: Exception | None = None
for attempt in range(attempts):
try:
return apply_ai_review(record_id, kimi_override)
except Exception as exc: # noqa: BLE001
last_exc = exc
if attempt + 1 < attempts:
time.sleep(4 + attempt * 4)
if last_exc:
raise last_exc
raise RuntimeError("AI核验失败")
def ai_target_ids(scope: str, record_id: int | None) -> list[int]:
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":
return [int(record_id)]
if scope == "ai_pending":
query = sql.SQL("SELECT id FROM {table} WHERE review_status = %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]
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] = []
if scope == "five":
where = sql.SQL("review_status = 'needs_review' AND id > %s")
params.append(record_id)
limit_sql = sql.SQL("LIMIT 5")
elif scope == "fifty":
where = sql.SQL("review_status = 'needs_review' AND id > %s")
params.append(record_id)
limit_sql = sql.SQL("LIMIT 50")
else:
if record_id:
where = sql.SQL("review_status = 'needs_review' AND id > %s")
params.append(record_id)
limit_sql = sql.SQL("")
query = sql.SQL("SELECT id FROM {table} WHERE {where} ORDER BY id {limit}").format(
table=table_identifier(),
where=where,
limit=limit_sql,
)
with connect() as conn, conn.cursor() as cur:
cur.execute(query, params)
rows = cur.fetchall()
return [int(row["id"]) for row in rows]
def update_ai_job(**updates: Any) -> dict[str, Any]:
with AI_JOB_LOCK:
AI_REVIEW_JOB.update(updates)
return dict(AI_REVIEW_JOB)
def is_ai_stop_error(message: str) -> bool:
lower = message.lower()
return any(marker in lower for marker in AI_STOP_ERROR_MARKERS)
def append_ai_job_error(record_id: int, message: str) -> None:
errors = AI_REVIEW_JOB.setdefault("errors", [])
errors.append({"record_id": record_id, "message": message})
if len(errors) > AI_JOB_ERROR_LIMIT:
del errors[: len(errors) - AI_JOB_ERROR_LIMIT]
def ai_cancel_requested() -> bool:
with AI_JOB_LOCK:
return bool(AI_REVIEW_JOB.get("cancel_requested"))
def run_ai_review_job(scope: str, ids: list[int], kimi_override: dict[str, Any] | None = None) -> None:
settings = public_kimi_settings(load_local_settings().get("kimi") or {})
concurrency = min(max(1, int(settings.get("concurrency") or 3)), max(1, len(ids)))
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,
total=len(ids),
processed=0,
ok=0,
pending=0,
failed=0,
concurrency=concurrency,
model=str((kimi_override or {}).get("model") or settings.get("model") or ""),
thinking_enabled=bool((kimi_override or {}).get("thinking_enabled") if (kimi_override or {}).get("thinking_enabled") is not None else settings.get("thinking_enabled")),
message="AI核验中",
errors=[],
started_at=datetime.now().isoformat(timespec="seconds"),
finished_at="",
last_record_id=None,
privacy_mode=privacy_mode,
)
record_queue: Queue[int] = Queue()
for record_id in ids:
record_queue.put(record_id)
def worker() -> None:
try:
while not stop_event.is_set():
if ai_cancel_requested():
with AI_JOB_LOCK:
AI_REVIEW_JOB["message"] = "AI核验正在中断..."
stop_event.set()
return
try:
record_id = record_queue.get_nowait()
except Empty:
return
try:
item = apply_ai_review_with_retry(record_id, kimi_override=kimi_override)
status = item.get("status")
with AI_JOB_LOCK:
AI_REVIEW_JOB["processed"] += 1
AI_REVIEW_JOB["last_record_id"] = record_id
if status == AI_NO_ISSUE_STATUS:
AI_REVIEW_JOB["ok"] += 1
elif status == AI_PENDING_STATUS:
AI_REVIEW_JOB["pending"] += 1
except Exception as exc: # noqa: BLE001
message = str(getattr(exc, "detail", exc))
with AI_JOB_LOCK:
AI_REVIEW_JOB["processed"] += 1
AI_REVIEW_JOB["failed"] += 1
AI_REVIEW_JOB["last_record_id"] = record_id
AI_REVIEW_JOB["message"] = message
append_ai_job_error(record_id, message)
if is_ai_stop_error(message):
AI_REVIEW_JOB["message"] = f"AI核验已暂停：{message}"
stop_event.set()
finally:
record_queue.task_done()
except Exception as exc: # noqa: BLE001
with AI_JOB_LOCK:
AI_REVIEW_JOB["message"] = str(exc)
workers = [
threading.Thread(target=worker, name=f"kimi-ai-worker-{index + 1}", daemon=True)
for index in range(concurrency)
]
for thread in workers:
thread.start()
for thread in workers:
thread.join()
refresh_status_snapshot(source="ai")
with AI_JOB_LOCK:
failed = int(AI_REVIEW_JOB.get("failed") or 0)
cancelled = bool(AI_REVIEW_JOB.get("cancel_requested"))
stopped = stop_event.is_set()
message = str(AI_REVIEW_JOB.get("message") or "")
update_ai_job(
running=False,
message="AI核验已中断" if cancelled else (message if stopped else ("AI核验完成" if failed == 0 else f"AI核验完成，失败 {failed} 条")),
finished_at=datetime.now().isoformat(timespec="seconds"),
)
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(
"""
UPDATE {table}
SET review_status = 'reviewed',
manual_corrected = TRUE,
review_logs = COALESCE(review_logs, '[]'::jsonb) || jsonb_build_array(
jsonb_build_object(
'id', %s || id::text,
'record_id', id,
'source_file', source_file,
'changed_at', %s,
'changed_by', 'web',
'manual_note', %s,
'changed_fields', jsonb_build_array(
jsonb_build_object('field', 'review_status', 'label', '复核状态', 'old', review_status, 'new', 'reviewed'),
jsonb_build_object('field', 'manual_corrected', 'label', '人工修正', 'old', COALESCE(manual_corrected, false), 'new', true)
)
)
)
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, record_ids))
count = cur.rowcount
conn.commit()
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')}): 已从患者首页复核工作台隐藏"
query = sql.SQL(
"""
UPDATE {table}
SET review_status = '已提交',
review_logs = COALESCE(review_logs, '[]'::jsonb) || jsonb_build_array(
jsonb_build_object(
'id', %s || id::text,
'record_id', id,
'source_file', source_file,
'changed_at', %s,
'changed_by', 'web',
'manual_note', %s,
'changed_fields', jsonb_build_array(
jsonb_build_object('field', 'review_status', 'label', '复核状态', 'old', review_status, 'new', '已提交')
)
)
)
WHERE review_status = 'reviewed'
"""
).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))
count = cur.rowcount
conn.commit()
refresh_status_snapshot(source="submit")
return int(count)
def parse_field_value(field: str, value: Any) -> Any:
meta = FIELD_META[field]
field_type = meta["type"]
if value == "":
return [] if field in JSON_FIELDS else None
if field in JSON_FIELDS:
if isinstance(value, str):
try:
return json.loads(value)
except json.JSONDecodeError as exc:
raise HTTPException(status_code=400, detail=f"{meta['label']} 不是合法 JSON：{exc}") from exc
return value
if field in INTEGER_FIELDS:
return None if value is None else int(value)
if field in NUMERIC_FIELDS:
return None if value is None else Decimal(str(value))
return value
def fetch_record(record_id: int) -> dict[str, Any]:
query = sql.SQL("SELECT * FROM {table} WHERE id = %s").format(table=table_identifier())
with connect() as conn, conn.cursor() as cur:
cur.execute(query, (record_id,))
row = cur.fetchone()
if not row:
raise HTTPException(status_code=404, detail="记录不存在")
record = row_to_json(dict(row))
record["pdf_url"] = f"/api/pdf/{record['source_file']}" if get_pdf_path(record["source_file"]) else ""
record["review_logs"] = fetch_review_logs(record_id)
record["last_activity_at"] = record["review_logs"][0].get("changed_at") if record["review_logs"] else None
return record
@app.get("/")
def index():
return FileResponse(STATIC_DIR / "index.html")
@app.get("/favicon.ico")
def favicon():
return Response(status_code=204)
@app.post("/api/auth/login")
def login(payload: LoginPayload, response: Response):
user = authenticate_user(payload.username, payload.password)
if not user:
raise HTTPException(status_code=401, detail="用户名或密码错误")
token = secrets.token_urlsafe(32)
SESSIONS[token] = {
**user,
"login_at": datetime.now().isoformat(timespec="seconds"),
}
response.set_cookie(
SESSION_COOKIE,
token,
httponly=True,
samesite="lax",
max_age=12 * 60 * 60,
)
return {"authenticated": True, "user": SESSIONS[token]}
@app.post("/api/auth/logout")
def logout(request: Request, response: Response):
token = request.cookies.get(SESSION_COOKIE, "")
if token:
SESSIONS.pop(token, None)
response.delete_cookie(SESSION_COOKIE)
return {"ok": True}
@app.get("/api/auth/me")
def auth_me(request: Request):
user = session_from_request(request)
if not user:
return {"authenticated": False, "user": None}
return {"authenticated": True, "user": user}
@app.get("/api/status")
def status():
data = load_local_settings()
snapshot = data.get("status_snapshot") or default_status_snapshot()
system = normalize_system_settings(data.get("system") or {})
snapshot["next_check_at"] = next_status_check_at(system)
try:
query = sql.SQL(
"""
SELECT
count(*) AS total,
count(*) FILTER (WHERE review_status IN ('needs_review', 'reviewed', 'AI已处理-OK', 'AI已处理-不OK', 'AI复核-无问题', 'AI复核-待确认') OR (
review_status = 'auto_pass'
AND COALESCE(review_logs, '[]'::jsonb) @> jsonb_build_array(jsonb_build_object('changed_by', 'AI'))
)) AS workbench_total,
count(*) FILTER (WHERE review_status IN ('needs_review', 'AI已处理-不OK', 'AI复核-待确认')) AS review_needed,
count(*) FILTER (WHERE review_status = 'needs_review') AS needs_review,
count(*) FILTER (WHERE review_status = 'auto_pass') AS auto_passed,
count(*) FILTER (WHERE (
review_status = 'auto_pass'
AND COALESCE(review_logs, '[]'::jsonb) @> jsonb_build_array(jsonb_build_object('changed_by', 'AI'))
) OR review_status IN ('AI已处理-OK', 'AI复核-无问题')) AS ai_passed,
count(*) FILTER (WHERE review_status IN ('AI已处理-不OK', 'AI复核-待确认')) AS ai_pending,
count(*) FILTER (WHERE review_status = 'reviewed') AS reviewed,
count(*) FILTER (WHERE review_status = '已提交') AS submitted,
count(*) FILTER (WHERE manual_corrected IS TRUE) AS manual_corrected
FROM {table}
"""
).format(table=table_identifier())
with connect() as conn, conn.cursor() as cur:
cur.execute(query)
snapshot.update(row_to_json(dict(cur.fetchone())))
snapshot["database"] = "online"
snapshot["message"] = snapshot.get("message") or "连接正常"
snapshot["checked_at"] = status_now().isoformat(timespec="seconds")
except Exception as exc: # noqa: BLE001
snapshot["database"] = "offline"
snapshot["message"] = str(exc)
return snapshot
@app.get("/api/schema")
def schema():
return {"groups": FIELD_GROUPS}
@app.get("/api/overview")
def overview():
query = sql.SQL(
"""
SELECT
count(*) AS total,
count(*) FILTER (WHERE review_status IN ('needs_review', 'reviewed', 'AI已处理-OK', 'AI已处理-不OK', 'AI复核-无问题', 'AI复核-待确认') OR (
review_status = 'auto_pass'
AND COALESCE(review_logs, '[]'::jsonb) @> jsonb_build_array(jsonb_build_object('changed_by', 'AI'))
)) AS review_queue,
count(*) FILTER (WHERE review_status = 'needs_review') AS needs_review,
count(*) FILTER (WHERE (
review_status = 'auto_pass'
AND COALESCE(review_logs, '[]'::jsonb) @> jsonb_build_array(jsonb_build_object('changed_by', 'AI'))
) OR review_status IN ('AI已处理-OK', 'AI复核-无问题')) AS ai_passed,
count(*) FILTER (WHERE review_status IN ('AI已处理-不OK', 'AI复核-待确认')) AS ai_pending,
count(*) FILTER (WHERE review_status = 'auto_corrected') AS auto_corrected,
count(*) FILTER (WHERE review_status = 'reviewed') AS reviewed,
count(*) FILTER (WHERE review_status = '已提交') AS submitted,
count(*) FILTER (WHERE review_status = 'auto_pass') AS auto_passed,
count(*) FILTER (WHERE manual_corrected IS TRUE) AS manual_corrected
FROM {table}
"""
).format(table=table_identifier())
with connect() as conn, conn.cursor() as cur:
cur.execute(query)
summary = row_to_json(dict(cur.fetchone()))
summary.update({"audit_total": 0, "audit_pending": 0, "audit_passed": 0, "audit_failed": 0, "audit_unsure": 0})
return {
"summary": summary,
"recent_logs": [],
}
def record_filter_sql(q: str = "", status_filter: str = "review_all") -> tuple[sql.Composable, list[Any]]:
clauses = []
params: list[Any] = []
if q:
like = f"%{q}%"
clauses.append("(source_file ILIKE %s OR inpatient_no ILIKE %s OR medical_record_no ILIKE %s OR patient_name ILIKE %s OR primary_diagnosis ILIKE %s)")
params.extend([like, like, like, like, like])
if status_filter == "review_all":
clauses.append("(review_status IN ('needs_review', 'reviewed', 'AI已处理-OK', 'AI已处理-不OK', 'AI复核-无问题', 'AI复核-待确认') OR (review_status = 'auto_pass' AND COALESCE(review_logs, '[]'::jsonb) @> jsonb_build_array(jsonb_build_object('changed_by', 'AI'))))")
elif status_filter != "all":
if status_filter == "reviewed":
clauses.append("review_status = 'reviewed'")
elif status_filter in {"ai_passed", "AI已处理-OK", "AI复核-无问题"}:
clauses.append("(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'))))")
else:
clauses.append("review_status = %s")
params.append(status_filter)
if not clauses:
return sql.SQL(""), params
return sql.SQL("WHERE ") + sql.SQL(" AND ").join(sql.SQL(clause) for clause in clauses), params
RECORD_LIST_ORDER_SQL = sql.SQL(
"""
ORDER BY
last_activity_at DESC NULLS LAST,
CASE review_status
WHEN 'needs_review' THEN 1
WHEN 'AI已处理-不OK' THEN 2
WHEN 'AI复核-待确认' THEN 2
WHEN 'AI已处理-OK' THEN 3
WHEN 'AI复核-无问题' THEN 3
WHEN 'auto_pass' THEN 3
WHEN 'auto_corrected' THEN 4
WHEN 'reviewed' THEN 5
WHEN '已提交' THEN 6
ELSE 7
END,
id
"""
)
@app.get("/api/records")
def list_records(q: str = "", status_filter: str = "review_all", limit: int = 300, offset: int = 0):
limit = max(50, min(int(limit), 500))
offset = max(0, int(offset))
where_sql, params = record_filter_sql(q, status_filter)
query = sql.SQL(
"""
SELECT
id, source_file, inpatient_no, medical_record_no, patient_name, review_status, manual_corrected,
major_department, discharge_dept, primary_diagnosis, primary_diagnosis_code,
contact_phone,
COALESCE(review_logs, '[]'::jsonb) @> jsonb_build_array(jsonb_build_object('changed_by', 'AI')) AS ai_reviewed,
NULLIF(review_logs->-1->>'changed_at', '')::timestamp AS last_activity_at
FROM {table}
{where}
{order_by}
LIMIT %s OFFSET %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, limit + 1, offset])
rows = [row_to_json(dict(row)) for row in cur.fetchall()]
has_more = len(rows) > limit
rows = rows[:limit]
for row in rows:
row["has_pdf"] = get_pdf_path(row["source_file"]) is not None
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:
raise HTTPException(status_code=400, detail="AI提问图片索引无效")
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 文件不存在")
ai_logs = [log for log in fetch_review_logs(record_id, limit=100) if log.get("ai_result")]
if log_index >= len(ai_logs):
raise HTTPException(status_code=404, detail="AI提问记录不存在")
result = ai_logs[log_index].get("ai_result") or {}
modules = (result.get("pdf_context") or {}).get("modules") or []
if module_index >= len(modules):
raise HTTPException(status_code=404, detail="AI提问图片不存在")
module = modules[module_index] or {}
bbox = module.get("bbox") or []
if not isinstance(bbox, list) or len(bbox) != 4:
raise HTTPException(status_code=400, detail="AI提问图片定位无效")
try:
import fitz # type: ignore[import-not-found]
clip = fitz.Rect([float(value) for value in bbox])
page_index = max(0, int(module.get("page") or 1) - 1)
except Exception as exc: # noqa: BLE001
raise HTTPException(status_code=400, detail="AI提问图片定位无法解析") from exc
return Response(
content=render_pdf_page_png(pdf_path, dpi=120, page_index=page_index, clip=clip),
media_type="image/png",
headers={"Cache-Control": "no-store", "X-Content-Type-Options": "nosniff"},
)
@app.post("/api/records/{record_id}")
def update_record(record_id: int, payload: UpdatePayload):
updates, changed_fields, before = build_record_updates(
record_id,
payload.fields,
payload.manual_note,
payload.note_prefix or "人工复核",
force_reviewed=True,
)
if not updates and not changed_fields:
raise HTTPException(status_code=400, detail="没有可保存字段")
with connect() as conn, conn.cursor() as cur:
apply_record_updates(cur, record_id, updates, changed_fields, before, payload.manual_note, force_reviewed=True)
conn.commit()
return {"ok": True, "record": fetch_record(record_id)}
@app.post("/api/audit/sample")
def create_audit_sample(source: str = "reviewed", count: int = 5):
ensure_workflow_tables()
if source not in {"reviewed", "auto_pass"}:
raise HTTPException(status_code=400, detail="抽查来源只能是 reviewed 或 auto_pass")
count = max(1, min(int(count), 50))
status_clause = "review_status = 'reviewed'" if source == "reviewed" else "review_status = 'auto_pass'"
query = sql.SQL(
"""
SELECT *
FROM {table}
WHERE {status_clause}
ORDER BY random()
LIMIT %s
"""
).format(table=table_identifier(), status_clause=sql.SQL(status_clause))
with connect() as conn, conn.cursor() as cur:
cur.execute(query, (count,))
rows = [dict(row) for row in cur.fetchall()]
return {
"audit_source": source,
"records": [fetch_record(int(row["id"])) for row in rows],
}
@app.get("/api/audit/logs")
def list_audit_logs(limit: int = 100):
return {"logs": fetch_audit_logs(max(1, min(int(limit), 500)))}
@app.post("/api/audit/classify")
def classify_audit(payload: AuditClassifyPayload):
ensure_workflow_tables()
if payload.audit_status not in {"passed", "failed", "unsure"}:
raise HTTPException(status_code=400, detail="抽查归类只能是 passed/failed/unsure")
if payload.audit_source not in {"reviewed", "auto_pass"}:
raise HTTPException(status_code=400, detail="抽查来源只能是 reviewed 或 auto_pass")
updates, changed_fields, before = build_record_updates(
payload.record_id,
payload.fields,
"",
"抽查",
force_reviewed=False,
)
snapshot = {
key: json_ready_deep(before.get(key))
for key in [
"source_file",
"medical_record_no",
"patient_name",
"primary_diagnosis",
"primary_diagnosis_code",
"discharge_diagnoses",
"operations",
"review_status",
]
}
now = datetime.now().isoformat(timespec="seconds")
log = {
"id": datetime.now().strftime("%Y%m%d%H%M%S%f"),
"record_id": payload.record_id,
"source_file": before.get("source_file"),
"audit_source": payload.audit_source,
"audit_status": payload.audit_status,
"audit_notes": payload.audit_notes.strip(),
"ai_result": None,
"snapshot": {**snapshot, "changed_fields": changed_fields},
"created_at": now,
"updated_at": now,
}
with connect() as conn, conn.cursor() as cur:
apply_record_updates(
cur,
payload.record_id,
updates,
changed_fields,
before,
payload.audit_notes if changed_fields else "",
force_reviewed=False,
)
cur.execute(
sql.SQL(
"""
UPDATE {table}
SET audit_logs = COALESCE(audit_logs, '[]'::jsonb) || %s::jsonb
WHERE id = %s
"""
).format(table=table_identifier()),
(
json.dumps([json_ready_deep(log)], ensure_ascii=False),
payload.record_id,
),
)
conn.commit()
return {"ok": True, "log": row_to_json(log), "record": fetch_record(payload.record_id)}
@app.post("/api/audit/logs/{audit_id}")
def update_audit_log(audit_id: int, payload: AuditPayload):
raise HTTPException(status_code=410, detail="抽查日志已并入主表，请使用归类保存接口")
@app.get("/api/settings")
def get_settings():
return public_settings()
@app.post("/api/settings/status/check")
def check_status_now():
return {"status_snapshot": refresh_status_snapshot(source="manual")}
@app.post("/api/settings/system")
def update_system_settings(payload: SystemSettingsPayload):
data = load_local_settings()
system = normalize_system_settings(data.get("system") or {})
system["status_check_time"] = normalize_status_check_time(payload.status_check_time)
data["system"] = system
snapshot = data.get("status_snapshot") or default_status_snapshot()
snapshot["next_check_at"] = next_status_check_at(system)
data["status_snapshot"] = snapshot
save_local_settings(data)
return public_settings()
@app.post("/api/settings/kimi")
def update_kimi_settings(payload: KimiSettingsPayload):
data = load_local_settings()
current_kimi = data.get("kimi") or {}
api_key = payload.api_key.strip() or str(current_kimi.get("api_key") or "").strip()
data["kimi"] = normalize_kimi_settings(
{
"enabled": payload.enabled,
"model": payload.model,
"api_base": payload.api_base or current_kimi.get("api_base") or DEFAULT_KIMI_API_BASE,
"api_key": api_key,
"concurrency": payload.concurrency,
"thinking_enabled": payload.thinking_enabled,
"ai_scope_mode": payload.ai_scope_mode,
"ai_action_modes": payload.ai_action_modes,
"ai_action_privacy_modes": payload.ai_action_privacy_modes,
}
)
save_local_settings(data)
return public_settings()
@app.get("/api/ai/config")
def ai_config():
data = load_local_settings()
return {"kimi": public_kimi_settings(data.get("kimi") or {})}
@app.get("/api/ai/review/status")
def ai_review_status():
with AI_JOB_LOCK:
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:
if AI_REVIEW_JOB.get("running"):
AI_REVIEW_JOB["cancel_requested"] = True
AI_REVIEW_JOB["message"] = "AI核验正在中断..."
return dict(AI_REVIEW_JOB)
@app.post("/api/ai/review/ack")
def ack_ai_review():
with AI_JOB_LOCK:
if AI_REVIEW_JOB.get("running"):
return dict(AI_REVIEW_JOB)
AI_REVIEW_JOB.update(
running=False,
cancel_requested=False,
scope="",
total=0,
processed=0,
ok=0,
pending=0,
failed=0,
concurrency=0,
message="",
errors=[],
started_at="",
finished_at="",
last_record_id=None,
)
return dict(AI_REVIEW_JOB)
@app.post("/api/ai/review/approve-no-issue")
def approve_ai_no_issue():
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")
def start_ai_review(payload: AiReviewPayload):
local_kimi_settings = load_local_settings().get("kimi") or {}
settings = public_kimi_settings(local_kimi_settings)
if not settings["available"]:
raise HTTPException(status_code=400, detail="AI 核验未启用或未配置 API Key")
if not ai_scope_allowed(str(settings.get("ai_scope_mode") or "all"), payload.scope):
raise HTTPException(status_code=403, detail="当前设置未开放这个 AI 处理范围")
with AI_JOB_LOCK:
if AI_REVIEW_JOB.get("running"):
raise HTTPException(status_code=409, detail="已有 AI 核验任务正在运行")
ids = ai_target_ids(payload.scope, payload.record_id)
if not ids:
raise HTTPException(status_code=400, detail="当前范围没有可 AI 核验的需复核记录")
privacy_modes = normalize_ai_action_privacy_modes(local_kimi_settings.get("ai_action_privacy_modes"))
privacy_mode = (
normalize_bool(payload.privacy_mode, privacy_modes.get(payload.scope, True))
if payload.privacy_mode is not None
else privacy_modes.get(payload.scope, True)
)
kimi_override = {
"model": payload.model,
"thinking_enabled": payload.thinking_enabled,
"privacy_mode": privacy_mode,
}
thread = threading.Thread(
target=run_ai_review_job,
args=(payload.scope, ids, kimi_override),
name="kimi-ai-review",
daemon=True,
)
thread.start()
time.sleep(0.1)
with AI_JOB_LOCK:
return dict(AI_REVIEW_JOB)
@app.post("/api/settings/submit-reviewed")
def submit_reviewed():
count = submit_reviewed_records()
return {"ok": True, "updated": count, "status_snapshot": load_local_settings().get("status_snapshot")}
@app.post("/api/settings/users")
def create_user(payload: UserPayload):
data = load_local_settings()
username = validate_local_username(payload.username, data)
if not payload.password:
raise HTTPException(status_code=400, detail="新用户必须设置密码")
user = {
"username": username,
"permissions": clean_permissions(payload.permissions),
"created_at": datetime.now().isoformat(timespec="seconds"),
}
user.update(password_hash(payload.password))
data.setdefault("users", []).append(user)
save_local_settings(data)
return public_settings()
@app.post("/api/settings/users/{username}")
def update_user(username: str, payload: UserUpdatePayload):
if username == admin_username():
raise HTTPException(status_code=400, detail="环境变量管理员不能在网页端编辑")
data = load_local_settings()
index = local_user_index(data, username)
if index is None:
raise HTTPException(status_code=404, detail="只能编辑本地配置用户")
user = data["users"][index]
new_username = validate_local_username(payload.username or username, data, current_username=username)
user["username"] = new_username
user["permissions"] = clean_permissions(payload.permissions)
if payload.password:
user.update(password_hash(payload.password))
user["updated_at"] = datetime.now().isoformat(timespec="seconds")
save_local_settings(data)
return public_settings()
@app.post("/api/settings/users/{username}/password")
def update_user_password(username: str, payload: PasswordPayload):
if username == admin_username():
raise HTTPException(status_code=400, detail="环境变量管理员密码请在 .env 中修改")
if not payload.password:
raise HTTPException(status_code=400, detail="密码不能为空")
data = load_local_settings()
index = local_user_index(data, username)
if index is None:
raise HTTPException(status_code=404, detail="只能修改本地配置用户")
data["users"][index].update(password_hash(payload.password))
data["users"][index]["updated_at"] = datetime.now().isoformat(timespec="seconds")
save_local_settings(data)
return public_settings()
@app.post("/api/settings/users/{username}/permissions")
def update_user_permissions(username: str, payload: PermissionPayload):
if username == admin_username():
raise HTTPException(status_code=400, detail="环境变量管理员不能在网页端编辑")
data = load_local_settings()
for user in data.get("users", []):
if user.get("username") == username:
user["permissions"] = clean_permissions(payload.permissions)
user["updated_at"] = datetime.now().isoformat(timespec="seconds")
save_local_settings(data)
return public_settings()
raise HTTPException(status_code=404, detail="只能修改本地配置用户")
@app.delete("/api/settings/users/{username}")
def delete_user(username: str):
if username == admin_username():
raise HTTPException(status_code=400, detail="环境变量管理员不能删除")
data = load_local_settings()
users = data.get("users", [])
next_users = [user for user in users if user.get("username") != username]
if len(next_users) == len(users):
raise HTTPException(status_code=404, detail="只能删除本地配置用户")
data["users"] = next_users
save_local_settings(data)
return public_settings()
@app.get("/api/pdf/{source_file:path}")
def pdf_file(source_file: str, request: Request):
referer = request.headers.get("referer", "")
host = request.headers.get("host", "")
if not referer:
raise HTTPException(status_code=403, detail="PDF 只能在工作台内预览")
if host and urlparse(referer).netloc != host:
raise HTTPException(status_code=403, detail="PDF 只能在同源工作台内预览")
path = get_pdf_path(source_file)
if not path:
raise HTTPException(status_code=404, detail="PDF 文件不存在")
return FileResponse(
path,
media_type="application/pdf",
headers={
"Content-Disposition": "inline",
"Cache-Control": "no-store",
"X-Content-Type-Options": "nosniff",
},
)
Reference in New Issue View Git Blame Copy Permalink