fix: complete the migration of the group_list configuration to use AstrBot Unified Message Origins (UMOs).

Changes Implemented:
Configuration: group_list now supports UMOs (e.g., qq_123456:GroupMessage:123456) while maintaining backward compatibility for simple group IDs.
Logic: ConfigManager and AutoScheduler were updated to handle UMOs for permission checking and group fetching.
Commands: The /分析设置 command now uses the full UMO when enabling/disabling analysis for a group.
This commit is contained in:
SXP-Simon
2026-02-08 20:19:07 +08:00
parent e24449cac1
commit c3cc2fd1f4
4 changed files with 95 additions and 25 deletions
+23 -9
View File
@@ -26,8 +26,11 @@ class ConfigManager:
"""获取群组列表(用于黑白名单)"""
return self.config.get("group_list", [])
def is_group_allowed(self, group_id: str) -> bool:
"""根据配置的白/黑名单判断是否允许在该群聊中使用"""
def is_group_allowed(self, group_id_or_umo: str) -> bool:
"""
根据配置的白/黑名单判断是否允许在该群聊中使用
支持传入 simple group_id 或 UMO (Unified Message Origin)
"""
mode = self.get_group_list_mode().lower()
if mode not in ("whitelist", "blacklist", "none"):
mode = "none"
@@ -37,19 +40,30 @@ class ConfigManager:
return True
glist = [str(g) for g in self.get_group_list()]
group_id_str = str(group_id)
target = str(group_id_or_umo)
# 解析目标 ID(如果是 UMO,提取最后一部分作为 ID)
# UMO 格式通常为: platform_id:GroupMessage:group_id
target_simple_id = target.split(":")[-1] if ":" in target else target
def _is_match(item: str, target: str, target_simple_id: str) -> bool:
# 1. 配置项是 UMO (包含 :) -> 必须精确匹配目标 UMO
if ":" in item:
return item == target
# 2. 配置项是 Simple ID (不含 :) -> 匹配目标的 Simple ID
# 这意味着 Simple ID 配置对所有平台生效 (向后兼容)
return item == target_simple_id
is_in_list = any(_is_match(item, target, target_simple_id) for item in glist)
if mode == "whitelist":
return group_id_str in glist if glist else False
return is_in_list
if mode == "blacklist":
return group_id_str not in glist if glist else True
return not is_in_list
return True
def get_max_concurrent_tasks(self) -> int:
"""获取自动分析最大并发数"""
return self.config.get("max_concurrent_tasks", 5)
def get_max_messages(self) -> int:
"""获取最大消息数量"""
return self.config.get("max_messages", 1000)
+33 -7
View File
@@ -87,7 +87,9 @@ class AutoScheduler:
logger.info(f"✅ 群 {group_id} 属于平台 {platform_id}")
return platform_id
else:
logger.debug(f"平台 {platform_id} 无法获取群 {group_id} 信息 (返回None)")
logger.debug(
f"平台 {platform_id} 无法获取群 {group_id} 信息 (返回None)"
)
continue
# 回退到原始逻辑 (Legacy)
@@ -193,10 +195,14 @@ class AutoScheduler:
# 始终获取所有群组并进行过滤
logger.info(f"自动分析使用 {group_list_mode} 模式,正在获取群列表...")
all_groups = await self._get_all_groups()
logger.info(f"共获取到 {len(all_groups)} 个群组: {all_groups}")
logger.info(f"共获取到 {len(all_groups)} 个群组")
enabled_groups = []
for group_id in all_groups:
if self.config_manager.is_group_allowed(group_id):
for platform_id, group_id in all_groups:
# 构造 UMO 进行检查
umo = f"{platform_id}:GroupMessage:{group_id}"
# 检查 UMO 是否允许 (ConfigManager 会自动处理 UMO 和 Simple ID 的匹配逻辑)
if self.config_manager.is_group_allowed(umo):
enabled_groups.append(group_id)
logger.info(
@@ -447,8 +453,12 @@ class AutoScheduler:
# 锁资源由 WeakValueDictionary 自动管理,无需手动清理
logger.info(f"{group_id} 自动分析完成")
async def _get_all_groups(self) -> list[str]:
"""获取所有bot实例所在的群列表"""
async def _get_all_groups(self) -> list[tuple[str, str]]:
"""
获取所有bot实例所在的群列表
Returns:
list[tuple[str, str]]: [(platform_id, group_id), ...]
"""
all_groups = set()
if (
@@ -475,6 +485,20 @@ class AutoScheduler:
):
call_action_func = bot_instance.api.call_action
# 特别处理 Discord 适配器 (如果有专门的方法)
if hasattr(bot_instance, "get_group_list"):
try:
result = await bot_instance.get_group_list()
if result:
for group_id in result:
all_groups.add((platform_id, str(group_id)))
logger.info(
f"平台 {platform_id} (Adapter) 成功获取 {len(result)} 个群组"
)
continue
except Exception as e:
logger.debug(f"平台 {platform_id} get_group_list 失败: {e}")
if call_action_func:
# 尝试 OneBot v11 get_group_list
try:
@@ -495,7 +519,9 @@ class AutoScheduler:
if isinstance(result, list):
for group in result:
if isinstance(group, dict) and "group_id" in group:
all_groups.add(str(group["group_id"]))
all_groups.add(
(platform_id, str(group["group_id"]))
)
logger.info(
f"平台 {platform_id} 成功获取 {len(result)} 个群组"
)