Skip to content

事件系统与模型源码解析 (Event System) ​

很多同学看文档最讨厌的就是只写一句“这里有个群事件”,然后字段全靠猜对不对?
在 developTools 中,所有事件都是由 Pydantic 驱动的强类型数据模型。OneBot 协议端送来的原始 JSON 字典,在进入你的处理器前就已经被解析得清清爽爽啦!

今天就直接把源码里的模型定义贴到你眼前,逐行配上大白话讲解,让你清清楚楚知道每一个字段的类型、含义与神仙用法!


一、基础模型与消息发送者源码 ​

在分析具体事件前,我们先看两个最常打交道的基础数据类——消息发送者 Sender 与引用回复 Reply:

python
# 位于 developTools/event/events.py

class Sender(BaseModel):
    user_id: Optional[int] = None       # 发送者的 QQ 号
    nickname: Optional[str] = None      # 用户昵称
    sex: Optional[str] = None           # 性别(male / female / unknown)
    age: Optional[int] = None           # 年龄
    card: Optional[str] = None          # 群名片 / 群备注(仅群聊有效,未设置则为空)
    area: Optional[str] = None          # 地区
    level: Optional[str] = None         # 成员等级
    role: Optional[str] = None          # 成员角色: "owner"(群主) / "admin"(管理) / "member"(普通成员)
    title: Optional[str] = None         # 专属头衔

    model_config = ConfigDict(extra="allow")


class Reply(BaseModel):
    time: int                           # 原消息发送时间戳
    message_type: str                   # 消息类型 ("group" / "private")
    message_id: int                     # 原消息 ID
    real_id: int                        # 真实消息 ID
    sender: Sender                      # 原消息发送者信息
    message: List[Dict[str, Any]]       # 原消息包含的内容列表

    model_config = ConfigDict(extra="allow")

    @validator("message", pre=True, always=True)
    def ensure_message_chain(cls, value):
        """自动将原始数据封装为 MessageChain 便捷操作链"""
        return MessageChain(value)

获取称呼的最佳实践

在群聊里称呼用户时,推荐优先使用 event.sender.card or event.sender.nickname or str(event.user_id),这样既尊重群友设置的个性群昵称,在未设置时也能优雅回落到 QQ 昵称!


二、消息事件 (Message Events) 源码详解 ​

消息事件是机器人处理最频繁的事件。所有消息事件都继承自 MessageEvent 基类。

1. MessageEvent 基类源码 ​

python
# 位于 developTools/event/events.py

class MessageEvent(BaseModel):
    """消息事件基类"""

    post_type: Literal["message"]       # 上报类型,固定为 "message"
    sub_type: str                       # 子类型,例如 "normal", "anonymous", "friend"
    user_id: int                        # 发送者的 QQ 号
    message_type: str                   # 消息类型 ("group" 或 "private")
    message_id: int                     # 消息唯一 ID(可用于撤回或引用)
    message: List[Dict[str, Any]]       # 原始消息段列表
    original_message: Optional[list] = None
    _raw_message: str                   # 包含 CQ 码的原始消息字符串
    font: int                           # 字体 ID
    sender: Sender                      # 发送者结构体(见上一节)
    to_me: bool = False                 # 是否艾特了 Bot 或者包含 Bot 昵称
    reply: Optional[Reply] = None       # 若本条消息引用回复了别人,则该字段不为空

    processed_message: List[Dict[str, Union[str, Dict]]] = []
    message_chain: MessageChain = []
    adapter_source: str = "onebot_v11"

    model_config = ConfigDict(extra="allow", arbitrary_types_allowed=True)

    @property
    def pure_text(self) -> str:
        """最常用!智能提取去除 CQ 码、图片等修饰后的纯文本"""
        return self.message_chain.fetch_text()

    @property
    def raw_message(self):
        return self._raw_message

    def get(self, message_type: str):
        """按类型快速提取消息段(例如 'image', 'record')"""
        result = [msg[message_type] for msg in self.processed_message if message_type in msg]
        if result and message_type == "image" and "url" not in result[0]:
            result[0]["url"] = result[0]["file"]
        return result if result else None

🌟 核心高频属性与方法: ​

  1. event.pure_text:你最经常用的属性!自动将消息里的纯文字部分拼接并去除多余空白,方便你做 if "签到" in event.pure_text 的逻辑判断。
  2. event.reply:当用户“引用回复”某条消息时,这个属性会有值。你可以通过 event.reply.message_id 获取被引用的消息 ID,通过 event.reply.sender.user_id 知道他回复的是谁!
  3. event.get("image"):如果消息里发了图片,调用 imgs = event.get("image") 就能直接拿到包含图片 URL 和文件名的字典列表!
  4. event.to_me:布尔值,代表该消息是否明确艾特了机器人。

