mirror of
https://github.com/nonebot/nonebot2.git
synced 2025-06-07 12:55:56 +00:00
💡 add adapter docstring
This commit is contained in:
parent
98ef09585a
commit
aef585c60c
@ -1,7 +1,6 @@
|
|||||||
"""
|
"""本模块定义了协议适配基类,各协议请继承以下基类。
|
||||||
## 协议适配基类
|
|
||||||
|
|
||||||
各协议请继承以下基类,并使用 `driver.register_adapter` 注册适配器
|
使用 {ref}`nonebot.drivers.Driver.register_adapter` 注册适配器
|
||||||
"""
|
"""
|
||||||
|
|
||||||
from typing import Iterable
|
from typing import Iterable
|
||||||
|
@ -18,56 +18,89 @@ from ._bot import Bot
|
|||||||
|
|
||||||
|
|
||||||
class Adapter(abc.ABC):
|
class Adapter(abc.ABC):
|
||||||
|
"""协议适配器基类。
|
||||||
|
|
||||||
|
通常,在 Adapter 中编写协议通信相关代码,如: 建立通信连接、处理接收与发送 data 等。
|
||||||
|
|
||||||
|
参数:
|
||||||
|
driver: {ref}`nonebot.drivers.Driver` 实例
|
||||||
|
kwargs: 其他由 {ref}`nonebot.drivers.Driver.register_adapter` 传入的额外参数
|
||||||
|
"""
|
||||||
|
|
||||||
def __init__(self, driver: Driver, **kwargs: Any):
|
def __init__(self, driver: Driver, **kwargs: Any):
|
||||||
self.driver: Driver = driver
|
self.driver: Driver = driver
|
||||||
|
"""{ref}`nonebot.drivers.Driver` 实例"""
|
||||||
self.bots: Dict[str, Bot] = {}
|
self.bots: Dict[str, Bot] = {}
|
||||||
|
"""本协议适配器已建立连接的 {ref}`nonebot.adapters._bot.Bot` 实例"""
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
def get_name(cls) -> str:
|
def get_name(cls) -> str:
|
||||||
|
"""当前协议适配器的名称"""
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def config(self) -> Config:
|
def config(self) -> Config:
|
||||||
|
"""全局 NoneBot 配置"""
|
||||||
return self.driver.config
|
return self.driver.config
|
||||||
|
|
||||||
def bot_connect(self, bot: Bot) -> None:
|
def bot_connect(self, bot: Bot) -> None:
|
||||||
|
"""告知 NoneBot 建立了一个新的 {ref}`nonebot.adapters._bot.Bot` 连接。
|
||||||
|
|
||||||
|
当有新的 {ref}`nonebot.adapters._bot.Bot` 实例连接建立成功时调用。
|
||||||
|
|
||||||
|
参数:
|
||||||
|
bot: {ref}`nonebot.adapters._bot.Bot` 实例
|
||||||
|
"""
|
||||||
self.driver._bot_connect(bot)
|
self.driver._bot_connect(bot)
|
||||||
self.bots[bot.self_id] = bot
|
self.bots[bot.self_id] = bot
|
||||||
|
|
||||||
def bot_disconnect(self, bot: Bot) -> None:
|
def bot_disconnect(self, bot: Bot) -> None:
|
||||||
|
"""告知 NoneBot {ref}`nonebot.adapters._bot.Bot` 连接已断开。
|
||||||
|
|
||||||
|
当有 {ref}`nonebot.adapters._bot.Bot` 实例连接断开时调用。
|
||||||
|
|
||||||
|
参数:
|
||||||
|
bot: {ref}`nonebot.adapters._bot.Bot` 实例
|
||||||
|
"""
|
||||||
self.driver._bot_disconnect(bot)
|
self.driver._bot_disconnect(bot)
|
||||||
self.bots.pop(bot.self_id, None)
|
self.bots.pop(bot.self_id, None)
|
||||||
|
|
||||||
def setup_http_server(self, setup: HTTPServerSetup):
|
def setup_http_server(self, setup: HTTPServerSetup):
|
||||||
|
"""设置一个 HTTP 服务器路由配置"""
|
||||||
if not isinstance(self.driver, ReverseDriver):
|
if not isinstance(self.driver, ReverseDriver):
|
||||||
raise TypeError("Current driver does not support http server")
|
raise TypeError("Current driver does not support http server")
|
||||||
self.driver.setup_http_server(setup)
|
self.driver.setup_http_server(setup)
|
||||||
|
|
||||||
def setup_websocket_server(self, setup: WebSocketServerSetup):
|
def setup_websocket_server(self, setup: WebSocketServerSetup):
|
||||||
|
"""设置一个 WebSocket 服务器路由配置"""
|
||||||
if not isinstance(self.driver, ReverseDriver):
|
if not isinstance(self.driver, ReverseDriver):
|
||||||
raise TypeError("Current driver does not support websocket server")
|
raise TypeError("Current driver does not support websocket server")
|
||||||
self.driver.setup_websocket_server(setup)
|
self.driver.setup_websocket_server(setup)
|
||||||
|
|
||||||
async def request(self, setup: Request) -> Response:
|
async def request(self, setup: Request) -> Response:
|
||||||
|
"""进行一个 HTTP 客户端请求"""
|
||||||
if not isinstance(self.driver, ForwardDriver):
|
if not isinstance(self.driver, ForwardDriver):
|
||||||
raise TypeError("Current driver does not support http client")
|
raise TypeError("Current driver does not support http client")
|
||||||
return await self.driver.request(setup)
|
return await self.driver.request(setup)
|
||||||
|
|
||||||
@asynccontextmanager
|
@asynccontextmanager
|
||||||
async def websocket(self, setup: Request) -> AsyncGenerator[WebSocket, None]:
|
async def websocket(self, setup: Request) -> AsyncGenerator[WebSocket, None]:
|
||||||
|
"""建立一个 WebSocket 客户端连接请求"""
|
||||||
if not isinstance(self.driver, ForwardDriver):
|
if not isinstance(self.driver, ForwardDriver):
|
||||||
raise TypeError("Current driver does not support websocket client")
|
raise TypeError("Current driver does not support websocket client")
|
||||||
async with self.driver.websocket(setup) as ws:
|
async with self.driver.websocket(setup) as ws:
|
||||||
yield ws
|
yield ws
|
||||||
|
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
async def _call_api(self, bot: Bot, api: str, **data) -> Any:
|
async def _call_api(self, bot: Bot, api: str, **data: Any) -> Any:
|
||||||
"""
|
"""`Adapter` 实际调用 api 的逻辑实现函数,实现该方法以调用 api。
|
||||||
`adapter` 实际调用 api 的逻辑实现函数,实现该方法以调用 api。
|
|
||||||
|
|
||||||
参数:
|
参数:
|
||||||
api: API 名称
|
api: API 名称
|
||||||
**data: API 数据
|
data: API 数据
|
||||||
"""
|
"""
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
|
|
||||||
|
__autodoc__ = {"Adapter._call_api": True}
|
||||||
|
@ -21,26 +21,23 @@ class _ApiCall(Protocol):
|
|||||||
|
|
||||||
|
|
||||||
class Bot(abc.ABC):
|
class Bot(abc.ABC):
|
||||||
"""
|
"""Bot 基类。
|
||||||
Bot 基类。用于处理上报消息,并提供 API 调用接口。
|
|
||||||
|
用于处理上报消息,并提供 API 调用接口。
|
||||||
|
|
||||||
|
参数:
|
||||||
|
adapter: 协议适配器实例
|
||||||
|
self_id: 机器人 ID
|
||||||
"""
|
"""
|
||||||
|
|
||||||
_calling_api_hook: Set[T_CallingAPIHook] = set()
|
_calling_api_hook: Set[T_CallingAPIHook] = set()
|
||||||
"""
|
"""call_api 时执行的函数"""
|
||||||
call_api 时执行的函数
|
|
||||||
"""
|
|
||||||
_called_api_hook: Set[T_CalledAPIHook] = set()
|
_called_api_hook: Set[T_CalledAPIHook] = set()
|
||||||
"""
|
"""call_api 后执行的函数"""
|
||||||
call_api 后执行的函数
|
|
||||||
"""
|
|
||||||
|
|
||||||
def __init__(self, adapter: "Adapter", self_id: str):
|
def __init__(self, adapter: "Adapter", self_id: str):
|
||||||
"""
|
|
||||||
参数:
|
|
||||||
self_id: 机器人 ID
|
|
||||||
request: request 连接对象
|
|
||||||
"""
|
|
||||||
self.adapter: "Adapter" = adapter
|
self.adapter: "Adapter" = adapter
|
||||||
|
"""协议适配器实例"""
|
||||||
self.self_id: str = self_id
|
self.self_id: str = self_id
|
||||||
"""机器人 ID"""
|
"""机器人 ID"""
|
||||||
|
|
||||||
@ -49,19 +46,20 @@ class Bot(abc.ABC):
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def type(self) -> str:
|
def type(self) -> str:
|
||||||
|
"""协议适配器名称"""
|
||||||
return self.adapter.get_name()
|
return self.adapter.get_name()
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def config(self) -> Config:
|
def config(self) -> Config:
|
||||||
|
"""全局 NoneBot 配置"""
|
||||||
return self.adapter.config
|
return self.adapter.config
|
||||||
|
|
||||||
async def call_api(self, api: str, **data: Any) -> Any:
|
async def call_api(self, api: str, **data: Any) -> Any:
|
||||||
"""
|
"""调用机器人 API 接口,可以通过该函数或直接通过 bot 属性进行调用
|
||||||
调用机器人 API 接口,可以通过该函数或直接通过 bot 属性进行调用
|
|
||||||
|
|
||||||
参数:
|
参数:
|
||||||
api: API 名称
|
api: API 名称
|
||||||
**data: API 数据
|
data: API 数据
|
||||||
|
|
||||||
用法:
|
用法:
|
||||||
```python
|
```python
|
||||||
@ -121,41 +119,44 @@ class Bot(abc.ABC):
|
|||||||
|
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
async def send(
|
async def send(
|
||||||
self, event: "Event", message: Union[str, "Message", "MessageSegment"], **kwargs
|
self,
|
||||||
|
event: "Event",
|
||||||
|
message: Union[str, "Message", "MessageSegment"],
|
||||||
|
**kwargs: Any,
|
||||||
) -> Any:
|
) -> Any:
|
||||||
"""
|
"""调用机器人基础发送消息接口
|
||||||
调用机器人基础发送消息接口
|
|
||||||
|
|
||||||
参数:
|
参数:
|
||||||
event: 上报事件
|
event: 上报事件
|
||||||
message: 要发送的消息
|
message: 要发送的消息
|
||||||
|
kwargs: 任意额外参数
|
||||||
"""
|
"""
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def on_calling_api(cls, func: T_CallingAPIHook) -> T_CallingAPIHook:
|
def on_calling_api(cls, func: T_CallingAPIHook) -> T_CallingAPIHook:
|
||||||
"""
|
"""调用 api 预处理。
|
||||||
调用 api 预处理。
|
|
||||||
|
|
||||||
参数:
|
插槽函数参数:
|
||||||
bot: 当前 bot 对象
|
|
||||||
api: 调用的 api 名称
|
- bot: 当前 bot 对象
|
||||||
data: api 调用的参数字典
|
- api: 调用的 api 名称
|
||||||
|
- data: api 调用的参数字典
|
||||||
"""
|
"""
|
||||||
cls._calling_api_hook.add(func)
|
cls._calling_api_hook.add(func)
|
||||||
return func
|
return func
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def on_called_api(cls, func: T_CalledAPIHook) -> T_CalledAPIHook:
|
def on_called_api(cls, func: T_CalledAPIHook) -> T_CalledAPIHook:
|
||||||
"""
|
"""调用 api 后处理。
|
||||||
调用 api 后处理。
|
|
||||||
|
|
||||||
参数:
|
插槽函数参数:
|
||||||
bot: 当前 bot 对象
|
|
||||||
exception: 调用 api 时发生的错误
|
- bot: 当前 bot 对象
|
||||||
api: 调用的 api 名称
|
- exception: 调用 api 时发生的错误
|
||||||
data: api 调用的参数字典
|
- api: 调用的 api 名称
|
||||||
result: api 调用的返回
|
- data: api 调用的参数字典
|
||||||
|
- result: api 调用的返回
|
||||||
"""
|
"""
|
||||||
cls._called_api_hook.add(func)
|
cls._called_api_hook.add(func)
|
||||||
return func
|
return func
|
||||||
|
@ -16,31 +16,26 @@ class Event(abc.ABC, BaseModel):
|
|||||||
|
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
def get_type(self) -> str:
|
def get_type(self) -> str:
|
||||||
"""
|
"""获取事件类型的方法,类型通常为 NoneBot 内置的四种类型。"""
|
||||||
获取事件类型的方法,类型通常为 NoneBot 内置的四种类型。
|
|
||||||
"""
|
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
def get_event_name(self) -> str:
|
def get_event_name(self) -> str:
|
||||||
"""
|
"""获取事件名称的方法。"""
|
||||||
获取事件名称的方法。
|
|
||||||
"""
|
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
def get_event_description(self) -> str:
|
def get_event_description(self) -> str:
|
||||||
"""
|
"""获取事件描述的方法,通常为事件具体内容。"""
|
||||||
获取事件描述的方法,通常为事件具体内容。
|
|
||||||
"""
|
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
def __str__(self) -> str:
|
def __str__(self) -> str:
|
||||||
return f"[{self.get_event_name()}]: {self.get_event_description()}"
|
return f"[{self.get_event_name()}]: {self.get_event_description()}"
|
||||||
|
|
||||||
def get_log_string(self) -> str:
|
def get_log_string(self) -> str:
|
||||||
"""
|
"""获取事件日志信息的方法。
|
||||||
获取事件日志信息的方法,通常你不需要修改这个方法,只有当希望 NoneBot 隐藏该事件日志时,可以抛出 `NoLogException` 异常。
|
|
||||||
|
通常你不需要修改这个方法,只有当希望 NoneBot 隐藏该事件日志时,可以抛出 `NoLogException` 异常。
|
||||||
|
|
||||||
异常:
|
异常:
|
||||||
NoLogException
|
NoLogException
|
||||||
@ -49,34 +44,27 @@ class Event(abc.ABC, BaseModel):
|
|||||||
|
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
def get_user_id(self) -> str:
|
def get_user_id(self) -> str:
|
||||||
"""
|
"""获取事件主体 id 的方法,通常是用户 id 。"""
|
||||||
获取事件主体 id 的方法,通常是用户 id 。
|
|
||||||
"""
|
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
def get_session_id(self) -> str:
|
def get_session_id(self) -> str:
|
||||||
"""
|
"""获取会话 id 的方法,用于判断当前事件属于哪一个会话,通常是用户 id、群组 id 组合。"""
|
||||||
获取会话 id 的方法,用于判断当前事件属于哪一个会话,通常是用户 id、群组 id 组合。
|
|
||||||
"""
|
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
def get_message(self) -> "Message":
|
def get_message(self) -> "Message":
|
||||||
"""
|
"""获取事件消息内容的方法。"""
|
||||||
获取事件消息内容的方法。
|
|
||||||
"""
|
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
def get_plaintext(self) -> str:
|
def get_plaintext(self) -> str:
|
||||||
"""
|
"""获取消息纯文本的方法。
|
||||||
获取消息纯文本的方法,通常不需要修改,默认通过 `get_message().extract_plain_text` 获取。
|
|
||||||
|
通常不需要修改,默认通过 `get_message().extract_plain_text` 获取。
|
||||||
"""
|
"""
|
||||||
return self.get_message().extract_plain_text()
|
return self.get_message().extract_plain_text()
|
||||||
|
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
def is_tome(self) -> bool:
|
def is_tome(self) -> bool:
|
||||||
"""
|
"""获取事件是否与机器人有关的方法。"""
|
||||||
获取事件是否与机器人有关的方法。
|
|
||||||
"""
|
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
@ -25,17 +25,14 @@ class MessageSegment(Mapping, abc.ABC, Generic[TM]):
|
|||||||
"""消息段基类"""
|
"""消息段基类"""
|
||||||
|
|
||||||
type: str
|
type: str
|
||||||
"""
|
"""消息段类型"""
|
||||||
消息段类型
|
|
||||||
"""
|
|
||||||
data: Dict[str, Any] = field(default_factory=lambda: {})
|
data: Dict[str, Any] = field(default_factory=lambda: {})
|
||||||
"""
|
"""消息段数据"""
|
||||||
消息段数据
|
|
||||||
"""
|
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
def get_message_class(cls) -> Type[TM]:
|
def get_message_class(cls) -> Type[TM]:
|
||||||
|
"""获取消息数组类型"""
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
@ -84,11 +81,16 @@ class MessageSegment(Mapping, abc.ABC, Generic[TM]):
|
|||||||
|
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
def is_text(self) -> bool:
|
def is_text(self) -> bool:
|
||||||
|
"""当前消息段是否为纯文本"""
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
|
|
||||||
class Message(List[TMS], abc.ABC):
|
class Message(List[TMS], abc.ABC):
|
||||||
"""消息数组"""
|
"""消息数组
|
||||||
|
|
||||||
|
参数:
|
||||||
|
message: 消息内容
|
||||||
|
"""
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self: TM,
|
self: TM,
|
||||||
@ -96,10 +98,6 @@ class Message(List[TMS], abc.ABC):
|
|||||||
*args,
|
*args,
|
||||||
**kwargs,
|
**kwargs,
|
||||||
):
|
):
|
||||||
"""
|
|
||||||
参数:
|
|
||||||
message: 消息内容
|
|
||||||
"""
|
|
||||||
super().__init__(*args, **kwargs)
|
super().__init__(*args, **kwargs)
|
||||||
if message is None:
|
if message is None:
|
||||||
return
|
return
|
||||||
@ -112,8 +110,9 @@ class Message(List[TMS], abc.ABC):
|
|||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def template(cls: Type[TM], format_string: Union[str, TM]) -> MessageTemplate[TM]:
|
def template(cls: Type[TM], format_string: Union[str, TM]) -> MessageTemplate[TM]:
|
||||||
"""
|
"""创建消息模板。
|
||||||
根据创建消息模板, 用法和 `str.format` 大致相同, 但是可以输出消息对象, 并且支持以 `Message` 对象作为消息模板
|
|
||||||
|
用法和 `str.format` 大致相同, 但是可以输出消息对象, 并且支持以 `Message` 对象作为消息模板
|
||||||
|
|
||||||
并且提供了拓展的格式化控制符, 可以用适用于该消息类型的 `MessageSegment` 的工厂方法创建消息
|
并且提供了拓展的格式化控制符, 可以用适用于该消息类型的 `MessageSegment` 的工厂方法创建消息
|
||||||
|
|
||||||
@ -137,13 +136,14 @@ class Message(List[TMS], abc.ABC):
|
|||||||
format_string: 格式化字符串
|
format_string: 格式化字符串
|
||||||
|
|
||||||
返回:
|
返回:
|
||||||
MessageFormatter[TM]: 消息格式化器
|
消息格式化器
|
||||||
"""
|
"""
|
||||||
return MessageTemplate(format_string, cls)
|
return MessageTemplate(format_string, cls)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
def get_segment_class(cls) -> Type[TMS]:
|
def get_segment_class(cls) -> Type[TMS]:
|
||||||
|
"""获取消息段类型"""
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
def __str__(self):
|
def __str__(self):
|
||||||
@ -160,6 +160,7 @@ class Message(List[TMS], abc.ABC):
|
|||||||
@staticmethod
|
@staticmethod
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
def _construct(msg: Union[str, Mapping, Iterable[Mapping], Any]) -> Iterable[TMS]:
|
def _construct(msg: Union[str, Mapping, Iterable[Mapping], Any]) -> Iterable[TMS]:
|
||||||
|
"""构造消息数组"""
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
def __add__(self: TM, other: Union[str, Mapping, Iterable[Mapping]]) -> TM:
|
def __add__(self: TM, other: Union[str, Mapping, Iterable[Mapping]]) -> TM:
|
||||||
@ -181,8 +182,7 @@ class Message(List[TMS], abc.ABC):
|
|||||||
return self
|
return self
|
||||||
|
|
||||||
def append(self: TM, obj: Union[str, TMS]) -> TM:
|
def append(self: TM, obj: Union[str, TMS]) -> TM:
|
||||||
"""
|
"""添加一个消息段到消息数组末尾。
|
||||||
添加一个消息段到消息数组末尾
|
|
||||||
|
|
||||||
参数:
|
参数:
|
||||||
obj: 要添加的消息段
|
obj: 要添加的消息段
|
||||||
@ -196,8 +196,7 @@ class Message(List[TMS], abc.ABC):
|
|||||||
return self
|
return self
|
||||||
|
|
||||||
def extend(self: TM, obj: Union[TM, Iterable[TMS]]) -> TM:
|
def extend(self: TM, obj: Union[TM, Iterable[TMS]]) -> TM:
|
||||||
"""
|
"""拼接一个消息数组或多个消息段到消息数组末尾。
|
||||||
拼接一个消息数组或多个消息段到消息数组末尾
|
|
||||||
|
|
||||||
参数:
|
参数:
|
||||||
obj: 要添加的消息数组
|
obj: 要添加的消息数组
|
||||||
@ -210,8 +209,9 @@ class Message(List[TMS], abc.ABC):
|
|||||||
return deepcopy(self)
|
return deepcopy(self)
|
||||||
|
|
||||||
def extract_plain_text(self: "Message[MessageSegment]") -> str:
|
def extract_plain_text(self: "Message[MessageSegment]") -> str:
|
||||||
"""
|
"""提取消息内纯文本消息"""
|
||||||
提取消息内纯文本消息
|
|
||||||
"""
|
|
||||||
|
|
||||||
return "".join(str(seg) for seg in self if seg.is_text())
|
return "".join(str(seg) for seg in self if seg.is_text())
|
||||||
|
|
||||||
|
|
||||||
|
__autodoc__ = {"MessageSegment.__str__": True}
|
||||||
|
@ -31,7 +31,12 @@ FormatSpecFunc_T = TypeVar("FormatSpecFunc_T", bound=FormatSpecFunc)
|
|||||||
|
|
||||||
|
|
||||||
class MessageTemplate(Formatter, Generic[TF]):
|
class MessageTemplate(Formatter, Generic[TF]):
|
||||||
"""消息模板格式化实现类"""
|
"""消息模板格式化实现类。
|
||||||
|
|
||||||
|
参数:
|
||||||
|
template: 模板
|
||||||
|
factory: 消息构造类型,默认为 `str`
|
||||||
|
"""
|
||||||
|
|
||||||
@overload
|
@overload
|
||||||
def __init__(
|
def __init__(
|
||||||
@ -46,13 +51,6 @@ class MessageTemplate(Formatter, Generic[TF]):
|
|||||||
...
|
...
|
||||||
|
|
||||||
def __init__(self, template, factory=str) -> None:
|
def __init__(self, template, factory=str) -> None:
|
||||||
"""
|
|
||||||
创建一个模板
|
|
||||||
|
|
||||||
参数:
|
|
||||||
template: 模板
|
|
||||||
factory: 消息构造类型,默认为 `str`
|
|
||||||
"""
|
|
||||||
self.template: TF = template
|
self.template: TF = template
|
||||||
self.factory: Type[TF] = factory
|
self.factory: Type[TF] = factory
|
||||||
self.format_specs: Dict[str, FormatSpecFunc] = {}
|
self.format_specs: Dict[str, FormatSpecFunc] = {}
|
||||||
@ -67,9 +65,7 @@ class MessageTemplate(Formatter, Generic[TF]):
|
|||||||
return spec
|
return spec
|
||||||
|
|
||||||
def format(self, *args: Any, **kwargs: Any) -> TF:
|
def format(self, *args: Any, **kwargs: Any) -> TF:
|
||||||
"""
|
"""根据模板和参数生成消息对象"""
|
||||||
根据模板和参数生成消息对象
|
|
||||||
"""
|
|
||||||
msg = self.factory()
|
msg = self.factory()
|
||||||
if isinstance(self.template, str):
|
if isinstance(self.template, str):
|
||||||
msg += self.vformat(self.template, args, kwargs)
|
msg += self.vformat(self.template, args, kwargs)
|
||||||
|
2
poetry.lock
generated
2
poetry.lock
generated
@ -562,7 +562,7 @@ six = ">=1.6.1"
|
|||||||
type = "git"
|
type = "git"
|
||||||
url = "https://github.com/nonebot/nb-autodoc.git"
|
url = "https://github.com/nonebot/nb-autodoc.git"
|
||||||
reference = "master"
|
reference = "master"
|
||||||
resolved_reference = "35bcf0e5c41aa59aa923e201b3935ea96afc6273"
|
resolved_reference = "9371179f1ae4a1659ae789187b1dd323b7751a02"
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "nodeenv"
|
name = "nodeenv"
|
||||||
|
Loading…
x
Reference in New Issue
Block a user