66 lines
3.0 KiB
Python
66 lines
3.0 KiB
Python
from astrbot.api.event import AstrMessageEvent
|
||
from astrbot.api.event.filter import event_message_type, EventMessageType
|
||
from astrbot.api.star import Context, Star, register
|
||
from astrbot.api import logger
|
||
from astrbot.core.platform import AstrBotMessage
|
||
from astrbot.core.message.components import At
|
||
|
||
def check_at_bot(message_obj:AstrBotMessage) -> bool:
|
||
"""
|
||
检测群聊消息是否@了当前机器人。
|
||
依据: plugin.md 中 AstrBotMessage 及消息链定义:
|
||
- message_obj.message 为 BaseMessageComponent 列表
|
||
- At(qq=...) 为 @ 消息段，message_obj.self_id 为机器人的标识 ID
|
||
当消息链中存在 At 且其 qq 等于 self_id 时，判定为@了机器人。
|
||
忽略 AtAll(@全体)。
|
||
"""
|
||
self_id = message_obj.self_id
|
||
segments = message_obj.message
|
||
for segment in segments:
|
||
if isinstance(segment, At) and str(segment.qq) == self_id:
|
||
logger.info(f"[check_at_bot] 消息 {message_obj.message_id} @了机器人 {self_id}")
|
||
return True
|
||
logger.info(f"[check_at_bot] 消息 {message_obj.message_id} 未@机器人 {self_id}")
|
||
return False
|
||
|
||
|
||
|
||
|
||
@register("qqfight-bot", "hxuanyu", "消息中转插件", "1.0.0")
|
||
class RelayPlugin(Star):
|
||
def __init__(self, context: Context, config=None):
|
||
super().__init__(context)
|
||
self.config = config or {}
|
||
# 读取群聊白名单和私聊黑名单，全部转为字符串集合，避免类型不一致
|
||
self.group_whitelist = set(str(x) for x in self.config.get("group_whitelist", []))
|
||
self.private_blacklist = set(str(x) for x in self.config.get("private_blacklist", []))
|
||
logger.info(f"[RelayPlugin] 群聊白名单: {self.group_whitelist}")
|
||
logger.info(f"[RelayPlugin] 私聊黑名单: {self.private_blacklist}")
|
||
|
||
@event_message_type(EventMessageType.GROUP_MESSAGE)
|
||
async def on_group_message(self, event: AstrMessageEvent):
|
||
|
||
group_id = str(event.message_obj.group_id)
|
||
at_bot = check_at_bot(event.message_obj)
|
||
|
||
if group_id not in self.group_whitelist:
|
||
logger.info(f"[RelayPlugin] 群聊 {group_id} 不在白名单，忽略消息")
|
||
return
|
||
if not at_bot:
|
||
logger.info(f"[RelayPlugin] 群聊 {group_id} 消息未@机器人，忽略消息")
|
||
return
|
||
logger.info(f"[RelayPlugin] 群聊 {group_id} 消息链: {event.message_obj.message}")
|
||
# TODO: 处理函数留空
|
||
|
||
@event_message_type(EventMessageType.PRIVATE_MESSAGE)
|
||
async def on_private_message(self, event: AstrMessageEvent):
|
||
sender = event.message_obj.sender
|
||
if hasattr(sender, 'id'):
|
||
sender_id = str(sender.id)
|
||
else:
|
||
sender_id = str(sender)
|
||
if sender_id in set(str(x) for x in self.private_blacklist):
|
||
logger.info(f"[RelayPlugin] 私聊 {sender_id} 在黑名单，忽略消息")
|
||
return
|
||
logger.info(f"[RelayPlugin] 私聊 {sender_id} 消息链: {event.message_obj.message}")
|
||
# TODO: 处理函数留空 |