210 lines
6.9 KiB
Python
210 lines
6.9 KiB
Python
#!/usr/bin/env python3
|
|
import csv
|
|
import sys
|
|
import tempfile
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
from openpyxl import Workbook, load_workbook
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from app.main import _clean_preview_rows
|
|
from app.processor import _summarize_workbook, run_processing
|
|
|
|
|
|
REPORT_HEADERS = [
|
|
"rptunitid",
|
|
"rechkdt",
|
|
"reportid",
|
|
"reporttype",
|
|
"req_reason",
|
|
"specimen_code",
|
|
"specimen_name",
|
|
"rptunitname",
|
|
"resultclass",
|
|
"pat_diag",
|
|
"alter_flag",
|
|
"emer_flag",
|
|
"sampled_dt",
|
|
]
|
|
|
|
DETAIL_HEADERS = ["reportid", "rpt_itemname", "result_str", "result_ref"]
|
|
|
|
|
|
def write_csv(path, headers, rows):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open("w", encoding="utf-8-sig", newline="") as file:
|
|
writer = csv.DictWriter(file, fieldnames=headers)
|
|
writer.writeheader()
|
|
writer.writerows(rows)
|
|
|
|
|
|
def detail_filename(patient_id, row):
|
|
return (
|
|
f"{patient_id}_{row['reporttype']}_{row['rptunitid']}_"
|
|
f"{row['reportid'].replace('_', '-')}_"
|
|
f"{row['rechkdt'].replace(' ', '-').replace(':', '-').replace('_', '-')}.csv"
|
|
)
|
|
|
|
|
|
def build_fixture(root):
|
|
patient_id = "0000000001"
|
|
write_csv(root / "Patients_info.csv", ["pat_name", "pat_no"], [{"pat_name": "验证患者", "pat_no": "1"}])
|
|
|
|
reports = [
|
|
{
|
|
"rptunitid": "20",
|
|
"rechkdt": "2026-01-01 10:00:00",
|
|
"reportid": "20260101-20-1",
|
|
"reporttype": "10",
|
|
"req_reason": "肝功十项[复]_电解质五项[复]_肾功三项[复]",
|
|
"sampled_dt": "2026-01-01 08:00:00",
|
|
},
|
|
{
|
|
"rptunitid": "20",
|
|
"rechkdt": "2026-01-01 10:05:00",
|
|
"reportid": "20260101-20-2",
|
|
"reporttype": "10",
|
|
"req_reason": "B型钠尿肽前体(Pro-BNP)测定",
|
|
"sampled_dt": "2026-01-01 08:05:00",
|
|
},
|
|
{
|
|
"rptunitid": "20",
|
|
"rechkdt": "2026-01-01 10:10:00",
|
|
"reportid": "20260101-20-3",
|
|
"reporttype": "10",
|
|
"req_reason": "未知组合检测",
|
|
"sampled_dt": "2026-01-01 08:10:00",
|
|
},
|
|
{
|
|
"rptunitid": "161",
|
|
"rechkdt": "2026-01-01 10:15:00",
|
|
"reportid": "20260101-161-4",
|
|
"reporttype": "10",
|
|
"req_reason": "甲功七项（化学发光法）[复]",
|
|
"sampled_dt": "2026-01-01 08:15:00",
|
|
},
|
|
{
|
|
"rptunitid": "161",
|
|
"rechkdt": "2026-01-01 10:15:00",
|
|
"reportid": "20260101-161-4",
|
|
"reporttype": "10",
|
|
"req_reason": "甲功七项（化学发光法）[复]",
|
|
"sampled_dt": "2026-01-01 08:15:00",
|
|
},
|
|
]
|
|
for row in reports:
|
|
for header in REPORT_HEADERS:
|
|
row.setdefault(header, "")
|
|
|
|
write_csv(root / "Tests_List" / f"{patient_id}.csv", REPORT_HEADERS, reports)
|
|
|
|
details = [
|
|
[{"reportid": reports[0]["reportid"], "rpt_itemname": "谷草转氨酶", "result_str": "21"}],
|
|
[{"reportid": reports[1]["reportid"], "rpt_itemname": "B型前脑尿钠肽", "result_str": "57.20"}],
|
|
[{"reportid": reports[2]["reportid"], "rpt_itemname": "神秘检测项目", "result_str": "42"}],
|
|
[
|
|
{"reportid": reports[3]["reportid"], "rpt_itemname": "甲状腺球蛋白", "result_str": "6.02"},
|
|
],
|
|
[
|
|
{"reportid": reports[4]["reportid"], "rpt_itemname": "甲状腺球蛋白", "result_str": "6.02"},
|
|
],
|
|
]
|
|
detail_dir = root / "Tests_Detail_List" / patient_id
|
|
for report, rows in zip(reports, details):
|
|
for row in rows:
|
|
row.setdefault("result_ref", "")
|
|
write_csv(detail_dir / detail_filename(patient_id, report), DETAIL_HEADERS, rows)
|
|
|
|
|
|
def zip_dir(source, target):
|
|
with zipfile.ZipFile(target, "w", zipfile.ZIP_DEFLATED) as zf:
|
|
for path in source.rglob("*"):
|
|
if path.is_file():
|
|
zf.write(path, path.relative_to(source).as_posix())
|
|
|
|
|
|
def rows_for(ws):
|
|
return [["" if value is None else str(value) for value in row] for row in ws.iter_rows(values_only=True)]
|
|
|
|
|
|
def assert_dynamic_routing(workbook_path):
|
|
workbook = load_workbook(workbook_path, read_only=True, data_only=True)
|
|
try:
|
|
liver_rows = rows_for(workbook["肝功"])
|
|
heart_rows = rows_for(workbook["心衰系列"])
|
|
immune_rows = rows_for(workbook["免疫系列"])
|
|
summary_text = "\n".join("\t".join(row) for row in rows_for(workbook["未检测到内容汇总"]))
|
|
|
|
liver_data = [row for row in liver_rows[1:] if row and row[0]]
|
|
heart_data = [row for row in heart_rows[1:] if row and row[0]]
|
|
tumor_data = [row for row in rows_for(workbook["各类肿瘤标志物"])[1:] if row and row[0]]
|
|
immune_data = [row for row in immune_rows[1:] if row and row[0]]
|
|
|
|
assert len(liver_data) == 1, liver_data
|
|
assert liver_data[0][3] == "肝功十项[复]_电解质五项[复]_肾功三项[复]"
|
|
assert "21" in liver_data[0]
|
|
assert "57.20" not in liver_data[0]
|
|
|
|
assert len(heart_data) == 1, heart_data
|
|
assert heart_data[0][3] == "B型钠尿肽前体(Pro-BNP)测定"
|
|
assert "57.20" in heart_data[0]
|
|
assert "肝功十项[复]_电解质五项[复]_肾功三项[复]" not in "\n".join("\t".join(row) for row in heart_data)
|
|
|
|
assert "神秘检测项目" in summary_text
|
|
assert "谷草转氨酶" not in summary_text
|
|
assert "B型前脑尿钠肽" not in summary_text
|
|
assert len(tumor_data) == 0, tumor_data
|
|
assert len(immune_data) == 1, immune_data
|
|
assert immune_data[0][3] == "甲功七项（化学发光法）[复]"
|
|
finally:
|
|
workbook.close()
|
|
|
|
|
|
def assert_preview_can_exceed_200(tmp_path):
|
|
assert _clean_preview_rows(500) == 500
|
|
|
|
workbook_path = tmp_path / "preview.xlsx"
|
|
workbook = Workbook()
|
|
sheet = workbook.active
|
|
sheet.title = "预览"
|
|
for index in range(250):
|
|
sheet.append([index])
|
|
workbook.save(workbook_path)
|
|
|
|
summary = _summarize_workbook(workbook_path, tmp_path, 250)
|
|
assert len(summary.sheets[0].preview) == 250
|
|
|
|
|
|
def main():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
tmp_path = Path(tmp)
|
|
data_dir = tmp_path / "data"
|
|
build_fixture(data_dir)
|
|
zip_path = tmp_path / "fixture.zip"
|
|
zip_dir(data_dir, zip_path)
|
|
|
|
job_dir = tmp_path / "job"
|
|
result = run_processing(
|
|
zip_path=zip_path,
|
|
job_dir=job_dir,
|
|
mode="auto",
|
|
data_type="auto",
|
|
result_name="Verify",
|
|
show_not_match=True,
|
|
show_all_infos=True,
|
|
preview_rows=500,
|
|
)
|
|
assert result.mode == "v1"
|
|
assert_dynamic_routing(job_dir / "output" / "Verify.xlsx")
|
|
assert_preview_can_exceed_200(tmp_path)
|
|
|
|
print("dynamic routing verification passed")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|