mirror of
https://github.com/zhenxun-org/zhenxun_bot.git
synced 2026-10-03 10:40:02 +08:00
perf👌: 完善系统管理api
This commit is contained in:
@@ -2,3 +2,4 @@ from .database import *
|
||||
from .main import *
|
||||
from .manage import *
|
||||
from .plugin_manage import *
|
||||
from .system import *
|
||||
|
||||
@@ -101,4 +101,4 @@ async def _(query: QueryModel) -> Result:
|
||||
async def _(plugin_name: Optional[str] = None) -> Result:
|
||||
if plugin_name:
|
||||
return Result.ok(SQL_DICT.get(plugin_name))
|
||||
return Result.ok(SQL_DICT)
|
||||
return Result.ok(str(SQL_DICT))
|
||||
|
||||
@@ -246,14 +246,14 @@ async def _(date_type: Optional[QueryDateType] = None) -> Result:
|
||||
|
||||
|
||||
@ws_router.websocket("/system_status")
|
||||
async def system_logs_realtime(websocket: WebSocket):
|
||||
async def system_logs_realtime(websocket: WebSocket, sleep: Optional[int] = 5):
|
||||
await websocket.accept()
|
||||
logger.debug("ws system_status is connect")
|
||||
try:
|
||||
while websocket.client_state == WebSocketState.CONNECTED:
|
||||
system_status = await get_system_status()
|
||||
await websocket.send_text(system_status.json())
|
||||
await asyncio.sleep(5)
|
||||
await asyncio.sleep(sleep)
|
||||
except (WebSocketDisconnect, ConnectionClosedError, ConnectionClosedOK):
|
||||
pass
|
||||
return
|
||||
|
||||
@@ -0,0 +1,26 @@
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import List, Optional
|
||||
|
||||
from fastapi import APIRouter
|
||||
|
||||
from ....base_model import Result
|
||||
from ....utils import authentication, get_system_disk
|
||||
from .model import DirFile
|
||||
|
||||
router = APIRouter(prefix="/system")
|
||||
|
||||
|
||||
|
||||
@router.get("/get_dir_list", dependencies=[authentication()], description="获取文件列表")
|
||||
async def _(path: Optional[str] = None) -> Result:
|
||||
base_path = Path(path) if path else Path()
|
||||
data_list = []
|
||||
for file in os.listdir(base_path):
|
||||
data_list.append(DirFile(is_file=not (base_path / file).is_dir(), name=file, parent=path))
|
||||
return Result.ok(data_list)
|
||||
|
||||
|
||||
@router.get("/get_resources_size", dependencies=[authentication()], description="获取文件列表")
|
||||
async def _(type: Optional[str] = None) -> Result:
|
||||
return Result.ok(await get_system_disk(type))
|
||||
@@ -0,0 +1,35 @@
|
||||
|
||||
|
||||
|
||||
from datetime import datetime
|
||||
from typing import Literal, Optional
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
class DirFile(BaseModel):
|
||||
|
||||
"""
|
||||
文件或文件夹
|
||||
"""
|
||||
|
||||
is_file: bool
|
||||
"""是否为文件"""
|
||||
name: str
|
||||
"""文件夹或文件名称"""
|
||||
parent: Optional[str] = None
|
||||
"""父级"""
|
||||
|
||||
class SystemFolderSize(BaseModel):
|
||||
"""
|
||||
资源文件占比
|
||||
"""
|
||||
|
||||
font_dir_size: float
|
||||
image_dir_size: float
|
||||
text_dir_size: float
|
||||
record_dir_size: float
|
||||
temp_dir_size: float
|
||||
data_dir_size: float
|
||||
log_dir_size: float
|
||||
check_time: datetime
|
||||
Reference in New Issue
Block a user