mirror of
https://github.com/fastapiadmin/FastapiAdmin.git
synced 2026-10-09 02:57:35 +00:00
zhangtao
60e6452255
refactor(前端): 统一表单搜索组件样式和布局 style: 优化表格和卡片样式，调整高度和间距 fix(上传): 修复文件上传组件name属性缺失问题 perf(IP定位): 优化IP定位逻辑，增加内网IP判断和重试机制 docs: 更新密码修改成功后的提示信息 chore: 清理无用代码和注释 ``` 这个提交消息包含了多个方面的改进： 1. 新增功能(feat)：为用户表添加了avatar字段 2. 重构(refactor)：统一了前端表单搜索组件的样式和布局 3. 样式优化(style)：调整了表格和卡片的样式 4. 问题修复(fix)：修复了文件上传组件name属性缺失的问题 5. 性能优化(perf)：改进了IP定位服务的逻辑 6. 文档更新(docs)：修改了密码修改成功后的提示信息 7. 杂项(chore)：清理了无用的代码和注释 消息遵循了提交规范，每个部分都简明扼要地描述了变更内容。
78 lines
2.8 KiB
Python
78 lines
2.8 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import json
|
|
from typing import Dict, List, Optional
|
|
from aioredis import Redis
|
|
|
|
from app.common.enums import RedisInitKeyConfig
|
|
from app.core.exceptions import CustomException
|
|
from app.api.v1.params.monitor.online_param import OnlineQueryParams
|
|
from app.api.v1.schemas.monitor.online_schema import OnlineOutSchema
|
|
from app.core.redis_crud import RedisCURD
|
|
from app.core.security import decode_access_token
|
|
from app.core.logger import logger
|
|
|
|
class OnlineService:
|
|
"""在线用户管理模块服务层"""
|
|
|
|
@classmethod
|
|
async def get_online_list_service(cls, redis: Redis, search: Optional[OnlineQueryParams] = None) -> List[Dict]:
|
|
"""
|
|
获取在线用户列表信息（支持分页和搜索）
|
|
"""
|
|
|
|
keys = await RedisCURD(redis).get_keys(f"{RedisInitKeyConfig.ACCESS_TOKEN.key}:*")
|
|
tokens = await RedisCURD(redis).mget(*keys)
|
|
|
|
online_users = []
|
|
for token in tokens:
|
|
if not token:
|
|
continue
|
|
try:
|
|
payload = decode_access_token(token=token)
|
|
session_info = json.loads(payload.sub)
|
|
if cls._match_search_conditions(session_info, search):
|
|
online_users.append(session_info)
|
|
except Exception as e:
|
|
logger.error(f"解析在线用户数据失败: {e}")
|
|
continue
|
|
# 按照 login_time 倒序排序
|
|
online_users.sort(key=lambda x: x.get('login_time', ''), reverse=True)
|
|
|
|
return online_users
|
|
|
|
|
|
@classmethod
|
|
async def delete_online_service(cls, redis: Redis, session_id: str) -> bool:
|
|
"""强制下线在线用户"""
|
|
# 删除 token
|
|
await RedisCURD(redis).delete(f"{RedisInitKeyConfig.ACCESS_TOKEN.key}:{session_id}")
|
|
await RedisCURD(redis).delete(f"{RedisInitKeyConfig.REFRESH_TOKEN.key}:{session_id}")
|
|
|
|
|
|
logger.info(f"强制下线用户会话: {session_id}")
|
|
return True
|
|
|
|
@staticmethod
|
|
def _match_search_conditions(online_info: Dict, search: Optional[OnlineQueryParams]) -> bool:
|
|
"""检查是否匹配搜索条件"""
|
|
if not search:
|
|
return True
|
|
|
|
if search.name and search.name[1]:
|
|
keyword = search.name[1].strip('%')
|
|
if keyword.lower() not in online_info.get("name", "").lower():
|
|
return False
|
|
|
|
if search.ipaddr and search.ipaddr[1]:
|
|
keyword = search.ipaddr[1].strip('%')
|
|
if keyword not in online_info.get("ipaddr", ""):
|
|
return False
|
|
|
|
if search.login_location and search.login_location[1]:
|
|
keyword = search.login_location[1].strip('%')
|
|
if keyword.lower() not in online_info.get("login_location", "").lower():
|
|
return False
|
|
|
|
return True
|