Tighten lab item routing and dedupe rows

.dockerignore

@@ -4,4 +4,4 @@ __pycache__/
 .venv/
 uploads/
 jobs/
-
+data/

README.md

@@ -1,6 +1,15 @@
 # HIS_Sur_Data_Deal
 
-网页端检测数据处理工具。上传 `待处理检测数据.zip` 后，服务会自动识别 V1/V2 数据结构，调用原处理脚本生成 Excel，并在网页中展示结果摘要、工作表统计和数据预览；用户可调整每个工作表的预览行数，并按需导出单个 Excel。
+网页端检测数据处理工具。上传 `待处理检测数据.zip` 后，系统会自动识别 V1/V2 数据结构，调用原处理脚本生成 Excel，并在网页中展示结果摘要、工作表统计和数据预览。
+
+处理完成后可在结果页：
+
+- 调整每个工作表的预览行数。
+- 选择是否显示基本工作表、未匹配检测内容项、未检测到内容汇总表。
+- 按姓名、采样时间、检测原因选择升序或降序排序。
+- 导出当前内容选择和排序后的单个 Excel。
+
+“未检测到内容汇总”会在每个“姓名 / 住院号 / 采样时间 / 检测原因”分组标题下单独排序，不会把不同检测原因的分组混在一起。
 
 ## 本地运行
 
@@ -9,7 +18,11 @@ pip install -r requirements.txt
 uvicorn app.main:app --host 0.0.0.0 --port 8000
 ```
 
-访问 `http://localhost:8000`。
+访问：
+
+```text
+http://localhost:8000
+```
 
 ## Docker 构建与运行
 
@@ -18,7 +31,145 @@ docker build -t his-sur-data-deal .
 docker run --rm -p 8000:8000 his-sur-data-deal
 ```
 
-## 推送镜像示例
+访问：
+
+```text
+http://localhost:8000
+```
+
+## Docker Compose 通用部署
+
+通用配置文件：
+
+```text
+docker_compose.yaml
+```
+
+这个配置是本地构建配置，不会从远程镜像仓库拉取 `his-sur-data-deal:latest`。运行时必须在项目源码目录执行，目录中需要包含：
+
+```text
+Dockerfile
+requirements.txt
+app/
+docker_compose.yaml
+```
+
+启动：
+
+```bash
+docker compose -f docker_compose.yaml up -d --build
+```
+
+默认映射：
+
+```text
+宿主机 8000 -> 容器 8000
+```
+
+访问：
+
+```text
+http://服务器IP:8000
+```
+
+## frpc + NPM 反向代理到公网域名
+
+如果本服务运行在内网机器，想通过公网服务器域名访问，可按下面链路处理：
+
+```text
+浏览器
+  -> 公网域名
+  -> 公网服务器 Nginx Proxy Manager
+  -> 公网服务器 frps 暴露端口
+  -> 内网机器 frpc
+  -> 内网 Docker 服务 127.0.0.1:8000
+```
+
+内网机器运行本项目：
+
+```bash
+docker compose -f docker_compose.yaml up -d --build
+```
+
+内网机器 frpc 示例：
+
+```toml
+serverAddr = "公网服务器IP"
+serverPort = 7000
+
+[[proxies]]
+name = "his-sur-data-deal"
+type = "tcp"
+localIP = "127.0.0.1"
+localPort = 8000
+remotePort = 18004
+```
+
+公网服务器 Nginx Proxy Manager 中新增 Proxy Host：
+
+```text
+Domain Names: 你的域名，例如 his.example.com
+Scheme: http
+Forward Hostname / IP: 127.0.0.1
+Forward Port: 18004
+Websockets Support: 可不开
+Block Common Exploits: 建议开启
+SSL: 按需申请 Let's Encrypt
+```
+
+如果公网服务器的 NPM 与 frps 不在同一台机器，则 `Forward Hostname / IP` 填写 frps 所在服务器的内网或公网 IP。
+
+## NAS Docker Compose 部署
+
+NAS 地址：
+
+```text
+http://192.168.31.5/
+```
+
+NAS 专用配置文件：
+
+```text
+docker_compose-Nas.yaml
+```
+
+这个配置同样是本地构建配置，不依赖远程镜像仓库。NAS 上请把整个项目源码放到下面目录：
+
+```text
+/share/Container/his_sur_data_deal
+```
+
+该目录中必须包含：
+
+```text
+/share/Container/his_sur_data_deal/Dockerfile
+/share/Container/his_sur_data_deal/requirements.txt
+/share/Container/his_sur_data_deal/app/
+/share/Container/his_sur_data_deal/docker_compose-Nas.yaml
+```
+
+如果只在 NAS 面板里粘贴 yaml，而没有把 `Dockerfile` 和 `app/` 一起放到上述路径，NAS 会报 `failed to read dockerfile: open Dockerfile: no such file or directory`。
+
+启动：
+
+```bash
+cd /share/Container/his_sur_data_deal
+docker compose -f docker_compose-Nas.yaml up -d --build
+```
+
+默认映射：
+
+```text
+NAS 4004 -> 容器 8000
+```
+
+访问：
+
+```text
+http://192.168.31.5:4004
+```
+
+## 镜像推送示例
 
 ```bash
 docker tag his-sur-data-deal 192.168.31.5:5002/admin/his-sur-data-deal:latest
@@ -27,19 +178,23 @@ docker push 192.168.31.5:5002/admin/his-sur-data-deal:latest
 
 ## 数据模式
 
-V1：zip 解压后包含 `Patients_info.csv`、`Tests_List`、`Tests_Detail_List`，输出一个汇总 Excel。
+V1：
 
-V2：zip 解压后包含 `Patients_info.csv`，并按患者目录分别保存检测汇总和具体检测，输出多个患者 Excel。
+- zip 解压后包含 `Patients_info.csv`、`Tests_List`、`Tests_Detail_List`。
+- 输出一个汇总 Excel。
 
-患者编号类型可选择自动识别。自动识别会读取 `Patients_info.csv` 中的 `pat_no`，并与 `Tests_List` 文件名或患者目录名比对：若更匹配 10 位补零编号，则使用 `pat_no`；若更匹配原始编号，则使用 `zhuyuanhao`。
+V2：
 
-系统默认输出完整结果；处理完成后可在结果页通过“内容选择”立即切换预览内容，并同步影响右侧“导出此 Excel”的文件内容。可选择是否保留：
+- zip 解压后包含 `Patients_info.csv`。
+- 按患者目录分别保存检测汇总和具体检测。
+- 输出多个患者 Excel。
 
-- 基本工作表
+患者编号类型可选择自动识别。自动识别会读取 `Patients_info.csv` 中的 `pat_no`，并与 `Tests_List` 文件名或患者目录名比对：如果更匹配 10 位补零编号，则使用 `pat_no`；如果更匹配原始编号，则使用 `zhuyuanhao`。
-- 未匹配检测内容项
-- 未检测到内容汇总表
 
-“未检测到内容汇总”会收集标准字段全部为 `Not_Find`、但存在未匹配检测内容的记录，并按检测原因排序汇总。
+## 结果规则
 
-未匹配检测内容会被规范化为独立列：表头为未匹配检测项目名，数据行仅保存对应检测值。
+- 系统默认输出完整结果。
-如果某个工作表没有任何未匹配检测项目，则不会显示“未匹配检测内容”标识列。标准字段全为 `Not_Find` 且没有未匹配值的空结果行会被移除。
+- `未匹配检测内容` 会规范化为独立列：表头为未匹配检测项目名，数据行只保存对应检测值。
+- 如果某个工作表没有任何未匹配检测项目，不显示 `未匹配检测内容` 标识列。
+- 标准检测字段全为 `Not_Find` 的记录会移入 `未检测到内容汇总`，不继续留在原基本工作表中。
+- 导出 Excel 会跟随页面上的内容选择和排序。

app/main.py

@@ -23,6 +23,16 @@ WORK_ROOT.mkdir(parents=True, exist_ok=True)
 
 APP_TITLE = "\u68c0\u6d4b\u6570\u636e\u5904\u7406"
 UNMATCHED = "\u672a\u5339\u914d\u68c0\u6d4b\u5185\u5bb9"
+SORT_OPTIONS = {
+    "name": "\u59d3\u540d",
+    "sample_time": "\u91c7\u6837\u65f6\u95f4",
+    "reason": "\u68c0\u6d4b\u539f\u56e0",
+}
+SORT_ORDERS = {
+    "asc": "\u5347\u5e8f",
+    "desc": "\u964d\u5e8f",
+}
+MAX_PREVIEW_ROWS = 10000
 
 app = FastAPI(title=APP_TITLE)
 
@@ -59,7 +69,7 @@ def index() -> str:
             </div>
             <div>
               <label for="preview_rows">\u6bcf\u4e2a\u5de5\u4f5c\u8868\u9884\u89c8\u884c\u6570</label>
-              <input id="preview_rows" name="preview_rows" type="number" min="5" max="200" value="20">
+              <input id="preview_rows" name="preview_rows" type="number" min="5" max="{MAX_PREVIEW_ROWS}" value="20">
             </div>
           </div>
           <button type="submit">\u5f00\u59cb\u5904\u7406</button>
