|
|
|
from fastapi import Request, Form
|
|
|
|
from fastapi import Depends
|
|
|
|
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
|
|
|
|
from jose import JWTError, jwt
|
|
|
|
import random
|
|
|
|
import uuid
|
|
|
|
from datetime import timedelta
|
|
|
|
from module_admin.service.user_service import *
|
|
|
|
from module_admin.entity.vo.login_vo import *
|
|
|
|
from module_admin.entity.vo.common_vo import CrudResponseModel
|
|
|
|
from module_admin.dao.login_dao import *
|
|
|
|
from exceptions.exception import LoginException, AuthException
|
|
|
|
from config.env import AppConfig, JwtConfig, RedisInitKeyConfig
|
|
|
|
from config.get_db import get_db
|
|
|
|
from utils.common_util import CamelCaseUtil
|
|
|
|
from utils.pwd_util import *
|
|
|
|
from utils.response_util import *
|
|
|
|
from utils.message_util import *
|
|
|
|
|
|
|
|
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")
|
|
|
|
|
|
|
|
|
|
|
|
class CustomOAuth2PasswordRequestForm(OAuth2PasswordRequestForm):
|
|
|
|
"""
|
|
|
|
自定义OAuth2PasswordRequestForm类,增加验证码及会话编号参数
|
|
|
|
"""
|
|
|
|
|
|
|
|
def __init__(
|
|
|
|
self,
|
|
|
|
grant_type: str = Form(default=None, regex="password"),
|
|
|
|
username: str = Form(),
|
|
|
|
password: str = Form(),
|
|
|
|
scope: str = Form(default=""),
|
|
|
|
client_id: Optional[str] = Form(default=None),
|
|
|
|
client_secret: Optional[str] = Form(default=None),
|
|
|
|
code: Optional[str] = Form(default=""),
|
|
|
|
uuid: Optional[str] = Form(default=""),
|
|
|
|
login_info: Optional[Dict[str, str]] = Form(default=None)
|
|
|
|
):
|
|
|
|
super().__init__(grant_type=grant_type, username=username, password=password,
|
|
|
|
scope=scope, client_id=client_id, client_secret=client_secret)
|
|
|
|
self.code = code
|
|
|
|
self.uuid = uuid
|
|
|
|
self.login_info = login_info
|
|
|
|
|
|
|
|
|
|
|
|
class LoginService:
|
|
|
|
"""
|
|
|
|
登录模块服务层
|
|
|
|
"""
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
async def authenticate_user(cls, request: Request, query_db: AsyncSession, login_user: UserLogin):
|
|
|
|
"""
|
|
|
|
根据用户名密码校验用户登录
|
|
|
|
:param request: Request对象
|
|
|
|
:param query_db: orm对象
|
|
|
|
:param login_user: 登录用户对象
|
|
|
|
:return: 校验结果
|
|
|
|
"""
|
|
|
|
await cls.__check_login_ip(request)
|
|
|
|
account_lock = await request.app.state.redis.get(
|
|
|
|
f"{RedisInitKeyConfig.ACCOUNT_LOCK.get('key')}:{login_user.user_name}")
|
|
|
|
if login_user.user_name == account_lock:
|
|
|
|
logger.warning("账号已锁定,请稍后再试")
|
|
|
|
raise LoginException(data="", message="账号已锁定,请稍后再试")
|
|
|
|
# 判断请求是否来自于api文档,如果是返回指定格式的结果,用于修复api文档认证成功后token显示undefined的bug
|
|
|
|
request_from_swagger = request.headers.get('referer').endswith('docs') if request.headers.get('referer') else False
|
|
|
|
request_from_redoc = request.headers.get('referer').endswith('redoc') if request.headers.get('referer') else False
|
|
|
|
# 判断是否开启验证码,开启则验证,否则不验证(dev模式下来自API文档的登录请求不检验)
|
|
|
|
if not login_user.captcha_enabled or ((request_from_swagger or request_from_redoc) and AppConfig.app_env == 'dev'):
|
|
|
|
pass
|
|
|
|
else:
|
|
|
|
await cls.__check_login_captcha(request, login_user)
|
|
|
|
user = await login_by_account(query_db, login_user.user_name)
|
|
|
|
print(user)
|
|
|
|
if not user:
|
|
|
|
logger.warning("用户不存在")
|
|
|
|
raise LoginException(data="", message="用户不存在")
|
|
|
|
if not PwdUtil.verify_password(login_user.password, user[0].password):
|
|
|
|
cache_password_error_count = await request.app.state.redis.get(
|
|
|
|
f"{RedisInitKeyConfig.PASSWORD_ERROR_COUNT.get('key')}:{login_user.user_name}")
|
|
|
|
password_error_counted = 0
|
|
|
|
if cache_password_error_count:
|
|
|
|
password_error_counted = cache_password_error_count
|
|
|
|
password_error_count = int(password_error_counted) + 1
|
|
|
|
await request.app.state.redis.set(
|
|
|
|
f"{RedisInitKeyConfig.PASSWORD_ERROR_COUNT.get('key')}:{login_user.user_name}", password_error_count,
|
|
|
|
ex=timedelta(minutes=10))
|
|
|
|
if password_error_count > 5:
|
|
|
|
await request.app.state.redis.delete(
|
|
|
|
f"{RedisInitKeyConfig.PASSWORD_ERROR_COUNT.get('key')}:{login_user.user_name}")
|
|
|
|
await request.app.state.redis.set(
|
|
|
|
f"{RedisInitKeyConfig.ACCOUNT_LOCK.get('key')}:{login_user.user_name}", login_user.user_name,
|
|
|
|
ex=timedelta(minutes=10))
|
|
|
|
logger.warning("10分钟内密码已输错超过5次,账号已锁定,请10分钟后再试")
|
|
|
|
raise LoginException(data="", message="10分钟内密码已输错超过5次,账号已锁定,请10分钟后再试")
|
|
|
|
logger.warning("密码错误")
|
|
|
|
raise LoginException(data="", message="密码错误")
|
|
|
|
if user[0].status == '1':
|
|
|
|
logger.warning("用户已停用")
|
|
|
|
raise LoginException(data="", message="用户已停用")
|
|
|
|
await request.app.state.redis.delete(
|
|
|
|
f"{RedisInitKeyConfig.PASSWORD_ERROR_COUNT.get('key')}:{login_user.user_name}")
|
|
|
|
return user
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
async def __check_login_ip(cls, request: Request):
|
|
|
|
"""
|
|
|
|
校验用户登录ip是否在黑名单内
|
|
|
|
:param request: Request对象
|
|
|
|
:return: 校验结果
|
|
|
|
"""
|
|
|
|
black_ip_value = await request.app.state.redis.get(
|
|
|
|
f"{RedisInitKeyConfig.SYS_CONFIG.get('key')}:sys.login.blackIPList")
|
|
|
|
black_ip_list = black_ip_value.split(',') if black_ip_value else []
|
|
|
|
if request.headers.get('X-Forwarded-For') in black_ip_list:
|
|
|
|
logger.warning("当前IP禁止登录")
|
|
|
|
raise LoginException(data="", message="当前IP禁止登录")
|
|
|
|
return True
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
async def __check_login_captcha(cls, request: Request, login_user: UserLogin):
|
|
|
|
"""
|
|
|
|
校验用户登录验证码
|
|
|
|
:param request: Request对象
|
|
|
|
:param login_user: 登录用户对象
|
|
|
|
:return: 校验结果
|
|
|
|
"""
|
|
|
|
captcha_value = await request.app.state.redis.get(
|
|
|
|
f"{RedisInitKeyConfig.CAPTCHA_CODES.get('key')}:{login_user.uuid}")
|
|
|
|
if not captcha_value:
|
|
|
|
logger.warning("验证码已失效")
|
|
|
|
raise LoginException(data="", message="验证码已失效")
|
|
|
|
if login_user.code != str(captcha_value):
|
|
|
|
logger.warning("验证码错误")
|
|
|
|
raise LoginException(data="", message="验证码错误")
|
|
|
|
return True
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
async def create_access_token(cls, data: dict, expires_delta: Union[timedelta, None] = None):
|
|
|
|
"""
|
|
|
|
根据登录信息创建当前用户token
|
|
|
|
:param data: 登录信息
|
|
|
|
:param expires_delta: token有效期
|
|
|
|
:return: token
|
|
|
|
"""
|
|
|
|
to_encode = data.copy()
|
|
|
|
if expires_delta:
|
|
|
|
expire = datetime.utcnow() + expires_delta
|
|
|
|
else:
|
|
|
|
expire = datetime.utcnow() + timedelta(minutes=30)
|
|
|
|
to_encode.update({"exp": expire})
|
|
|
|
encoded_jwt = jwt.encode(to_encode, JwtConfig.jwt_secret_key, algorithm=JwtConfig.jwt_algorithm)
|
|
|
|
return encoded_jwt
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
async def get_current_user(cls, request: Request = Request, token: str = Depends(oauth2_scheme),
|
|
|
|
query_db: AsyncSession = Depends(get_db)):
|
|
|
|
"""
|
|
|
|
根据token获取当前用户信息
|
|
|
|
:param request: Request对象
|
|
|
|
:param token: 用户token
|
|
|
|
:param query_db: orm对象
|
|
|
|
:return: 当前用户信息对象
|
|
|
|
:raise: 令牌异常AuthException
|
|
|
|
"""
|
|
|
|
# if token[:6] != 'Bearer':
|
|
|
|
# logger.warning("用户token不合法")
|
|
|
|
# raise AuthException(data="", message="用户token不合法")
|
|
|
|
try:
|
|
|
|
if token.startswith('Bearer'):
|
|
|
|
token = token.split(' ')[1]
|
|
|
|
payload = jwt.decode(token, JwtConfig.jwt_secret_key, algorithms=[JwtConfig.jwt_algorithm])
|
|
|
|
user_id: str = payload.get("user_id")
|
|
|
|
session_id: str = payload.get("session_id")
|
|
|
|
if user_id is None:
|
|
|
|
logger.warning("用户token不合法")
|
|
|
|
raise AuthException(data="", message="用户token不合法")
|
|
|
|
token_data = TokenData(user_id=int(user_id))
|
|
|
|
except JWTError:
|
|
|
|
logger.warning("用户token已失效,请重新登录")
|
|
|
|
raise AuthException(data="", message="用户token已失效,请重新登录")
|
|
|
|
query_user = await UserDao.get_user_by_id(query_db, user_id=token_data.user_id)
|
|
|
|
if query_user.get('user_basic_info') is None:
|
|
|
|
logger.warning("用户token不合法")
|
|
|
|
raise AuthException(data="", message="用户token不合法")
|
|
|
|
if AppConfig.app_same_time_login:
|
|
|
|
redis_token = await request.app.state.redis.get(f"{RedisInitKeyConfig.ACCESS_TOKEN.get('key')}:{session_id}")
|
|
|
|
else:
|
|
|
|
# 此方法可实现同一账号同一时间只能登录一次
|
|
|
|
redis_token = await request.app.state.redis.get(f"{RedisInitKeyConfig.ACCESS_TOKEN.get('key')}:{query_user.get('user_basic_info').user_id}")
|
|
|
|
if token == redis_token:
|
|
|
|
if AppConfig.app_same_time_login:
|
|
|
|
await request.app.state.redis.set(f"{RedisInitKeyConfig.ACCESS_TOKEN.get('key')}:{session_id}", redis_token,
|
|
|
|
ex=timedelta(minutes=JwtConfig.jwt_redis_expire_minutes))
|
|
|
|
else:
|
|
|
|
await request.app.state.redis.set(f"{RedisInitKeyConfig.ACCESS_TOKEN.get('key')}:{query_user.get('user_basic_info').user_id}", redis_token,
|
|
|
|
ex=timedelta(minutes=JwtConfig.jwt_redis_expire_minutes))
|
|
|
|
|
|
|
|
role_id_list = [item.role_id for item in query_user.get('user_role_info')]
|
|
|
|
if 1 in role_id_list:
|
|
|
|
permissions = ['*:*:*']
|
|
|
|
else:
|
|
|
|
permissions = [row.perms for row in query_user.get('user_menu_info')]
|
|
|
|
post_ids = ','.join([str(row.post_id) for row in query_user.get('user_post_info')])
|
|
|
|
role_ids = ','.join([str(row.role_id) for row in query_user.get('user_role_info')])
|
|
|
|
roles = [row.role_key for row in query_user.get('user_role_info')]
|
|
|
|
|
|
|
|
current_user = CurrentUserModel(
|
|
|
|
permissions=permissions,
|
|
|
|
roles=roles,
|
|
|
|
user=UserInfoModel(
|
|
|
|
**CamelCaseUtil.transform_result(query_user.get('user_basic_info')),
|
|
|
|
postIds=post_ids,
|
|
|
|
roleIds=role_ids,
|
|
|
|
dept=CamelCaseUtil.transform_result(query_user.get('user_dept_info')),
|
|
|
|
role=CamelCaseUtil.transform_result(query_user.get('user_role_info'))
|
|
|
|
)
|
|
|
|
)
|
|
|
|
return current_user
|
|
|
|
else:
|
|
|
|
logger.warning("用户token已失效,请重新登录")
|
|
|
|
raise AuthException(data="", message="用户token已失效,请重新登录")
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
async def get_current_user_routers(cls, user_id: int, query_db: AsyncSession):
|
|
|
|
"""
|
|
|
|
根据用户id获取当前用户路由信息
|
|
|
|
:param user_id: 用户id
|
|
|
|
:param query_db: orm对象
|
|
|
|
:return: 当前用户路由信息对象
|
|
|
|
"""
|
|
|
|
query_user = await UserDao.get_user_by_id(query_db, user_id=user_id)
|
|
|
|
user_router_menu = sorted([row for row in query_user.get('user_menu_info') if row.menu_type in ['M', 'C']], key=lambda x: x.order_num)
|
|
|
|
menus = cls.__generate_menus(0, user_router_menu)
|
|
|
|
user_router = cls.__generate_user_router_menu(menus)
|
|
|
|
return [router.model_dump(exclude_unset=True, by_alias=True) for router in user_router]
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
def __generate_menus(cls, pid: int, permission_list: List[SysMenu]):
|
|
|
|
"""
|
|
|
|
工具方法:根据菜单信息生成菜单信息树形嵌套数据
|
|
|
|
:param pid: 菜单id
|
|
|
|
:param permission_list: 菜单列表信息
|
|
|
|
:return: 菜单信息树形嵌套数据
|
|
|
|
"""
|
|
|
|
menu_list: List[MenuTreeModel] = []
|
|
|
|
for permission in permission_list:
|
|
|
|
if permission.parent_id == pid:
|
|
|
|
children = cls.__generate_menus(permission.menu_id, permission_list)
|
|
|
|
menu_list_data = MenuTreeModel(**CamelCaseUtil.transform_result(permission))
|
|
|
|
if children:
|
|
|
|
menu_list_data.children = children
|
|
|
|
menu_list.append(menu_list_data)
|
|
|
|
|
|
|
|
return menu_list
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
def __generate_user_router_menu(cls, permission_list: List[MenuTreeModel]):
|
|
|
|
"""
|
|
|
|
工具方法:根据菜单树信息生成路由信息树形嵌套数据
|
|
|
|
:param permission_list: 菜单树列表信息
|
|
|
|
:return: 路由信息树形嵌套数据
|
|
|
|
"""
|
|
|
|
router_list: List[RouterModel] = []
|
|
|
|
for permission in permission_list:
|
|
|
|
router = RouterModel(
|
|
|
|
hidden=True if permission.visible == '1' else False,
|
|
|
|
name=RouterUtil.get_router_name(permission),
|
|
|
|
path=RouterUtil.get_router_path(permission),
|
|
|
|
component=RouterUtil.get_component(permission),
|
|
|
|
query=permission.query,
|
|
|
|
meta=MetaModel(
|
|
|
|
title=permission.menu_name,
|
|
|
|
icon=permission.icon,
|
|
|
|
noCache=True if permission.is_cache == 1 else False,
|
|
|
|
link=permission.path if RouterUtil.is_http(permission.path) else None
|
|
|
|
)
|
|
|
|
)
|
|
|
|
c_menus = permission.children
|
|
|
|
if c_menus and permission.menu_type == 'M':
|
|
|
|
router.always_show = True
|
|
|
|
router.redirect = 'noRedirect'
|
|
|
|
router.children = cls.__generate_user_router_menu(c_menus)
|
|
|
|
elif RouterUtil.is_menu_frame(permission):
|
|
|
|
router.meta = None
|
|
|
|
children_list: List[RouterModel] = []
|
|
|
|
children = RouterModel(
|
|
|
|
path=permission.path,
|
|
|
|
component=permission.component,
|
|
|
|
name=permission.path.capitalize(),
|
|
|
|
meta=MetaModel(
|
|
|
|
title=permission.menu_name,
|
|
|
|
icon=permission.icon,
|
|
|
|
noCache=True if permission.is_cache == 1 else False,
|
|
|
|
link=permission.path if RouterUtil.is_http(permission.path) else None
|
|
|
|
),
|
|
|
|
query=permission.query
|
|
|
|
)
|
|
|
|
children_list.append(children)
|
|
|
|
router.children = children_list
|
|
|
|
elif permission.parent_id == 0 and RouterUtil.is_inner_link(permission):
|
|
|
|
router.meta = MetaModel(
|
|
|
|
title=permission.menu_name,
|
|
|
|
icon=permission.icon
|
|
|
|
)
|
|
|
|
router.path = '/'
|
|
|
|
children_list: List[RouterModel] = []
|
|
|
|
router_path = RouterUtil.inner_link_replace_each(permission.path)
|
|
|
|
children = RouterModel(
|
|
|
|
path=router_path,
|
|
|
|
component='InnerLink',
|
|
|
|
name=router_path.capitalize(),
|
|
|
|
meta=MetaModel(
|
|
|
|
title=permission.menu_name,
|
|
|
|
icon=permission.icon,
|
|
|
|
link=permission.path if RouterUtil.is_http(permission.path) else None
|
|
|
|
)
|
|
|
|
)
|
|
|
|
children_list.append(children)
|
|
|
|
router.children = children_list
|
|
|
|
|
|
|
|
router_list.append(router)
|
|
|
|
|
|
|
|
return router_list
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
async def register_user_services(cls, request: Request, query_db: AsyncSession, user_register: UserRegister):
|
|
|
|
"""
|
|
|
|
用户注册services
|
|
|
|
:param request: Request对象
|
|
|
|
:param query_db: orm对象
|
|
|
|
:param user_register: 注册用户对象
|
|
|
|
:return: 注册结果
|
|
|
|
"""
|
|
|
|
register_enabled = True if await request.app.state.redis.get(
|
|
|
|
f"{RedisInitKeyConfig.SYS_CONFIG.get('key')}:sys.account.registerUser") == 'true' else False
|
|
|
|
captcha_enabled = True if await request.app.state.redis.get(
|
|
|
|
f"{RedisInitKeyConfig.SYS_CONFIG.get('key')}:sys.account.captchaEnabled") == 'true' else False
|
|
|
|
if user_register.password == user_register.confirm_password:
|
|
|
|
if register_enabled:
|
|
|
|
if captcha_enabled:
|
|
|
|
captcha_value = await request.app.state.redis.get(
|
|
|
|
f"{RedisInitKeyConfig.CAPTCHA_CODES.get('key')}:{user_register.uuid}")
|
|
|
|
if not captcha_value:
|
|
|
|
logger.warning("验证码已失效")
|
|
|
|
return CrudResponseModel(is_success=False, message='验证码已失效')
|
|
|
|
elif user_register.code != str(captcha_value):
|
|
|
|
logger.warning("验证码错误")
|
|
|
|
return CrudResponseModel(is_success=False, message='验证码错误')
|
|
|
|
add_user = AddUserModel(
|
|
|
|
userName=user_register.username,
|
|
|
|
nickName=user_register.username,
|
|
|
|
password=PwdUtil.get_password_hash(user_register.password)
|
|
|
|
)
|
|
|
|
result = await UserService.add_user_services(query_db, add_user)
|
|
|
|
return result
|
|
|
|
else:
|
|
|
|
result = dict(is_success=False, message='注册程序已关闭,禁止注册')
|
|
|
|
else:
|
|
|
|
result = dict(is_success=False, message='两次输入的密码不一致')
|
|
|
|
|
|
|
|
return CrudResponseModel(**result)
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
async def get_sms_code_services(cls, request: Request, query_db: AsyncSession, user: ResetUserModel):
|
|
|
|
"""
|
|
|
|
获取短信验证码service
|
|
|
|
:param request: Request对象
|
|
|
|
:param query_db: orm对象
|
|
|
|
:param user: 用户对象
|
|
|
|
:return: 短信验证码对象
|
|
|
|
"""
|
|
|
|
redis_sms_result = await request.app.state.redis.get(
|
|
|
|
f"{RedisInitKeyConfig.SMS_CODE.get('key')}:{user.session_id}")
|
|
|
|
if redis_sms_result:
|
|
|
|
return SmsCode(**dict(is_success=False, sms_code='', session_id='', message='短信验证码仍在有效期内'))
|
|
|
|
is_user = await UserDao.get_user_by_name(query_db, user.user_name)
|
|
|
|
if is_user:
|
|
|
|
sms_code = str(random.randint(100000, 999999))
|
|
|
|
session_id = str(uuid.uuid4())
|
|
|
|
await request.app.state.redis.set(f"{RedisInitKeyConfig.SMS_CODE.get('key')}:{session_id}", sms_code,
|
|
|
|
ex=timedelta(minutes=2))
|
|
|
|
# 此处模拟调用短信服务
|
|
|
|
message_service(sms_code)
|
|
|
|
|
|
|
|
return SmsCode(**dict(is_success=True, sms_code=sms_code, session_id=session_id, message='获取成功'))
|
|
|
|
|
|
|
|
return SmsCode(**dict(is_success=False, sms_code='', session_id='', message='用户不存在'))
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
async def forget_user_services(cls, request: Request, query_db: AsyncSession, forget_user: ResetUserModel):
|
|
|
|
"""
|
|
|
|
用户忘记密码services
|
|
|
|
:param request: Request对象
|
|
|
|
:param query_db: orm对象
|
|
|
|
:param forget_user: 重置用户对象
|
|
|
|
:return: 重置结果
|
|
|
|
"""
|
|
|
|
redis_sms_result = await request.app.state.redis.get(
|
|
|
|
f"{RedisInitKeyConfig.SMS_CODE.get('key')}:{forget_user.session_id}")
|
|
|
|
if forget_user.sms_code == redis_sms_result:
|
|
|
|
forget_user.password = PwdUtil.get_password_hash(forget_user.password)
|
|
|
|
forget_user.user_id = (await UserDao.get_user_by_name(query_db, forget_user.user_name)).user_id
|
|
|
|
edit_result = await UserService.reset_user_services(query_db, forget_user)
|
|
|
|
result = edit_result.dict()
|
|
|
|
elif not redis_sms_result:
|
|
|
|
result = dict(is_success=False, message='短信验证码已过期')
|
|
|
|
else:
|
|
|
|
await request.app.state.redis.delete(f"{RedisInitKeyConfig.SMS_CODE.get('key')}:{forget_user.session_id}")
|
|
|
|
result = dict(is_success=False, message='短信验证码不正确')
|
|
|
|
|
|
|
|
return CrudResponseModel(**result)
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
async def logout_services(cls, request: Request, session_id: str):
|
|
|
|
"""
|
|
|
|
退出登录services
|
|
|
|
:param request: Request对象
|
|
|
|
:param session_id: 会话编号
|
|
|
|
:return: 退出登录结果
|
|
|
|
"""
|
|
|
|
await request.app.state.redis.delete(f"{RedisInitKeyConfig.ACCESS_TOKEN.get('key')}:{session_id}")
|
|
|
|
# await request.app.state.redis.delete(f'{current_user.user.user_id}_access_token')
|
|
|
|
# await request.app.state.redis.delete(f'{current_user.user.user_id}_session_id')
|
|
|
|
|
|
|
|
return True
|
|
|
|
|
|
|
|
|
|
|
|
class RouterUtil:
|
|
|
|
"""
|
|
|
|
路由处理工具类
|
|
|
|
"""
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
def get_router_name(cls, menu: MenuTreeModel):
|
|
|
|
"""
|
|
|
|
获取路由名称
|
|
|
|
:param menu: 菜单数对象
|
|
|
|
:return: 路由名称
|
|
|
|
"""
|
|
|
|
router_name = menu.path.capitalize()
|
|
|
|
if cls.is_menu_frame(menu):
|
|
|
|
router_name = ''
|
|
|
|
|
|
|
|
return router_name
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
def get_router_path(cls, menu: MenuTreeModel):
|
|
|
|
"""
|
|
|
|
获取路由地址
|
|
|
|
:param menu: 菜单数对象
|
|
|
|
:return: 路由地址
|
|
|
|
"""
|
|
|
|
# 内链打开外网方式
|
|
|
|
router_path = menu.path
|
|
|
|
if menu.parent_id != 0 and cls.is_inner_link(menu):
|
|
|
|
router_path = cls.inner_link_replace_each(router_path)
|
|
|
|
# 非外链并且是一级目录(类型为目录)
|
|
|
|
if menu.parent_id == 0 and menu.menu_type == 'M' and menu.is_frame == 1:
|
|
|
|
router_path = f'/{menu.path}'
|
|
|
|
# 非外链并且是一级目录(类型为菜单)
|
|
|
|
elif cls.is_menu_frame(menu):
|
|
|
|
router_path = '/'
|
|
|
|
return router_path
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
def get_component(cls, menu: MenuTreeModel):
|
|
|
|
"""
|
|
|
|
获取组件信息
|
|
|
|
:param menu: 菜单数对象
|
|
|
|
:return: 组件信息
|
|
|
|
"""
|
|
|
|
component = 'Layout'
|
|
|
|
if menu.component and not cls.is_menu_frame(menu):
|
|
|
|
component = menu.component
|
|
|
|
elif menu.component and menu.parent_id != 0 and cls.is_inner_link(menu):
|
|
|
|
component = 'InnerLink'
|
|
|
|
elif menu.component and cls.is_parent_view(menu):
|
|
|
|
component = 'ParentView'
|
|
|
|
return component
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
def is_menu_frame(cls, menu: MenuTreeModel):
|
|
|
|
"""
|
|
|
|
判断是否为菜单内部跳转
|
|
|
|
:param menu: 菜单数对象
|
|
|
|
:return: 是否为菜单内部跳转
|
|
|
|
"""
|
|
|
|
return menu.parent_id == 0 and menu.menu_type == 'C' and menu.is_frame == 1
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
def is_inner_link(cls, menu: MenuTreeModel):
|
|
|
|
"""
|
|
|
|
判断是否为内链组件
|
|
|
|
:param menu: 菜单数对象
|
|
|
|
:return: 是否为内链组件
|
|
|
|
"""
|
|
|
|
return menu.is_frame == 1 and cls.is_http(menu.path)
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
def is_parent_view(cls, menu: MenuTreeModel):
|
|
|
|
"""
|
|
|
|
判断是否为parent_view组件
|
|
|
|
:param menu: 菜单数对象
|
|
|
|
:return: 是否为parent_view组件
|
|
|
|
"""
|
|
|
|
return menu.parent_id != 0 and menu.menu_type == 'M'
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
def is_http(cls, link: str):
|
|
|
|
"""
|
|
|
|
判断是否为http(s)://开头
|
|
|
|
:param link: 链接
|
|
|
|
:return: 是否为http(s)://开头
|
|
|
|
"""
|
|
|
|
return link.startswith('http://') or link.startswith('https://')
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
def inner_link_replace_each(cls, path: str):
|
|
|
|
"""
|
|
|
|
内链域名特殊字符替换
|
|
|
|
:param path: 内链域名
|
|
|
|
:return: 替换后的内链域名
|
|
|
|
"""
|
|
|
|
old_values = ["http://", "https://", "www.", ".", ":"]
|
|
|
|
new_values = ["", "", "", "/", "/"]
|
|
|
|
for old, new in zip(old_values, new_values):
|
|
|
|
path = path.replace(old, new)
|
|
|
|
return path
|