Diuse/zhenxun_bot
1
0
Fork 0
mirror of https://github.com/zhenxun-org/zhenxun_bot.git synced 2025-12-14 21:52:56 +08:00
Code Issues Actions 7 Packages Projects Releases Wiki Activity

🎨 代码优化 (#1698)

Browse Source
This commit is contained in:
HibiKier 2024-10-15 03:44:30 +08:00 committed by GitHub
parent 4a52c4825b
commit fe1634eb2d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 352 additions and 129 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
zhenxun/builtin_plugins/web_ui/api/tabs/dashboard/__init__.py
View File

@ -4,6 +4,7 @@ from nonebot import require
from fastapi import APIRouter
from tortoise.functions import Count
from tortoise.expressions import RawSQL
from fastapi.responses import JSONResponse
from zhenxun.models.statistics import Statistics
from zhenxun.models.chat_history import ChatHistory
@ -12,7 +13,7 @@ from zhenxun.models.bot_connect_log import BotConnectLog
from .data_source import BotManage
from ....utils import authentication
from ....base_model import Result, QueryModel, BaseResultModel
from .model import ChatCallMonthCount, QueryChatCallCount, AllChatAndCallCount
from .model import BotInfo, ChatCallMonthCount, QueryChatCallCount, AllChatAndCallCount
require("plugin_store")
@ -22,9 +23,11 @@ router = APIRouter(prefix="/dashboard")
@router.get(
"/get_bot_list",
dependencies=[authentication()],
response_model=Result[list[BotInfo]],
response_class=JSONResponse,
deprecated="获取bot列表", # type: ignore
)
async def _() -> Result:
async def _() -> Result[list[BotInfo]]:
try:
return Result.ok(await BotManage.get_bot_list(), "拿到信息啦!")
except Exception as e:
@ -34,9 +37,11 @@ async def _() -> Result:
@router.get(
"/get_chat_and_call_count",
dependencies=[authentication()],
response_model=Result[QueryChatCallCount],
response_class=JSONResponse,
description="获取聊天/调用记录的全部和今日数量",
)
async def _(bot_id: str | None = None) -> Result:
async def _(bot_id: str | None = None) -> Result[QueryChatCallCount]:
now = datetime.now()
query = ChatHistory
if bot_id:
@ -65,9 +70,11 @@ async def _(bot_id: str | None = None) -> Result:
@router.get(
"/get_all_chat_and_call_count",
dependencies=[authentication()],
response_model=Result[AllChatAndCallCount],
response_class=JSONResponse,
description="获取聊天/调用记录的全部数据次数",
)
async def _(bot_id: str | None = None) -> Result:
async def _(bot_id: str | None = None) -> Result[AllChatAndCallCount]:
now = datetime.now()
query = ChatHistory
if bot_id:
@ -108,9 +115,11 @@ async def _(bot_id: str | None = None) -> Result:
@router.get(
"/get_chat_and_call_month",
dependencies=[authentication()],
response_model=Result[ChatCallMonthCount],
response_class=JSONResponse,
deprecated="获取聊天/调用记录的一个月数量", # type: ignore
)
async def _(bot_id: str | None = None) -> Result:
async def _(bot_id: str | None = None) -> Result[ChatCallMonthCount]:
now = datetime.now()
filter_date = now - timedelta(days=30, hours=now.hour, minutes=now.minute)
chat_query = ChatHistory
@ -158,9 +167,11 @@ async def _(bot_id: str | None = None) -> Result:
@router.post(
"/get_connect_log",
dependencies=[authentication()],
response_model=Result[BaseResultModel],
response_class=JSONResponse,
deprecated="获取Bot连接记录", # type: ignore
)
async def _(query: QueryModel) -> Result:
async def _(query: QueryModel) -> Result[BaseResultModel]:
total = await BotConnectLog.all().count()
if total % query.size:
total += 1

47
zhenxun/builtin_plugins/web_ui/api/tabs/database/__init__.py
View File

@ -2,6 +2,7 @@ import nonebot
from tortoise import Tortoise
from nonebot.drivers import Driver
from fastapi import Request, APIRouter
from fastapi.responses import JSONResponse
from tortoise.exceptions import OperationalError
from zhenxun.models.task_info import TaskInfo
@ -62,26 +63,40 @@ async def _():
@router.get(
"/get_table_list", dependencies=[authentication()], description="获取数据库表"
"/get_table_list",
dependencies=[authentication()],
response_model=Result[list[dict]],
response_class=JSONResponse,
description="获取数据库表",
)
async def _() -> Result:
async def _() -> Result[list[dict]]:
db = Tortoise.get_connection("default")
query = await db.execute_query_dict(SELECT_TABLE_SQL)
return Result.ok(query)
@router.get(
"/get_table_column", dependencies=[authentication()], description="获取表字段"
"/get_table_column",
dependencies=[authentication()],
response_model=Result[list[dict]],
response_class=JSONResponse,
description="获取表字段",
)
async def _(table_name: str) -> Result:
async def _(table_name: str) -> Result[list[dict]]:
db = Tortoise.get_connection("default")
# print(SELECT_TABLE_COLUMN_SQL.format(table_name))
query = await db.execute_query_dict(SELECT_TABLE_COLUMN_SQL.format(table_name))
return Result.ok(query)
@router.post("/exec_sql", dependencies=[authentication()], description="执行sql")
async def _(sql: SqlText, request: Request) -> Result:
@router.post(
"/exec_sql",
dependencies=[authentication()],
response_model=Result[list[dict]],
response_class=JSONResponse,
description="执行sql",
)
async def _(sql: SqlText, request: Request) -> Result[list[dict]]:
ip = request.client.host if request.client else "unknown"
try:
if sql.sql.lower().startswith("select"):
@ -98,8 +113,14 @@ async def _(sql: SqlText, request: Request) -> Result:
return Result.warning_(f"sql执行错误: {e}")
@router.post("/get_sql_log", dependencies=[authentication()], description="sql日志列表")
async def _(query: QueryModel) -> Result:
@router.post(
"/get_sql_log",
dependencies=[authentication()],
response_model=Result[BaseResultModel],
response_class=JSONResponse,
description="sql日志列表",
)
async def _(query: QueryModel) -> Result[BaseResultModel]:
total = await SqlLog.all().count()
if total % query.size:
total += 1
@ -112,8 +133,14 @@ async def _(query: QueryModel) -> Result:
return Result.ok(BaseResultModel(total=total, data=data))
@router.get("/get_common_sql", dependencies=[authentication()], description="常用sql")
async def _(plugin_name: str | None = None) -> Result:
@router.get(
"/get_common_sql",
dependencies=[authentication()],
response_model=Result[dict],
response_class=JSONResponse,
description="常用sql",
)
async def _(plugin_name: str | None = None) -> Result[dict]:
if plugin_name:
return Result.ok(SQL_DICT.get(plugin_name))
return Result.ok(str(SQL_DICT))

134
zhenxun/builtin_plugins/web_ui/api/tabs/main/__init__.py
View File

@ -6,7 +6,9 @@ from datetime import datetime, timedelta
import nonebot
from fastapi import APIRouter
from nonebot.config import Config
from tortoise.functions import Count
from fastapi.responses import JSONResponse
from websockets.exceptions import ConnectionClosedOK, ConnectionClosedError
from starlette.websockets import WebSocket, WebSocketState, WebSocketDisconnect
@ -22,7 +24,14 @@ from ....base_model import Result
from .data_source import bot_live
from ....utils import authentication, get_system_status
from ....config import AVA_URL, GROUP_AVA_URL, QueryDateType
from .model import BaseInfo, HotPlugin, QueryCount, ActiveGroup, NonebotData
from .model import (
BaseInfo,
HotPlugin,
QueryCount,
ActiveGroup,
NonebotData,
TemplateBaseInfo,
)
driver = nonebot.get_driver()
run_time = time.time()
@ -31,8 +40,14 @@ ws_router = APIRouter()
router = APIRouter(prefix="/main")
@router.get("/get_base_info", dependencies=[authentication()], description="基础信息")
async def _(bot_id: str | None = None) -> Result:
@router.get(
"/get_base_info",
dependencies=[authentication()],
response_model=Result[list[BaseInfo]],
response_class=JSONResponse,
description="基础信息",
)
async def _(bot_id: str | None = None) -> Result[list[BaseInfo]]:
"""获取Bot基础信息
参数:
@ -42,13 +57,13 @@ async def _(bot_id: str | None = None) -> Result:
Result: 获取指定bot信息与bot列表
"""
global run_time
bot_list: list[BaseInfo] = []
bot_list: list[TemplateBaseInfo] = []
if bots := nonebot.get_bots():
select_bot: BaseInfo
for key, bot in bots.items():
for _, bot in bots.items():
login_info = await bot.get_login_info()
bot_list.append(
BaseInfo(
TemplateBaseInfo(
bot=bot, # type: ignore
self_id=bot.self_id,
nickname=login_info["nickname"],
@ -93,18 +108,23 @@ async def _(bot_id: str | None = None) -> Result:
day_call = await Statistics.filter(
create_time__gte=now - timedelta(hours=now.hour)
).count()
select_bot.config = driver.config
select_bot.day_call = day_call
select_bot.connect_count = await BotConnectLog.filter(
bot_id=select_bot.self_id
).count()
return Result.ok(bot_list, "拿到信息啦!")
return Result.ok([BaseInfo(**e.dict()) for e in bot_list], "拿到信息啦!")
return Result.warning_("无Bot连接...")
@router.get(
"/get_all_ch_count", dependencies=[authentication()], description="获取接收消息数量"
"/get_all_chat_count",
dependencies=[authentication()],
response_model=Result[QueryCount],
response_class=JSONResponse,
description="获取接收消息数量",
)
async def _(bot_id: str | None = None) -> Result:
async def _(bot_id: str | None = None) -> Result[QueryCount]:
now = datetime.now()
query = ChatHistory
if bot_id:
@ -134,9 +154,13 @@ async def _(bot_id: str | None = None) -> Result:
@router.get(
"/get_all_call_count", dependencies=[authentication()], description="获取调用次数"
"/get_all_call_count",
dependencies=[authentication()],
response_model=Result[QueryCount],
response_class=JSONResponse,
description="获取调用次数",
)
async def _(bot_id: str | None = None) -> Result:
async def _(bot_id: str | None = None) -> Result[QueryCount]:
now = datetime.now()
query = Statistics
if bot_id:
@ -166,44 +190,13 @@ async def _(bot_id: str | None = None) -> Result:
@router.get(
"/get_ch_count", dependencies=[authentication()], description="获取接收消息数量"
"get_fg_count",
dependencies=[authentication()],
response_model=Result[dict[str, int]],
response_class=JSONResponse,
description="好友/群组数量",
)
async def _(bot_id: str, query_type: QueryDateType | None = None) -> Result:
if nonebot.get_bot(bot_id):
if not query_type:
return Result.ok(await ChatHistory.filter(bot_id=bot_id).count())
now = datetime.now()
if query_type == QueryDateType.DAY:
return Result.ok(
await ChatHistory.filter(
bot_id=bot_id, create_time__gte=now - timedelta(hours=now.hour)
).count()
)
if query_type == QueryDateType.WEEK:
return Result.ok(
await ChatHistory.filter(
bot_id=bot_id, create_time__gte=now - timedelta(days=7)
).count()
)
if query_type == QueryDateType.MONTH:
return Result.ok(
await ChatHistory.filter(
bot_id=bot_id, create_time__gte=now - timedelta(days=30)
).count()
)
if query_type == QueryDateType.YEAR:
return Result.ok(
await ChatHistory.filter(
bot_id=bot_id, create_time__gte=now - timedelta(days=365)
).count()
)
return Result.warning_("无Bot连接...")
@router.get(
"get_fg_count", dependencies=[authentication()], description="好友/群组数量"
)
async def _(bot_id: str) -> Result:
async def _(bot_id: str) -> Result[dict[str, int]]:
if bots := nonebot.get_bots():
if bot_id not in bots:
return Result.warning_("指定Bot未连接...")
@ -219,29 +212,49 @@ async def _(bot_id: str) -> Result:
return Result.warning_("无Bot连接...")
@router.get("/get_nb_data", dependencies=[authentication()], description="获取nb数据")
async def _() -> Result:
@router.get(
"/get_nb_data",
dependencies=[authentication()],
response_model=Result[NonebotData],
response_class=JSONResponse,
description="获取nb数据",
)
async def _() -> Result[NonebotData]:
return Result.ok(NonebotData(config=driver.config, run_time=int(run_time)))
@router.get("/get_nb_config", dependencies=[authentication()], description="获取nb配置")
async def _() -> Result:
@router.get(
"/get_nb_config",
dependencies=[authentication()],
response_model=Result[Config],
response_class=JSONResponse,
description="获取nb配置",
)
async def _() -> Result[Config]:
return Result.ok(driver.config)
@router.get(
"/get_run_time", dependencies=[authentication()], description="获取nb运行时间"
"/get_run_time",
dependencies=[authentication()],
response_model=Result[int],
response_class=JSONResponse,
description="获取nb运行时间",
)
async def _() -> Result:
async def _() -> Result[int]:
return Result.ok(int(run_time))
@router.get(
"/get_active_group", dependencies=[authentication()], description="获取活跃群聊"
"/get_active_group",
dependencies=[authentication()],
response_model=Result[list[ActiveGroup]],
response_class=JSONResponse,
description="获取活跃群聊",
)
async def _(
date_type: QueryDateType | None = None, bot_id: str | None = None
) -> Result:
) -> Result[list[ActiveGroup]]:
query = ChatHistory
now = datetime.now()
if bot_id:
@ -287,11 +300,15 @@ async def _(
@router.get(
"/get_hot_plugin", dependencies=[authentication()], description="获取热门插件"
"/get_hot_plugin",
dependencies=[authentication()],
response_model=Result[list[HotPlugin]],
response_class=JSONResponse,
description="获取热门插件",
)
async def _(
date_type: QueryDateType | None = None, bot_id: str | None = None
) -> Result:
) -> Result[list[HotPlugin]]:
query = Statistics
now = datetime.now()
if bot_id:
@ -336,4 +353,3 @@ async def system_logs_realtime(websocket: WebSocket, sleep: int = 5):
system_status = await get_system_status()
await websocket.send_text(system_status.json())
await asyncio.sleep(sleep)
return

14
zhenxun/builtin_plugins/web_ui/api/tabs/main/model.py
View File

@ -18,8 +18,6 @@ class BaseInfo(BaseModel):
基础信息
"""
bot: Bot
"""Bot"""
self_id: str
"""SELF ID"""
nickname: str
@ -48,7 +46,8 @@ class BaseInfo(BaseModel):
is_select: bool = False
"""当前选择"""
config: Config | None
"""nb配置"""
day_call: int = 0
"""今日调用插件次数"""
version: str = "unknown"
@ -58,6 +57,15 @@ class BaseInfo(BaseModel):
arbitrary_types_allowed = True
class TemplateBaseInfo(BaseInfo):
"""
基础信息
"""
bot: Bot
"""bot"""
class QueryCount(BaseModel):
"""
聊天记录数量

105
zhenxun/builtin_plugins/web_ui/api/tabs/manage/__init__.py
View File

@ -1,6 +1,7 @@
import nonebot
from fastapi import APIRouter
from tortoise.functions import Count
from fastapi.responses import JSONResponse
from nonebot.adapters.onebot.v11 import ActionFailed
from zhenxun.services.log import logger
@ -41,7 +42,11 @@ router = APIRouter(prefix="/manage")
@router.get(
"/get_group_list", dependencies=[authentication()], description="获取群组列表"
"/get_group_list",
dependencies=[authentication()],
response_model=Result[list[GroupResult]],
response_class=JSONResponse,
description="获取群组列表",
)
async def _(bot_id: str) -> Result:
"""
@ -65,9 +70,13 @@ async def _(bot_id: str) -> Result:
@router.post(
"/update_group", dependencies=[authentication()], description="修改群组信息"
"/update_group",
dependencies=[authentication()],
response_model=Result[str],
response_class=JSONResponse,
description="修改群组信息",
)
async def _(group: UpdateGroup) -> Result:
async def _(group: UpdateGroup) -> Result[str]:
try:
group_id = group.group_id
if db_group := await GroupConsole.get_group(group_id):
@ -91,9 +100,13 @@ async def _(group: UpdateGroup) -> Result:
@router.get(
"/get_friend_list", dependencies=[authentication()], description="获取好友列表"
"/get_friend_list",
dependencies=[authentication()],
response_model=Result[list[Friend]],
response_class=JSONResponse,
description="获取好友列表",
)
async def _(bot_id: str) -> Result:
async def _(bot_id: str) -> Result[list[Friend]]:
"""
获取群信息
"""
@ -118,9 +131,13 @@ async def _(bot_id: str) -> Result:
@router.get(
"/get_request_count", dependencies=[authentication()], description="获取请求数量"
"/get_request_count",
dependencies=[authentication()],
response_model=Result[dict[str, int]],
response_class=JSONResponse,
description="获取请求数量",
)
async def _() -> Result:
async def _() -> Result[dict[str, int]]:
f_count = await FgRequest.filter(
request_type=RequestType.FRIEND, handle_type__isnull=True
).count()
@ -135,9 +152,13 @@ async def _() -> Result:
@router.get(
"/get_request_list", dependencies=[authentication()], description="获取请求列表"
"/get_request_list",
dependencies=[authentication()],
response_model=Result[ReqResult],
response_class=JSONResponse,
description="获取请求列表",
)
async def _() -> Result:
async def _() -> Result[ReqResult]:
try:
req_result = ReqResult()
data_list = await FgRequest.filter(handle_type__isnull=True).all()
@ -179,7 +200,11 @@ async def _() -> Result:
@router.post(
"/clear_request", dependencies=[authentication()], description="清空请求列表"
"/clear_request",
dependencies=[authentication()],
response_model=Result,
response_class=JSONResponse,
description="清空请求列表",
)
async def _(cr: ClearRequest) -> Result:
await FgRequest.filter(
@ -188,7 +213,13 @@ async def _(cr: ClearRequest) -> Result:
return Result.ok(info="成功清除了数据!")
@router.post("/refuse_request", dependencies=[authentication()], description="拒绝请求")
@router.post(
"/refuse_request",
dependencies=[authentication()],
response_model=Result,
response_class=JSONResponse,
description="拒绝请求",
)
async def _(parma: HandleRequest) -> Result:
try:
if bots := nonebot.get_bots():
@ -209,14 +240,24 @@ async def _(parma: HandleRequest) -> Result:
return Result.fail(f"{type(e)}: {e}")
@router.post("/delete_request", dependencies=[authentication()], description="忽略请求")
@router.post(
"/delete_request",
dependencies=[authentication()],
response_model=Result,
response_class=JSONResponse,
description="忽略请求",
)
async def _(parma: HandleRequest) -> Result:
await FgRequest.ignore(parma.id)
return Result.ok(info="成功处理了请求!")
@router.post(
"/approve_request", dependencies=[authentication()], description="同意请求"
"/approve_request",
dependencies=[authentication()],
response_model=Result,
response_class=JSONResponse,
description="同意请求",
)
async def _(parma: HandleRequest) -> Result:
try:
@ -247,7 +288,13 @@ async def _(parma: HandleRequest) -> Result:
return Result.fail(f"{type(e)}: {e}")
@router.post("/leave_group", dependencies=[authentication()], description="退群")
@router.post(
"/leave_group",
dependencies=[authentication()],
response_model=Result,
response_class=JSONResponse,
description="退群",
)
async def _(param: LeaveGroup) -> Result:
try:
if bots := nonebot.get_bots():
@ -266,7 +313,13 @@ async def _(param: LeaveGroup) -> Result:
return Result.fail(f"{type(e)}: {e}")
@router.post("/delete_friend", dependencies=[authentication()], description="删除好友")
@router.post(
"/delete_friend",
dependencies=[authentication()],
response_model=Result,
response_class=JSONResponse,
description="删除好友",
)
async def _(param: DeleteFriend) -> Result:
try:
if bots := nonebot.get_bots():
@ -286,9 +339,13 @@ async def _(param: DeleteFriend) -> Result:
@router.get(
"/get_friend_detail", dependencies=[authentication()], description="获取好友详情"
"/get_friend_detail",
dependencies=[authentication()],
response_model=Result[UserDetail],
response_class=JSONResponse,
description="获取好友详情",
)
async def _(bot_id: str, user_id: str) -> Result:
async def _(bot_id: str, user_id: str) -> Result[UserDetail]:
if bots := nonebot.get_bots():
if bot_id in bots:
if fd := [
@ -329,9 +386,13 @@ async def _(bot_id: str, user_id: str) -> Result:
@router.get(
"/get_group_detail", dependencies=[authentication()], description="获取群组详情"
"/get_group_detail",
dependencies=[authentication()],
response_model=Result[GroupDetail],
response_class=JSONResponse,
description="获取群组详情",
)
async def _(bot_id: str, group_id: str) -> Result:
async def _(bot_id: str, group_id: str) -> Result[GroupDetail]:
if not (bots := nonebot.get_bots()):
return Result.warning_("无Bot连接...")
if bot_id not in bots:
@ -418,7 +479,11 @@ async def _(bot_id: str, group_id: str) -> Result:
@router.post(
"/send_message", dependencies=[authentication()], description="获取群组详情"
"/send_message",
dependencies=[authentication()],
response_model=Result,
response_class=JSONResponse,
description="获取群组详情",
)
async def _(param: SendMessage) -> Result:
if not (bots := nonebot.get_bots()):

41
zhenxun/builtin_plugins/web_ui/api/tabs/plugin_manage/__init__.py
View File

@ -2,6 +2,7 @@ import re
import cattrs
from fastapi import Query, APIRouter
from fastapi.responses import JSONResponse
from zhenxun.services.log import logger
from zhenxun.configs.config import Config
@ -25,11 +26,13 @@ router = APIRouter(prefix="/plugin")
@router.get(
"/get_plugin_list",
dependencies=[authentication()],
response_model=Result[list[PluginInfo]],
response_class=JSONResponse,
deprecated="获取插件列表", # type: ignore
)
async def _(
plugin_type: list[PluginType] = Query(None), menu_type: str | None = None
) -> Result:
) -> Result[list[PluginInfo]]:
try:
plugin_list: list[PluginInfo] = []
query = DbPluginInfo
@ -61,9 +64,11 @@ async def _(
@router.get(
"/get_plugin_count",
dependencies=[authentication()],
response_model=Result[int],
response_class=JSONResponse,
deprecated="获取插件数量", # type: ignore
)
async def _() -> Result:
async def _() -> Result[int]:
plugin_count = PluginCount()
plugin_count.normal = await DbPluginInfo.filter(
plugin_type=PluginType.NORMAL, load_status=True
@ -82,7 +87,11 @@ async def _() -> Result:
@router.post(
"/update_plugin", dependencies=[authentication()], description="更新插件参数"
"/update_plugin",
dependencies=[authentication()],
response_model=Result,
response_class=JSONResponse,
description="更新插件参数",
)
async def _(plugin: UpdatePlugin) -> Result:
try:
@ -113,7 +122,13 @@ async def _(plugin: UpdatePlugin) -> Result:
return Result.ok(info="已经帮你写好啦!")
@router.post("/change_switch", dependencies=[authentication()], description="开关插件")
@router.post(
"/change_switch",
dependencies=[authentication()],
response_model=Result,
response_class=JSONResponse,
description="开关插件",
)
async def _(param: PluginSwitch) -> Result:
db_plugin = await DbPluginInfo.get_or_none(module=param.module, load_status=True)
if not db_plugin:
@ -129,9 +144,13 @@ async def _(param: PluginSwitch) -> Result:
@router.get(
"/get_plugin_menu_type", dependencies=[authentication()], description="获取插件类型"
"/get_plugin_menu_type",
dependencies=[authentication()],
response_model=Result[list[str]],
response_class=JSONResponse,
description="获取插件类型",
)
async def _() -> Result:
async def _() -> Result[list[str]]:
menu_type_list = []
result = await DbPluginInfo.annotate().values_list("menu_type", flat=True)
for r in result:
@ -140,8 +159,14 @@ async def _() -> Result:
return Result.ok(menu_type_list)
@router.get("/get_plugin", dependencies=[authentication()], description="获取插件详情")
async def _(module: str) -> Result:
@router.get(
"/get_plugin",
dependencies=[authentication()],
response_model=Result[PluginDetail],
response_class=JSONResponse,
description="获取插件详情",
)
async def _(module: str) -> Result[PluginDetail]:
db_plugin = await DbPluginInfo.get_or_none(module=module, load_status=True)
if not db_plugin:
return Result.fail("插件不存在...")

11
zhenxun/builtin_plugins/web_ui/api/tabs/plugin_manage/store.py
View File

@ -1,5 +1,6 @@
from nonebot import require
from fastapi import APIRouter
from fastapi.responses import JSONResponse
from zhenxun.models.plugin_info import PluginInfo
@ -13,9 +14,11 @@ router = APIRouter(prefix="/store")
@router.get(
"/get_plugin_store",
dependencies=[authentication()],
response_model=Result[dict],
response_class=JSONResponse,
deprecated="获取插件商店插件信息", # type: ignore
)
async def _() -> Result:
async def _() -> Result[dict]:
try:
require("plugin_store")
from zhenxun.builtin_plugins.plugin_store import ShopManage
@ -36,6 +39,8 @@ async def _() -> Result:
@router.post(
"/install_plugin",
dependencies=[authentication()],
response_model=Result,
response_class=JSONResponse,
deprecated="安装插件", # type: ignore
)
async def _(param: PluginIr) -> Result:
@ -52,6 +57,8 @@ async def _(param: PluginIr) -> Result:
@router.post(
"/update_plugin",
dependencies=[authentication()],
response_model=Result,
response_class=JSONResponse,
deprecated="更新插件", # type: ignore
)
async def _(param: PluginIr) -> Result:
@ -68,6 +75,8 @@ async def _(param: PluginIr) -> Result:
@router.post(
"/remove_plugin",
dependencies=[authentication()],
response_model=Result,
response_class=JSONResponse,
deprecated="移除插件", # type: ignore
)
async def _(param: PluginIr) -> Result:

91
zhenxun/builtin_plugins/web_ui/api/tabs/system/__init__.py
View File

@ -4,10 +4,11 @@ from pathlib import Path
import aiofiles
from fastapi import APIRouter
from fastapi.responses import JSONResponse
from zhenxun.utils._build_image import BuildImage
from ....base_model import Result
from ....base_model import Result, SystemFolderSize
from ....utils import authentication, get_system_disk
from .model import AddFile, DirFile, SaveFile, DeleteFile, RenameFile
@ -17,9 +18,13 @@ IMAGE_TYPE = ["jpg", "jpeg", "png", "gif", "bmp", "webp", "svg"]
@router.get(
"/get_dir_list", dependencies=[authentication()], description="获取文件列表"
"/get_dir_list",
dependencies=[authentication()],
response_model=Result[list[DirFile]],
response_class=JSONResponse,
description="获取文件列表",
)
async def _(path: str | None = None) -> Result:
async def _(path: str | None = None) -> Result[list[DirFile]]:
base_path = Path(path) if path else Path()
data_list = []
for file in os.listdir(base_path):
@ -37,13 +42,23 @@ async def _(path: str | None = None) -> Result:
@router.get(
"/get_resources_size", dependencies=[authentication()], description="获取文件列表"
"/get_resources_size",
dependencies=[authentication()],
response_model=Result[SystemFolderSize],
response_class=JSONResponse,
description="获取文件列表",
)
async def _(full_path: str | None = None) -> Result:
async def _(full_path: str | None = None) -> Result[SystemFolderSize]:
return Result.ok(await get_system_disk(full_path))
@router.post("/delete_file", dependencies=[authentication()], description="删除文件")
@router.post(
"/delete_file",
dependencies=[authentication()],
response_model=Result,
response_class=JSONResponse,
description="删除文件",
)
async def _(param: DeleteFile) -> Result:
path = Path(param.full_path)
if not path or not path.exists():
@ -56,7 +71,11 @@ async def _(param: DeleteFile) -> Result:
@router.post(
"/delete_folder", dependencies=[authentication()], description="删除文件夹"
"/delete_folder",
dependencies=[authentication()],
response_model=Result,
response_class=JSONResponse,
description="删除文件夹",
)
async def _(param: DeleteFile) -> Result:
path = Path(param.full_path)
@ -69,7 +88,13 @@ async def _(param: DeleteFile) -> Result:
return Result.warning_(f"删除失败: {e!s}")
@router.post("/rename_file", dependencies=[authentication()], description="重命名文件")
@router.post(
"/rename_file",
dependencies=[authentication()],
response_model=Result,
response_class=JSONResponse,
description="重命名文件",
)
async def _(param: RenameFile) -> Result:
path = (
(Path(param.parent) / param.old_name) if param.parent else Path(param.old_name)
@ -84,7 +109,11 @@ async def _(param: RenameFile) -> Result:
@router.post(
"/rename_folder", dependencies=[authentication()], description="重命名文件夹"
"/rename_folder",
dependencies=[authentication()],
response_model=Result,
response_class=JSONResponse,
description="重命名文件夹",
)
async def _(param: RenameFile) -> Result:
path = (
@ -100,7 +129,13 @@ async def _(param: RenameFile) -> Result:
return Result.warning_(f"重命名失败: {e!s}")
@router.post("/add_file", dependencies=[authentication()], description="新建文件")
@router.post(
"/add_file",
dependencies=[authentication()],
response_model=Result,
response_class=JSONResponse,
description="新建文件",
)
async def _(param: AddFile) -> Result:
path = (Path(param.parent) / param.name) if param.parent else Path(param.name)
if path.exists():
@ -112,7 +147,13 @@ async def _(param: AddFile) -> Result:
return Result.warning_(f"新建文件失败: {e!s}")
@router.post("/add_folder", dependencies=[authentication()], description="新建文件夹")
@router.post(
"/add_folder",
dependencies=[authentication()],
response_model=Result,
response_class=JSONResponse,
description="新建文件夹",
)
async def _(param: AddFile) -> Result:
path = (Path(param.parent) / param.name) if param.parent else Path(param.name)
if path.exists():
@ -124,7 +165,13 @@ async def _(param: AddFile) -> Result:
return Result.warning_(f"新建文件夹失败: {e!s}")
@router.get("/read_file", dependencies=[authentication()], description="读取文件")
@router.get(
"/read_file",
dependencies=[authentication()],
response_model=Result[str],
response_class=JSONResponse,
description="读取文件",
)
async def _(full_path: str) -> Result:
path = Path(full_path)
if not path.exists():
@ -136,8 +183,14 @@ async def _(full_path: str) -> Result:
return Result.warning_(f"读取文件失败: {e!s}")
@router.post("/save_file", dependencies=[authentication()], description="读取文件")
async def _(param: SaveFile) -> Result:
@router.post(
"/save_file",
dependencies=[authentication()],
response_model=Result[str],
response_class=JSONResponse,
description="读取文件",
)
async def _(param: SaveFile) -> Result[str]:
path = Path(param.full_path)
try:
async with aiofiles.open(path, "w", encoding="utf-8") as f:
@ -147,8 +200,14 @@ async def _(param: SaveFile) -> Result:
return Result.warning_(f"保存文件失败: {e!s}")
@router.get("/get_image", dependencies=[authentication()], description="读取图片base64")
async def _(full_path: str) -> Result:
@router.get(
"/get_image",
dependencies=[authentication()],
response_model=Result[str],
response_class=JSONResponse,
description="读取图片base64",
)
async def _(full_path: str) -> Result[str]:
path = Path(full_path)
if not path.exists():
return Result.warning_("文件不存在...")

15
zhenxun/builtin_plugins/web_ui/base_model.py
View File

@ -1,11 +1,12 @@
from datetime import datetime
from typing_extensions import Self
from typing import Any, Generic, TypeVar
from pydantic import BaseModel, validator
T = TypeVar("T")
RT = TypeVar("RT")
class User(BaseModel):
username: str
@ -17,7 +18,7 @@ class Token(BaseModel):
token_type: str
class Result(BaseModel):
class Result(Generic[RT], BaseModel):
"""
总体返回
"""
@ -30,19 +31,21 @@ class Result(BaseModel):
"""info"""
warning: str | None = None
"""警告信息"""
data: Any = None
data: RT = None
"""返回数据"""
@classmethod
def warning_(cls, info: str, code: int = 200) -> Self:
def warning_(cls, info: str, code: int = 200) -> "Result[RT]":
return cls(suc=True, warning=info, code=code)
@classmethod
def fail(cls, info: str = "异常错误", code: int = 500) -> Self:
def fail(cls, info: str = "异常错误", code: int = 500) -> "Result[RT]":
return cls(suc=False, info=info, code=code)
@classmethod
def ok(cls, data: Any = None, info: str = "操作成功", code: int = 200) -> Self:
def ok(
cls, data: Any = None, info: str = "操作成功", code: int = 200
) -> "Result[RT]":
return cls(suc=True, info=info, code=code, data=data)