87 lines
2.5 KiB
Python
87 lines
2.5 KiB
Python
# ===========================================
|
||
# 班级操行分管理系统 - 后端服务
|
||
#
|
||
# 开发者: Canglan
|
||
# 联系方式: admin@sea-studio.top
|
||
# 版权归属: Sea Network Technology Studio
|
||
# 许可证: MIT License
|
||
#
|
||
# 版权所有 © Sea Network Technology Studio
|
||
# ===========================================
|
||
|
||
from jose import jwt, JWTError
|
||
from datetime import datetime, timedelta
|
||
from typing import Optional, Dict, Any
|
||
|
||
from config import settings
|
||
from utils.logger import get_logger
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
|
||
class JWTHandler:
|
||
"""JWT Token处理类"""
|
||
|
||
@staticmethod
|
||
def create_token(user_id: int, username: str, user_type: str, student_id: int = None, role: str = None, real_name: str = None) -> str:
|
||
"""
|
||
创建JWT Token
|
||
"""
|
||
payload = {
|
||
'user_id': user_id,
|
||
'username': username,
|
||
'user_type': user_type,
|
||
'student_id': student_id,
|
||
'role': role,
|
||
'real_name': real_name,
|
||
'exp': datetime.utcnow() + timedelta(minutes=settings.JWT_EXPIRE_MINUTES),
|
||
'iat': datetime.utcnow(),
|
||
'iss': settings.APP_NAME
|
||
}
|
||
|
||
token = jwt.encode(
|
||
payload,
|
||
settings.JWT_SECRET_KEY,
|
||
algorithm=settings.JWT_ALGORITHM
|
||
)
|
||
return token
|
||
|
||
@staticmethod
|
||
def verify_token(token: str) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
验证JWT Token
|
||
返回: 解码后的payload,失败返回None
|
||
"""
|
||
try:
|
||
payload = jwt.decode(
|
||
token,
|
||
settings.JWT_SECRET_KEY,
|
||
algorithms=[settings.JWT_ALGORITHM],
|
||
options={'verify_exp': True}
|
||
)
|
||
return payload
|
||
except jwt.ExpiredSignatureError:
|
||
logger.warning("JWT Token已过期")
|
||
return None
|
||
except jwt.JWTError as e:
|
||
logger.warning(f"JWT Token验证失败: {e}")
|
||
return None
|
||
|
||
@staticmethod
|
||
def get_user_id_from_token(token: str) -> Optional[int]:
|
||
"""从Token中获取用户ID"""
|
||
payload = JWTHandler.verify_token(token)
|
||
if payload:
|
||
return payload.get('user_id')
|
||
return None
|
||
|
||
@staticmethod
|
||
def get_user_type_from_token(token: str) -> Optional[str]:
|
||
"""从Token中获取用户类型"""
|
||
payload = JWTHandler.verify_token(token)
|
||
if payload:
|
||
return payload.get('user_type')
|
||
return None
|
||
|
||
|
||
jwt_handler = JWTHandler() |