Archived
1
0
Fork 0

Module add hide

This commit is contained in:
多羅狼 2024-02-19 13:43:03 +08:00 committed by GitHub
parent 8bf300e451
commit 07afbc5ea3
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 98 additions and 427 deletions

View file

@ -145,6 +145,7 @@ def module(
developers: Union[str, list, tuple, None] = None,
required_admin: bool = False,
base: bool = False,
hide: bool = False,
required_superuser: bool = False,
required_base_superuser: bool = False,
available_for: Union[str, list, tuple] = '*',
@ -161,10 +162,12 @@ def module(
:param developers: 模块作者
:param required_admin: 此命令是否需要群组管理员权限
:param base: 将此命令设为基础命令设为基础命令后此命令将被强制开启
:param base: 将此命令设为隐藏命令设为隐藏命令后此命令在帮助列表不可见
:param required_superuser: 将此命令设为机器人的超级管理员才可执行
:param required_base_superuser: 将此命令设为机器人的基础超级管理员才可执行
:param available_for: 此命令支持的平台列表
:param exclude_from: 此命令排除的平台列表
:param support_languages: 此命令支持的语言列表
:return: 此类型的模块
"""
module = Module(alias=alias,
@ -173,6 +176,7 @@ def module(
recommend_modules=recommend_modules,
developers=developers,
base=base,
hide=hide,
required_admin=required_admin,
required_superuser=required_superuser,
required_base_superuser=required_base_superuser,

View file

@ -1,363 +1,58 @@
from __future__ import annotations
import asyncio
from typing import List, Union, Dict, Coroutine, Awaitable
from typing import Union, Dict, List
from core.exceptions import FinishedException
from .chain import MessageChain
from apscheduler.triggers.combining import AndTrigger, OrTrigger
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.date import DateTrigger
from apscheduler.triggers.interval import IntervalTrigger
from .component_matches import *
class MsgInfo:
__slots__ = ["target_id", "sender_id", "sender_name", "target_from", "sender_info", "sender_from", "client_name",
"message_id", "reply_id"]
def convert2lst(elements: Union[str, list, tuple]) -> list:
if isinstance(elements, str):
return [elements]
elif isinstance(elements, tuple):
return list(elements)
return elements
class Module:
def __init__(self,
target_id: Union[int, str],
sender_id: Union[int, str],
sender_name: str,
target_from: str,
sender_from: str,
client_name: str,
message_id: Union[int, str],
reply_id: Union[int, str] = None):
self.target_id = target_id
self.sender_id = sender_id
self.sender_name = sender_name
self.target_from = target_from
self.sender_from = sender_from
self.client_name = client_name
self.message_id = message_id
self.reply_id = reply_id
def __repr__(self):
return f'MsgInfo(target_id={self.target_id}, sender_id={self.sender_id}, sender_name={self.sender_name},' \
f' target_from={self.target_from}, sender_from={self.sender_from}, client_name={self.client_name}, ' \
f'message_id={self.message_id}, reply_id={self.reply_id})'
bind_prefix: str,
alias: Union[str, list, tuple, dict, None] = None,
desc: str = None,
recommend_modules: Union[str, list, tuple, None] = None,
developers: Union[str, list, tuple, None] = None,
required_admin: bool = False,
base: bool = False,
hide: bool = False,
required_superuser: bool = False,
required_base_superuser: bool = False,
available_for: Union[str, list, tuple, None] = '*',
exclude_from: Union[str, list, tuple, None] = '',
support_languages: Union[str, list, tuple, None] = None):
self.bind_prefix: str = bind_prefix
if isinstance(alias, str):
alias = {alias: bind_prefix}
elif isinstance(alias, (tuple, list)):
alias = {x: bind_prefix for x in alias}
self.alias: Dict[str, str] = alias
self.desc: str = desc
self.recommend_modules: List[str] = convert2lst(recommend_modules)
self.developers: List[str] = convert2lst(developers)
self.required_admin: bool = required_admin
self.base: bool = base
self.hide: bool = hide
self.required_superuser: bool = required_superuser
self.required_base_superuser: bool = required_base_superuser
self.available_for: List[str] = convert2lst(available_for)
self.exclude_from: List[str] = convert2lst(exclude_from)
self.support_languages: List[str] = convert2lst(support_languages)
self.command_list = CommandMatches()
self.regex_list = RegexMatches()
self.schedule_list = ScheduleMatches()
self.hooks_list = HookMatches()
class Session:
def __init__(self, message, target, sender):
self.message = message
self.target = target
self.sender = sender
def __repr__(self):
return f'Session(message={self.message}, target={self.target}, sender={self.sender})'
class FinishedSession:
def __init__(self, session, message_id: Union[List[int], List[str], int, str], result):
self.session = session
if isinstance(message_id, (int, str)):
message_id = [message_id]
self.message_id = message_id
self.result = result
async def delete(self):
"""
用于删除这条消息
"""
...
def __str__(self):
return f"FinishedSession(message_id={self.message_id}, result={self.result})"
class MessageSession:
"""
消息会话囊括了处理一条消息所需要的东西
"""
__slots__ = (
"target", "session", "trigger_msg", "parsed_msg", "matched_msg", "sent", "prefixes", "options",
"enabled_modules", "muted", "custom_admins", "data", "locale", "timestamp", "tmp", "timezone_offset",
"_tz_offset")
parsed_msg: Dict[str, Union[str, list]]
def __init__(self,
target: MsgInfo,
session: Session):
self.target = target
self.session = session
self.sent: List[MessageChain] = []
self.prefixes: List[str] = []
self.options: dict = {}
self.enabled_modules: List[str] = []
self.muted: bool = False
self.timestamp: float = 0
self.tmp = {}
async def send_message(self,
message_chain: MessageChain,
quote=True,
disable_secret_check=False,
allow_split_image=True,
callback: Coroutine = None) -> FinishedSession:
"""
用于向消息发送者回复消息
:param message_chain: 消息链若传入str则自动创建一条带有Plain元素的消息链
:param quote: 是否引用传入dict中的消息默认为True
:param disable_secret_check: 是否禁用消息检查默认为False
:param allow_split_image: 是否允许拆分图片发送此参数作接口兼容用仅telegram平台使用了切割
:param callback: 回调函数用于在消息发送完成后回复本消息执行的函数
:return: 被发送的消息链
"""
raise NotImplementedError
async def finish(self,
message_chain: MessageChain = None,
quote: bool = True,
disable_secret_check: bool = False,
allow_split_image: bool = True,
callback: Union[Awaitable, None] = None):
"""
用于向消息发送者回复消息并终结会话模块后续代码不再执行
:param message_chain: 消息链若传入str则自动创建一条带有Plain元素的消息链
:param quote: 是否引用传入dict中的消息默认为True
:param disable_secret_check: 是否禁用消息检查默认为False
:param allow_split_image: 是否允许拆分图片发送此参数作接口兼容用仅telegram平台使用了切割
:param callback: 回调函数用于在消息发送完成后回复本消息执行的函数
:return: 被发送的消息链
"""
...
f = None
if message_chain:
f = await self.send_message(message_chain, disable_secret_check=disable_secret_check, quote=quote,
allow_split_image=allow_split_image, callback=callback)
raise FinishedException(f)
async def send_direct_message(self, message_chain, disable_secret_check=False, allow_split_image=True,
callback: Coroutine = None):
"""
用于向消息发送者直接发送消息
:param message_chain: 消息链若传入str则自动创建一条带有Plain元素的消息链
:param disable_secret_check: 是否禁用消息检查默认为False
:param allow_split_image: 是否允许拆分图片发送此参数作接口兼容用仅Telegram平台使用了切割
:param callback: 回调函数用于在消息发送完成后回复本消息执行的函数
:return: 被发送的消息链
"""
await self.send_message(message_chain, disable_secret_check=disable_secret_check, quote=False,
allow_split_image=allow_split_image, callback=callback)
async def wait_confirm(self, message_chain=None, quote=True, delete=True, timeout=120, append_instruction=True):
"""
一次性模板用于等待触发对象确认
:param message_chain: 需要发送的确认消息可不填
:param quote: 是否引用传入dict中的消息默认为True
:param delete: 是否在触发后删除消息默认为True
:param timeout: 超时时间
:return: 若对象发送confirm_command中的其一文本时返回True反之则返回False
"""
raise NotImplementedError
async def wait_next_message(self, message_chain=None, quote=True, delete=False, timeout=120,
append_instruction=True):
"""
一次性模板用于等待对象的下一条消息
:param message_chain: 需要发送的确认消息可不填
:param quote: 是否引用传入dict中的消息默认为True
:param delete: 是否在触发后删除消息默认为False
:param timeout: 超时时间
:return: 下一条消息的MessageChain对象
"""
raise NotImplementedError
async def wait_reply(self, message_chain, quote=True, delete=False, timeout=120, all_=False,
append_instruction=True):
"""
一次性模板用于等待触发对象回复消息
:param message_chain: 需要发送的确认消息可不填
:param quote: 是否引用传入dict中的消息默认为True
:param delete: 是否在触发后删除消息默认为False
:param timeout: 超时时间
:param all_: 是否设置触发对象为对象内的所有人默认为False
:return: 回复消息的MessageChain对象
"""
raise NotImplementedError
async def wait_anyone(self, message_chain=None, quote=False, delete=False, timeout=120):
"""
一次性模板用于等待触发发送者所属对象内所有成员确认
:param message_chain: 需要发送的确认消息可不填
:param quote: 是否引用传入dict中的消息默认为False
:param delete: 是否在触发后删除消息默认为False
:param timeout: 超时时间
:return: 任意人的MessageChain对象
"""
raise NotImplementedError
def as_display(self, text_only=False) -> str:
"""
用于将消息转换为一般文本格式
:param text_only: 是否只保留纯文本默认为False
"""
raise NotImplementedError
async def to_message_chain(self) -> MessageChain:
"""
用于将session.message中的消息文本转换为MessageChain
"""
raise NotImplementedError
async def delete(self):
"""
用于删除这条消息
"""
raise NotImplementedError
async def check_permission(self):
"""
用于检查消息发送者在对象内的权限
"""
raise NotImplementedError
async def check_native_permission(self):
"""
用于检查消息发送者原本在聊天平台中是否具有管理员权限
"""
raise NotImplementedError
async def fake_forward_msg(self, nodelist):
"""
用于发送假转发消息QQ
"""
raise NotImplementedError
async def get_text_channel_list(self):
"""
用于获取子文字频道列表QQ
"""
raise NotImplementedError
def ts2strftime(self, timestamp: float, date=True, iso=False, time=True, seconds=True, timezone=True):
"""
用于将时间戳转换为可读的时间格式
:param timestamp: 时间戳
:param date: 是否显示日期
:param iso: 是否以ISO格式显示
:param time: 是否显示时间
:param seconds: 是否显示秒
:param timezone: 是否显示时区
"""
raise NotImplementedError
class Typing:
def __init__(self, msg):
"""
:param msg: 本条消息由于此class需要被一同传入下游方便调用所以作为子class存在将来可能会有其他的解决办法
"""
async def __aenter__(self):
pass
async def __aexit__(self, exc_type, exc_val, exc_tb):
pass
def check_super_user(self):
"""
用于检查消息发送者是否为超级用户
"""
raise NotImplementedError
@staticmethod
async def sleep(s):
await asyncio.sleep(s)
async def call_api(self, action, **params):
raise NotImplementedError
sendMessage = send_message
sendDirectMessage = send_direct_message
waitConfirm = wait_confirm
waitNextMessage = wait_next_message
waitReply = wait_reply
waitAnyone = wait_anyone
asDisplay = as_display
toMessageChain = to_message_chain
checkPermission = check_permission
checkNativePermission = check_native_permission
checkSuperUser = check_super_user
class Feature:
"""
此消息来自的客户端所支持的消息特性一览用于不同平台适用特性判断如QQ支持消息类型的语音而Discord不支持
"""
image = ...
voice = ...
embed = ...
forward = ...
delete = ...
quote = ...
wait = ...
def __str__(self):
return "Message(target={}, session={}, sent={})".format(self.target, self.session, self.sent)
class FetchedSession:
def __init__(self, target_from, target_id, sender_from=None, sender_id=None):
if not sender_from:
sender_from = target_from
if not sender_id:
sender_id = target_id
self.target = MsgInfo(target_id=f'{target_from}|{target_id}',
sender_id=f'{target_from}|{sender_id}',
target_from=target_from,
sender_from=sender_from,
sender_name='', client_name='', reply_id=None, message_id=0)
self.session = Session(message=False, target=target_id, sender=sender_id)
self.parent = MessageSession(self.target, self.session)
async def send_direct_message(self, message_chain, disable_secret_check=False, allow_split_image=True):
"""
用于向获取对象发送消息
:param message_chain: 消息链若传入str则自动创建一条带有Plain元素的消息链
:param disable_secret_check: 是否禁用消息检查默认为False
:param allow_split_image: 是否允许拆分图片发送此参数作接口兼容用仅telegram平台使用了切割
:return: 被发送的消息链
"""
return await self.parent.send_direct_message(message_chain, disable_secret_check,
allow_split_image=allow_split_image)
sendDirectMessage = send_direct_message
class FetchTarget:
name = ''
@staticmethod
async def fetch_target(target_id, sender_id=None) -> FetchedSession:
"""
尝试从数据库记录的对象ID中取得对象消息会话实际此会话中的消息文本会被设为False因为本来就没有
"""
raise NotImplementedError
@staticmethod
async def fetch_target_list(target_list: list) -> List[FetchedSession]:
"""
尝试从数据库记录的对象ID中取得对象消息会话实际此会话中的消息文本会被设为False因为本来就没有
"""
raise NotImplementedError
@staticmethod
async def post_message(module_name, message, user_list: List[FetchedSession] = None, i18n=False, **kwargs):
"""
尝试向开启此模块的对象发送一条消息
:param module_name: 模块名
:param message: 消息文本
:param user_list: 用户列表
:param i18n: 是否使用i18n若为True则message为i18n的key或为指定语言的dict映射表k=语言v=文本
"""
raise NotImplementedError
class ModuleHookContext:
"""
模块任务上下文主要用于传递模块任务的参数
"""
def __init__(self, args: dict):
self.args = args
__all__ = ["FetchTarget", "MsgInfo", "MessageSession", "Session", "FetchedSession", "FinishedSession",
"ModuleHookContext", "MessageChain"]
__all__ = ["Module", "AndTrigger", "OrTrigger", "DateTrigger",
"CronTrigger", "IntervalTrigger"]

View file

@ -81,7 +81,7 @@ async def config_modules(msg: Bot.MessageSession):
for function in modules_:
if function[0] == '_':
continue
if modules_[function].base or modules_[function].required_superuser:
if modules_[function].base or modules_[function].hide or modules_[function].required_superuser:
continue
enable_list.append(function)
else:
@ -140,7 +140,7 @@ async def config_modules(msg: Bot.MessageSession):
for function in modules_:
if function[0] == '_':
continue
if modules_[function].base or modules_[function].required_superuser:
if modules_[function].base or modules_[function].hide or modules_[function].required_superuser:
continue
disable_list.append(function)
else:
@ -428,9 +428,11 @@ async def _(msg: Bot.MessageSession):
appends.append('\n'.join(malias) if malias else '')
if module_.developers:
appends.append(msg.locale.t('message.delimiter').join(module_.developers))
if module_.base and not (module_.required_superuser or module_.required_base_superuser):
if module_.base and not (
module_.hide or module_.required_superuser or module_.required_base_superuser):
essential.append(appends)
if x in target_enabled_list and not (module_.required_superuser or module_.required_base_superuser):
if x in target_enabled_list and not (
module_.hide or module_.required_superuser or module_.required_base_superuser):
m.append(appends)
if essential:
tables.append(ImageTable(
@ -462,14 +464,14 @@ async def _(msg: Bot.MessageSession):
help_msg = [msg.locale.t("core.message.help.legacy.base")]
essential = []
for x in module_list:
if module_list[x].base and not (
if module_list[x].base and not (module_list[x].hide or
module_list[x].required_superuser or module_list[x].required_base_superuser):
essential.append(module_list[x].bind_prefix)
help_msg.append(' | '.join(essential))
help_msg.append(msg.locale.t("core.message.help.legacy.external"))
module_ = []
for x in module_list:
if x in target_enabled_list and not (
if x in target_enabled_list and not (module_list[x].hide or
module_list[x].required_superuser or module_list[x].required_base_superuser):
module_.append(x)
help_msg.append(' | '.join(module_))
@ -507,7 +509,7 @@ async def modules_help(msg: Bot.MessageSession, legacy):
module_ = module_list[x]
if x[0] == '_':
continue
if module_.base or module_.required_superuser or module_.required_base_superuser:
if module_.base or module_.hide or module_.required_superuser or module_.required_base_superuser:
continue
appends = [module_.bind_prefix]
doc_ = []
@ -561,7 +563,8 @@ async def modules_help(msg: Bot.MessageSession, legacy):
for x in module_list:
if x[0] == '_':
continue
if module_list[x].base or module_list[x].required_superuser or module_list[x].required_base_superuser:
if module_list[x].base or module_list[x].hide or \
module_list[x].required_superuser or module_list[x].required_base_superuser:
continue
module_.append(module_list[x].bind_prefix)
help_msg.append(' | '.join(module_))

View file

@ -2,8 +2,7 @@ from core.builtins import Bot
from core.component import module
from .mcv import mcv, mcbv, mcdv, mcev, mclgv
m = module(
bind_prefix='mcv',
m = module('mcv',
developers=['OasisAkari', 'Dianliang233'],
recommend_modules=['mcbv'])
@ -13,8 +12,7 @@ async def mcv_loader(msg: Bot.MessageSession):
await msg.finish(await mcv(msg))
mb = module(
bind_prefix='mcbv',
mb = module('mcbv',
developers=['OasisAkari', 'Dianliang233'],
recommend_modules=['mcv'])
@ -24,9 +22,9 @@ async def mcbv_loader(msg: Bot.MessageSession):
await msg.finish(await mcbv(msg))
md = module(
bind_prefix='mcdv',
developers=['OasisAkari', 'Dianliang233'])
md = module('mcdv',
developers=['OasisAkari', 'Dianliang233'],
hide=True)
@md.command('{{mcv.help.mcdv}}')
@ -34,9 +32,9 @@ async def mcdv_loader(msg: Bot.MessageSession):
await msg.finish(await mcdv(msg))
me = module(
bind_prefix='mcev',
developers=['OasisAkari', 'Dianliang233'])
me = module('mcev',
developers=['OasisAkari', 'Dianliang233'],
hide=True)
@me.command('{{mcv.help.mcev}}')
@ -44,9 +42,9 @@ async def mcev_loader(msg: Bot.MessageSession):
await msg.finish(await mcev(msg))
mlg = module(
bind_prefix='mclgv',
developers=['OasisAkari', 'Dianliang233'])
mlg = module('mclgv',
developers=['OasisAkari', 'Dianliang233'],
hide=True)
@mlg.command('{{mcv.help.mclgv}}')

View file

@ -3,8 +3,9 @@ from core.component import module
mcv_rss = module('mcv_rss',
developers=['OasisAkari', 'Dianliang233'],
recommend_modules=['mcv_jira_rss', 'mcbv_jira_rss', 'mcdv_jira_rss', 'mclgv_jira_rss'],
desc='{mcv_rss.help.mcv_rss.desc}', alias='mcvrss')
recommend_modules=['mcv_jira_rss'],
desc='{mcv_rss.help.mcv_rss.desc}',
alias='mcvrss')
@mcv_rss.hook()
@ -14,7 +15,8 @@ async def mcv_rss(fetch: Bot.FetchTarget, ctx: Bot.ModuleHookContext):
mcbv_rss = module('mcbv_rss', developers=['OasisAkari'],
recommend_modules=['mcbv_jira_rss'],
desc='{mcv_rss.help.mcbv_rss.desc}', alias='mcbvrss')
desc='{mcv_rss.help.mcbv_rss.desc}',
alias='mcbvrss')
@mcbv_rss.hook()
@ -22,9 +24,11 @@ async def mcbv_rss(fetch: Bot.FetchTarget, ctx: Bot.ModuleHookContext):
await fetch.post_message('mcbv_rss', **ctx.args)
mcv_jira_rss = module('mcv_jira_rss', developers=['OasisAkari', 'Dianliang233'],
recommend_modules=['mcv_rss', 'mcbv_jira_rss', 'mcdv_jira_rss', 'mclgv_jira_rss'],
desc='{mcv_rss.help.mcv_jira_rss.desc}', alias='mcvjirarss')
mcv_jira_rss = module('mcv_jira_rss',
developers=['OasisAkari', 'Dianliang233'],
recommend_modules=['mcv_rss'],
desc='{mcv_rss.help.mcv_jira_rss.desc}',
alias='mcvjirarss')
@mcv_jira_rss.hook()
@ -34,8 +38,9 @@ async def mcv_jira_rss(fetch: Bot.FetchTarget, ctx: Bot.ModuleHookContext):
mcbv_jira_rss = module('mcbv_jira_rss',
developers=['OasisAkari', 'Dianliang233'],
recommend_modules=['mcv_rss', 'mcv_jira_rss', 'mcdv_jira_rss', 'mclgv_jira_rss'],
desc='{mcv_rss.help.mcbv_jira_rss.desc}', alias='mcbvjirarss')
recommend_modules=['mcbv_rss'],
desc='{mcv_rss.help.mcbv_jira_rss.desc}',
alias='mcbvjirarss')
@mcbv_jira_rss.hook()
@ -45,8 +50,9 @@ async def mcbv_jira_rss(fetch: Bot.FetchTarget, ctx: Bot.ModuleHookContext):
mcdv_jira_rss = module('mcdv_jira_rss',
developers=['OasisAkari', 'Dianliang233'],
recommend_modules=['mcv_rss', 'mcbv_jira_rss', 'mcv_jira_rss', 'mclgv_jira_rss'],
desc='{mcv_rss.help.mcdv_jira_rss.desc}', alias='mcdvjirarss')
desc='{mcv_rss.help.mcdv_jira_rss.desc}',
alias='mcdvjirarss',
hide=True)
@mcdv_jira_rss.hook()
@ -56,8 +62,9 @@ async def mcdv_jira_rss(fetch: Bot.FetchTarget, ctx: Bot.ModuleHookContext):
mclgv_jira_rss = module('mclgv_jira_rss',
developers=['OasisAkari', 'Dianliang233'],
recommend_modules=['mcv_rss', 'mcbv_jira_rss', 'mcv_jira_rss', 'mcdv_jira_rss'],
desc='{mcv_rss.help.mclgv_jira_rss.desc}', alias='mclgvjirarss')
desc='{mcv_rss.help.mclgv_jira_rss.desc}',
alias='mclgvjirarss',
hide=True)
@mclgv_jira_rss.hook()

View file

@ -1,23 +0,0 @@
import secrets
from core.builtins import Bot
from core.component import module
from core.petal import gained_petal
stone = module('stone', developers=['OasisAkari'], desc='{stone.help.desc}')
@stone.command()
@stone.regex(r'打水漂')
async def _(msg: Bot.MessageSession):
count = secrets.randbelow(11)
if count == 0:
send = msg.locale.t('stone.message.skip.nothing')
else:
send = msg.locale.t('stone.message.skip', count=count)
if count == 10:
if g := (g_msg := await gained_petal(msg, 1)):
send += '\n' + g
await msg.finish(send)

View file

@ -1,5 +0,0 @@
{
"stone.help.desc": "Stone skipping.",
"stone.message.skip": "You threw a stone and it floated ${count} time(s).",
"stone.message.skip.nothing": "You threw a stone and nothing happened."
}

View file

@ -1,5 +0,0 @@
{
"stone.help.desc": "打水漂。",
"stone.message.skip": "你扔了一块石头,漂了 ${count} 下。",
"stone.message.skip.nothing": "你扔了一块石头,什么也没有发生。"
}

View file

@ -1,5 +0,0 @@
{
"stone.help.desc": "打水漂。",
"stone.message.skip": "你扔了一塊石頭,漂了 ${count} 下。",
"stone.message.skip.nothing": "你扔了一塊石頭,什麼也沒有發生。"
}

View file

@ -1,3 +1,5 @@
import traceback
from config import Config
from core.logger import Logger
from core.scheduler import Scheduler, CronTrigger