TriM-Organization/LiteyukiBot-TriM
2
0
Fork 0
mirror of https://github.com/TriM-Organization/LiteyukiBot-TriM.git synced 2024-09-20 21:22:29 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat: 添加了对markdown的简单封装

Browse Source
This commit is contained in:
snowyServer 2024-03-20 12:30:17 +08:00
parent e351465d97
commit fdefedf288
6 changed files with 166 additions and 121 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

182
src/liteyuki_main/webdash.py
View File

@ -1,91 +1,91 @@
import nonebot
import psutil
from dash import Dash, Input, Output, dcc, html
from starlette.middleware.wsgi import WSGIMiddleware
from src.utils.language import Language
from src.utils.tools import convert_size
app = nonebot.get_app()
def get_system_info():
cpu_percent = psutil.cpu_percent()
memory_info = psutil.virtual_memory()
memory_percent = memory_info.percent
return {
"cpu_percent" : cpu_percent,
"memory_percent": memory_percent
}
@app.get("/system_info")
async def system_info():
return get_system_info()
lang = Language()
dash_app = Dash(__name__)
dash_app.layout = dash_app.layout = html.Div(children=[
html.H1(children=lang.get("main.monitor.title"), style={
'textAlign': 'center'
}),
dcc.Graph(id='live-update-graph'),
dcc.Interval(
id='interval-component',
interval=1 * 1000, # in milliseconds
n_intervals=0
)
])
@dash_app.callback(Output('live-update-graph', 'figure'),
[Input('interval-component', 'n_intervals')])
def update_graph_live(n):
lang = Language()
system_inf = get_system_info()
dash_app.layout = html.Div(children=[
html.H1(children=lang.get("main.monitor.title"), style={
'textAlign': 'center'
}),
dcc.Graph(id='live-update-graph'),
dcc.Interval(
id='interval-component',
interval=2 * 1000, # in milliseconds
n_intervals=0
)
])
mem = psutil.virtual_memory()
cpu_f = psutil.cpu_freq()
figure = {
'data' : [
{
'x' : [f"{cpu_f.current / 1000:.2f}GHz {psutil.cpu_count(logical=False)}c{psutil.cpu_count()}t"],
'y' : [system_inf['cpu_percent']],
'type': 'bar',
'name': f"{lang.get('main.monitor.cpu')} {lang.get('main.monitor.usage')}"
},
{
'x' : [f"{convert_size(mem.used, add_unit=False)}/{convert_size(mem.total)}({mem.used / mem.total * 100:.2f}%)"],
'y' : [system_inf['memory_percent']],
'type': 'bar',
'name': f"{lang.get('main.monitor.memory')} {lang.get('main.monitor.usage')}"
},
],
'layout': {
'title': lang.get('main.monitor.description'),
# 'xaxis': {
# 'range': [0, 10]
# }, # 设置x轴的范围
'yaxis': {
'range': [0, 100]
}, # 设置y轴的范围
}
}
return figure
app.mount("/", WSGIMiddleware(dash_app.server))
# import nonebot
# import psutil
# from dash import Dash, Input, Output, dcc, html
# from starlette.middleware.wsgi import WSGIMiddleware
#
# from src.utils.language import Language
# from src.utils.tools import convert_size
#
# app = nonebot.get_app()
#
#
# def get_system_info():
# cpu_percent = psutil.cpu_percent()
# memory_info = psutil.virtual_memory()
# memory_percent = memory_info.percent
# return {
# "cpu_percent" : cpu_percent,
# "memory_percent": memory_percent
# }
#
#
# @app.get("/system_info")
# async def system_info():
# return get_system_info()
#
#
# lang = Language()
# dash_app = Dash(__name__)
# dash_app.layout = dash_app.layout = html.Div(children=[
# html.H1(children=lang.get("main.monitor.title"), style={
# 'textAlign': 'center'
# }),
#
# dcc.Graph(id='live-update-graph'),
# dcc.Interval(
# id='interval-component',
# interval=1 * 1000, # in milliseconds
# n_intervals=0
# )
# ])
#
#
# @dash_app.callback(Output('live-update-graph', 'figure'),
# [Input('interval-component', 'n_intervals')])
# def update_graph_live(n):
# lang = Language()
# system_inf = get_system_info()
# dash_app.layout = html.Div(children=[
# html.H1(children=lang.get("main.monitor.title"), style={
# 'textAlign': 'center'
# }),
#
# dcc.Graph(id='live-update-graph'),
# dcc.Interval(
# id='interval-component',
# interval=2 * 1000, # in milliseconds
# n_intervals=0
# )
# ])
# mem = psutil.virtual_memory()
# cpu_f = psutil.cpu_freq()
# figure = {
# 'data' : [
# {
# 'x' : [f"{cpu_f.current / 1000:.2f}GHz {psutil.cpu_count(logical=False)}c{psutil.cpu_count()}t"],
# 'y' : [system_inf['cpu_percent']],
# 'type': 'bar',
# 'name': f"{lang.get('main.monitor.cpu')} {lang.get('main.monitor.usage')}"
#
# },
# {
# 'x' : [f"{convert_size(mem.used, add_unit=False)}/{convert_size(mem.total)}({mem.used / mem.total * 100:.2f}%)"],
# 'y' : [system_inf['memory_percent']],
# 'type': 'bar',
# 'name': f"{lang.get('main.monitor.memory')} {lang.get('main.monitor.usage')}"
# },
# ],
# 'layout': {
# 'title': lang.get('main.monitor.description'),
# # 'xaxis': {
# # 'range': [0, 10]
# # }, # 设置x轴的范围
# 'yaxis': {
# 'range': [0, 100]
# }, # 设置y轴的范围
# }
# }
# return figure
#
#
# app.mount("/", WSGIMiddleware(dash_app.server))