2. 群消息事件:GroupMessageEvent 源码 ​

python
class GroupMessageEvent(MessageEvent):
    """群消息事件"""

    message_type: Literal["group"]      # 固定为 "group"
    group_id: int                       # 触发消息的 QQ 群号
    anonymous: Optional[Anonymous] = None # 若为匿名发言则包含匿名信息,否则为 None

实战监听示例: ​

python
from developTools.event.events import GroupMessageEvent
from developTools.message.message_components import Text, At

@bot.on(GroupMessageEvent)
async def on_group_msg(event: GroupMessageEvent):
    # 打印消息来源群与发送者
    print(f"群 {event.group_id} 的 {event.sender.nickname} 说: {event.pure_text}")
    
    # 简单的打招呼示例
    if event.pure_text == "早安":
        await bot.send(event, [At(event.user_id), Text(" 早上好呀~ 今天也要元气满满哦!")])

3. 私聊消息事件:PrivateMessageEvent 源码 ​

python
class PrivateMessageEvent(MessageEvent):
    """私聊消息事件"""

    message_type: Literal["private"]    # 固定为 "private"

实战监听示例: ​

python
from developTools.event.events import PrivateMessageEvent

@bot.on(PrivateMessageEvent)
async def on_private_msg(event: PrivateMessageEvent):
    # 私聊中直接回复
    if event.pure_text == "ping":
        await bot.send(event, "pong! 私聊通道畅通无阻哦~")

三、通知提醒事件 (Notice Events) 源码详解 ​

当群里发生“撤回消息”、“戳一戳”、“成员进出群”或“管理变更”时,协议端会上报通知事件。所有通知事件继承自 NoticeEvent:

python
class NoticeEvent(EventBase):
    """通知事件基类"""
    post_type: Literal["notice"]        # 固定为 "notice"
    notice_type: str                    # 通知细分类型

1. 戳一戳事件:PokeNotifyEvent 源码 ​

python
class PokeNotifyEvent(NoticeEvent):
    """戳一戳提醒事件"""

    sub_type: Literal["poke"]
    target_id: int                      # 被戳的人的 QQ 号(若被戳的是 Bot,则 target_id == bot.id)
    user_id: int                        # 发起戳一戳的人的 QQ 号
    group_id: Optional[int] = None       # 若在群内戳一戳则为群号;私聊双击头像则为 None
    raw_info: list = None

实战:反弹戳一戳 ​

python
from developTools.event.events import PokeNotifyEvent

@bot.on(PokeNotifyEvent)
async def on_poke(event: PokeNotifyEvent):
    # 如果被戳的目标是机器人本体
    if event.target_id == bot.id:
        if event.group_id:
            # 群内反手戳回去!
            await bot.group_poke(event.group_id, event.user_id)
        else:
            # 好友私聊反手戳回去!
            await bot.friend_poke(event.user_id)

2. 群禁言事件:GroupBanNoticeEvent 源码 ​

python
class GroupBanNoticeEvent(NoticeEvent):
    """群禁言事件"""

    notice_type: Literal["group_ban"]
    sub_type: str                       # "ban"(禁言) 或 "lift_ban"(解除禁言)
    user_id: int                        # 被禁言的成员 QQ(全员禁言时为 0)
    group_id: int                       # 群号
    operator_id: int                    # 执行操作的管理员/群主 QQ 号
    duration: int                       # 禁言时长(单位:秒,为 0 时表示解除禁言)

3. 成员增减事件:GroupIncreaseNoticeEvent & GroupDecreaseNoticeEvent 源码 ​

python
class GroupIncreaseNoticeEvent(NoticeEvent):
    """群成员增加事件(新人入群 / 邀请入群)"""

    notice_type: Literal["group_increase"]
    sub_type: str                       # "approve"(管理员审批入群) 或 "invite"(群员邀请入群)
    user_id: int                        # 加入的新成员 QQ 号
    group_id: int                       # 对应群号
    operator_id: int                    # 审批人或邀请人的 QQ 号


class GroupDecreaseNoticeEvent(NoticeEvent):
    """群成员减少事件(主动退群 / 被踢出群)"""

    notice_type: Literal["group_decrease"]
    sub_type: str                       # "leave"(主动退群) / "kick"(被踢) / "kick_me"(Bot自己被移出群)
    user_id: int                        # 离开的成员 QQ 号
    group_id: int                       # 对应群号
    operator_id: int                    # 操作者 QQ 号(主动退群时通常与 user_id 相同)

