From a66cdb04e0af63bfbf1a670f626224249a003f55 Mon Sep 17 00:00:00 2001 From: InfinityPacer Date: Mon, 31 Aug 2026 07:20:42 +0800 Subject: [PATCH 1/2] fix(notification): stabilize channel identity --- app/api/endpoints/notification.py | 61 ++++++++++++++++++++- app/modules/__init__.py | 10 +++- app/modules/wechatclawbot/__init__.py | 18 ++++++- app/modules/wechatclawbot/wechatclawbot.py | 43 +++++++++++++-- app/schemas/system.py | 3 ++ app/schemas/types.py | 2 + tests/test_notification_config_identity.py | 63 ++++++++++++++++++++++ 7 files changed, 192 insertions(+), 8 deletions(-) create mode 100644 tests/test_notification_config_identity.py diff --git a/app/api/endpoints/notification.py b/app/api/endpoints/notification.py index 9ce32fae5..96941cbdb 100644 --- a/app/api/endpoints/notification.py +++ b/app/api/endpoints/notification.py @@ -1,16 +1,75 @@ from typing import Any, Dict -from fastapi import Depends +from fastapi import Body, Depends from app.api.dependencies.auth import get_current_active_superuser from app.api.response import ResponseAPIRouter +from app.application.configuration import get_configured_system_config from app.chain.notification import NotificationChain from app.schemas.common import ManageRequest as _SchemaManageRequest from app.schemas.response import Response as _SchemaResponse +from app.schemas.system import NotificationConf +from app.schemas.types import SystemConfigKey router = ResponseAPIRouter() +def _normalize_configs(value: list[NotificationConf]) -> list[dict[str, Any]]: + """规整通知配置身份和名称，拒绝会导致运行目录覆盖的输入。""" + normalized: list[dict[str, Any]] = [] + names: set[str] = set() + for config in value: + name = (config.name or "").strip() + if not name: + raise ValueError("通知渠道名称不能为空") + key = name.casefold() + if key in names: + raise ValueError(f"通知渠道名称重复：{name}") + names.add(key) + data = config.model_dump() + data["name"] = name + normalized.append(data) + return normalized + + +@router.post( + "/config", + summary="保存通知渠道并同步登录缓存", + response_model=_SchemaResponse[Dict[str, Any]], +) +async def save_config( + configs: list[NotificationConf] = Body(...), + _: object = Depends(get_current_active_superuser), +): + """一次保存通知配置，并由渠道模块处理改名与删除缓存。""" + try: + current = _normalize_configs(configs) + previous = get_configured_system_config().get(SystemConfigKey.Notifications) or [] + previous_models = [] + for item in previous: + if not isinstance(item, dict): + continue + legacy = dict(item) + if not legacy.get("id"): + legacy["id"] = f"legacy-{legacy.get('type') or 'notification'}-{legacy.get('name') or 'unnamed'}" + previous_models.append(NotificationConf(**legacy)) + previous_data = _normalize_configs(previous_models) + result = NotificationChain().manage_channel( + channel="WechatClawBot", + action="reconcile_config", + previous=previous_data, + current=current, + ) + if not result.get("success"): + return _SchemaResponse(success=False, message=result.get("message")) + saved = await get_configured_system_config().async_set(SystemConfigKey.Notifications, current) + if saved is False: + return _SchemaResponse(success=False, message="通知配置保存失败") + return _SchemaResponse(success=True, message="通知配置保存成功", data={"value": current}) + except ValueError as error: + return _SchemaResponse(success=False, message=str(error)) + + @router.post( "/manage", summary="通知渠道统一管理", diff --git a/app/modules/__init__.py b/app/modules/__init__.py index 53f63c008..ad49cd91c 100644 --- a/app/modules/__init__.py +++ b/app/modules/__init__.py @@ -1,6 +1,6 @@ import threading from abc import abstractmethod, ABCMeta -from typing import Generic, Tuple, Union, TypeVar, Type, Dict, Optional, Callable +from typing import Any, Generic, Tuple, Union, TypeVar, Type, Dict, Optional, Callable from pathlib import Path from app.runtime.extensions.service import ServiceConfigHelper @@ -148,11 +148,17 @@ class ServiceBase(Generic[TService, TConf], metaclass=ABCMeta): # 通过服务类型或工厂函数来创建实例 if isinstance(service_type, type): # 如果传入的是类类型，调用构造函数实例化 - self._instances[conf.name] = service_type(name=conf.name, **conf.config) + self._instances[conf.name] = service_type( + **self._service_kwargs(conf) + ) else: # 如果传入的是工厂函数，直接调用工厂函数 self._instances[conf.name] = service_type(conf) + def _service_kwargs(self, conf: TConf) -> Dict[str, Any]: + """构建类服务实例参数，允许具体模块传递稳定配置身份。""" + return {"name": conf.name, **conf.config} + def get_instances(self) -> Dict[str, TService]: """ 获取服务实例列表 diff --git a/app/modules/wechatclawbot/__init__.py b/app/modules/wechatclawbot/__init__.py index 0f4071501..e494e8f50 100644 --- a/app/modules/wechatclawbot/__init__.py +++ b/app/modules/wechatclawbot/__init__.py @@ -77,6 +77,10 @@ class WechatClawBotModule(_MessageChannelModuleBase[WechatClawBot]): """微信 ClawBot 的连接探测返回 (状态, 信息)。""" return client.test_connection() + def _service_kwargs(self, conf) -> Dict[str, Any]: + """将稳定通知身份传递给 ClawBot 客户端。""" + return {"name": conf.name, "identity": conf.id, **conf.config} + def init_setting(self) -> Tuple[str, Union[str, bool]]: """初始化模块设置。""" pass @@ -112,6 +116,11 @@ class WechatClawBotModule(_MessageChannelModuleBase[WechatClawBot]): ) return {"success": success, "message": message} + if action == NotificationAction.RECONCILE_CONFIG: + return WechatClawBot.reconcile_cached_states( + params.get("previous") or [], params.get("current") or [] + ) + client, errmsg = self._resolve_client(params) if not client: return {"success": False, "message": errmsg} @@ -149,6 +158,11 @@ class WechatClawBotModule(_MessageChannelModuleBase[WechatClawBot]): if candidate_names: for candidate in candidate_names: config = self.get_config(candidate) + if not config: + config = next( + (item for item in self.get_configs().values() if item.id == candidate), + None, + ) if not config: continue client = self.get_instance(config.name) @@ -170,10 +184,12 @@ class WechatClawBotModule(_MessageChannelModuleBase[WechatClawBot]): def _build_temp_client(self, params: Dict[str, Any]) -> Optional[Any]: """基于表单参数创建临时客户端，用于未保存配置时的扫码状态预览。""" source_name = str(params.get("source") or params.get("fallback_source") or "").strip() + fallback_name = str(params.get("fallback_source") or "").strip() if not source_name: return None return WechatClawBot( - name=source_name, + name=fallback_name or source_name, + identity=source_name, WECHATCLAWBOT_BASE_URL=params.get("WECHATCLAWBOT_BASE_URL"), WECHATCLAWBOT_DEFAULT_TARGET=params.get("WECHATCLAWBOT_DEFAULT_TARGET"), WECHATCLAWBOT_ADMINS=params.get("WECHATCLAWBOT_ADMINS"), diff --git a/app/modules/wechatclawbot/wechatclawbot.py b/app/modules/wechatclawbot/wechatclawbot.py index e4e81c8de..71c07a93f 100644 --- a/app/modules/wechatclawbot/wechatclawbot.py +++ b/app/modules/wechatclawbot/wechatclawbot.py @@ -1443,9 +1443,9 @@ class WechatClawBot: _poll_join_timeout_seconds = 5 @classmethod - def _build_cache_key(cls, config_name: str) -> str: - """根据配置名称构建缓存键。""" - safe_name = hashlib.md5(str(config_name or "wechatclawbot").encode("utf-8")).hexdigest()[:12] + def _build_cache_key(cls, identity: str) -> str: + """根据稳定渠道身份构建缓存键。""" + safe_name = hashlib.md5(str(identity or "wechatclawbot").encode("utf-8")).hexdigest()[:12] return f"__wechatclawbot_state_{safe_name}__" def __init__( @@ -1455,11 +1455,13 @@ class WechatClawBot: WECHATCLAWBOT_ADMINS: Optional[str] = None, WECHATCLAWBOT_POLL_TIMEOUT: Optional[int] = None, name: Optional[str] = None, + identity: Optional[str] = None, auto_start_polling: bool = True, **kwargs, ): """初始化微信 ClawBot 实例及相关参数。""" self._config_name = name or "wechatclawbot" + self._identity = str(identity or self._config_name) self._base_url = (WECHATCLAWBOT_BASE_URL or self._default_base_url).rstrip("/") self._default_target = (WECHATCLAWBOT_DEFAULT_TARGET or "").strip() or None self._auto_start_polling = bool(auto_start_polling) @@ -1472,7 +1474,8 @@ class WechatClawBot: self._poll_timeout = max(10, int(WECHATCLAWBOT_POLL_TIMEOUT or 25)) except Exception: self._poll_timeout = 25 - self._cache_key = self._build_cache_key(self._config_name) + self._cache_key = self._build_cache_key(self._identity) + self._legacy_cache_key = self._build_cache_key(self._config_name) self._filecache = FileCache() self._lock = threading.Lock() self._stop_event = threading.Event() @@ -1484,6 +1487,11 @@ class WechatClawBot: def _load_state(self) -> Dict[str, Any]: """从文件缓存加载登录状态。""" content = self._filecache.get(self._cache_key) + if not content and self._legacy_cache_key != self._cache_key: + content = self._filecache.get(self._legacy_cache_key) + if content: + self._filecache.set(self._cache_key, content) + self._filecache.delete(self._legacy_cache_key) if not content: return { "bot_token": None, @@ -1919,6 +1927,33 @@ class WechatClawBot: cache.delete(source_key) return True, f"已将微信 ClawBot 登录缓存从 {source_name} 迁移到 {target_name}" + @classmethod + def reconcile_cached_states(cls, previous: list[dict], current: list[dict]) -> Dict[str, Any]: + """按稳定身份迁移改名缓存，并清理已删除渠道的登录状态。""" + cache = FileCache() + old_by_id = {str(item.get("id")): item for item in previous if item.get("id")} + current_ids = {str(item.get("id")) for item in current if item.get("id")} + for item in current: + if item.get("type") != "wechatclawbot": + continue + identity = str(item.get("id") or item.get("name") or "").strip() + old = old_by_id.get(identity) + if old and old.get("name") and old.get("name") != item.get("name"): + legacy_key = cls._build_cache_key(str(old["name"])) + target_key = cls._build_cache_key(identity) + if not cache.exists(target_key) and cache.exists(legacy_key): + content = cache.get(legacy_key) + if content: + cache.set(target_key, content) + cache.delete(legacy_key) + for item in previous: + if item.get("type") != "wechatclawbot" or str(item.get("id")) in current_ids: + continue + for identity in (item.get("id"), item.get("name")): + if identity: + cache.delete(cls._build_cache_key(str(identity))) + return {"success": True, "message": "通知渠道缓存已同步"} + @staticmethod def _decode_ref_payload(ref: str, kind: str) -> Optional[Dict[str, Any]]: prefix = f"wxclaw://{kind}/" diff --git a/app/schemas/system.py b/app/schemas/system.py index e7f95b667..cab95223c 100644 --- a/app/schemas/system.py +++ b/app/schemas/system.py @@ -1,6 +1,7 @@ from dataclasses import dataclass from datetime import datetime as _DateTime from typing import Optional, Any, Literal +from uuid import uuid4 from pydantic import BaseModel, Field, field_validator @@ -89,6 +90,8 @@ class NotificationConf(BaseModel): 通知配置 """ + # 稳定渠道身份；名称变化时保持不变，用于运行态凭据和跨页面引用。 + id: str = Field(default_factory=lambda: str(uuid4())) # 名称 name: Optional[str] = None # 类型 telegram/wechat/feishu/vocechat/synologychat/slack/webpush/qqbot diff --git a/app/schemas/types.py b/app/schemas/types.py index a19897a45..c95a58317 100644 --- a/app/schemas/types.py +++ b/app/schemas/types.py @@ -515,6 +515,8 @@ class NotificationAction(str, Enum): TEST_CONNECTION = "test_connection" # 迁移渠道名变更前的登录缓存 MIGRATE_CACHE = "migrate_cache" + # 同步通知配置变更产生的缓存迁移和清理 + RECONCILE_CONFIG = "reconcile_config" class StorageAction(str, Enum): diff --git a/tests/test_notification_config_identity.py b/tests/test_notification_config_identity.py new file mode 100644 index 000000000..11cbc9e50 --- /dev/null +++ b/tests/test_notification_config_identity.py @@ -0,0 +1,63 @@ +"""通知渠道稳定身份与缓存同步契约。""" + +import pytest + +from app.api.endpoints.notification import _normalize_configs +from app.modules.wechatclawbot.wechatclawbot import WechatClawBot +from app.schemas.system import NotificationConf + + +def test_notification_config_normalizes_names_and_rejects_duplicates(): + configs = [NotificationConf(name=" Alpha ", type="telegram")] + assert _normalize_configs(configs)[0]["name"] == "Alpha" + + with pytest.raises(ValueError, match="重复"): + _normalize_configs( + [ + NotificationConf(name="Alpha", type="telegram"), + NotificationConf(name=" alpha ", type="wechat"), + ] + ) + + +def test_reconcile_cached_states_migrates_renamed_identity_and_cleans_removed(monkeypatch): + values = {} + + class FakeCache: + def exists(self, key): + return key in values + + def get(self, key): + return values.get(key) + + def set(self, key, value): + values[key] = value + + def delete(self, key): + values.pop(key, None) + + monkeypatch.setattr( + "app.modules.wechatclawbot.wechatclawbot.FileCache", FakeCache + ) + old_name = WechatClawBot._build_cache_key("旧名") + removed_id = WechatClawBot._build_cache_key("removed-id") + values[old_name] = b"state" + values[removed_id] = b"stale" + + result = WechatClawBot.reconcile_cached_states( + [ + {"id": "stable-id", "name": "旧名", "type": "wechatclawbot"}, + {"id": "removed-id", "name": "已删除", "type": "wechatclawbot"}, + ], + [{"id": "stable-id", "name": "新名", "type": "wechatclawbot"}], + ) + + assert result["success"] is True + assert values[WechatClawBot._build_cache_key("stable-id")] == b"state" + assert old_name not in values + assert removed_id not in values + + +@pytest.mark.parametrize("identity", ["stable-id", "display-name"]) +def test_cache_key_is_stable_for_same_identity(identity): + assert WechatClawBot._build_cache_key(identity) == WechatClawBot._build_cache_key(identity) From 655a4fb2046cd10c8c51e633b462dff8984ad82a Mon Sep 17 00:00:00 2001 From: InfinityPacer Date: Mon, 31 Aug 2026 07:33:40 +0800 Subject: [PATCH 2/2] fix(notification): keep config save independent from cache sync --- app/api/endpoints/notification.py | 24 ++++++--- tests/test_notification_config_identity.py | 59 +++++++++++++++++++++- 2 files changed, 74 insertions(+), 9 deletions(-) diff --git a/app/api/endpoints/notification.py b/app/api/endpoints/notification.py index 96941cbdb..32cf1345d 100644 --- a/app/api/endpoints/notification.py +++ b/app/api/endpoints/notification.py @@ -6,6 +6,7 @@ from app.api.dependencies.auth import get_current_active_superuser from app.api.response import ResponseAPIRouter from app.application.configuration import get_configured_system_config from app.chain.notification import NotificationChain +from app.modules.wechatclawbot.wechatclawbot import WechatClawBot from app.schemas.common import ManageRequest as _SchemaManageRequest from app.schemas.response import Response as _SchemaResponse from app.schemas.system import NotificationConf @@ -54,17 +55,24 @@ async def save_config( legacy["id"] = f"legacy-{legacy.get('type') or 'notification'}-{legacy.get('name') or 'unnamed'}" previous_models.append(NotificationConf(**legacy)) previous_data = _normalize_configs(previous_models) - result = NotificationChain().manage_channel( - channel="WechatClawBot", - action="reconcile_config", - previous=previous_data, - current=current, - ) - if not result.get("success"): - return _SchemaResponse(success=False, message=result.get("message")) saved = await get_configured_system_config().async_set(SystemConfigKey.Notifications, current) if saved is False: return _SchemaResponse(success=False, message="通知配置保存失败") + try: + # 配置写入成功后再同步 ClawBot 缓存，避免持久化失败留下半完成的登录状态。 + result = WechatClawBot.reconcile_cached_states(previous_data, current) + except Exception as error: + return _SchemaResponse( + success=False, + message=f"通知配置已保存，但缓存同步失败，请重试：{error}", + data={"value": current}, + ) + if not result.get("success"): + return _SchemaResponse( + success=False, + message=f"通知配置已保存，但缓存同步失败，请重试：{result.get('message')}", + data={"value": current}, + ) return _SchemaResponse(success=True, message="通知配置保存成功", data={"value": current}) except ValueError as error: return _SchemaResponse(success=False, message=str(error)) diff --git a/tests/test_notification_config_identity.py b/tests/test_notification_config_identity.py index 11cbc9e50..11f3649e5 100644 --- a/tests/test_notification_config_identity.py +++ b/tests/test_notification_config_identity.py @@ -2,7 +2,7 @@ import pytest -from app.api.endpoints.notification import _normalize_configs +from app.api.endpoints.notification import _normalize_configs, save_config from app.modules.wechatclawbot.wechatclawbot import WechatClawBot from app.schemas.system import NotificationConf @@ -61,3 +61,60 @@ def test_reconcile_cached_states_migrates_renamed_identity_and_cleans_removed(mo @pytest.mark.parametrize("identity", ["stable-id", "display-name"]) def test_cache_key_is_stable_for_same_identity(identity): assert WechatClawBot._build_cache_key(identity) == WechatClawBot._build_cache_key(identity) + + +@pytest.mark.asyncio +async def test_save_config_does_not_touch_cache_when_persistence_fails(monkeypatch): + calls = {"reconcile": 0} + + class FakeConfig: + def get(self, _key): + return [] + + async def async_set(self, _key, _value): + return False + + monkeypatch.setattr( + "app.api.endpoints.notification.get_configured_system_config", + lambda: FakeConfig(), + ) + + def fail_if_reconciled(_previous, _current): + calls["reconcile"] += 1 + return {"success": True} + + monkeypatch.setattr(WechatClawBot, "reconcile_cached_states", fail_if_reconciled) + response = await save_config([NotificationConf(name="Alpha", type="telegram")], None) + + assert response.success is False + assert response.message == "通知配置保存失败" + assert calls["reconcile"] == 0 + + +@pytest.mark.asyncio +async def test_save_config_saves_channels_without_loaded_clawbot_module(monkeypatch): + calls = [] + + class FakeConfig: + def get(self, _key): + return [] + + async def async_set(self, _key, value): + calls.append(value) + return True + + monkeypatch.setattr( + "app.api.endpoints.notification.get_configured_system_config", + lambda: FakeConfig(), + ) + monkeypatch.setattr( + WechatClawBot, + "reconcile_cached_states", + lambda _previous, _current: {"success": True}, + ) + + response = await save_config([NotificationConf(name="TG", type="telegram")], None) + + assert response.success is True + assert response.data["value"][0]["name"] == "TG" + assert len(calls) == 1