diff --git a/README.md b/README.md
index c038208..33ac320 100644
--- a/README.md
+++ b/README.md
@@ -33,7 +33,7 @@ V2：zip 解压后包含 `Patients_info.csv`，并按患者目录分别保存检
患者编号类型可选择自动识别。自动识别会读取 `Patients_info.csv` 中的 `pat_no`，并与 `Tests_List` 文件名或患者目录名比对：若更匹配 10 位补零编号，则使用 `pat_no`；若更匹配原始编号，则使用 `zhuyuanhao`。
-系统默认输出完整结果；处理完成后可在结果页通过“内容选择”立即切换预览内容，并同步影响右侧“导出此 Excel”的文件内容。可选择是否保留：
+系统默认输出完整结果；处理完成后可在结果页通过“内容选择”立即切换预览内容，并同步影响右侧“导出此 Excel”的文件内容。结果页还可以按姓名、采样时间或检测原因选择升序/降序排序；“未检测到内容汇总”会在每个分组标题下单独排序，不会把不同检测原因的分组混在一起。可选择是否保留：
- 基本工作表
- 未匹配检测内容项
diff --git a/app/main.py b/app/main.py
index b5c66d0..8c0cc3c 100644
--- a/app/main.py
+++ b/app/main.py
@@ -23,6 +23,15 @@ 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",
+}
app = FastAPI(title=APP_TITLE)
@@ -117,16 +126,22 @@ def result_page(
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 +151,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 +162,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 +172,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 +187,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 +198,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
@@ -207,12 +232,26 @@ 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,6 +269,16 @@ def _render_result(
+
+
+
diff --git a/app/processor.py b/app/processor.py
index 28661a9..b70a138 100644
--- a/app/processor.py
+++ b/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
@@ -17,6 +18,8 @@ PROCESSOR_DIR = Path(__file__).resolve().parent / "processors"
SUMMARY_SHEET_NAME = "\u672a\u68c0\u6d4b\u5230\u5185\u5bb9\u6c47\u603b"
UNMATCHED_HEADER = "\u672a\u5339\u914d\u68c0\u6d4b\u5185\u5bb9"
UNMATCHED_FILL = PatternFill(fill_type="solid", fgColor="FCE4D6")
+SORT_FIELDS = {"none", "name", "sample_time", "reason"}
+SORT_ORDERS = {"asc", "desc"}
class ProcessingError(Exception):
@@ -59,6 +62,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 +147,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 +167,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 +185,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 +200,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 +217,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 +241,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 +258,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 +277,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 +299,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,6 +456,8 @@ 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:
@@ -455,6 +479,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()
@@ -631,6 +656,134 @@ def _remove_unmatched_columns(workbook) -> None:
sheet.delete_cols(unmatched_col + 1, sheet.max_column - unmatched_col)
+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: