Module mirai.models

此模块提供 model 层封装。

model 层架构在 bus 和 adapters 之上，将 mirai-api-http 传回的原始数据解析为 Python 对象， 并支持用 Python 对象作为参数调用 API。

model 层使用 pydantic 进行数据解析。

Expand source code 
# -*- coding: utf-8 -*-
"""
此模块提供 model 层封装。

model 层架构在 bus 和 adapters 之上，将 mirai-api-http 传回的原始数据解析为 Python 对象，
并支持用 Python 对象作为参数调用 API。

model 层使用 pydantic 进行数据解析。
"""
from mirai.models.bus import ModelEventBus
from mirai.models.entities import (
    Client, Entity, Friend, Group, GroupMember, Permission
)
from mirai.models.events import (
    BotEvent, BotGroupPermissionChangeEvent, BotInvitedJoinGroupRequestEvent,
    BotJoinGroupEvent, BotLeaveEventActive, BotLeaveEventKick, BotMuteEvent,
    BotOfflineEventActive, BotOfflineEventDropped, BotOfflineEventForce,
    BotOnlineEvent, BotReloginEvent, BotUnmuteEvent, ClientKind, CommandEvent,
    CommandExecutedEvent, Event, FriendEvent, FriendInputStatusChangedEvent,
    FriendMessage, FriendNickChangedEvent, FriendRecallEvent,
    FriendSyncMessage, GroupAllowAnonymousChatEvent,
    GroupAllowConfessTalkEvent, GroupAllowMemberInviteEvent,
    GroupEntranceAnnouncementChangeEvent, GroupEvent, GroupMessage,
    GroupMuteAllEvent, GroupNameChangeEvent, GroupRecallEvent,
    GroupSyncMessage, MemberCardChangeEvent, MemberHonorChangeEvent,
    MemberJoinEvent, MemberJoinRequestEvent, MemberLeaveEventKick,
    MemberLeaveEventQuit, MemberMuteEvent, MemberPermissionChangeEvent,
    MemberSpecialTitleChangeEvent, MemberUnmuteEvent, MessageEvent,
    NewFriendRequestEvent, NudgeEvent, OtherClientEvent, OtherClientMessage,
    OtherClientOfflineEvent, OtherClientOnlineEvent, RequestEvent,
    StrangerMessage, StrangerSyncMessage, TempMessage, TempSyncMessage,
    BotLeaveEventDisband
)
from mirai.models.message import (
    App, At, AtAll, Dice, Face, File, FlashImage, Forward, ForwardMessageNode,
    Image, Json, MarketFace, MessageChain, MessageComponent, MiraiCode,
    MusicShare, MusicShareKind, Plain, Poke, PokeNames, Quote, Source, Unknown,
    Voice, Xml, deserialize, serialize
)

__all__ = [
    'Entity',
    'Friend',
    'Group',
    'GroupMember',
    'Permission',
    'Client',
    'BotEvent',
    'BotGroupPermissionChangeEvent',
    'BotInvitedJoinGroupRequestEvent',
    'BotJoinGroupEvent',
    'BotLeaveEventActive',
    'BotLeaveEventKick',
    'BotMuteEvent',
    'BotOfflineEventActive',
    'BotOfflineEventDropped',
    'BotOfflineEventForce',
    'BotOnlineEvent',
    'BotReloginEvent',
    'BotUnmuteEvent',
    'CommandEvent',
    'CommandExecutedEvent',
    'Event',
    'FriendEvent',
    'FriendSyncMessage',
    'GroupSyncMessage',
    'StrangerSyncMessage',
    'TempSyncMessage',
    'FriendInputStatusChangedEvent',
    'FriendMessage',
    'FriendNickChangedEvent',
    'FriendRecallEvent',
    'GroupAllowAnonymousChatEvent',
    'GroupAllowConfessTalkEvent',
    'GroupAllowMemberInviteEvent',
    'GroupEntranceAnnouncementChangeEvent',
    'GroupEvent',
    'GroupMessage',
    'GroupMuteAllEvent',
    'GroupNameChangeEvent',
    'GroupRecallEvent',
    'MemberCardChangeEvent',
    'MemberHonorChangeEvent',
    'MemberJoinEvent',
    'MemberJoinRequestEvent',
    'MemberLeaveEventKick',
    'MemberLeaveEventQuit',
    'MemberMuteEvent',
    'MemberPermissionChangeEvent',
    'MemberSpecialTitleChangeEvent',
    'MemberUnmuteEvent',
    'MessageEvent',
    'NudgeEvent',
    'NewFriendRequestEvent',
    'OtherClientEvent',
    'OtherClientMessage',
    'OtherClientOnlineEvent',
    'OtherClientOfflineEvent',
    'ClientKind',
    'RequestEvent',
    'StrangerMessage',
    'TempMessage',
    'App',
    'At',
    'AtAll',
    'Dice',
    'Face',
    'File',
    'FlashImage',
    'Forward',
    'ForwardMessageNode',
    'Image',
    'Json',
    'MessageChain',
    'MessageComponent',
    'MiraiCode',
    'MusicShareKind',
    'MusicShare',
    'MarketFace',
    'Plain',
    'PokeNames',
    'Poke',
    'Quote',
    'Source',
    'Unknown',
    'Voice',
    'Xml',
    'serialize',
    'deserialize',
    'ModelEventBus',
    'BotLeaveEventDisband'
]

Sub-modules

mirai.models.api

此模块提供 API 调用与返回数据解析相关。

mirai.models.base

此模块提供 YiriMirai 中使用的 pydantic 模型的基类。

mirai.models.bus

此模块提供模型事件总线相关。 …

mirai.models.entities

此模块提供实体和配置项模型。

mirai.models.events

此模块提供事件模型。

mirai.models.message

此模块提供消息链相关。

Functions

def deserialize(s: str) ‑> str

mirai 码去转义。

Args

s
待去转义的字符串。

Returns

str
去转义后的字符串。
Expand source code 
def deserialize(s: str) -> str:
    """mirai 码去转义。

    Args:
        s: 待去转义的字符串。

    Returns:
        str: 去转义后的字符串。
    """
    return re.sub(
        r'\\([\[\]:,\\])', lambda match: match.group(1),
        s.replace('\\n', '\n').replace('\\r', '\r')
    )
def serialize(s: str) ‑> str

mirai 码转义。

Args

s
待转义的字符串。

Returns

str
去转义后的字符串。
Expand source code 
def serialize(s: str) -> str:
    """mirai 码转义。

    Args:
        s: 待转义的字符串。

    Returns:
        str: 去转义后的字符串。
    """
    return re.sub(r'[\[\]:,\\]', lambda match: '\\' + match.group(0),
                  s).replace('\n', '\\n').replace('\r', '\\r')

Classes

class App (*args, **kwargs)

应用。

Expand source code 
class App(MessageComponent):
    """应用。"""
    type: str = "App"
    """消息组件类型。"""
    content: str
    """内容。"""
    def as_json(self):
        from json import loads as json_loads
        return json_loads(self.content)

    def __str__(self):
        try:
            json = self.as_json()
            return json.get('prompt', '[应用消息]')
        except:
            return '[应用消息]'

    def as_mirai_code(self) -> str:
        return f'[mirai:app:{serialize(self.content)}]'

Ancestors

Class variables

var content : str

内容。

Methods

def as_json(self)
Expand source code 
def as_json(self):
    from json import loads as json_loads
    return json_loads(self.content)

Inherited members

class At (*args, **kwargs)

At某人。

Expand source code 
class At(MessageComponent):
    """At某人。"""
    type: str = "At"
    """消息组件类型。"""
    target: int
    """群员 QQ 号。"""
    display: Optional[str] = None
    """At时显示的文字，发送消息时无效，自动使用群名片。"""
    def __eq__(self, other):
        return isinstance(other, At) and self.target == other.target

    def __str__(self):
        return f"@{self.display or self.target}"

    def as_mirai_code(self) -> str:
        return f"[mirai:at:{self.target}]"

Ancestors

Class variables

var display : Optional[str]

At时显示的文字，发送消息时无效，自动使用群名片。

var target : int

群员 QQ 号。

Inherited members

class AtAll (*args, **kwargs)

At全体。

Expand source code 
class AtAll(MessageComponent):
    """At全体。"""
    type: str = "AtAll"
    """消息组件类型。"""
    def __str__(self):
        return "@全体成员"

    def as_mirai_code(self) -> str:
        return f"[mirai:atall]"

Ancestors

Inherited members

class BotEvent (*args, **kwargs)

Bot 自身事件。

Args

type
事件名。
qq
Bot 的 QQ 号。
Expand source code 
class BotEvent(Event):
    """Bot 自身事件。

    Args:
        type: 事件名。
        qq: Bot 的 QQ 号。
    """
    type: str
    """事件名。"""
    qq: int
    """Bot 的 QQ 号。"""

Ancestors

Subclasses

Class variables

var qq : int

Bot 的 QQ 号。

Inherited members

class BotGroupPermissionChangeEvent (*args, **kwargs)

Bot在群里的权限被改变。

Args

type
事件名。
origin
原权限。
current
新权限。
group
事件对应的群。
Expand source code 
class BotGroupPermissionChangeEvent(GroupEvent):
    """ Bot在群里的权限被改变。

    Args:
        type: 事件名。
        origin: 原权限。
        current: 新权限。
        group: 事件对应的群。
    """
    type: str = 'BotGroupPermissionChangeEvent'
    """事件名。"""
    origin: Permission
    """原权限。"""
    current: Permission
    """新权限。"""
    group: Group
    """事件对应的群。"""

Ancestors

Class variables

var current : Permission

新权限。

var group : Group

事件对应的群。

var origin : Permission

原权限。

Inherited members

class BotInvitedJoinGroupRequestEvent (*args, **kwargs)

Bot 被邀请入群申请。

Args

type
事件名。
event_id
事件标识，响应该事件时的标识。
from_id
邀请人 QQ 号。
group_id
被邀请进入群的群号。
group_name
被邀请进入群的群名称。
nick
邀请人（好友）的昵称。
message
邀请消息。
Expand source code 
class BotInvitedJoinGroupRequestEvent(RequestEvent):
    """Bot 被邀请入群申请。

    Args:
        type: 事件名。
        event_id: 事件标识，响应该事件时的标识。
        from_id: 邀请人 QQ 号。
        group_id: 被邀请进入群的群号。
        group_name: 被邀请进入群的群名称。
        nick: 邀请人（好友）的昵称。
        message: 邀请消息。
    """
    type: str = 'BotInvitedJoinGroupRequestEvent'
    """事件名。"""
    event_id: int
    """事件标识，响应该事件时的标识。"""
    from_id: int
    """邀请人 QQ 号。"""
    group_id: int
    """被邀请进入群的群号。"""
    group_name: str
    """被邀请进入群的群名称。"""
    nick: str
    """邀请人（好友）的昵称。"""
    message: str
    """邀请消息。"""

Ancestors

Class variables

var group_name : str

被邀请进入群的群名称。

var message : str

邀请消息。

var nick : str

邀请人（好友）的昵称。

Inherited members

class BotJoinGroupEvent (*args, **kwargs)

Bot 加入了一个新群。

Args

type
事件名。
group
Bot 加入的群。
Expand source code 
class BotJoinGroupEvent(GroupEvent):
    """Bot 加入了一个新群。

    Args:
        type: 事件名。
        group: Bot 加入的群。
    """
    type: str = 'BotJoinGroupEvent'
    """事件名。"""
    group: Group
    """Bot 加入的群。"""
    invitor: Optional[GroupMember] = None
    """邀请者。"""

Ancestors

Class variables

var group : Group

Bot 加入的群。

var invitor : Optional[GroupMember]

邀请者。

Inherited members

class BotLeaveEventActive (*args, **kwargs)

Bot 主动退出一个群。

Args

type
事件名。
group
Bot 退出的群。
Expand source code 
class BotLeaveEventActive(GroupEvent):
    """Bot 主动退出一个群。

    Args:
        type: 事件名。
        group: Bot 退出的群。
    """
    type: str = 'BotLeaveEventActive'
    """事件名。"""
    group: Group
    """Bot 退出的群。"""

Ancestors

Class variables

var group : Group

Bot 退出的群。

Inherited members

class BotLeaveEventDisband (*args, **kwargs)

Bot 因为群解散而离群

Args

group
群对象
operator
操作人
Expand source code 
class BotLeaveEventDisband(GroupEvent):
    """Bot 因为群解散而离群

    Args:
        group: 群对象
        operator: 操作人
    """
    type: str = 'BotLeaveEventDisband'
    """事件名。"""
    group: Group
    """被解散的群聊"""
    operator: Optional[GroupMember]
    """操作者"""

Ancestors

Class variables

var group : Group

被解散的群聊

var operator : Optional[GroupMember]

操作者

Inherited members

class BotLeaveEventKick (*args, **kwargs)

Bot 被踢出一个群。

Args

type
事件名。
group
Bot 被踢出的群。
Expand source code 
class BotLeaveEventKick(GroupEvent):
    """Bot 被踢出一个群。

    Args:
        type: 事件名。
        group: Bot 被踢出的群。
    """
    type: str = 'BotLeaveEventKick'
    """事件名。"""
    group: Group
    """Bot 被踢出的群。"""
    operator: Optional[GroupMember]
    """踢出 Bot 的管理员。"""

Ancestors

Class variables

var group : Group

Bot 被踢出的群。

var operator : Optional[GroupMember]

踢出 Bot 的管理员。

Inherited members

class BotMuteEvent (*args, **kwargs)

Bot 被禁言。

Args

type
事件名。
duration_seconds
禁言时间，单位秒。
operator
禁言的操作者。
Expand source code 
class BotMuteEvent(GroupEvent):
    """Bot 被禁言。

    Args:
        type: 事件名。
        duration_seconds: 禁言时间，单位秒。
        operator: 禁言的操作者。
    """
    type: str = 'BotMuteEvent'
    """事件名。"""
    duration_seconds: int
    """禁言时间，单位秒。"""
    operator: Optional[GroupMember]
    """禁言的操作者。"""

Ancestors

Class variables

var duration_seconds : int

禁言时间，单位秒。

var operator : Optional[GroupMember]

禁言的操作者。

Inherited members

class BotOfflineEventActive (*args, **kwargs)

Bot 主动离线。

Args

type
事件名。
qq
主动离线的 Bot 的 QQ 号。
Expand source code 
class BotOfflineEventActive(BotEvent):
    """Bot 主动离线。

    Args:
        type: 事件名。
        qq: 主动离线的 Bot 的 QQ 号。
    """
    type: str = 'BotOfflineEventActive'
    """事件名。"""
    qq: int
    """主动离线的 Bot 的 QQ 号。"""

Ancestors

Inherited members

class BotOfflineEventDropped (*args, **kwargs)

Bot 被服务器断开或因网络问题而掉线。

Args

type
事件名。
qq
被服务器断开或因网络问题而掉线的 Bot 的 QQ 号。
Expand source code 
class BotOfflineEventDropped(BotEvent):
    """Bot 被服务器断开或因网络问题而掉线。

    Args:
        type: 事件名。
        qq: 被服务器断开或因网络问题而掉线的 Bot 的 QQ 号。
    """
    type: str = 'BotOfflineEventDropped'
    """事件名。"""
    qq: int
    """被服务器断开或因网络问题而掉线的 Bot 的 QQ 号。"""

Ancestors

Inherited members

class BotOfflineEventForce (*args, **kwargs)

Bot被挤下线。

Args

type
事件名。
qq
被挤下线的 Bot 的 QQ 号。
Expand source code 
class BotOfflineEventForce(BotEvent):
    """Bot被挤下线。

    Args:
        type: 事件名。
        qq: 被挤下线的 Bot 的 QQ 号。
    """
    type: str = 'BotOfflineEventForce'
    """事件名。"""
    qq: int
    """被挤下线的 Bot 的 QQ 号。"""

Ancestors

Inherited members

class BotOnlineEvent (*args, **kwargs)

Bot 登陆成功。

Args

