#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 把患者首页 PDF 转为图片，并生成图片-结构化字段对照核验清单。 依赖系统命令 pdftoppm，通常由 poppler-utils 提供。 """ from __future__ import annotations import argparse import csv import html import json import re import shutil import subprocess from dataclasses import dataclass from pathlib import Path from string import Template from typing import Any PROJECT_ROOT = Path(__file__).resolve().parents[2] DEFAULT_INPUT_DIR = PROJECT_ROOT / "待处理-患者首页PDF" DEFAULT_RESULT_DIR = PROJECT_ROOT / "数据处理结果区" DEFAULT_IMAGE_REVIEW_DIR = DEFAULT_RESULT_DIR / "06_PDF图片对照" @dataclass(frozen=True) class FieldCheck: group: str key: str level: str location_hint: str FIELD_CHECKS = [ FieldCheck("基本信息", "医疗机构", "recommended", "首页抬头及组织机构代码附近"), FieldCheck("基本信息", "医疗付费方式", "recommended", "首页左上付费方式"), FieldCheck("基本信息", "健康卡号", "recommended", "首页左上健康卡号"), FieldCheck("基本信息", "住院次数", "recommended", "首页左上住院次数"), FieldCheck("基本信息", "病案号", "required", "首页左上病案号"), FieldCheck("基本信息", "姓名", "required", "基本信息行"), FieldCheck("基本信息", "性别", "required", "基本信息行"), FieldCheck("基本信息", "出生日期", "required", "基本信息行"), FieldCheck("基本信息", "年龄", "required", "基本信息行"), FieldCheck("基本信息", "国籍", "recommended", "基本信息行"), FieldCheck("基本信息", "身份证号", "required", "基本信息行"), FieldCheck("基本信息", "职业", "recommended", "基本信息行"), FieldCheck("基本信息", "婚姻代码", "recommended", "基本信息行"), FieldCheck("基本信息", "出生地", "recommended", "地址信息区"), FieldCheck("基本信息", "籍贯", "recommended", "地址信息区"), FieldCheck("基本信息", "民族", "recommended", "地址信息区"), FieldCheck("地址联系人", "现住址", "required", "现住址行"), FieldCheck("地址联系人", "现住址电话", "recommended", "现住址行"), FieldCheck("地址联系人", "现住址邮编", "recommended", "现住址行"), FieldCheck("地址联系人", "户口地址", "recommended", "户口地址行"), FieldCheck("地址联系人", "户口地址邮编", "recommended", "户口地址行"), FieldCheck("地址联系人", "工作单位及地址", "recommended", "工作单位及地址行"), FieldCheck("地址联系人", "单位电话", "recommended", "工作单位及地址行"), FieldCheck("地址联系人", "单位邮编", "recommended", "工作单位及地址行"), FieldCheck("地址联系人", "联系人姓名", "required", "联系人信息行"), FieldCheck("地址联系人", "联系人关系", "recommended", "联系人信息行"), FieldCheck("地址联系人", "联系人地址", "recommended", "联系人信息行"), FieldCheck("地址联系人", "联系人电话", "required", "联系人信息行"), FieldCheck("入出院", "入院途径代码", "recommended", "入院途径勾选项"), FieldCheck("入出院", "入院时间", "required", "入院记录行"), FieldCheck("入出院", "入院科别", "required", "入院记录行"), FieldCheck("入出院", "入院病房", "recommended", "入院记录行"), FieldCheck("入出院", "转科科别", "optional", "转科科别行"), FieldCheck("入出院", "出院时间", "required", "出院记录行"), FieldCheck("入出院", "出院科别", "required", "出院记录行"), FieldCheck("入出院", "出院病房", "recommended", "出院记录行"), FieldCheck("入出院", "实际住院天数", "required", "出院记录行"), FieldCheck("入出院", "大科室", "required", "由入院/出院科别映射"), FieldCheck("诊断手术", "门急诊诊断", "recommended", "诊断区顶部"), FieldCheck("诊断手术", "门急诊诊断编码", "recommended", "诊断区顶部"), FieldCheck("诊断手术", "主要诊断名称", "required", "出院诊断表第一行"), FieldCheck("诊断手术", "主要诊断编码", "required", "出院诊断表第一行"), FieldCheck("诊断手术", "主要诊断入院病情", "required", "出院诊断表第一行"), FieldCheck("诊断手术", "出院诊断", "recommended", "出院诊断表全部行"), FieldCheck("诊断手术", "手术操作", "recommended", "手术及操作表"), FieldCheck("诊断手术", "损伤中毒外部原因", "optional", "损伤中毒外部原因行"), FieldCheck("诊断手术", "损伤中毒疾病编码", "optional", "损伤中毒外部原因行"), FieldCheck("诊断手术", "病理诊断", "optional", "病理诊断行"), FieldCheck("诊断手术", "病理诊断编码", "optional", "病理诊断行"), FieldCheck("诊断手术", "病理号", "optional", "病理号"), FieldCheck("质控信息", "药物过敏代码", "recommended", "药物过敏勾选项"), FieldCheck("质控信息", "过敏药物", "optional", "过敏药物填写处"), FieldCheck("质控信息", "死亡患者尸检代码", "recommended", "死亡患者尸检勾选项"), FieldCheck("质控信息", "血型代码", "recommended", "血型勾选项"), FieldCheck("质控信息", "Rh代码", "recommended", "Rh勾选项"), FieldCheck("质控信息", "科主任", "recommended", "医师签名区"), FieldCheck("质控信息", "主任副主任医师", "recommended", "医师签名区"), FieldCheck("质控信息", "主治医师", "recommended", "医师签名区"), FieldCheck("质控信息", "住院医师", "recommended", "医师签名区"), FieldCheck("质控信息", "责任护士", "recommended", "护理签名区"), FieldCheck("质控信息", "编码员", "recommended", "编码员签名区"), FieldCheck("质控信息", "病案质量代码", "recommended", "病案质量勾选项"), FieldCheck("质控信息", "质控医师", "recommended", "质控签名区"), FieldCheck("质控信息", "质控护士", "recommended", "质控签名区"), FieldCheck("质控信息", "质控日期", "recommended", "质控日期"), FieldCheck("离院费用", "离院方式代码", "recommended", "离院方式勾选项"), FieldCheck("离院费用", "出院31天内再住院计划代码", "recommended", "再住院计划勾选项"), FieldCheck("离院费用", "再住院计划目的", "optional", "再住院计划目的"), FieldCheck("离院费用", "入院前昏迷天数", "optional", "昏迷时间区"), FieldCheck("离院费用", "入院后昏迷天数", "optional", "昏迷时间区"), FieldCheck("离院费用", "总费用", "optional", "费用区"), FieldCheck("离院费用", "自付金额", "optional", "费用区"), FieldCheck("离院费用", "费用明细", "optional", "费用明细区"), ] def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="患者首页 PDF 转图片并生成字段对照核验报告") parser.add_argument("-i", "--input-dir", type=Path, default=DEFAULT_INPUT_DIR, help="PDF 输入目录") parser.add_argument("-r", "--result-dir", type=Path, default=DEFAULT_RESULT_DIR, help="解析结果根目录") parser.add_argument("-o", "--output-dir", type=Path, default=DEFAULT_IMAGE_REVIEW_DIR, help="图片对照输出目录") parser.add_argument("--dpi", type=int, default=180, help="图片分辨率，默认 180") parser.add_argument("--format", choices=["png", "jpeg"], default="png", help="图片格式，默认 png") parser.add_argument("--force", action="store_true", help="重新生成已存在图片") parser.add_argument("--strict-recommended", action="store_true", help="把建议字段缺项也写入对照建议") return parser.parse_args() def run(cmd: list[str]) -> subprocess.CompletedProcess[str]: return subprocess.run(cmd, check=True, text=True, capture_output=True) def numeric_page_key(path: Path) -> tuple[int, str]: match = re.search(r"-(\d+)\.(png|jpe?g)$", path.name, flags=re.IGNORECASE) if not match: return (10**9, path.name) return (int(match.group(1)), path.name) def convert_pdf_to_images(pdf_path: Path, output_dir: Path, dpi: int, image_format: str, force: bool) -> tuple[list[Path], str]: pdftoppm = shutil.which("pdftoppm") if not pdftoppm: return [], "未找到 pdftoppm，请安装 poppler-utils。" output_dir.mkdir(parents=True, exist_ok=True) extension = "jpg" if image_format == "jpeg" else "png" existing = sorted(output_dir.glob(f"page-*.{extension}"), key=numeric_page_key) if existing and not force: return existing, "" if force: for old_image in output_dir.glob("page-*.*"): old_image.unlink() prefix = output_dir / "page" cmd = [pdftoppm, "-r", str(dpi), f"-{image_format}", str(pdf_path), str(prefix)] try: run(cmd) except subprocess.CalledProcessError as exc: detail = exc.stderr.strip() or exc.stdout.strip() or str(exc) return [], f"图片转换失败：{detail}" images = sorted(output_dir.glob(f"page-*.{extension}"), key=numeric_page_key) normalized_images: list[Path] = [] for index, image_path in enumerate(images, start=1): target = output_dir / f"page-{index:03d}.{extension}" if image_path != target: if target.exists(): target.unlink() image_path.rename(target) normalized_images.append(target) return normalized_images, "" if normalized_images else "图片转换后未找到输出文件。" def load_record(result_dir: Path, pdf_path: Path) -> tuple[dict[str, Any], Path | None, str]: json_path = result_dir / "02_单份JSON" / f"{pdf_path.stem}.json" if not json_path.exists(): return {}, None, "未找到结构化 JSON，请先运行 02_解析入库 步骤。" try: return json.loads(json_path.read_text(encoding="utf-8")), json_path, "" except json.JSONDecodeError as exc: return {}, json_path, f"结构化 JSON 读取失败：{exc}" def is_blank(value: Any) -> bool: if value is None: return True if isinstance(value, str): text = value.strip() return text == "" or text in {"-", "--", "null", "None", "[]", "{}"} if isinstance(value, (list, tuple, set, dict)): return len(value) == 0 return False def value_preview(value: Any, max_length: int = 120) -> str: if value is None: return "" if isinstance(value, (list, dict)): text = json.dumps(value, ensure_ascii=False) else: text = str(value) text = re.sub(r"\s+", " ", text).strip() if len(text) > max_length: return text[: max_length - 1] + "…" return text def build_field_rows(record: dict[str, Any]) -> tuple[list[dict[str, str]], list[str], list[str]]: rows: list[dict[str, str]] = [] missing_required: list[str] = [] missing_recommended: list[str] = [] for check in FIELD_CHECKS: value = record.get(check.key) missing = is_blank(value) if missing and check.level == "required": missing_required.append(check.key) elif missing and check.level == "recommended": missing_recommended.append(check.key) rows.append( { "group": check.group, "key": check.key, "level": check.level, "location_hint": check.location_hint, "value": value_preview(value), "missing": "是" if missing else "否", } ) return rows, missing_required, missing_recommended def as_csv_text(value: Any) -> str: if value is None: return "" if isinstance(value, (list, dict)): return json.dumps(value, ensure_ascii=False) return str(value) def relative_to(path: Path, start: Path) -> str: try: return path.relative_to(start).as_posix() except ValueError: return path.as_posix() def html_escape(value: Any) -> str: return html.escape(str(value), quote=True) def make_case_summary( pdf_path: Path, images: list[Path], image_error: str, record: dict[str, Any], json_path: Path | None, json_error: str, output_dir: Path, strict_recommended: bool, ) -> dict[str, Any]: field_rows, missing_required, missing_recommended = build_field_rows(record) if record else ([], [], []) review_notes = record.get("复核备注", []) if record else [] if not isinstance(review_notes, list): review_notes = [review_notes] suggestions: list[str] = [] if image_error: suggestions.append(image_error) if json_error: suggestions.append(json_error) if missing_required: suggestions.append("核心字段缺项，需对照图片补齐：" + "、".join(missing_required)) if strict_recommended and missing_recommended: suggestions.append("建议字段缺项，抽样确认是否首页未填写：" + "、".join(missing_recommended)) if record.get("复核状态") and record.get("复核状态") != "auto_pass": suggestions.append("该记录已有复核状态：" + str(record.get("复核状态"))) if review_notes: suggestions.append("已有复核备注：" + "；".join(value_preview(note, 80) for note in review_notes)) if not suggestions: suggestions.append("图片与结构化字段抽查一致即可。") return { "pdf_path": pdf_path, "source_file": pdf_path.name, "images": images, "image_dir": images[0].parent if images else output_dir / "图片" / pdf_path.stem, "first_image": images[0] if images else None, "page_count": len(images), "image_error": image_error, "record": record, "json_path": json_path, "json_error": json_error, "field_rows": field_rows, "missing_required": missing_required, "missing_recommended": missing_recommended, "suggestions": suggestions, } def write_index_csv(cases: list[dict[str, Any]], output_dir: Path) -> Path: csv_path = output_dir / "患者首页_PDF图片对照索引.csv" fieldnames = [ "源文件", "页数", "图片目录", "首页图片", "结构化JSON", "病案号", "姓名", "复核状态", "复核备注", "核心缺项", "建议核对缺项", "对照建议", ] with csv_path.open("w", encoding="utf-8-sig", newline="") as fp: writer = csv.DictWriter(fp, fieldnames=fieldnames) writer.writeheader() for case in cases: record = case["record"] writer.writerow( { "源文件": case["source_file"], "页数": case["page_count"], "图片目录": relative_to(case["image_dir"], output_dir), "首页图片": relative_to(case["first_image"], output_dir) if case["first_image"] else "", "结构化JSON": relative_to(case["json_path"], output_dir) if case["json_path"] else "", "病案号": record.get("病案号", ""), "姓名": record.get("姓名", ""), "复核状态": record.get("复核状态", ""), "复核备注": as_csv_text(record.get("复核备注", [])), "核心缺项": "、".join(case["missing_required"]), "建议核对缺项": "、".join(case["missing_recommended"]), "对照建议": "；".join(case["suggestions"]), } ) return csv_path def render_field_table(field_rows: list[dict[str, str]]) -> str: if not field_rows: return '

