# +====================================================================== # | DjangoAdmin 企业数字化开发平台 [ 赋能开发者，助力企业发展 ] # +---------------------------------------------------------------------- # | Copyright (c) 2021-2026 小蚂蚁云团队 版权所有 # +---------------------------------------------------------------------- # | 开源协议: Apache-2.0 # | 重要声明: 未经书面许可，不得移除或修改本框架相关版权标识。 # +---------------------------------------------------------------------- # | 官方网站: https://www.djangoadmin.cn # +---------------------------------------------------------------------- # | 研发团队: 小蚂蚁云团队 荣誉出品 # +====================================================================== # | 版权和免责声明: # | 一、知识产权声明 # | 本团队对该软件框架产品（以下简称“本产品”）拥有完整知识产权，包括但不限于商标权、专利权、 # | 著作权、商业秘密等。前述权利均受中华人民共和国及相关国际法律法规保护。未经本团队书面授权， # | 任何个人或单位不得基于本产品本身申请任何形式的知识产权。 # | 二、使用限制与禁止条款 # | 您理解并同意，禁止将本产品用于以下任何行为： # | 1. 任何违反中华人民共和国法律法规的用途； # | 2. 任何侵害他人合法权益（包括但不限于人身权、财产权、知识产权等）的行为； # | 3. 任何危害国家安全、公共利益及社会道德风尚的行为。 # | 三、责任豁免 # | 因使用本产品而产生的任何损失或法律纠纷（包括但不限于直接损失、间接损失、附带损失、衍生损失 # | 以及因违约、侵权等引发的责任），本团队均不承担任何法律责任。若因违规使用本产品导致任何第三 # | 方提起索赔或行政机关进行调查，您应独立承担全部责任。 # | 四、违约与法律责任 # | 如发现任何违反本声明的行为，本团队有权： # | 1. 无条件配合公安机关及其他有权机关的调查取证工作； # | 2. 依法追究相关方的民事责任，并要求赔偿全部损失； # | 3. 保留进一步采取刑事控告等法律措施的权利。 # | 五、许可范围说明 # | 1、被授权方应视本产品源代码为核心机密信息，负有最高标准的保密与保管义务。禁止以任何形式（包 # | 括但不限于直接或间接的分享、复制、传播、分发、转售、出租、开源、反向工程、反编译、反汇编或通 # | 过网络服务提供给任何第三方）泄露、披露或处置源代码的全部或部分内容。任何未经授权的复制、使用、 # | 传播行为均构成对本团队知识产权的严重侵权，本团队有权依法追究其全部法律责任，并要求赔偿因此造 # | 成的一切经济损失及商誉损害。 # | 2、本产品仅限于公司或个人在法律允许范围内，用于内部且合规的软件产品研发。具体权利义务以《软件 # | 许可协议》附件为准。 # +====================================================================== """ 用户管理服务模块 ============================================================= 该模块实现了用户管理相关的所有业务逻辑。 包括用户的分页查询、详情查询、添加、更新、删除、状态设置、密码重置等功能。 用户是系统的基本主体，包含基本信息、部门、职级、岗位、角色等关联信息， 通过用户管理可以维护系统用户账号及其权限分配。 命名规范： - 函数参数使用驼峰命名（如 userId, pageNo） - 表单字段使用下划线命名（如 realname, username）- 与数据库字段保持一致 - 返回数据使用驼峰命名（如 createUser, updateTime）- 与前端规范保持一致 - 私有函数以下划线开头（如 _apply_filters, _build_page_data） 功能模块： 1. 分页查询：支持按姓名、性别、状态筛选，按ID排序 2. 详情查询：根据ID获取单个用户的详细信息，包含角色、地区等关联信息 3. 数据维护：用户的添加、更新、删除（支持批量） 4. 状态管理：单独更新用户的启用/禁用状态 5. 密码管理：重置用户密码为默认密码 """ import logging from django.core.paginator import Paginator, PageNotAnInteger, InvalidPage, EmptyPage from application.constants import GENDER_LIST from application.dept.models import Dept from application.level.models import Level from application.position.models import Position from application.user import forms from application.user.models import User from application.user_role import services as user_role_services from application.user_role.models import UserRole from constant.constants import PAGE_SIZE, DEFAULT_PASSWORD from utils import R, regular from utils.common import get_file_url, format_datetime, parse_request_body, save_file from utils.password import encrypt_password, generate_salt from utils.security import get_username # ============================================================= # 分页查询服务 # ============================================================= def get_user_page(request): """ 查询用户分页数据 Args: request: HttpRequest对象，包含查询参数 - pageNo: 当前页码（可选，默认为1） - pageSize: 每页记录数（可选，默认为PAGE_SIZE） - realname: 用户姓名（可选，模糊查询） - gender: 性别（可选，精确匹配） - status: 用户状态（可选，精确匹配） Returns: R: 包含分页数据的JSON响应 """ try: # 获取分页参数 page_no = int(request.GET.get('pageNo', 1)) page_size = int(request.GET.get('pageSize', PAGE_SIZE)) # 构建基础查询（只查询未删除的记录） query = User.objects.filter(is_delete=False) # 应用筛选条件 query = _apply_filters(query, request) # 按ID升序排列 query = query.order_by("id") # 执行分页查询 paginator = Paginator(query, page_size) try: user_list = paginator.page(page_no) except (PageNotAnInteger, InvalidPage, EmptyPage): # 页码无效时默认返回第一页 user_list = paginator.page(1) page_no = 1 # 构建并返回分页数据 page_data = _build_page_data(user_list, paginator, page_no, page_size) return R.ok(data=page_data) except Exception as e: logging.error(f"查询用户分页数据异常: {str(e)}") return R.failed(msg="查询失败，请稍后重试") def _apply_filters(queryset, request): """ 应用查询筛选条件 Args: queryset: 初始查询集 request: HttpRequest对象，包含筛选参数 Returns: QuerySet: 应用筛选后的查询集 """ # 用户姓名模糊筛选 realname = request.GET.get('realname') if realname: queryset = queryset.filter(realname__contains=realname) # 用户状态筛选 status = request.GET.get('status') if status and status.isdigit(): queryset = queryset.filter(status=int(status)) return queryset def _build_page_data(user_list, paginator, page_no, page_size): """ 构建分页返回数据 Args: user_list: 当前页用户列表 paginator: 分页器对象 page_no: 当前页码 page_size: 每页大小 Returns: dict: 分页数据 """ # 预加载关联数据字典，减少循环内查询 dept_dict = _get_dept_dict() level_dict = _get_level_dict() position_dict = _get_position_dict() records = [] for item in user_list: # 获取用户角色列表 role_list = user_role_services.get_user_role_list(item.id) # 城市编码列表 city_list = [item.province_code, item.city_code, item.district_code] records.append({ 'id': item.id, 'username': item.username, 'realname': item.realname, 'avatar': get_file_url(item.avatar), 'gender': item.gender, 'genderName': GENDER_LIST.get(item.gender), 'mobile': item.mobile, 'email': item.email, 'roles': role_list, 'deptId': item.dept_id, 'deptName': dept_dict.get(item.dept_id), 'levelId': item.level_id, 'levelName': level_dict.get(item.level_id), 'positionId': item.position_id, 'positionName': position_dict.get(item.position_id), 'status': item.status, 'sort': item.sort, 'city': city_list, 'address': item.address, 'createUser': item.create_user, 'createTime': format_datetime(item.create_time), 'updateUser': item.update_user, 'updateTime': format_datetime(item.update_time), }) return { 'records': records, 'total': paginator.count, 'size': page_size, 'current': page_no, 'pages': paginator.num_pages } def _get_dept_dict(): """获取部门ID到名称的映射字典（私有辅助函数）""" try: dept_list = Dept.objects.filter(is_delete=False).values('id', 'name') return {dept['id']: dept['name'] for dept in dept_list} except Exception as e: logging.error(f"获取部门字典异常: {str(e)}") return {} def _get_level_dict(): """获取职级ID到名称的映射字典（私有辅助函数）""" try: level_list = Level.objects.filter(is_delete=False).values('id', 'name') return {level['id']: level['name'] for level in level_list} except Exception as e: logging.error(f"获取职级字典异常: {str(e)}") return {} def _get_position_dict(): """获取岗位ID到名称的映射字典（私有辅助函数）""" try: position_list = Position.objects.filter(is_delete=False).values('id', 'name') return {position['id']: position['name'] for position in position_list} except Exception as e: logging.error(f"获取岗位字典异常: {str(e)}") return {} # ============================================================= # 详情查询服务 # ============================================================= def get_user_detail(user_id): """ 根据ID查询用户详情 Args: user_id: 用户ID Returns: dict/None: 用户详情字典，不存在时返回None """ try: # 查询未删除的用户记录 user = User.objects.filter(is_delete=False, id=user_id).first() if not user: return None # 获取用户角色数据 role_list = UserRole.objects.filter(user_id=user.id).values_list('role_id', flat=True) roles = list(role_list) # 城市编码列表 city_list = [user.province_code, user.city_code, user.district_code, user.street_code] # 返回详情数据 return { 'id': user.id, 'username': user.username, 'realname': user.realname, 'gender': user.gender, 'avatar': get_file_url(user.avatar) if user.avatar else "", 'mobile': user.mobile, 'email': user.email, 'deptId': user.dept_id, 'levelId': user.level_id, 'positionId': user.position_id, 'provinceCode': user.province_code, 'cityCode': user.city_code, 'districtCode': user.district_code, 'streetCode': user.street_code, 'address': user.address, 'intro': user.intro, 'status': user.status, 'note': user.note, 'sort': user.sort, 'roles': roles, 'city': city_list, 'createUser': user.create_user, 'createTime': format_datetime(user.create_time), 'updateUser': user.update_user, 'updateTime': format_datetime(user.update_time), } except Exception as e: logging.error(f"查询用户详情异常，ID: {user_id}, 错误: {str(e)}") return None # ============================================================= # 添加服务 # ============================================================= def add_user(request): """ 添加用户 处理流程： 1. 解析请求体中的JSON数据 2. 表单验证 3. 检查用户名和手机号唯一性 4. 处理头像图片 5. 处理行政区划 6. 创建数据库记录 7. 创建用户角色关系 Args: request: HttpRequest对象，包含用户信息的JSON数据 Returns: R: 操作结果（成功/失败） """ # 解析请求参数 data, error = parse_request_body(request) if error: return R.failed(msg=error) # 表单验证 form = forms.UserForm(data) if not form.is_valid(): return R.failed(msg=regular.get_err(form)) try: cleaned_data = form.cleaned_data # 检查用户名是否已存在 username = cleaned_data.get('username') if User.objects.filter(username=username, is_delete=False).exists(): return R.failed(msg=f"用户名 '{username}' 已存在") # 检查手机号是否已存在 mobile = cleaned_data.get('mobile') if mobile and User.objects.filter(mobile=mobile, is_delete=False).exists(): return R.failed(msg=f"手机号 '{mobile}' 已存在") # 处理头像 avatar = cleaned_data.get('avatar') if avatar: avatar = save_file(avatar, "user") # 处理行政区划 city = cleaned_data.get('city', []) if len(city) != 4: return R.failed("请选择完整的所在城市（省/市/区/街道）") # 处理密码 password = cleaned_data.get('password') if not password: password = DEFAULT_PASSWORD # 生成密码盐值和加密密码 salt = generate_salt() # 创建用户记录 user = User.objects.create( realname=cleaned_data.get('realname'), gender=cleaned_data.get('gender'), avatar=avatar, mobile=mobile, email=cleaned_data.get('email'), dept_id=cleaned_data.get('deptId'), level_id=cleaned_data.get('levelId'), position_id=cleaned_data.get('positionId'), province_code=city[0], city_code=city[1], district_code=city[2], street_code=city[3], address=cleaned_data.get('address'), username=username, password=encrypt_password(password, salt), salt=salt, intro=cleaned_data.get('intro'), status=cleaned_data.get('status', 1), note=cleaned_data.get('note'), sort=cleaned_data.get('sort', 0), create_user=get_username(request), update_user=get_username(request) ) # 创建用户角色数据 roles = cleaned_data.get('roles', []) _add_user_roles(user.id, roles, get_username(request)) return R.ok(msg="创建成功") except Exception as e: logging.error(f"添加用户异常: {str(e)}") return R.failed(msg="添加失败，请稍后重试") # ============================================================= # 更新服务 # ============================================================= def update_user(request): """ 更新用户 处理流程： 1. 解析请求体中的JSON数据 2. 验证用户ID 3. 表单验证 4. 查询现有记录 5. 检查用户名和手机号唯一性（排除自身） 6. 处理头像 7. 处理行政区划 8. 处理密码（如果提供则更新） 9. 更新各字段 10. 保存更新 11. 更新用户角色关系 Args: request: HttpRequest对象，包含用户ID和更新信息的JSON数据 Returns: R: 操作结果（成功/失败） """ # 解析请求参数 data, error = parse_request_body(request) if error: return R.failed(msg=error) # 验证用户ID user_id = data.get('id') if not user_id or not str(user_id).isdigit() or int(user_id) <= 0: return R.failed("用户ID无效") user_id = int(user_id) # 表单验证 form = forms.UserForm(data) if not form.is_valid(): return R.failed(msg=regular.get_err(form)) try: cleaned_data = form.cleaned_data # 查询用户记录 user = User.objects.filter(id=user_id, is_delete=False).first() if not user: return R.failed("用户不存在") # 检查用户名是否与其他用户冲突（排除自身） username = cleaned_data.get('username') if username != user.username: if User.objects.filter(username=username, is_delete=False).exclude(id=user_id).exists(): return R.failed(msg=f"用户名 '{username}' 已存在") # 检查手机号是否与其他用户冲突（排除自身） mobile = cleaned_data.get('mobile') if mobile and mobile != user.mobile: if User.objects.filter(mobile=mobile, is_delete=False).exclude(id=user_id).exists(): return R.failed(msg=f"手机号 '{mobile}' 已存在") # 处理头像 avatar = cleaned_data.get('avatar') if avatar and avatar != user.avatar: avatar = save_file(avatar, "user") else: avatar = user.avatar # 处理行政区划 city = cleaned_data.get('city', []) if len(city) != 4: return R.failed("请选择完整的所在城市（省/市/区/街道）") # 更新用户信息 user.realname = cleaned_data.get('realname') user.gender = cleaned_data.get('gender') user.avatar = avatar user.mobile = mobile user.email = cleaned_data.get('email') user.dept_id = cleaned_data.get('deptId') user.level_id = cleaned_data.get('levelId') user.position_id = cleaned_data.get('positionId') user.province_code = city[0] user.city_code = city[1] user.district_code = city[2] user.street_code = city[3] user.address = cleaned_data.get('address') user.username = username # 处理密码（如果提供了新密码） password = cleaned_data.get('password') if password: salt = generate_salt() user.password = encrypt_password(password, salt) user.salt = salt user.intro = cleaned_data.get('intro') user.status = cleaned_data.get('status') user.note = cleaned_data.get('note') user.sort = cleaned_data.get('sort') user.update_user = get_username(request) user.save() # 更新用户角色数据 roles = cleaned_data.get('roles', []) _add_user_roles(user_id, roles, get_username(request)) return R.ok(msg="更新成功") except Exception as e: logging.error(f"更新用户异常，ID: {user_id}, 错误: {str(e)}") return R.failed(msg="更新失败，请稍后重试") def _add_user_roles(user_id, role_ids, uname): """ 批量添加用户角色关系（私有辅助函数） Args: user_id: 用户ID role_ids: 角色ID列表 uname: 操作人用户名 """ try: # 删除原有用户角色关系 UserRole.objects.filter(user_id=user_id).delete() # 创建新的用户角色关系 if role_ids: user_roles = [] for role_id in role_ids: if role_id: user_roles.append(UserRole( user_id=user_id, role_id=role_id, create_user=uname, update_user=uname )) if user_roles: UserRole.objects.bulk_create(user_roles) except Exception as e: logging.error(f"添加用户角色关系异常，user_id: {user_id}, roles: {role_ids}, 错误: {str(e)}") raise # ============================================================= # 删除服务（支持批量） # ============================================================= def delete_users(user_ids): """ 删除用户（批量删除） 采用逻辑删除方式（is_delete=True），而非物理删除。 支持逗号分隔的多个ID。 Args: user_ids: 用户ID字符串，多个ID用逗号分隔 例如: "1" 或 "1,2,3" Returns: R: 操作结果，包含删除数量 """ if not user_ids: return R.failed("记录ID不存在") try: # 解析ID列表 id_list = [int(id_str.strip()) for id_str in user_ids.split(',') if id_str.strip()] if not id_list: return R.failed("无效的记录ID") # 批量更新删除标识（逻辑删除） updated_count = User.objects.filter( id__in=id_list, is_delete=False ).update(is_delete=True) return R.ok(msg=f"本次共删除{updated_count}条数据") except ValueError: return R.failed("记录ID格式错误") except Exception as e: logging.error(f"删除用户异常，IDs: {user_ids}, 错误: {str(e)}") return R.failed(msg="删除失败，请稍后重试") # ============================================================= # 状态更新服务 # ============================================================= def UserStatus(request): """ 设置用户状态（启用/禁用） 单独更新用户的启用/禁用状态，不涉及其他字段。 Args: request: HttpRequest对象 - id: 用户ID - status: 目标状态值（1-启用, 2-禁用） Returns: R: 操作结果（成功/失败） """ # 解析请求参数 data, error = parse_request_body(request) if error: return R.failed(msg=error) # 表单验证 form = forms.UserStatusForm(data=data) if not form.is_valid(): return R.failed(msg=regular.get_err(form)) try: user_id = int(form.cleaned_data.get('id')) status = int(form.cleaned_data.get("status")) # 查询用户记录 user = User.objects.filter(id=user_id, is_delete=False).first() if not user: return R.failed("记录不存在") # 更新状态 user.status = status user.update_user = get_username(request) user.save() msg = "启用成功" if status == 1 else "禁用成功" return R.ok(msg=msg) except Exception as e: logging.error(f"设置用户状态异常，ID: {user_id}, 错误: {str(e)}") return R.failed(msg="状态更新失败，请稍后重试") # ============================================================= # 密码重置服务 # ============================================================= def reset_user_password(request): """ 重置用户密码为默认密码 将用户密码重置为系统配置的默认密码。 Args: request: HttpRequest对象，包含用户ID Returns: R: 操作结果（成功/失败） """ # 解析请求参数 data, error = parse_request_body(request) if error: return R.failed(msg=error) user_id = data.get('userId') if not user_id or not str(user_id).isdigit() or int(user_id) <= 0: return R.failed("用户ID无效") user_id = int(user_id) try: # 查询用户记录 user = User.objects.filter(is_delete=False, id=user_id).first() if not user: return R.failed("用户不存在") # 重置密码为默认密码 salt = generate_salt() user.password = encrypt_password(DEFAULT_PASSWORD, salt) user.salt = salt user.update_user = get_username(request) user.save() return R.ok(msg="重置密码成功") except Exception as e: logging.error(f"重置用户密码异常，ID: {user_id}, 错误: {str(e)}") return R.failed(msg="重置密码失败，请稍后重试")