type
事件名。
qq
登陆成功的 Bot 的 QQ 号。
Expand source code 
class BotOnlineEvent(BotEvent):
    """Bot 登陆成功。

    Args:
        type: 事件名。
        qq: 登陆成功的 Bot 的 QQ 号。
    """
    type: str = 'BotOnlineEvent'
    """事件名。"""
    qq: int
    """登陆成功的 Bot 的 QQ 号。"""

Ancestors

Inherited members

class BotReloginEvent (*args, **kwargs)

Bot 主动重新登录。

Args

type
事件名。
qq
主动重新登录的 Bot 的 QQ 号。
Expand source code 
class BotReloginEvent(BotEvent):
    """Bot 主动重新登录。

    Args:
        type: 事件名。
        qq: 主动重新登录的 Bot 的 QQ 号。
    """
    type: str = 'BotReloginEvent'
    """事件名。"""
    qq: int
    """主动重新登录的 Bot 的 QQ 号。"""

Ancestors

Inherited members

class BotUnmuteEvent (*args, **kwargs)

Bot 被取消禁言。

Args

type
事件名。
operator
取消禁言的操作者。
Expand source code 
class BotUnmuteEvent(GroupEvent):
    """Bot 被取消禁言。

    Args:
        type: 事件名。
        operator: 取消禁言的操作者。
    """
    type: str = 'BotUnmuteEvent'
    """事件名。"""
    operator: Optional[GroupMember]
    """取消禁言的操作者。"""

Ancestors

Class variables

var operator : Optional[GroupMember]

取消禁言的操作者。

Inherited members

class Client (*args, **kwargs)

来自其他客户端的用户。

Expand source code 
class Client(Entity):
    """来自其他客户端的用户。"""
    id: int
    """识别 id。"""
    platform: str
    """来源平台。"""
    def get_avatar_url(self) -> str:
        raise NotImplementedError

    def get_name(self) -> str:
        return self.platform

Ancestors

Class variables

var platform : str

来源平台。

Inherited members

class ClientKind (value, names=None, *, module=None, qualname=None, type=None, start=1)

详细设备类型。

Expand source code 
class ClientKind(int, Enum):
    """详细设备类型。"""
    ANDROID_PAD = 68104
    AOL_CHAOJIHUIYUAN = 73730
    AOL_HUIYUAN = 73474
    AOL_SQQ = 69378
    CAR = 65806
    HRTX_IPHONE = 66566
    HRTX_PC = 66561
    MC_3G = 65795
    MISRO_MSG = 69634
    MOBILE_ANDROID = 65799
    MOBILE_ANDROID_NEW = 72450
    MOBILE_HD = 65805
    MOBILE_HD_NEW = 71426
    MOBILE_IPAD = 68361
    MOBILE_IPAD_NEW = 72194
    MOBILE_IPHONE = 67586
    MOBILE_OTHER = 65794
    MOBILE_PC_QQ = 65793
    MOBILE_PC_TIM = 77313
    MOBILE_WINPHONE_NEW = 72706
    QQ_FORELDER = 70922
    QQ_SERVICE = 71170
    TV_QQ = 69130
    WIN8 = 69899
    WINPHONE = 65804

Ancestors

  • builtins.int
  • enum.Enum

Class variables

var ANDROID_PAD
var AOL_CHAOJIHUIYUAN
var AOL_HUIYUAN
var AOL_SQQ
var CAR
var HRTX_IPHONE
var HRTX_PC
var MC_3G
var MISRO_MSG
var MOBILE_ANDROID
var MOBILE_ANDROID_NEW
var MOBILE_HD
var MOBILE_HD_NEW
var MOBILE_IPAD
var MOBILE_IPAD_NEW
var MOBILE_IPHONE
var MOBILE_OTHER
var MOBILE_PC_QQ
var MOBILE_PC_TIM
var MOBILE_WINPHONE_NEW
var QQ_FORELDER
var QQ_SERVICE
var TV_QQ
var WIN8
var WINPHONE
class CommandEvent (*args, **kwargs)

命令事件。

Args

type
事件名。
Expand source code 
class CommandEvent(Event):
    """命令事件。

    Args:
        type: 事件名。
    """
    type: str
    """事件名。"""

Ancestors

Subclasses

Inherited members

class CommandExecutedEvent (*args, **kwargs)

命令被执行。

Args

type
事件名。
name
命令名称。
friend
发送命令的好友, 从控制台发送为 None。
member
发送命令的群成员, 从控制台发送为 None。
args
命令执行时的参数。
Expand source code 
class CommandExecutedEvent(CommandEvent):
    """命令被执行。

    Args:
        type: 事件名。
        name: 命令名称。
        friend: 发送命令的好友, 从控制台发送为 None。
        member: 发送命令的群成员, 从控制台发送为 None。
        args: 命令执行时的参数。
    """
    type: str = 'CommandExecutedEvent'
    """事件名。"""
    name: str
    """命令名称。"""
    friend: Optional[Friend]
    """发送命令的好友, 从控制台发送为 None。"""
    member: Optional[GroupMember]
    """发送命令的群成员, 从控制台发送为 None。"""
    args: MessageChain
    """命令执行时的参数。"""

Ancestors

Class variables

var args : MessageChain

命令执行时的参数。

var friend : Optional[Friend]

发送命令的好友, 从控制台发送为 None。

var member : Optional[GroupMember]

发送命令的群成员, 从控制台发送为 None。

var name : str

命令名称。

Inherited members

class Dice (*args, **kwargs)

骰子。

Expand source code 
class Dice(MessageComponent):
    """骰子。"""
    type: str = "Dice"
    """消息组件类型。"""
    value: int
    """点数。"""
    def __str__(self):
        return f'[骰子{self.value}]'

    def as_mirai_code(self) -> str:
        return f'[mirai:dice:{self.value}]'

Ancestors

Class variables

var value : int

点数。

Inherited members

class Entity (*args, **kwargs)

实体，表示一个用户或群。

Expand source code 
class Entity(MiraiBaseModel):
    """实体，表示一个用户或群。"""
    id: int
    """QQ 号或群号。"""
    @abc.abstractmethod
    def get_avatar_url(self) -> str:
        """头像图片链接。"""

    @abc.abstractmethod
    def get_name(self) -> str:
        """名称。"""

Ancestors

  • MiraiBaseModel
  • pydantic.main.BaseModel
  • pydantic.utils.Representation

Subclasses

Class variables

var id : int

QQ 号或群号。

Methods

def get_avatar_url(self) ‑> str

头像图片链接。

Expand source code 
@abc.abstractmethod
def get_avatar_url(self) -> str:
    """头像图片链接。"""
def get_name(self) ‑> str

名称。

Expand source code 
@abc.abstractmethod
def get_name(self) -> str:
    """名称。"""
class Event (*args, **kwargs)

事件基类。

Args

type
事件名。
Expand source code 
class Event(MiraiIndexedModel):
    """事件基类。

    Args:
        type: 事件名。
    """
    type: str
    """事件名。"""
    def __repr__(self):
        return self.__class__.__name__ + '(' + ', '.join(
            (
                f'{k}={repr(v)}'
                for k, v in self.__dict__.items() if k != 'type' and v
            )
        ) + ')'

    @classmethod
    def parse_subtype(cls, obj: dict) -> 'Event':
        try:
            return cast(Event, super().parse_subtype(obj))
        except ValueError:
            return Event(type=obj['type'])

    @classmethod
    def get_subtype(cls, name: str) -> Type['Event']:
        try:
            return cast(Type[Event], super().get_subtype(name))
        except ValueError:
            return Event

Ancestors

Subclasses

Class variables

var type : str

事件名。

Inherited members

class Face (*args, **kwargs)

表情。

Expand source code 
class Face(MessageComponent):
    """表情。"""
    type: str = "Face"
    """消息组件类型。"""
    face_id: Optional[int] = None
    """QQ 表情编号，可选，优先度高于 name。"""
    name: Optional[str] = None
    """QQ表情名称，可选。"""
    def __init__(self, *args, **kwargs):
        if len(args) == 1:
            if isinstance(args[0], str):
                self.name = args[0]
            elif isinstance(args[0], int):
                self.face_id = args[0]
            super().__init__(**kwargs)
        super().__init__(*args, **kwargs)

    def __eq__(self, other):
        return isinstance(other, Face) and \
            (self.face_id == other.face_id or self.name == other.name)

    def __str__(self):
        if self.name:
            return f'[{self.name}]'
        return '[表情]'

    def as_mirai_code(self):
        return f"[mirai:face:{self.face_id}]"

Ancestors

Class variables

var face_id : Optional[int]

QQ 表情编号，可选，优先度高于 name。

var name : Optional[str]

QQ表情名称，可选。

Inherited members

class File (*args, **kwargs)

文件。

Expand source code 
class File(MessageComponent):
    """文件。"""
    type: str = "File"
    """消息组件类型。"""
    id: str
    """文件识别 ID。"""
    name: str
    """文件名称。"""
    size: int
    """文件大小。"""
    def __str__(self):
        return f'[文件]{self.name}'

Ancestors

Class variables

var id : str

文件识别 ID。

var name : str

文件名称。

var size : int

文件大小。

Inherited members

class FlashImage (*args, **kwargs)

闪照。

Expand source code 
class FlashImage(Image):
    """闪照。"""
    type: str = "FlashImage"
    """消息组件类型。"""
    image_id: Optional[str] = None
    """图片的 image_id，群图片与好友图片格式不同。不为空时将忽略 url 属性。"""
    url: Optional[HttpUrl] = None
    """图片的 URL，发送时可作网络图片的链接；接收时为腾讯图片服务器的链接，可用于图片下载。"""
    path: Optional[str] = None
    """图片的路径，发送本地图片，路径相对于 `plugins/MiraiAPIHTTP/images`。"""
    base64: Optional[str] = None
    """图片的 Base64 编码。"""
    width: int
    """图片的宽度"""
    height: int
    """图片的高度"""
    def __str__(self):
        return '[闪照]'

    def as_mirai_code(self) -> str:
        return f"[mirai:flash:{self.image_id}]"

    def as_image(self) -> Image:
        return Image(self.image_id, self.url)

Ancestors

Methods

def as_image(self) ‑> Image
Expand source code 
def as_image(self) -> Image:
    return Image(self.image_id, self.url)

Inherited members

class Forward (*args, **kwargs)

合并转发。

Expand source code 
class Forward(MessageComponent):
    """合并转发。"""
    type: str = "Forward"
    """消息组件类型。"""
    display: Optional[ForwardMessageDiaplay]
    """转发组件的显示信息。"""
    node_list: List[ForwardMessageNode]
    """转发消息节点列表。"""
    def __init__(self, *args, **kwargs):
        if len(args) == 1:
            kwargs['node_list'] = args[0]
            args=()
        if 'display' not in kwargs:
            kwargs['display'] = ForwardMessageDiaplay.create(kwargs['node_list'])
        super().__init__(*args, **kwargs)

    def __str__(self):
        return '[聊天记录]'

Ancestors

Class variables

var display : Optional[mirai.models.message.ForwardMessageDiaplay]

转发组件的显示信息。

var node_list : List[ForwardMessageNode]

转发消息节点列表。

Inherited members

class ForwardMessageNode (*args, **kwargs)

合并转发中的一条消息。

Expand source code 
class ForwardMessageNode(MiraiBaseModel):
    """合并转发中的一条消息。"""
    sender_id: Optional[int] = None
    """发送人QQ号。"""
    sender_name: Optional[str] = None
    """显示名称。"""
    message_chain: Optional[MessageChain] = None
    """消息内容。"""
    message_id: Optional[int] = None
    """消息的 message_id，可以只使用此属性，从缓存中读取消息内容。"""
    time: Optional[datetime] = None
    """发送时间。"""
    @validator('message_chain', check_fields=False)
    def _validate_message_chain(cls, value: Union[MessageChain, list]):
        if isinstance(value, list):
            return MessageChain.parse_obj(value)
        return value

    @classmethod
    def create(
        cls, sender: Union[Friend, GroupMember], message: MessageChain
    ) -> 'ForwardMessageNode':
        """从消息链生成转发消息。

        Args:
            sender: 发送人。
            message: 消息内容。

        Returns:
            ForwardMessageNode: 生成的一条消息。
        """
        return ForwardMessageNode(
            sender_id=sender.id,
            sender_name=sender.get_name(),
            message_chain=message
        )

Ancestors

  • MiraiBaseModel
  • pydantic.main.BaseModel
  • pydantic.utils.Representation

Class variables

var message_chain : Optional[MessageChain]

消息内容。

var message_id : Optional[int]

消息的 message_id，可以只使用此属性，从缓存中读取消息内容。

var sender_id : Optional[int]

发送人QQ号。

var sender_name : Optional[str]

显示名称。

var time : Optional[datetime.datetime]

发送时间。

Static methods

def create(sender: Union[Friend, GroupMember], message: MessageChain) ‑> ForwardMessageNode

从消息链生成转发消息。

Args

sender
发送人。
message
消息内容。

Returns

ForwardMessageNode
生成的一条消息。
Expand source code 
@classmethod
def create(
    cls, sender: Union[Friend, GroupMember], message: MessageChain
) -> 'ForwardMessageNode':
    """从消息链生成转发消息。

    Args:
        sender: 发送人。
        message: 消息内容。

    Returns:
        ForwardMessageNode: 生成的一条消息。
    """
    return ForwardMessageNode(
        sender_id=sender.id,
        sender_name=sender.get_name(),
        message_chain=message
    )
class Friend (*args, **kwargs)

好友。

Expand source code 
class Friend(Entity):
    """好友。"""
    id: int
    """QQ 号。"""
    nickname: Optional[str]
    """昵称。"""
    remark: Optional[str]
    """备注。"""
    def get_avatar_url(self) -> str:
        return f'http://q4.qlogo.cn/g?b=qq&nk={self.id}&s=140'

    def get_name(self) -> str:
        return self.nickname or self.remark or ''

Ancestors

Class variables

var nickname : Optional[str]

昵称。

var remark : Optional[str]

备注。

Inherited members

class FriendEvent (*args, **kwargs)

好友事件。

Args

type
事件名。
friend
事件对应的好友。
Expand source code 
class FriendEvent(Event):
    """好友事件。

    Args:
        type: 事件名。
        friend: 事件对应的好友。
    """
    type: str
    """事件名。"""
    friend: Friend
    """事件对应的好友。"""

Ancestors

Subclasses

Class variables

var friend : Friend

事件对应的好友。

Inherited members

class FriendInputStatusChangedEvent (*args, **kwargs)

好友输入状态改变。

Args

type
事件名。
friend
事件对应的好友。
inputting
是否正在输入。
Expand source code 
class FriendInputStatusChangedEvent(FriendEvent):
    """好友输入状态改变。

    Args:
        type: 事件名。
        friend: 事件对应的好友。
        inputting: 是否正在输入。
    """
    type: str = 'FriendInputStatusChangedEvent'
    """事件名。"""
    friend: Friend
    """事件对应的好友。"""
    inputting: bool
    """是否正在输入。"""

Ancestors

Class variables

var inputting : bool

是否正在输入。

Inherited members

class FriendMessage (*args, **kwargs)

好友消息。

Args

type
事件名。
sender
发送消息的好友。
message_chain
消息内容。
Expand source code 
class FriendMessage(MessageEvent):
    """好友消息。

    Args:
        type: 事件名。
        sender: 发送消息的好友。
        message_chain: 消息内容。
    """
    type: str = 'FriendMessage'
    """事件名。"""
    sender: Friend
    """发送消息的好友。"""
    message_chain: MessageChain
    """消息内容。"""

Ancestors

Class variables

var sender : Friend

发送消息的好友。

Inherited members

class FriendNickChangedEvent (*args, **kwargs)

好友昵称改变。

Args

type
事件名。
friend
事件对应的好友。
from_
原昵称。
to
新昵称。
Expand source code 
class FriendNickChangedEvent(FriendEvent):
    """好友昵称改变。

    Args:
        type: 事件名。
        friend: 事件对应的好友。
        from_: 原昵称。
        to: 新昵称。
    """
    type: str = 'FriendNickChangedEvent'
    """事件名。"""
    friend: Friend
    """事件对应的好友。"""
    from_: str
    """原昵称。"""
    to: str
    """新昵称。"""

Ancestors

Class variables

var from_ : str

原昵称。

var to : str

新昵称。

Inherited members

class FriendRecallEvent (*args, **kwargs)

好友消息撤回。

Args

type
事件名。
author_id
原消息发送者的 QQ 号。
message_id
原消息 message_id。
time
原消息发送时间。
operator
好友 QQ 号或 Bot QQ 号。
Expand source code 
class FriendRecallEvent(Event):
    """好友消息撤回。

    Args:
        type: 事件名。
        author_id: 原消息发送者的 QQ 号。
        message_id: 原消息 message_id。
        time: 原消息发送时间。
        operator: 好友 QQ 号或 Bot QQ 号。
    """
    type: str = 'FriendRecallEvent'
    """事件名。"""
    # 按照文档顺序，这个事件确实应该在这个位置。
    # 而且它不符合 FriendEvent 的形式，就放在这里吧。
    author_id: int
    """原消息发送者的 QQ 号。"""
    message_id: int
    """原消息 message_id。"""
    time: datetime
    """原消息发送时间。"""
    operator: int
    """好友 QQ 号或 Bot QQ 号。"""

Ancestors

Class variables

var author_id : int

原消息发送者的 QQ 号。

var message_id : int

原消息 message_id。

var operator : int

好友 QQ 号或 Bot QQ 号。

var time : datetime.datetime

原消息发送时间。

Inherited members

class FriendSyncMessage (*args, **kwargs)

其他客户端发送的好友消息。

Args

type
事件名。
subject
发送消息的目标好友。
message_chain
消息内容。
Expand source code 
class FriendSyncMessage(MessageEvent):
    """其他客户端发送的好友消息。

    Args:
        type: 事件名。
        subject: 发送消息的目标好友。
        message_chain: 消息内容。
    """
    type: str = 'FriendSyncMessage'
    """事件名。"""
    subject: Friend
    """发送消息的目标好友。"""
    message_chain: MessageChain
    """消息内容。"""

Ancestors

Class variables

var subject : Friend

发送消息的目标好友。

Inherited members

class Group (*args, **kwargs)

群。

Expand source code 
class Group(Entity):
    """群。"""
    id: int
    """群号。"""
    name: str
    """群名称。"""
    permission: Permission
    """Bot 在群中的权限。"""
    def get_avatar_url(self) -> str:
        return f'https://p.qlogo.cn/gh/{self.id}/{self.id}/'

    def get_name(self) -> str:
        return self.name

Ancestors

Class variables

var name : str

群名称。

var permission : Permission

Bot 在群中的权限。

Inherited members

class GroupAllowAnonymousChatEvent (*args, **kwargs)

匿名聊天。

Args

type
事件名。
origin
原本是否允许匿名聊天。
current
现在是否允许匿名聊天。
group
匿名聊天状态改变的群。
operator
操作者，为 None 表示 Bot 操作。
Expand source code 
class GroupAllowAnonymousChatEvent(GroupEvent):
    """匿名聊天。

    Args:
        type: 事件名。
        origin: 原本是否允许匿名聊天。
        current: 现在是否允许匿名聊天。
        group: 匿名聊天状态改变的群。
        operator: 操作者，为 None 表示 Bot 操作。
    """
    type: str = 'GroupAllowAnonymousChatEvent'
    """事件名。"""
    origin: bool
    """原本是否允许匿名聊天。"""
    current: bool
    """现在是否允许匿名聊天。"""
    group: Group
    """匿名聊天状态改变的群。"""
    operator: Optional[GroupMember]
    """操作者，为 None 表示 Bot 操作。"""

Ancestors

Class variables

var current : bool

现在是否允许匿名聊天。

var group : Group

匿名聊天状态改变的群。

var operator : Optional[GroupMember]

操作者，为 None 表示 Bot 操作。

var origin : bool

原本是否允许匿名聊天。

Inherited members

class GroupAllowConfessTalkEvent (*args, **kwargs)

坦白说。

Args

type
事件名。
origin
原本是否允许坦白说。
current
现在是否允许坦白说。
group
坦白说状态改变的群。
is_by_bot
是否 Bot 进行该操作。
Expand source code 
class GroupAllowConfessTalkEvent(GroupEvent):
    """坦白说。

    Args:
        type: 事件名。
        origin: 原本是否允许坦白说。
        current: 现在是否允许坦白说。
        group: 坦白说状态改变的群。
        is_by_bot: 是否 Bot 进行该操作。
    """
    type: str = 'GroupAllowConfessTalkEvent'
    """事件名。"""
    origin: bool
    """原本是否允许坦白说。"""
    current: bool
    """现在是否允许坦白说。"""
    group: Group
    """坦白说状态改变的群。"""
    is_by_bot: bool
    """是否是 Bot 进行该操作。"""

Ancestors

Class variables

var current : bool

现在是否允许坦白说。

var group : Group

坦白说状态改变的群。

var is_by_bot : bool

是否是 Bot 进行该操作。

var origin : bool

原本是否允许坦白说。

Inherited members

class GroupAllowMemberInviteEvent (*args, **kwargs)

允许群员邀请好友加群。

Args

type
事件名。
origin
原本是否允许群员邀请好友加群。
current
现在是否允许群员邀请好友加群。
group
允许群员邀请好友加群状态改变的群。
operator
操作者，为 None 表示 Bot 操作。
Expand source code 
class GroupAllowMemberInviteEvent(GroupEvent):
    """允许群员邀请好友加群。

    Args:
        type: 事件名。
        origin: 原本是否允许群员邀请好友加群。
        current: 现在是否允许群员邀请好友加群。
        group: 允许群员邀请好友加群状态改变的群。
        operator: 操作者，为 None 表示 Bot 操作。
    """
    type: str = 'GroupAllowMemberInviteEvent'
    """事件名。"""
    origin: bool
    """原本是否允许群员邀请好友加群。"""
    current: bool
    """现在是否允许群员邀请好友加群。"""
    group: Group
    """允许群员邀请好友加群状态改变的群。"""
    operator: Optional[GroupMember]
    """操作者，为 None 表示 Bot 操作。"""

Ancestors

Class variables

var current : bool

现在是否允许群员邀请好友加群。

var group : Group

允许群员邀请好友加群状态改变的群。

var operator : Optional[GroupMember]

操作者，为 None 表示 Bot 操作。

var origin : bool

原本是否允许群员邀请好友加群。

Inherited members

class GroupEntranceAnnouncementChangeEvent (*args, **kwargs)

某群入群公告改变。

Args

type
事件名。
origin
原公告。
current
新公告。
group
群公告改变的群。
operator
操作者，为 None 表示 Bot 操作。
Expand source code 
class GroupEntranceAnnouncementChangeEvent(GroupEvent):
    """某群入群公告改变。

    Args:
        type: 事件名。
        origin: 原公告。
        current: 新公告。
        group: 群公告改变的群。
        operator: 操作者，为 None 表示 Bot 操作。
    """
    type: str = 'GroupEntranceAnnouncementChangeEvent'
    """事件名。"""
    origin: str
    """原公告。"""
    current: str
    """新公告。"""
    group: Group
    """群公告改变的群。"""
    operator: Optional[GroupMember]
    """操作者，为 None 表示 Bot 操作。"""

Ancestors

Class variables

var current : str

新公告。

var group : Group

群公告改变的群。

var operator : Optional[GroupMember]

操作者，为 None 表示 Bot 操作。

var origin : str

原公告。

Inherited members

class GroupEvent (*args, **kwargs)

群事件。

Args

type
事件名。
group
事件对应的群。
Expand source code 
class GroupEvent(Event):
    """群事件。

    Args:
        type: 事件名。
        group: 事件对应的群。
    """
    # group: Group
    # 一个奇怪的现象：群事件不一定有 group，它可能藏在 opeartor.group 里

    type: str
    """事件名。"""
    def __getattr__(self, name) -> Union[Group, Any]:
        if name == 'group':
            member = getattr(self, 'operator',
                             None) or getattr(self, 'member', None)
            if member:
                return member.group
        return getattr(super(), name)

Ancestors

Subclasses

Inherited members

class GroupMember (*args, **kwargs)

群成员。

Expand source code 
class GroupMember(Entity):
    """群成员。"""
    id: int
    """QQ 号。"""
    member_name: str
    """群成员名称。"""
    permission: Permission
    """Bot 在群中的权限。"""
    group: Group
    """群。"""
    special_title: str = ''
    """群头衔。"""
    join_timestamp: datetime = datetime.utcfromtimestamp(0)
    """加入群的时间。"""
    last_speak_timestamp: datetime = datetime.utcfromtimestamp(0)
    """最后一次发言的时间。"""
    mute_time_remaining: int = 0
    """禁言剩余时间。"""
    def get_avatar_url(self) -> str:
        return f'http://q4.qlogo.cn/g?b=qq&nk={self.id}&s=140'

    def get_name(self) -> str:
        return self.member_name

Ancestors

Subclasses

Class variables

var group : Group

群。

var join_timestamp : datetime.datetime

加入群的时间。

var last_speak_timestamp : datetime.datetime

最后一次发言的时间。

var member_name : str

群成员名称。

var mute_time_remaining : int

禁言剩余时间。

var permission : Permission

Bot 在群中的权限。

var special_title : str

群头衔。

Inherited members

class GroupMessage (*args, **kwargs)

群消息。

Args

type
事件名。
sender
发送消息的群成员。
message_chain
消息内容。
Expand source code 
class GroupMessage(MessageEvent):
    """群消息。

    Args:
        type: 事件名。
        sender: 发送消息的群成员。
        message_chain: 消息内容。
    """
    type: str = 'GroupMessage'
    """事件名。"""
    sender: GroupMember
    """发送消息的群成员。"""
    message_chain: MessageChain
    """消息内容。"""
    @property
    def group(self) -> Group:
        return self.sender.group

Ancestors

Class variables

var sender : GroupMember

发送消息的群成员。

Instance variables

var group : Group
Expand source code 
@property
def group(self) -> Group:
    return self.sender.group

Inherited members

class GroupMuteAllEvent (*args, **kwargs)

全员禁言。

Args

type
事件名。
origin
原本是否处于全员禁言。
current
现在是否处于全员禁言。
group
全员禁言的群。
operator
操作者，为 None 表示 Bot 操作。
Expand source code 
class GroupMuteAllEvent(GroupEvent):
    """全员禁言。

    Args:
        type: 事件名。
        origin: 原本是否处于全员禁言。
        current: 现在是否处于全员禁言。
        group: 全员禁言的群。
        operator: 操作者，为 None 表示 Bot 操作。
    """
    type: str = 'GroupMuteAllEvent'
    """事件名。"""
    origin: bool
    """原本是否处于全员禁言。"""
    current: bool
    """现在是否处于全员禁言。"""
    group: Group
    """全员禁言的群。"""
    operator: Optional[GroupMember]
    """操作者，为 None 表示 Bot 操作。"""

Ancestors

Class variables

var current : bool

现在是否处于全员禁言。

var group : Group

全员禁言的群。

var operator : Optional[GroupMember]

操作者，为 None 表示 Bot 操作。

var origin : bool

原本是否处于全员禁言。

Inherited members

class GroupNameChangeEvent (*args, **kwargs)

某个群名改变。

Args

type
事件名。
origin
原群名。
current
新群名。
group
群名改名的群。
operator
操作者，为 None 表示 Bot 操作。
Expand source code 
class GroupNameChangeEvent(GroupEvent):
    """某个群名改变。

    Args:
        type: 事件名。
        origin: 原群名。
        current: 新群名。
        group: 群名改名的群。
        operator: 操作者，为 None 表示 Bot 操作。
    """
    type: str = 'GroupNameChangeEvent'
    """事件名。"""
    origin: str
    """原群名。"""
    current: str
    """新群名。"""
    group: Group
    """群名改名的群。"""
    operator: Optional[GroupMember]
    """操作者，为 None 表示 Bot 操作。"""

Ancestors

Class variables

var current : str

新群名。

var group : Group

群名改名的群。

var operator : Optional[GroupMember]

操作者，为 None 表示 Bot 操作。

var origin : str

原群名。

Inherited members

class GroupRecallEvent (*args, **kwargs)

群消息撤回。

Args

type
事件名。
author_id
原消息发送者的 QQ 号。
message_id
原消息 message_id。
time
原消息发送时间。
group
消息撤回所在的群。
operator
消息撤回的操作者，为 None 表示 Bot 操作。
Expand source code 
class GroupRecallEvent(GroupEvent):
    """群消息撤回。

    Args:
        type: 事件名。
        author_id: 原消息发送者的 QQ 号。
        message_id: 原消息 message_id。
        time: 原消息发送时间。
        group: 消息撤回所在的群。
        operator: 消息撤回的操作者，为 None 表示 Bot 操作。
    """
    type: str = 'GroupRecallEvent'
    """事件名。"""
    author_id: int
    """原消息发送者的 QQ 号。"""
    message_id: int
    """原消息 message_id。"""
    time: datetime
    """原消息发送时间。"""
    group: Group
    """消息撤回所在的群。"""
    operator: Optional[GroupMember]
    """消息撤回的操作者，为 None 表示 Bot 操作。"""

Ancestors

Class variables

var author_id : int

原消息发送者的 QQ 号。

var group : Group

消息撤回所在的群。

var message_id : int

原消息 message_id。

var operator : Optional[GroupMember]

消息撤回的操作者，为 None 表示 Bot 操作。

var time : datetime.datetime

原消息发送时间。

Inherited members

class GroupSyncMessage (*args, **kwargs)

其他客户端发送的群消息。

Args

type
事件名。
subject
发送消息的目标群组。
message_chain
消息内容。
Expand source code 
class GroupSyncMessage(MessageEvent):
    """其他客户端发送的群消息。

    Args:
        type: 事件名。
        subject: 发送消息的目标群组。
        message_chain: 消息内容。
    """
    type: str = 'GroupSyncMessage'
    """事件名。"""
    subject: Group
    """发送消息的目标群组。"""
    message_chain: MessageChain
    """消息内容。"""
    @property
    def group(self) -> Group:
        return self.subject

Ancestors

Class variables

var subject : Group

发送消息的目标群组。

Instance variables

var group : Group
Expand source code 
@property
def group(self) -> Group:
    return self.subject

Inherited members

class Image (*args, **kwargs)

图片。