20
src/plugins/liteyuki_markdowntest.py
View File

@ -1,6 +1,7 @@
import nonebot
from nonebot import on_command
from nonebot import on_command, on_message
from nonebot.adapters.onebot.v11 import MessageSegment
from nonebot.exception import FinishedException
from nonebot.params import CommandArg
from nonebot.permission import SUPERUSER
@ -9,6 +10,8 @@ from src.utils.typing import T_Message, T_Bot, v11, T_MessageEvent
md_test = on_command("mdts", aliases={"会话md"}, permission=SUPERUSER)
md_group = on_command("mdg", aliases={"群md"}, permission=SUPERUSER)
md_conv = on_command("md",block=False, permission=SUPERUSER)
placeholder = {
"[": "[",
@ -27,4 +30,17 @@ async def _(bot: T_Bot, event: T_MessageEvent, arg: v11.Message = CommandArg()):
bot,
message_type=event.message_type,
session_id=event.user_id if event.message_type == "private" else event.group_id
)
)
ignore_msg_ids = []
last_sent = None
@md_conv.handle()
async def _(bot: v11.Bot, event: v11.MessageEvent, arg: v11.Message = CommandArg()):
if str(event.user_id) == str(bot.self_id) and str(bot.self_id) in ["2751454815"]:
nonebot.logger.info("开始处理：%s" % str(event.message_id))
data, fdata = await send_markdown(str(arg), bot, event.message_type,
event.group_id if event.message_type == "group" else event.user_id)
await bot.delete_msg(message_id=event.message_id)
nonebot.logger.info("已处理：%s, %s" % (str(data["message_id"]), str(fdata["message_id"])))

40
src/plugins/liteyuki_plugin_eventpush.py
View File

