92 lines
3.6 KiB
Python
92 lines
3.6 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
|
||
import astrbot.api.message_components as Comp
|
||
|
||
|
||
def _check_at_bot(message_obj) -> bool:
|
||
"""
|
||
检测群聊消息是否@了当前机器人。
|
||
依据: plugin.md 中 AstrBotMessage 及消息链定义:
|
||
- message_obj.message 为 BaseMessageComponent 列表
|
||
- At(qq=...) 为 @ 消息段，message_obj.self_id 为机器人的标识 ID
|
||
当消息链中存在 At 且其 qq 等于 self_id 时，判定为@了机器人。
|
||
忽略 AtAll(@全体)。
|
||
"""
|
||
try:
|
||
self_id = str(getattr(message_obj, "self_id", "")) or ""
|
||
segments = getattr(message_obj, "message", []) or []
|
||
except Exception:
|
||
return False
|
||
|
||
if not self_id or not isinstance(segments, list):
|
||
return False
|
||
|
||
for seg in segments:
|
||
# 适配对象消息段
|
||
try:
|
||
if isinstance(seg, Comp.At):
|
||
# 排除 @all
|
||
if isinstance(seg, Comp.AtAll):
|
||
continue
|
||
if str(seg.qq) == self_id:
|
||
return True
|
||
except Exception:
|
||
pass
|
||
|
||
# 适配字典形式的消息段（稳健性处理）
|
||
try:
|
||
if isinstance(seg, dict):
|
||
seg_type = str(seg.get("type", "")).lower()
|
||
data = seg.get("data", {}) or {}
|
||
if seg_type == "at":
|
||
qq = str(data.get("qq", ""))
|
||
if qq and qq.lower() != "all" and qq == self_id:
|
||
return True
|
||
except Exception:
|
||
pass
|
||
|
||
return False
|
||
|
||
|
||
|
||
|
||
@register("relay", "YourName", "消息中转插件", "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: 处理函数留空 |