Expand source code 
class Image(MessageComponent):
    """图片。"""
    type: str = "Image"
    """消息组件类型。"""
    image_id: Optional[str] = None
    """图片的 image_id，群图片与好友图片格式不同。不为空时将忽略 url 属性。"""
    url: Optional[HttpUrl] = None
    """图片的 URL，发送时可作网络图片的链接；接收时为腾讯图片服务器的链接，可用于图片下载。"""
    path: Union[str, Path, None] = None
    """图片的路径，发送本地图片。"""
    base64: Optional[str] = None
    """图片的 Base64 编码。"""
    width: int
    """图片的宽度"""
    height: int
    """图片的高度"""
    def __eq__(self, other):
        return isinstance(
            other, Image
        ) and self.type == other.type and self.uuid == other.uuid

    def __str__(self):
        return '[图片]'

    def as_mirai_code(self) -> str:
        return f"[mirai:image:{self.image_id}]"

    @validator('path')
    def validate_path(cls, path: Union[str, Path, None]):
        """修复 path 参数的行为，使之相对于 YiriMirai 的启动路径。"""
        if path:
            try:
                return str(Path(path).resolve(strict=True))
            except FileNotFoundError:
                raise ValueError(f"无效路径：{path}")
        else:
            return path

    @property
    def uuid(self):
        image_id = self.image_id
        if image_id[0] == '{':  # 群图片
            image_id = image_id[1:37]
        elif image_id[0] == '/':  # 好友图片
            image_id = image_id[1:]
        return image_id

    def as_group_image(self) -> str:
        return f"{{{self.uuid.upper()}}}.jpg"

    def as_friend_image(self) -> str:
        return f"/{self.uuid.lower()}"

    def as_flash_image(self) -> "FlashImage":
        return FlashImage(
            image_id=self.image_id,
            url=self.url,
            path=self.path,
            base64=self.base64
        )

    async def download(
        self,
        filename: Union[str, Path, None] = None,
        directory: Union[str, Path, None] = None,
        determine_type: bool = True
    ):
        """下载图片到本地。

        Args:
            filename: 下载到本地的文件路径。与 `directory` 二选一。
            directory: 下载到本地的文件夹路径。与 `filename` 二选一。
            determine_type: 是否自动根据图片类型确定拓展名，默认为 True。
        """
        if not self.url:
            logger.warning(f'图片 `{self.uuid}` 无 url 参数，下载失败。')
            return

        import httpx
        async with httpx.AsyncClient() as client:
            response = await client.get(self.url)
            response.raise_for_status()
            content = response.content

            if filename:
                path = Path(filename)
                if determine_type:
                    import imghdr
                    path = path.with_suffix(
                        '.' + str(imghdr.what(None, content))
                    )
                path.parent.mkdir(parents=True, exist_ok=True)
            elif directory:
                import imghdr
                path = Path(directory)
                path.mkdir(parents=True, exist_ok=True)
                path = path / f'{self.uuid}.{imghdr.what(None, content)}'
            else:
                raise ValueError("请指定文件路径或文件夹路径！")

            import aiofiles
            async with aiofiles.open(path, 'wb') as f:
                await f.write(content)

            return path

    @classmethod
    async def from_local(
        cls,
        filename: Union[str, Path, None] = None,
        content: Optional[bytes] = None,
    ) -> "Image":
        """从本地文件路径加载图片，以 base64 的形式传递。

        Args:
            filename: 从本地文件路径加载图片，与 `content` 二选一。
            content: 从本地文件内容加载图片，与 `filename` 二选一。

        Returns:
            Image: 图片对象。
        """
        if content:
            pass
        elif filename:
            path = Path(filename)
            import aiofiles
            async with aiofiles.open(path, 'rb') as f:
                content = await f.read()
        else:
            raise ValueError("请指定图片路径或图片内容！")
        import base64
        img = cls(base64=base64.b64encode(content).decode())
        return img

    @classmethod
    def from_unsafe_path(cls, path: Union[str, Path]) -> "Image":
        """从不安全的路径加载图片。

        Args:
            path: 从不安全的路径加载图片。

        Returns:
            Image: 图片对象。
        """
        return cls.construct(path=str(path))

Ancestors

Subclasses

Class variables

var base64 : Optional[str]

图片的 Base64 编码。

var height : int

图片的高度

var image_id : Optional[str]

图片的 image_id，群图片与好友图片格式不同。不为空时将忽略 url 属性。

var path : Union[str, pathlib.Path, NoneType]

图片的路径，发送本地图片。

var url : Optional[pydantic.networks.HttpUrl]

图片的 URL，发送时可作网络图片的链接；接收时为腾讯图片服务器的链接，可用于图片下载。

var width : int

图片的宽度

Static methods

async def from_local(filename: Union[str, pathlib.Path, NoneType] = None, content: Optional[bytes] = None) ‑> Image

从本地文件路径加载图片，以 base64 的形式传递。

Args

filename
从本地文件路径加载图片，与 content 二选一。
content
从本地文件内容加载图片，与 filename 二选一。

Returns

Image
图片对象。
Expand source code 
@classmethod
async def from_local(
    cls,
    filename: Union[str, Path, None] = None,
    content: Optional[bytes] = None,
) -> "Image":
    """从本地文件路径加载图片，以 base64 的形式传递。

    Args:
        filename: 从本地文件路径加载图片，与 `content` 二选一。
        content: 从本地文件内容加载图片，与 `filename` 二选一。

    Returns:
        Image: 图片对象。
    """
    if content:
        pass
    elif filename:
        path = Path(filename)
        import aiofiles
        async with aiofiles.open(path, 'rb') as f:
            content = await f.read()
    else:
        raise ValueError("请指定图片路径或图片内容！")
    import base64
    img = cls(base64=base64.b64encode(content).decode())
    return img
def from_unsafe_path(path: Union[str, pathlib.Path]) ‑> Image

从不安全的路径加载图片。

Args

path
从不安全的路径加载图片。

Returns

Image
图片对象。
Expand source code 
@classmethod
def from_unsafe_path(cls, path: Union[str, Path]) -> "Image":
    """从不安全的路径加载图片。

    Args:
        path: 从不安全的路径加载图片。

    Returns:
        Image: 图片对象。
    """
    return cls.construct(path=str(path))
def validate_path(path: Union[str, pathlib.Path, NoneType])

修复 path 参数的行为，使之相对于 YiriMirai 的启动路径。

Expand source code 
@validator('path')
def validate_path(cls, path: Union[str, Path, None]):
    """修复 path 参数的行为，使之相对于 YiriMirai 的启动路径。"""
    if path:
        try:
            return str(Path(path).resolve(strict=True))
        except FileNotFoundError:
            raise ValueError(f"无效路径：{path}")
    else:
        return path

Instance variables

var uuid
Expand source code 
@property
def uuid(self):
    image_id = self.image_id
    if image_id[0] == '{':  # 群图片
        image_id = image_id[1:37]
    elif image_id[0] == '/':  # 好友图片
        image_id = image_id[1:]
    return image_id

Methods

def as_flash_image(self) ‑> FlashImage
Expand source code 
def as_flash_image(self) -> "FlashImage":
    return FlashImage(
        image_id=self.image_id,
        url=self.url,
        path=self.path,
        base64=self.base64
    )
def as_friend_image(self) ‑> str
Expand source code 
def as_friend_image(self) -> str:
    return f"/{self.uuid.lower()}"
def as_group_image(self) ‑> str
Expand source code 
def as_group_image(self) -> str:
    return f"{{{self.uuid.upper()}}}.jpg"
async def download(self, filename: Union[str, pathlib.Path, NoneType] = None, directory: Union[str, pathlib.Path, NoneType] = None, determine_type: bool = True)

下载图片到本地。

Args

filename
下载到本地的文件路径。与 directory 二选一。
directory
下载到本地的文件夹路径。与 filename 二选一。
determine_type
是否自动根据图片类型确定拓展名，默认为 True。
Expand source code 
async def download(
    self,
    filename: Union[str, Path, None] = None,
    directory: Union[str, Path, None] = None,
    determine_type: bool = True
):
    """下载图片到本地。

    Args:
        filename: 下载到本地的文件路径。与 `directory` 二选一。
        directory: 下载到本地的文件夹路径。与 `filename` 二选一。
        determine_type: 是否自动根据图片类型确定拓展名，默认为 True。
    """
    if not self.url:
        logger.warning(f'图片 `{self.uuid}` 无 url 参数，下载失败。')
        return

    import httpx
    async with httpx.AsyncClient() as client:
        response = await client.get(self.url)
        response.raise_for_status()
        content = response.content

        if filename:
            path = Path(filename)
            if determine_type:
                import imghdr
                path = path.with_suffix(
                    '.' + str(imghdr.what(None, content))
                )
            path.parent.mkdir(parents=True, exist_ok=True)
        elif directory:
            import imghdr
            path = Path(directory)
            path.mkdir(parents=True, exist_ok=True)
            path = path / f'{self.uuid}.{imghdr.what(None, content)}'
        else:
            raise ValueError("请指定文件路径或文件夹路径！")

        import aiofiles
        async with aiofiles.open(path, 'wb') as f:
            await f.write(content)

        return path

Inherited members

class Json (*args, **kwargs)

JSON。

Expand source code 
class Json(MessageComponent):
    """JSON。"""
    type: str = "Json"
    """消息组件类型。"""
    json_: str
    """JSON 文本。"""

Ancestors

Class variables

var json_ : str

JSON 文本。

Inherited members

class MarketFace (*args, **kwargs)

商店表情（目前只支持接收）。

Expand source code 
class MarketFace(MessageComponent):
    """商店表情（目前只支持接收）。"""
    type: str = "MarketFace"
    """消息组件类型。"""
    id: int
    """商店表情编号。"""
    name: str
    """商店表情名称。"""

Ancestors

Class variables

var id : int

商店表情编号。

var name : str

商店表情名称。

Inherited members

class MemberCardChangeEvent (*args, **kwargs)

群名片改动。

Args

type
事件名。
origin
原本名片。
current
现在名片。
member
改动的群成员。
group
事件发生的群。
Expand source code 
class MemberCardChangeEvent(GroupEvent):
    """群名片改动。

    Args:
        type: 事件名。
        origin: 原本名片。
        current: 现在名片。
        member: 改动的群成员。
        group: 事件发生的群。
    """
    type: str = 'MemberCardChangeEvent'
    """事件名。"""
    origin: str
    """原本名片。"""
    current: str
    """现在名片。"""
    member: GroupMember
    """改动的群成员。"""

Ancestors

Class variables

var current : str

现在名片。

var member : GroupMember

改动的群成员。

var origin : str

原本名片。

Inherited members

class MemberHonorChangeEvent (*args, **kwargs)

群员称号改变。

Args

type
事件名。
member
称号改变的群成员。
action
称号变化行为：achieve 获得称号，lose 失去称号。
group
事件发生的群。
honor
称号名称。
Expand source code 
class MemberHonorChangeEvent(GroupEvent):
    """群员称号改变。

    Args:
        type: 事件名。
        member: 称号改变的群成员。
        action: 称号变化行为：achieve 获得称号，lose 失去称号。
        group: 事件发生的群。
        honor: 称号名称。
    """
    type: str = 'MemberHonorChangeEvent'
    """事件名。"""
    member: GroupMember
    """称号改变的群成员。"""
    action: Literal['achieve', 'lose']
    """称号变化行为：achieve 获得称号，lose 失去称号。"""
    honor: str
    """称号名称。"""

Ancestors

Class variables

var action : Literal['achieve', 'lose']

称号变化行为：achieve 获得称号，lose 失去称号。

var honor : str

称号名称。

var member : GroupMember

称号改变的群成员。

Inherited members

class MemberJoinEvent (*args, **kwargs)

新人入群。

Args

type
事件名。
member
加入的群成员。
group
加入的群。
Expand source code 
class MemberJoinEvent(GroupEvent):
    """新人入群。

    Args:
        type: 事件名。
        member: 加入的群成员。
        group: 加入的群。
    """
    type: str = 'MemberJoinEvent'
    """事件名。"""
    member: GroupMember
    """加入的群成员。"""
    invitor: Optional[GroupMember]
    """邀请者。"""

Ancestors

Class variables

var invitor : Optional[GroupMember]

邀请者。

var member : GroupMember

加入的群成员。

Inherited members

class MemberJoinRequestEvent (*args, **kwargs)

用户入群申请（Bot需要有管理员权限）。

Args

type
事件名。
event_id
事件标识，响应该事件时的标识。
from_id
申请人 QQ 号。
group_id
申请人申请入群的群号。
group_name
申请人申请入群的群名称。
nick
申请人的昵称或群名片。
message
申请消息。
Expand source code 
class MemberJoinRequestEvent(RequestEvent):
    """用户入群申请（Bot需要有管理员权限）。

    Args:
        type: 事件名。
        event_id: 事件标识，响应该事件时的标识。
        from_id: 申请人 QQ 号。
        group_id: 申请人申请入群的群号。
        group_name: 申请人申请入群的群名称。
        nick: 申请人的昵称或群名片。
        message: 申请消息。
    """
    type: str = 'MemberJoinRequestEvent'
    """事件名。"""
    event_id: int
    """事件标识，响应该事件时的标识。"""
    from_id: int
    """申请人 QQ 号。"""
    group_id: int
    """申请人申请入群的群号。"""
    group_name: str
    """申请人申请入群的群名称。"""
    nick: str
    """申请人的昵称或群名片。"""
    message: str
    """申请消息。"""

Ancestors

Class variables

var group_name : str

申请人申请入群的群名称。

var message : str

申请消息。

var nick : str

申请人的昵称或群名片。

Inherited members

class MemberLeaveEventKick (*args, **kwargs)

成员被踢出群（该成员不是Bot）。

Args

type
事件名。
member
被踢出的群成员。
group
事件发生的群。
operator
被踢出的群的管理员。
Expand source code 
class MemberLeaveEventKick(GroupEvent):
    """成员被踢出群（该成员不是Bot）。

    Args:
        type: 事件名。
        member: 被踢出的群成员。
        group: 事件发生的群。
        operator: 被踢出的群的管理员。
    """
    type: str = 'MemberLeaveEventKick'
    """事件名。"""
    member: GroupMember
    """被踢出的群成员。"""
    operator: Optional[GroupMember]
    """被踢出的群的管理员。"""

Ancestors

Class variables

var member : GroupMember

被踢出的群成员。

var operator : Optional[GroupMember]

被踢出的群的管理员。

Inherited members

class MemberLeaveEventQuit (*args, **kwargs)

成员主动离群（该成员不是Bot）。

Args

type
事件名。
member
离群的群成员。
group
事件发生的群。
Expand source code 
class MemberLeaveEventQuit(GroupEvent):
    """成员主动离群（该成员不是Bot）。

    Args:
        type: 事件名。
        member: 离群的群成员。
        group: 事件发生的群。
    """
    type: str = 'MemberLeaveEventQuit'
    """事件名。"""
    member: GroupMember
    """离群的群成员。"""

Ancestors

Class variables

var member : GroupMember

离群的群成员。

Inherited members

class MemberMuteEvent (*args, **kwargs)

群成员被禁言（该成员不是Bot）。

Args

type
事件名。
duration_seconds
禁言时间，单位为秒。
Expand source code 
class MemberMuteEvent(GroupEvent):
    """群成员被禁言（该成员不是Bot）。

    Args:
        type: 事件名。
        duration_seconds: 禁言时间，单位为秒。
    """
    type: str = 'MemberMuteEvent'
    """事件名。"""
    duration_seconds: int
    """禁言时间，单位为秒。"""

Ancestors

Class variables

var duration_seconds : int

禁言时间，单位为秒。

Inherited members

class MemberPermissionChangeEvent (*args, **kwargs)

成员权限改变（该成员不是Bot）。

Args

type
事件名。
origin
原本权限。
current
现在权限。
member
权限改变的群成员。
group
事件发生的群。
Expand source code 
class MemberPermissionChangeEvent(GroupEvent):
    """成员权限改变（该成员不是Bot）。

    Args:
        type: 事件名。
        origin: 原本权限。
        current: 现在权限。
        member: 权限改变的群成员。
        group: 事件发生的群。
    """
    type: str = 'MemberPermissionChangeEvent'
    """事件名。"""
    origin: Permission
    """原本权限。"""
    current: Permission
    """现在权限。"""
    member: GroupMember
    """权限改变的群成员。"""

Ancestors

Class variables

var current : Permission

现在权限。

var member : GroupMember

权限改变的群成员。

var origin : Permission

原本权限。

Inherited members

class MemberSpecialTitleChangeEvent (*args, **kwargs)

群头衔改动（只有群主有操作权限）。

Args

type
事件名。
origin
原本头衔。
current
现在头衔。
member
头衔改动的群成员。
group
事件发生的群。
Expand source code 
class MemberSpecialTitleChangeEvent(GroupEvent):
    """群头衔改动（只有群主有操作权限）。

    Args:
        type: 事件名。
        origin: 原本头衔。
        current: 现在头衔。
        member: 头衔改动的群成员。
        group: 事件发生的群。
    """
    type: str = 'MemberSpecialTitleChangeEvent'
    """事件名。"""
    origin: str
    """原本头衔。"""
    current: str
    """现在头衔。"""
    member: GroupMember
    """头衔改动的群成员。"""

Ancestors

Class variables

var current : str

现在头衔。

var member : GroupMember

头衔改动的群成员。

var origin : str

原本头衔。

Inherited members

class MemberUnmuteEvent (*args, **kwargs)

群成员被取消禁言（该成员不是Bot）。

Args

type
事件名。
member
被取消禁言的群成员。
group
事件发生的群。
Expand source code 
class MemberUnmuteEvent(GroupEvent):
    """群成员被取消禁言（该成员不是Bot）。

    Args:
        type: 事件名。
        member: 被取消禁言的群成员。
        group: 事件发生的群。
    """
    type: str = 'MemberUnmuteEvent'
    """事件名。"""
    member: GroupMember
    """被取消禁言的群成员。"""
    operator: Optional[GroupMember]
    """被取消禁言的群管理员。"""

