调整目录结构
This commit is contained in:
zhengxinonly
@@ -0,0 +1,87 @@
|
||||
from applications.common.utils.validate import xss_escape
|
||||
from applications.extensions import db
|
||||
from applications.models import Dept, DeptSchema
|
||||
from applications.models import User
|
||||
from applications.common.curd import model_to_dicts
|
||||
|
||||
|
||||
def get_dept_dict():
|
||||
dept = Dept.query.order_by(Dept.sort).all()
|
||||
res = model_to_dicts(Schema=DeptSchema, model=dept)
|
||||
return res
|
||||
|
||||
|
||||
def save_dept(req):
|
||||
address = xss_escape(req.get("address"))
|
||||
deptName = xss_escape(req.get("deptName"))
|
||||
email = xss_escape(req.get("email"))
|
||||
leader = xss_escape(req.get("leader"))
|
||||
parentId = xss_escape(req.get("parentId"))
|
||||
phone = xss_escape(req.get("phone"))
|
||||
sort = xss_escape(req.get("sort"))
|
||||
status = xss_escape(req.get("status"))
|
||||
dept = Dept(
|
||||
parent_id=parentId,
|
||||
dept_name=deptName,
|
||||
sort=sort,
|
||||
leader=leader,
|
||||
phone=phone,
|
||||
email=email,
|
||||
status=status,
|
||||
address=address
|
||||
)
|
||||
r = db.session.add(dept)
|
||||
db.session.commit()
|
||||
return r
|
||||
|
||||
|
||||
def get_dept_by_id(id):
|
||||
d = Dept.query.filter_by(id=id).first()
|
||||
return d
|
||||
|
||||
|
||||
# 启动权限
|
||||
def enable_status(id):
|
||||
enable = 1
|
||||
d = Dept.query.filter_by(id=id).update({"status": enable})
|
||||
if d:
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# 停用权限
|
||||
def disable_status(id):
|
||||
enable = 0
|
||||
d = Dept.query.filter_by(id=id).update({"status": enable})
|
||||
if d:
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def update_dept(json):
|
||||
id = json.get("deptId"),
|
||||
data = {
|
||||
"dept_name": xss_escape(json.get("deptName")),
|
||||
"sort": xss_escape(json.get("sort")),
|
||||
"leader": xss_escape(json.get("leader")),
|
||||
"phone": xss_escape(json.get("phone")),
|
||||
"email": xss_escape(json.get("email")),
|
||||
"status": xss_escape(json.get("status")),
|
||||
"address": xss_escape(json.get("address"))
|
||||
}
|
||||
d = Dept.query.filter_by(id=id).update(data)
|
||||
if not d:
|
||||
return False
|
||||
db.session.commit()
|
||||
return True
|
||||
|
||||
|
||||
def remove_dept(id):
|
||||
d = Dept.query.filter_by(id=id).delete()
|
||||
if not d:
|
||||
return False
|
||||
User.query.filter_by(dept_id=id).update({"dept_id": None})
|
||||
db.session.commit()
|
||||
return True
|
||||
@@ -0,0 +1,145 @@
|
||||
from applications.common.utils.validate import xss_escape
|
||||
from applications.extensions import db
|
||||
from applications.models import DictType, DictData, DictTypeSchema, DictDataSchema
|
||||
from applications.common.curd import model_to_dicts
|
||||
|
||||
|
||||
def get_dict(typecode: str):
|
||||
dict_list = []
|
||||
if DictType.query.filter_by(type_code=typecode, enable=1).first():
|
||||
dicts = DictData.query.filter_by(type_code=typecode, enable=1).all()
|
||||
for d in dicts:
|
||||
dict_dict = {"key": d.data_label, "value": d.data_value}
|
||||
dict_list.append(dict_dict)
|
||||
else:
|
||||
return None
|
||||
return dict_list
|
||||
|
||||
|
||||
def get_dict_type(page, limit, type_name):
|
||||
dict_all = DictType.query
|
||||
if type_name:
|
||||
dict_all = dict_all.filter(DictType.type_name.like('%' + type_name + '%'))
|
||||
dict_all = dict_all.paginate(page=page,
|
||||
per_page=limit,
|
||||
error_out=False)
|
||||
count = DictType.query.count()
|
||||
data = model_to_dicts(Schema=DictTypeSchema, model=dict_all.items)
|
||||
return data, count
|
||||
|
||||
|
||||
def get_dict_data(page, limit, type_code):
|
||||
dict_all = DictData.query.filter_by(type_code=type_code).paginate(page=page,
|
||||
per_page=limit,
|
||||
error_out=False)
|
||||
count = DictType.query.count()
|
||||
dict_dict = model_to_dicts(Schema=DictDataSchema, model=dict_all.items)
|
||||
return dict_dict, count
|
||||
|
||||
|
||||
# 增加 dict type
|
||||
def save_dict_type(req_json):
|
||||
description = xss_escape(req_json.get("description"))
|
||||
enable = xss_escape(req_json.get("enable"))
|
||||
type_code = xss_escape(req_json.get("typeCode"))
|
||||
type_name = xss_escape(req_json.get("typeName"))
|
||||
d = DictType(type_name=type_name, type_code=type_code, enable=enable, description=description)
|
||||
db.session.add(d)
|
||||
db.session.commit()
|
||||
return d.id
|
||||
|
||||
|
||||
# 编辑字典类型
|
||||
def update_dict_type(req_json):
|
||||
id = xss_escape(req_json.get("id"))
|
||||
description = xss_escape(req_json.get("description"))
|
||||
enable = xss_escape(req_json.get("enable"))
|
||||
type_code = xss_escape(req_json.get("typeCode"))
|
||||
type_name = xss_escape(req_json.get("typeName"))
|
||||
DictType.query.filter_by(id=id).update({
|
||||
"description": description,
|
||||
"enable": enable,
|
||||
"type_code": type_code,
|
||||
"type_name": type_name
|
||||
})
|
||||
db.session.commit()
|
||||
return
|
||||
|
||||
|
||||
def enable_dict_type_status(id):
|
||||
enable = 1
|
||||
res = DictType.query.filter_by(id=id).update({"enable": enable})
|
||||
if res:
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def disable_dict_type_status(id):
|
||||
enable = 0
|
||||
res = DictType.query.filter_by(id=id).update({"enable": enable})
|
||||
if res:
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# 删除字典类型
|
||||
def delete_type_by_id(id):
|
||||
type_code = DictType.query.filter_by(id=id).first().type_code
|
||||
DictData.query.filter_by(type_code=type_code).delete()
|
||||
res = DictType.query.filter_by(id=id).delete()
|
||||
db.session.commit()
|
||||
return res
|
||||
|
||||
|
||||
# 增加dictdata
|
||||
def save_dict_data(req_json):
|
||||
data_label = xss_escape(req_json.get("dataLabel"))
|
||||
data_value = xss_escape(req_json.get("dataValue"))
|
||||
enable = xss_escape(req_json.get("enable"))
|
||||
remark = xss_escape(req_json.get("remark"))
|
||||
type_code = xss_escape(req_json.get("typeCode"))
|
||||
d = DictData(data_label=data_label, data_value=data_value, enable=enable, remark=remark, type_code=type_code)
|
||||
db.session.add(d)
|
||||
db.session.commit()
|
||||
return d.id
|
||||
|
||||
|
||||
# 编辑字典数据
|
||||
def update_dict_data(req_json):
|
||||
id = req_json.get("dataId")
|
||||
DictData.query.filter_by(id=id).update({
|
||||
"data_label": xss_escape(req_json.get("dataLabel")),
|
||||
"data_value": xss_escape(req_json.get("dataValue")),
|
||||
"enable": xss_escape(req_json.get("enable")),
|
||||
"remark": xss_escape(req_json.get("remark")),
|
||||
"type_code": xss_escape(req_json.get("typeCode"))
|
||||
})
|
||||
db.session.commit()
|
||||
return
|
||||
|
||||
|
||||
def enable_dict_data_status(id):
|
||||
enable = 1
|
||||
res = DictData.query.filter_by(id=id).update({"enable": enable})
|
||||
if res:
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def disable_dict_data_status(id):
|
||||
enable = 0
|
||||
res = DictData.query.filter_by(id=id).update({"enable": enable})
|
||||
if res:
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# 删除dictdata
|
||||
def delete_data_by_id(id):
|
||||
res = DictData.query.filter_by(id=id).delete()
|
||||
db.session.commit()
|
||||
return res
|
||||
@@ -0,0 +1,36 @@
|
||||
import os
|
||||
from flask import current_app
|
||||
from sqlalchemy import desc
|
||||
|
||||
from applications.common.utils.upload import photos
|
||||
from applications.extensions import db
|
||||
from applications.models import Photo, PhotoSchema
|
||||
from applications.common.curd import model_to_dicts
|
||||
|
||||
|
||||
def get_photo(page, limit):
|
||||
photo = Photo.query.order_by(desc(Photo.create_time)).paginate(page=page, per_page=limit, error_out=False)
|
||||
count = Photo.query.count()
|
||||
data = model_to_dicts(Schema=PhotoSchema, model=photo.items)
|
||||
return data, count
|
||||
|
||||
|
||||
def upload_one(photo, mime):
|
||||
filename = photos.save(photo)
|
||||
file_url = photos.url(filename)
|
||||
|
||||
upload_url = current_app.config.get("UPLOADED_PHOTOS_DEST")
|
||||
size = os.path.getsize(upload_url + '/' + filename)
|
||||
photo = Photo(name=filename, href=file_url, mime=mime, size=size)
|
||||
db.session.add(photo)
|
||||
db.session.commit()
|
||||
return file_url
|
||||
|
||||
|
||||
def delete_photo_by_id(_id):
|
||||
photo_name = Photo.query.filter_by(id=_id).first().name
|
||||
photo = Photo.query.filter_by(id=_id).delete()
|
||||
db.session.commit()
|
||||
upload_url = current_app.config.get("UPLOADED_PHOTOS_DEST")
|
||||
os.remove(upload_url + '/' + photo_name)
|
||||
return photo
|
||||
@@ -0,0 +1,146 @@
|
||||
from io import BytesIO
|
||||
from flask import session, make_response, current_app
|
||||
from flask_login import current_user
|
||||
|
||||
from applications.common.utils.gen_captcha import gen_captcha
|
||||
from applications.models.admin_power import PowerSchema
|
||||
|
||||
|
||||
# 授权路由存入session
|
||||
def add_auth_session():
|
||||
role = current_user.role
|
||||
user_power = []
|
||||
for i in role:
|
||||
if i.enable == 0:
|
||||
continue
|
||||
for p in i.power:
|
||||
if p.enable == 0:
|
||||
continue
|
||||
user_power.append(p.code)
|
||||
session['permissions'] = user_power
|
||||
|
||||
|
||||
# 生成菜单树
|
||||
def make_menu_tree():
|
||||
# power0 = Power.query.filter(
|
||||
# Power.type == 0,
|
||||
# ).all()
|
||||
# power1 = Power.query.filter(
|
||||
# Power.type == 1
|
||||
# ).all()
|
||||
role = current_user.role
|
||||
power0 = []
|
||||
power1 = []
|
||||
for i in role:
|
||||
if i.enable == 0:
|
||||
continue
|
||||
for p in i.power:
|
||||
if p.enable == 0:
|
||||
continue
|
||||
if int(p.type) == 0:
|
||||
power0.append(p)
|
||||
else:
|
||||
power1.append(p)
|
||||
|
||||
power_schema = PowerSchema(many=True) # 用已继承ma.ModelSchema类的自定制类生成序列化类
|
||||
power0_dict = power_schema.dump(power0) # 生成可序列化对象
|
||||
power1_dict = power_schema.dump(power1) # 生成可序列化对象
|
||||
power0_dict = sorted(power0_dict, key=lambda i: i['sort'])
|
||||
power1_dict = sorted(power1_dict, key=lambda i: i['sort'])
|
||||
# print(power0)
|
||||
# print(power1)
|
||||
|
||||
menu = []
|
||||
|
||||
for p0 in power0_dict:
|
||||
for p1 in power1_dict:
|
||||
if p0.get('id') == p1.get('parent_id'):
|
||||
if p0.get("children") is None:
|
||||
p0['children'] = []
|
||||
p0['children'].append(p1)
|
||||
menu.append(p0)
|
||||
return menu
|
||||
|
||||
|
||||
# 生成验证码
|
||||
def get_captcha():
|
||||
code, image = gen_captcha()
|
||||
out = BytesIO()
|
||||
session["code"] = code
|
||||
image.save(out, 'png')
|
||||
out.seek(0)
|
||||
resp = make_response(out.read())
|
||||
resp.content_type = 'image/png'
|
||||
return resp, code
|
||||
|
||||
|
||||
def get_render_config():
|
||||
# 网站配置
|
||||
config = dict(logo={
|
||||
# 网站名称
|
||||
"title": current_app.config.get("SYSTEM_NAME"),
|
||||
# 网站图标
|
||||
"image": "/static/admin/admin/images/logo.png"
|
||||
# 菜单配置
|
||||
}, menu={
|
||||
# 菜单数据来源
|
||||
"data": "/admin/menu",
|
||||
"collaspe": True,
|
||||
# 是否同时只打开一个菜单目录
|
||||
"accordion": True,
|
||||
"method": "GET",
|
||||
# 是否开启多系统菜单模式
|
||||
"control": False,
|
||||
# 默认选中的菜单项
|
||||
"select": "0",
|
||||
# 是否开启异步菜单，false 时 data 属性设置为菜单数据，false 时为 json 文件或后端接口
|
||||
"async": True
|
||||
}, tab={
|
||||
# 是否开启多选项卡
|
||||
"muiltTab": True,
|
||||
# 切换选项卡时，是否刷新页面状态
|
||||
"keepState": True,
|
||||
# 是否开启 Tab 记忆
|
||||
"session": True,
|
||||
# 最大可打开的选项卡数量
|
||||
"tabMax": 30,
|
||||
"index": {
|
||||
# 标识 ID , 建议与菜单项中的 ID 一致
|
||||
"id": "10",
|
||||
# 页面地址
|
||||
"href": "/admin/welcome",
|
||||
# 标题
|
||||
"title": "首页"
|
||||
}
|
||||
}, theme={
|
||||
# 默认主题色，对应 colors 配置中的 ID 标识
|
||||
"defaultColor": "2",
|
||||
# 默认的菜单主题 dark-theme 黑 / light-theme 白
|
||||
"defaultMenu": "dark-theme",
|
||||
# 是否允许用户切换主题，false 时关闭自定义主题面板
|
||||
"allowCustom": True
|
||||
}, colors=[{
|
||||
"id": "1",
|
||||
"color": "#2d8cf0"
|
||||
},
|
||||
{
|
||||
"id": "2",
|
||||
"color": "#5FB878"
|
||||
},
|
||||
{
|
||||
"id": "3",
|
||||
"color": "#1E9FFF"
|
||||
}, {
|
||||
"id": "4",
|
||||
"color": "#FFB800"
|
||||
}, {
|
||||
"id": "5",
|
||||
"color": "darkgray"
|
||||
}
|
||||
], links=current_app.config.get("SYSTEM_PANEL_LINKS"), other={
|
||||
# 主页动画时长
|
||||
"keepLoad": 1200,
|
||||
# 布局顶部主题
|
||||
"autoHead": False
|
||||
}, header=False)
|
||||
return config
|
||||
@@ -0,0 +1,112 @@
|
||||
from applications.common.utils.validate import xss_escape
|
||||
from applications.extensions import db
|
||||
from applications.models.admin_power import Power, PowerSchema2
|
||||
from applications.models import Role
|
||||
from applications.common.curd import model_to_dicts
|
||||
|
||||
|
||||
def get_power_dict():
|
||||
power = Power.query.all()
|
||||
res = model_to_dicts(Schema=PowerSchema2, model=power)
|
||||
return res
|
||||
|
||||
|
||||
# 选择父节点
|
||||
def select_parent():
|
||||
power = Power.query.all()
|
||||
res = model_to_dicts(Schema=PowerSchema2, model=power)
|
||||
res.append({"powerId": 0, "powerName": "顶级权限", "parentId": -1})
|
||||
return res
|
||||
|
||||
|
||||
# 增加权限
|
||||
def save_power(req):
|
||||
icon = xss_escape(req.get("icon"))
|
||||
openType = xss_escape(req.get("openType"))
|
||||
parentId = xss_escape(req.get("parentId"))
|
||||
powerCode = xss_escape(req.get("powerCode"))
|
||||
powerName = xss_escape(req.get("powerName"))
|
||||
powerType = xss_escape(req.get("powerType"))
|
||||
powerUrl = xss_escape(req.get("powerUrl"))
|
||||
sort = xss_escape(req.get("sort"))
|
||||
power = Power(
|
||||
icon=icon,
|
||||
open_type=openType,
|
||||
parent_id=parentId,
|
||||
code=powerCode,
|
||||
name=powerName,
|
||||
type=powerType,
|
||||
url=powerUrl,
|
||||
sort=sort,
|
||||
enable=1
|
||||
)
|
||||
r = db.session.add(power)
|
||||
db.session.commit()
|
||||
return r
|
||||
|
||||
|
||||
# 根据id查询权限
|
||||
def get_power_by_id(id):
|
||||
p = Power.query.filter_by(id=id).first()
|
||||
return p
|
||||
|
||||
|
||||
# 更新权限
|
||||
def update_power(req_json):
|
||||
id = req_json.get("powerId")
|
||||
data = {
|
||||
"icon": xss_escape(req_json.get("icon")),
|
||||
"open_type": xss_escape(req_json.get("openType")),
|
||||
"parent_id": xss_escape(req_json.get("parentId")),
|
||||
"code": xss_escape(req_json.get("powerCode")),
|
||||
"name": xss_escape(req_json.get("powerName")),
|
||||
"type": xss_escape(req_json.get("powerType")),
|
||||
"url": xss_escape(req_json.get("powerUrl")),
|
||||
"sort": xss_escape(req_json.get("sort"))
|
||||
}
|
||||
# print(data)
|
||||
power = Power.query.filter_by(id=id).update(data)
|
||||
db.session.commit()
|
||||
# print(power)
|
||||
return power
|
||||
|
||||
|
||||
# 启动权限
|
||||
def enable_status(id):
|
||||
enable = 1
|
||||
user = Power.query.filter_by(id=id).update({"enable": enable})
|
||||
if user:
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# 停用权限
|
||||
def disable_status(id):
|
||||
enable = 0
|
||||
user = Power.query.filter_by(id=id).update({"enable": enable})
|
||||
if user:
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# 删除权限（目前没有判断父节点自动删除子节点）
|
||||
def remove_power(id):
|
||||
power = Power.query.filter_by(id=id).first()
|
||||
role_id_list = []
|
||||
roles = power.role
|
||||
for role in roles:
|
||||
role_id_list.append(role.id)
|
||||
roles = Role.query.filter(Role.id.in_(role_id_list)).all()
|
||||
for p in roles:
|
||||
power.role.remove(p)
|
||||
r = Power.query.filter_by(id=id).delete()
|
||||
db.session.commit()
|
||||
return r
|
||||
|
||||
|
||||
# 批量删除权限
|
||||
def batch_remove(ids):
|
||||
for id in ids:
|
||||
remove_power(id)
|
||||
@@ -0,0 +1,149 @@
|
||||
from sqlalchemy import and_
|
||||
|
||||
from applications.common.utils.validate import xss_escape
|
||||
from applications.extensions import db
|
||||
from applications.models import Role, RoleSchema
|
||||
from applications.models.admin_power import Power, PowerSchema2
|
||||
from applications.models import User
|
||||
|
||||
# 获取角色对象
|
||||
from applications.common.curd import model_to_dicts
|
||||
|
||||
|
||||
def get_role_data(page, limit, filters):
|
||||
role = Role.query.filter(and_(*[getattr(Role, k).like(v) for k, v in filters.items()])).paginate(page=page,
|
||||
per_page=limit,
|
||||
error_out=False)
|
||||
count = Role.query.count()
|
||||
return role, count
|
||||
|
||||
|
||||
# 获取角色dict
|
||||
def get_role_data_dict(page, limit, filters):
|
||||
role, count = get_role_data(page, limit, filters)
|
||||
data = model_to_dicts(Schema=RoleSchema, model=role.items)
|
||||
return data, count
|
||||
|
||||
|
||||
# 增加角色
|
||||
def add_role(req):
|
||||
details = xss_escape(req.get("details"))
|
||||
enable = xss_escape(req.get("enable"))
|
||||
roleCode = xss_escape(req.get("roleCode"))
|
||||
roleName = xss_escape(req.get("roleName"))
|
||||
sort = xss_escape(req.get("sort"))
|
||||
role = Role(
|
||||
details=details,
|
||||
enable=enable,
|
||||
code=roleCode,
|
||||
name=roleName,
|
||||
sort=sort
|
||||
)
|
||||
db.session.add(role)
|
||||
db.session.commit()
|
||||
|
||||
|
||||
# 通过id获取角色
|
||||
def get_role_by_id(id):
|
||||
r = Role.query.filter_by(id=id).first()
|
||||
return r
|
||||
|
||||
|
||||
# 更新角色
|
||||
def update_role(req_json):
|
||||
id = req_json.get("roleId")
|
||||
data = {
|
||||
"code": xss_escape(req_json.get("roleCode")),
|
||||
"name": xss_escape(req_json.get("roleName")),
|
||||
"sort": xss_escape(req_json.get("sort")),
|
||||
"enable": xss_escape(req_json.get("enable")),
|
||||
"details": xss_escape(req_json.get("details"))
|
||||
}
|
||||
role = Role.query.filter_by(id=id).update(data)
|
||||
db.session.commit()
|
||||
return role
|
||||
|
||||
|
||||
# 获取角色的权限
|
||||
def get_role_power(id):
|
||||
role = Role.query.filter_by(id=id).first()
|
||||
check_powers = role.power
|
||||
check_powers_list = []
|
||||
for cp in check_powers:
|
||||
check_powers_list.append(cp.id)
|
||||
powers = Power.query.all()
|
||||
power_schema = PowerSchema2(many=True) # 用已继承ma.ModelSchema类的自定制类生成序列化类
|
||||
output = power_schema.dump(powers) # 生成可序列化对象
|
||||
for i in output:
|
||||
if int(i.get("powerId")) in check_powers_list:
|
||||
i["checkArr"] = "1"
|
||||
else:
|
||||
i["checkArr"] = "0"
|
||||
return output
|
||||
|
||||
|
||||
# 更新角色权限
|
||||
def update_role_power(id, power_list):
|
||||
role = Role.query.filter_by(id=id).first()
|
||||
power_id_list = []
|
||||
for p in role.power:
|
||||
power_id_list.append(p.id)
|
||||
# print(p.id)
|
||||
# print(power_id_list)
|
||||
powers = Power.query.filter(Power.id.in_(power_id_list)).all()
|
||||
for p in powers:
|
||||
role.power.remove(p)
|
||||
powers = Power.query.filter(Power.id.in_(power_list)).all()
|
||||
for p in powers:
|
||||
role.power.append(p)
|
||||
db.session.commit()
|
||||
|
||||
|
||||
# 启动角色
|
||||
def enable_status(id):
|
||||
enable = 1
|
||||
role = Role.query.filter_by(id=id).update({"enable": enable})
|
||||
if role:
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# 停用角色
|
||||
def disable_status(id):
|
||||
enable = 0
|
||||
role = Role.query.filter_by(id=id).update({"enable": enable})
|
||||
if role:
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# 删除角色
|
||||
def remove_role(id):
|
||||
role = Role.query.filter_by(id=id).first()
|
||||
# 删除该角色的权限
|
||||
power_id_list = []
|
||||
for p in role.power:
|
||||
power_id_list.append(p.id)
|
||||
|
||||
powers = Power.query.filter(Power.id.in_(power_id_list)).all()
|
||||
for p in powers:
|
||||
role.power.remove(p)
|
||||
user_id_list = []
|
||||
for u in role.user:
|
||||
user_id_list.append(u.id)
|
||||
users = User.query.filter(User.id.in_(user_id_list)).all()
|
||||
for u in users:
|
||||
role.user.remove(u)
|
||||
r = Role.query.filter_by(id=id).delete()
|
||||
db.session.commit()
|
||||
return r
|
||||
|
||||
|
||||
# 批量删除
|
||||
def batch_remove(ids):
|
||||
# role = Role.query.filter(Role.id.in_(ids)).delete(synchronize_session=False)
|
||||
# db.session.commit()
|
||||
for id in ids:
|
||||
remove_role(id)
|
||||
@@ -0,0 +1,159 @@
|
||||
from flask import jsonify
|
||||
from flask_login import current_user
|
||||
from sqlalchemy import and_, desc
|
||||
from applications.extensions import db
|
||||
from applications.models import User, UserSchema
|
||||
from applications.models import Role
|
||||
from applications.models import AdminLog
|
||||
|
||||
# 获取用户的 sqlalchemy 对象分页器
|
||||
from applications.common.curd import model_to_dicts
|
||||
|
||||
|
||||
def get_user_data(page, limit, filters, deptId):
|
||||
if deptId:
|
||||
user = User.query.filter_by(dept_id=deptId).filter(
|
||||
and_(*[getattr(User, k).like(v) for k, v in filters.items()])).paginate(page=page,
|
||||
per_page=limit,
|
||||
error_out=False)
|
||||
else:
|
||||
user = User.query.filter(and_(*[getattr(User, k).like(v) for k, v in filters.items()])).paginate(page=page,
|
||||
per_page=limit,
|
||||
error_out=False)
|
||||
count = User.query.count()
|
||||
return user, count
|
||||
|
||||
|
||||
# 获取用户的dict数据分页器
|
||||
def get_user_data_dict(page, limit, filters, deptId):
|
||||
user, count = get_user_data(page, limit, filters, deptId)
|
||||
data = model_to_dicts(Schema=UserSchema, model=user.items)
|
||||
return data, count
|
||||
|
||||
|
||||
# 通过名称获取用户
|
||||
def get_user_by_name(username):
|
||||
return User.query.filter_by(username=username).first()
|
||||
|
||||
|
||||
# 获取当前用户日志
|
||||
def get_current_user_logs():
|
||||
log = AdminLog.query.filter_by(url='/admin/login').filter_by(uid=current_user.id).order_by(
|
||||
desc(AdminLog.create_time)).limit(10)
|
||||
return log
|
||||
|
||||
|
||||
# 判断用户是否存在
|
||||
def is_user_exists(username):
|
||||
res = User.query.filter_by(username=username).count()
|
||||
return bool(res)
|
||||
|
||||
|
||||
# 增加用户
|
||||
def add_user(username, realName, password):
|
||||
user = User(username=username, realname=realName)
|
||||
user.set_password(password)
|
||||
db.session.add(user)
|
||||
db.session.commit()
|
||||
return user.id
|
||||
|
||||
|
||||
# 增加用户角色
|
||||
def add_user_role(id, roles_list):
|
||||
user = User.query.filter_by(id=id).first()
|
||||
roles = Role.query.filter(Role.id.in_(roles_list)).all()
|
||||
for r in roles:
|
||||
user.role.append(r)
|
||||
db.session.commit()
|
||||
|
||||
|
||||
# 更新用户头像
|
||||
def update_avatar(url):
|
||||
r = User.query.filter_by(id=current_user.id).update({"avatar": url})
|
||||
db.session.commit()
|
||||
return r
|
||||
|
||||
|
||||
# 更新用户信息
|
||||
def update_user(id, username, realname, deptId):
|
||||
user = User.query.filter_by(id=id).update({'username': username, 'realname': realname, 'dept_id': deptId})
|
||||
db.session.commit()
|
||||
return user
|
||||
|
||||
|
||||
# 更新当前用户信息
|
||||
def update_current_user_info(req_json):
|
||||
r = User.query.filter_by(id=current_user.id).update(
|
||||
{"realname": req_json.get("realName"), "remark": req_json.get("details")})
|
||||
db.session.commit()
|
||||
return r
|
||||
|
||||
|
||||
# 修改当前用户密码
|
||||
def edit_password(res_json):
|
||||
if res_json.get("newPassword") == '':
|
||||
return jsonify(success=False, msg="新密码不得为空")
|
||||
if res_json.get("newPassword") != res_json.get("confirmPassword"):
|
||||
return jsonify(success=False, msg="俩次密码不一样")
|
||||
user = current_user
|
||||
is_right = user.validate_password(res_json.get("oldPassword"))
|
||||
if not is_right:
|
||||
return jsonify(success=False, msg="旧密码错误")
|
||||
user.set_password(res_json.get("newPassword"))
|
||||
db.session.add(user)
|
||||
db.session.commit()
|
||||
return jsonify(success=True, msg="更改成功")
|
||||
|
||||
|
||||
# 删除用户
|
||||
def delete_by_id(id):
|
||||
user = User.query.filter_by(id=id).first()
|
||||
roles_id = []
|
||||
for role in user.role:
|
||||
roles_id.append(role.id)
|
||||
roles = Role.query.filter(Role.id.in_(roles_id)).all()
|
||||
for r in roles:
|
||||
user.role.remove(r)
|
||||
res = User.query.filter_by(id=id).delete()
|
||||
db.session.commit()
|
||||
return res
|
||||
|
||||
|
||||
# 启用用户
|
||||
def enable_status(id):
|
||||
enable = 1
|
||||
user = User.query.filter_by(id=id).update({"enable": enable})
|
||||
if user:
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# 停用用户
|
||||
def disable_status(id):
|
||||
enable = 0
|
||||
user = User.query.filter_by(id=id).update({"enable": enable})
|
||||
if user:
|
||||
db.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# 批量删除
|
||||
def batch_remove(ids):
|
||||
for id in ids:
|
||||
delete_by_id(id)
|
||||
|
||||
|
||||
def update_user_role(id, roles_list):
|
||||
user = User.query.filter_by(id=id).first()
|
||||
roles_id = []
|
||||
for role in user.role:
|
||||
roles_id.append(role.id)
|
||||
roles = Role.query.filter(Role.id.in_(roles_id)).all()
|
||||
for r in roles:
|
||||
user.role.remove(r)
|
||||
roles = Role.query.filter(Role.id.in_(roles_list)).all()
|
||||
for r in roles:
|
||||
user.role.append(r)
|
||||
db.session.commit()
|
||||
@@ -0,0 +1,56 @@
|
||||
from flask_login import current_user
|
||||
|
||||
from applications.extensions import db
|
||||
from applications.models import AdminLog
|
||||
|
||||
|
||||
def login_log(request, uid, is_access):
|
||||
info = {
|
||||
'method': request.method,
|
||||
'url': request.path,
|
||||
'ip': request.remote_addr,
|
||||
'user_agent': request.headers.get('User-Agent'),
|
||||
'desc': request.form.get('username'),
|
||||
'uid': uid,
|
||||
'success': int(is_access)
|
||||
|
||||
}
|
||||
log = AdminLog(
|
||||
url=info.get('url'),
|
||||
ip=info.get('ip'),
|
||||
user_agent=info.get('user_agent'),
|
||||
desc=info.get('desc'),
|
||||
uid=info.get('uid'),
|
||||
method=info.get('method'),
|
||||
success=info.get('success')
|
||||
)
|
||||
db.session.add(log)
|
||||
db.session.flush()
|
||||
db.session.commit()
|
||||
return log.id
|
||||
|
||||
|
||||
def admin_log(request, is_access):
|
||||
info = {
|
||||
'method': request.method,
|
||||
'url': request.path,
|
||||
'ip': request.remote_addr,
|
||||
'user_agent': request.headers.get('User-Agent'),
|
||||
'desc': str(dict(request.values)),
|
||||
'uid': current_user.id,
|
||||
'success': int(is_access)
|
||||
|
||||
}
|
||||
log = AdminLog(
|
||||
url=info.get('url'),
|
||||
ip=info.get('ip'),
|
||||
user_agent=info.get('user_agent'),
|
||||
desc=info.get('desc'),
|
||||
uid=info.get('uid'),
|
||||
method=info.get('method'),
|
||||
success=info.get('success')
|
||||
)
|
||||
db.session.add(log)
|
||||
db.session.commit()
|
||||
|
||||
return log.id
|
||||
@@ -0,0 +1,5 @@
|
||||
def model_to_dicts(Schema, model):
|
||||
# 如果是分页器返回，需要传入model.items
|
||||
common_schema = Schema(many=True) # 用已继承ma.ModelSchema类的自定制类生成序列化类
|
||||
output = common_schema.dump(model) # 生成可序列化对象
|
||||
return output
|
||||
@@ -0,0 +1,17 @@
|
||||
#
|
||||
MYSQL_HOST = '127.0.0.1'
|
||||
MYSQL_PORT = '3306'
|
||||
MYSQL_DATABASE = 'PearAdminFlask'
|
||||
MYSQL_USERNAME = 'root'
|
||||
MYSQL_PASSWORD = 'root'
|
||||
|
||||
MYSQL_DB_URI = "mysql+pymysql://{username}:{password}@{host}:{port}/{db}?charset=utf8".format(username=USERNAME,
|
||||
password=PASSWORD,
|
||||
host=HOST,
|
||||
port=PORT,
|
||||
db=DATABASE)
|
||||
|
||||
SQLALCHEMY_DATABASE_URI = MYSQL_DB_URI
|
||||
SQLALCHEMY_TRACK_MODIFICATIONS = True
|
||||
SQLALCHEMY_ECHO = False
|
||||
SQLALCHEMY_POOL_RECYCLE = 8
|
||||
@@ -0,0 +1,14 @@
|
||||
from captcha.image import ImageCaptcha
|
||||
|
||||
from PIL import Image
|
||||
from random import choices
|
||||
|
||||
|
||||
def gen_captcha(content='0123456789'):
|
||||
""" 生成验证码 """
|
||||
image = ImageCaptcha()
|
||||
# 获取字符串
|
||||
captcha_text = "".join(choices(content, k=4))
|
||||
# 生成图像
|
||||
captcha_image = Image.open(image.generate(captcha_text))
|
||||
return captcha_text, captcha_image
|
||||
@@ -0,0 +1,24 @@
|
||||
from flask import jsonify
|
||||
|
||||
|
||||
def success_api(msg: str = "成功"):
|
||||
""" 成功响应 默认值”成功“ """
|
||||
return jsonify(success=True, message=msg)
|
||||
|
||||
|
||||
def fail_api(msg: str = "失败"):
|
||||
""" 失败响应 默认值“失败” """
|
||||
return jsonify(success=False, message=msg)
|
||||
|
||||
|
||||
def table_api(msg: str = "", count=0, data=None, limit=10):
|
||||
""" 动态表格渲染响应 """
|
||||
res = {
|
||||
'message': msg,
|
||||
'code': 0,
|
||||
'data': data,
|
||||
'count': count,
|
||||
'limit': limit
|
||||
|
||||
}
|
||||
return jsonify(res)
|
||||
@@ -0,0 +1,25 @@
|
||||
from functools import wraps
|
||||
from flask import abort, request, jsonify, session
|
||||
from flask_login import login_required
|
||||
from applications.common.admin_log import admin_log
|
||||
|
||||
|
||||
def authorize(power: str, log: bool = False):
|
||||
def decorator(func):
|
||||
@login_required
|
||||
@wraps(func)
|
||||
def wrapper(*args, **kwargs):
|
||||
if not power in session.get('permissions'):
|
||||
if log:
|
||||
admin_log(request=request, is_access=False)
|
||||
if request.method == 'GET':
|
||||
abort(403)
|
||||
else:
|
||||
return jsonify(success=False, msg="权限不足!")
|
||||
if log:
|
||||
admin_log(request=request, is_access=True)
|
||||
return func(*args, **kwargs)
|
||||
|
||||
return wrapper
|
||||
|
||||
return decorator
|
||||
@@ -0,0 +1,4 @@
|
||||
from flask_uploads import UploadSet, IMAGES
|
||||
|
||||
|
||||
photos = UploadSet('photos', IMAGES)
|
||||
@@ -0,0 +1,20 @@
|
||||
# xss过滤
|
||||
from flask import abort, make_response, jsonify
|
||||
|
||||
|
||||
def xss_escape(s: str):
|
||||
if s is None:
|
||||
return None
|
||||
else:
|
||||
return s.replace("&", "&").replace(">", ">").replace("<", "<").replace("'", "'").replace('"',
|
||||
""")
|
||||
|
||||
|
||||
def check_data(schema, data):
|
||||
errors = schema.validate(data)
|
||||
for k, v in errors.items():
|
||||
for i in v:
|
||||
# print("{}{}".format(k, i))
|
||||
msg = "{}{}".format(k, i)
|
||||
if errors:
|
||||
abort(make_response(jsonify(result=False, msg=msg), 200))
|
||||
Reference in New Issue
Block a user