未生成字段核验表。

' parts = [ '', "", "", ] for row in field_rows: missing_class = " is-missing" if row["missing"] == "是" and row["level"] != "optional" else "" parts.append( "" "" "".format( missing_class=missing_class.strip(), group=html_escape(row["group"]), key=html_escape(row["key"]), level=html_escape(row["level"]), location=html_escape(row["location_hint"]), value=html_escape(row["value"]), missing=html_escape(row["missing"]), ) ) parts.append("
组字段级别首页位置值缺项
{group}{key}{level}{location}{value}{missing}
") return "\n".join(parts) def render_html(cases: list[dict[str, Any]], output_dir: Path) -> Path: html_path = output_dir / "患者首页_PDF图片对照.html" total = len(cases) with_required_missing = sum(1 for case in cases if case["missing_required"]) with_review_status = sum(1 for case in cases if case["record"].get("复核状态") not in {"", "auto_pass", None}) failed_images = sum(1 for case in cases if case["image_error"]) nav_rows = [] case_sections = [] for index, case in enumerate(cases, start=1): anchor = f"case-{index}" record = case["record"] status = record.get("复核状态", "未解析") nav_rows.append( "{source}{name}{mrn}{status}{required}".format( anchor=anchor, source=html_escape(case["source_file"]), name=html_escape(record.get("姓名", "")), mrn=html_escape(record.get("病案号", "")), status=html_escape(status), required=html_escape("、".join(case["missing_required"]) or "无"), ) ) images_html = [] for image_path in case["images"]: rel_image = relative_to(image_path, output_dir) images_html.append( '
{alt}
{caption}
'.format( src=html_escape(rel_image), alt=html_escape(f"{case['source_file']} {image_path.name}"), caption=html_escape(image_path.name), ) ) if not images_html: images_html.append('

