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>
This commit is contained in:
zlwuandCursor committed 2026-08-21 16:55:47 +08:00
1 parent 4a194aa25e
commit eb4233f23f
1 file changed
+102
+102
View File
@@ -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,