164 lines
5.2 KiB
Python
164 lines
5.2 KiB
Python
# app/services/user_service.py
|
|
|
|
from app.models.user import User, Role, db
|
|
from sqlalchemy import or_
|
|
from datetime import datetime
|
|
|
|
|
|
class UserService:
|
|
@staticmethod
|
|
def get_users(page=1, per_page=10, search_query=None, status=None, role_id=None):
|
|
"""
|
|
获取用户列表，支持分页、搜索和过滤
|
|
"""
|
|
query = User.query
|
|
|
|
# 搜索条件
|
|
if search_query:
|
|
query = query.filter(or_(
|
|
User.username.like(f'%{search_query}%'),
|
|
User.email.like(f'%{search_query}%'),
|
|
User.nickname.like(f'%{search_query}%'),
|
|
User.phone.like(f'%{search_query}%')
|
|
))
|
|
|
|
# 状态过滤
|
|
if status is not None:
|
|
query = query.filter(User.status == status)
|
|
|
|
# 角色过滤
|
|
if role_id is not None:
|
|
query = query.filter(User.role_id == role_id)
|
|
|
|
# 分页
|
|
pagination = query.order_by(User.id.desc()).paginate(
|
|
page=page, per_page=per_page, error_out=False
|
|
)
|
|
|
|
return pagination
|
|
|
|
@staticmethod
|
|
def get_user_by_id(user_id):
|
|
"""通过ID获取用户"""
|
|
return User.query.get(user_id)
|
|
|
|
@staticmethod
|
|
def update_user(user_id, data):
|
|
"""更新用户信息"""
|
|
user = User.query.get(user_id)
|
|
if not user:
|
|
return False, "用户不存在"
|
|
|
|
try:
|
|
# 更新可编辑字段
|
|
if 'email' in data:
|
|
# 检查邮箱是否已被其他用户使用
|
|
existing = User.query.filter(User.email == data['email'], User.id != user_id).first()
|
|
if existing:
|
|
return False, "邮箱已被使用"
|
|
user.email = data['email']
|
|
|
|
if 'phone' in data:
|
|
# 检查手机号是否已被其他用户使用
|
|
existing = User.query.filter(User.phone == data['phone'], User.id != user_id).first()
|
|
if existing:
|
|
return False, "手机号已被使用"
|
|
user.phone = data['phone']
|
|
|
|
if 'nickname' in data and data['nickname']:
|
|
user.nickname = data['nickname']
|
|
|
|
# 只有管理员可以修改这些字段
|
|
if 'role_id' in data:
|
|
user.role_id = data['role_id']
|
|
|
|
if 'status' in data:
|
|
user.status = data['status']
|
|
|
|
if 'password' in data and data['password']:
|
|
user.set_password(data['password'])
|
|
|
|
user.updated_at = datetime.now()
|
|
db.session.commit()
|
|
return True, "用户信息更新成功"
|
|
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
return False, f"更新失败: {str(e)}"
|
|
|
|
@staticmethod
|
|
def change_user_status(user_id, status):
|
|
"""变更用户状态 (启用/禁用)"""
|
|
user = User.query.get(user_id)
|
|
if not user:
|
|
return False, "用户不存在"
|
|
|
|
try:
|
|
user.status = status
|
|
user.updated_at = datetime.now()
|
|
db.session.commit()
|
|
status_text = "启用" if status == 1 else "禁用"
|
|
return True, f"用户已{status_text}"
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
return False, f"状态变更失败: {str(e)}"
|
|
|
|
@staticmethod
|
|
def delete_user(user_id):
|
|
"""删除用户 (软删除，将状态设为-1)"""
|
|
user = User.query.get(user_id)
|
|
if not user:
|
|
return False, "用户不存在"
|
|
|
|
try:
|
|
user.status = -1 # 软删除，设置状态为-1
|
|
user.updated_at = datetime.now()
|
|
db.session.commit()
|
|
return True, "用户已删除"
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
return False, f"删除失败: {str(e)}"
|
|
|
|
@staticmethod
|
|
def get_all_roles():
|
|
"""获取所有角色"""
|
|
return Role.query.all()
|
|
|
|
@staticmethod
|
|
def create_role(role_name, description=None):
|
|
"""创建新角色"""
|
|
existing = Role.query.filter_by(role_name=role_name).first()
|
|
if existing:
|
|
return False, "角色名已存在"
|
|
|
|
try:
|
|
role = Role(role_name=role_name, description=description)
|
|
db.session.add(role)
|
|
db.session.commit()
|
|
return True, "角色创建成功"
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
return False, f"创建失败: {str(e)}"
|
|
|
|
@staticmethod
|
|
def update_role(role_id, role_name, description=None):
|
|
"""更新角色信息"""
|
|
role = Role.query.get(role_id)
|
|
if not role:
|
|
return False, "角色不存在"
|
|
|
|
# 检查角色名是否已被使用
|
|
existing = Role.query.filter(Role.role_name == role_name, Role.id != role_id).first()
|
|
if existing:
|
|
return False, "角色名已存在"
|
|
|
|
try:
|
|
role.role_name = role_name
|
|
if description is not None:
|
|
role.description = description
|
|
db.session.commit()
|
|
return True, "角色更新成功"
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
return False, f"更新失败: {str(e)}"
|