feat(trendradar): 新增空权限插件并接入webhook群权限校验

- 新建 trendradar_permission 空插件,仅注册 TRENDRADAR_WEBHOOK 功能用于后台群级开关

- webhook 发送前强制校验群权限,未开启群加入 blocked_groups 并拦截

- 更新对接文档,补充权限开关的启用步骤与返回字段说明
This commit is contained in:
liuwei
2026-04-21 16:37:23 +08:00
parent ace2af7dba
commit 055406d207
5 changed files with 148 additions and 3 deletions

View File

@@ -0,0 +1,85 @@
# -*- coding: utf-8 -*-
from typing import Any, Dict, List, Optional, Tuple
from loguru import logger
from base.plugin_common.message_plugin_interface import MessagePluginInterface
from base.plugin_common.plugin_interface import PluginStatus
class TrendRadarPermissionPlugin(MessagePluginInterface):
"""TrendRadar Webhook 权限占位插件。
设计说明:
1. 本插件不处理任何聊天消息,也不提供命令;
2. 唯一职责是注册 FeatureTRENDRADAR_WEBHOOK
3. 该 Feature 会出现在群插件权限配置页,便于按群启停 webhook 下发。
"""
FEATURE_KEY = "TRENDRADAR_WEBHOOK"
FEATURE_DESCRIPTION = "📡 TrendRadar Webhook推送 [群级开关]"
@property
def name(self) -> str:
return "TrendRadar权限开关"
@property
def version(self) -> str:
return "1.0.0"
@property
def description(self) -> str:
return "仅用于注册TrendRadar webhook群权限不处理消息。"
@property
def author(self) -> str:
return "ABOT Team"
@property
def command_prefix(self) -> Optional[str]:
return ""
@property
def commands(self) -> List[str]:
return []
@property
def feature_key(self) -> Optional[str]:
return self.FEATURE_KEY
@property
def feature_description(self) -> Optional[str]:
return self.FEATURE_DESCRIPTION
def __init__(self):
super().__init__()
self.feature = self.register_feature()
self.enable = True
def initialize(self, context: Dict[str, Any]) -> bool:
"""初始化插件。
这里只读取 enable 配置,默认启用。
"""
self.LOG = logger
cfg = self._config.get("TrendRadarPermission", {})
self.enable = bool(cfg.get("enable", True))
self.LOG.info(f"[{self.name}] 初始化完成enable={self.enable}, feature={self.feature_key}")
return True
def start(self) -> bool:
self.status = PluginStatus.RUNNING
return True
def stop(self) -> bool:
self.status = PluginStatus.STOPPED
return True
def can_process(self, message: Dict[str, Any]) -> bool:
"""该插件不处理任何消息。"""
return False
async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
"""空实现:永远不处理。"""
return False, None