feat(认证): 实现免登录快速登录功能

- 后端添加免登录用户列表、生成Token和免登录接口
- 前端添加快速登录标签页,支持用户头像快速登录
- 优化登录页面样式和交互体验
- 修复日志处理器移除时的异常处理问题
- 调整任务调度页面按钮文字和样式
This commit is contained in:
zhangtao
2026-03-03 00:18:21 +08:00
parent 4b9ebfd4ed
commit 2add3bed3b
11 changed files with 714 additions and 120 deletions
@@ -13,12 +13,14 @@ from app.core.router_class import OperationLogRoute
from app.core.security import CustomOAuth2PasswordRequestForm
from .schema import (
AutoLoginTokenSchema,
AutoLoginUserSchema,
CaptchaOutSchema,
JWTOutSchema,
LogoutPayloadSchema,
RefreshTokenPayloadSchema,
)
from .service import CaptchaService, LoginService
from .service import AutoLoginService, CaptchaService, LoginService
AuthRouter = APIRouter(route_class=OperationLogRoute, prefix="/auth", tags=["认证授权"])
@@ -150,3 +152,84 @@ async def logout_controller(
log.info("退出成功")
return SuccessResponse(msg="退出成功")
return ErrorResponse(msg="退出失败")
@AuthRouter.get(
"/auto-login/users",
summary="获取免登录用户列表",
description="获取可用于免登录快速登录的用户列表",
response_model=list[AutoLoginUserSchema],
)
async def get_auto_login_users_controller(
db: Annotated[AsyncSession, Depends(db_getter)],
) -> JSONResponse:
"""
获取免登录用户列表
参数:
- db (AsyncSession): 数据库会话对象
返回:
- list[AutoLoginUserSchema]: 免登录用户列表
"""
users = await AutoLoginService.get_auto_login_users_service(db=db)
return SuccessResponse(data=users, msg="获取成功")
@AuthRouter.post(
"/auto-login/token",
summary="获取免登录Token",
description="根据用户ID生成免登录Token",
response_model=AutoLoginTokenSchema,
)
async def get_auto_login_token_controller(
redis: Annotated[Redis, Depends(redis_getter)],
db: Annotated[AsyncSession, Depends(db_getter)],
user_id: int,
) -> JSONResponse:
"""
获取免登录Token
参数:
- redis (Redis): Redis客户端对象
- db (AsyncSession): 数据库会话对象
- user_id (int): 用户ID
返回:
- AutoLoginTokenSchema: 免登录Token和用户信息
"""
result = await AutoLoginService.create_auto_login_token_service(
redis=redis, db=db, user_id=user_id
)
return SuccessResponse(data=result, msg="获取成功")
@AuthRouter.post(
"/auto-login",
summary="免登录",
description="使用免登录Token快速登录",
response_model=JWTOutSchema,
)
async def auto_login_controller(
request: Request,
redis: Annotated[Redis, Depends(redis_getter)],
db: Annotated[AsyncSession, Depends(db_getter)],
token: str,
) -> JSONResponse:
"""
免登录
参数:
- request (Request): FastAPI请求对象
- redis (Redis): Redis客户端对象
- db (AsyncSession): 数据库会话对象
- token (str): 免登录Token
返回:
- JWTOutSchema: JWT令牌信息
"""
login_token = await AutoLoginService.auto_login_service(
request=request, redis=redis, db=db, token=token
)
log.info("用户免登录成功")
return SuccessResponse(data=login_token.model_dump(), msg="登录成功")
@@ -61,3 +61,23 @@ class CaptchaOutSchema(BaseModel):
enable: bool = Field(default=True, description="是否启用验证码")
key: str = Field(..., min_length=1, description="验证码唯一标识")
img_base: str = Field(..., min_length=1, description="Base64编码的验证码图片")
class AutoLoginUserSchema(BaseModel):
"""免登录用户信息模型"""
model_config = ConfigDict(from_attributes=True)
id: int = Field(..., description="用户ID")
username: str = Field(..., description="用户名")
name: str = Field(..., description="用户姓名")
avatar: str | None = Field(default=None, description="头像")
class AutoLoginTokenSchema(BaseModel):
"""免登录Token响应模型"""
model_config = ConfigDict(from_attributes=True)
token: str = Field(..., description="免登录Token")
user: AutoLoginUserSchema = Field(..., description="用户信息")
@@ -28,6 +28,8 @@ from app.utils.ip_local_util import IpLocalUtil
from .schema import (
AuthSchema,
AutoLoginTokenSchema,
AutoLoginUserSchema,
CaptchaOutSchema,
JWTOutSchema,
JWTPayloadSchema,
@@ -401,3 +403,162 @@ class CaptchaService:
await RedisCURD(redis).delete(redis_key)
log.info(f"验证码校验成功,key:{key}")
return True
class AutoLoginService:
"""免登录服务"""
# 免登录Token前缀
AUTO_LOGIN_PREFIX = "fastapiadmin:auto_login:"
# Token有效期(秒) - 5分钟
TOKEN_EXPIRE = 300
@classmethod
async def get_auto_login_users_service(cls, db: AsyncSession) -> list[AutoLoginUserSchema]:
"""
获取免登录用户列表
参数:
- db (AsyncSession): 数据库会话对象
返回:
- list[AutoLoginUserSchema]: 用户列表
"""
from sqlalchemy import select
from app.api.v1.module_system.user.model import UserModel
# 查询所有启用的用户
stmt = select(UserModel).where(UserModel.status == "0").order_by(UserModel.id)
result = await db.execute(stmt)
users = result.scalars().all()
return [
AutoLoginUserSchema(
id=user.id,
username=user.username,
name=user.name,
avatar=user.avatar,
)
for user in users
]
@classmethod
async def create_auto_login_token_service(
cls, redis: Redis, db: AsyncSession, user_id: int
) -> AutoLoginTokenSchema:
"""
创建免登录Token
参数:
- request (Request): FastAPI请求对象
- redis (Redis): Redis客户端对象
- db (AsyncSession): 数据库会话对象
- user_id (int): 用户ID
返回:
- AutoLoginTokenSchema: 免登录Token和用户信息
异常:
- CustomException: 用户不存在或已停用时抛出异常
"""
from sqlalchemy import select
from app.api.v1.module_system.user.model import UserModel
# 查询用户
stmt = select(UserModel).where(UserModel.id == user_id)
result = await db.execute(stmt)
user = result.scalar_one_or_none()
if not user:
raise CustomException(msg="用户不存在")
if user.status == "1":
raise CustomException(msg="用户已被停用")
# 生成免登录Token
import uuid
token = str(uuid.uuid4())
token_key = f"{cls.AUTO_LOGIN_PREFIX}{token}"
# 存储到Redis,设置5分钟过期
token_data = {
"user_id": user.id,
"username": user.username,
"created_at": datetime.now().isoformat(),
}
await RedisCURD(redis).set(
key=token_key,
value=json.dumps(token_data),
expire=cls.TOKEN_EXPIRE,
)
log.info(f"创建免登录Token成功,用户:{user.username}")
return AutoLoginTokenSchema(
token=token,
user=AutoLoginUserSchema(
id=user.id,
username=user.username,
name=user.name,
avatar=user.avatar,
),
)
@classmethod
async def auto_login_service(
cls, request: Request, redis: Redis, db: AsyncSession, token: str
) -> JWTOutSchema:
"""
免登录
参数:
- request (Request): FastAPI请求对象
- redis (Redis): Redis客户端对象
- db (AsyncSession): 数据库会话对象
- token (str): 免登录Token
返回:
- JWTOutSchema: JWT令牌信息
异常:
- CustomException: Token无效或过期时抛出异常
"""
from sqlalchemy import select
from app.api.v1.module_system.user.model import UserModel
# 验证Token
token_key = f"{cls.AUTO_LOGIN_PREFIX}{token}"
token_data_str = await RedisCURD(redis).get(token_key)
if not token_data_str:
raise CustomException(msg="免登录Token已过期或无效")
token_data = json.loads(token_data_str)
user_id = token_data.get("user_id")
# 查询用户
stmt = select(UserModel).where(UserModel.id == user_id)
result = await db.execute(stmt)
user = result.scalar_one_or_none()
if not user:
raise CustomException(msg="用户不存在")
if user.status == "1":
raise CustomException(msg="用户已被停用")
# 删除已使用的Token
await RedisCURD(redis).delete(token_key)
# 使用LoginService创建token
jwt_token = await LoginService.create_token_service(
request=request, redis=redis, user=user, login_type="PC端"
)
log.info(f"用户{user.username}免登录成功")
return jwt_token