Ancestors

Class variables

var member : GroupMember

被取消禁言的群成员。

var operator : Optional[GroupMember]

被取消禁言的群管理员。

Inherited members

class MessageChain

消息链。

一个构造消息链的例子：

message_chain = MessageChain([
    AtAll(),
    Plain("Hello World!"),
])

Plain 可以省略。

message_chain = MessageChain([
    AtAll(),
    "Hello World!",
])

在调用 API 时，参数中需要 MessageChain 的，也可以使用 List[MessageComponent] 代替。 例如，以下两种写法是等价的：

await bot.send_friend_message(12345678, [
    Plain("Hello World!")
])

await bot.send_friend_message(12345678, MessageChain([
    Plain("Hello World!")
]))

使用str(message_chain)获取消息链的字符串表示，字符串采用 mirai 码格式， 并自动按照 mirai 码的规定转义。 参看 mirai 的文档。 获取未转义的消息链字符串，可以使用deserialize()(str(message_chain))。

可以使用 for 循环遍历消息链中的消息组件。

for component in message_chain:
    print(repr(component))

可以使用 == 运算符比较两个消息链是否相同。

another_msg_chain = MessageChain([
    {
        "type": "AtAll"
    }, {
        "type": "Plain",
        "text": "Hello World!"
    },
])
print(message_chain == another_msg_chain)
'True'

可以使用 in 运算检查消息链中： 1. 是否有某个消息组件。 2. 是否有某个类型的消息组件。 3. 是否有某个子消息链。 4. 对应的 mirai 码中是否有某子字符串。

if AtAll in message_chain:
    print('AtAll')

if At(bot.qq) in message_chain:
    print('At Me')

if MessageChain([At(bot.qq), Plain('Hello!')]) in message_chain:
    print('Hello!')

if 'Hello' in message_chain:
    print('Hi!')

消息链的 has 方法和 in 等价。

if message_chain.has(AtAll):
    print('AtAll')

也可以使用 >= 和 <=运算符：

if MessageChain([At(bot.qq), Plain('Hello!')]) <= message_chain:
    print('Hello!')

需注意，此处的子消息链匹配会把 Plain 看成一个整体，而不是匹配其文本的一部分。 如需对文本进行部分匹配，请采用 mirai 码字符串匹配的方式。

消息链对索引操作进行了增强。以消息组件类型为索引，获取消息链中的全部该类型的消息组件。

plain_list = message_chain[Plain]
'[Plain("Hello World!")]'

以 类型, 数量 为索引，获取前至多多少个该类型的消息组件。

plain_list_first = message_chain[Plain, 1]
'[Plain("Hello World!")]'

消息链的 get 方法和索引操作等价。

plain_list_first = message_chain.get(Plain)
'[Plain("Hello World!")]'

消息链的 get 方法还可指定第二个参数 count，这相当于以 类型, 数量 为索引。

plain_list_first = message_chain.get(Plain, 1)
# 这等价于
plain_list_first = message_chain[Plain, 1]

可以用加号连接两个消息链。

MessageChain(['Hello World!']) + MessageChain(['Goodbye World!'])
# 返回 MessageChain([Plain("Hello World!"), Plain("Goodbye World!")])

可以用 * 运算符复制消息链。

MessageChain(['Hello World!']) * 2
# 返回 MessageChain([Plain("Hello World!"), Plain("Hello World!")])

除此之外，消息链还支持很多 list 拥有的操作，比如 index 和 count。

message_chain = MessageChain([
    AtAll(),
    "Hello World!",
])
message_chain.index(Plain)
# 返回 0
message_chain.count(Plain)
# 返回 1

消息链对这些操作进行了拓展。在传入元素的地方，一般都可以传入元素的类型。

Expand source code 
class MessageChain(MiraiBaseModel):
    """消息链。

    一个构造消息链的例子：
    ```py
    message_chain = MessageChain([
        AtAll(),
        Plain("Hello World!"),
    ])
    ```

    `Plain` 可以省略。
    ```py
    message_chain = MessageChain([
        AtAll(),
        "Hello World!",
    ])
    ```

    在调用 API 时，参数中需要 MessageChain 的，也可以使用 `List[MessageComponent]` 代替。
    例如，以下两种写法是等价的：
    ```py
    await bot.send_friend_message(12345678, [
        Plain("Hello World!")
    ])
    ```
    ```py
    await bot.send_friend_message(12345678, MessageChain([
        Plain("Hello World!")
    ]))
    ```

    使用`str(message_chain)`获取消息链的字符串表示，字符串采用 mirai 码格式，
    并自动按照 mirai 码的规定转义。
    参看 mirai 的[文档](https://github.com/mamoe/mirai/blob/dev/docs/Messages.md#mirai-%E7%A0%81)。
    获取未转义的消息链字符串，可以使用`deserialize(str(message_chain))`。

    可以使用 for 循环遍历消息链中的消息组件。
    ```py
    for component in message_chain:
        print(repr(component))
    ```

    可以使用 `==` 运算符比较两个消息链是否相同。
    ```py
    another_msg_chain = MessageChain([
        {
            "type": "AtAll"
        }, {
            "type": "Plain",
            "text": "Hello World!"
        },
    ])
    print(message_chain == another_msg_chain)
    'True'
    ```

    可以使用 `in` 运算检查消息链中：
    1. 是否有某个消息组件。
    2. 是否有某个类型的消息组件。
    3. 是否有某个子消息链。
    4. 对应的 mirai 码中是否有某子字符串。

    ```py
    if AtAll in message_chain:
        print('AtAll')

    if At(bot.qq) in message_chain:
        print('At Me')

    if MessageChain([At(bot.qq), Plain('Hello!')]) in message_chain:
        print('Hello!')

    if 'Hello' in message_chain:
        print('Hi!')
    ```

    消息链的 `has` 方法和 `in` 等价。
    ```py
    if message_chain.has(AtAll):
        print('AtAll')
    ```

    也可以使用 `>=` 和 `<= `运算符：
    ```py
    if MessageChain([At(bot.qq), Plain('Hello!')]) <= message_chain:
        print('Hello!')
    ```

    需注意，此处的子消息链匹配会把 Plain 看成一个整体，而不是匹配其文本的一部分。
    如需对文本进行部分匹配，请采用 mirai 码字符串匹配的方式。

    消息链对索引操作进行了增强。以消息组件类型为索引，获取消息链中的全部该类型的消息组件。
    ```py
    plain_list = message_chain[Plain]
    '[Plain("Hello World!")]'
    ```

    以 `类型, 数量` 为索引，获取前至多多少个该类型的消息组件。
    ```py
    plain_list_first = message_chain[Plain, 1]
    '[Plain("Hello World!")]'
    ```

    消息链的 `get` 方法和索引操作等价。
    ```py
    plain_list_first = message_chain.get(Plain)
    '[Plain("Hello World!")]'
    ```

    消息链的 `get` 方法还可指定第二个参数 `count`，这相当于以 `类型, 数量` 为索引。
    ```py
    plain_list_first = message_chain.get(Plain, 1)
    # 这等价于
    plain_list_first = message_chain[Plain, 1]
    ```

    可以用加号连接两个消息链。
    ```py
    MessageChain(['Hello World!']) + MessageChain(['Goodbye World!'])
    # 返回 MessageChain([Plain("Hello World!"), Plain("Goodbye World!")])
    ```

    可以用 `*` 运算符复制消息链。
    ```py
    MessageChain(['Hello World!']) * 2
    # 返回 MessageChain([Plain("Hello World!"), Plain("Hello World!")])
    ```

    除此之外，消息链还支持很多 list 拥有的操作，比如 `index` 和 `count`。
    ```py
    message_chain = MessageChain([
        AtAll(),
        "Hello World!",
    ])
    message_chain.index(Plain)
    # 返回 0
    message_chain.count(Plain)
    # 返回 1
    ```

    消息链对这些操作进行了拓展。在传入元素的地方，一般都可以传入元素的类型。
    """
    __root__: List[MessageComponent]

    @staticmethod
    def _parse_message_chain(msg_chain: Iterable):
        result = []
        for msg in msg_chain:
            if isinstance(msg, dict):
                result.append(MessageComponent.parse_subtype(msg))
            elif isinstance(msg, MessageComponent):
                result.append(msg)
            elif isinstance(msg, str):
                result.append(Plain(msg))
            else:
                raise TypeError(
                    f"消息链中元素需为 dict 或 str 或 MessageComponent，当前类型：{type(msg)}"
                )
        return result

    @validator('__root__', always=True, pre=True)
    def _parse_component(cls, msg_chain):
        if isinstance(msg_chain, (str, MessageComponent)):
            msg_chain = [msg_chain]
        if not msg_chain:
            msg_chain = []
        return cls._parse_message_chain(msg_chain)

    @classmethod
    def parse_obj(cls, msg_chain: Iterable):
        """通过列表形式的消息链，构造对应的 `MessageChain` 对象。

        Args:
            msg_chain: 列表形式的消息链。
        """
        result = cls._parse_message_chain(msg_chain)
        return cls(__root__=result)

    def __init__(self, __root__: Iterable[MessageComponent] = None):
        super().__init__(__root__=__root__)

    def __str__(self):
        return "".join(str(component) for component in self.__root__)

    def as_mirai_code(self) -> str:
        """将消息链转换为 mirai 码字符串。

        该方法会自动转换消息链中的元素。

        Returns:
            mirai 码字符串。
        """
        return "".join(
            component.as_mirai_code() for component in self.__root__
        )

    def __repr__(self):
        return f'{self.__class__.__name__}({self.__root__!r})'

    def __iter__(self):
        yield from self.__root__

    @overload
    def get(self, index: int) -> MessageComponent:
        ...

    @overload
    def get(self, index: slice) -> List[MessageComponent]:
        ...

    @overload
    def get(self, index: Type[TMessageComponent]) -> List[TMessageComponent]:
        ...

    @overload
    def get(
        self, index: Tuple[Type[TMessageComponent], int]
    ) -> List[TMessageComponent]:
        ...

    def get(
        self,
        index: Union[int, slice, Type[TMessageComponent],
                     Tuple[Type[TMessageComponent], int]],
        count: Optional[int] = None
    ) -> Union[MessageComponent, List[MessageComponent],
               List[TMessageComponent]]:
        """获取消息链中的某个（某些）消息组件，或某类型的消息组件。

        Args:
            index (`Union[int, slice, Type[TMessageComponent], Tuple[Type[TMessageComponent], int]]`):
                如果为 `int`，则返回该索引处的消息组件。
                如果为 `slice`，则返回该索引范围处的消息组件。
                如果为 `Type[TMessageComponent]`，则返回该类型的全部消息组件。
                如果为 `Tuple[Type[TMessageComponent], int]`，则返回该类型的至多 `index[1]` 个消息组件。

            count: 如果为 `int`，则返回至多 `count` 个消息组件。

        Returns:
            MessageComponent: 返回指定索引处的消息组件。
            List[MessageComponent]: 返回指定索引范围的消息组件。
            List[TMessageComponent]: 返回指定类型的消息组件构成的列表。
        """
        # 正常索引
        if isinstance(index, int):
            return self.__root__[index]
        # 切片索引
        if isinstance(index, slice):
            return self.__root__[index]
        # 指定 count
        if count:
            if isinstance(index, type):
                index = (index, count)
            elif isinstance(index, tuple):
                index = (index[0], count if count < index[1] else index[1])
        # 索引对象为 MessageComponent 类，返回所有对应 component
        if isinstance(index, type):
            return [
                component for component in self if type(component) is index
            ]
        # 索引对象为 MessageComponent 和 int 构成的 tuple， 返回指定数量的 component
        if isinstance(index, tuple):
            components = (
                component for component in self if type(component) is index[0]
            )
            return [
                component for component, _ in zip(components, range(index[1]))
            ]
        raise TypeError(f"消息链索引需为 int 或 MessageComponent，当前类型：{type(index)}")

    def get_first(self,
                  t: Type[TMessageComponent]) -> Optional[TMessageComponent]:
        """获取消息链中第一个符合类型的消息组件。"""
        for component in self:
            if isinstance(component, t):
                return component
        return None

    @overload
    def __getitem__(self, index: int) -> MessageComponent:
        ...

    @overload
    def __getitem__(self, index: slice) -> List[MessageComponent]:
        ...

    @overload
    def __getitem__(self,
                    index: Type[TMessageComponent]) -> List[TMessageComponent]:
        ...

    @overload
    def __getitem__(
        self, index: Tuple[Type[TMessageComponent], int]
    ) -> List[TMessageComponent]:
        ...

    def __getitem__(
        self, index: Union[int, slice, Type[TMessageComponent],
                           Tuple[Type[TMessageComponent], int]]
    ) -> Union[MessageComponent, List[MessageComponent],
               List[TMessageComponent]]:
        return self.get(index)

    def __setitem__(
        self, key: Union[int, slice],
        value: Union[MessageComponent, str, Iterable[Union[MessageComponent,
                                                           str]]]
    ):
        if isinstance(value, str):
            value = Plain(value)
        if isinstance(value, Iterable):
            value = (Plain(c) if isinstance(c, str) else c for c in value)
        self.__root__[key] = value  # type: ignore

    def __delitem__(self, key: Union[int, slice]):
        del self.__root__[key]

    def __reversed__(self) -> Iterable[MessageComponent]:
        return reversed(self.__root__)

    def has(
        self, sub: Union[MessageComponent, Type[MessageComponent],
                         'MessageChain', str]
    ) -> bool:
        """判断消息链中：
        1. 是否有某个消息组件。
        2. 是否有某个类型的消息组件。
        3. 是否有某个子消息链。
        4. 对应的 mirai 码中是否有某子字符串。

        Args:
            sub (`Union[MessageComponent, Type[MessageComponent], 'MessageChain', str]`):
                若为 `MessageComponent`，则判断该组件是否在消息链中。
                若为 `Type[MessageComponent]`，则判断该组件类型是否在消息链中。
                若为 `MessageChain`，则判断该子消息链是否在消息链中。
                若为 `str`，则判断对应的 mirai 码中是否有某子字符串。

        Returns:
            bool: 是否找到。
        """
        if isinstance(sub, type):  # 检测消息链中是否有某种类型的对象
            for i in self:
                if type(i) is sub:
                    return True
            return False
        if isinstance(sub, MessageComponent):  # 检查消息链中是否有某个组件
            for i in self:
                if i == sub:
                    return True
            return False
        if isinstance(sub, MessageChain):  # 检查消息链中是否有某个子消息链
            return bool(kmp(self, sub))
        if isinstance(sub, str):  # 检查消息中有无指定字符串子串
            return sub in deserialize(str(self))
        raise TypeError(f"类型不匹配，当前类型：{type(sub)}")

    def __contains__(self, sub) -> bool:
        return self.has(sub)

    def __ge__(self, other):
        return other in self

    def __len__(self) -> int:
        return len(self.__root__)

    def __add__(
        self, other: Union['MessageChain', MessageComponent, str]
    ) -> 'MessageChain':
        if isinstance(other, MessageChain):
            return self.__class__(self.__root__ + other.__root__)
        if isinstance(other, str):
            return self.__class__(self.__root__ + [Plain(other)])
        if isinstance(other, MessageComponent):
            return self.__class__(self.__root__ + [other])
        return NotImplemented

    def __radd__(self, other: Union[MessageComponent, str]) -> 'MessageChain':
        if isinstance(other, MessageComponent):
            return self.__class__([other] + self.__root__)
        if isinstance(other, str):
            return self.__class__(
                [cast(MessageComponent, Plain(other))] + self.__root__
            )
        return NotImplemented

    def __mul__(self, other: int):
        if isinstance(other, int):
            return self.__class__(self.__root__ * other)
        return NotImplemented

    def __rmul__(self, other: int):
        return self.__mul__(other)

    def __iadd__(self, other: Iterable[Union[MessageComponent, str]]):
        self.extend(other)

    def __imul__(self, other: int):
        if isinstance(other, int):
            self.__root__ *= other
        return NotImplemented

    def index(
        self,
        x: Union[MessageComponent, Type[MessageComponent]],
        i: int = 0,
        j: int = -1
    ) -> int:
        """返回 x 在消息链中首次出现项的索引号（索引号在 i 或其后且在 j 之前）。

        Args:
            x (`Union[MessageComponent, Type[MessageComponent]]`):
                要查找的消息元素或消息元素类型。
            i: 从哪个位置开始查找。
            j: 查找到哪个位置结束。

        Returns:
            int: 如果找到，则返回索引号。

        Raises:
            ValueError: 没有找到。
            TypeError: 类型不匹配。
        """
        if isinstance(x, type):
            l = len(self)
            if i < 0:
                i += l
            if i < 0:
                i = 0
            if j < 0:
                j += l
            if j > l:
                j = l
            for index in range(i, j):
                if type(self[index]) is x:
                    return index
            raise ValueError("消息链中不存在该类型的组件。")
        if isinstance(x, MessageComponent):
            return self.__root__.index(x, i, j)
        raise TypeError(f"类型不匹配，当前类型：{type(x)}")

    def count(self, x: Union[MessageComponent, Type[MessageComponent]]) -> int:
        """返回消息链中 x 出现的次数。

        Args:
            x (`Union[MessageComponent, Type[MessageComponent]]`):
                要查找的消息元素或消息元素类型。

        Returns:
            int: 次数。
        """
        if isinstance(x, type):
            return sum(1 for i in self if type(i) is x)
        if isinstance(x, MessageComponent):
            return self.__root__.count(x)
        raise TypeError(f"类型不匹配，当前类型：{type(x)}")

    def extend(self, x: Iterable[Union[MessageComponent, str]]):
        """将另一个消息链中的元素添加到消息链末尾。

        Args:
            x: 另一个消息链，也可为消息元素或字符串元素的序列。
        """
        self.__root__.extend(Plain(c) if isinstance(c, str) else c for c in x)

    def append(self, x: Union[MessageComponent, str]):
        """将一个消息元素或字符串元素添加到消息链末尾。

        Args:
            x: 消息元素或字符串元素。
        """
        self.__root__.append(Plain(x) if isinstance(x, str) else x)

    def insert(self, i: int, x: Union[MessageComponent, str]):
        """将一个消息元素或字符串添加到消息链中指定位置。

        Args:
            i: 插入位置。
            x: 消息元素或字符串元素。
        """
        self.__root__.insert(i, Plain(x) if isinstance(x, str) else x)

    def pop(self, i: int = -1) -> MessageComponent:
        """从消息链中移除并返回指定位置的元素。

        Args:
            i: 移除位置。默认为末尾。

        Returns:
            MessageComponent: 移除的元素。
        """
        return self.__root__.pop(i)

    def remove(self, x: Union[MessageComponent, Type[MessageComponent]]):
        """从消息链中移除指定元素或指定类型的一个元素。

        Args:
            x: 指定的元素或元素类型。
        """
        if isinstance(x, type):
            self.pop(self.index(x))
        if isinstance(x, MessageComponent):
            self.__root__.remove(x)

    def exclude(
        self,
        x: Union[MessageComponent, Type[MessageComponent]],
        count: int = -1
    ) -> 'MessageChain':
        """返回移除指定元素或指定类型的元素后剩余的消息链。

        Args:
            x: 指定的元素或元素类型。
            count: 至多移除的数量。默认为全部移除。

        Returns:
            MessageChain: 剩余的消息链。
        """
        def _exclude():
            nonlocal count
            x_is_type = isinstance(x, type)
            for c in self:
                if count > 0 and ((x_is_type and type(c) is x) or c == x):
                    count -= 1
                    continue
                yield c

        return self.__class__(_exclude())

    def reverse(self):
        """将消息链原地翻转。"""
        self.__root__.reverse()

    @classmethod
    def join(cls, *args: Iterable[Union[str, MessageComponent]]):
        return cls(
            Plain(c) if isinstance(c, str) else c
            for c in itertools.chain(*args)
        )

    @property
    def source(self) -> Optional['Source']:
        """获取消息链中的 `Source` 对象。"""
        return self.get_first(Source)

    @property
    def message_id(self) -> int:
        """获取消息链的 message_id，若无法获取，返回 -1。"""
        source = self.source
        return source.id if source else -1

