From 69f4319023d824a3c1ea61541d640b66237153d4 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Wed, 13 May 2026 16:15:22 +0800 Subject: [PATCH] =?UTF-8?q?feat(synologychat):=20=E5=AE=8C=E6=88=90Synolog?= =?UTF-8?q?y=20Chat=E9=80=82=E9=85=8D=E5=99=A8=E7=9A=84=E5=A4=9A=E7=BB=B4?= =?UTF-8?q?=E5=BA=A6=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 此提交对Synology Chat适配器进行了全面改进： 1. 新增消息去重、分布式轮询租约、bot名称配置等功能 2. 优化URL提取逻辑，自动清理尾部标点符号 3. 重构发送逻辑，提取通用重试工具函数并优化SID缓存 4. 完善文档提示与配置项，新增轮询租约类型支持 5. 修复认证API路径硬编码问题，调整交互组件提示文案 6. 增加Webhook模式下的DSM客户端兜底初始化 7. 优化导入顺序与代码结构，清理冗余空行 --- .../adapters/synologychat/__init__.py | 13 +- .../channels/adapters/synologychat/adapter.py | 95 ++++++---- .../channels/adapters/synologychat/auth.py | 10 +- .../adapters/synologychat/config_schema.py | 11 ++ .../channels/adapters/synologychat/lease.py | 13 +- .../channels/adapters/synologychat/monitor.py | 5 + .../adapters/synologychat/normalize.py | 4 +- .../channels/adapters/synologychat/prompt.py | 4 +- .../channels/adapters/synologychat/send.py | 171 +++++++++--------- .../adapters/synologychat/setup_wizard.py | 9 +- .../channels/adapters/synologychat/target.py | 1 - .../adapters/synologychat/webhook_send.py | 39 ++-- 12 files changed, 224 insertions(+), 151 deletions(-) diff --git a/backend/package/yuxi/channels/adapters/synologychat/__init__.py b/backend/package/yuxi/channels/adapters/synologychat/__init__.py index fb0800e5..835a3436 100644 --- a/backend/package/yuxi/channels/adapters/synologychat/__init__.py +++ b/backend/package/yuxi/channels/adapters/synologychat/__init__.py @@ -1,10 +1,11 @@ from yuxi.channels.adapters.synologychat.adapter import SynologyChatAdapter -from yuxi.channels.adapters.synologychat.webhook_handler import ( - clear_invalid_token_limiter_for_test, - InvalidTokenRateLimiter, -) -from yuxi.channels.adapters.synologychat.setup_wizard import patch_config +from yuxi.channels.adapters.synologychat.dedup import MessageDeduplicator, should_process_event from yuxi.channels.adapters.synologychat.security import SynologyChatSecurityPolicy +from yuxi.channels.adapters.synologychat.setup_wizard import patch_config +from yuxi.channels.adapters.synologychat.webhook_handler import ( + InvalidTokenRateLimiter, + clear_invalid_token_limiter_for_test, +) __all__ = [ "SynologyChatAdapter", @@ -12,4 +13,6 @@ __all__ = [ "InvalidTokenRateLimiter", "clear_invalid_token_limiter_for_test", "patch_config", + "MessageDeduplicator", + "should_process_event", ] diff --git a/backend/package/yuxi/channels/adapters/synologychat/adapter.py b/backend/package/yuxi/channels/adapters/synologychat/adapter.py index d79b7de1..c1c5a828 100644 --- a/backend/package/yuxi/channels/adapters/synologychat/adapter.py +++ b/backend/package/yuxi/channels/adapters/synologychat/adapter.py @@ -20,15 +20,36 @@ from typing import Any import httpx +from yuxi.channels.adapters.synologychat.accounts import DEFAULT_ACCOUNT_ID, list_account_ids, resolve_account +from yuxi.channels.adapters.synologychat.approval import ApprovalManager +from yuxi.channels.adapters.synologychat.auth import apply_env_defaults +from yuxi.channels.adapters.synologychat.client import DSMClient, DSMClientError +from yuxi.channels.adapters.synologychat.dedup import MessageDeduplicator +from yuxi.channels.adapters.synologychat.directory import list_groups as _directory_list_groups +from yuxi.channels.adapters.synologychat.directory import list_peers as _directory_list_peers +from yuxi.channels.adapters.synologychat.lease import PollingLease, with_polling_lease +from yuxi.channels.adapters.synologychat.monitor import polling_loop +from yuxi.channels.adapters.synologychat.normalize import normalize_event +from yuxi.channels.adapters.synologychat.probe import probe_dsm +from yuxi.channels.adapters.synologychat.prompt import get_format_hints +from yuxi.channels.adapters.synologychat.security import SynologyChatSecurityPolicy +from yuxi.channels.adapters.synologychat.send import send_media, send_stream_block, send_with_retry +from yuxi.channels.adapters.synologychat.session import resolve_session_route +from yuxi.channels.adapters.synologychat.webhook_handler import ( + extract_token_from_request, + handle_webhook_event, + verify_token, +) +from yuxi.channels.adapters.synologychat.webhook_send import send_via_incoming_webhook from yuxi.channels.base import BaseChannelAdapter from yuxi.channels.capabilities import ChannelCapabilities -from yuxi.channels.meta import ChannelMeta -from yuxi.channels.infra.circuit_breaker import CircuitBreaker, CircuitBreakerOpenError from yuxi.channels.exceptions import ( ChannelAuthenticationError, ChannelConnectionError, ChannelNotConnectedError, ) +from yuxi.channels.infra.circuit_breaker import CircuitBreaker, CircuitBreakerOpenError +from yuxi.channels.meta import ChannelMeta from yuxi.channels.models import ( ChannelMessage, ChannelResponse, @@ -38,27 +59,6 @@ from yuxi.channels.models import ( HealthStatus, ) from yuxi.channels.registry import register_builtin_adapter -from yuxi.channels.adapters.synologychat.client import DSMClient, DSMClientError -from yuxi.channels.adapters.synologychat.monitor import polling_loop -from yuxi.channels.adapters.synologychat.normalize import normalize_event -from yuxi.channels.adapters.synologychat.probe import probe_dsm -from yuxi.channels.adapters.synologychat.security import SynologyChatSecurityPolicy -from yuxi.channels.adapters.synologychat.send import send_media, send_stream_block, send_with_retry -from yuxi.channels.adapters.synologychat.session import resolve_session_route -from yuxi.channels.adapters.synologychat.auth import apply_env_defaults -from yuxi.channels.adapters.synologychat.accounts import list_account_ids, resolve_account, DEFAULT_ACCOUNT_ID -from yuxi.channels.adapters.synologychat.lease import PollingLease, with_polling_lease -from yuxi.channels.adapters.synologychat.approval import ApprovalManager -from yuxi.channels.adapters.synologychat.dedup import MessageDeduplicator -from yuxi.channels.adapters.synologychat.directory import list_peers as _directory_list_peers -from yuxi.channels.adapters.synologychat.directory import list_groups as _directory_list_groups -from yuxi.channels.adapters.synologychat.prompt import get_format_hints -from yuxi.channels.adapters.synologychat.webhook_handler import ( - extract_token_from_request, - handle_webhook_event, - verify_token, -) -from yuxi.channels.adapters.synologychat.webhook_send import send_via_incoming_webhook from yuxi.utils.datetime_utils import utc_now_naive from yuxi.utils.logging_config import logger @@ -139,6 +139,7 @@ class SynologyChatAdapter(BaseChannelAdapter): self._last_source_ip: str = "unknown" self._send_lock = asyncio.Lock() self._last_send_ts: dict[str, float] = {} + self._stream_chunk_state: dict[str, dict[str, int]] = {} async def connect(self) -> None: if self._status == ChannelStatus.CONNECTED: @@ -201,6 +202,7 @@ class SynologyChatAdapter(BaseChannelAdapter): self._circuit_breaker, security_filter=self._security.check, account_id=self._account_id, + dedup=self._dedup, ) lease_type = self.config.get("polling_lease_type", "memory") @@ -267,6 +269,23 @@ class SynologyChatAdapter(BaseChannelAdapter): f"webhook_path_source={self._webhook_path_source}). " f"Expecting webhook events at path: {webhook_path}" ) + + dsm_url = self.config.get("dsm_url", "") + if dsm_url: + try: + self._http_client = httpx.AsyncClient( + base_url=dsm_url, + timeout=httpx.Timeout(10.0, read=30.0), + verify=self.config.get("verify_ssl", True), + ) + self._api_info = await probe_dsm(self._http_client, dsm_url) + if self._api_info: + self._dsm_client = DSMClient(self._http_client, dsm_url, self.config, self._api_info) + await self._dsm_client.login() + logger.info("[SynologyChat] DSM client initialized as fallback in webhook mode") + except Exception as e: + logger.warning(f"[SynologyChat] DSM fallback init failed (non-fatal): {e}") + except ChannelAuthenticationError: self._status = ChannelStatus.ERROR raise @@ -282,8 +301,15 @@ class SynologyChatAdapter(BaseChannelAdapter): logger.info(f"[SynologyChat] Stopping channel '{self.config.get('name', self.channel_id)}'") try: - if self._poll_task and not self._poll_task.done(): - self._poll_task.cancel() + if self._poll_task: + try: + done = self._poll_task.done() + if not isinstance(done, bool): + done = False + except Exception: + done = False + if not done: + self._poll_task.cancel() try: await self._poll_task except asyncio.CancelledError: @@ -367,6 +393,12 @@ class SynologyChatAdapter(BaseChannelAdapter): if not self._dsm_client: return DeliveryResult(success=False, error="DSM client not initialized") + stream_key = f"{chat_id}:{msg_id}" + if stream_key not in self._stream_chunk_state: + self._stream_chunk_state[stream_key] = {"index": 0, "total": 0} + state = self._stream_chunk_state[stream_key] + state["index"] += 1 + text = _format_markdown_to_chunk(chunk, finished) return await send_stream_block( @@ -375,6 +407,8 @@ class SynologyChatAdapter(BaseChannelAdapter): text, self.config, self._circuit_breaker, + chunk_index=state["index"], + chunk_total=state["total"] or state["index"], send_lock=self._send_lock, last_send_ts=self._last_send_ts, account_id=self._account_id, @@ -439,9 +473,8 @@ class SynologyChatAdapter(BaseChannelAdapter): # ---- Normalize ---- def normalize_inbound(self, raw: dict) -> ChannelMessage: - return normalize_event( - raw, self.channel_id, self.channel_type, self.config.get("username"), self.config.get("trigger_word") - ) + bot_name = self.config.get("bot_name") or self.config.get("name", "ForcePilot") + return normalize_event(raw, self.channel_id, self.channel_type, bot_name, self.config.get("trigger_word")) def _normalize(self, raw: dict) -> ChannelMessage: return self.normalize_inbound(raw) @@ -776,15 +809,13 @@ class SynologyChatAdapter(BaseChannelAdapter): webhook_url = self.config.get("incoming_webhook_url", "") if webhook_url: try: - result = await send_via_incoming_webhook( - webhook_url, error_text, client=self._http_client - ) + result = await send_via_incoming_webhook(webhook_url, error_text, client=self._http_client) if not result.success: logger.warning(f"[SynologyChat] Webhook error reply failed: {result.error}") except Exception as e: logger.warning(f"[SynologyChat] Webhook error reply exception: {e}") else: - logger.warning(f"[SynologyChat] Cannot send error reply: no DSM client and no webhook URL") + logger.warning("[SynologyChat] Cannot send error reply: no DSM client and no webhook URL") def _format_markdown_to_chunk(text: str, finished: bool) -> str: diff --git a/backend/package/yuxi/channels/adapters/synologychat/auth.py b/backend/package/yuxi/channels/adapters/synologychat/auth.py index a076b1b0..1d2dcb72 100644 --- a/backend/package/yuxi/channels/adapters/synologychat/auth.py +++ b/backend/package/yuxi/channels/adapters/synologychat/auth.py @@ -29,6 +29,10 @@ from yuxi.channels.adapters.synologychat.probe import resolve_api_version from yuxi.channels.exceptions import ChannelAuthenticationError from yuxi.utils.logging_config import logger +_DSM_AUTH_PATH = "/webapi/auth.cgi" +_DSM_QUERY_PATH = "/webapi/query.cgi" +_DSM_ENTRY_PATH = "/webapi/entry.cgi" + _ENV_MAP = { "DSM_URL": "dsm_url", @@ -129,7 +133,7 @@ async def dsm_login( try: response = await http_client.get( - f"{base_url}/webapi/auth.cgi", + f"{base_url}{_DSM_AUTH_PATH}", params=params, ) response.raise_for_status() @@ -159,7 +163,7 @@ async def dsm_logout( try: response = await http_client.get( - f"{base_url}/webapi/auth.cgi", + f"{base_url}{_DSM_AUTH_PATH}", params={ "api": "SYNO.API.Auth", "version": str(auth_version), @@ -189,7 +193,7 @@ async def ensure_valid_sid( for attempt in range(max_retries): try: response = await http_client.get( - f"{base_url}/webapi/query.cgi", + f"{base_url}{_DSM_QUERY_PATH}", params={ "api": "SYNO.Chat.External", "version": str(chat_version), diff --git a/backend/package/yuxi/channels/adapters/synologychat/config_schema.py b/backend/package/yuxi/channels/adapters/synologychat/config_schema.py index 5ae5c6db..447e22c0 100644 --- a/backend/package/yuxi/channels/adapters/synologychat/config_schema.py +++ b/backend/package/yuxi/channels/adapters/synologychat/config_schema.py @@ -72,6 +72,11 @@ class SynologyChatConfig(BaseModel): description="Channel instance name", json_schema_extra={"label": "Channel Name", "category": "general"}, ) + bot_name: str = Field( + default="", + description="Bot display name in Synology Chat (used for @mention detection)", + json_schema_extra={"label": "Bot Name", "category": "general"}, + ) default_agent_id: str = Field( default="default", description="Default agent ID for routing incoming messages", @@ -101,6 +106,12 @@ class SynologyChatConfig(BaseModel): description="Maximum backoff seconds when polling encounters errors", json_schema_extra={"label": "Max Polling Backoff (s)", "category": "polling"}, ) + polling_lease_type: str = Field( + default="memory", + pattern=r"^(memory|redis)$", + description="Polling lease type for distributed coordination: 'memory' or 'redis'", + json_schema_extra={"label": "Polling Lease Type", "category": "polling"}, + ) text_chunk_limit: int = Field( default=4000, ge=100, diff --git a/backend/package/yuxi/channels/adapters/synologychat/lease.py b/backend/package/yuxi/channels/adapters/synologychat/lease.py index 7dc727cf..19621dc6 100644 --- a/backend/package/yuxi/channels/adapters/synologychat/lease.py +++ b/backend/package/yuxi/channels/adapters/synologychat/lease.py @@ -57,11 +57,13 @@ class RedisPollingLease: self._lease_key = lease_key self._ttl = ttl_seconds self._instance_id = f"{id(self)}:{time.time()}" + self._last_known_held: bool = False async def acquire(self) -> bool: try: result = await self._redis.set(self._lease_key, self._instance_id, nx=True, ex=self._ttl) - return bool(result) + self._last_known_held = bool(result) + return self._last_known_held except Exception as e: logger.warning(f"[SynologyChat] Redis lease acquire failed: {e}") return False @@ -76,6 +78,7 @@ class RedisPollingLease: end """ await self._redis.eval(script, 1, self._lease_key, self._instance_id) + self._last_known_held = False except Exception as e: logger.warning(f"[SynologyChat] Redis lease release failed: {e}") @@ -89,19 +92,21 @@ class RedisPollingLease: end """ result = await self._redis.eval(script, 1, self._lease_key, self._instance_id, self._ttl) - return bool(result) + self._last_known_held = bool(result) + return self._last_known_held except Exception as e: logger.warning(f"[SynologyChat] Redis lease renew failed: {e}") return False @property def is_held(self) -> bool: - return True + return self._last_known_held async def check_held(self) -> bool: try: current = await self._redis.get(self._lease_key) - return current is not None and current.decode() == self._instance_id + self._last_known_held = current is not None and current.decode() == self._instance_id + return self._last_known_held except Exception: return False diff --git a/backend/package/yuxi/channels/adapters/synologychat/monitor.py b/backend/package/yuxi/channels/adapters/synologychat/monitor.py index 58cbc2b1..2bbffdf4 100644 --- a/backend/package/yuxi/channels/adapters/synologychat/monitor.py +++ b/backend/package/yuxi/channels/adapters/synologychat/monitor.py @@ -13,6 +13,7 @@ from collections.abc import Awaitable, Callable from typing import Any from yuxi.channels.adapters.synologychat.client import DSMClient +from yuxi.channels.adapters.synologychat.dedup import MessageDeduplicator from yuxi.channels.infra.circuit_breaker import CircuitBreaker, CircuitBreakerOpenError from yuxi.channels.models import ChannelMessage, ChannelStatus from yuxi.utils.logging_config import logger @@ -28,6 +29,7 @@ async def polling_loop( security_filter: Callable[[ChannelMessage], bool] | None = None, max_inflight: int = 10, account_id: str = "", + dedup: MessageDeduplicator | None = None, ) -> None: poll_interval = config.get("polling_interval_seconds", 3) max_backoff = config.get("polling_max_backoff_seconds", 60) @@ -69,6 +71,9 @@ async def polling_loop( events = data.get("events", []) for event in events: try: + message_id = str(event.get("message_id", "")) + if dedup and message_id and dedup.check_and_mark(message_id): + continue message = normalize_fn(event) if security_filter and not security_filter(message): continue diff --git a/backend/package/yuxi/channels/adapters/synologychat/normalize.py b/backend/package/yuxi/channels/adapters/synologychat/normalize.py index 5470d974..f3ce068b 100644 --- a/backend/package/yuxi/channels/adapters/synologychat/normalize.py +++ b/backend/package/yuxi/channels/adapters/synologychat/normalize.py @@ -25,6 +25,7 @@ from yuxi.utils.datetime_utils import utc_now_naive _MENTION_RE = re.compile(r"@(\w[\w.-]{0,31})") _URL_RE = re.compile(r"https?://[^\s]+") +_URL_CLEAN_RE = re.compile(r'[.,;:!?。，；：！？)"\'】」』]+$') _FILE_IMAGE_MIMES = frozenset( { @@ -87,7 +88,8 @@ def extract_mentions(text: str, bot_name: str | None = None) -> MentionsInfo: def extract_urls(text: str) -> list[str]: - return _URL_RE.findall(text) + urls = _URL_RE.findall(text) + return [_URL_CLEAN_RE.sub("", url) for url in urls] _INJECTION_PATTERNS = [ diff --git a/backend/package/yuxi/channels/adapters/synologychat/prompt.py b/backend/package/yuxi/channels/adapters/synologychat/prompt.py index d8cc8d10..e2bd16f6 100644 --- a/backend/package/yuxi/channels/adapters/synologychat/prompt.py +++ b/backend/package/yuxi/channels/adapters/synologychat/prompt.py @@ -18,7 +18,7 @@ def get_format_hints() -> str: "- 链接：使用 格式，例如 \n" "- 不支持 Markdown 表格、代码块、图片嵌入等高级语法\n" "- 单条消息限制 4000 字符，长回复请分段发送\n" - "- 不支持按钮、卡片等交互组件\n" + "- 按钮、卡片等交互组件暂未启用\n" ) @@ -33,5 +33,5 @@ def get_format_hints_english() -> str: "- Links: use format, e.g. \n" "- Markdown tables, code blocks, inline images are NOT supported\n" "- Single message limit: 4000 characters, split long replies\n" - "- Interactive components (buttons, cards) are NOT supported\n" + "- Interactive components (buttons, cards) are not yet enabled\n" ) diff --git a/backend/package/yuxi/channels/adapters/synologychat/send.py b/backend/package/yuxi/channels/adapters/synologychat/send.py index 3870c763..162d9438 100644 --- a/backend/package/yuxi/channels/adapters/synologychat/send.py +++ b/backend/package/yuxi/channels/adapters/synologychat/send.py @@ -12,6 +12,7 @@ import ipaddress import random import socket import time +from collections.abc import Awaitable, Callable from typing import Any from urllib.parse import urlparse @@ -26,6 +27,18 @@ from yuxi.channels.models import ChannelResponse, DeliveryResult from yuxi.utils.logging_config import logger _DEFAULT_MIN_SEND_INTERVAL_MS = 500 +_SID_CACHE_TTL = 300 + +_last_sid_refresh: dict[str, float] = {} + + +async def _ensure_fresh_sid(client: DSMClient, account_id: str = "default") -> None: + now = time.monotonic() + last = _last_sid_refresh.get(account_id, 0) + if now - last > _SID_CACHE_TTL: + await client.refresh_sid() + _last_sid_refresh[account_id] = now + _PRIVATE_IP_RANGES = [ ipaddress.IPv4Network("10.0.0.0/8"), @@ -134,6 +147,49 @@ def _build_text(response: ChannelResponse, config: dict[str, Any]) -> str: return text[:text_chunk_limit] +async def _execute_with_retry( + send_fn: Callable[[], Awaitable[dict[str, Any]]], + max_attempts: int, + min_delay_ms: int, + max_delay_ms: int, + jitter: float, + circuit_breaker: CircuitBreaker | None = None, +) -> DeliveryResult: + last_error = None + for attempt in range(max_attempts): + try: + if circuit_breaker: + result = await circuit_breaker.call(send_fn) + else: + result = await send_fn() + + if result.get("success"): + message_id = result.get("data", {}).get("message_id") + return DeliveryResult(success=True, message_id=message_id) + + err_code = result.get("error", {}).get("code", 0) + last_error = f"DSM error code: {err_code}" + if err_code in (105, 101): + return DeliveryResult(success=False, error=last_error) + + except CircuitBreakerOpenError: + return DeliveryResult(success=False, error="Circuit breaker open") + except DSMNonRetryableError as e: + logger.error(f"[SynologyChat] Non-retryable send error: {e}") + return DeliveryResult(success=False, error=str(e)) + except DSMClientError as e: + last_error = str(e) + except Exception as e: + last_error = str(e) + + if attempt < max_attempts - 1: + delay = _calc_retry_delay(attempt, min_delay_ms, max_delay_ms, jitter) + logger.warning(f"Send retry {attempt + 1}/{max_attempts} after {delay:.1f}s: {last_error}") + await asyncio.sleep(delay) + + return DeliveryResult(success=False, error=last_error or "Send failed after retries") + + async def send_with_retry( client: DSMClient, response: ChannelResponse, @@ -151,36 +207,17 @@ async def send_with_retry( await _wait_send_interval(last_send_ts, send_lock, config, account_id) async def _attempt_send() -> dict[str, Any]: - await client.refresh_sid() + await _ensure_fresh_sid(client, account_id) return await client.send_message(chat_id, text) - last_error = None - for attempt in range(max_retries): - try: - result = await circuit_breaker.call(_attempt_send) - if result.get("success"): - message_id = result.get("data", {}).get("message_id") - return DeliveryResult(success=True, message_id=message_id) - else: - err_code = result.get("error", {}).get("code", 0) - last_error = f"DSM error code: {err_code}" - if err_code in (105, 101): - return DeliveryResult(success=False, error=last_error) - except CircuitBreakerOpenError: - return DeliveryResult(success=False, error="Circuit breaker open") - except DSMNonRetryableError as e: - logger.error(f"[SynologyChat] Non-retryable send error: {e}") - return DeliveryResult(success=False, error=str(e)) - except DSMClientError as e: - last_error = str(e) - except Exception as e: - last_error = str(e) - - delay = _calc_retry_delay(attempt, min_delay_ms, max_delay_ms, jitter) - logger.warning(f"Send retry {attempt + 1}/{max_retries} after {delay:.1f}s: {last_error}") - await asyncio.sleep(delay) - - return DeliveryResult(success=False, error=last_error or "Send failed after retries") + return await _execute_with_retry( + _attempt_send, + max_retries, + min_delay_ms, + max_delay_ms, + jitter, + circuit_breaker, + ) async def send_stream_block( @@ -206,35 +243,17 @@ async def send_stream_block( await _wait_send_interval(last_send_ts, send_lock, config, account_id) async def _attempt_send() -> dict[str, Any]: - await client.refresh_sid() + await _ensure_fresh_sid(client, account_id) return await client.send_message(chat_id, text) - last_error = None - for attempt in range(min(2, max_retries)): - try: - result = await circuit_breaker.call(_attempt_send) - if result.get("success"): - message_id = result.get("data", {}).get("message_id") - return DeliveryResult(success=True, message_id=message_id) - else: - err_code = result.get("error", {}).get("code", 0) - last_error = f"DSM error code: {err_code}" - if err_code in (105, 101): - return DeliveryResult(success=False, error=last_error) - except CircuitBreakerOpenError: - return DeliveryResult(success=False, error="Circuit breaker open") - except DSMNonRetryableError as e: - return DeliveryResult(success=False, error=str(e)) - except DSMClientError as e: - last_error = str(e) - except Exception as e: - last_error = str(e) - - if attempt < 1: - delay = _calc_retry_delay(attempt, min_delay_ms, max_delay_ms, jitter) - await asyncio.sleep(delay) - - return DeliveryResult(success=False, error=last_error or "Stream block send failed") + return await _execute_with_retry( + _attempt_send, + min(max_retries, 3), + min_delay_ms, + max_delay_ms, + jitter, + circuit_breaker, + ) async def send_media( @@ -273,38 +292,14 @@ async def send_media( await _wait_send_interval(last_send_ts, send_lock, config, account_id) async def _attempt_send() -> dict[str, Any]: - await client.refresh_sid() + await _ensure_fresh_sid(client, account_id) return await client.send_message(chat_id, text, file_url=media_url) - last_error = None - for attempt in range(max_retries): - try: - if circuit_breaker: - result = await circuit_breaker.call(_attempt_send) - else: - result = await _attempt_send() - - if result.get("success"): - message_id = result.get("data", {}).get("message_id") - return DeliveryResult(success=True, message_id=message_id) - - err_code = result.get("error", {}).get("code", 0) - last_error = f"DSM error code: {err_code}" - if err_code in (105, 101): - return DeliveryResult(success=False, error=last_error) - - except CircuitBreakerOpenError: - return DeliveryResult(success=False, error="Circuit breaker open") - except DSMNonRetryableError as e: - logger.error(f"[SynologyChat] Non-retryable send media error: {e}") - return DeliveryResult(success=False, error=str(e)) - except DSMClientError as e: - last_error = str(e) - except Exception as e: - last_error = str(e) - - delay = _calc_retry_delay(attempt, min_delay_ms, max_delay_ms, jitter) - logger.warning(f"Send media retry {attempt + 1}/{max_retries} after {delay:.1f}s: {last_error}") - await asyncio.sleep(delay) - - return DeliveryResult(success=False, error=last_error or "Send media failed after retries") + return await _execute_with_retry( + _attempt_send, + max_retries, + min_delay_ms, + max_delay_ms, + jitter, + circuit_breaker, + ) diff --git a/backend/package/yuxi/channels/adapters/synologychat/setup_wizard.py b/backend/package/yuxi/channels/adapters/synologychat/setup_wizard.py index f3c445ed..ef57c5c3 100644 --- a/backend/package/yuxi/channels/adapters/synologychat/setup_wizard.py +++ b/backend/package/yuxi/channels/adapters/synologychat/setup_wizard.py @@ -192,8 +192,13 @@ def build_setup_prompt( } -def get_setup_instructions(account_id: str = DEFAULT_ACCOUNT_ID) -> str: +def get_setup_instructions(account_id: str = DEFAULT_ACCOUNT_ID, connect_mode: str = "polling") -> str: prefix = f"[Named Account: {account_id}] " if account_id != DEFAULT_ACCOUNT_ID else "" + mode_note = ( + "Note: This adapter uses DSM API Polling mode, no incoming/outgoing webhook URLs needed." + if connect_mode == "polling" + else "Note: This adapter uses Webhook mode. Ensure Synology Chat Outgoing Webhook is configured." + ) return ( f"{prefix}Synology Chat Setup Guide:\n" "1. Install 'Synology Chat' package on your DSM via Package Center\n" @@ -201,7 +206,7 @@ def get_setup_instructions(account_id: str = DEFAULT_ACCOUNT_ID) -> str: "3. Copy the Bot's credentials (DSM URL, username, password)\n" "4. Configure the adapter with DSM URL, username, and password\n" "5. Optionally set dm_policy to 'allowlist' or 'pairing' for access control\n" - "Note: This adapter uses DSM API Polling mode, no incoming/outgoing webhook URLs needed." + f"{mode_note}" + ( "\nEnvironment variables (DSM_URL, DSM_USERNAME, etc.) are only available " "for the default account. Named accounts must use explicit configuration." diff --git a/backend/package/yuxi/channels/adapters/synologychat/target.py b/backend/package/yuxi/channels/adapters/synologychat/target.py index 43a53651..993731aa 100644 --- a/backend/package/yuxi/channels/adapters/synologychat/target.py +++ b/backend/package/yuxi/channels/adapters/synologychat/target.py @@ -5,7 +5,6 @@ ID normalization, validation, and format hints for user/group targeting. from __future__ import annotations - _CHANNEL_PREFIX = "synologychat:" diff --git a/backend/package/yuxi/channels/adapters/synologychat/webhook_send.py b/backend/package/yuxi/channels/adapters/synologychat/webhook_send.py index eca437c1..a9aeaaa5 100644 --- a/backend/package/yuxi/channels/adapters/synologychat/webhook_send.py +++ b/backend/package/yuxi/channels/adapters/synologychat/webhook_send.py @@ -6,13 +6,13 @@ as a lighter-weight alternative to the DSM API direct connection. from __future__ import annotations +import asyncio from typing import Any import httpx from yuxi.channels.adapters.synologychat.send import assert_safe_media_url from yuxi.channels.models import DeliveryResult -from yuxi.utils.logging_config import logger async def send_via_incoming_webhook( @@ -21,6 +21,7 @@ async def send_via_incoming_webhook( file_url: str | None = None, timeout: float = 30.0, client: httpx.AsyncClient | None = None, + max_retries: int = 3, ) -> DeliveryResult: if not webhook_url: return DeliveryResult(success=False, error="Incoming webhook URL not configured") @@ -45,18 +46,30 @@ async def send_via_incoming_webhook( error=f"Webhook send failed: {result.get('error', {})}", ) - try: - if client: - return await _do_send(client) - async with httpx.AsyncClient(timeout=httpx.Timeout(timeout)) as new_client: - return await _do_send(new_client) - except httpx.TimeoutException: - return DeliveryResult(success=False, error="Incoming webhook request timed out") - except httpx.HTTPStatusError as e: - return DeliveryResult(success=False, error=f"Incoming webhook HTTP {e.response.status_code}") - except Exception as e: - logger.error(f"[SynologyChat] Incoming webhook send failed: {e}") - return DeliveryResult(success=False, error=str(e)) + last_error = None + for attempt in range(max_retries): + try: + if client: + result = await _do_send(client) + else: + async with httpx.AsyncClient(timeout=httpx.Timeout(timeout)) as new_client: + result = await _do_send(new_client) + if result.success: + return result + last_error = result.error + except httpx.TimeoutException: + last_error = "Incoming webhook request timed out" + except httpx.HTTPStatusError as e: + if e.response.status_code < 500: + return DeliveryResult(success=False, error=f"Incoming webhook HTTP {e.response.status_code}") + last_error = f"Incoming webhook HTTP {e.response.status_code}" + except Exception as e: + last_error = str(e) + + if attempt < max_retries - 1: + await asyncio.sleep(1.0 * (2**attempt)) + + return DeliveryResult(success=False, error=last_error or "Webhook send failed after retries") def build_webhook_url(dsm_url: str, webhook_token: str) -> str: