from __future__ import annotations import json import os import base64 import hashlib import secrets import re import threading import time import urllib.error import urllib.request 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 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"(? 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 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 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. PDF局部截图可能因遮挡、截屏边界或缩放只显示编码/文字前半段；如果 PDF 显示值是结构化字段值的前缀，且没有相反证据，应判为 match/ok，不要写“需确认完整编码”，也不要建议把结构化字段截短。 10. 入院途径、离院方式、入院病情等代码字段只核对代码本身；例如 PDF 显示“2（门诊）”而结构化字段为“2”就是一致，不要要求复核代码与中文标签关系。 11. 手术操作日期只有在日历日期确实早于入院日期时才算问题；同日或晚于入院日期均为正常，不要推测月份应改为其他月份。 12. 手术及操作编码允许带 x 和 001/002/005/006 等扩展后缀；如果原始内容显示“54.5100x ... 005”这类拆开的后缀，应建议写入完整编码“54.5100x005”，不要要求人工确认扩展码是否有效。 13. remaining_issues 只写当前文档复核人应该特别注意的内容；不要写如何修改，不要重复 suggested_updates，不要写“无需补录/无需处理/首页原貌”这类已判定无问题的说明。 14. 不要编造 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 ai_compact_compare_value(value: Any) -> str: text = str(value or "").strip() text = text.replace("．", ".").replace("Ｘ", "X").replace("ｘ", "x").replace("×", "x") return re.sub(r"\s+", "", text).upper() def ai_pdf_value_is_structured_prefix(pdf_value: Any, structured_value: Any) -> bool: pdf_text = ai_compact_compare_value(pdf_value) structured_text = ai_compact_compare_value(structured_value) if len(pdf_text) < 3 or len(structured_text) <= len(pdf_text): return False return structured_text.startswith(pdf_text) def ai_text_pdf_prefix_issue(text: Any) -> bool: compact = re.sub(r"\s+", "", str(text or "")) if not compact: return False patterns = [ r"PDF(?:显示|中显示|可见|值)?(?:为|是|[:：])?([A-Za-z0-9][A-Za-z0-9.+\-*/xX]*)[，,；;。、]?(?:结构化字段|结构化值|字段|系统字段)(?:为|是|[:：])?([A-Za-z0-9][A-Za-z0-9.+\-*/xX]*)", r"PDF[^，,；;。]*?([A-Za-z0-9][A-Za-z0-9.+\-*/xX]{2,})[^，,；;。]*?(?:结构化字段|结构化值|字段|系统字段)[^，,；;。]*?([A-Za-z0-9][A-Za-z0-9.+\-*/xX]{2,})", ] return any( ai_pdf_value_is_structured_prefix(match.group(1), match.group(2)) for pattern in patterns for match in re.finditer(pattern, compact, flags=re.IGNORECASE) ) def ai_pdf_prefix_truncation_issue(value: Any) -> bool: if isinstance(value, dict): pdf_value = value.get("pdf_value") or value.get("pdf") or value.get("PDF值") or value.get("图片值") structured_value = ( value.get("structured_value") or value.get("structured") or value.get("field_value") or value.get("结构化值") or value.get("结构化字段") ) if ai_pdf_value_is_structured_prefix(pdf_value, structured_value): return True return ai_text_pdf_prefix_issue(value) return ai_text_pdf_prefix_issue(value) def ai_extract_leading_code(value: Any) -> str: text = str(value or "").strip() text = text.replace("（", "(").replace("）", ")").replace("：", ":") match = re.match(r"^\s*([A-Za-z]?\d+(?:[.\-xX]\d+)*)", text) return ai_compact_compare_value(match.group(1)) if match else "" def ai_code_label_text(value: Any) -> bool: text = str(value or "") return bool(re.search(r"(代码|编码|入院途径|离院方式|入院病情|[A-Za-z_][A-Za-z0-9_]*_code)", text, flags=re.IGNORECASE)) def ai_code_label_values_match(pdf_value: Any, structured_value: Any) -> bool: pdf_code = ai_extract_leading_code(pdf_value) structured_code = ai_extract_leading_code(structured_value) if not pdf_code or not structured_code or pdf_code != structured_code: return False pdf_text = str(pdf_value or "") structured_text = str(structured_value or "") return pdf_text != structured_text and ( bool(re.search(r"[（(][^）)]{1,12}[）)]", pdf_text)) or ai_code_label_text(pdf_text + structured_text) ) def ai_text_code_label_consistent_issue(text: Any) -> bool: raw = str(text or "") if not raw or "PDF" not in raw or not ai_code_label_text(raw): return False if re.search(r"(缺失|为空|未填|未填写|漏填)", raw): return False pdf_match = re.search( r"PDF[^，,；;。]*?(?:显示|勾选|可见|值)?[^，,；;。]*?(?:为|是|=)?\s*[\"'“”‘’]?([A-Za-z]?\d+(?:[.\-xX]\d+)*)(?:[）)\"'“”‘’]|[（(][^）)]{1,12}[）)])?", raw, flags=re.IGNORECASE, ) structured_match = re.search( r"(?:结构化字段|结构化值|结构化|[A-Za-z_][A-Za-z0-9_]*_code|[A-Za-z_][A-Za-z0-9_]*字段值|字段值)[^，,；;。]*?(?:是否为|为|是|=)?\s*[\"'“”‘’]?([A-Za-z]?\d+(?:[.\-xX]\d+)*)", raw, flags=re.IGNORECASE, ) if structured_match and not pdf_match and re.search(r"(与PDF一致|与PDF相符)", raw): return True if pdf_match and not structured_match and re.search(r"(与结构化一致|与PDF一致|是否与结构化一致|是否匹配|是否正确)", raw): return True if not pdf_match or not structured_match: return False return ai_compact_compare_value(pdf_match.group(1)) == ai_compact_compare_value(structured_match.group(1)) def ai_code_label_consistent_issue(value: Any) -> bool: if isinstance(value, dict): pdf_value = value.get("pdf_value") or value.get("pdf") or value.get("PDF值") or value.get("图片值") structured_value = ( value.get("structured_value") or value.get("structured") or value.get("field_value") or value.get("结构化值") or value.get("结构化字段") ) if ai_code_label_values_match(pdf_value, structured_value) and ai_code_label_text(ai_join_text(value)): return True return ai_text_code_label_consistent_issue(ai_join_text(value)) return ai_text_code_label_consistent_issue(value) def ai_parse_date_token(value: str) -> date | None: match = re.match(r"(\d{4})[-/年](\d{1,2})[-/月](\d{1,2})", value.strip()) if not match: return None try: return date(int(match.group(1)), int(match.group(2)), int(match.group(3))) except ValueError: return None def ai_dates_in_text(text: str) -> list[tuple[date, int]]: dates: list[tuple[date, int]] = [] for match in re.finditer(r"\d{4}[-/年]\d{1,2}[-/月]\d{1,2}", text): parsed = ai_parse_date_token(match.group(0)) if parsed: dates.append((parsed, match.start())) return dates def ai_operation_date_not_early_issue(value: Any) -> bool: text = ai_join_text(value) if not text or not re.search(r"(手术操作日期|手术日期|手术时间)", text): return False if not re.search(r"(早于入院|入院前|同一天|是否早于)", text): return False if re.search(r"PDF.{0,24}\d{4}[-/年]\d{1,2}[-/月]\d{1,2}.{0,24}结构化(?:字段|值)", text): return False admission_match = re.search(r"入院(?:时间|日期)?[^0-9]{0,12}(\d{4}[-/年]\d{1,2}[-/月]\d{1,2})", text) if not admission_match: return False admission_date = ai_parse_date_token(admission_match.group(1)) if not admission_date: return False operation_dates = [ item_date for item_date, position in ai_dates_in_text(text) if position < admission_match.start() ] if not operation_dates: operation_dates = [ item_date for item_date, position in ai_dates_in_text(text) if position != admission_match.start(1) ] if not operation_dates: return False return all(item_date >= admission_date for item_date in operation_dates) def ai_false_positive_issue(value: Any) -> bool: return ( ai_pdf_prefix_truncation_issue(value) or ai_code_label_consistent_issue(value) or ai_operation_date_not_early_issue(value) ) def ai_has_nonblank_suggested_updates(parsed: dict[str, Any]) -> bool: suggested_updates = parsed.get("suggested_updates") if not isinstance(suggested_updates, list): return False return any(isinstance(item, dict) and not blank_ai_value(suggested_update_value(item)) for item in suggested_updates) def ai_problem_evidence(parsed: dict[str, Any]) -> list[dict[str, Any]]: evidence = parsed.get("evidence") if not isinstance(evidence, list): return [] problem_results = {"mismatch", "uncertain", "missing", "problem", "not_match", "not ok", "not_ok"} return [ item for item in evidence if isinstance(item, dict) and str(item.get("result") or "").strip().lower() in problem_results ] def ai_only_false_positive(parsed: dict[str, Any]) -> bool: if ai_has_nonblank_suggested_updates(parsed): return False issues = ai_remaining_issues(parsed) if issues: return all(ai_false_positive_issue(issue) for issue in issues) evidence = ai_problem_evidence(parsed) if evidence: return all(ai_false_positive_issue(item) for item in evidence) return ai_false_positive_issue(parsed.get("summary")) or ai_false_positive_issue(parsed.get("decision")) def ai_only_pdf_prefix_truncation(parsed: dict[str, Any]) -> bool: return ai_only_false_positive(parsed) 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) and not ai_false_positive_issue(item) ] elif remaining and ai_needs_review_text(remaining) and not ai_false_positive_issue(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"} and not ai_false_positive_issue(item): 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_only_false_positive(parsed): return False 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 ai_only_false_positive(parsed): return AI_OK_STATUS if classification in ok_values or decision in ok_values or resolution in {"false_positive", "ok", "no_issue", "误报", "无问题", "通过"}: return AI_OK_STATUS if classification in problem_values or decision in problem_values or resolution in {"confirmed_problem", "uncertain", "problem", "待确认", "已证实"}: if not ai_remaining_issues(parsed) and ai_has_nonblank_suggested_updates(parsed): return AI_OK_STATUS if ( not ai_remaining_issues(parsed) and not any(not ai_false_positive_issue(item) for item in ai_problem_evidence(parsed)) and not ai_has_unresolved_problem(parsed) ): return AI_OK_STATUS return AI_PROBLEM_STATUS if ai_remaining_issues(parsed) or ai_has_unresolved_problem(parsed): return AI_PROBLEM_STATUS suggested_updates = parsed.get("suggested_updates") if isinstance(suggested_updates, list) and suggested_updates: return AI_OK_STATUS if any(not ai_false_positive_issue(item) for item in ai_problem_evidence(parsed)): 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 normalize_operation_code(value: Any) -> str: text = str(value or "").strip() text = re.sub(r"\s+", " ", text) match = re.match(r"^([A-Za-z]?\d{1,3}\.\d{1,4}x?)\s+(\d{3})$", text, flags=re.IGNORECASE) if match: return f"{match.group(1)}{match.group(2)}" return text 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 field == "operations" and key == "手术操作编码": value = normalize_operation_code(value) if ai_pdf_value_is_structured_prefix(value, compare_old): continue if comparable(compare_old) == comparable(value): continue rows[index][key] = value new_value = rows label = f"{FIELD_META[field]['label']}[{index + 1}].{key}" if index is None and ai_pdf_value_is_structured_prefix(value, compare_old): continue if comparable(compare_old) == comparable(value): continue working[field] = new_value 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/{record_id}") def get_record(record_id: int): return {"record": fetch_record(record_id)} @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", }, )