Ancestors

  • MiraiBaseModel
  • pydantic.main.BaseModel
  • pydantic.utils.Representation

Static methods

def join(*args: Iterable[Union[MessageComponent, str]])
Expand source code 
@classmethod
def join(cls, *args: Iterable[Union[str, MessageComponent]]):
    return cls(
        Plain(c) if isinstance(c, str) else c
        for c in itertools.chain(*args)
    )
def parse_obj(msg_chain: Iterable)

通过列表形式的消息链，构造对应的 MessageChain 对象。

Args

msg_chain
列表形式的消息链。
Expand source code 
@classmethod
def parse_obj(cls, msg_chain: Iterable):
    """通过列表形式的消息链，构造对应的 `MessageChain` 对象。

    Args:
        msg_chain: 列表形式的消息链。
    """
    result = cls._parse_message_chain(msg_chain)
    return cls(__root__=result)

Instance variables

var message_id : int

获取消息链的 message_id，若无法获取，返回 -1。

Expand source code 
@property
def message_id(self) -> int:
    """获取消息链的 message_id，若无法获取，返回 -1。"""
    source = self.source
    return source.id if source else -1
var source : Optional[Source]

获取消息链中的 Source 对象。

Expand source code 
@property
def source(self) -> Optional['Source']:
    """获取消息链中的 `Source` 对象。"""
    return self.get_first(Source)

Methods

def append(self, x: Union[MessageComponent, str])

将一个消息元素或字符串元素添加到消息链末尾。

Args

x
消息元素或字符串元素。
Expand source code 
def append(self, x: Union[MessageComponent, str]):
    """将一个消息元素或字符串元素添加到消息链末尾。

    Args:
        x: 消息元素或字符串元素。
    """
    self.__root__.append(Plain(x) if isinstance(x, str) else x)
def as_mirai_code(self) ‑> str

将消息链转换为 mirai 码字符串。

该方法会自动转换消息链中的元素。

Returns

mirai 码字符串。

Expand source code 
def as_mirai_code(self) -> str:
    """将消息链转换为 mirai 码字符串。

    该方法会自动转换消息链中的元素。

    Returns:
        mirai 码字符串。
    """
    return "".join(
        component.as_mirai_code() for component in self.__root__
    )
def count(self, x: Union[MessageComponent, Type[MessageComponent]]) ‑> int

返回消息链中 x 出现的次数。

Args

x (Union[MessageComponent, Type[MessageComponent]]): 要查找的消息元素或消息元素类型。

Returns

int
次数。
Expand source code 
def count(self, x: Union[MessageComponent, Type[MessageComponent]]) -> int:
    """返回消息链中 x 出现的次数。

    Args:
        x (`Union[MessageComponent, Type[MessageComponent]]`):
            要查找的消息元素或消息元素类型。

    Returns:
        int: 次数。
    """
    if isinstance(x, type):
        return sum(1 for i in self if type(i) is x)
    if isinstance(x, MessageComponent):
        return self.__root__.count(x)
    raise TypeError(f"类型不匹配，当前类型：{type(x)}")
def exclude(self, x: Union[MessageComponent, Type[MessageComponent]], count: int = -1) ‑> MessageChain

返回移除指定元素或指定类型的元素后剩余的消息链。

Args

x
指定的元素或元素类型。
count
至多移除的数量。默认为全部移除。

Returns

MessageChain
剩余的消息链。
Expand source code 
def exclude(
    self,
    x: Union[MessageComponent, Type[MessageComponent]],
    count: int = -1
) -> 'MessageChain':
    """返回移除指定元素或指定类型的元素后剩余的消息链。

    Args:
        x: 指定的元素或元素类型。
        count: 至多移除的数量。默认为全部移除。

    Returns:
        MessageChain: 剩余的消息链。
    """
    def _exclude():
        nonlocal count
        x_is_type = isinstance(x, type)
        for c in self:
            if count > 0 and ((x_is_type and type(c) is x) or c == x):
                count -= 1
                continue
            yield c

    return self.__class__(_exclude())
def extend(self, x: Iterable[Union[MessageComponent, str]])

将另一个消息链中的元素添加到消息链末尾。

Args

x
另一个消息链，也可为消息元素或字符串元素的序列。
Expand source code 
def extend(self, x: Iterable[Union[MessageComponent, str]]):
    """将另一个消息链中的元素添加到消息链末尾。

    Args:
        x: 另一个消息链，也可为消息元素或字符串元素的序列。
    """
    self.__root__.extend(Plain(c) if isinstance(c, str) else c for c in x)
def get(self, index: Union[int, slice, Type[~TMessageComponent], Tuple[Type[~TMessageComponent], int]], count: Optional[int] = None) ‑> Union[MessageComponent, List[MessageComponent], List[~TMessageComponent]]

获取消息链中的某个（某些）消息组件，或某类型的消息组件。

Args

index (Union[int, slice, Type[TMessageComponent], Tuple[Type[TMessageComponent], int]]): 如果为 int，则返回该索引处的消息组件。 如果为 slice，则返回该索引范围处的消息组件。 如果为 Type[TMessageComponent]，则返回该类型的全部消息组件。 如果为 Tuple[Type[TMessageComponent], int]，则返回该类型的至多 index[1] 个消息组件。

count
如果为 int，则返回至多 count 个消息组件。

Returns

MessageComponent
返回指定索引处的消息组件。
List[MessageComponent]
返回指定索引范围的消息组件。
List[TMessageComponent]
返回指定类型的消息组件构成的列表。
Expand source code 
def get(
    self,
    index: Union[int, slice, Type[TMessageComponent],
                 Tuple[Type[TMessageComponent], int]],
    count: Optional[int] = None
) -> Union[MessageComponent, List[MessageComponent],
           List[TMessageComponent]]:
    """获取消息链中的某个（某些）消息组件，或某类型的消息组件。

    Args:
        index (`Union[int, slice, Type[TMessageComponent], Tuple[Type[TMessageComponent], int]]`):
            如果为 `int`，则返回该索引处的消息组件。
            如果为 `slice`，则返回该索引范围处的消息组件。
            如果为 `Type[TMessageComponent]`，则返回该类型的全部消息组件。
            如果为 `Tuple[Type[TMessageComponent], int]`，则返回该类型的至多 `index[1]` 个消息组件。

        count: 如果为 `int`，则返回至多 `count` 个消息组件。

    Returns:
        MessageComponent: 返回指定索引处的消息组件。
        List[MessageComponent]: 返回指定索引范围的消息组件。
        List[TMessageComponent]: 返回指定类型的消息组件构成的列表。
    """
    # 正常索引
    if isinstance(index, int):
        return self.__root__[index]
    # 切片索引
    if isinstance(index, slice):
        return self.__root__[index]
    # 指定 count
    if count:
        if isinstance(index, type):
            index = (index, count)
        elif isinstance(index, tuple):
            index = (index[0], count if count < index[1] else index[1])
    # 索引对象为 MessageComponent 类，返回所有对应 component
    if isinstance(index, type):
        return [
            component for component in self if type(component) is index
        ]
    # 索引对象为 MessageComponent 和 int 构成的 tuple， 返回指定数量的 component
    if isinstance(index, tuple):
        components = (
            component for component in self if type(component) is index[0]
        )
        return [
            component for component, _ in zip(components, range(index[1]))
        ]
    raise TypeError(f"消息链索引需为 int 或 MessageComponent，当前类型：{type(index)}")
def get_first(self, t: Type[~TMessageComponent]) ‑> Optional[~TMessageComponent]

获取消息链中第一个符合类型的消息组件。

Expand source code 
def get_first(self,
              t: Type[TMessageComponent]) -> Optional[TMessageComponent]:
    """获取消息链中第一个符合类型的消息组件。"""
    for component in self:
        if isinstance(component, t):
            return component
    return None
def has(self, sub: Union[MessageComponent, Type[MessageComponent], ForwardRef('MessageChain'), str]) ‑> bool

判断消息链中： 1. 是否有某个消息组件。 2. 是否有某个类型的消息组件。 3. 是否有某个子消息链。 4. 对应的 mirai 码中是否有某子字符串。

Args

sub (Union[MessageComponent, Type[MessageComponent], 'MessageChain', str]): 若为 MessageComponent，则判断该组件是否在消息链中。 若为 Type[MessageComponent]，则判断该组件类型是否在消息链中。 若为 MessageChain，则判断该子消息链是否在消息链中。 若为 str，则判断对应的 mirai 码中是否有某子字符串。

Returns

bool
是否找到。
Expand source code 
def has(
    self, sub: Union[MessageComponent, Type[MessageComponent],
                     'MessageChain', str]
) -> bool:
    """判断消息链中：
    1. 是否有某个消息组件。
    2. 是否有某个类型的消息组件。
    3. 是否有某个子消息链。
    4. 对应的 mirai 码中是否有某子字符串。

    Args:
        sub (`Union[MessageComponent, Type[MessageComponent], 'MessageChain', str]`):
            若为 `MessageComponent`，则判断该组件是否在消息链中。
            若为 `Type[MessageComponent]`，则判断该组件类型是否在消息链中。
            若为 `MessageChain`，则判断该子消息链是否在消息链中。
            若为 `str`，则判断对应的 mirai 码中是否有某子字符串。

    Returns:
        bool: 是否找到。
    """
    if isinstance(sub, type):  # 检测消息链中是否有某种类型的对象
        for i in self:
            if type(i) is sub:
                return True
        return False
    if isinstance(sub, MessageComponent):  # 检查消息链中是否有某个组件
        for i in self:
            if i == sub:
                return True
        return False
    if isinstance(sub, MessageChain):  # 检查消息链中是否有某个子消息链
        return bool(kmp(self, sub))
    if isinstance(sub, str):  # 检查消息中有无指定字符串子串
        return sub in deserialize(str(self))
    raise TypeError(f"类型不匹配，当前类型：{type(sub)}")
def index(self, x: Union[MessageComponent, Type[MessageComponent]], i: int = 0, j: int = -1) ‑> int

返回 x 在消息链中首次出现项的索引号（索引号在 i 或其后且在 j 之前）。

Args

x (Union[MessageComponent, Type[MessageComponent]]):
要查找的消息元素或消息元素类型。
i
从哪个位置开始查找。
j
查找到哪个位置结束。

Returns

int
如果找到，则返回索引号。

Raises

ValueError
没有找到。
TypeError
类型不匹配。
Expand source code 
def index(
    self,
    x: Union[MessageComponent, Type[MessageComponent]],
    i: int = 0,
    j: int = -1
) -> int:
    """返回 x 在消息链中首次出现项的索引号（索引号在 i 或其后且在 j 之前）。

    Args:
        x (`Union[MessageComponent, Type[MessageComponent]]`):
            要查找的消息元素或消息元素类型。
        i: 从哪个位置开始查找。
        j: 查找到哪个位置结束。

    Returns:
        int: 如果找到，则返回索引号。

    Raises:
        ValueError: 没有找到。
        TypeError: 类型不匹配。
    """
    if isinstance(x, type):
        l = len(self)
        if i < 0:
            i += l
        if i < 0:
            i = 0
        if j < 0:
            j += l
        if j > l:
            j = l
        for index in range(i, j):
            if type(self[index]) is x:
                return index
        raise ValueError("消息链中不存在该类型的组件。")
    if isinstance(x, MessageComponent):
        return self.__root__.index(x, i, j)
    raise TypeError(f"类型不匹配，当前类型：{type(x)}")
def insert(self, i: int, x: Union[MessageComponent, str])

将一个消息元素或字符串添加到消息链中指定位置。

Args

i
插入位置。
x
消息元素或字符串元素。
Expand source code 
def insert(self, i: int, x: Union[MessageComponent, str]):
    """将一个消息元素或字符串添加到消息链中指定位置。

    Args:
        i: 插入位置。
        x: 消息元素或字符串元素。
    """
    self.__root__.insert(i, Plain(x) if isinstance(x, str) else x)
