djangoadmin/DjangoAdmin_Flask_EleVue
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: P4c 最终批次-6 user 模块（最后一个）——P4 全部完成

Browse Source 
【改造】get_user_page/add/update/status/delete 按新范式；
update_user_info/update_password/reset_user_password 入参 data + BizError；
get_info 委托 UserRepository
【验收】flake8 零告警；应用启动；user 分页/详情运行时正常；12 passed

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
DjangoAdmin Dev
2026-08-07 17:32:54 +08:00
co-authored by Claude
parent b9ce9dc510
commit 2bf2179611
4 changed files with 290 additions and 338 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/modules/system/user/models.py
-1
View File
@@ -50,7 +50,6 @@ from src.common.base_db import BaseDB
from src.common.base_model import BaseModel from src.common.base_model import BaseModel
from config import DB_PREFIX from config import DB_PREFIX
from src.core.database import db from src.core.database import db
from src.common.utils.case_converter import convert_camel_to_snake
# ============================================================ # ============================================================
src/modules/system/user/repository.py
+15
View File
@@ -0,0 +1,15 @@
"""
User 仓储模块
=============================================================
数据访问层，继承 BaseRepository 获得通用 CRUD 能力。
"""
from src.common.base_repository import BaseRepository
from src.modules.system.user.models import User
class UserRepository(BaseRepository):
"""User 仓储"""
model = User
src/modules/system/user/router.py
+41 -22
View File
@@ -56,21 +56,24 @@
application.middleware.check_permission - 权限校验中间件 application.middleware.check_permission - 权限校验中间件
""" """
from flask import Blueprint from flask import Blueprint, request
from application.middleware.check_demo import check_demo from application.middleware.check_demo import check_demo
from application.middleware.check_permission import check_permission from application.middleware.check_permission import check_permission
from src.modules.system.user import service as user from src.modules.system.user import service as user
from src.common.utils.request import parse_batch_ids from src.common.utils.request import parse_batch_ids, parse_request_body
user_blue = Blueprint('user', __name__, url_prefix='/user')
from src.core import response as R from src.core import response as R
user_blue = Blueprint("user", __name__, url_prefix="/user")
# ============================================================ # ============================================================
# 用户分页查询 # 用户分页查询
# ============================================================ # ============================================================
@user_blue.route('/page', methods=["GET"])
@user_blue.route("/page", methods=["GET"])
@check_permission("sys:user:page") @check_permission("sys:user:page")
def page(): def page():
""" """
@@ -82,15 +85,15 @@ def page():
Returns: Returns:
JSON: 用户分页数据，包含 records（数据列表）、total（总记录数）等 JSON: 用户分页数据，包含 records（数据列表）、total（总记录数）等
""" """
result = user.get_user_page() return R.ok(data=user.get_user_page(request.args))
return result
# ============================================================ # ============================================================
# 用户详情查询 # 用户详情查询
# ============================================================ # ============================================================
@user_blue.route('/detail/<int:id>', methods=["GET"])
@user_blue.route("/detail/<int:id>", methods=["GET"])
@check_permission("sys:user:detail") @check_permission("sys:user:detail")
def detail(id): def detail(id):
""" """
@@ -110,7 +113,8 @@ def detail(id):
# 用户添加 # 用户添加
# ============================================================ # ============================================================
@user_blue.route('/add', methods=["POST"])
@user_blue.route("/add", methods=["POST"])
@check_permission("sys:user:add") @check_permission("sys:user:add")
@check_demo @check_demo
def add(): def add():
@@ -123,15 +127,19 @@ def add():
Returns: Returns:
JSON: 操作结果 JSON: 操作结果
""" """
result = user.add_user() json_data, err = parse_request_body()
return result if err:
return R.failed(msg=err)
user.add_user(json_data)
return R.ok(msg="添加成功")
# ============================================================ # ============================================================
# 用户更新 # 用户更新
# ============================================================ # ============================================================
@user_blue.route('/update', methods=["PUT"])
@user_blue.route("/update", methods=["PUT"])
@check_permission("sys:user:update") @check_permission("sys:user:update")
@check_demo @check_demo
def update(): def update():
@@ -144,15 +152,19 @@ def update():
Returns: Returns:
JSON: 操作结果 JSON: 操作结果
""" """
result = user.update_user() json_data, err = parse_request_body()
return result if err:
return R.failed(msg=err)
user.update_user(json_data)
return R.ok(msg="更新成功")
# ============================================================ # ============================================================
# 用户删除 # 用户删除
# ============================================================ # ============================================================
@user_blue.route('/delete/<string:id>', methods=["DELETE"])
@user_blue.route("/delete/<string:id>", methods=["DELETE"])
@check_permission("sys:user:delete") @check_permission("sys:user:delete")
@check_demo @check_demo
def delete(id): def delete(id):
@@ -165,15 +177,16 @@ def delete(id):
Returns: Returns:
JSON: 操作结果 JSON: 操作结果
""" """
result = user.delete_users(id) count = user.delete_users(id)
return result return R.ok(msg=f"本次共删除{count}条数据")
# ============================================================ # ============================================================
# 用户批量删除 # 用户批量删除
# ============================================================ # ============================================================
@user_blue.route('/batchDelete', methods=["DELETE"])
@user_blue.route("/batchDelete", methods=["DELETE"])
@check_permission("sys:user:batch_delete") @check_permission("sys:user:batch_delete")
@check_demo @check_demo
def batch_delete(): def batch_delete():
@@ -192,14 +205,16 @@ def batch_delete():
id_str, err = parse_batch_ids() id_str, err = parse_batch_ids()
if err: if err:
return err return err
return user.delete_users(id_str) count = user.delete_users(id_str)
return R.ok(msg=f"本次共删除{count}条数据")
# ============================================================ # ============================================================
# 用户状态设置 # 用户状态设置
# ============================================================ # ============================================================
@user_blue.route('/status', methods=["PUT"])
@user_blue.route("/status", methods=["PUT"])
@check_permission("sys:user:status") @check_permission("sys:user:status")
@check_demo @check_demo
def status(): def status():
@@ -212,15 +227,19 @@ def status():
Returns: Returns:
JSON: 操作结果 JSON: 操作结果
""" """
result = user.update_user_status() json_data, err = parse_request_body()
return result if err:
return R.failed(msg=err)
user.update_user_status(json_data)
return R.ok(msg="更新成功")
# ============================================================ # ============================================================
# 用户密码重置 # 用户密码重置
# ============================================================ # ============================================================
@user_blue.route('/resetPwd', methods=["PUT"])
@user_blue.route("/resetPwd", methods=["PUT"])
@check_permission("sys:user:resetPwd") @check_permission("sys:user:resetPwd")
@check_demo @check_demo
def reset_pwd(): def reset_pwd():
src/modules/system/user/service.py
+234 -315
View File
@@ -64,8 +64,6 @@
import logging import logging
from datetime import datetime, timezone from datetime import datetime, timezone
from flask import request
from sqlalchemy import and_
from src.common.constants.dictionary_constants import GENDER_LIST from src.common.constants.dictionary_constants import GENDER_LIST
from src.modules.system.user.schema import UserForm, UserStatusForm, UpdatePwdForm, UserInfoForm, ResetPwdForm from src.modules.system.user.schema import UserForm, UserStatusForm, UpdatePwdForm, UserInfoForm, ResetPwdForm
@@ -73,19 +71,20 @@ from src.modules.system.dept.models import Dept
from src.modules.system.level.models import Level from src.modules.system.level.models import Level
from src.modules.system.position.models import Position from src.modules.system.position.models import Position
from src.modules.system.user.models import User from src.modules.system.user.models import User
from src.modules.system.user.repository import UserRepository
from src.core.exceptions import BizError
from src.modules.association.user_role.models import UserRole from src.modules.association.user_role.models import UserRole
from src.modules.association.user_role import service as user_role from src.modules.association.user_role import service as user_role
from src.modules.system.menu.service import invalidate_user_cache from src.modules.system.menu.service import invalidate_user_cache
from src.common.constants.common import DEFAULT_PASSWORD, PAGE_SIZE from src.common.constants.common import DEFAULT_PASSWORD, PAGE_SIZE
from src.core.database import db from src.core.database import db
from src.core import response as R
from src.common.utils import regular from src.common.utils import regular
from src.core import password from src.core import password
from src.common.utils.case_converter import convert_camel_to_snake from src.common.utils.case_converter import convert_camel_to_snake
from src.common.utils.rich_text import get_file_url, save_file from src.common.utils.rich_text import get_file_url, save_file
from src.common.utils.model_utils import models_to_list, model_to_dict from src.common.utils.model_utils import models_to_list, model_to_dict
from src.core.password import verify_password, encrypt_password, generate_salt from src.core.password import verify_password, encrypt_password, generate_salt
from src.common.utils.request import parse_request_body, parse_id_list, like_contains from src.common.utils.request import parse_id_list, like_contains
from src.core.security import get_username, get_user_id from src.core.security import get_username, get_user_id
# 获取日志记录器 # 获取日志记录器
@@ -96,7 +95,8 @@ logger = logging.getLogger(__name__)
# 用户分页查询 # 用户分页查询
# ============================================================ # ============================================================
def get_user_page():
def get_user_page(params: dict) -> dict:
""" """
查询用户分页数据 查询用户分页数据
@@ -118,31 +118,26 @@ def get_user_page():
"pages": 总页数 "pages": 总页数
} }
""" """
try: # 获取分页参数
# 获取分页参数 page_no = int(params.get("page", 1))
page_no = int(request.args.get('page', 1)) page_size = int(params.get("limit", PAGE_SIZE))
page_size = int(request.args.get('limit', PAGE_SIZE))
# 应用筛选条件 # 应用筛选条件
query = _apply_filters() query = _apply_filters(params)
# 记录总数 # 记录总数
total = query.count() total = query.count()
# 计算总页数 # 计算总页数
pages = (total + page_size - 1) // page_size if total > 0 else 0 pages = (total + page_size - 1) // page_size if total > 0 else 0
# 分页查询 # 分页查询
user_list = query.limit(page_size).offset((page_no - 1) * page_size).all() user_list = query.limit(page_size).offset((page_no - 1) * page_size).all()
# 构建返回数据 # 构建返回数据
page_data = _build_page_data(user_list, total, page_size, page_no, pages) page_data = _build_page_data(user_list, total, page_size, page_no, pages)
return R.ok(data=page_data) return page_data
except Exception as e:
logger.error(f"查询用户分页数据异常: {str(e)}")
return R.failed(msg="查询失败，请稍后重试")
def _apply_filters(): def _apply_filters(params: dict):
""" """
应用查询筛选条件 应用查询筛选条件
@@ -160,17 +155,17 @@ def _apply_filters():
query = User.query.filter(User.is_delete == 0) query = User.query.filter(User.is_delete == 0)
# 用户姓名模糊筛选 # 用户姓名模糊筛选
realname = request.args.get('realname') realname = params.get("realname")
if realname: if realname:
query = query.filter(User.realname.like(like_contains(realname), escape='\\')) query = query.filter(User.realname.like(like_contains(realname), escape="\\"))
# 性别精确筛选 # 性别精确筛选
gender = request.args.get('gender') gender = params.get("gender")
if gender: if gender:
query = query.filter(User.gender == gender) query = query.filter(User.gender == gender)
# 用户状态精确筛选 # 用户状态精确筛选
status = request.args.get('status') status = params.get("status")
if status: if status:
query = query.filter(User.status == status) query = query.filter(User.status == status)
@@ -197,13 +192,7 @@ def _build_page_data(user_list, total, page_size, page_no, pages):
# 补充关联数据 # 补充关联数据
records = _enrich_user_records(records) records = _enrich_user_records(records)
return { return {"records": records, "total": total, "size": page_size, "current": page_no, "pages": pages}
'records': records,
'total': total,
'size': page_size,
'current': page_no,
'pages': pages
}
def _enrich_user_records(records): def _enrich_user_records(records):
@@ -225,27 +214,27 @@ def _enrich_user_records(records):
position_map = _get_position_map() position_map = _get_position_map()
# 批量查询所有用户的角色关系（1次查询替代N次） # 批量查询所有用户的角色关系（1次查询替代N次）
user_ids = [data.get('id') for data in records] user_ids = [data.get("id") for data in records]
role_map = user_role.get_user_roles_batch(user_ids) role_map = user_role.get_user_roles_batch(user_ids)
for data in records: for data in records:
# 城市编码列表 # 城市编码列表
city_list = [data.get('provinceCode'), data.get('cityCode'), data.get('districtCode'), data.get('streetCode')] city_list = [data.get("provinceCode"), data.get("cityCode"), data.get("districtCode"), data.get("streetCode")]
# 性别名称 # 性别名称
data['genderName'] = GENDER_LIST.get(data.get('gender')) data["genderName"] = GENDER_LIST.get(data.get("gender"))
# 头像URL # 头像URL
data['avatar'] = get_file_url(data.get('avatar')) data["avatar"] = get_file_url(data.get("avatar"))
# 部门名称 # 部门名称
data['deptName'] = dept_map.get(data.get('deptId')) data["deptName"] = dept_map.get(data.get("deptId"))
# 职级名称 # 职级名称
data['levelName'] = level_map.get(data.get('levelId')) data["levelName"] = level_map.get(data.get("levelId"))
# 岗位名称 # 岗位名称
data['positionName'] = position_map.get(data.get('positionId')) data["positionName"] = position_map.get(data.get("positionId"))
# 用户角色列表 # 用户角色列表
data['roles'] = role_map.get(data.get('id'), []) data["roles"] = role_map.get(data.get("id"), [])
# 行政区划 # 行政区划
data['city'] = city_list data["city"] = city_list
return records return records
@@ -272,6 +261,7 @@ def _get_position_map():
# 用户详情查询 # 用户详情查询
# ============================================================ # ============================================================
def get_user_detail(user_id): def get_user_detail(user_id):
""" """
根据ID查询用户详情（返回字典格式） 根据ID查询用户详情（返回字典格式）
@@ -295,7 +285,7 @@ def get_user_detail(user_id):
# 移除密码字段 # 移除密码字段
if data: if data:
data.pop('password', None) data.pop("password", None)
# 补充关联数据 # 补充关联数据
data = _enrich_user_detail(data) data = _enrich_user_detail(data)
@@ -320,15 +310,15 @@ def _enrich_user_detail(data):
return data return data
# 头像URL # 头像URL
data['avatar'] = get_file_url(data.get('avatar')) data["avatar"] = get_file_url(data.get("avatar"))
# 获取用户角色数据（复用 user_role 服务） # 获取用户角色数据（复用 user_role 服务）
role_list = user_role.get_user_role_list(data.get('id')) role_list = user_role.get_user_role_list(data.get("id"))
data['roles'] = [role['id'] for role in role_list] data["roles"] = [role["id"] for role in role_list]
# 行政区划 # 行政区划
city_list = [data.get('provinceCode'), data.get('cityCode'), data.get('districtCode'), data.get('streetCode')] city_list = [data.get("provinceCode"), data.get("cityCode"), data.get("districtCode"), data.get("streetCode")]
data['city'] = city_list data["city"] = city_list
return data return data
@@ -337,6 +327,7 @@ def _enrich_user_detail(data):
# 用户信息查询（根据ID）- 核心查询方法 # 用户信息查询（根据ID）- 核心查询方法
# ============================================================ # ============================================================
def get_info(user_id: int): def get_info(user_id: int):
""" """
根据ID查询用户信息（返回模型对象） 根据ID查询用户信息（返回模型对象）
@@ -362,27 +353,10 @@ def get_info(user_id: int):
if user: if user:
print(user.realname) print(user.realname)
""" """
# 参数有效性验证 return UserRepository.get_by_id(user_id)
if not user_id or user_id <= 0:
return None
try:
# 查询未删除的用户
user = User.query.filter(
and_(User.id == user_id, User.is_delete == 0)
).first()
return user
except Exception as e:
logger.error(f"查询用户信息异常，ID: {user_id}, 错误: {str(e)}")
return None
# ============================================================ def add_user(data: dict) -> None:
# 用户添加
# ============================================================
def add_user():
""" """
添加用户 添加用户
@@ -396,69 +370,58 @@ def add_user():
Returns: Returns:
R: 操作结果 R: 操作结果
""" """
try:
# 获取参数
json_data, error = parse_request_body()
if error:
return R.failed(msg=error)
# 表单验证 # 表单验证
form = UserForm(data=json_data) form = UserForm(data=data)
if not form.validate(): if not form.validate():
err_msg = regular.get_err(form) raise BizError(regular.get_err(form))
return R.failed(msg=err_msg)
# 图片处理 # 图片处理
avatar = form.data.get('avatar') avatar = form.data.get("avatar")
if avatar: if avatar:
form.avatar.data = save_file(avatar, "user") form.avatar.data = save_file(avatar, "user")
# 密码加密（未设置密码时使用默认密码） # 密码加密（未设置密码时使用默认密码）
pwd = form.data.get('password') or DEFAULT_PASSWORD pwd = form.data.get("password") or DEFAULT_PASSWORD
salt = password.generate_salt() salt = password.generate_salt()
form.password.data = encrypt_password(pwd, salt) form.password.data = encrypt_password(pwd, salt)
form.salt.data = salt form.salt.data = salt
# 处理行政区划 # 处理行政区划
city = form.data.get('city', []) city = form.data.get("city", [])
if len(city) != 4: if len(city) != 4:
return R.failed("请选择完整的所在城市（省/市/区/街道）") raise BizError("请选择完整的所在城市（省/市/区/街道）")
form.provinceCode.data = city[0] form.provinceCode.data = city[0]
form.cityCode.data = city[1] form.cityCode.data = city[1]
form.districtCode.data = city[2] form.districtCode.data = city[2]
form.streetCode.data = city[3] form.streetCode.data = city[3]
# 用户角色 # 用户角色
roles = form.data.get('roles') roles = form.data.get("roles")
# 移除不需要直接保存到User表的字段 # 移除不需要直接保存到User表的字段
form_data = form.data.copy() form_data = form.data.copy()
form_data.pop('city', None) form_data.pop("city", None)
form_data.pop('roles', None) form_data.pop("roles", None)
# 创建用户 # 创建用户
user = User(**form_data) user = User(**form_data)
user.create_user = get_username() user.create_user = get_username()
user.save() user.save()
# 创建用户角色数据 # 创建用户角色数据
_add_user_role(user.id, roles) _add_user_role(user.id, roles)
# 清除该用户权限缓存 # 清除该用户权限缓存
invalidate_user_cache(user.id) invalidate_user_cache(user.id)
return R.ok(msg="添加成功")
except Exception as e:
logger.error(f"添加用户异常: {str(e)}")
return R.failed(msg="添加用户失败，请稍后重试")
# ============================================================ # ============================================================
# 用户更新 # 用户更新
# ============================================================ # ============================================================
def update_user():
def update_user(data: dict) -> None:
""" """
更新用户 更新用户
@@ -473,83 +436,72 @@ def update_user():
Returns: Returns:
R: 操作结果 R: 操作结果
""" """
try:
# 获取参数
json_data, error = parse_request_body()
if error:
return R.failed(msg=error)
# 表单验证 # 表单验证
form = UserForm(data=json_data) form = UserForm(data=data)
if not form.validate(): if not form.validate():
return R.failed(msg=regular.get_err(form)) raise BizError(regular.get_err(form))
# 获取用户ID # 获取用户ID
user_id = form.data.get('id') user_id = form.data.get("id")
if not user_id or int(user_id) <= 0: if not user_id or int(user_id) <= 0:
return R.failed("用户ID无效") raise BizError("用户ID无效")
user_id = int(user_id) user_id = int(user_id)
# 复用 get_info 方法验证用户是否存在 # 复用 get_info 方法验证用户是否存在
user = get_info(user_id) user = get_info(user_id)
if not user: if not user:
return R.failed("用户不存在") raise BizError("用户不存在")
# 图片处理 # 图片处理
avatar = form.data.get('avatar') avatar = form.data.get("avatar")
if avatar: if avatar:
form.avatar.data = save_file(avatar, "user") form.avatar.data = save_file(avatar, "user")
# 密码加密（仅当有传入新密码时） # 密码加密（仅当有传入新密码时）
pwd = form.data.get('password') pwd = form.data.get("password")
if pwd: if pwd:
salt = password.generate_salt() salt = password.generate_salt()
form.password.data = encrypt_password(pwd, salt) form.password.data = encrypt_password(pwd, salt)
form.salt.data = salt form.salt.data = salt
else: else:
# 密码不填时保持原密码值不做更新 # 密码不填时保持原密码值不做更新
form_data = form.data.copy() form_data = form.data.copy()
form_data.pop('password', None) form_data.pop("password", None)
form_data.pop('salt', None) form_data.pop("salt", None)
# 处理行政区划 # 处理行政区划
city = form.data.get('city', []) city = form.data.get("city", [])
if len(city) != 4: if len(city) != 4:
return R.failed("请选择完整的所在城市（省/市/区/街道）") raise BizError("请选择完整的所在城市（省/市/区/街道）")
form.provinceCode.data = city[0] form.provinceCode.data = city[0]
form.cityCode.data = city[1] form.cityCode.data = city[1]
form.districtCode.data = city[2] form.districtCode.data = city[2]
form.streetCode.data = city[3] form.streetCode.data = city[3]
# 用户角色 # 用户角色
roles = form.data.get('roles') roles = form.data.get("roles")
# 移除不需要直接更新到User表的字段 # 移除不需要直接更新到User表的字段
form_data = convert_camel_to_snake(form.data.copy()) form_data = convert_camel_to_snake(form.data.copy())
form_data.pop('id', None) form_data.pop("id", None)
form_data.pop('city', None) form_data.pop("city", None)
form_data.pop('roles', None) form_data.pop("roles", None)
# 更新用户信息 # 更新用户信息
form_data['update_time'] = datetime.now(tz=timezone.utc) form_data["update_time"] = datetime.now(tz=timezone.utc)
form_data['update_user'] = get_username() form_data["update_user"] = get_username()
result = User.query.filter_by(id=user_id).update(form_data) result = User.query.filter_by(id=user_id).update(form_data)
if not result: if not result:
return R.failed("更新失败") raise BizError("更新失败")
# 更新用户角色数据 # 更新用户角色数据
_add_user_role(user_id, roles) _add_user_role(user_id, roles)
# 清除该用户权限缓存 # 清除该用户权限缓存
invalidate_user_cache(user_id) invalidate_user_cache(user_id)
return R.ok(msg="更新成功")
except Exception as e:
logger.error(f"更新用户异常: {str(e)}")
return R.failed(msg="更新用户失败，请稍后重试")
def _add_user_role(user_id, roles): def _add_user_role(user_id, roles):
@@ -566,17 +518,15 @@ def _add_user_role(user_id, roles):
# 批量插入 # 批量插入
if roles: if roles:
db.session.bulk_insert_mappings( db.session.bulk_insert_mappings(UserRole, [{"user_id": user_id, "role_id": role_id} for role_id in roles])
UserRole,
[{'user_id': user_id, 'role_id': role_id} for role_id in roles]
)
# ============================================================ # ============================================================
# 用户删除 # 用户删除
# ============================================================ # ============================================================
def delete_users(user_ids):
def delete_users(user_ids: str) -> int:
""" """
删除用户（批量删除） 删除用户（批量删除）
@@ -589,14 +539,18 @@ def delete_users(user_ids):
Returns: Returns:
R: 操作结果 R: 操作结果
""" """
return User.batch_delete_r(user_ids) user_ids_list = parse_id_list(user_ids)
if not user_ids_list:
raise BizError("无效的记录ID")
return User.batch_delete(user_ids_list)
# ============================================================ # ============================================================
# 用户状态设置 # 用户状态设置
# ============================================================ # ============================================================
def update_user_status():
def update_user_status(data: dict) -> None:
""" """
设置用户状态 设置用户状态
@@ -605,45 +559,34 @@ def update_user_status():
Returns: Returns:
R: 操作结果 R: 操作结果
""" """
try:
# 获取参数
json_data, error = parse_request_body()
if error:
return R.failed(msg=error)
# 表单验证 # 表单验证
form = UserStatusForm(data=json_data) form = UserStatusForm(data=data)
if not form.validate(): if not form.validate():
err_msg = regular.get_err(form) raise BizError(regular.get_err(form))
return R.failed(msg=err_msg)
user_id = int(form.data['id']) user_id = int(form.data["id"])
status = int(form.data['status']) status = int(form.data["status"])
# 复用 get_info 方法验证用户是否存在 # 复用 get_info 方法验证用户是否存在
user = get_info(user_id) user = get_info(user_id)
if not user: if not user:
return R.failed("用户不存在") raise BizError("用户不存在")
# 更新状态 # 更新状态
update_data = {"status": status, "update_time": datetime.now(tz=timezone.utc), "update_user": get_username()} update_data = {"status": status, "update_time": datetime.now(tz=timezone.utc), "update_user": get_username()}
result = User.query.filter_by(id=user_id).update(update_data) result = User.query.filter_by(id=user_id).update(update_data)
if not result: if not result:
return R.failed("设置失败") raise BizError("设置失败")
return R.ok(msg="设置成功")
except Exception as e:
logger.error(f"设置用户状态异常: {str(e)}")
return R.failed(msg="设置用户状态失败，请稍后重试")
# ============================================================ # ============================================================
# 用户信息更新（个人信息） # 用户信息更新（个人信息）
# ============================================================ # ============================================================
def update_user_info():
def update_user_info(data: dict) -> None:
""" """
更新当前用户信息 更新当前用户信息
@@ -652,54 +595,43 @@ def update_user_info():
Returns: Returns:
R: 操作结果 R: 操作结果
""" """
try:
# 获取参数
json_data, error = parse_request_body()
if error:
return R.failed(msg=error)
# 表单验证 # 表单验证
form = UserInfoForm(data=json_data) form = UserInfoForm(data=data)
if not form.validate(): if not form.validate():
err_msg = regular.get_err(form) raise BizError(regular.get_err(form))
return R.failed(msg=err_msg)
# 用户ID（当前登录用户） # 用户ID（当前登录用户）
user_id = get_user_id() user_id = get_user_id()
# 复用 get_info 方法验证用户是否存在 # 复用 get_info 方法验证用户是否存在
user = get_info(user_id) user = get_info(user_id)
if not user: if not user:
return R.failed("用户不存在") raise BizError("用户不存在")
# 更新用户信息 # 更新用户信息
update_data = { update_data = {
'realname': form.realname.data, "realname": form.realname.data,
'gender': form.gender.data, "gender": form.gender.data,
'mobile': form.mobile.data, "mobile": form.mobile.data,
'email': form.email.data, "email": form.email.data,
'address': form.address.data, "address": form.address.data,
'intro': form.intro.data, "intro": form.intro.data,
'update_time': datetime.now(tz=timezone.utc), "update_time": datetime.now(tz=timezone.utc),
'update_user': get_username() "update_user": get_username(),
} }
result = User.query.filter_by(id=user_id).update(update_data) result = User.query.filter_by(id=user_id).update(update_data)
if not result: if not result:
return R.failed("更新失败") raise BizError("更新失败")
return R.ok(msg="更新成功")
except Exception as e:
logger.error(f"更新用户信息异常: {str(e)}")
return R.failed(msg="更新用户信息失败，请稍后重试")
# ============================================================ # ============================================================
# 密码修改 # 密码修改
# ============================================================ # ============================================================
def update_password():
def update_password(data: dict) -> None:
""" """
修改当前用户密码 修改当前用户密码
@@ -708,51 +640,45 @@ def update_password():
Returns: Returns:
R: 操作结果 R: 操作结果
""" """
try:
# 获取参数
json_data, error = parse_request_body()
if error:
return R.failed(msg=error)
# 表单验证 # 表单验证
form = UpdatePwdForm(data=json_data) form = UpdatePwdForm(data=data)
if not form.validate(): if not form.validate():
err_msg = regular.get_err(form) raise BizError(regular.get_err(form))
return R.failed(msg=err_msg)
# 用户ID（当前登录用户） # 用户ID（当前登录用户）
user_id = get_user_id() user_id = get_user_id()
# 根据ID查询用户 # 根据ID查询用户
user = get_info(user_id) user = get_info(user_id)
if not user: if not user:
return R.failed("用户不存在") raise BizError("用户不存在")
# 验证原始密码 # 验证原始密码
if not verify_password(form.oldPassword.data, user.salt, user.password): if not verify_password(form.oldPassword.data, user.salt, user.password):
return R.failed("原始密码不正确") raise BizError("原始密码不正确")
# 加密新密码 # 加密新密码
new_salt = generate_salt() new_salt = generate_salt()
password = encrypt_password(form.newPassword.data, new_salt) password = encrypt_password(form.newPassword.data, new_salt)
update_data = {"password": password, "salt": new_salt, "update_time": datetime.now(tz=timezone.utc), "update_user": get_username()} update_data = {
result = User.query.filter_by(id=user_id).update(update_data) "password": password,
"salt": new_salt,
"update_time": datetime.now(tz=timezone.utc),
"update_user": get_username(),
}
result = User.query.filter_by(id=user_id).update(update_data)
if not result: if not result:
return R.failed("更新失败") raise BizError("更新失败")
return R.ok(msg="更新成功")
except Exception as e:
logger.error(f"修改密码异常: {str(e)}")
return R.failed(msg="修改密码失败，请稍后重试")
# ============================================================ # ============================================================
# 密码重置 # 密码重置
# ============================================================ # ============================================================
def reset_user_password():
def reset_user_password(data: dict) -> None:
""" """
重置用户密码 重置用户密码
@@ -761,36 +687,29 @@ def reset_user_password():
Returns: Returns:
R: 操作结果 R: 操作结果
""" """
try:
# 获取参数
json_data, error = parse_request_body()
if error:
return R.failed(msg=error)
# 表单验证 # 表单验证
form = ResetPwdForm(data=json_data) form = ResetPwdForm(data=data)
if not form.validate(): if not form.validate():
err_msg = regular.get_err(form) raise BizError(regular.get_err(form))
return R.failed(msg=err_msg)
user_id = form.userId.data user_id = form.userId.data
# 复用 get_info 方法验证用户是否存在 # 复用 get_info 方法验证用户是否存在
user = get_info(user_id) user = get_info(user_id)
if not user: if not user:
return R.failed("用户不存在") raise BizError("用户不存在")
# 加密新密码（默认密码123456） # 加密新密码（默认密码123456）
new_salt = generate_salt() new_salt = generate_salt()
password = encrypt_password(DEFAULT_PASSWORD, new_salt) password = encrypt_password(DEFAULT_PASSWORD, new_salt)
update_data = {"password": password, "salt": new_salt, "update_time": datetime.now(tz=timezone.utc), "update_user": get_username()} update_data = {
result = User.query.filter_by(id=user_id).update(update_data) "password": password,
"salt": new_salt,
"update_time": datetime.now(tz=timezone.utc),
"update_user": get_username(),
}
result = User.query.filter_by(id=user_id).update(update_data)
if not result: if not result:
return R.failed("重置密码失败") raise BizError("重置密码失败")
return R.ok(msg="重置密码成功")
except Exception as e:
logger.error(f"重置密码异常: {str(e)}")
return R.failed(msg="重置密码失败，请稍后重试")