事件系统与模型源码解析 (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🌟 核心高频属性与方法:
event.pure_text:你最经常用的属性!自动将消息里的纯文字部分拼接并去除多余空白,方便你做if "签到" in event.pure_text的逻辑判断。event.reply:当用户“引用回复”某条消息时,这个属性会有值。你可以通过event.reply.message_id获取被引用的消息 ID,通过event.reply.sender.user_id知道他回复的是谁!event.get("image"):如果消息里发了图片,调用imgs = event.get("image")就能直接拿到包含图片 URL 和文件名的字典列表!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 已经成功苏醒啦!")