Merge pull request #6510 from InfinityPacer/codex/fix/notification-channel-identity-v3

This commit is contained in:
jxxghp
2026-08-31 07:41:14 +08:00
committed by GitHub
7 changed files with 257 additions and 8 deletions

View File

@@ -1,16 +1,83 @@
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.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
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)
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))
@router.post(
"/manage",
summary="通知渠道统一管理",

View File

@@ -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]:
"""
获取服务实例列表

View File

@@ -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"),

View File

@@ -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}/"

View File

@@ -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

View File

@@ -515,6 +515,8 @@ class NotificationAction(str, Enum):
TEST_CONNECTION = "test_connection"
# 迁移渠道名变更前的登录缓存
MIGRATE_CACHE = "migrate_cache"
# 同步通知配置变更产生的缓存迁移和清理
RECONCILE_CONFIG = "reconcile_config"
class StorageAction(str, Enum):

View File

@@ -0,0 +1,120 @@
"""通知渠道稳定身份与缓存同步契约。"""
import pytest
from app.api.endpoints.notification import _normalize_configs, save_config
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)
@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