实战:入群欢迎 ​

python
@bot.on(GroupIncreaseNoticeEvent)
async def welcome_new_member(event: GroupIncreaseNoticeEvent):
    await bot.send_group_message(
        event.group_id, 
        [Text(f"欢迎新人 [CQ:at,qq={event.user_id}] 进群!请先阅读群公告哦~")]
    )

4. 消息撤回事件:GroupRecallNoticeEvent & FriendRecallNoticeEvent 源码 ​

python
class GroupRecallNoticeEvent(NoticeEvent):
    """群消息撤回事件"""

    notice_type: Literal["group_recall"]
    user_id: int                        # 消息原作者的 QQ 号
    group_id: int                       # 群号
    operator_id: int                    # 执行撤回操作的人(可能是群主/管理员代撤回)
    message_id: int                     # 被撤回的消息 ID


class FriendRecallNoticeEvent(NoticeEvent):
    """好友私聊消息撤回事件"""

    notice_type: Literal["friend_recall"]
    user_id: int                        # 撤回消息的好友 QQ 号
    message_id: int                     # 被撤回的消息 ID

四、请求事件 (Request Events) 源码详解 ​

当有人主动申请加机器人为好友,或者邀请机器人进群时触发。

1. 加好友请求:FriendRequestEvent 源码 ​

python
class FriendRequestEvent(RequestEvent):
    """加好友请求事件"""

    post_type: Literal["request"]       # 固定为 "request"
    request_type: Literal["friend"]     # 固定为 "friend"
    user_id: int                        # 申请人 QQ 号
    flag: str                           # 请求标识符(用于调用审批 API)
    comment: Optional[str] = None       # 申请附带的验证消息内容

实战:自动通过好友验证并打招呼 ​

python
@bot.on(FriendRequestEvent)
async def auto_approve_friend(event: FriendRequestEvent):
    print(f"收到来自 {event.user_id} 的好友申请,验证信息: {event.comment}")
    # 调用 bot.handle_friend_request 直接同意!
    await bot.handle_friend_request(flag=event.flag, approve=True, remark="群友小可爱")

2. 加群与邀请请求:GroupRequestEvent 源码 ​

python
class GroupRequestEvent(RequestEvent):
    """加群申请 / 邀请入群事件"""

    post_type: Literal["request"]
    request_type: Literal["group"]
    sub_type: str                       # "add"(有人申请加群) 或 "invite"(有人邀请 Bot 进群)
    group_id: int                       # 目标群号
    user_id: int                        # 申请人或邀请人 QQ 号
    flag: str                           # 请求标识符(用于调用审批 API)
    comment: Optional[str] = None       # 申请理由

实战:审批加群/邀请 ​

python
@bot.on(GroupRequestEvent)
async def handle_group_req(event: GroupRequestEvent):
    if event.sub_type == "invite":
        # 如果是别人邀请 Bot 进群,直接同意
        await bot.set_group_add_request(flag=event.flag, approve=True)
    elif event.sub_type == "add":
        # 配合管理员权限自动处理加群申请
        if "暗号正确" in (event.comment or ""):
            await bot.set_group_add_request(flag=event.flag, approve=True)
        else:
            await bot.set_group_add_request(flag=event.flag, approve=False, reason="入群暗号错误")

五、生命周期与元事件 (Meta Events) 源码详解 ​

元事件反映了 Bot 底层引擎与协议端的生命状态:

python
class LifecycleMetaEvent(MetaEvent):
    """生命周期事件:包含连接建立、心跳等"""

    meta_event_type: Literal["lifecycle"]
    sub_type: str                       # "connect"(连接建立) / "enable" / "disable"


class HeartbeatMetaEvent(MetaEvent):
    """心跳元事件:协议端定期发送的状态报告"""

    meta_event_type: Literal["heartbeat"]
    status: Status                      # 包含 online(bool) 与 good(bool)
    interval: int                       # 下次心跳毫秒间隔

实战:连接成功后向管理汇报启动状态 ​

python
@bot.on(LifecycleMetaEvent)
async def on_connect(event: LifecycleMetaEvent):
    if event.sub_type == "connect":
        print("🎉 成功连上 OneBot 协议端,机器人已准备就绪!")
        # 可以向主人的私聊推送一条唤醒通知
        # await bot.send_friend_message(master_id, "主人~ Yucca 已经成功苏醒啦!")