refactor(wechatclawbot): 渠道管理改为模块方法契约,endpoint 不再接触客户端对象

- 收消息方向保持 Telegram 模式:模块自建连接轮询网关,转发本地消息 API
- 管理方向(状态/二维码/退出/连通性)封装为 wechatclawbot_status 等模块契约方法,
  实例解析与临时客户端回退封闭在模块内部,统一返回 {success, message} dict
- 新建 NotificationChain 薄分发链,MessageChain 移除全部 clawbot 方法
- notification endpoint 改为纯 chain 调用,不再获取任何客户端对象
This commit is contained in:
jxxghp
2026-08-16 05:45:58 +08:00
parent 6504cb36f5
commit a66cbe6192
4 changed files with 254 additions and 90 deletions
+1 -55
View File
@@ -1850,58 +1850,4 @@ class MessageChain(ChainBase):
return base64.b64decode(payload)
except Exception as e:
logger.error(e)
return None
def get_wechatclawbot_client(
self,
source: Optional[str] = None,
fallback_source: Optional[str] = None,
WECHATCLAWBOT_BASE_URL: Optional[str] = None,
WECHATCLAWBOT_DEFAULT_TARGET: Optional[str] = None,
WECHATCLAWBOT_ADMINS: Optional[str] = None,
WECHATCLAWBOT_POLL_TIMEOUT: Optional[int] = None,
allow_temporary: bool = False,
):
"""获取已加载的微信 ClawBot 客户端,必要时退回到临时客户端。"""
source_name = str(source or "").strip() or None
client = self.run_module(
"wechatclawbot_client",
source=source,
fallback_source=fallback_source,
)
if client:
return client, None
if allow_temporary:
temp_client = self.run_module(
"wechatclawbot_temp_client",
source=source_name or str(fallback_source or "").strip() or None,
WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL,
WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET,
WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS,
WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT,
)
if temp_client:
return temp_client, None
if source_name:
return None, f"未找到名为 {source_name} 的微信 ClawBot 通知配置"
return None, "微信 ClawBot 通知未启用或配置尚未保存,请先保存并启用当前渠道"
def migrate_wechatclawbot_cache(
self,
old_name: str,
new_name: str,
cleanup_old: bool = False,
overwrite: bool = False,
):
"""在通知名称变更时迁移对应的微信 ClawBot 登录缓存。"""
return self.run_module(
"wechatclawbot_migrate_cache",
old_name=old_name,
new_name=new_name,
cleanup_old=cleanup_old,
overwrite=overwrite,
)
return None
+114
View File
@@ -0,0 +1,114 @@
from typing import Any, Dict, Tuple
from app.chain import ChainBase
class NotificationChain(ChainBase):
"""
通知渠道管理链,仅做模块方法名契约的薄分发,渠道连接与能力全部封闭在模块内部
"""
def get_wechatclawbot_status(
self,
source=None,
fallback_source=None,
WECHATCLAWBOT_BASE_URL=None,
WECHATCLAWBOT_DEFAULT_TARGET=None,
WECHATCLAWBOT_ADMINS=None,
WECHATCLAWBOT_POLL_TIMEOUT=None,
refresh_remote: bool = True,
auto_generate_qrcode: bool = True,
) -> Dict[str, Any]:
"""查询微信 ClawBot 登录状态与二维码。"""
result = self.run_module(
"wechatclawbot_status",
source=source,
fallback_source=fallback_source,
WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL,
WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET,
WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS,
WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT,
refresh_remote=refresh_remote,
auto_generate_qrcode=auto_generate_qrcode,
)
return result or {"success": False, "message": "微信 ClawBot 通知未启用或配置尚未保存"}
def refresh_wechatclawbot_qrcode(
self,
source=None,
fallback_source=None,
WECHATCLAWBOT_BASE_URL=None,
WECHATCLAWBOT_DEFAULT_TARGET=None,
WECHATCLAWBOT_ADMINS=None,
WECHATCLAWBOT_POLL_TIMEOUT=None,
) -> Dict[str, Any]:
"""刷新微信 ClawBot 登录二维码。"""
result = self.run_module(
"wechatclawbot_refresh_qrcode",
source=source,
fallback_source=fallback_source,
WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL,
WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET,
WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS,
WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT,
)
return result or {"success": False, "message": "微信 ClawBot 通知未启用或配置尚未保存"}
def logout_wechatclawbot(
self,
source=None,
fallback_source=None,
WECHATCLAWBOT_BASE_URL=None,
WECHATCLAWBOT_DEFAULT_TARGET=None,
WECHATCLAWBOT_ADMINS=None,
WECHATCLAWBOT_POLL_TIMEOUT=None,
) -> Dict[str, Any]:
"""退出微信 ClawBot 登录。"""
result = self.run_module(
"wechatclawbot_logout",
source=source,
fallback_source=fallback_source,
WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL,
WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET,
WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS,
WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT,
)
return result or {"success": False, "message": "微信 ClawBot 通知未启用或配置尚未保存"}
def test_wechatclawbot_connection(
self,
source=None,
fallback_source=None,
WECHATCLAWBOT_BASE_URL=None,
WECHATCLAWBOT_DEFAULT_TARGET=None,
WECHATCLAWBOT_ADMINS=None,
WECHATCLAWBOT_POLL_TIMEOUT=None,
) -> Dict[str, Any]:
"""测试微信 ClawBot 当前登录态是否可用。"""
result = self.run_module(
"wechatclawbot_test_connection",
source=source,
fallback_source=fallback_source,
WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL,
WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET,
WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS,
WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT,
)
return result or {"success": False, "message": "微信 ClawBot 通知未启用或配置尚未保存"}
def migrate_wechatclawbot_cache(
self,
old_name: str,
new_name: str,
cleanup_old: bool = False,
overwrite: bool = False,
) -> Tuple[bool, str]:
"""在通知名称变更时迁移对应的微信 ClawBot 登录缓存。"""
result = self.run_module(
"wechatclawbot_migrate_cache",
old_name=old_name,
new_name=new_name,
cleanup_old=cleanup_old,
overwrite=overwrite,
)
return result or (False, "微信 ClawBot 通知未启用")