From eb4233f23f8f1cdbba1dd2411c77e97284434594 Mon Sep 17 00:00:00 2001
From: zlwu <704541015@qq.com>
Date: Fri, 21 Aug 2026 16:55:47 +0800
Subject: [PATCH] feat(admin): add users list Excel export endpoint

Reuse list filters, tenant and data scope; support ids and visible columns.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 backend/app/api/admin.py | 102 +++++++++++++++++++++++++++++++++++++++
 1 file changed, 102 insertions(+)

diff --git a/backend/app/api/admin.py b/backend/app/api/admin.py
index 3496abe8e..cf7a53692 100644
--- a/backend/app/api/admin.py
+++ b/backend/app/api/admin.py
@@ -1,6 +1,7 @@
 """运营端管理接口（租户、角色下拉等）"""
 import os
 import uuid
+from datetime import datetime
 from typing import Any, List, Optional
 
 from fastapi import APIRouter, Depends, HTTPException, Query, UploadFile, File, Form
@@ -35,6 +36,12 @@ from app.schemas.menu_favorite import (
     MenuFavoriteReorderBody,
 )
 from app.services.web_auth_roles import _map_role_name_to_web
+from app.utils.excel_export import (
+    ExportColumn,
+    ExportValidationError,
+    export_rows_to_xlsx_response,
+    filter_export_columns,
+)
 
 router = APIRouter(prefix="/web/admin", tags=["运营端管理"])
 
@@ -277,6 +284,26 @@ class UserUpdateBody(BaseModel):
     post_ids: Optional[List[str]] = Field(None, alias="postIds")
 
 
+USER_EXPORT_COLUMNS = [
+    ExportColumn(title="用户账号", field="userNo"),
+    ExportColumn(title="用户名", field="nickName"),
+    ExportColumn(title="所属组织", field="deptName"),
+    ExportColumn(title="角色", field="roles"),
+    ExportColumn(title="手机号", field="phoneNo"),
+    ExportColumn(title="状态", field="status", dict_type="sys_status"),
+    ExportColumn(title="创建时间", field="createTime"),
+]
+
+
+class UserExportBody(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    keyword: Optional[str] = None
+    dept_id: Optional[str] = Field(None, alias="deptId")
+    status: Optional[str] = None
+    ids: Optional[List[str]] = None
+    columns: Optional[List[str]] = None
+
+
 @router.get("/users")
 def list_users(
     page: int = Query(1, ge=1),
@@ -335,6 +362,81 @@ def list_users(
     return success_response({"list": result_list, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
 
 
+@router.post("/users/export")
+def export_users(
+    body: UserExportBody,
+    user: dict = Depends(require_menu_code("system:user:list")),
+    db: Session = Depends(get_db),
+):
+    """导出用户列表（当前筛选全量，或 ids 已选）。"""
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        query = db.query(SysUser).filter(
+            SysUser.del_flag == 0,
+            SysUser.tenant_no == tenant_no,
+        )
+        kw = (body.keyword or "").strip()
+        if kw:
+            like = f"%{kw}%"
+            query = query.filter(
+                or_(
+                    SysUser.user_no.like(like),
+                    SysUser.nick_name.like(like),
+                    SysUser.phone_no.like(like),
+                )
+            )
+        if body.dept_id:
+            query = query.filter(SysUser.dept_id == body.dept_id)
+        if body.status:
+            query = query.filter(SysUser.status == body.status)
+        if body.ids:
+            query = query.filter(SysUser.id.in_(body.ids))
+        query = apply_data_scope(query, SysUser, user, db, tenant_no)
+        rows_orm = query.order_by(SysUser.create_time.desc()).all()
+        dept_ids = list({r.dept_id for r in rows_orm if r.dept_id})
+        dept_map: dict[str, str] = {}
+        if dept_ids:
+            for d in db.query(SysDept).filter(SysDept.id.in_(dept_ids), SysDept.del_flag == 0).all():
+                dept_map[d.id] = d.dept_name or ""
+        result_list = [
+            user_svc._row_to_dict(r, dept_name=dept_map.get(r.dept_id or "", "")) for r in rows_orm
+        ]
+        user_ids = [d["id"] for d in result_list if d.get("id")]
+        role_map: dict[str, list[str]] = {uid: [] for uid in user_ids}
+        role_name_by_id: dict[str, str] = {}
+        if user_ids:
+            for ur in (
+                db.query(SysUserRole)
+                .filter(
+                    SysUserRole.user_id.in_(user_ids),
+                    SysUserRole.del_flag == 0,
+                )
+                .all()
+            ):
+                role_map.setdefault(ur.user_id, []).append(ur.role_id)
+            role_ids = list({rid for ids in role_map.values() for rid in ids})
+            if role_ids:
+                for role in db.query(SysRole).filter(SysRole.id.in_(role_ids), SysRole.del_flag == 0).all():
+                    role_name_by_id[role.id] = role.role_name or role.id
+        for d in result_list:
+            names = [role_name_by_id.get(rid, rid) for rid in role_map.get(d["id"], [])]
+            d["roles"] = "、".join(names)
+        cols = filter_export_columns(USER_EXPORT_COLUMNS, body.columns)
+        date_str = datetime.now().strftime("%Y%m%d")
+        return export_rows_to_xlsx_response(
+            db,
+            tenant_no=tenant_no,
+            sheet_title="用户列表",
+            columns=cols,
+            rows=result_list,
+            filename_stem=f"用户列表_{date_str}",
+        )
+    except ExportValidationError as e:
+        return error_response(400, str(e))
+    except Exception as e:
+        return error_response(500, str(e))
+
+
 @router.get("/users/{user_id}")
 def get_user_by_id(
     user_id: str,