refactor(chain): 处理链功能域 mixin 化,清理未使用导入并根治兼容层循环导入

- ChainBase 拆分为 RecognitionMixin/MessageProcessingMixin/NotificationMixin
- TransferChain 拆分为 7 个功能 mixin(_mixins.py),SubscribeChain 音乐订阅域拆出 _music.py
- 斜杠命令交互四件套收敛为 InteractionChainMixin 委托,会话管理器移至 application 层,chain 层不再 re-export
- 模块基础类收敛到 app/modules/_base(notification/mediaserver 语义重命名)
- 清理 app/chain/__init__.py 24 个未使用导入,修正 49 处测试 patch 目标到实际命名空间
- 兼容层 legacy 符号不再并入 __all__,根治 schemas 初始化反向拉起 application.transfer 的循环导入
- 修复 bangumi 集数为字符串时 set_bangumi_info 抛 TypeError
- 新增重复代码等架构门禁测试;capability 清单校验排除下划线内部目录
This commit is contained in:
jxxghp
2026-08-16 16:30:16 +08:00
parent 24671f8f18
commit 7e851dbfa7
102 changed files with 6041 additions and 5888 deletions
+4 -44
View File
@@ -9,7 +9,7 @@ from app.application.messaging.agent import (
resolve_config_principal_ids,
)
from app.runtime.log import logger
from app.modules import _ModuleBase, _MessageBase
from app.modules._base import _MessageChannelModuleBase
from app.modules.vocechat.vocechat import VoceChat
from app.schemas import MessageChannel, CommingMessage, Notification
from app.schemas.types import ModuleType
@@ -21,7 +21,9 @@ register_channel_admin_resolver(
)
class VoceChatModule(_ModuleBase, _MessageBase[VoceChat]):
class VoceChatModule(_MessageChannelModuleBase[VoceChat]):
# 管理员配置键,与渠道 resolver 保持一致
_admin_config_key = "VOCECHAT_ADMINS"
_IMAGE_SUFFIXES = (
".png",
".jpg",
@@ -83,51 +85,9 @@ class VoceChatModule(_ModuleBase, _MessageBase[VoceChat]):
def stop(self):
pass
def test(self) -> Optional[Tuple[bool, str]]:
"""
测试模块连接性
"""
if not self.get_instances():
return None
for name, client in self.get_instances().items():
state = client.get_state()
if not state:
return False, f"VoceChat {name} 未就绪"
return True, ""
def init_setting(self) -> Tuple[str, Union[str, bool]]:
pass
@staticmethod
def _get_admins(config: Optional[dict]) -> List[str]:
"""
解析 VoceChat 管理员配置,兼容逗号分隔和首尾空白。
"""
return [
admin.strip()
for admin in str((config or {}).get("VOCECHAT_ADMINS") or "").split(",")
if admin.strip()
]
@classmethod
def _should_reject_admin_command(
cls,
config: Optional[dict],
*user_ids: Optional[Union[str, int]],
) -> bool:
"""
判断 VoceChat 斜杠命令是否应因非管理员身份被拒绝。
"""
admins = cls._get_admins(config)
if not admins:
return False
candidates = [
str(user_id).strip()
for user_id in user_ids
if user_id is not None and str(user_id).strip()
]
return not any(candidate in admins for candidate in candidates)
@staticmethod
def _send_admin_denied(
client: Optional[VoceChat], userid: Optional[Union[str, int]]