未生成图片。

') suggestion_items = "\n".join(f"
  • {html_escape(note)}
    • " for note in case["suggestions"]) summary_items = [ ("病案号", record.get("病案号", "")), ("姓名", record.get("姓名", "")), ("大科室", record.get("大科室", "")), ("出院科别", record.get("出院科别", "")), ("主要诊断", record.get("主要诊断名称", "")), ("主要诊断编码", record.get("主要诊断编码", "")), ("复核状态", status), ] summary_html = "\n".join( f"
    {html_escape(label)}
    {html_escape(value_preview(value, 160))}
    " for label, value in summary_items ) case_sections.append( """

    {source}

    {name} · {mrn} · {status}

    返回顶部
    {images}
    {summary}

    对照建议

      {suggestions}
    {field_table}
    """.format( anchor=anchor, source=html_escape(case["source_file"]), name=html_escape(record.get("姓名", "")), mrn=html_escape(record.get("病案号", "")), status=html_escape(status), images="\n".join(images_html), summary=summary_html, suggestions=suggestion_items, field_table=render_field_table(case["field_rows"]), ) ) page_template = Template( """ 患者首页 PDF 图片对照核验

    患者首页 PDF 图片对照核验

    PDF：$total 核心缺项：$with_required_missing 需复核状态：$with_review_status 图片失败：$failed_images
    $nav_rows
    源文件姓名病案号复核状态核心缺项
    $case_sections
    """ ) html_path.write_text( page_template.substitute( total=total, with_required_missing=with_required_missing, with_review_status=with_review_status, failed_images=failed_images, nav_rows="\n".join(nav_rows), case_sections="\n".join(case_sections), ), encoding="utf-8", ) return html_path def main() -> int: args = parse_args() input_dir = args.input_dir.resolve() result_dir = args.result_dir.resolve() output_dir = args.output_dir.resolve() image_root = output_dir / "图片" output_dir.mkdir(parents=True, exist_ok=True) pdf_files = sorted(input_dir.glob("*.pdf")) if not pdf_files: raise SystemExit(f"未找到 PDF：{input_dir}") cases: list[dict[str, Any]] = [] for pdf_path in pdf_files: case_image_dir = image_root / pdf_path.stem images, image_error = convert_pdf_to_images(pdf_path, case_image_dir, args.dpi, args.format, args.force) record, json_path, json_error = load_record(result_dir, pdf_path) cases.append( make_case_summary( pdf_path=pdf_path, images=images, image_error=image_error, record=record, json_path=json_path, json_error=json_error, output_dir=output_dir, strict_recommended=args.strict_recommended, ) ) index_csv = write_index_csv(cases, output_dir) html_path = render_html(cases, output_dir) required_missing_count = sum(1 for case in cases if case["missing_required"]) print(f"PDF 数量：{len(cases)}") print(f"图片输出：{image_root}") print(f"对照索引：{index_csv}") print(f"HTML 对照页：{html_path}") print(f"核心缺项病例数：{required_missing_count}") return 0 if __name__ == "__main__": raise SystemExit(main())