@@ -113,20 +123,26 @@ async def process(
 @app.get("/result/{job_id}", response_class=HTMLResponse)
 def result_page(
     job_id: str,
-    preview_rows: int = Query(20, ge=5, le=200),
+    preview_rows: int = Query(20, ge=5, le=MAX_PREVIEW_ROWS),
     include_basic_sheets: bool = Query(True),
     include_unmatched_items: bool = Query(True),
     include_summary_sheet: bool = Query(True),
+    sort_by: str = Query("sample_time"),
+    sort_order: str = Query("asc"),
 ) -> str:
     job_dir = _get_job_dir(job_id)
     try:
         rows = _clean_preview_rows(preview_rows)
+        clean_sort_by = _clean_sort_by(sort_by)
+        clean_sort_order = _clean_sort_order(sort_order)
         result = summarize_job(
             job_dir,
             rows,
             include_basic_sheets=include_basic_sheets,
             include_unmatched_items=include_unmatched_items,
             include_summary_sheet=include_summary_sheet,
+            sort_by=clean_sort_by,
+            sort_order=clean_sort_order,
         )
     except ProcessingError as exc:
         raise HTTPException(status_code=404, detail=str(exc)) from exc
@@ -136,6 +152,8 @@ def result_page(
         include_basic_sheets=include_basic_sheets,
         include_unmatched_items=include_unmatched_items,
         include_summary_sheet=include_summary_sheet,
+        sort_by=clean_sort_by,
+        sort_order=clean_sort_order,
     )
 
 
@@ -145,6 +163,8 @@ def download_all(
     include_basic_sheets: bool = Query(True),
     include_unmatched_items: bool = Query(True),
     include_summary_sheet: bool = Query(True),
+    sort_by: str = Query("sample_time"),
+    sort_order: str = Query("asc"),
 ) -> FileResponse:
     job_dir = _get_job_dir(job_id)
     try:
@@ -153,6 +173,8 @@ def download_all(
             include_basic_sheets=include_basic_sheets,
             include_unmatched_items=include_unmatched_items,
             include_summary_sheet=include_summary_sheet,
+            sort_by=_clean_sort_by(sort_by),
+            sort_order=_clean_sort_order(sort_order),
         )
     except ProcessingError as exc:
         raise HTTPException(status_code=404, detail=str(exc)) from exc
@@ -166,6 +188,8 @@ def download_file(
     include_basic_sheets: bool = Query(True),
     include_unmatched_items: bool = Query(True),
     include_summary_sheet: bool = Query(True),
+    sort_by: str = Query("sample_time"),
+    sort_order: str = Query("asc"),
 ) -> FileResponse:
     job_dir = _get_job_dir(job_id)
     try:
@@ -175,6 +199,8 @@ def download_file(
             include_basic_sheets=include_basic_sheets,
             include_unmatched_items=include_unmatched_items,
             include_summary_sheet=include_summary_sheet,
+            sort_by=_clean_sort_by(sort_by),
+            sort_order=_clean_sort_order(sort_order),
         )
     except ProcessingError as exc:
         raise HTTPException(status_code=404, detail=str(exc)) from exc
@@ -200,19 +226,33 @@ def _get_job_dir(job_id: str) -> Path:
 
 
 def _clean_preview_rows(preview_rows: int) -> int:
-    return max(5, min(int(preview_rows or 20), 200))
+    return max(5, min(int(preview_rows or 20), MAX_PREVIEW_ROWS))
 
 
 def _checked(value: bool) -> str:
     return "checked" if value else ""
 
 
+def _selected(value: str, current: str) -> str:
+    return "selected" if value == current else ""
+
+
+def _clean_sort_by(sort_by: str) -> str:
+    return sort_by if sort_by in SORT_OPTIONS else "sample_time"
+
+
+def _clean_sort_order(sort_order: str) -> str:
+    return sort_order if sort_order in SORT_ORDERS else "asc"
+
+
 def _render_result(
     result: ProcessingResult,
     preview_rows: int = 20,
     include_basic_sheets: bool = True,
     include_unmatched_items: bool = True,
     include_summary_sheet: bool = True,
+    sort_by: str = "sample_time",
+    sort_order: str = "asc",
 ) -> str:
     total_sheets = sum(len(file.sheets) for file in result.files)
     total_rows = sum(max(sheet.rows - 1, 0) for file in result.files for sheet in file.sheets)
@@ -230,10 +270,20 @@ def _render_result(
         <label><input class="export-option" type="checkbox" data-param="include_basic_sheets" {_checked(include_basic_sheets)}> \u57fa\u672c\u5de5\u4f5c\u8868</label>
         <label><input class="export-option" type="checkbox" data-param="include_unmatched_items" {_checked(include_unmatched_items)}> \u672a\u5339\u914d\u68c0\u6d4b\u5185\u5bb9\u9879</label>
         <label><input class="export-option" type="checkbox" data-param="include_summary_sheet" {_checked(include_summary_sheet)}> \u672a\u68c0\u6d4b\u5230\u5185\u5bb9\u6c47\u603b\u8868</label>
+        <label for="sort_by">\u6392\u5e8f</label>
+        <select id="sort_by" class="result-option" data-param="sort_by">
+          <option value="name" {_selected("name", sort_by)}>\u6309\u59d3\u540d</option>
+          <option value="sample_time" {_selected("sample_time", sort_by)}>\u6309\u91c7\u6837\u65f6\u95f4</option>
+          <option value="reason" {_selected("reason", sort_by)}>\u6309\u68c0\u6d4b\u539f\u56e0</option>
+        </select>
+        <select id="sort_order" class="result-option" data-param="sort_order">
+          <option value="asc" {_selected("asc", sort_order)}>\u5347\u5e8f</option>
+          <option value="desc" {_selected("desc", sort_order)}>\u964d\u5e8f</option>
+        </select>
       </section>
       <form class="row-control" method="get" action="/result/{html.escape(result.job_id)}">
         <label for="preview_rows">\u6bcf\u4e2a\u5de5\u4f5c\u8868\u9884\u89c8\u884c\u6570</label>
-        <input id="preview_rows" name="preview_rows" type="number" min="5" max="200" value="{preview_rows}">
+        <input id="preview_rows" name="preview_rows" type="number" min="5" max="{MAX_PREVIEW_ROWS}" value="{preview_rows}">
         <button type="submit">\u5237\u65b0\u9884\u89c8</button>
       </form>
       <section class="hint">\u7ed3\u679c\u4e2d\u201c{UNMATCHED}\u201d\u5217\u4e3a\u989d\u5916\u8f85\u52a9\u4fe1\u606f\uff0c\u7528\u6765\u6807\u660e\u672a\u5f52\u5165\u6807\u51c6\u9879\u76ee\u7684\u68c0\u6d4b\u5185\u5bb9\u3002</section>
@@ -411,6 +461,11 @@ def _page_shell(
       margin: 0;
       font-weight: 500;
     }}
+    .export-options select {{
+      width: auto;
+      min-width: 128px;
+      padding: 8px 10px;
+    }}
     .checks label {{
       display: inline-flex;
       align-items: center;
@@ -591,6 +646,9 @@ def _page_shell(
       const options = Array.from(document.querySelectorAll('.export-option'));
       const params = new URLSearchParams();
       options.forEach((option) => params.set(option.dataset.param, option.checked ? 'true' : 'false'));
+      document.querySelectorAll('.result-option').forEach((option) => {{
+        params.set(option.dataset.param, option.value);
+      }});
       document.querySelectorAll('.export-link').forEach((link) => {{
         const base = link.dataset.base;
         const sep = base.includes('?') ? '&' : '?';
@@ -606,6 +664,9 @@ def _page_shell(
       document.querySelectorAll('.export-option').forEach((option) => {{
         url.searchParams.set(option.dataset.param, option.checked ? 'true' : 'false');
       }});
+      document.querySelectorAll('.result-option').forEach((option) => {{
+        url.searchParams.set(option.dataset.param, option.value);
+      }});
       window.location.href = url.toString();
     }}
     const resultForm = document.querySelector('.row-control');
@@ -619,6 +680,10 @@ def _page_shell(
       updateExportLinks();
       refreshResultWithOptions();
     }}));
+    document.querySelectorAll('.result-option').forEach((option) => option.addEventListener('change', () => {{
+      updateExportLinks();
+      refreshResultWithOptions();
+    }}));
     updateExportLinks();
   </script>
 </body>

app/processor.py

@@ -6,6 +6,7 @@ import sys
 import uuid
 import zipfile
 from dataclasses import dataclass
+from datetime import datetime
 from pathlib import Path
 
 from openpyxl import load_workbook
@@ -15,8 +16,12 @@ from openpyxl.styles import Font
 
 PROCESSOR_DIR = Path(__file__).resolve().parent / "processors"
 SUMMARY_SHEET_NAME = "\u672a\u68c0\u6d4b\u5230\u5185\u5bb9\u6c47\u603b"
+UNASSIGNED_SHEET_NAME = "\u672a\u5f52\u5c5e\u68c0\u6d4b\u5185\u5bb9"
 UNMATCHED_HEADER = "\u672a\u5339\u914d\u68c0\u6d4b\u5185\u5bb9"
 UNMATCHED_FILL = PatternFill(fill_type="solid", fgColor="FCE4D6")
+MAX_PREVIEW_ROWS = 10000
+SORT_FIELDS = {"none", "name", "sample_time", "reason"}
+SORT_ORDERS = {"asc", "desc"}
 
 
 class ProcessingError(Exception):
@@ -59,6 +64,8 @@ def run_processing(
     include_basic_sheets: bool = True,
     include_unmatched_items: bool = True,
     include_summary_sheet: bool = True,
+    sort_by: str = "sample_time",
+    sort_order: str = "asc",
 ) -> ProcessingResult:
     if mode not in {"auto", "v1", "v2"}:
         raise ProcessingError("处理模式不正确。")
@@ -142,6 +149,8 @@ def run_processing(
             include_basic_sheets=include_basic_sheets,
             include_unmatched_items=include_unmatched_items,
             include_summary_sheet=include_summary_sheet,
+            sort_by=sort_by,
+            sort_order=sort_order,
         )
 
     result_zip = job_dir / "result.zip"
@@ -160,6 +169,8 @@ def create_result_zip(
     include_basic_sheets: bool = True,
     include_unmatched_items: bool = True,
     include_summary_sheet: bool = True,
+    sort_by: str = "sample_time",
+    sort_order: str = "asc",
 ) -> Path:
     output_dir = job_dir / "output"
     result_zip = job_dir / "result.zip"
@@ -176,6 +187,8 @@ def create_result_zip(
                 include_basic_sheets=include_basic_sheets,
                 include_unmatched_items=include_unmatched_items,
                 include_summary_sheet=include_summary_sheet,
+                sort_by=sort_by,
+                sort_order=sort_order,
             )
 
     result_zip = export_dir / "result.zip"
@@ -189,6 +202,8 @@ def summarize_job(
     include_basic_sheets: bool = True,
     include_unmatched_items: bool = True,
     include_summary_sheet: bool = True,
+    sort_by: str = "sample_time",
+    sort_order: str = "asc",
 ) -> ProcessingResult:
     output_dir = job_dir / "output"
     if not output_dir.exists():
@@ -204,6 +219,8 @@ def summarize_job(
                 include_basic_sheets=include_basic_sheets,
                 include_unmatched_items=include_unmatched_items,
                 include_summary_sheet=include_summary_sheet,
+                sort_by=sort_by,
+                sort_order=sort_order,
             )
 
     xlsx_files = sorted(preview_dir.rglob("*.xlsx"))
@@ -226,6 +243,8 @@ def find_output_file(
     include_basic_sheets: bool = True,
     include_unmatched_items: bool = True,
     include_summary_sheet: bool = True,
+    sort_by: str = "sample_time",
+    sort_order: str = "asc",
 ) -> Path:
     output_dir = (job_dir / "output").resolve()
     target = (output_dir / relpath).resolve()
@@ -241,6 +260,8 @@ def find_output_file(
         include_basic_sheets=include_basic_sheets,
         include_unmatched_items=include_unmatched_items,
         include_summary_sheet=include_summary_sheet,
+        sort_by=sort_by,
+        sort_order=sort_order,
     )
     return export_target
 
@@ -258,6 +279,8 @@ def _apply_export_options(
     include_basic_sheets: bool,
     include_unmatched_items: bool,
     include_summary_sheet: bool,
+    sort_by: str,
+    sort_order: str,
 ) -> None:
     if not include_basic_sheets and not include_summary_sheet:
         raise ProcessingError("\u81f3\u5c11\u9700\u8981\u5bfc\u51fa\u57fa\u672c\u5de5\u4f5c\u8868\u6216\u672a\u68c0\u6d4b\u5230\u5185\u5bb9\u6c47\u603b\u8868\u3002")
@@ -278,6 +301,7 @@ def _apply_export_options(
         if not workbook.worksheets:
             workbook.create_sheet(SUMMARY_SHEET_NAME)
 
+        _sort_workbook(workbook, sort_by, sort_order)
         workbook.save(path)
     finally:
         workbook.close()
@@ -434,12 +458,16 @@ def _postprocess_workbook(
     include_basic_sheets: bool,
     include_unmatched_items: bool,
     include_summary_sheet: bool,
+    sort_by: str,
+    sort_order: str,
 ) -> None:
     workbook = load_workbook(path)
     try:
         _normalize_unmatched_columns(workbook)
-        _remove_empty_not_find_rows(workbook)
+        _deduplicate_regular_rows(workbook)
         summary_records = _collect_summary_records(workbook)
+        _remove_not_found_rows(workbook)
+        _remove_empty_unmatched_columns(workbook)
 
         if not include_unmatched_items:
             _remove_unmatched_columns(workbook)
@@ -455,6 +483,7 @@ def _postprocess_workbook(
         if not workbook.worksheets:
             workbook.create_sheet(SUMMARY_SHEET_NAME)
 
+        _sort_workbook(workbook, sort_by, sort_order)
         workbook.save(path)
     finally:
         workbook.close()
@@ -484,13 +513,14 @@ def _normalize_unmatched_columns(workbook) -> None:
                 if item_name not in item_names:
                     item_names.append(item_name)
 
+        if not item_names:
+            if sheet.max_column >= marker_col:
+                sheet.delete_cols(marker_col, sheet.max_column - marker_col + 1)
+            continue
+
         if sheet.max_column >= marker_col:
             sheet.delete_cols(marker_col, sheet.max_column - marker_col + 1)
 
-        if not item_names:
-            sheet.delete_cols(marker_col, 1)
-            continue
-
         sheet.cell(1, marker_col).value = UNMATCHED_HEADER
         sheet.cell(1, marker_col).font = Font(bold=True)
         sheet.cell(1, marker_col).fill = UNMATCHED_FILL
@@ -509,7 +539,7 @@ def _normalize_unmatched_columns(workbook) -> None:
                 cell.fill = UNMATCHED_FILL
 
 
-def _remove_empty_not_find_rows(workbook) -> None:
+def _remove_not_found_rows(workbook) -> None:
     for sheet in workbook.worksheets:
         if sheet.title == SUMMARY_SHEET_NAME or sheet.max_row < 2:
             continue
@@ -522,25 +552,37 @@ def _remove_empty_not_find_rows(workbook) -> None:
                 _cell_text(sheet.cell(row_index, col).value)
                 for col in range(5, standard_end + 1)
             ]
-            unmatched_values = []
+            if _all_standard_values_missing(standard_values):
-            if unmatched_index is not None:
-                unmatched_values = [
-                    _cell_text(sheet.cell(row_index, col).value)
-                    for col in range(unmatched_index + 2, sheet.max_column + 1)
-                ]
-            if _all_standard_values_missing(standard_values) and not any(unmatched_values):
                 sheet.delete_rows(row_index, 1)
 
 
+def _deduplicate_regular_rows(workbook) -> None:
+    for sheet in workbook.worksheets:
+        if sheet.title == SUMMARY_SHEET_NAME or sheet.max_row < 3:
+            continue
+
+        seen = set()
+        for row_index in range(sheet.max_row, 1, -1):
+            values = tuple(
+                _cell_text(sheet.cell(row_index, col).value)
+                for col in range(1, sheet.max_column + 1)
+            )
+            if not any(values):
+                continue
+            if values in seen:
+                sheet.delete_rows(row_index, 1)
+            else:
+                seen.add(values)
+
+
 def _collect_summary_records(workbook) -> list[dict[str, object]]:
     records: list[dict[str, object]] = []
     for sheet in workbook.worksheets:
-        if sheet.title == SUMMARY_SHEET_NAME or sheet.max_row < 2:
+        if sheet.title != UNASSIGNED_SHEET_NAME or sheet.max_row < 2:
             continue
         header = [_cell_text(sheet.cell(1, col).value) for col in range(1, sheet.max_column + 1)]
         unmatched_col = _find_header_index(header, UNMATCHED_HEADER)
-        if unmatched_col is None:
+        standard_end = unmatched_col if unmatched_col is not None else sheet.max_column
-            continue
 
         for row_index in range(2, sheet.max_row + 1):
             base_values = [_cell_text(sheet.cell(row_index, col).value) for col in range(1, 5)]
@@ -548,17 +590,12 @@ def _collect_summary_records(workbook) -> list[dict[str, object]]:
                 continue
             standard_values = [
                 _cell_text(sheet.cell(row_index, col).value)
-                for col in range(5, unmatched_col + 1)
+                for col in range(5, standard_end + 1)
             ]
-            item_values = {
+            if standard_values and not _all_standard_values_missing(standard_values):
-                _cell_text(sheet.cell(1, col).value): _cell_text(sheet.cell(row_index, col).value)
-                for col in range(unmatched_col + 2, sheet.max_column + 1)
-                if _cell_text(sheet.cell(1, col).value)
-            }
-            item_values = {name: value for name, value in item_values.items() if value}
-            if not item_values:
                 continue
-            if not _all_standard_values_missing(standard_values):
+            item_values = _summary_item_values(sheet, row_index, header, unmatched_col, standard_end)
+            if not item_values:
                 continue
             records.append(
                 {
@@ -573,6 +610,31 @@ def _collect_summary_records(workbook) -> list[dict[str, object]]:
     return records
 
 
+def _summary_item_values(
+    sheet,
+    row_index: int,
+    header: list[str],
+    unmatched_col: int | None,
+    standard_end: int,
+) -> dict[str, str]:
+    item_values: dict[str, str] = {}
+    if unmatched_col is not None:
+        for col in range(unmatched_col + 2, sheet.max_column + 1):
+            item_name = _cell_text(sheet.cell(1, col).value)
+            item_value = _cell_text(sheet.cell(row_index, col).value)
+            if item_name and item_value:
+                item_values[item_name] = item_value
+    if item_values:
+        return item_values
+
+    for col in range(5, standard_end + 1):
+        item_name = header[col - 1] if col - 1 < len(header) else ""
+        item_value = _cell_text(sheet.cell(row_index, col).value)
+        if item_name and item_value:
+            item_values[item_name] = item_value
+    return item_values
+
+
 def _replace_summary_sheet(workbook, records: list[dict[str, object]]) -> None:
     if SUMMARY_SHEET_NAME in workbook.sheetnames:
         workbook.remove(workbook[SUMMARY_SHEET_NAME])
@@ -631,6 +693,163 @@ def _remove_unmatched_columns(workbook) -> None:
             sheet.delete_cols(unmatched_col + 1, sheet.max_column - unmatched_col)
 
 
+def _remove_empty_unmatched_columns(workbook) -> None:
+    for sheet in workbook.worksheets:
+        if sheet.title == SUMMARY_SHEET_NAME or sheet.max_row < 1:
+            continue
+        header = [_cell_text(sheet.cell(1, col).value) for col in range(1, sheet.max_column + 1)]
+        unmatched_col = _find_header_index(header, UNMATCHED_HEADER)
+        if unmatched_col is None:
+            continue
+
+        first_col = unmatched_col + 1
+        item_cols = [
+            col
+            for col in range(first_col + 1, sheet.max_column + 1)
+            if _cell_text(sheet.cell(1, col).value)
+        ]
+        used_item_cols = [
+            col
+            for col in item_cols
+            if any(_cell_text(sheet.cell(row, col).value) for row in range(2, sheet.max_row + 1))
+        ]
+        if not used_item_cols:
+            sheet.delete_cols(first_col, sheet.max_column - unmatched_col)
+            continue
+
+        for col in reversed(item_cols):
+            if col not in used_item_cols:
+                sheet.delete_cols(col, 1)
+
+
+def _sort_workbook(workbook, sort_by: str, sort_order: str) -> None:
+    sort_by = sort_by if sort_by in SORT_FIELDS else "sample_time"
+    sort_order = sort_order if sort_order in SORT_ORDERS else "asc"
+    if sort_by == "none":
+        return
+
+    reverse = sort_order == "desc"
+    for sheet in workbook.worksheets:
+        if sheet.max_row < 3:
+            continue
+        if sheet.title == SUMMARY_SHEET_NAME:
+            _sort_summary_sheet(sheet, sort_by, reverse)
+        else:
+            _sort_regular_sheet(sheet, sort_by, reverse)
+
+
+def _sort_regular_sheet(sheet, sort_by: str, reverse: bool) -> None:
+    header = [_cell_text(sheet.cell(1, col).value) for col in range(1, sheet.max_column + 1)]
+    sort_col = _sort_column_from_header(header, sort_by)
+    if sort_col is None:
+        return
+
+    rows = _read_rows(sheet, 2, sheet.max_row)
+    _sort_rows(rows, sort_col, sort_by, reverse)
+    _write_rows(sheet, 2, rows)
+
+
+def _sort_summary_sheet(sheet, sort_by: str, reverse: bool) -> None:
+    row_index = 1
+    while row_index <= sheet.max_row:
+        row = [_cell_text(sheet.cell(row_index, col).value) for col in range(1, min(sheet.max_column, 4) + 1)]
+        if not _is_summary_header_values(row):
+            row_index += 1
+            continue
+
+        header = [_cell_text(sheet.cell(row_index, col).value) for col in range(1, sheet.max_column + 1)]
+        sort_col = _sort_column_from_header(header, sort_by)
+        if sort_col is None:
+            row_index += 1
+            continue
+
+        start = row_index + 1
+        end = start
+        while end <= sheet.max_row:
+            next_row = [
+                _cell_text(sheet.cell(end, col).value)
+                for col in range(1, min(sheet.max_column, 4) + 1)
+            ]
+            if _is_summary_header_values(next_row):
+                break
+            end += 1
+
+        if end > start:
+            rows = _read_rows(sheet, start, end - 1)
+            _sort_rows(rows, sort_col, sort_by, reverse)
+            _write_rows(sheet, start, rows)
+        row_index = end
+
+
+def _sort_column_from_header(header: list[str], sort_by: str) -> int | None:
+    if sort_by == "name":
+        names = ["姓名"]
+    elif sort_by == "sample_time":
+        names = ["采样时间"]
+    elif sort_by == "reason":
+        names = ["检测原因"]
+    else:
+        return None
+
+    for index, value in enumerate(header):
+        if any(value == name or value.startswith(name) for name in names):
+            return index
+    return None
+
+
+def _read_rows(sheet, start: int, end: int) -> list[list[object]]:
+    return [
+        [sheet.cell(row_index, col).value for col in range(1, sheet.max_column + 1)]
+        for row_index in range(start, end + 1)
+    ]
+
+
+def _write_rows(sheet, start: int, rows: list[list[object]]) -> None:
+    for offset, row in enumerate(rows):
+        row_index = start + offset
+        for col, value in enumerate(row, start=1):
+            sheet.cell(row_index, col).value = value
+
+
+def _sort_rows(rows: list[list[object]], sort_col: int, sort_by: str, reverse: bool) -> None:
+    rows.sort(key=lambda row: _sort_value(row[sort_col], sort_by))
+    if reverse:
+        filled = [row for row in rows if _cell_text(row[sort_col])]
+        empty = [row for row in rows if not _cell_text(row[sort_col])]
+        filled.reverse()
+        rows[:] = filled + empty
+
+
+def _sort_value(value: object, sort_by: str) -> tuple[int, object]:
+    text = _cell_text(value)
+    if not text:
+        return (1, "")
+    if sort_by == "sample_time":
+        parsed = _parse_datetime(text)
+        if parsed is not None:
+            return (0, parsed.isoformat())
+    return (0, text)
+
+
+def _parse_datetime(value: str) -> datetime | None:
+    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y/%m/%d %H:%M:%S", "%Y-%m-%d", "%Y/%m/%d"):
+        try:
+            return datetime.strptime(value, fmt)
+        except ValueError:
+            continue
+    return None
+
+
+def _is_summary_header_values(row: list[str]) -> bool:
+    return (
+        len(row) >= 4
+        and row[0] == "姓名"
+        and row[1] == "住院号"
+        and row[2] == "采样时间"
+        and row[3].startswith("检测原因")
+    )
+
+
 def _find_header_index(header: list[str], name: str) -> int | None:
     for index, value in enumerate(header):
         if value == name:
@@ -677,7 +896,7 @@ def _summarize_workbook(path: Path, output_dir: Path, preview_rows: int) -> Exce
     try:
         for sheet in workbook.worksheets:
             preview: list[list[str]] = []
-            for row in sheet.iter_rows(max_row=max(2, min(preview_rows, 200)), values_only=True):
+            for row in sheet.iter_rows(max_row=max(2, min(preview_rows, MAX_PREVIEW_ROWS)), values_only=True):
                 preview.append([_cell_to_text(value) for value in row])
             sheets.append(
                 SheetSummary(

app/processors/V1-ALL_convert_Lab_Test_data.py

@@ -3,6 +3,7 @@
 import csv, sys, os, copy, re 
 import os.path as osp
 from openpyxl import Workbook, load_workbook
+from dynamic_router import UNASSIGNED_SHEET_NAME, UNMATCHED_HEADER, append_routed_report, route_detail_rows
 
 # 向特定excel的sheet中添加内容
 workbook = None  # 全局变量，初始值为 None
@@ -158,8 +159,8 @@ ALL_tests = [# 血细胞 # (1):血细胞分析+五分类；(2):血细胞分析+
             {'test_rptunitid':[str(3), str(110),str(90)], 'test_check_name':'凝血', 'test_check_list':["凝血酶原时间", "凝血酶原活动度", "血浆凝血酶原时间比值", "凝血酶原国际标准化比值", "活化部分凝血活酶时间", "活化部分凝血活酶比值", "凝血酶时间", "凝血酶时间比值", "纤维蛋白原含量", "D-二聚体测定", "纤维蛋白原降解产物", "凝血酶生成时间","凝血速率","血小板功能"],\
              'test_check_list_all':["凝血酶原时间*", ["凝血酶原活动度", r"凝血酶原活度\(%\)"], ["凝血酶原比值", "血浆凝血酶原时间比值"], ["凝血酶原标准化比值", "凝血酶原国际标准化比值"], ["活化部分凝血活酶时间*", "活化部分凝血酶时间*"], "活化部分凝血活酶比值", "凝血酶时间", "凝血酶时间比值", "纤维蛋白原含量", [r"D-二聚体\(sysmex\)", "D-二聚体测定*"], [r"纤维蛋白\(原\)降解产物", "纤维蛋白原降解产物"], "凝血酶生成时间","凝血速率","血小板功能"], 'test_result_col_name':"result_str"},
             # 肝功 # (5):平诊肝功十四项+平诊电解质八项+平诊肾功七项；(20)急诊肝功十二项_急诊肾功五项[复]_急诊电解质七项[复]；东院区：（108）：传染性指标检测八项
-            {'test_rptunitid':[str(5), str(20), str(108)], 'test_check_name':'肝功', 'test_check_list':["谷草转氨酶", "谷丙转氨酶", "谷草/谷丙", "碱性磷酸酶", "γ谷氨酰氨转肽酶", "总胆红素", "直接胆红素", "间接胆红素", "胆碱脂酶", "总胆固醇", "总蛋白", "白蛋白", "球蛋白", "白球比", "总胆汁酸", "a-L-岩藻糖苷酶", "前白蛋白", "超氧化物歧化酶", "尿素", "肌酐", "胱抑素C", "葡萄糖", "尿酸", "钾", "钠", "氯", "磷", "钙", "镁", "二氧化碳结合率", "离子间隙", "糖化白蛋白", "视黄醇结合蛋白", "eGFR(CKD-EPI)", "a-l岩藻糖苷酶", "超氧化物歧化酶", "eGFR(MDRD)", r"eGFR单位ml/min/1.73m^2", "8597", "乳酸测定", "乳酸脱氢酶", "羟丁酸脱氢酶", "肌酸激酶", "肌酸激酶同工酶", "载脂蛋白A", "载脂蛋白B", "载脂蛋白E", "脂蛋白（a）", "缺血修饰白蛋白", "甘油三酯", "高密度脂蛋白", "低密度脂蛋白", ], \
+            {'test_rptunitid':[str(5), str(20), str(108)], 'test_check_name':'肝功', 'test_check_list':["谷草转氨酶", "谷丙转氨酶", "谷草/谷丙", "碱性磷酸酶", "γ谷氨酰氨转肽酶", "总胆红素", "直接胆红素", "间接胆红素", "胆碱酯酶", "总胆固醇", "总蛋白", "白蛋白", "球蛋白", "白球比", "总胆汁酸", "a-L-岩藻糖苷酶", "前白蛋白", "超氧化物歧化酶", "尿素", "肌酐", "胱抑素C", "葡萄糖", "尿酸", "钾", "钠", "氯", "磷", "钙", "镁", "二氧化碳结合率", "离子间隙", "糖化白蛋白", "视黄醇结合蛋白", "eGFR(CKD-EPI)", "a-l岩藻糖苷酶", "超氧化物歧化酶", "eGFR(MDRD)", r"eGFR单位ml/min/1.73m^2", "8597", "乳酸测定", "乳酸脱氢酶", "羟丁酸脱氢酶", "肌酸激酶", "肌酸激酶同工酶", "载脂蛋白A", "载脂蛋白B", "载脂蛋白E", "脂蛋白（a）", "缺血修饰白蛋白", "甘油三酯", "高密度脂蛋白", "低密度脂蛋白", ], \
-             'test_check_list_all':["谷草转氨酶", "谷丙转氨酶", "谷草/谷丙", "碱性磷酸酶", ["γ谷氨酰氨转肽酶", "γ-谷氨酰转肽酶"], "总胆红素", "直接胆红素", "间接胆红素", "胆碱脂酶", "总胆固醇", "总蛋白", "白蛋白", "球蛋白", "白球比", "总胆汁酸", "a-L-岩藻糖苷酶", "前白蛋白", "超氧化物歧化酶", "尿素", "肌酐", "胱抑素C", "葡萄糖", "尿酸", "钾", "钠", "氯", "磷", "钙", "镁", "二氧化碳结合率", "离子间隙", "糖化白蛋白", "视黄醇结合蛋白", r"eGFR\(CKD-EPI\)", "a-l岩藻糖苷酶", "超氧化物歧化酶", r"eGFR\(MDRD\)", r"eGFR单位ml/min/1.73m\^2", "8597", "乳酸测定", "乳酸脱氢酶", "羟丁酸脱氢酶", "肌酸激酶", "肌酸激酶同工酶", "载脂蛋白A", "载脂蛋白B", "载脂蛋白E", "脂蛋白（a）", "缺血修饰白蛋白", "甘油三酯", "高密度脂蛋白", "低密度脂蛋白", ], 'test_result_col_name':"result_str"},
+             'test_check_list_all':["谷草转氨酶", "谷丙转氨酶", "谷草/谷丙", "碱性磷酸酶", ["γ谷氨酰氨转肽酶", "γ-谷氨酰转肽酶"], "总胆红素", "直接胆红素", "间接胆红素", "胆碱酯酶", "总胆固醇", "总蛋白", "白蛋白", "球蛋白", "白球比", "总胆汁酸", "a-L-岩藻糖苷酶", "前白蛋白", "超氧化物歧化酶", "尿素", "肌酐", "胱抑素C", "葡萄糖", "尿酸", "钾", "钠", "氯", "磷", "钙", "镁", "二氧化碳结合率", "离子间隙", "糖化白蛋白", "视黄醇结合蛋白", r"eGFR\(CKD-EPI\)", "a-l岩藻糖苷酶", "超氧化物歧化酶", r"eGFR\(MDRD\)", r"eGFR单位ml/min/1.73m\^2", "8597", "乳酸测定", "乳酸脱氢酶", "羟丁酸脱氢酶", "肌酸激酶", "肌酸激酶同工酶", "载脂蛋白A", "载脂蛋白B", "载脂蛋白E", "脂蛋白（a）", "缺血修饰白蛋白", "甘油三酯", "高密度脂蛋白", "低密度脂蛋白", ], 'test_result_col_name':"result_str"},
             # 各类肿瘤标志物 # (6):各类肿瘤标志物，e.g.肿瘤标志物肺癌六项[复]、肿瘤标志物十项(男)；东院区：（108）：肿瘤标志物肺癌六项[复]
             {'test_rptunitid':[str(6), str(108)], 'test_check_name':'各类肿瘤标志物', 'test_check_list':["胃泌素释放肽前体(罗氏)", "鳞状上皮细胞癌抗原(罗氏)", "癌胚抗原", "甲胎蛋白", "糖类抗原125", "糖类抗原199", "糖类抗原724", "细胞角蛋白19片段", "神经元特异性烯醇化酶测定", "总前列腺特异性抗原", "游离前列腺抗原", "游离/总", "糖类抗原153","人附睾蛋白","绝经前 ROMA","绝经后 ROMA","铁蛋白","叶酸","维生素B12","高尔基体蛋白73测定","胃蛋白酶原Ⅰ","胃蛋白酶原Ⅱ","胃蛋白酶原Ⅰ/Ⅱ","降钙素","甲状腺球蛋白"],\
              'test_check_list_all':[r"胃泌素释放肽前体\(罗氏\)", ["鳞状上皮细胞癌相关抗原", r"鳞状上皮细胞癌抗原\(罗氏\)"], "癌胚抗原", "甲胎蛋白", "糖类抗原125", "糖类抗原199", "糖类抗原724", "细胞角蛋白19片段", "神经元特异性烯醇化酶测定", "总前列腺特异性抗原", "游离前列腺抗原", "游离/总", "糖类抗原153","人附睾蛋白","绝经前 ROMA","绝经后 ROMA","铁蛋白","叶酸","维生素B12","高尔基体蛋白73测定","胃蛋白酶原Ⅰ","胃蛋白酶原Ⅱ","胃蛋白酶原Ⅰ/Ⅱ","降钙素","甲状腺球蛋白"], 'test_result_col_name':"result_str"},
@@ -215,6 +216,7 @@ for test in ALL_tests: # 各项检测信息
         add_content_to_excel(result_save_pth, test['test_check_name'], front_content + basic_content + test["test_check_list"] + ['未匹配检测内容'])
     else:
         add_content_to_excel(result_save_pth, test['test_check_name'], front_content + basic_content + test["test_check_list"])
+add_content_to_excel(result_save_pth, UNASSIGNED_SHEET_NAME, front_content + basic_content + [UNMATCHED_HEADER])
 
 
 ####### 获取所有患者头部信息（Front_line） #######
@@ -251,113 +253,36 @@ for pat_no in pat_no_col:
         Error(str(pat_no)+"的excel_head为空")
         exit()
     print("__处理患者头为__:", excel_head)
-    # 遍历所有rptunitid
+    with open(osp.join(patno_dir, pat_no+'.csv'), "r", encoding='utf-8-sig') as file:
-    for test in ALL_tests:
+        reader = list(csv.DictReader(file))
-        # 特定检查rptunitid
+        print("患者检查总次数为", len(reader))
-        test_rptunitid = test['test_rptunitid']
-        if not isinstance(test_rptunitid, list) and not isinstance(test_rptunitid, tuple):
-            test_rptunitid = [test_rptunitid]
-        # 检查名
-        test_check_name = test['test_check_name']
-        # 检查项
-        test_check_list_all = test['test_check_list_all']
-        # 检查项_名
-        test_check_list = test['test_check_list']
-        if len(test_check_list_all) != len(test_check_list):
-            Error("test_check_list_all长度和test_check_list不同")
-            print(f"test_check_list_all:{test_check_list_all}\ntest_check_list:{test_check_list}")
-            exit()
-        # 检查结果所在列的列名
-        test_result_col_name = test['test_result_col_name']
-        # 存放检查结果的dict
-        test_check_result = {}
-        
-        print("获取患者 ", test_check_name, " 检查结果")
-    
-        ### 打开patno/pat_no.csv文件获取更详细信息 ###
-        with open(osp.join(patno_dir, pat_no+'.csv'), "r", encoding='utf-8-sig') as file:
-            reader = csv.DictReader(file) # 读取文件
-            # 汇总特定检查行
-            test_rptunitid_rows  = [ row for row in reader if row['rptunitid'] in test_rptunitid ]
-            print("患者检查 ", test_check_name, "次数为", len(test_rptunitid_rows))
 
-            # 遍历所有特定检查行，提取关键信息
+    for i, row_1 in enumerate(reader):
-            rows_not_match = [] # 每一行内不匹配的内容
+        print("处理患者第", i+1, "次检查")
-            for i in range(len(test_rptunitid_rows)):
+        excel_basic = []
-                print("处理患者第", i+1, "次检查")
+        for item in Basic_line:
-                row_1 = test_rptunitid_rows[i]
+            excel_basic.append(row_1[Basic_line[item]])
-                sampled_dt = row_1['sampled_dt'] # 获取时间信息
-                ###### basic信息头生成 ######
-                excel_basic = []
-                for item in Basic_line:
-                    excel_basic.append(row_1[Basic_line[item]])
-                
-                ### 打开Patient_detail_infos/pat_no/文件 ###，获取特定检查文件存储路径
-                if data_type == "zhuyuanhao":
-                    row_file_path = os.path.join(Patient_detail_infos_dir, pat_no, 'None' + "_" + row_1['reporttype'] + "_" + row_1['rptunitid'] + "_" + row_1['reportid'].replace("_","-") + "_" + row_1['rechkdt'].replace(" ","-").replace(":","-").replace("_","-") +".csv")
-                elif data_type == "pat_no":
-                    row_file_path = os.path.join(Patient_detail_infos_dir, pat_no, pat_no + "_" + row_1['reporttype'] + "_" + row_1['rptunitid'] + "_" + row_1['reportid'].replace("_","-") + "_" + row_1['rechkdt'].replace(" ","-").replace(":","-").replace("_","-") +".csv")
-                with open(row_file_path, "r") as row_file:
-                    row_reader_ = csv.DictReader(row_file) # 读取文件，迭代器只能读取一次
-                    row_reader = []
-                    for r in row_reader_:
-                        row_reader.append(r)
-                    # 寻找所有检查项的值
-                    match = False # 检测内容中是否含有检测项目
-                    rows_not_match = [] # 每一行内不匹配的内容
-                    # 遍历所有行，查看对应检查结果
-                    for row_2 in row_reader:
-                        row_test_name_exist = False
-                        row_test_name = row_2['rpt_itemname']
-                        # 遍历所有待检查项
-                        for j in range(len(test_check_list_all)):
-                            # 检查项名称
-                            test_result_name = test_check_list[j]
-                            # 遍历所有完整版检查项目
-                            test_checks = test_check_list_all[j]
-                            if isinstance(test_checks, str):
-                                test_checks = [test_checks]
-                            for test_check in test_checks:
-                                # print(test_check, row_2['rpt_itemname'],"___")
-                                if match_re(row_test_name, test_check):# test_check == row_test_name: # 检测项目属于其中
-                                    match = True
-                                    row_test_name_exist = True
-                                    # temp临时存储变量，如果其为'.'，则结果变为'None'
-                                    temp = row_2[test_result_col_name]
-                                    if (temp == '.' or temp == '') :
-                                        temp = 'None' 
-                                    test_check_result[test_result_name] = temp
-                                    break
-                            # 如果结果中存在row_test_name_exist的话，退出循环
-                            if row_test_name_exist == True:
-                                break
-                        # 如果没有row_test_name_exist的话，将信息加入行信息中
-                        if row_test_name_exist == False:
-                            rows_not_match.append(f"{row_test_name}：{row_2.get(test_result_col_name, '')}")
 
-                    # 如果没有寻找到对应test_check_result的话，设置为Not_Find
+        if data_type == "zhuyuanhao":
-                    for test_result_name in test_check_list:
+            row_file_path = os.path.join(Patient_detail_infos_dir, pat_no, 'None' + "_" + row_1['reporttype'] + "_" + row_1['rptunitid'] + "_" + row_1['reportid'].replace("_","-") + "_" + row_1['rechkdt'].replace(" ","-").replace(":","-").replace("_","-") +".csv")
-                        if not test_result_name in  test_check_result:
+        elif data_type == "pat_no":
-                            test_check_result[test_result_name] = 'Not_Find'
+            row_file_path = os.path.join(Patient_detail_infos_dir, pat_no, pat_no + "_" + row_1['reporttype'] + "_" + row_1['rptunitid'] + "_" + row_1['reportid'].replace("_","-") + "_" + row_1['rechkdt'].replace(" ","-").replace(":","-").replace("_","-") +".csv")
 
-                    # 如果没有检测到相匹配内容，且不输出所有信息，continue
+        with open(row_file_path, "r", encoding='utf-8-sig') as row_file:
-                    if (match == False and show_all_infos == False):
+            detail_rows = list(csv.DictReader(row_file))
-                        continue
+
-                    # 进行进一步操作 ing...    
+        sheet_results, unassigned_items = route_detail_rows(detail_rows, ALL_tests, row_1.get("req_reason", ""))
-                    
+        append_routed_report(
-                    
+            add_content_to_excel,
-                    # 抽取结果格式转换
+            result_save_pth,
-                    excel_fromat_result = []
+            ALL_tests,
-                    for test_item in test['test_check_list']:
+            excel_head,
-                        excel_fromat_result.append(test_check_result[test_item])
+            excel_basic,
-                    # print("输出结果：", excel_fromat_result)
+            sheet_results,
-                    if show_not_match == True:
+            unassigned_items,
-                        add_content_to_excel(result_save_pth, test['test_check_name'], excel_head + excel_basic + excel_fromat_result + rows_not_match)
+            show_not_match,
-                    else:
+        )
-                        add_content_to_excel(result_save_pth, test['test_check_name'], excel_head + excel_basic + excel_fromat_result)
     
     # 每处理一个患者数据，保存相关信息
 save_excel()                
      
-
-

app/processors/V2-Every_Pat_File_convert_Lab_Test_data.py

@@ -4,6 +4,7 @@ import csv, sys, os, copy, re, argparse
 import os.path as osp
 from openpyxl import Workbook, load_workbook
 from V2_Data import Front_line, Basic_line, ALL_tests
+from dynamic_router import UNASSIGNED_SHEET_NAME, UNMATCHED_HEADER, append_routed_report, route_detail_rows
 
 # 向特定excel的sheet中添加内容
 workbook = None  # 全局变量，初始值为 None
@@ -192,6 +193,7 @@ for pat_no in os.listdir(file_dir): # 遍历 file_dir 下的所有文件和文
             add_content_to_excel(result_save_pth, test['test_check_name'], front_content + basic_content + test["test_check_list"] + ['未匹配检测内容'])
         else:
             add_content_to_excel(result_save_pth, test['test_check_name'], front_content + basic_content + test["test_check_list"])
+    add_content_to_excel(result_save_pth, UNASSIGNED_SHEET_NAME, front_content + basic_content + [UNMATCHED_HEADER])
 
     # 信息头生成
     excel_head = []
@@ -216,110 +218,35 @@ for pat_no in os.listdir(file_dir): # 遍历 file_dir 下的所有文件和文
     ####### 获取所有患者检测信息（base_line，pat_no/pato_no.csv） #######
     # 遍历所有患者（pat_no）
         
-    # 遍历所有rptunitid
+    with open(patno_pth, "r", encoding="utf-8-sig") as file:
-    for test in ALL_tests:
+        report_rows = list(csv.DictReader(file))
-        # 特定检查rptunitid
+        print("患者检查总次数为", len(report_rows))
-        test_rptunitid = test['test_rptunitid']
-        if not isinstance(test_rptunitid, list) and not isinstance(test_rptunitid, tuple):
-            test_rptunitid = [test_rptunitid]
-        # 检查名
-        test_check_name = test['test_check_name']
-        # 检查项
-        test_check_list_all = test['test_check_list_all']
-        # 检查项_名
-        test_check_list = test['test_check_list']
-        if len(test_check_list_all) != len(test_check_list):
-            Error("test_check_list_all长度和test_check_list不同", error_dir=osp.join(pat_file_dir, "Error.txt"))
-            print(f"test_check_list_all:{test_check_list_all}\ntest_check_list:{test_check_list}")
-            exit()
-        # 检查结果所在列的列名
-        test_result_col_name = test['test_result_col_name']
-        # 存放检查结果的dict
-        test_check_result = {}
-        
-        print("获取患者 ", test_check_name, " 检查结果")
-    
-        ### 打开patno/pat_no.csv文件获取更详细信息 ###
-        with open(patno_pth, "r", encoding="utf-8-sig") as file:
-            reader = csv.DictReader(file) # 读取文件
-            # 汇总特定检查行
-            test_rptunitid_rows  = [ row for row in reader if row['rptunitid'] in test_rptunitid ]
-            print("患者检查 ", test_check_name, "次数为", len(test_rptunitid_rows))
 
-            # 遍历所有特定检查行，提取关键信息
+    for i, row_1 in enumerate(report_rows):
-            rows_not_match = [] # 每一行内不匹配的内容
+        print("处理患者第", i+1, "次检查")
-            for i in range(len(test_rptunitid_rows)):
+        excel_basic = []
-                print("处理患者第", i+1, "次检查")
+        for item in Basic_line:
-                row_1 = test_rptunitid_rows[i]
+            excel_basic.append(row_1[Basic_line[item]])
-                sampled_dt = row_1['sampled_dt'] # 获取时间信息
-                ###### basic信息头生成 ######
-                excel_basic = []
-                for item in Basic_line:
-                    excel_basic.append(row_1[Basic_line[item]])
-                
-                ### 打开Patient_detail_infos/pat_no/文件 ###，获取特定检查文件存储路径
-                if data_type == "zhuyuanhao":
-                    row_file_path = os.path.join(Patient_detail_infos_dir, 'None' + "_" + row_1['reporttype'] + "_" + row_1['rptunitid'] + "_" + row_1['reportid'].replace("_","-") + "_" + row_1['rechkdt'].replace(" ","-").replace(":","-").replace("_","-") +".csv")
-                elif data_type == "pat_no":
-                    row_file_path = os.path.join(Patient_detail_infos_dir, pat_no + "_" + row_1['reporttype'] + "_" + row_1['rptunitid'] + "_" + row_1['reportid'].replace("_","-") + "_" + row_1['rechkdt'].replace(" ","-").replace(":","-").replace("_","-") +".csv")
-                with open(row_file_path, "r", encoding="utf-8-sig") as row_file:
-                    row_reader_ = csv.DictReader(row_file) # 读取文件，迭代器只能读取一次
-                    row_reader = []
-                    for r in row_reader_:
-                        row_reader.append(r)
-                    # 寻找所有检查项的值
-                    match = False # 检测内容中是否含有检测项目
-                    rows_not_match = [] # 每一行内不匹配的内容
-                    # 遍历所有行，查看对应检查结果
-                    for row_2 in row_reader:
-                        row_test_name_exist = False
-                        row_test_name = row_2['rpt_itemname']
-                        # 遍历所有待检查项
-                        for j in range(len(test_check_list_all)):
-                            # 检查项名称
-                            test_result_name = test_check_list[j]
-                            # 遍历所有完整版检查项目
-                            test_checks = test_check_list_all[j]
-                            if isinstance(test_checks, str):
-                                test_checks = [test_checks]
-                            for test_check in test_checks:
-                                # print(test_check, row_2['rpt_itemname'],"___")
-                                if match_re(row_test_name, test_check):# test_check == row_test_name: # 检测项目属于其中
-                                    match = True
-                                    row_test_name_exist = True
-                                    # temp临时存储变量，如果其为'.'，则结果变为'None'
-                                    temp = row_2[test_result_col_name]
-                                    if (temp == '.' or temp == '') :
-                                        temp = 'None' 
-                                    test_check_result[test_result_name] = temp
-                                    break
-                            # 如果结果中存在row_test_name_exist的话，退出循环
-                            if row_test_name_exist == True:
-                                break
-                        # 如果没有row_test_name_exist的话，将信息加入行信息中
-                        if row_test_name_exist == False:
-                            rows_not_match.append(f"{row_test_name}：{row_2.get(test_result_col_name, '')}")
 
-                    # 如果没有寻找到对应test_check_result的话，设置为Not_Find
+        if data_type == "zhuyuanhao":
-                    for test_result_name in test_check_list:
+            row_file_path = os.path.join(Patient_detail_infos_dir, 'None' + "_" + row_1['reporttype'] + "_" + row_1['rptunitid'] + "_" + row_1['reportid'].replace("_","-") + "_" + row_1['rechkdt'].replace(" ","-").replace(":","-").replace("_","-") +".csv")
-                        if not test_result_name in  test_check_result:
+        elif data_type == "pat_no":
-                            test_check_result[test_result_name] = 'Not_Find'
+            row_file_path = os.path.join(Patient_detail_infos_dir, pat_no + "_" + row_1['reporttype'] + "_" + row_1['rptunitid'] + "_" + row_1['reportid'].replace("_","-") + "_" + row_1['rechkdt'].replace(" ","-").replace(":","-").replace("_","-") +".csv")
 
-                    # 如果没有检测到相匹配内容，且不输出所有信息，continue
+        with open(row_file_path, "r", encoding="utf-8-sig") as row_file:
-                    if (match == False and show_all_infos == False):
+            detail_rows = list(csv.DictReader(row_file))
-                        continue
+
-                    # 进行进一步操作 ing...    
+        sheet_results, unassigned_items = route_detail_rows(detail_rows, ALL_tests, row_1.get("req_reason", ""))
-                    
+        append_routed_report(
-                    
+            add_content_to_excel,
-                    # 抽取结果格式转换
+            result_save_pth,
-                    excel_fromat_result = []
+            ALL_tests,
-                    for test_item in test['test_check_list']:
+            excel_head,
-                        excel_fromat_result.append(test_check_result[test_item])
+            excel_basic,
-                    # print("输出结果：", excel_fromat_result)
+            sheet_results,
-                    if show_not_match == True:
+            unassigned_items,
-                        add_content_to_excel(result_save_pth, test['test_check_name'], excel_head + excel_basic + excel_fromat_result + rows_not_match)
+            show_not_match,
-                    else:
+        )
-                        add_content_to_excel(result_save_pth, test['test_check_name'], excel_head + excel_basic + excel_fromat_result)
     
     # 每处理一个患者数据，保存相关信息
     save_excel()

app/processors/V2_Data.py

@@ -15,8 +15,8 @@ ALL_tests = [# 血细胞 # (1):血细胞分析+五分类；(2):血细胞分析+
             {'test_rptunitid':[str(3), str(110),str(90)], 'test_check_name':'凝血', 'test_check_list':["凝血酶原时间", "凝血酶原活动度", "血浆凝血酶原时间比值", "凝血酶原国际标准化比值", "活化部分凝血活酶时间", "活化部分凝血活酶比值", "凝血酶时间", "凝血酶时间比值", "纤维蛋白原含量", "D-二聚体测定", "纤维蛋白原降解产物", "凝血酶生成时间","凝血速率","血小板功能"],\
              'test_check_list_all':["凝血酶原时间*", ["凝血酶原活动度", r"凝血酶原活度\(%\)"], ["凝血酶原比值", "血浆凝血酶原时间比值"], ["凝血酶原标准化比值", "凝血酶原国际标准化比值"], ["活化部分凝血活酶时间*", "活化部分凝血酶时间*"], "活化部分凝血活酶比值", "凝血酶时间", "凝血酶时间比值", "纤维蛋白原含量", [r"D-二聚体\(sysmex\)", "D-二聚体测定*"], [r"纤维蛋白\(原\)降解产物", "纤维蛋白原降解产物"], "凝血酶生成时间","凝血速率","血小板功能"], 'test_result_col_name':"result_str"},
             # 肝功 # (5):平诊肝功十四项+平诊电解质八项+平诊肾功七项；(20)急诊肝功十二项_急诊肾功五项[复]_急诊电解质七项[复]；东院区：（108）：传染性指标检测八项
-            {'test_rptunitid':[str(5), str(20), str(108)], 'test_check_name':'肝功', 'test_check_list':["谷草转氨酶", "谷丙转氨酶", "谷草/谷丙", "碱性磷酸酶", "γ谷氨酰氨转肽酶", "总胆红素", "直接胆红素", "间接胆红素", "胆碱脂酶", "总胆固醇", "总蛋白", "白蛋白", "球蛋白", "白球比", "总胆汁酸", "a-L-岩藻糖苷酶", "前白蛋白", "超氧化物歧化酶", "尿素", "肌酐", "胱抑素C", "葡萄糖", "尿酸", "钾", "钠", "氯", "磷", "钙", "镁", "二氧化碳结合率", "离子间隙", "糖化白蛋白", "视黄醇结合蛋白", "eGFR(CKD-EPI)", "a-l岩藻糖苷酶", "超氧化物歧化酶", "eGFR(MDRD)", r"eGFR单位ml/min/1.73m^2", "8597", "乳酸测定", "乳酸脱氢酶", "羟丁酸脱氢酶", "肌酸激酶", "肌酸激酶同工酶", "载脂蛋白A", "载脂蛋白B", "载脂蛋白E", "脂蛋白（a）", "缺血修饰白蛋白", "甘油三酯", "高密度脂蛋白", "低密度脂蛋白", ], \
+            {'test_rptunitid':[str(5), str(20), str(108)], 'test_check_name':'肝功', 'test_check_list':["谷草转氨酶", "谷丙转氨酶", "谷草/谷丙", "碱性磷酸酶", "γ谷氨酰氨转肽酶", "总胆红素", "直接胆红素", "间接胆红素", "胆碱酯酶", "总胆固醇", "总蛋白", "白蛋白", "球蛋白", "白球比", "总胆汁酸", "a-L-岩藻糖苷酶", "前白蛋白", "超氧化物歧化酶", "尿素", "肌酐", "胱抑素C", "葡萄糖", "尿酸", "钾", "钠", "氯", "磷", "钙", "镁", "二氧化碳结合率", "离子间隙", "糖化白蛋白", "视黄醇结合蛋白", "eGFR(CKD-EPI)", "a-l岩藻糖苷酶", "超氧化物歧化酶", "eGFR(MDRD)", r"eGFR单位ml/min/1.73m^2", "8597", "乳酸测定", "乳酸脱氢酶", "羟丁酸脱氢酶", "肌酸激酶", "肌酸激酶同工酶", "载脂蛋白A", "载脂蛋白B", "载脂蛋白E", "脂蛋白（a）", "缺血修饰白蛋白", "甘油三酯", "高密度脂蛋白", "低密度脂蛋白", ], \
-             'test_check_list_all':["谷草转氨酶", "谷丙转氨酶", "谷草/谷丙", "碱性磷酸酶", ["γ谷氨酰氨转肽酶", "γ-谷氨酰转肽酶"], "总胆红素", "直接胆红素", "间接胆红素", "胆碱脂酶", "总胆固醇", "总蛋白", "白蛋白", "球蛋白", "白球比", "总胆汁酸", "a-L-岩藻糖苷酶", "前白蛋白", "超氧化物歧化酶", "尿素", "肌酐", "胱抑素C", "葡萄糖", "尿酸", "钾", "钠", "氯", "磷", "钙", "镁", "二氧化碳结合率", "离子间隙", "糖化白蛋白", "视黄醇结合蛋白", r"eGFR\(CKD-EPI\)", "a-l岩藻糖苷酶", "超氧化物歧化酶", r"eGFR\(MDRD\)", r"eGFR单位ml/min/1.73m\^2", "8597", "乳酸测定", "乳酸脱氢酶", "羟丁酸脱氢酶", "肌酸激酶", "肌酸激酶同工酶", "载脂蛋白A", "载脂蛋白B", "载脂蛋白E", "脂蛋白（a）", "缺血修饰白蛋白", "甘油三酯", "高密度脂蛋白", "低密度脂蛋白", ], 'test_result_col_name':"result_str"},
+             'test_check_list_all':["谷草转氨酶", "谷丙转氨酶", "谷草/谷丙", "碱性磷酸酶", ["γ谷氨酰氨转肽酶", "γ-谷氨酰转肽酶"], "总胆红素", "直接胆红素", "间接胆红素", "胆碱酯酶", "总胆固醇", "总蛋白", "白蛋白", "球蛋白", "白球比", "总胆汁酸", "a-L-岩藻糖苷酶", "前白蛋白", "超氧化物歧化酶", "尿素", "肌酐", "胱抑素C", "葡萄糖", "尿酸", "钾", "钠", "氯", "磷", "钙", "镁", "二氧化碳结合率", "离子间隙", "糖化白蛋白", "视黄醇结合蛋白", r"eGFR\(CKD-EPI\)", "a-l岩藻糖苷酶", "超氧化物歧化酶", r"eGFR\(MDRD\)", r"eGFR单位ml/min/1.73m\^2", "8597", "乳酸测定", "乳酸脱氢酶", "羟丁酸脱氢酶", "肌酸激酶", "肌酸激酶同工酶", "载脂蛋白A", "载脂蛋白B", "载脂蛋白E", "脂蛋白（a）", "缺血修饰白蛋白", "甘油三酯", "高密度脂蛋白", "低密度脂蛋白", ], 'test_result_col_name':"result_str"},
             # 各类肿瘤标志物 # (6):各类肿瘤标志物，e.g.肿瘤标志物肺癌六项[复]、肿瘤标志物十项(男)；东院区：（108）：肿瘤标志物肺癌六项[复]
             {'test_rptunitid':[str(6), str(108)], 'test_check_name':'各类肿瘤标志物', 'test_check_list':["胃泌素释放肽前体(罗氏)", "鳞状上皮细胞癌抗原(罗氏)", "癌胚抗原", "甲胎蛋白", "糖类抗原125", "糖类抗原199", "糖类抗原724", "细胞角蛋白19片段", "神经元特异性烯醇化酶测定", "总前列腺特异性抗原", "游离前列腺抗原", "游离/总", "糖类抗原153","人附睾蛋白","绝经前 ROMA","绝经后 ROMA","铁蛋白","叶酸","维生素B12","高尔基体蛋白73测定","胃蛋白酶原Ⅰ","胃蛋白酶原Ⅱ","胃蛋白酶原Ⅰ/Ⅱ","降钙素","甲状腺球蛋白"],\
              'test_check_list_all':[r"胃泌素释放肽前体\(罗氏\)", ["鳞状上皮细胞癌相关抗原", r"鳞状上皮细胞癌抗原\(罗氏\)"], "癌胚抗原", "甲胎蛋白", "糖类抗原125", "糖类抗原199", "糖类抗原724", "细胞角蛋白19片段", "神经元特异性烯醇化酶测定", "总前列腺特异性抗原", "游离前列腺抗原", "游离/总", "糖类抗原153","人附睾蛋白","绝经前 ROMA","绝经后 ROMA","铁蛋白","叶酸","维生素B12","高尔基体蛋白73测定","胃蛋白酶原Ⅰ","胃蛋白酶原Ⅱ","胃蛋白酶原Ⅰ/Ⅱ","降钙素","甲状腺球蛋白"], 'test_result_col_name':"result_str"},

app/processors/dynamic_router.py

@@ -0,0 +1,137 @@
+import re
+
+
+UNMATCHED_HEADER = "未匹配检测内容"
+UNASSIGNED_SHEET_NAME = "未归属检测内容"
+
+CATEGORY_REASON_KEYWORDS = {
+    "血细胞": ["血细胞", "血常规"],
+    "凝血": ["凝血"],
+    "肝功": ["肝功", "肾功", "电解质", "葡萄糖", "心肌酶"],
+    "各类肿瘤标志物": ["肿瘤", "标志物", "癌胚", "甲状旁腺", "降钙素", "鳞状细胞"],
+    "七抗": ["七抗", "自身抗体"],
+    "传染指标": ["传染"],
+    "血气分析+生化分析": ["血气"],
+    "感染指标": ["感染", "新冠", "冠状病毒", "结核", "细菌", "病毒", "HPV", "C反应蛋白", "降钙素原"],
+    "基因检测指标": ["基因", "CYP"],
+    "心衰系列": ["心衰", "B型", "BNP", "Pro-BNP", "钠尿肽", "肌钙蛋白"],
+    "普通指标": ["血型", "隐血", "卡式"],
+    "免疫系列": ["甲功", "甲状腺", "促甲状腺", "抗甲状腺"],
+    "特殊指标": ["细胞因子", "白介素", "血管内皮"],
+    "内分泌代谢系列": ["内分泌", "代谢", "儿茶酚胺", "ANCA"],
+    "用药指导": ["用药", "VKORC", "CYP2C9", "ALDH2", "ApoE", "SLCO"],
+}
+
+
+def match_re(value, pattern):
+    return re.match(str(pattern), str(value or "")) is not None
+
+
+def clean_result(value):
+    if value in (None, "", "."):
+        return "None"
+    return str(value)
+
+
+def detail_value(row):
+    for key in ("result_str", "result_ref", "result_txt", "result1"):
+        value = row.get(key, "")
+        if value not in (None, ""):
+            return clean_result(value)
+    return ""
+
+
+def route_detail_rows(detail_rows, all_tests, reason=""):
+    sheet_results = {}
+    unassigned_items = []
+    tests_by_name = {test["test_check_name"]: test for test in all_tests}
+    candidates_by_row = []
+    candidate_counts = {}
+
+    for detail_row in detail_rows:
+        candidates = _match_candidates(detail_row, all_tests)
+        candidates_by_row.append((detail_row, candidates))
+        for candidate in candidates:
+            candidate_counts[candidate["sheet_name"]] = candidate_counts.get(candidate["sheet_name"], 0) + 1
+
+    for detail_row, candidates in candidates_by_row:
+        item_name = detail_row.get("rpt_itemname", "")
+        if not candidates:
+            unassigned_items.append(f"{item_name}：{detail_value(detail_row)}")
+            continue
+
+        candidate = _choose_candidate(candidates, reason, candidate_counts)
+        test = tests_by_name[candidate["sheet_name"]]
+        sheet_results.setdefault(candidate["sheet_name"], {})[candidate["result_name"]] = clean_result(
+            detail_row.get(test["test_result_col_name"], "")
+        )
+
+    return sheet_results, unassigned_items
+
+
+def _match_candidates(detail_row, all_tests):
+    item_name = detail_row.get("rpt_itemname", "")
+    candidates = []
+
+    for test_index, test in enumerate(all_tests):
+        test_check_list = test["test_check_list"]
+        test_check_list_all = test["test_check_list_all"]
+
+        for item_index, checks in enumerate(test_check_list_all):
+            if isinstance(checks, str):
+                checks = [checks]
+            if any(match_re(item_name, pattern) for pattern in checks):
+                candidates.append(
+                    {
+                        "sheet_name": test["test_check_name"],
+                        "result_name": test_check_list[item_index],
+                        "test_index": test_index,
+                        "item_index": item_index,
+                    }
+                )
+                break
+
+    return candidates
+
+
+def _choose_candidate(candidates, reason, candidate_counts):
+    if len(candidates) == 1:
+        return candidates[0]
+
+    reason = str(reason or "")
+
+    def score(candidate):
+        sheet_name = candidate["sheet_name"]
+        keywords = CATEGORY_REASON_KEYWORDS.get(sheet_name, [])
+        reason_score = 100 if any(keyword and keyword in reason for keyword in keywords) else 0
+        density_score = candidate_counts.get(sheet_name, 0)
+        return (reason_score, density_score, -candidate["test_index"], -candidate["item_index"])
+
+    return max(candidates, key=score)
+
+
+def append_routed_report(
+    add_content_to_excel,
+    result_save_path,
+    all_tests,
+    excel_head,
+    excel_basic,
+    sheet_results,
+    unassigned_items,
+    show_not_match,
+):
+    for test in all_tests:
+        sheet_name = test["test_check_name"]
+        if sheet_name not in sheet_results:
+            continue
+
+        result_values = sheet_results[sheet_name]
+        row = excel_head + excel_basic + [
+            result_values.get(test_item, "Not_Find") for test_item in test["test_check_list"]
+        ]
+        if show_not_match:
+            row += unassigned_items
+        add_content_to_excel(result_save_path, sheet_name, row)
+
+    if unassigned_items and not sheet_results:
+        add_content_to_excel(result_save_path, UNASSIGNED_SHEET_NAME, excel_head + excel_basic + unassigned_items)

docker_compose-Nas.yaml

@@ -0,0 +1,34 @@
+services:
+  his_sur_data_deal:
+    build:
+      context: /share/Container/his_sur_data_deal
+      dockerfile: Dockerfile
+      args:
+        HTTP_PROXY: http://192.168.31.7:7893
+        HTTPS_PROXY: http://192.168.31.7:7893
+        NO_PROXY: localhost,127.0.0.1,192.168.31.0/24,his_sur_data_deal
+    pull_policy: never
+    image: his-sur-data-deal:latest
+    container_name: his_sur_data_deal
+    restart: unless-stopped
+    ports:
+      - "4004:8000"
+    volumes:
+      - /share/Container/his_sur_data_deal/data/jobs:/tmp/his_sur_data_deal_jobs
+    environment:
+      TZ: Asia/Shanghai
+      HTTP_PROXY: http://192.168.31.7:7893
+      HTTPS_PROXY: http://192.168.31.7:7893
+      http_proxy: http://192.168.31.7:7893
+      https_proxy: http://192.168.31.7:7893
+      NO_PROXY: localhost,127.0.0.1,192.168.31.0/24,his_sur_data_deal
+    healthcheck:
+      test:
+        [
+          "CMD-SHELL",
+          "python -c \"import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/health', timeout=5).read()\"",
+        ]
+      interval: 10s
+      timeout: 5s
+      retries: 6
+      start_period: 20s

docker_compose.yaml

@@ -0,0 +1,28 @@
+services:
+  his_sur_data_deal:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    pull_policy: never
+    image: his-sur-data-deal:latest
+    container_name: his_sur_data_deal
+    restart: unless-stopped
+    ports:
+      - "8000:8000"
+    volumes:
+      - his_sur_data_deal_jobs:/tmp/his_sur_data_deal_jobs
+    environment:
+      TZ: Asia/Shanghai
+    healthcheck:
+      test:
+        [
+          "CMD-SHELL",
+          "python -c \"import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/health', timeout=5).read()\"",
+        ]
+      interval: 10s
+      timeout: 5s
+      retries: 6
+      start_period: 20s
+
+volumes:
+  his_sur_data_deal_jobs:

tests/verify_dynamic_routing.py

@@ -0,0 +1,209 @@
+#!/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()