From 4a194aa25e81f82584189b4820d7a28563015c48 Mon Sep 17 00:00:00 2001
From: zlwu <704541015@qq.com>
Date: Fri, 21 Aug 2026 16:52:50 +0800
Subject: [PATCH] feat(backend): add list export column filter and row-count
 guards

Shared helpers enforce empty/over-limit export rules and allow
clients to request visible columns by field key.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 backend/app/utils/excel_export.py          | 32 ++++++++++++++++
 backend/tests/test_excel_export_helpers.py | 43 ++++++++++++++++++++++
 2 files changed, 75 insertions(+)
 create mode 100644 backend/tests/test_excel_export_helpers.py

diff --git a/backend/app/utils/excel_export.py b/backend/app/utils/excel_export.py
index c5a7a0825..849b62668 100644
--- a/backend/app/utils/excel_export.py
+++ b/backend/app/utils/excel_export.py
@@ -13,6 +13,12 @@ from app.models.models import SysDictData
 
 CellFormatter = Callable[[Any], Any]
 
+EXPORT_MAX_ROWS = 50000
+
+
+class ExportValidationError(ValueError):
+    """导出业务校验失败（空数据 / 超限等），路由层转为 400。"""
+
 
 @dataclass(frozen=True)
 class ExportColumn:
@@ -121,6 +127,29 @@ def workbook_to_streaming_response(
     )
 
 
+def filter_export_columns(
+    all_columns: Sequence[ExportColumn],
+    requested: Optional[Sequence[str]],
+) -> List[ExportColumn]:
+    """按前端可见列 prop 裁剪；未知 field 忽略；requested 为空则返回全部默认列。"""
+    if not requested:
+        return list(all_columns)
+    by_field = {c.field: c for c in all_columns}
+    out: List[ExportColumn] = []
+    for key in requested:
+        col = by_field.get(str(key))
+        if col is not None:
+            out.append(col)
+    return out if out else list(all_columns)
+
+
+def assert_export_row_count(count: int) -> None:
+    if count <= 0:
+        raise ExportValidationError("没有可导出的数据")
+    if count > EXPORT_MAX_ROWS:
+        raise ExportValidationError(f"导出超过 {EXPORT_MAX_ROWS} 条，请缩小筛选范围")
+
+
 def export_rows_to_xlsx_response(
     db: Session,
     *,
@@ -132,6 +161,9 @@ def export_rows_to_xlsx_response(
     include_legacy_filename: bool = True,
 ) -> StreamingResponse:
     """CRM 列表导出统一入口：表头 + 行数据 + 字典 label 转换 + 文件流响应。"""
+    assert_export_row_count(len(rows))
+    if not columns:
+        raise ExportValidationError("没有可导出的列")
     resolver = DictLabelResolver(db, columns, tenant_no=tenant_no)
     wb = _build_workbook(sheet_title, columns, rows, resolver)
     return workbook_to_streaming_response(
diff --git a/backend/tests/test_excel_export_helpers.py b/backend/tests/test_excel_export_helpers.py
new file mode 100644
index 000000000..49efa1e00
--- /dev/null
+++ b/backend/tests/test_excel_export_helpers.py
@@ -0,0 +1,43 @@
+import pytest
+
+from app.utils.excel_export import (
+    EXPORT_MAX_ROWS,
+    ExportColumn,
+    ExportValidationError,
+    filter_export_columns,
+    assert_export_row_count,
+)
+
+
+COLS = [
+    ExportColumn(title="账号", field="userNo"),
+    ExportColumn(title="姓名", field="nickName"),
+    ExportColumn(title="状态", field="status", dict_type="sys_status"),
+]
+
+
+def test_filter_export_columns_keeps_order_and_ignores_unknown():
+    out = filter_export_columns(COLS, ["status", "unknown", "userNo"])
+    assert [c.field for c in out] == ["status", "userNo"]
+
+
+def test_filter_export_columns_none_returns_all():
+    out = filter_export_columns(COLS, None)
+    assert [c.field for c in out] == ["userNo", "nickName", "status"]
+
+
+def test_assert_export_row_count_empty():
+    with pytest.raises(ExportValidationError) as ei:
+        assert_export_row_count(0)
+    assert "没有可导出的数据" in str(ei.value)
+
+
+def test_assert_export_row_count_over_limit():
+    with pytest.raises(ExportValidationError) as ei:
+        assert_export_row_count(EXPORT_MAX_ROWS + 1)
+    assert "50000" in str(ei.value)
+
+
+def test_assert_export_row_count_ok():
+    assert_export_row_count(1)
+    assert_export_row_count(EXPORT_MAX_ROWS)