TriM-Organization/LiteyukiBot-TriM
2
0
Fork 0
mirror of https://github.com/TriM-Organization/LiteyukiBot-TriM.git synced 2024-11-11 01:27:29 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Core适配satori

Browse Source
This commit is contained in:
Expliyh 2024-05-16 20:09:20 +08:00
parent 24722447da
commit 251bfaf410
No known key found for this signature in database
GPG Key ID: F30DE0CFF6157916
5 changed files with 123 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
liteyuki/liteyuki_main/core.py
View File

@ -37,6 +37,7 @@ markdown_image = common_db.where_one(StoredConfig(), default=StoredConfig()).con
),
permission=SUPERUSER
).handle()
# Satori OK
async def _(bot: T_Bot, matcher: Matcher, result: Arparma):
if result.main_args.get("text"):
await matcher.finish(Message(unescape(result.main_args.get("text"))))
@ -51,9 +52,11 @@ async def _(bot: T_Bot, matcher: Matcher, result: Arparma):
),
permission=SUPERUSER
).handle()
# Satori OK
async def _(bot: T_Bot, event: T_MessageEvent):
# 使用git pull更新
ulang = get_user_lang(str(event.user_id))
ulang = get_user_lang(str(event.user.id if isinstance(event, satori.event.Event) else event.user_id))
success, logs = update_liteyuki()
reply = "Liteyuki updated!\n"
reply += f"```\n{logs}\n```\n"
@ -70,6 +73,7 @@ async def _(bot: T_Bot, event: T_MessageEvent):
),
permission=SUPERUSER
).handle()
# Satori OK
async def _(matcher: Matcher, bot: T_Bot, event: T_MessageEvent):
await matcher.send("Liteyuki reloading")
temp_data = common_db.where_one(TempConfig(), default=TempConfig())
@ -79,8 +83,8 @@ async def _(matcher: Matcher, bot: T_Bot, event: T_MessageEvent):
"reload": True,
"reload_time": time.time(),
"reload_bot_id": bot.self_id,
"reload_session_type": event.message_type,
"reload_session_id" : event.group_id if event.message_type == "group" else event.user_id,
"reload_session_type": satori_utils.get_message_type(event),
"reload_session_id": (event.group_id if event.message_type == "group" else event.user_id) if not isinstance(event,satori.event.Event) else event.channel.id,
"delta_time": 0
}
)
@ -112,8 +116,9 @@ async def _(matcher: Matcher, bot: T_Bot, event: T_MessageEvent):
),
permission=SUPERUSER
).handle()
# Satori OK
async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot, matcher: Matcher):
ulang = get_user_lang(str(event.user_id))
ulang = get_user_lang(str(satori_utils.get_user_id(event)))
stored_config: StoredConfig = common_db.where_one(StoredConfig(), default=StoredConfig())
if result.subcommands.get("set"):
key, value = result.subcommands.get("set").args.get("key"), result.subcommands.get("set").args.get("value")
@ -159,15 +164,17 @@ async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot, matcher: Matcher
),
permission=SUPERUSER
).handle()
# Satori OK
async def _(event: T_MessageEvent, matcher: Matcher):
global markdown_image
# 切换图片模式，False以图片形式发送，True以markdown形式发送
ulang = get_user_lang(str(event.user_id))
ulang = get_user_lang(str(satori_utils.get_user_id(event)))
stored_config: StoredConfig = common_db.where_one(StoredConfig(), default=StoredConfig())
stored_config.config["markdown_image"] = not stored_config.config.get("markdown_image", False)
markdown_image = stored_config.config["markdown_image"]
common_db.save(stored_config)
await matcher.finish(ulang.get("liteyuki.image_mode_on" if stored_config.config["markdown_image"] else "liteyuki.image_mode_off"))
await matcher.finish(
ulang.get("liteyuki.image_mode_on" if stored_config.config["markdown_image"] else "liteyuki.image_mode_off"))
@on_alconna(
@ -176,6 +183,7 @@ async def _(event: T_MessageEvent, matcher: Matcher):
),
aliases={"轻雪文档"},
).handle()
# Satori OK
async def _(matcher: Matcher):
await matcher.finish("https://bot.liteyuki.icu/usage")
@ -228,7 +236,8 @@ async def _(result: Arparma, bot: T_Bot, event: T_MessageEvent, matcher: Matcher
@Bot.on_calling_api # 图片模式检测
async def test_for_md_image(bot: T_Bot, api: str, data: dict):
# 截获大图发送，转换为markdown发送
if api in ["send_msg", "send_private_msg", "send_group_msg"] and markdown_image and data.get("user_id") != bot.self_id:
if api in ["send_msg", "send_private_msg", "send_group_msg"] and markdown_image and data.get(
"user_id") != bot.self_id:
if api == "send_msg" and data.get("message_type") == "private" or api == "send_private_msg":
session_type = "private"
session_id = data.get("user_id")
@ -241,10 +250,12 @@ async def test_for_md_image(bot: T_Bot, api: str, data: dict):
file: str = data["message"][0].data.get("file")
# file:// http:// base64://
if file.startswith("http"):
result = await md.send_md(await md.image_async(file), bot, message_type=session_type, session_id=session_id)
result = await md.send_md(await md.image_async(file), bot, message_type=session_type,
session_id=session_id)
elif file.startswith("file"):
file = file.replace("file://", "")
result = await md.send_image(open(file, "rb").read(), bot, message_type=session_type, session_id=session_id)
result = await md.send_image(open(file, "rb").read(), bot, message_type=session_type,
session_id=session_id)
elif file.startswith("base64"):
file_bytes = base64.b64decode(file.replace("base64://", ""))
result = await md.send_image(file_bytes, bot, message_type=session_type, session_id=session_id)
@ -283,6 +294,12 @@ async def _(bot: T_Bot):
reload_session_id = temp_data.data.get("reload_session_id", 0)
delta_time = temp_data.data.get("delta_time", 0)
common_db.save(temp_data) # 更新数据
if isinstance(bot,satori.Bot):
await bot.send_message(
channel_id=reload_session_id,
message="Liteyuki reloaded in %.2f s" % delta_time
)
else:
await bot.call_api(
"send_msg",
message_type=reload_session_type,

25
liteyuki/utils/message/message.py
View File

@ -7,6 +7,7 @@ from PIL import Image
import aiohttp
import nonebot
from nonebot import require
from nonebot.adapters import satori
from nonebot.adapters.onebot import v11
from typing import Any, Type
@ -65,8 +66,17 @@ class MarkdownMessage:
"""
formatted_md = v11.unescape(markdown).replace("\n", r"\n").replace('"', r'\\\"')
if event is not None and message_type is None:
if isinstance(event, satori.event.Event):
message_type = "private" if event.guild is None else "group"
group_id = event.guild.id if event.guild is not None else None
else:
assert event is not None
message_type = event.message_type
session_id = event.user_id if event.message_type == "private" else event.group_id
group_id = event.group_id if message_type == "group" else None
user_id = event.user.id if isinstance(event, satori.event.Event) else event.user_id
session_id = user_id if message_type == "private" else group_id
else:
pass
try:
raise TencentBannedMarkdownError("Tencent banned markdown")
forward_id = await bot.call_api(
@ -117,6 +127,13 @@ class MarkdownMessage:
width=540,
device_scale_factor=4
)
if isinstance(bot, satori.Bot):
msg_seg = satori.MessageSegment.image(raw=md_image_bytes,mime="image/png")
data = await bot.send(
event=event,
message=msg_seg
)
else:
data = await bot.send_msg(
message_type=message_type,
group_id=session_id,
@ -164,7 +181,8 @@ class MarkdownMessage:
if method == 2:
base64_string = base64.b64encode(image).decode("utf-8")
data = await bot.call_api("upload_image", file=f"base64://{base64_string}")
await MarkdownMessage.send_md(MarkdownMessage.image(data, Image.open(io.BytesIO(image)).size), bot, event=event, message_type=message_type,
await MarkdownMessage.send_md(MarkdownMessage.image(data, Image.open(io.BytesIO(image)).size), bot,
event=event, message_type=message_type,
session_id=session_id, **kwargs)
# 其他实现端方案
@ -178,7 +196,8 @@ class MarkdownMessage:
image_url = (await bot.get_msg(message_id=image_message_id))["message"][0]["data"]["url"]
image_size = Image.open(io.BytesIO(image)).size
image_md = MarkdownMessage.image(image_url, image_size)
return await MarkdownMessage.send_md(image_md, bot, message_type=message_type, session_id=session_id, event=event, **kwargs)
return await MarkdownMessage.send_md(image_md, bot, message_type=message_type, session_id=session_id,
event=event, **kwargs)
if data is None:
data = await bot.send_msg(

2
liteyuki/utils/satori_utils/__init__.py
View File

@ -1 +1,3 @@
from .user_info import user_infos
from .get_message_type import get_message_type
from .event_tools import *

10
liteyuki/utils/satori_utils/event_tools.py Normal file
View File

@ -0,0 +1,10 @@
from nonebot.adapters import satori
from liteyuki.utils.base.ly_typing import T_MessageEvent
def get_user_id(event: T_MessageEvent):
if isinstance(event, satori.event.Event):
return event.user.id
else:
return event.user_id

10
liteyuki/utils/satori_utils/get_message_type.py Normal file
View File

@ -0,0 +1,10 @@
from nonebot.adapters import satori
from liteyuki.utils.base.ly_typing import T_MessageEvent
def get_message_type(event: T_MessageEvent) -> str:
if isinstance(event, satori.event.Event):
return "private" if event.guild is None else "group"
else:
return event.message_type