def pop(self, i: int = -1) ‑> MessageComponent

从消息链中移除并返回指定位置的元素。

Args

i
移除位置。默认为末尾。

Returns

MessageComponent
移除的元素。
Expand source code 
def pop(self, i: int = -1) -> MessageComponent:
    """从消息链中移除并返回指定位置的元素。

    Args:
        i: 移除位置。默认为末尾。

    Returns:
        MessageComponent: 移除的元素。
    """
    return self.__root__.pop(i)
def remove(self, x: Union[MessageComponent, Type[MessageComponent]])

从消息链中移除指定元素或指定类型的一个元素。

Args

x
指定的元素或元素类型。
Expand source code 
def remove(self, x: Union[MessageComponent, Type[MessageComponent]]):
    """从消息链中移除指定元素或指定类型的一个元素。

    Args:
        x: 指定的元素或元素类型。
    """
    if isinstance(x, type):
        self.pop(self.index(x))
    if isinstance(x, MessageComponent):
        self.__root__.remove(x)
def reverse(self)

将消息链原地翻转。

Expand source code 
def reverse(self):
    """将消息链原地翻转。"""
    self.__root__.reverse()
class MessageComponent (*args, **kwargs)

消息组件。

Expand source code 
class MessageComponent(MiraiIndexedModel, metaclass=MessageComponentMetaclass):
    """消息组件。"""
    type: str
    """消息组件类型。"""
    def __str__(self):
        return ''

    def as_mirai_code(self) -> str:
        """转化为 mirai 码。"""
        return ''

    def __repr__(self):
        return self.__class__.__name__ + '(' + ', '.join(
            (
                f'{k}={repr(v)}'
                for k, v in self.__dict__.items() if k != 'type' and v
            )
        ) + ')'

    def __init__(self, *args, **kwargs):
        # 解析参数列表，将位置参数转化为具名参数
        parameter_names = self.__parameter_names__
        if len(args) > len(parameter_names):
            raise TypeError(
                f'`{self.type}`需要{len(parameter_names)}个参数，但传入了{len(args)}个。'
            )
        for name, value in zip(parameter_names, args):
            if name in kwargs:
                raise TypeError(f'在 `{self.type}` 中，具名参数 `{name}` 与位置参数重复。')
            kwargs[name] = value

        super().__init__(**kwargs)

Ancestors

Subclasses

Class variables

var type : str

消息组件类型。

Methods

def as_mirai_code(self) ‑> str

转化为 mirai 码。

Expand source code 
def as_mirai_code(self) -> str:
    """转化为 mirai 码。"""
    return ''

Inherited members

class MessageEvent (*args, **kwargs)

消息事件。

Args

type
事件名。
message_chain
消息内容。
Expand source code 
class MessageEvent(Event):
    """消息事件。

    Args:
        type: 事件名。
        message_chain: 消息内容。
    """
    type: str
    """事件名。"""
    message_chain: MessageChain
    """消息内容。"""

Ancestors

Subclasses

Class variables

var message_chain : MessageChain

消息内容。

Inherited members

class MiraiCode (*args, **kwargs)

Mirai 码。

Expand source code 
class MiraiCode(MessageComponent):
    """Mirai 码。"""
    type: str = "MiraiCode"
    """消息组件类型。"""
    code: str
    """Mirai 码。"""
    def __str__(self):
        return serialize(self.code)

    def __repr__(self):
        return f'MiraiCode({self.code!r})'

Ancestors

Class variables

var code : str

Mirai 码。

Inherited members

class ModelEventBus

模型事件总线，实现底层事件总线上的事件再分发，以将事件解析到 Event 对象。

ModelEventBus 在注册事件处理器时，可使用 Event 类或事件名。关于可用的 Event 类， 参见模块 mirai.models.events。

模型事件总线支持的事件处理器接受唯一的参数event，该参数是一个 Event 对象，包含触发的事件的信息。

事件触发时，会自动按照 Event 类的继承关系向上级传播。

Args

event_chain_generator
一个函数， 输入事件名，返回一个生成此事件所在事件链的全部事件的事件名的生成器， 默认行为是事件链只包含单一事件。
Expand source code 
class ModelEventBus(EventBus):
    """模型事件总线，实现底层事件总线上的事件再分发，以将事件解析到 Event 对象。

    `ModelEventBus` 在注册事件处理器时，可使用 `Event` 类或事件名。关于可用的 `Event` 类，
    参见模块 `mirai.models.events`。

    模型事件总线支持的事件处理器接受唯一的参数`event`，该参数是一个 `Event` 对象，包含触发的事件的信息。

    事件触发时，会自动按照 `Event` 类的继承关系向上级传播。
    """
    def __init__(self):
        self.base_bus = EventBus(event_chain_generator=event_chain_parents)
        self._middlewares = defaultdict(type(None))

    def subscribe(
        self,
        event_type: Union[Type[Event], str],
        func: Callable,
        priority: int = 0
    ) -> None:
        """注册事件处理器。

        Args:
            event: 事件类型。
            func: 事件处理器。
            priority: 优先级，小者优先。
        """
        if isinstance(event_type, str):
            event_type = cast(Type[Event], Event.get_subtype(event_type))

        async def middleware(event: dict):
            """中间件。负责与底层 bus 沟通，将 event dict 解析为 Event 对象。"""
            event_model = cast(Event, Event.parse_subtype(event))
            logger.debug(f'收到事件 {event_model.type}。')
            return await async_with_exception(func(event_model))

        self._middlewares[func] = middleware
        self.base_bus.subscribe(event_type.__name__, middleware, priority)
        logger.debug(f'注册事件 {event_type.__name__} at {func}。')

    def unsubscribe(
        self, event_type: Union[Type[Event], str], func: Callable
    ) -> None:
        """移除事件处理器。

        Args:
            event_type: 事件类型。
            func: 事件处理器。
        """
        if isinstance(event_type, str):
            event_type = cast(Type[Event], Event.get_subtype(event_type))

        self.base_bus.unsubscribe(event_type.__name__, self._middlewares[func])
        del self._middlewares[func]
        logger.debug(f'解除事件注册 {event_type.__name__} at {func}。')

    def on(
        self,
        event_type: Union[Type[Event], str],
        priority: int = 0,
    ) -> Callable:
        """以装饰器的方式注册事件处理器。

        例如：
        ```py
        @bus.on(FriendMessage)
        def my_event_handler(event: FriendMessage):
            print(event.sender.id)
        ```

        Args:
            event_type: 事件类型或事件名。
            priority: 优先级，小者优先。
        """
        def decorator(func: Callable) -> Callable:
            self.subscribe(event_type, func, priority)
            return func

        return decorator

    async def emit(self, event: Union[Event, str], *args,
                   **kwargs) -> List[Awaitable[Any]]:
        """触发一个事件。

        Args:
            event: 要触发的事件。
            *args: 参数。
            **kwargs: 参数。
        """
        if isinstance(event, str):
            return await super().emit(event, *args, **kwargs)

        return await self.base_bus.emit(
            event.type, event.dict(by_alias=True, exclude_none=True)
        )

Ancestors

Methods

async def emit(self, event: Union[Event, str], *args, **kwargs) ‑> List[Awaitable[Any]]

触发一个事件。

Args

event
要触发的事件。
*args
参数。
**kwargs
参数。
Expand source code 
async def emit(self, event: Union[Event, str], *args,
               **kwargs) -> List[Awaitable[Any]]:
    """触发一个事件。

    Args:
        event: 要触发的事件。
        *args: 参数。
        **kwargs: 参数。
    """
    if isinstance(event, str):
        return await super().emit(event, *args, **kwargs)

    return await self.base_bus.emit(
        event.type, event.dict(by_alias=True, exclude_none=True)
    )
def on(self, event_type: Union[Type[Event], str], priority: int = 0) ‑> Callable

以装饰器的方式注册事件处理器。

例如：

@bus.on(FriendMessage)
def my_event_handler(event: FriendMessage):
    print(event.sender.id)

Args

event_type
事件类型或事件名。
priority
优先级，小者优先。
Expand source code 
def on(
    self,
    event_type: Union[Type[Event], str],
    priority: int = 0,
) -> Callable:
    """以装饰器的方式注册事件处理器。

    例如：
    ```py
    @bus.on(FriendMessage)
    def my_event_handler(event: FriendMessage):
        print(event.sender.id)
    ```

    Args:
        event_type: 事件类型或事件名。
        priority: 优先级，小者优先。
    """
    def decorator(func: Callable) -> Callable:
        self.subscribe(event_type, func, priority)
        return func

    return decorator
def subscribe(self, event_type: Union[Type[Event], str], func: Callable, priority: int = 0) ‑> NoneType

注册事件处理器。

Args

event
事件类型。
func
事件处理器。
priority
优先级，小者优先。
Expand source code 
def subscribe(
    self,
    event_type: Union[Type[Event], str],
    func: Callable,
    priority: int = 0
) -> None:
    """注册事件处理器。

    Args:
        event: 事件类型。
        func: 事件处理器。
        priority: 优先级，小者优先。
    """
    if isinstance(event_type, str):
        event_type = cast(Type[Event], Event.get_subtype(event_type))

    async def middleware(event: dict):
        """中间件。负责与底层 bus 沟通，将 event dict 解析为 Event 对象。"""
        event_model = cast(Event, Event.parse_subtype(event))
        logger.debug(f'收到事件 {event_model.type}。')
        return await async_with_exception(func(event_model))

    self._middlewares[func] = middleware
    self.base_bus.subscribe(event_type.__name__, middleware, priority)
    logger.debug(f'注册事件 {event_type.__name__} at {func}。')
def unsubscribe(self, event_type: Union[Type[Event], str], func: Callable) ‑> NoneType

移除事件处理器。

Args

event_type
事件类型。
func
事件处理器。
Expand source code 
def unsubscribe(
    self, event_type: Union[Type[Event], str], func: Callable
) -> None:
    """移除事件处理器。

    Args:
        event_type: 事件类型。
        func: 事件处理器。
    """
    if isinstance(event_type, str):
        event_type = cast(Type[Event], Event.get_subtype(event_type))

    self.base_bus.unsubscribe(event_type.__name__, self._middlewares[func])
    del self._middlewares[func]
    logger.debug(f'解除事件注册 {event_type.__name__} at {func}。')
class MusicShare (*args, **kwargs)

音乐分享。

Expand source code 
class MusicShare(MessageComponent):
    """音乐分享。"""
    type: str = "MusicShare"
    """消息组件类型。"""
    kind: MusicShareKind
    """音乐分享的来源。"""
    title: str
    """标题。"""
    summary: str
    """歌手。"""
    jump_url: HttpUrl
    """跳转路径。"""
    picture_url: HttpUrl
    """封面路径。"""
    music_url: HttpUrl
    """音源路径。"""
    brief: str = ""
    """在消息列表中显示的内容。"""
    def __str__(self):
        return self.brief

Ancestors

Class variables

var brief : str

在消息列表中显示的内容。

var jump_url : pydantic.networks.HttpUrl

跳转路径。

var kind : MusicShareKind

音乐分享的来源。

var music_url : pydantic.networks.HttpUrl

音源路径。

var picture_url : pydantic.networks.HttpUrl

封面路径。

var summary : str

歌手。

var title : str

标题。

Inherited members

class MusicShareKind (value, names=None, *, module=None, qualname=None, type=None, start=1)

音乐分享的来源。

Expand source code 
class MusicShareKind(str, Enum):
    """音乐分享的来源。"""
    NeteaseCloudMusic = "NeteaseCloudMusic"
    QQMusic = "QQMusic"
    MiguMusic = "MiguMusic"
    KugouMusic = "KugouMusic"
    KuwoMusic = "KuwoMusic"

Ancestors

  • builtins.str
  • enum.Enum

Class variables

var KugouMusic
var KuwoMusic
var MiguMusic
var NeteaseCloudMusic
var QQMusic
class NewFriendRequestEvent (*args, **kwargs)

添加好友申请。

Args

type
事件名。
event_id
事件标识，响应该事件时的标识。
from_id
申请人 QQ 号。
group_id
申请人如果通过某个群添加好友，该项为该群群号；否则为0。
nick
申请人的昵称或群名片。
message
申请消息。
Expand source code 
class NewFriendRequestEvent(RequestEvent):
    """添加好友申请。

    Args:
        type: 事件名。
        event_id: 事件标识，响应该事件时的标识。
        from_id: 申请人 QQ 号。
        group_id: 申请人如果通过某个群添加好友，该项为该群群号；否则为0。
        nick: 申请人的昵称或群名片。
        message: 申请消息。
    """
    type: str = 'NewFriendRequestEvent'
    """事件名。"""
    event_id: int
    """事件标识，响应该事件时的标识。"""
    from_id: int
    """申请人 QQ 号。"""
    group_id: int
    """申请人如果通过某个群添加好友，该项为该群群号；否则为0。"""
    nick: str
    """申请人的昵称或群名片。"""
    message: str
    """申请消息。"""

Ancestors

Class variables

var message : str

申请消息。

var nick : str

申请人的昵称或群名片。

Inherited members

class NudgeEvent (*args, **kwargs)

头像戳一戳事件。

Args

type
事件名。
from_id
动作发出者的 QQ 号。
target
动作目标的 QQ 号。
subject
来源。
action
戳一戳类型。
suffix
自定义戳一戳内容。
Expand source code 
class NudgeEvent(Event):
    """头像戳一戳事件。

    Args:
        type: 事件名。
        from_id: 动作发出者的 QQ 号。
        target: 动作目标的 QQ 号。
        subject: 来源。
        action: 戳一戳类型。
        suffix: 自定义戳一戳内容。
    """
    type: str = "NudgeEvent"
    """事件名。"""
    from_id: int
    """动作发出者的 QQ 号。"""
    target: int
    """动作目标的 QQ 号。"""
    subject: Subject
    """来源。"""
    action: str
    """戳一戳类型。"""
    suffix: str
    """自定义戳一戳内容。"""

Ancestors

Class variables

var action : str

戳一戳类型。

var from_id : int

动作发出者的 QQ 号。

var subject : Subject

来源。

var suffix : str

自定义戳一戳内容。

var target : int

动作目标的 QQ 号。

Inherited members

class OtherClientEvent (*args, **kwargs)

其它客户端事件。

Args

type
事件名。
client
其他设备。
Expand source code 
class OtherClientEvent(Event):
    """其它客户端事件。

    Args:
        type: 事件名。
        client: 其他设备。
    """
    type: str
    """事件名。"""
    client: Client
    """其他设备。"""

Ancestors

Subclasses

  • mirai.models.events.OtherClientOfflineEvent
  • mirai.models.events.OtherClientOnlineEvent

Class variables

var client : mirai.models.entities.Client

其他设备。

Inherited members

class OtherClientMessage (*args, **kwargs)

其他客户端消息。

Args

type
事件名。
sender
发送消息的人。
message_chain
消息内容。
Expand source code 
class OtherClientMessage(MessageEvent):
    """其他客户端消息。

    Args:
        type: 事件名。
        sender: 发送消息的人。
        message_chain: 消息内容。
    """
    type: str = 'OtherClientMessage'
    """事件名。"""
    sender: Client
    """发送消息的人。"""
    message_chain: MessageChain
    """消息内容。"""

Ancestors

Class variables

var sender : mirai.models.entities.Client

发送消息的人。

Inherited members

class OtherClientOfflineEvent (*args, **kwargs)

其它客户端下线事件。

Args

type
事件名。
client
其他设备。
Expand source code 
class OtherClientOfflineEvent(OtherClientEvent):
    """其它客户端下线事件。

    Args:
        type: 事件名。
        client: 其他设备。
    """
    type: str = 'OtherClientOfflineEvent'
    """事件名。"""
    client: Client
    """其他设备。"""

Ancestors

Class variables

var client : mirai.models.entities.Client

其他设备。

Inherited members

class OtherClientOnlineEvent (*args, **kwargs)

其它客户端上线事件。

Args

