import secrets from typing import List from django.contrib.auth.hashers import make_password from django.db import IntegrityError from django.db.models import Q from django.shortcuts import get_object_or_404 from loguru import logger from ninja import Router from ninja.errors import HttpError from ninja.pagination import paginate, PageNumberPagination from users.models import User, Department from users.permission import require_permissions from users.schema import UserUpdate, UserOut, UserCreate, UserResetPassword from utils.dept_utils import get_dept_children_ids router = Router(tags=['users']) @router.get("/", response=List[UserOut], summary="获取用户列表") @paginate(PageNumberPagination, page_size=20) @require_permissions("user:list") def list_users(request, dept_id: int = None, q: str = None): users = User.objects.all() logger.debug(f"users: {len(users)}") logger.debug(f"dept_id: {dept_id}, q: {q}") if dept_id: current_dept = get_object_or_404(Department, id=dept_id) dept_ids = get_dept_children_ids(dept_id) if current_dept.is_root: # 包含未分配用户 users = users.filter(Q(dept_id__in=dept_ids) | Q(dept__isnull=True)) else: users = users.filter(dept_id__in=dept_ids) logger.debug(f"users: {len(users)}") if q: users = users.filter( Q(username__icontains=q) | Q(cname__icontains=q) | Q(phone__icontains=q) ) logger.debug(f"users: {len(users)}") return users.order_by("dept_id") @router.post("/") @require_permissions("user:create") def create_user(request, data: UserCreate): try: if not data.password: data.password = secrets.token_urlsafe(8)[:8] logger.debug(f"用户名:{data.username}, 密码:{data.password}") user = User.objects.create_user( username=data.username, email=data.email, password=make_password(data.password) ) for k, v in data.dict(exclude={"password", "roles"}).items(): setattr(user, k, v) user.save() if data.roles: user.roles.set(data.roles) return { "detail": f"用户创建成功!密码:{data.password}" } except IntegrityError as e: raise HttpError(500, "用户已存在!") @router.post("/reset-password/", summary="重置用户密码") @require_permissions("user:reset_password") def reset_password(request, payload: UserResetPassword): user = get_object_or_404(User, id=payload.user_id) if payload.new_password: new_pass = payload.new_password else: new_pass = secrets.token_urlsafe(8)[:12] # 自动生成 user.password = make_password(new_pass) user.save() # 返回新密码(仅用于通知,前端可显示) return {"detail": f"密码已重置: {new_pass}"} @router.get("/{id}", response=UserOut) @require_permissions("user:view") def get_user(request, id: int): obj = get_object_or_404(User, id=id) data = obj.__dict__ data["roles"] = list(obj.roles.values_list("id", flat=True)) return data @router.put("/{id}") @require_permissions("user:update") def update_user(request, id: int, data: UserUpdate): user = get_object_or_404(User, id=id) if data.password: user.set_password(data.password) for k, v in data.dict(exclude_unset=True, exclude={"password", "roles"}).items(): if k == "dept": k = "dept_id" setattr(user, k, v) user.save() if data.roles is not None: user.roles.set(data.roles) if data.password: return {"detail": f"密码已重置: {data.password}"} return {"success": True} @router.delete("/{id}") @require_permissions("user:delete") def delete_user(request, id: int): User.objects.filter(id=id).delete() return {"success": True}