"""医生核对包：目录结构、CSV 规格、问题清单的每种来源、逐字节拷贝、TIC结果、没装 CEUS 的构建、PDF。 预期值照任务书 03 写：目录和文件名、列名、问题清单的「环节」与「问题」原文、关键帧文件名的秒数 （`video_time_s` 保留一位小数）都按任务书的文字手写，不从实现里推。 假数据（6 例）： - P001：进库；定位到 D1、D2、F1，其余 4 个槽位未取到；病灶去暗区曲线拟合失败 （parameter_on_boundary）；紫圈不可用。 - P002：CEUS 第 2 步失败（C15 跳过）。P003：CEUS 运行里没有结果目录（C15 跳过）。 - P004：进库；7 个槽位都取到，D2、F1 的时间改成 5.04、12.46 秒（测一位小数）；拟合都成功；紫圈可用。 - P005：MR 表诊断与另两张表不同（C10 跳过）；CEUS 运行里却有它完整的结果目录（被跳过但定量成功）。 - P006：进库；7 个槽位都取到；病灶整体曲线拟合失败（credible_peak_not_found）；紫圈可用。 """ import csv import hashlib import json import re import shutil import subprocess from datetime import datetime from pathlib import Path from typing import Any import pytest import yaml from cadx.answers import read_aliases from cadx.build import BuildSummary, run_build from cadx.library import read_ultrasound from cadx.review import run_review_pack from cadx.review._pdf import markdown_to_pdf, report_markdown from fake_ceus_run import UNLOCATED_REASON, summary, write_run from fake_delivery import DIAGNOSIS_COLUMN, REPO_ROOT, make_delivery, write_config BUILD_NOW = datetime(2026, 10, 17, 9, 0, 0).astimezone() PACK_NOW = datetime(2026, 10, 17, 10, 0, 0).astimezone() RUN_ID = "20261017-080000-abcdef0" ALL_SLOTS = ("B0", "D1", "D2", "D3", "D4", "F1", "F2") SUMMARY_HEADER = [ "病例号", "视频号", "化名", "是否进入资料库", "没进的原因", "CEUS 定量是否成功", "CEUS 失败在第几步", *ALL_SLOTS, ] PROBLEM_HEADER = ["病例号", "视频号", "环节", "问题"] ANSWER_HEADERS = { "化名对照.csv": ["化名", "病例号", "视频号"], "诊断.csv": ["化名", "病例号", "临床表诊断编码", "超声表诊断编码", "MR表诊断编码", "诊断名"], "跳过病例.csv": ["病例", "检查编号", "检查说明", "文件", "列", "涉及值"], "报告复核句.csv": ["化名", "路", "命中的词", "所在句子"], } CASE_FILES = {"临床信息.csv", "超声报告.pdf", "MR报告.pdf", "关键图.png"} CEUS_FILES = {"关键帧", "TIC曲线.png", "TIC结果.csv", "回放视频.mp4"} def _rewrite_summary(run: Path, video_id: str, located: tuple[str, ...], edit: Any) -> None: content = summary(video_id, located) edit(content) (run / video_id / "summary.json").write_text(json.dumps(content), encoding="utf-8") def _fail_fit(curve: str, code: str) -> Any: def edit(content: dict[str, Any]) -> None: content["fits"][curve].update(success=False, failure_reason=code) return edit def _p004(content: dict[str, Any]) -> None: content["reference_valid"] = True content["anchor_frames"]["D2"]["frame_time_video_s"] = 5.04 content["anchor_frames"]["F1"]["frame_time_video_s"] = 12.46 def _p006(content: dict[str, Any]) -> None: _fail_fit("roi_all", "credible_peak_not_found")(content) content["reference_valid"] = True def _build_with_ceus(root: Path) -> tuple[BuildSummary, Path]: delivery = make_delivery(6) delivery.case(5).mr[DIAGNOSIS_COLUMN] = 2 delivery_dir = delivery.write(root / "delivery") run = write_run( root / "runs" / RUN_ID, delivery_dir, {"U001": ("D1", "D2", "F1"), "U004": ALL_SLOTS, "U005": ALL_SLOTS, "U006": ALL_SLOTS}, [("U002", 2, "两半都没有绿框")], ) _rewrite_summary( run, "U001", ("D1", "D2", "F1"), _fail_fit("roi_perf", "parameter_on_boundary") ) _rewrite_summary(run, "U004", ALL_SLOTS, _p004) _rewrite_summary(run, "U006", ALL_SLOTS, _p006) return run_build(write_config(root, delivery_dir, 1, None, run), REPO_ROOT, BUILD_NOW), run def _pack_config(root: Path, library_dir: Path, answers_dir: Path) -> Path: config = { "library_dir": str(library_dir), "answers_dir": str(answers_dir), "output_root": str(root / "review_pack"), } path = root / "review_pack.yaml" path.write_text(yaml.safe_dump(config, allow_unicode=True), encoding="utf-8") return path def _tree_hashes(folder: Path) -> dict[str, str]: return { p.relative_to(folder).as_posix(): hashlib.sha256(p.read_bytes()).hexdigest() for p in sorted(folder.rglob("*")) if p.is_file() } class Packed: """构建一次、出一次包，各条测试共用。""" def __init__(self, root: Path) -> None: self.build, self.run = _build_with_ceus(root) self.library_before = _tree_hashes(self.build.library_dir) self.answers_before = _tree_hashes(self.build.answers_dir) config = _pack_config(root, self.build.library_dir, self.build.answers_dir) self.config = config self.result = run_review_pack(config, REPO_ROOT, PACK_NOW) self.pack = self.result.pack_dir self.alias = {r.case_no: r.alias for r in read_aliases(self.build.answers_dir)} @pytest.fixture(scope="module") def packed(tmp_path_factory: pytest.TempPathFactory) -> Packed: return Packed(tmp_path_factory.mktemp("review")) def _csv(path: Path) -> list[list[str]]: raw = path.read_bytes() assert raw.startswith(b"\xef\xbb\xbf"), path with path.open(encoding="utf-8-sig", newline="") as handle: return list(csv.reader(handle)) def test_directory_layout(packed: Packed) -> None: pack = packed.pack assert pack.name == f"医生核对包_{packed.build.build_id}" assert sorted(p.name for p in packed.result.run_dir.iterdir()) == sorted( ["snapshot", pack.name] ) assert {p.name for p in pack.iterdir()} == { "说明.pdf", "问题清单.csv", "病例总表.csv", "标准答案", "病例", } assert {p.name for p in (pack / "标准答案").iterdir()} == set(ANSWER_HEADERS) assert {p.name for p in (pack / "病例").iterdir()} == {"P001", "P004", "P006"} for case in ("P001", "P004", "P006"): assert {p.name for p in (pack / "病例" / case).iterdir()} == CASE_FILES | CEUS_FILES assert {p.name for p in (pack / "病例" / "P001" / "关键帧").iterdir()} == { "D1_第2.5秒.png", "D2_第5.0秒.png", "F1_第12.5秒.png", } assert {p.name for p in (pack / "病例" / "P004" / "关键帧").iterdir()} == { "B0_第0.0秒.png", "D1_第2.5秒.png", "D2_第5.0秒.png", "D3_第7.5秒.png", "D4_第10.0秒.png", "F1_第12.5秒.png", "F2_第15.0秒.png", } def test_every_csv_has_bom_and_no_placeholder_for_missing(packed: Packed) -> None: csvs = sorted(packed.pack.rglob("*.csv")) assert len(csvs) == 2 + 4 + 3 * 2 for path in csvs: for row in _csv(path): assert not {"null", "None", "NaN", "nan"} & set(row), path assert _csv(packed.pack / "病例总表.csv")[0] == SUMMARY_HEADER assert _csv(packed.pack / "问题清单.csv")[0] == PROBLEM_HEADER for name, header in ANSWER_HEADERS.items(): assert _csv(packed.pack / "标准答案" / name)[0] == header clinical = _csv(packed.pack / "病例" / "P001" / "临床信息.csv") assert len(clinical) == 2 and "Case_ID" not in clinical[0] and "Diagnosis" not in clinical[0] assert clinical[0][:3] == ["Sex", "Age", "Symptom_Pain"] # 假交付包里没测的化验是 None：留空。 assert clinical[1][clinical[0].index("Amylase")] == "" def test_case_summary(packed: Packed) -> None: rows = _csv(packed.pack / "病例总表.csv")[1:] blank = [""] * 7 a = packed.alias assert rows == [ [ "P001", "U001", a["001"], "是", "", "是", "", "未取到", "2.5", "5.0", "未取到", "未取到", "12.5", "未取到", ], ["P002", "U002", "", "否", "C15 CEUS 定量结果齐全", "否", "2", *blank], ["P003", "U003", "", "否", "C15 CEUS 定量结果齐全", "", "", *blank], [ "P004", "U004", a["004"], "是", "", "是", "", "0.0", "2.5", "5.04", "7.5", "10.0", "12.46", "15.0", ], ["P005", "U005", "", "否", "C10 跨表一致（诊断、年龄、性别）", "是", "", *blank], [ "P006", "U006", a["006"], "是", "", "是", "", "0.0", "2.5", "5.0", "7.5", "10.0", "12.5", "15.0", ], ] def test_problem_list_has_every_source(packed: Packed) -> None: rows = _csv(packed.pack / "问题清单.csv")[1:] skipped = {row[0]: row for row in _csv(packed.pack / "标准答案" / "跳过病例.csv")[1:]} unlocated = [ ["P001", "U001", "关键帧", f"{slot} 未取到：{UNLOCATED_REASON}"] for slot in ("B0", "D3", "D4", "F2") ] assert rows == [ ["P001", "U001", "TIC 拟合", "病灶去暗区曲线未完成拟合（parameter_on_boundary）"], *unlocated, ["P001", "U001", "紫圈", "紫圈曲线不可用"], [ "P002", "U002", "装配资料库", f"C15 CEUS 定量结果齐全；文件：{RUN_ID}/failed.csv；" '涉及值："第 2 步失败：两半都没有绿框"', ], ["P002", "U002", "CEUS 第 2 步", "第 2 步失败：两半都没有绿框"], [ "P003", "U003", "装配资料库", f"C15 CEUS 定量结果齐全；文件：{RUN_ID}/U003；" '涉及值："CEUS 运行目录里没有这一例的结果目录"', ], [ "P005", "U005", "装配资料库", f"C10 跨表一致（诊断、年龄、性别）；文件：{skipped['005'][3]}；" f"列：{skipped['005'][4]}；涉及值：{skipped['005'][5]}", ], ["P006", "U006", "TIC 拟合", "病灶整体曲线未完成拟合（credible_peak_not_found）"], ] counts = {"装配资料库": 3, "CEUS 第 2 步": 1, "TIC 拟合": 2, "关键帧": 4, "紫圈": 1} assert packed.result.problems_by_stage == dict(sorted(counts.items())) def test_copies_are_byte_identical(packed: Packed) -> None: for case_no, video_id in (("001", "U001"), ("004", "U004")): case_dir = packed.pack / "病例" / f"P{case_no}" route = read_ultrasound(packed.build.library_dir, packed.alias[case_no]) assert (case_dir / "关键图.png").read_bytes() == route.key_image.read_bytes() assert route.tic is not None copied = {p.name.split("_")[0]: p for p in (case_dir / "关键帧").iterdir()} assert sorted(copied) == sorted(route.keyframes) for slot, source in route.keyframes.items(): assert copied[slot].read_bytes() == source.read_bytes() seconds = route.tic["keyframes"][slot]["video_time_s"] assert copied[slot].name == f"{slot}_第{round(seconds, 1)}秒.png" run_case = packed.run / video_id assert (case_dir / "TIC曲线.png").read_bytes() == (run_case / "tic.png").read_bytes() assert (case_dir / "回放视频.mp4").read_bytes() == (run_case / "overlay.mp4").read_bytes() answers = packed.build.answers_dir for source, name in ( ("aliases.csv", "化名对照.csv"), ("diagnoses.csv", "诊断.csv"), ("skipped.csv", "跳过病例.csv"), ("review.csv", "报告复核句.csv"), ): assert (packed.pack / "标准答案" / name).read_bytes() == (answers / source).read_bytes() def _tool_output(tool: str, *args: str) -> str: path = shutil.which(tool) assert path is not None, f"找不到 {tool}" return subprocess.run([path, *args], check=True, capture_output=True, text=True).stdout # noqa: S603 def _expected_cell(value: object) -> str: if value is None: return "" if isinstance(value, bool): return "true" if value else "false" return value if isinstance(value, str) else repr(value) def test_tic_results_match_tic_json(packed: Packed) -> None: for case_no in ("001", "004", "006"): tic = read_ultrasound(packed.build.library_dir, packed.alias[case_no]).tic assert tic is not None rows = _csv(packed.pack / "病例" / f"P{case_no}" / "TIC结果.csv") assert rows[0] == ["项目", "值"] # 4 条曲线 × 2 + 病灶 2 条 ×（3 个口径 + 11 个指标 + 3 个辅助时刻）+ 紫圈 1 + 7 槽位 × 3。 assert len(rows) - 1 == 8 + 34 + 1 + 21 for item, value in rows[1:]: path = item[item.rindex("（") + 1 : -1] node: Any = tic for key in path.split("."): node = None if node is None else node[key] assert value == _expected_cell(node), (case_no, item) rows = dict(_csv(packed.pack / "病例" / "P001" / "TIC结果.csv")[1:]) assert rows["病灶整体曲线 拟合成功（fits.roi_all.success）"] == "true" assert rows["病灶去暗区曲线 拟合成功（fits.roi_perf.success）"] == "false" assert rows["紫圈曲线可用（pan_curve_usable）"] == "false" assert rows["关键帧 D4 未取到的原因（keyframes.D4.reason）"] == UNLOCATED_REASON def test_pdfs_use_noto_cjk_and_carry_ids(packed: Packed) -> None: pdfs = sorted(packed.pack.rglob("*.pdf")) assert len(pdfs) == packed.result.pdfs == 1 + 3 * 2 for pdf in pdfs: assert pdf.read_bytes().startswith(b"%PDF") and pdf.stat().st_size > 1000 readme = packed.pack / "说明.pdf" fonts = _tool_output("pdffonts", str(readme)) assert "NotoSerifCJK" in fonts and "NotoSansCJK" in fonts text = _tool_output("pdftotext", str(readme), "-") assert packed.build.build_id in text and RUN_ID in text and "parameter_on_boundary" in text report = _tool_output("pdftotext", str(packed.pack / "病例" / "P001" / "超声报告.pdf"), "-") assert report.startswith("病例号 P001 超声描述") def test_snapshot_and_no_writes_to_libraries(packed: Packed) -> None: snapshot = packed.result.run_dir / "snapshot" assert (snapshot / "review_pack.yaml").read_bytes() == packed.config.read_bytes() record = json.loads((snapshot / "review.json").read_text(encoding="utf-8")) assert record["build_id"] == packed.build.build_id and record["ceus_run_id"] == RUN_ID assert record["pandoc"].startswith("pandoc ") and "LibreOffice" in record["soffice"] assert (snapshot / "finished.json").is_file() assert _tree_hashes(packed.build.library_dir) == packed.library_before assert _tree_hashes(packed.build.answers_dir) == packed.answers_before def test_existing_run_dir_stops(packed: Packed) -> None: with pytest.raises(FileExistsError): run_review_pack(packed.config, REPO_ROOT, PACK_NOW) def test_build_without_ceus(tmp_path: Path) -> None: delivery_dir = make_delivery(2).write(tmp_path / "delivery") build = run_build(write_config(tmp_path, delivery_dir, 1, None, None), REPO_ROOT, BUILD_NOW) result = run_review_pack( _pack_config(tmp_path, build.library_dir, build.answers_dir), REPO_ROOT, PACK_NOW ) for case in ("P001", "P002"): assert {p.name for p in (result.pack_dir / "病例" / case).iterdir()} == CASE_FILES alias = {r.case_no: r.alias for r in read_aliases(build.answers_dir)} rows = _csv(result.pack_dir / "病例总表.csv")[1:] assert [row[:5] for row in rows] == [ ["P001", "U001", alias["001"], "是", ""], ["P002", "U002", alias["002"], "是", ""], ] assert all(row[5:] == [""] * 9 for row in rows) assert _csv(result.pack_dir / "问题清单.csv") == [PROBLEM_HEADER] assert result.problems_by_stage == {} and result.video_bytes == 0 def test_build_ids_differ_stops(tmp_path: Path) -> None: config = _pack_config(tmp_path, tmp_path / "library" / "a", tmp_path / "answers" / "b") with pytest.raises(ValueError, match="不是同一次构建"): run_review_pack(config, REPO_ROOT, PACK_NOW) assert not (tmp_path / "review_pack").exists() def test_report_pdf_keeps_markdown_characters(tmp_path: Path) -> None: """报告正文里的 Markdown 记号、引号、连字符、换行照抄，不被 pandoc 当成格式。 行首空格抽回文字时只剩一个（pdftotext 按版面估空格），所以每行比较去掉首尾空白后的内容。 """ text = ( '1. 囊壁*增厚* #2 a_b_c x [链接](y) -- "引号" ... `码` | 竖线\n 第二行行首两个空格' ) target = tmp_path / "报告.pdf" markdown_to_pdf({target: report_markdown("病例号 P001 超声描述", text)}) lines = _tool_output("pdftotext", "-layout", str(target), "-").splitlines() assert [line.strip() for line in lines if line.strip()][1:] == [ line.strip() for line in text.splitlines() ] def _gap_after_ab(pdf: Path) -> float: """pdftotext -bbox 里词「ab」右边界到下一个词左边界的距离（点）。""" boxes = re.findall( r'xMin="([0-9.]+)"[^>]*xMax="([0-9.]+)"[^>]*>([^<]*)', _tool_output("pdftotext", "-bbox", str(pdf), "-"), ) words = [(float(x0), float(x1), w) for x0, x1, w in boxes] index = next(i for i, (_, _, w) in enumerate(words) if w == "ab") return words[index + 1][0] - words[index][1] def test_report_pdf_keeps_repeated_spaces(tmp_path: Path) -> None: """一行里连续三个空格不被 pandoc 并成一个。 pdftotext 抽文字时按间距估空格，三个空格也只吐出一个，所以量版面：「ab」与「cd」之间的间距是 单个空格那份 PDF 的约三倍。 """ one, three = tmp_path / "一个空格.pdf", tmp_path / "三个空格.pdf" markdown_to_pdf( { one: report_markdown("病例号 P001 超声描述", "ab cd"), three: report_markdown("病例号 P001 超声描述", "ab cd"), } ) assert _gap_after_ab(three) / _gap_after_ab(one) == pytest.approx(3.0, abs=0.3) def test_report_with_tab_stops() -> None: with pytest.raises(ValueError, match="病例号 P001 超声描述：报告原文里有 Tab"): report_markdown("病例号 P001 超声描述", "囊壁厚\t约2mm")