@ -7,8 +7,7 @@ from nonebot_plugin_alconna import on_alconna
from src.utils.data import LiteModel
from src.utils.message import send_markdown
from src.utils.typing import T_Bot, T_MessageEvent
pushes = []
from src.utils.data import Database
class Node(LiteModel):
@ -23,8 +22,12 @@ class Node(LiteModel):
class Push(LiteModel):
source: Node
target: Node
inde: int
pushes_db = Database("data/pushes.ldb")
pushes_db.auto_migrate(Push, Node)
alc = Alconna(
"lep",
Subcommand(
@ -55,17 +58,20 @@ async def _(result: Arparma):
if source and target:
source = source.split(".")
target = target.split(".")
pushes.append(Push(
push1 = Push(
source=Node(bot_id=source[0], session_type=source[1], session_id=source[2]),
target=Node(bot_id=target[0], session_type=target[1], session_id=target[2])
))
target=Node(bot_id=target[0], session_type=target[1], session_id=target[2]),
inde=len(pushes_db.all(Push, default=[]))
)
pushes_db.save(push1)
if result.subcommands["add"].args.get("-b"):
pushes.append(Push(
if result.subcommands["add"].args.get("bidirectional"):
push2 = Push(
source=Node(bot_id=target[0], session_type=target[1], session_id=target[2]),
target=Node(bot_id=source[0], session_type=source[1], session_id=source[2])
))
target=Node(bot_id=source[0], session_type=source[1], session_id=source[2]),
inde=len(pushes_db.all(Push, default=[]))
)
pushes_db.save(push2)
await add_push.finish("添加成功")
else:
await add_push.finish("参数缺失")
@ -73,27 +79,31 @@ async def _(result: Arparma):
index = result.subcommands["rm"].args.get("index")
if index is not None:
try:
pushes.pop(index)
pushes_db.delete(Push, "inde = ?", index)
await add_push.finish("删除成功")
except IndexError:
await add_push.finish("索引错误")
else:
await add_push.finish("参数缺失")
elif result.subcommands.get("list"):
await add_push.finish("\n".join([f"{i} {push.source.bot_id}.{push.source.session_type}.{push.source.session_id} -> "
f"{push.target.bot_id}.{push.target.session_type}.{push.target.session_id}" for i, push in enumerate(pushes)]))
await add_push.finish(
"\n".join([f"{push.inde} {push.source.bot_id}.{push.source.session_type}.{push.source.session_id} -> "
f"{push.target.bot_id}.{push.target.session_type}.{push.target.session_id}" for i, push in
enumerate(pushes_db.all(Push, default=[]))]))
else:
await add_push.finish("参数错误")
@on_message(block=False).handle()
async def _(event: T_MessageEvent, bot: T_Bot):
for push in pushes:
for push in pushes_db.all(Push, default=[]):
if str(push.source) == f"{bot.self_id}.{event.message_type}.{event.user_id if event.message_type == 'private' else event.group_id}":
bot2 = nonebot.get_bot(push.target.bot_id)
msg_formatted = ""
for l in str(event.message).split("\n"):
msg_formatted += f"**{l.strip()}**\n"
push_message = f"{msg_formatted}\n\n> From {event.sender.nickname}@{push.source.session_type}.{push.source.session_id}\n> Bot {bot.self_id}"
push_message = (
f"> From {event.sender.nickname}@{push.source.session_type}.{push.source.session_id}\n> Bot {bot.self_id}\n\n"
f"{msg_formatted}")
await send_markdown(push_message, bot2, push.target.session_type, push.target.session_id)
return

39
src/utils/data.py
View File

@ -4,10 +4,10 @@ import sqlite3
import types
from abc import ABC
from collections.abc import Iterable
from typing import Any
import nonebot
from pydantic import BaseModel
from typing_extensions import Any
BaseIterable = list | tuple | set | dict
@ -16,7 +16,7 @@ class LiteModel(BaseModel):
"""轻量级模型基类
类型注解统一使用Python3.9的PEP585标准，如需使用泛型请使用typing模块的泛型类型
"""
id: Any = None
id: int = None
class BaseORMAdapter(ABC):
@ -72,7 +72,7 @@ class BaseORMAdapter(ABC):
raise NotImplementedError
class SqliteORMDatabase(BaseORMAdapter):
class Database(BaseORMAdapter):
"""SQLiteORM适配器，严禁使用`FORIEGNID`和`JSON`作为主键前缀，严禁使用`$ID:`作为字符串值前缀
Attributes:
@ -86,6 +86,14 @@ class SqliteORMDatabase(BaseORMAdapter):
bool : 'INTEGER',
list : 'TEXT'
}
DEFAULT_TYPE = {
'TEXT': '',
'INTEGER': 0,
'REAL': 0.0
}
FOREIGNID = 'FOREIGNID'
JSON = 'JSON'
ID = '$ID'
@ -143,6 +151,8 @@ class SqliteORMDatabase(BaseORMAdapter):
if field not in table_fields:
nonebot.logger.debug(f'ALTER TABLE {table_name} ADD COLUMN {field} {type_}')
self.cursor.execute(f'ALTER TABLE {table_name} ADD COLUMN {field} {type_}')
# 在原有的行中添加新字段对应类型的默认值，从DEFAULT_TYPE中获取
self.cursor.execute(f'UPDATE {table_name} SET {field} = ? WHERE {field} IS NULL', (self.DEFAULT_TYPE.get(type_, ""),))
# 检测多余字段，除了id字段
for field in table_fields:
@ -230,11 +240,12 @@ class SqliteORMDatabase(BaseORMAdapter):
"""
table_name = model.__name__
self.cursor.execute(f'SELECT * FROM {table_name} WHERE {conditions}', args)
if data := self.cursor.fetchone():
if row_data := self.cursor.fetchone():
data = dict(row_data)
return model(**self.convert_to_dict(data))
return default
def all(self, model: type(LiteModel), conditions, *args, default: Any = None) -> list[LiteModel] | None:
def all(self, model: type(LiteModel), conditions=None, *args, default: Any = None) -> list[LiteModel] | None:
"""查询所有数据
Args:
@ -246,9 +257,17 @@ class SqliteORMDatabase(BaseORMAdapter):
Returns: 数据
"""
table_name = model.__name__
self.cursor.execute(f'SELECT * FROM {table_name} WHERE {conditions}', args)
data = self.cursor.fetchall()
return [model(**self.convert_to_dict(d)) for d in data] if data else default
# 检测表是否存在，否则返回None
if conditions:
self.cursor.execute(f'SELECT * FROM {table_name} WHERE {conditions}', args)
else:
self.cursor.execute(f'SELECT * FROM {table_name}')
if row_datas := self.cursor.fetchall():
datas = [dict(row_data) for row_data in row_datas]
return [model(**self.convert_to_dict(d)) for d in datas] if datas else default
return default
def delete(self, model: type(LiteModel), conditions, *args):
"""删除数据
@ -299,7 +318,7 @@ class SqliteORMDatabase(BaseORMAdapter):
for k, v in d.items():
if k.startswith(self.FOREIGNID):
new_d[k.replace(self.FOREIGNID, '')] = load(
dict(self.cursor.execute(f'SELECT * FROM {v.split(":")[1]} WHERE id = ?', (v.split(":")[2],)).fetchone()))
dict(self.cursor.execute(f'SELECT * FROM {v.split(":",2)[1]} WHERE id = ?', (v.split(":",2)[2],)).fetchone()))
elif k.startswith(self.JSON):
new_d[k.replace(self.JSON, '')] = load(json.loads(v))
else:
@ -308,7 +327,7 @@ class SqliteORMDatabase(BaseORMAdapter):
new_d = []
for i, v in enumerate(d):
if isinstance(v, str) and v.startswith(self.ID):
new_d.append(load(dict(self.cursor.execute(f'SELECT * FROM {v.split(":")[1]} WHERE id = ?', (v.split(":")[2],)).fetchone())))
new_d.append(load(dict(self.cursor.execute(f'SELECT * FROM {v.split(":",2)[1]} WHERE id = ?', (v.split(":",2)[2],)).fetchone())))
elif isinstance(v, BaseIterable):
new_d.append(load(v))
else:

2
src/utils/data_manager.py
View File

@ -1,6 +1,6 @@
import os
from src.utils.data import LiteModel, SqliteORMDatabase as DB
from src.utils.data import LiteModel, Database as DB
DATA_PATH = "data/liteyuki"

4
src/utils/message.py
View File

@ -6,7 +6,7 @@ from .tools import de_escape
from .typing import T_Bot
async def send_markdown(markdown: str, bot: T_Bot, message_type: str, session_id: str | int) -> dict[str, Any]:
async def send_markdown(markdown: str, bot: T_Bot, message_type: str, session_id: str | int) -> tuple[dict[str, Any], dict[str, Any]]:
formatted_md = de_escape(markdown).replace("\n", r"\n").replace("\"", r'\\\"')
try:
forward_data = await bot.call_api(
@ -67,4 +67,4 @@ async def send_markdown(markdown: str, bot: T_Bot, message_type: str, session_id
else:
nonebot.logger.error("send_markdown: bot type not supported")
data = {}
return data
return data, forward_data