mirror of
https://github.com/RockChinQ/QChatGPT.git
synced 2024-11-16 19:57:04 +08:00
163 lines
4.3 KiB
Python
163 lines
4.3 KiB
Python
from __future__ import annotations
|
||
|
||
# MessageSource的适配器
|
||
import typing
|
||
import abc
|
||
|
||
import mirai
|
||
|
||
from ..core import app
|
||
|
||
|
||
preregistered_adapters: list[typing.Type[MessageSourceAdapter]] = []
|
||
|
||
def adapter_class(
|
||
name: str
|
||
):
|
||
def decorator(cls: typing.Type[MessageSourceAdapter]) -> typing.Type[MessageSourceAdapter]:
|
||
cls.name = name
|
||
preregistered_adapters.append(cls)
|
||
return cls
|
||
return decorator
|
||
|
||
|
||
class MessageSourceAdapter(metaclass=abc.ABCMeta):
|
||
name: str
|
||
|
||
bot_account_id: int
|
||
|
||
config: dict
|
||
|
||
ap: app.Application
|
||
|
||
def __init__(self, config: dict, ap: app.Application):
|
||
self.config = config
|
||
self.ap = ap
|
||
|
||
async def send_message(
|
||
self,
|
||
target_type: str,
|
||
target_id: str,
|
||
message: mirai.MessageChain
|
||
):
|
||
"""发送消息
|
||
|
||
Args:
|
||
target_type (str): 目标类型,`person`或`group`
|
||
target_id (str): 目标ID
|
||
message (mirai.MessageChain): YiriMirai库的消息链
|
||
"""
|
||
raise NotImplementedError
|
||
|
||
async def reply_message(
|
||
self,
|
||
message_source: mirai.MessageEvent,
|
||
message: mirai.MessageChain,
|
||
quote_origin: bool = False
|
||
):
|
||
"""回复消息
|
||
|
||
Args:
|
||
message_source (mirai.MessageEvent): YiriMirai消息源事件
|
||
message (mirai.MessageChain): YiriMirai库的消息链
|
||
quote_origin (bool, optional): 是否引用原消息. Defaults to False.
|
||
"""
|
||
raise NotImplementedError
|
||
|
||
async def is_muted(self, group_id: int) -> bool:
|
||
"""获取账号是否在指定群被禁言"""
|
||
raise NotImplementedError
|
||
|
||
def register_listener(
|
||
self,
|
||
event_type: typing.Type[mirai.Event],
|
||
callback: typing.Callable[[mirai.Event, MessageSourceAdapter], None]
|
||
):
|
||
"""注册事件监听器
|
||
|
||
Args:
|
||
event_type (typing.Type[mirai.Event]): YiriMirai事件类型
|
||
callback (typing.Callable[[mirai.Event], None]): 回调函数,接收一个参数,为YiriMirai事件
|
||
"""
|
||
raise NotImplementedError
|
||
|
||
def unregister_listener(
|
||
self,
|
||
event_type: typing.Type[mirai.Event],
|
||
callback: typing.Callable[[mirai.Event, MessageSourceAdapter], None]
|
||
):
|
||
"""注销事件监听器
|
||
|
||
Args:
|
||
event_type (typing.Type[mirai.Event]): YiriMirai事件类型
|
||
callback (typing.Callable[[mirai.Event], None]): 回调函数,接收一个参数,为YiriMirai事件
|
||
"""
|
||
raise NotImplementedError
|
||
|
||
async def run_async(self):
|
||
"""异步运行"""
|
||
raise NotImplementedError
|
||
|
||
async def kill(self) -> bool:
|
||
"""关闭适配器
|
||
|
||
Returns:
|
||
bool: 是否成功关闭,热重载时若此函数返回False则不会重载MessageSource底层
|
||
"""
|
||
raise NotImplementedError
|
||
|
||
|
||
class MessageConverter:
|
||
"""消息链转换器基类"""
|
||
@staticmethod
|
||
def yiri2target(message_chain: mirai.MessageChain):
|
||
"""将YiriMirai消息链转换为目标消息链
|
||
|
||
Args:
|
||
message_chain (mirai.MessageChain): YiriMirai消息链
|
||
|
||
Returns:
|
||
typing.Any: 目标消息链
|
||
"""
|
||
raise NotImplementedError
|
||
|
||
@staticmethod
|
||
def target2yiri(message_chain: typing.Any) -> mirai.MessageChain:
|
||
"""将目标消息链转换为YiriMirai消息链
|
||
|
||
Args:
|
||
message_chain (typing.Any): 目标消息链
|
||
|
||
Returns:
|
||
mirai.MessageChain: YiriMirai消息链
|
||
"""
|
||
raise NotImplementedError
|
||
|
||
|
||
class EventConverter:
|
||
"""事件转换器基类"""
|
||
|
||
@staticmethod
|
||
def yiri2target(event: typing.Type[mirai.Event]):
|
||
"""将YiriMirai事件转换为目标事件
|
||
|
||
Args:
|
||
event (typing.Type[mirai.Event]): YiriMirai事件
|
||
|
||
Returns:
|
||
typing.Any: 目标事件
|
||
"""
|
||
raise NotImplementedError
|
||
|
||
@staticmethod
|
||
def target2yiri(event: typing.Any) -> mirai.Event:
|
||
"""将目标事件的调用参数转换为YiriMirai的事件参数对象
|
||
|
||
Args:
|
||
event (typing.Any): 目标事件
|
||
|
||
Returns:
|
||
typing.Type[mirai.Event]: YiriMirai事件
|
||
"""
|
||
raise NotImplementedError
|