mirror of
https://github.com/TriM-Organization/LiteyukiBot-TriM.git
synced 2024-11-24 16:15:03 +08:00
📝 将从事件中获取信息的工具函数移动到单独的 utils.event
This commit is contained in:
parent
077658c68d
commit
309397b72c
@ -17,7 +17,7 @@ from liteyuki.utils.base.language import get_user_lang
|
||||
from liteyuki.utils.base.ly_typing import T_Bot, T_MessageEvent
|
||||
from liteyuki.utils.message.message import MarkdownMessage as md, broadcast_to_superusers
|
||||
from liteyuki.utils.base.reloader import Reloader
|
||||
from liteyuki.utils import satori_utils
|
||||
from liteyuki.utils import event as event_utils, satori_utils
|
||||
from .api import update_liteyuki
|
||||
|
||||
require("nonebot_plugin_alconna")
|
||||
@ -83,7 +83,7 @@ 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": satori_utils.get_message_type(event),
|
||||
"reload_session_type": event_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
|
||||
}
|
||||
@ -118,7 +118,7 @@ async def _(matcher: Matcher, bot: T_Bot, event: T_MessageEvent):
|
||||
).handle()
|
||||
# Satori OK
|
||||
async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot, matcher: Matcher):
|
||||
ulang = get_user_lang(str(satori_utils.get_user_id(event)))
|
||||
ulang = get_user_lang(str(event_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")
|
||||
@ -168,7 +168,7 @@ async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot, matcher: Matcher
|
||||
async def _(event: T_MessageEvent, matcher: Matcher):
|
||||
global markdown_image
|
||||
# 切换图片模式,False以图片形式发送,True以markdown形式发送
|
||||
ulang = get_user_lang(str(satori_utils.get_user_id(event)))
|
||||
ulang = get_user_lang(str(event_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"]
|
||||
|
@ -5,7 +5,7 @@ import aiofiles
|
||||
import nonebot.plugin
|
||||
from nonebot.adapters import satori
|
||||
|
||||
from liteyuki.utils import satori_utils
|
||||
from liteyuki.utils import event as event_utils
|
||||
from liteyuki.utils.base.data import LiteModel
|
||||
from liteyuki.utils.base.data_manager import GlobalPlugin, Group, User, group_db, plugin_db, user_db
|
||||
from liteyuki.utils.base.ly_typing import T_MessageEvent
|
||||
@ -140,12 +140,12 @@ def set_plugin_session_enable(event: T_MessageEvent, plugin_name: str, enable: b
|
||||
Returns:
|
||||
|
||||
"""
|
||||
if satori_utils.get_message_type(event) == "group":
|
||||
session = group_db.where_one(Group(), "group_id = ?", str(satori_utils.get_group_id(event)),
|
||||
default=Group(group_id=str(satori_utils.get_group_id(event))))
|
||||
if event_utils.get_message_type(event) == "group":
|
||||
session = group_db.where_one(Group(), "group_id = ?", str(event_utils.get_group_id(event)),
|
||||
default=Group(group_id=str(event_utils.get_group_id(event))))
|
||||
else:
|
||||
session = user_db.where_one(User(), "user_id = ?", str(satori_utils.get_user_id(event)),
|
||||
default=User(user_id=str(satori_utils.get_user_id(event))))
|
||||
session = user_db.where_one(User(), "user_id = ?", str(event_utils.get_user_id(event)),
|
||||
default=User(user_id=str(event_utils.get_user_id(event))))
|
||||
default_enable = get_plugin_default_enable(plugin_name)
|
||||
if default_enable:
|
||||
if enable:
|
||||
@ -158,12 +158,12 @@ def set_plugin_session_enable(event: T_MessageEvent, plugin_name: str, enable: b
|
||||
else:
|
||||
session.enabled_plugins.remove(plugin_name)
|
||||
|
||||
if satori_utils.get_message_type(event) == "group":
|
||||
__group_data[str(satori_utils.get_group_id(event))] = session
|
||||
if event_utils.get_message_type(event) == "group":
|
||||
__group_data[str(event_utils.get_group_id(event))] = session
|
||||
print(session)
|
||||
group_db.save(session)
|
||||
else:
|
||||
__user_data[str(satori_utils.get_user_id(event))] = session
|
||||
__user_data[str(event_utils.get_user_id(event))] = session
|
||||
user_db.save(session)
|
||||
|
||||
|
||||
|
@ -1,6 +1,5 @@
|
||||
import nonebot
|
||||
|
||||
from liteyuki.utils import satori_utils
|
||||
from nonebot.message import event_preprocessor
|
||||
# from nonebot_plugin_alconna.typings import Event
|
||||
from liteyuki.utils.base.ly_typing import T_MessageEvent
|
||||
|
@ -1,9 +1,9 @@
|
||||
from nonebot import Bot, require
|
||||
from liteyuki.utils.message.npl import convert_duration, convert_time_to_seconds
|
||||
from .stat_api import *
|
||||
from ...utils import satori_utils
|
||||
from ...utils.base.language import Language
|
||||
from ...utils.base.ly_typing import T_MessageEvent
|
||||
from liteyuki.utils import event as event_utils
|
||||
from liteyuki.utils.base.language import Language
|
||||
from liteyuki.utils.base.ly_typing import T_MessageEvent
|
||||
|
||||
require("nonebot_plugin_alconna")
|
||||
|
||||
@ -45,7 +45,7 @@ stat_msg = on_alconna(
|
||||
|
||||
@stat_msg.assign("message")
|
||||
async def _(result: Arparma, event: T_MessageEvent, bot: Bot):
|
||||
ulang = Language(satori_utils.get_user_id(event))
|
||||
ulang = Language(event_utils.get_user_id(event))
|
||||
|
||||
try:
|
||||
duration = convert_time_to_seconds(result.other_args.get("duration", "2d")) # 秒数
|
||||
@ -58,7 +58,7 @@ async def _(result: Arparma, event: T_MessageEvent, bot: Bot):
|
||||
bot_id = result.other_args.get("bot_id")
|
||||
|
||||
if group_id in ["current", "c"]:
|
||||
group_id = str(satori_utils.get_group_id(event))
|
||||
group_id = str(event_utils.get_group_id(event))
|
||||
|
||||
if group_id in ["all", "a"]:
|
||||
group_id = "all"
|
||||
|
@ -9,7 +9,7 @@ from liteyuki.utils.base.ly_typing import v11, satori
|
||||
from liteyuki.utils.base.ly_typing import T_Bot, T_MessageEvent
|
||||
|
||||
from .common import MessageEventModel, msg_db
|
||||
from liteyuki.utils import satori_utils
|
||||
from liteyuki.utils import event as event_utils
|
||||
|
||||
require("nonebot_plugin_alconna")
|
||||
|
||||
@ -64,6 +64,6 @@ async def satori_event_monitor(bot: satori.Bot, event: satori.MessageEvent):
|
||||
message_id=str(event.message.id),
|
||||
message=event.message,
|
||||
message_text=event.message.content,
|
||||
message_type=satori_utils.get_message_type(event),
|
||||
message_type=event_utils.get_message_type(event),
|
||||
)
|
||||
msg_db.save(mem)
|
||||
|
@ -4,8 +4,8 @@ from liteyuki.utils.base.resource import get_path
|
||||
from liteyuki.utils.message.html_tool import template2image
|
||||
from liteyuki.utils.base.language import get_user_lang
|
||||
from .api import *
|
||||
from ...utils import satori_utils
|
||||
from ...utils.base.ly_typing import T_Bot, T_MessageEvent
|
||||
from liteyuki.utils import event as event_utils
|
||||
from liteyuki.utils.base.ly_typing import T_Bot, T_MessageEvent
|
||||
|
||||
require("nonebot_plugin_alconna")
|
||||
from nonebot_plugin_alconna import on_alconna, Alconna, Args, Subcommand, Arparma, UniMessage
|
||||
@ -28,7 +28,7 @@ status_alc = on_alconna(
|
||||
|
||||
@status_alc.handle()
|
||||
async def _(event: T_MessageEvent, bot: T_Bot):
|
||||
ulang = get_user_lang(satori_utils.get_user_id(event))
|
||||
ulang = get_user_lang(event_utils.get_user_id(event))
|
||||
if ulang.lang_code in status_card_cache:
|
||||
image = status_card_cache[ulang.lang_code]
|
||||
else:
|
||||
|
@ -9,7 +9,7 @@ from liteyuki.utils.base.language import Language, change_user_lang, get_all_lan
|
||||
from liteyuki.utils.base.ly_typing import T_Bot, T_MessageEvent
|
||||
from liteyuki.utils.message.message import MarkdownMessage as md
|
||||
from .const import representative_timezones_list
|
||||
from ...utils import satori_utils
|
||||
from liteyuki.utils import event as event_utils
|
||||
|
||||
require("nonebot_plugin_alconna")
|
||||
from nonebot_plugin_alconna import Alconna, Args, Arparma, Subcommand, on_alconna
|
||||
@ -42,13 +42,13 @@ class Profile(LiteModel):
|
||||
|
||||
@profile_alc.handle()
|
||||
async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot):
|
||||
user: User = user_db.where_one(User(), "user_id = ?", satori_utils.get_user_id(event),
|
||||
default=User(user_id=str(satori_utils.get_user_id(event))))
|
||||
ulang = get_user_lang(str(satori_utils.get_user_id(event)))
|
||||
user: User = user_db.where_one(User(), "user_id = ?", event_utils.get_user_id(event),
|
||||
default=User(user_id=str(event_utils.get_user_id(event))))
|
||||
ulang = get_user_lang(str(event_utils.get_user_id(event)))
|
||||
if result.subcommands.get("set"):
|
||||
if result.subcommands["set"].args.get("value"):
|
||||
# 对合法性进行校验后设置
|
||||
r = set_profile(result.args["key"], result.args["value"], str(satori_utils.get_user_id(event)))
|
||||
r = set_profile(result.args["key"], result.args["value"], str(event_utils.get_user_id(event)))
|
||||
if r:
|
||||
user.profile[result.args["key"]] = result.args["value"]
|
||||
user_db.save(user) # 数据库保存
|
||||
|
@ -11,7 +11,7 @@ from liteyuki.utils.base.data_manager import User, user_db
|
||||
from liteyuki.utils.base.language import Language, get_user_lang
|
||||
from liteyuki.utils.base.resource import get_path
|
||||
from liteyuki.utils.message.html_tool import template2image
|
||||
from ...utils import satori_utils
|
||||
from liteyuki.utils import event as event_utils
|
||||
|
||||
require("nonebot_plugin_alconna")
|
||||
from nonebot_plugin_alconna import on_alconna, Alconna, Args, MultiVar, Arparma
|
||||
@ -43,11 +43,11 @@ async def _(event: T_MessageEvent, matcher: Matcher):
|
||||
|
||||
|
||||
async def get_weather_now_card(matcher: Matcher, event: T_MessageEvent, keyword: list[str], tip: bool = True):
|
||||
ulang = get_user_lang(satori_utils.get_user_id(event))
|
||||
ulang = get_user_lang(event_utils.get_user_id(event))
|
||||
qw_lang = get_qw_lang(ulang.lang_code)
|
||||
key = get_config("weather_key")
|
||||
is_dev = get_memory_data("weather.is_dev", True)
|
||||
user: User = user_db.where_one(User(), "user_id = ?", satori_utils.get_user_id(event), default=User())
|
||||
user: User = user_db.where_one(User(), "user_id = ?", event_utils.get_user_id(event), default=User())
|
||||
# params
|
||||
unit = user.profile.get("unit", "m")
|
||||
stored_location = user.profile.get("location", None)
|
||||
|
1
liteyuki/utils/event/__init__.py
Normal file
1
liteyuki/utils/event/__init__.py
Normal file
@ -0,0 +1 @@
|
||||
from .get_info import *
|
@ -15,3 +15,10 @@ def get_group_id(event: T_MessageEvent):
|
||||
return event.guild.id
|
||||
else:
|
||||
return event.group_id
|
||||
|
||||
|
||||
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
|
@ -1,5 +1,3 @@
|
||||
from .user_info import user_infos
|
||||
from .get_message_type import get_message_type
|
||||
from .event_tools import *
|
||||
from .count_friends import count_friends
|
||||
from .count_groups import count_groups
|
||||
|
@ -1,10 +0,0 @@
|
||||
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
|
Loading…
Reference in New Issue
Block a user