jayhgq/BookSystem
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
main
BookSystem/backend/auth/init_data.py
jayhgq 9586c7467a feat: 菜单管理添加排序功能 
- Permission 模型新增 sort 字段
- 后端 API 按 sort 升序排列树形菜单
- 前端菜单管理页面增加排序列
- 初始数据添加排序号
- 修复 GET /permissions/{id} 详情接口 bug

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-20 08:33:51 +00:00

452 lines
20 KiB
Python
Raw Permalink Blame History

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, text
from models import Role, User, Permission, SystemSetting
from utils.password import get_password_hash
async def init_permissions(db: AsyncSession):
"""初始化权限数据（树形结构）"""
# 定义需要初始化的权限（树形结构）
# parentid: 0 = 顶级模块, >0 = 子权限
permission_tree = [
# ========================================
# 顶级模块 (DB ID: 1,2,3)
# ========================================
{"name": "权限管理", "code": "perm", "parentid": 0, "route": "/admin/permissions", "description": "权限管理模块", "sort": 1},
{"name": "系统设置", "code": "system", "parentid": 0, "route": "/admin/settings", "description": "系统设置模块", "sort": 2},
{"name": "日志管理", "code": "log", "parentid": 0, "route": "/admin/logs", "description": "日志管理模块", "sort": 3},
# ========================================
# 权限管理 -> 二级子模块 (parentid=1, DB ID: 4,5,6)
# ========================================
{"name": "用户管理", "code": "perm:user", "parentid": 1, "route": "/admin/users", "description": "用户管理子模块", "sort": 1},
{"name": "角色管理", "code": "perm:role", "parentid": 1, "route": "/admin/roles", "description": "角色管理子模块", "sort": 2},
{"name": "菜单管理", "code": "perm:menu", "parentid": 1, "route": "/admin/menu", "description": "菜单管理子模块", "sort": 3},
# ========================================
# 用户管理操作 (parentid=4, DB ID: 7-14)
# ========================================
{"name": "查看列表", "code": "perm:user:list", "parentid": 4, "description": "查看用户列表", "sort": 1},
{"name": "查看详情", "code": "perm:user:detail", "parentid": 4, "description": "查看用户详情", "sort": 2},
{"name": "搜索", "code": "perm:user:search", "parentid": 4, "description": "搜索用户", "sort": 3},
{"name": "创建", "code": "perm:user:create", "parentid": 4, "description": "创建新用户", "sort": 4},
{"name": "编辑", "code": "perm:user:edit", "parentid": 4, "description": "编辑用户信息", "sort": 5},
{"name": "删除", "code": "perm:user:delete", "parentid": 4, "description": "删除用户", "sort": 6},
{"name": "导出", "code": "perm:user:export", "parentid": 4, "description": "导出用户数据", "sort": 7},
{"name": "批量操作", "code": "perm:user:batch", "parentid": 4, "description": "批量操作用户", "sort": 8},
# ========================================
# 角色管理操作 (parentid=5, DB ID: 15-22)
# ========================================
{"name": "查看列表", "code": "perm:role:list", "parentid": 5, "description": "查看角色列表", "sort": 1},
{"name": "查看详情", "code": "perm:role:detail", "parentid": 5, "description": "查看角色详情", "sort": 2},
{"name": "搜索", "code": "perm:role:search", "parentid": 5, "description": "搜索角色", "sort": 3},
{"name": "创建", "code": "perm:role:create", "parentid": 5, "description": "创建新角色", "sort": 4},
{"name": "编辑", "code": "perm:role:edit", "parentid": 5, "description": "编辑角色信息", "sort": 5},
{"name": "删除", "code": "perm:role:delete", "parentid": 5, "description": "删除角色", "sort": 6},
{"name": "导出", "code": "perm:role:export", "parentid": 5, "description": "导出角色数据", "sort": 7},
{"name": "分配权限", "code": "perm:role:assign", "parentid": 5, "description": "为角色分配权限", "sort": 8},
# ========================================
# 菜单管理操作 (parentid=6, DB ID: 23-29)
# ========================================
{"name": "查看列表", "code": "perm:menu:list", "parentid": 6, "description": "查看菜单列表", "sort": 1},
{"name": "查看详情", "code": "perm:menu:detail", "parentid": 6, "description": "查看菜单详情", "sort": 2},
{"name": "搜索", "code": "perm:menu:search", "parentid": 6, "description": "搜索菜单", "sort": 3},
{"name": "创建", "code": "perm:menu:create", "parentid": 6, "description": "创建新菜单", "sort": 4},
{"name": "编辑", "code": "perm:menu:edit", "parentid": 6, "description": "编辑菜单信息", "sort": 5},
{"name": "删除", "code": "perm:menu:delete", "parentid": 6, "description": "删除菜单", "sort": 6},
{"name": "导出", "code": "perm:menu:export", "parentid": 6, "description": "导出菜单数据", "sort": 7},
# ========================================
# 系统设置 -> 子权限 (parentid=2, DB ID: 30)
# ========================================
{"name": "系统设置", "code": "system:view", "parentid": 2, "description": "查看和修改系统设置", "sort": 1},
# ========================================
# 日志管理 -> 子模块 (parentid=3, DB ID: 31,32,37)
# ========================================
{"name": "登录日志", "code": "log:login", "parentid": 3, "route": "/admin/login-logs", "description": "登录日志查看", "sort": 1},
{"name": "操作日志", "code": "log:operation", "parentid": 3, "route": "/admin/operation-logs", "description": "操作日志查看", "sort": 2},
{"name": "清理日志", "code": "log:clean", "parentid": 3, "description": "清理旧日志", "sort": 3},
# ========================================
# 登录日志操作 (parentid=31, DB ID: 33,34)
# ========================================
{"name": "查看列表", "code": "log:login:list", "parentid": 31, "description": "查看登录日志列表", "sort": 1},
{"name": "搜索", "code": "log:login:search", "parentid": 31, "description": "搜索登录日志", "sort": 2},
# ========================================
# 操作日志操作 (parentid=32, DB ID: 35,36)
# ========================================
{"name": "查看列表", "code": "log:operation:list", "parentid": 32, "description": "查看操作日志列表", "sort": 1},
{"name": "搜索", "code": "log:operation:search", "parentid": 32, "description": "搜索操作日志", "sort": 2},
]
# 检查现有权限
result = await db.execute(select(Permission))
existing_permissions = result.scalars().all()
existing_permission_codes = {perm.code: perm for perm in existing_permissions}
# 创建或更新权限
created_count = 0
updated_count = 0
for perm_data in permission_tree:
perm_code = perm_data["code"]
if perm_code not in existing_permission_codes:
# 权限不存在，需要创建
db.add(Permission(**perm_data))
created_count += 1
else:
# 权限存在，检查是否需要更新
existing_perm = existing_permission_codes[perm_code]
update_needed = False
if existing_perm.name != perm_data["name"]:
existing_perm.name = perm_data["name"]
update_needed = True
if existing_perm.parentid != perm_data["parentid"]:
existing_perm.parentid = perm_data["parentid"]
update_needed = True
if existing_perm.route != perm_data.get("route"):
existing_perm.route = perm_data.get("route")
update_needed = True
if existing_perm.description != perm_data["description"]:
existing_perm.description = perm_data["description"]
update_needed = True
if existing_perm.sort != perm_data.get("sort", 0):
existing_perm.sort = perm_data.get("sort", 0)
update_needed = True
if update_needed:
updated_count += 1
if created_count > 0 or updated_count > 0:
await db.commit()
print(f"创建了 {created_count} 个权限，更新了 {updated_count} 个权限")
else:
print("权限数据已存在且无需更新，跳过初始化")
print("权限数据初始化/更新完成")
async def init_role_permissions(db: AsyncSession):
"""初始化角色权限分配"""
from sqlalchemy.orm import selectinload
# 获取管理员角色（使用预加载避免懒加载问题）
result = await db.execute(
select(Role)
.filter(Role.name == "管理员")
.options(selectinload(Role.permissions))
)
admin_role = result.scalar_one_or_none()
if not admin_role:
print("管理员角色不存在，请先初始化角色数据")
return
# 获取所有权限
result = await db.execute(select(Permission))
all_permissions = result.scalars().all()
if not all_permissions:
print("没有找到任何权限，请先初始化权限数据")
return
# 获取管理员角色当前的权限（使用预加载的数据）
admin_permission_codes = set()
if admin_role.permissions:
admin_permission_codes = {perm.code for perm in admin_role.permissions}
# 检查是否需要添加权限
permissions_to_add = []
for perm in all_permissions:
if perm.code not in admin_permission_codes:
permissions_to_add.append(perm)
if permissions_to_add:
# 直接设置权限（替换而非追加，确保权限完整）
admin_role.permissions = all_permissions
await db.commit()
print(f"为管理员角色分配了 {len(permissions_to_add)} 个权限")
else:
print("管理员角色权限已完整，无需更新")
print("角色权限初始化/更新完成")
async def init_roles(db: AsyncSession):
"""初始化角色数据"""
# 定义需要初始化的角色
required_roles = [
{"name": "管理员", "creator": "system"},
{"name": "普通用户", "creator": "system"},
{"name": "访客", "creator": "system"},
{"name": "会员", "creator": "system"},
]
# 检查现有角色
result = await db.execute(select(Role))
existing_roles = result.scalars().all()
existing_role_names = {role.name for role in existing_roles}
# 检查需要创建或更新的角色
roles_to_add = []
roles_to_update = []
for role_data in required_roles:
role_name = role_data["name"]
if role_name not in existing_role_names:
# 角色不存在，需要创建
roles_to_add.append(Role(**role_data))
else:
# 角色存在，检查是否需要更新
existing_role = next(
role for role in existing_roles if role.name == role_name
)
if existing_role.creator != role_data["creator"]:
# 需要更新
existing_role.creator = role_data["creator"]
roles_to_update.append(existing_role)
# 执行创建和更新操作
if roles_to_add:
for role in roles_to_add:
db.add(role)
await db.commit()
print(f"创建了 {len(roles_to_add)} 个角色")
if roles_to_update:
await db.commit()
print(f"更新了 {len(roles_to_update)} 个角色")
if not roles_to_add and not roles_to_update:
print("角色数据已存在且无需更新，跳过初始化")
print("角色数据初始化/更新完成")
async def init_admin_user(db: AsyncSession):
"""初始化管理员用户"""
# 获取管理员角色
result = await db.execute(select(Role).filter(Role.name == "管理员"))
admin_role = result.scalar_one_or_none()
if not admin_role:
print("管理员角色不存在，请先初始化角色数据")
return
# 定义需要的管理员用户数据
admin_data = {
"username": "admin",
"nickname": "系统管理员",
"email": "admin@example.com",
"phone": "13800138000",
"password_hash": get_password_hash("admin"),
"role_id": admin_role.id,
"avatar": "",
"wx_openid": "",
}
# 检查管理员用户是否存在
result = await db.execute(select(User).filter(User.username == "admin"))
existing_admin = result.scalar_one_or_none()
if not existing_admin:
# 管理员不存在，创建
admin_user = User(**admin_data)
db.add(admin_user)
await db.commit()
print("创建了管理员用户")
else:
# 管理员存在，检查是否需要更新
update_needed = False
# 检查字段是否需要更新（不包括密码，因为密码只在首次创建时设置）
if existing_admin.nickname != admin_data["nickname"]:
existing_admin.nickname = admin_data["nickname"]
update_needed = True
if existing_admin.email != admin_data["email"]:
existing_admin.email = admin_data["email"]
update_needed = True
if existing_admin.phone != admin_data["phone"]:
existing_admin.phone = admin_data["phone"]
update_needed = True
if existing_admin.role_id != admin_data["role_id"]:
existing_admin.role_id = admin_data["role_id"]
update_needed = True
if existing_admin.avatar != admin_data["avatar"]:
existing_admin.avatar = admin_data["avatar"]
update_needed = True
if existing_admin.wx_openid != admin_data["wx_openid"]:
existing_admin.wx_openid = admin_data["wx_openid"]
update_needed = True
if update_needed:
await db.commit()
print("更新了管理员用户")
else:
print("管理员用户已存在且无需更新，跳过初始化")
print("管理员用户初始化/更新完成")
async def init_test_user(db: AsyncSession):
"""初始化测试用户"""
# 获取普通用户角色
result = await db.execute(select(Role).filter(Role.name == "普通用户"))
user_role = result.scalar_one_or_none()
if not user_role:
print("普通用户角色不存在，请先初始化角色数据")
return
# 定义需要的测试用户数据
test_data = {
"username": "test",
"nickname": "测试用户",
"email": "test@example.com",
"phone": "13800138001",
"password_hash": get_password_hash("test"),
"role_id": user_role.id,
"avatar": "",
"wx_openid": "",
}
# 检查测试用户是否存在
result = await db.execute(select(User).filter(User.username == "test"))
existing_test_user = result.scalar_one_or_none()
if not existing_test_user:
# 测试用户不存在，创建
test_user = User(**test_data)
db.add(test_user)
await db.commit()
print("创建了测试用户")
else:
# 测试用户存在，检查是否需要更新
update_needed = False
# 检查字段是否需要更新（不包括密码，因为密码只在首次创建时设置）
if existing_test_user.nickname != test_data["nickname"]:
existing_test_user.nickname = test_data["nickname"]
update_needed = True
if existing_test_user.email != test_data["email"]:
existing_test_user.email = test_data["email"]
update_needed = True
if existing_test_user.phone != test_data["phone"]:
existing_test_user.phone = test_data["phone"]
update_needed = True
if existing_test_user.role_id != test_data["role_id"]:
existing_test_user.role_id = test_data["role_id"]
update_needed = True
if existing_test_user.avatar != test_data["avatar"]:
existing_test_user.avatar = test_data["avatar"]
update_needed = True
if existing_test_user.wx_openid != test_data["wx_openid"]:
existing_test_user.wx_openid = test_data["wx_openid"]
update_needed = True
if update_needed:
await db.commit()
print("更新了测试用户")
else:
print("测试用户已存在且无需更新，跳过初始化")
print("测试用户初始化/更新完成")
async def init_system_settings(db: AsyncSession):
"""初始化系统设置数据"""
# 定义需要初始化的系统设置
system_settings = [
{
"key": "系统标题",
"value": "图书管理系统",
"type": "string",
"description": "系统显示标题"
},
{
"key": "页脚信息",
"value": "Copyright © 2026 图书管理系统 版权所有 | 技术支持：强蕊科技",
"type": "string",
"description": "页脚信息"
},
{
"key": "ICP信息",
"value": "京ICP备2023000000号",
"type": "string",
"description": "ICP备案信息"
},
{
"key": "启用手机短信注册",
"value": "false",
"type": "boolean",
"description": "是否允许通过手机短信验证码注册"
},
{
"key": "启用邮箱注册",
"value": "false",
"type": "boolean",
"description": "是否允许通过邮箱验证链接注册"
},
{
"key": "系统Logo",
"value": "",
"type": "image",
"description": "系统Logo图片"
}
]
# 检查现有系统设置
result = await db.execute(select(SystemSetting))
existing_settings = result.scalars().all()
existing_setting_keys = {setting.key: setting for setting in existing_settings}
# 创建或更新系统设置
created_count = 0
updated_count = 0
for setting_data in system_settings:
setting_key = setting_data["key"]
if setting_key not in existing_setting_keys:
# 设置不存在，需要创建
db.add(SystemSetting(**setting_data))
created_count += 1
else:
# 设置存在，检查是否需要更新
existing_setting = existing_setting_keys[setting_key]
update_needed = False
if existing_setting.value != setting_data["value"]:
existing_setting.value = setting_data["value"]
update_needed = True
if existing_setting.type != setting_data["type"]:
existing_setting.type = setting_data["type"]
update_needed = True
if existing_setting.description != setting_data.get("description"):
existing_setting.description = setting_data.get("description")
update_needed = True
if update_needed:
updated_count += 1
if created_count > 0 or updated_count > 0:
await db.commit()
print(f"创建了 {created_count} 个系统设置，更新了 {updated_count} 个系统设置")
else:
print("系统设置数据已存在且无需更新，跳过初始化")
print("系统设置数据初始化/更新完成")
async def init_all_data(db: AsyncSession):
"""初始化所有数据"""
print("开始初始化数据库数据...")
await init_roles(db)
await init_permissions(db)
await init_role_permissions(db)
await init_admin_user(db)
await init_test_user(db)
await init_system_settings(db)
print("数据库数据初始化完成")
Reference in New Issue View Git Blame Copy Permalink