type
事件名。
client
其他设备。
kind
详细设备类型。
Expand source code 
class OtherClientOnlineEvent(OtherClientEvent):
    """其它客户端上线事件。

    Args:
        type: 事件名。
        client: 其他设备。
        kind: 详细设备类型。
    """
    type: str = 'OtherClientOnlineEvent'
    """事件名。"""
    client: Client
    """其他设备。"""
    kind: Optional[ClientKind] = None
    """详细设备类型。"""

Ancestors

Class variables

var client : mirai.models.entities.Client

其他设备。

var kind : Optional[mirai.models.events.ClientKind]

详细设备类型。

Inherited members

class Permission (value, names=None, *, module=None, qualname=None, type=None, start=1)

群成员身份权限。

Expand source code 
class Permission(str, Enum):
    """群成员身份权限。"""
    Member = "MEMBER"
    """成员。"""
    Administrator = "ADMINISTRATOR"
    """管理员。"""
    Owner = "OWNER"
    """群主。"""
    def __repr__(self) -> str:
        return repr(self.value)

Ancestors

  • builtins.str
  • enum.Enum

Class variables

var Administrator

管理员。

var Member

成员。

var Owner

群主。

class Plain (*args, **kwargs)

纯文本。

Expand source code 
class Plain(MessageComponent):
    """纯文本。"""
    type: str = "Plain"
    """消息组件类型。"""
    text: str
    """文字消息。"""
    def __str__(self):
        return self.text

    def as_mirai_code(self) -> str:
        return serialize(self.text)

    def __repr__(self):
        return f'Plain({self.text!r})'

Ancestors

Class variables

var text : str

文字消息。

Inherited members

class Poke (*args, **kwargs)

戳一戳。

Expand source code 
class Poke(MessageComponent):
    """戳一戳。"""
    type: str = "Poke"
    """消息组件类型。"""
    name: PokeNames
    """名称。"""
    @property
    def poke_type(self):
        return POKE_TYPE[self.name]

    @property
    def poke_id(self):
        return POKE_ID[self.name]

    def __str__(self):
        return f'[{POKE_NAME[self.name]}]'

    def as_mirai_code(self) -> str:
        return f'[mirai:poke:{self.name},{self.poke_type},{self.poke_id}]'

Ancestors

Class variables

var name : PokeNames

名称。

Instance variables

var poke_id
Expand source code 
@property
def poke_id(self):
    return POKE_ID[self.name]
var poke_type
Expand source code 
@property
def poke_type(self):
    return POKE_TYPE[self.name]

Inherited members

class PokeNames (value, names=None, *, module=None, qualname=None, type=None, start=1)

戳一戳名称。

Expand source code 
class PokeNames(str, Enum):
    """戳一戳名称。"""
    ChuoYiChuo = "ChuoYiChuo"
    BiXin = "BiXin"
    DianZan = "DianZan"
    XinSui = "XinSui"
    LiuLiuLiu = "LiuLiuLiu"
    FangDaZhao = "FangDaZhao"
    BaoBeiQiu = "BaoBeiQiu"
    Rose = "Rose"
    ZhaoHuanShu = "ZhaoHuanShu"
    RangNiPi = "RangNiPi"
    JeiYin = "JeiYin"
    ShouLei = "ShouLei"
    GouYin = "GouYin"
    ZhuaYiXia = "ZhuaYiXia"
    SuiPing = "SuiPing"
    QiaoMen = "QiaoMen"

    def as_component(self) -> 'Poke':
        return Poke(name=self.value)

Ancestors

  • builtins.str
  • enum.Enum

Class variables

var BaoBeiQiu
var BiXin
var ChuoYiChuo
var DianZan
var FangDaZhao
var GouYin
var JeiYin
var LiuLiuLiu
var QiaoMen
var RangNiPi
var Rose
var ShouLei
var SuiPing
var XinSui
var ZhaoHuanShu
var ZhuaYiXia

Methods

def as_component(self) ‑> Poke
Expand source code 
def as_component(self) -> 'Poke':
    return Poke(name=self.value)
class Quote (*args, **kwargs)

引用。

Expand source code 
class Quote(MessageComponent):
    """引用。"""
    type: str = "Quote"
    """消息组件类型。"""
    id: Optional[int] = None
    """被引用回复的原消息的 message_id。"""
    group_id: Optional[int] = None
    """被引用回复的原消息所接收的群号，当为好友消息时为0。"""
    sender_id: Optional[int] = None
    """被引用回复的原消息的发送者的QQ号。"""
    target_id: Optional[int] = None
    """被引用回复的原消息的接收者者的QQ号（或群号）。"""
    origin: MessageChain
    """被引用回复的原消息的消息链对象。"""
    @validator("origin", always=True, pre=True)
    def origin_formater(cls, v):
        return MessageChain.parse_obj(v)

Ancestors

Class variables

var group_id : Optional[int]

被引用回复的原消息所接收的群号，当为好友消息时为0。

var id : Optional[int]

被引用回复的原消息的 message_id。

var origin : MessageChain

被引用回复的原消息的消息链对象。

var sender_id : Optional[int]

被引用回复的原消息的发送者的QQ号。

var target_id : Optional[int]

被引用回复的原消息的接收者者的QQ号（或群号）。

Static methods

def origin_formater(v)
Expand source code 
@validator("origin", always=True, pre=True)
def origin_formater(cls, v):
    return MessageChain.parse_obj(v)

Inherited members

class RequestEvent (*args, **kwargs)

申请事件。

Args

type
事件名。
event_id
事件标识，响应该事件时的标识。
Expand source code 
class RequestEvent(Event):
    """申请事件。

    Args:
        type: 事件名。
        event_id: 事件标识，响应该事件时的标识。
    """
    type: str
    """事件名。"""
    event_id: int
    """事件标识，响应该事件时的标识。"""
    from_id: int
    """申请人 QQ 号。"""
    group_id: int
    """申请人群号，可能为0。"""

Ancestors

Subclasses

Class variables

var event_id : int

事件标识，响应该事件时的标识。

var from_id : int

申请人 QQ 号。

var group_id : int

申请人群号，可能为0。

Inherited members

class Source (*args, **kwargs)

源。包含消息的基本信息。

Expand source code 
class Source(MessageComponent):
    """源。包含消息的基本信息。"""
    type: str = "Source"
    """消息组件类型。"""
    id: int
    """消息的识别号，用于引用回复（Source 类型永远为 MessageChain 的第一个元素）。"""
    time: datetime
    """消息时间。"""

Ancestors

Class variables

var id : int

消息的识别号，用于引用回复（Source 类型永远为 MessageChain 的第一个元素）。

var time : datetime.datetime

消息时间。

Inherited members

class StrangerMessage (*args, **kwargs)

陌生人消息。

Args

type
事件名。
sender
发送消息的人。
message_chain
消息内容。
Expand source code 
class StrangerMessage(MessageEvent):
    """陌生人消息。

    Args:
        type: 事件名。
        sender: 发送消息的人。
        message_chain: 消息内容。
    """
    type: str = 'StrangerMessage'
    """事件名。"""
    sender: Friend
    """发送消息的人。"""
    message_chain: MessageChain
    """消息内容。"""

Ancestors

Class variables

var sender : Friend

发送消息的人。

Inherited members

class StrangerSyncMessage (*args, **kwargs)

其他客户端发送的陌生人消息。

Args

type
事件名。
subject
发送消息的目标。
message_chain
消息内容。
Expand source code 
class StrangerSyncMessage(MessageEvent):
    """其他客户端发送的陌生人消息。

    Args:
        type: 事件名。
        subject: 发送消息的目标。
        message_chain: 消息内容。
    """
    type: str = 'StrangerSyncMessage'
    """事件名。"""
    subject: Friend
    """发送消息的目标。"""
    message_chain: MessageChain
    """消息内容。"""

Ancestors

Class variables

var subject : Friend

发送消息的目标。

Inherited members

class TempMessage (*args, **kwargs)

群临时消息。

Args

type
事件名。
sender
发送消息的群成员。
message_chain
消息内容。
Expand source code 
class TempMessage(MessageEvent):
    """群临时消息。

    Args:
        type: 事件名。
        sender: 发送消息的群成员。
        message_chain: 消息内容。
    """
    type: str = 'TempMessage'
    """事件名。"""
    sender: GroupMember
    """发送消息的群成员。"""
    message_chain: MessageChain
    """消息内容。"""
    @property
    def group(self) -> Group:
        return self.sender.group

Ancestors

Class variables

var sender : GroupMember

发送消息的群成员。

Instance variables

var group : Group
Expand source code 
@property
def group(self) -> Group:
    return self.sender.group

Inherited members

class TempSyncMessage (*args, **kwargs)

其他客户端发送的群临时消息。

Args

type
事件名。
subject
发送消息的目标群成员。
message_chain
消息内容。
Expand source code 
class TempSyncMessage(MessageEvent):
    """其他客户端发送的群临时消息。

    Args:
        type: 事件名。
        subject: 发送消息的目标群成员。
        message_chain: 消息内容。
    """
    type: str = 'TempSyncMessage'
    """事件名。"""
    subject: GroupMember
    """发送消息的目标群成员。"""
    message_chain: MessageChain
    """消息内容。"""
    @property
    def group(self) -> Group:
        return self.subject.group

Ancestors

Class variables

var subject : GroupMember

发送消息的目标群成员。

Instance variables

var group : Group
Expand source code 
@property
def group(self) -> Group:
    return self.subject.group

Inherited members

class Unknown (*args, **kwargs)

未知。

Expand source code 
class Unknown(MessageComponent):
    """未知。"""
    type: str = "Unknown"
    """消息组件类型。"""
    text: str
    """文本。"""

Ancestors

Class variables

var text : str

文本。

Inherited members

class Voice (*args, **kwargs)

语音。

Expand source code 
class Voice(MessageComponent):
    """语音。"""
    type: str = "Voice"
    """消息组件类型。"""
    voice_id: Optional[str] = None
    """语音的 voice_id，不为空时将忽略 url 属性。"""
    url: Optional[str] = None
    """语音的 URL，发送时可作网络语音的链接；接收时为腾讯语音服务器的链接，可用于语音下载。"""
    path: Optional[str] = None
    """语音的路径，发送本地语音。"""
    base64: Optional[str] = None
    """语音的 Base64 编码。"""
    length: Optional[int] = None
    """语音的长度，单位为秒。"""
    @validator('path')
    def validate_path(cls, path: Optional[str]):
        """修复 path 参数的行为，使之相对于 YiriMirai 的启动路径。"""
        if path:
            try:
                return str(Path(path).resolve(strict=True))
            except FileNotFoundError:
                raise ValueError(f"无效路径：{path}")
        else:
            return path

    def __str__(self):
        return '[语音]'

    async def download(
        self,
        filename: Union[str, Path, None] = None,
        directory: Union[str, Path, None] = None
    ):
        """下载语音到本地。

        语音采用 silk v3 格式，silk 格式的编码解码请使用 [graiax-silkcoder](https://pypi.org/project/graiax-silkcoder/)。

        Args:
            filename: 下载到本地的文件路径。与 `directory` 二选一。
            directory: 下载到本地的文件夹路径。与 `filename` 二选一。
        """
        if not self.url:
            logger.warning(f'语音 `{self.voice_id}` 无 url 参数，下载失败。')
            return

        import httpx
        async with httpx.AsyncClient() as client:
            response = await client.get(self.url)
            response.raise_for_status()
            content = response.content

            if filename:
                path = Path(filename)
                path.parent.mkdir(parents=True, exist_ok=True)
            elif directory:
                path = Path(directory)
                path.mkdir(parents=True, exist_ok=True)
                path = path / f'{self.voice_id}.silk'
            else:
                raise ValueError("请指定文件路径或文件夹路径！")

            import aiofiles
            async with aiofiles.open(path, 'wb') as f:
                await f.write(content)

    @classmethod
    async def from_local(
        cls,
        filename: Union[str, Path, None] = None,
        content: Optional[bytes] = None,
    ) -> "Voice":
        """从本地文件路径加载语音，以 base64 的形式传递。

        Args:
            filename: 从本地文件路径加载语音，与 `content` 二选一。
            content: 从本地文件内容加载语音，与 `filename` 二选一。
        """
        if content:
            pass
        if filename:
            path = Path(filename)
            import aiofiles
            async with aiofiles.open(path, 'rb') as f:
                content = await f.read()
        else:
            raise ValueError("请指定语音路径或语音内容！")
        import base64
        img = cls(base64=base64.b64encode(content).decode())
        return img

Ancestors

Class variables

var base64 : Optional[str]

语音的 Base64 编码。

var length : Optional[int]

语音的长度，单位为秒。

var path : Optional[str]

语音的路径，发送本地语音。

var url : Optional[str]

语音的 URL，发送时可作网络语音的链接；接收时为腾讯语音服务器的链接，可用于语音下载。

var voice_id : Optional[str]

语音的 voice_id，不为空时将忽略 url 属性。

Static methods

async def from_local(filename: Union[str, pathlib.Path, NoneType] = None, content: Optional[bytes] = None) ‑> Voice

从本地文件路径加载语音，以 base64 的形式传递。

Args

filename
从本地文件路径加载语音，与 content 二选一。
content
从本地文件内容加载语音，与 filename 二选一。
Expand source code 
@classmethod
async def from_local(
    cls,
    filename: Union[str, Path, None] = None,
    content: Optional[bytes] = None,
) -> "Voice":
    """从本地文件路径加载语音，以 base64 的形式传递。

    Args:
        filename: 从本地文件路径加载语音，与 `content` 二选一。
        content: 从本地文件内容加载语音，与 `filename` 二选一。
    """
    if content:
        pass
    if filename:
        path = Path(filename)
        import aiofiles
        async with aiofiles.open(path, 'rb') as f:
            content = await f.read()
    else:
        raise ValueError("请指定语音路径或语音内容！")
    import base64
    img = cls(base64=base64.b64encode(content).decode())
    return img
def validate_path(path: Optional[str])

修复 path 参数的行为，使之相对于 YiriMirai 的启动路径。

Expand source code 
@validator('path')
def validate_path(cls, path: Optional[str]):
    """修复 path 参数的行为，使之相对于 YiriMirai 的启动路径。"""
    if path:
        try:
            return str(Path(path).resolve(strict=True))
        except FileNotFoundError:
            raise ValueError(f"无效路径：{path}")
    else:
        return path

Methods

async def download(self, filename: Union[str, pathlib.Path, NoneType] = None, directory: Union[str, pathlib.Path, NoneType] = None)

下载语音到本地。

语音采用 silk v3 格式，silk 格式的编码解码请使用 graiax-silkcoder。

Args

filename
下载到本地的文件路径。与 directory 二选一。
directory
下载到本地的文件夹路径。与 filename 二选一。
Expand source code 
async def download(
    self,
    filename: Union[str, Path, None] = None,
    directory: Union[str, Path, None] = None
):
    """下载语音到本地。

    语音采用 silk v3 格式，silk 格式的编码解码请使用 [graiax-silkcoder](https://pypi.org/project/graiax-silkcoder/)。

    Args:
        filename: 下载到本地的文件路径。与 `directory` 二选一。
        directory: 下载到本地的文件夹路径。与 `filename` 二选一。
    """
    if not self.url:
        logger.warning(f'语音 `{self.voice_id}` 无 url 参数，下载失败。')
        return

    import httpx
    async with httpx.AsyncClient() as client:
        response = await client.get(self.url)
        response.raise_for_status()
        content = response.content

        if filename:
            path = Path(filename)
            path.parent.mkdir(parents=True, exist_ok=True)
        elif directory:
            path = Path(directory)
            path.mkdir(parents=True, exist_ok=True)
            path = path / f'{self.voice_id}.silk'
        else:
            raise ValueError("请指定文件路径或文件夹路径！")

        import aiofiles
        async with aiofiles.open(path, 'wb') as f:
            await f.write(content)

Inherited members

class Xml (*args, **kwargs)

XML。

Expand source code 
class Xml(MessageComponent):
    """XML。"""
    type: str = "Xml"
    """消息组件类型。"""
    xml: str
    """XML文本。"""

Ancestors

Class variables

var xml : str

XML文本。

Inherited members