From 6cb56d5e4107bfe92b74b47eeaa0eb3db45628fc Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Wed, 13 May 2026 16:18:25 +0800 Subject: [PATCH] =?UTF-8?q?refactor(zalo-adapter):=20=E6=95=B4=E7=90=86Zal?= =?UTF-8?q?o=E7=94=A8=E6=88=B7=E9=80=82=E9=85=8D=E5=99=A8=E4=BB=A3?= =?UTF-8?q?=E7=A0=81=E5=B9=B6=E4=BC=98=E5=8C=96=E9=87=8D=E8=BF=9E=E9=80=BB?= =?UTF-8?q?=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 调整导入顺序与补全缺失依赖 2. 新增Zalo目标前缀映射与最大重连延迟配置 3. 优化反应消息ID获取逻辑与桥接URL检查 4. 改进WebSocket重连退避策略与错误处理 5. 重构发送消息逻辑，统一附件处理与音频发送流程 6. 调整类初始化代码位置与导入依赖顺序 --- .../channels/adapters/zalo_user/adapter.py | 12 ++--- .../yuxi/channels/adapters/zalo_user/cards.py | 2 +- .../channels/adapters/zalo_user/constants.py | 2 + .../channels/adapters/zalo_user/directory.py | 3 +- .../zalo_user/interactive_dispatch.py | 2 +- .../channels/adapters/zalo_user/monitor.py | 12 ++++- .../channels/adapters/zalo_user/normalize.py | 6 ++- .../yuxi/channels/adapters/zalo_user/send.py | 52 ++++++++++++------- .../adapters/zalo_user/status_issues.py | 11 ++-- 9 files changed, 67 insertions(+), 35 deletions(-) diff --git a/backend/package/yuxi/channels/adapters/zalo_user/adapter.py b/backend/package/yuxi/channels/adapters/zalo_user/adapter.py index 03f569d1..09e2d927 100644 --- a/backend/package/yuxi/channels/adapters/zalo_user/adapter.py +++ b/backend/package/yuxi/channels/adapters/zalo_user/adapter.py @@ -30,6 +30,7 @@ from yuxi.channels.registry import register_builtin_adapter from yuxi.utils.datetime_utils import utc_now_naive from yuxi.utils.logging_config import logger +from .allow_from import resolve_allow_from_entries from .backoff import BackoffManager from .bridge import BridgeClient from .constants import ( @@ -44,7 +45,6 @@ from .directory import get_self_info, list_group_members, list_groups, list_peer from .exec_approval import create_exec_approval from .group_cache import group_context_cache from .group_sync import sync_group_list -from .allow_from import resolve_allow_from_entries from .interactive_dispatch import ( default_approval_handler, default_model_selector_handler, @@ -67,8 +67,8 @@ from .session import ( check_mention_required, resolve_agent_route, ) -from .sticker_handler import augment_sticker_message from .status_issues import collect_status_issues +from .sticker_handler import augment_sticker_message from .text_styles import ( has_markdown_syntax, markdown_to_zalo_styles, @@ -133,6 +133,10 @@ class ZaloUserAdapter(BaseChannelAdapter): def __init__(self, config: dict[str, Any] | None = None): super().__init__(config) + self._config = config or {} + self._profile = _env_fallback( + ["ZALOUSER_PROFILE", "ZCA_PROFILE", "ZALO_PROFILE"], self._config, "profile", "default" + ) self._status = ChannelStatus.DISCONNECTED self._bridge: BridgeClient | None = None self._login_flow: LoginFlow | None = None @@ -152,10 +156,6 @@ class ZaloUserAdapter(BaseChannelAdapter): self._last_connected_at: float | None = None self._last_message_at: float | None = None self._last_transport_activity_at: float | None = None - self._config = config or {} - self._profile = _env_fallback( - ["ZALOUSER_PROFILE", "ZCA_PROFILE", "ZALO_PROFILE"], self._config, "profile", "default" - ) self._enabled = self._config.get("enabled", True) self._msg_id_tracker = MessageIdTracker() self._outbound_seq = OutboundSequencer() diff --git a/backend/package/yuxi/channels/adapters/zalo_user/cards.py b/backend/package/yuxi/channels/adapters/zalo_user/cards.py index 4c0549ac..7d351cd7 100644 --- a/backend/package/yuxi/channels/adapters/zalo_user/cards.py +++ b/backend/package/yuxi/channels/adapters/zalo_user/cards.py @@ -116,8 +116,8 @@ def template_confirmation_card( from .inline_buttons import ( ButtonGroup, ButtonStyle, - callback_button, InlineButtonRow, + callback_button, ) card = ZaloCard( diff --git a/backend/package/yuxi/channels/adapters/zalo_user/constants.py b/backend/package/yuxi/channels/adapters/zalo_user/constants.py index 1858507d..8d42797f 100644 --- a/backend/package/yuxi/channels/adapters/zalo_user/constants.py +++ b/backend/package/yuxi/channels/adapters/zalo_user/constants.py @@ -9,6 +9,7 @@ WS_MAX_MESSAGE_SIZE = 10 * 1024 * 1024 WS_PING_INTERVAL = 30 WS_PING_TIMEOUT = 10 WS_RECONNECT_DELAY = 5 +MAX_RECONNECT_DELAY = 120 WATCHDOG_INTERVAL = 30 WATCHDOG_MAX_IDLE = 35 @@ -70,6 +71,7 @@ TARGET_PREFIXES: dict[str, str] = { "dm:": "user", "u:": "user", "u-": "user", + "zalo:": "user", "group:": "group", "g:": "group", "g-": "group", diff --git a/backend/package/yuxi/channels/adapters/zalo_user/directory.py b/backend/package/yuxi/channels/adapters/zalo_user/directory.py index 61319f4e..d9954f5c 100644 --- a/backend/package/yuxi/channels/adapters/zalo_user/directory.py +++ b/backend/package/yuxi/channels/adapters/zalo_user/directory.py @@ -1,8 +1,7 @@ from __future__ import annotations -from typing import Any - import asyncio +from typing import Any from yuxi.utils.logging_config import logger diff --git a/backend/package/yuxi/channels/adapters/zalo_user/interactive_dispatch.py b/backend/package/yuxi/channels/adapters/zalo_user/interactive_dispatch.py index 5fd13135..605a2eb8 100644 --- a/backend/package/yuxi/channels/adapters/zalo_user/interactive_dispatch.py +++ b/backend/package/yuxi/channels/adapters/zalo_user/interactive_dispatch.py @@ -1,7 +1,7 @@ from __future__ import annotations +from collections.abc import Awaitable, Callable from typing import Any -from collections.abc import Callable, Awaitable from yuxi.channels.models import ChannelMessage from yuxi.utils.logging_config import logger diff --git a/backend/package/yuxi/channels/adapters/zalo_user/monitor.py b/backend/package/yuxi/channels/adapters/zalo_user/monitor.py index 95cb11b7..8df4dbac 100644 --- a/backend/package/yuxi/channels/adapters/zalo_user/monitor.py +++ b/backend/package/yuxi/channels/adapters/zalo_user/monitor.py @@ -12,6 +12,7 @@ from yuxi.channels.models import ChannelMessage from yuxi.utils.logging_config import logger from .constants import ( + MAX_RECONNECT_DELAY, WATCHDOG_INTERVAL, WATCHDOG_MAX_IDLE, WS_MAX_MESSAGE_SIZE, @@ -107,6 +108,7 @@ class MessageMonitor: logger.info("[ZaloUser] Message monitor stopped") async def _listen_loop(self) -> None: + consecutive_failures = 0 while self._running: try: async with websockets.connect( @@ -116,6 +118,7 @@ class MessageMonitor: max_size=WS_MAX_MESSAGE_SIZE, ) as ws: self._ws = ws + consecutive_failures = 0 self._last_activity = time.monotonic() logger.info(f"[ZaloUser] WebSocket connected: {self._ws_url}") async for raw_message in ws: @@ -126,9 +129,14 @@ class MessageMonitor: except asyncio.CancelledError: break except Exception as e: - logger.warning(f"[ZaloUser] WebSocket disconnected: {e}") + consecutive_failures += 1 + delay = min(WS_RECONNECT_DELAY * (2 ** min(consecutive_failures - 1, 5)), MAX_RECONNECT_DELAY) + logger.warning( + f"[ZaloUser] WebSocket disconnected (failures={consecutive_failures}), " + f"reconnecting in {delay}s: {e}" + ) if self._running: - await asyncio.sleep(WS_RECONNECT_DELAY) + await asyncio.sleep(delay) logger.info("[ZaloUser] Message monitor loop ended") diff --git a/backend/package/yuxi/channels/adapters/zalo_user/normalize.py b/backend/package/yuxi/channels/adapters/zalo_user/normalize.py index 4a464629..07306277 100644 --- a/backend/package/yuxi/channels/adapters/zalo_user/normalize.py +++ b/backend/package/yuxi/channels/adapters/zalo_user/normalize.py @@ -220,7 +220,11 @@ def _normalize_reaction(channel_id: str, raw_payload: dict[str, Any]) -> Channel channel_type=ChannelType.ZALO_USER, channel_user_id=str(raw_payload.get("from_id", "")), channel_chat_id=str(raw_payload.get("conversation_id", "")), - channel_message_id=str(raw_payload.get("reaction_message_id") or raw_payload.get("message_id", "")), + channel_message_id=str( + raw_payload.get("reaction_message_id") + if raw_payload.get("reaction_message_id") is not None + else raw_payload.get("message_id", "") + ), ) raw_timestamp = raw_payload.get("timestamp") diff --git a/backend/package/yuxi/channels/adapters/zalo_user/send.py b/backend/package/yuxi/channels/adapters/zalo_user/send.py index b544867a..710a5f82 100644 --- a/backend/package/yuxi/channels/adapters/zalo_user/send.py +++ b/backend/package/yuxi/channels/adapters/zalo_user/send.py @@ -3,7 +3,9 @@ from __future__ import annotations import asyncio import os import time -from typing import Any, TYPE_CHECKING +from typing import TYPE_CHECKING, Any + +import httpx from yuxi.channels.models import ( ChannelResponse, @@ -19,6 +21,15 @@ from .rate_limiter import RateLimiter if TYPE_CHECKING: from yuxi.channels.infra.circuit_breaker import CircuitBreaker +_ATTACHMENT_BRIDGE_TYPES: dict[MessageType, str] = { + MessageType.IMAGE: "image", + MessageType.VIDEO: "video", + MessageType.AUDIO: "audio", + MessageType.FILE: "file", +} + +_NON_RETRIABLE_ERRORS = {"invalid_cookie", "session_expired", "bad_request", "forbidden", "unauthorized"} + class OutboundSequencer: def __init__(self): @@ -80,7 +91,16 @@ async def _send_one( result = await bridge.send_message(payload) if result.success: return result + error_tag = (result.error or "").lower() + if any(tag in error_tag for tag in _NON_RETRIABLE_ERRORS): + return result last_error = result.error + except httpx.HTTPStatusError as e: + if 400 <= e.response.status_code < 500: + return DeliveryResult(success=False, error=f"Client error: {e}") + last_error = str(e) + except (httpx.ConnectError, httpx.TimeoutException) as e: + last_error = str(e) except Exception as e: last_error = str(e) logger.warning(f"[ZaloUser] Send attempt {attempt + 1}/{max_attempts} failed: {e}") @@ -149,14 +169,12 @@ def _build_send_payload(response: ChannelResponse, config: dict[str, Any]) -> di if custom_styles and isinstance(custom_styles, list): payload["custom_text_styles"] = custom_styles - _ATTACHMENT_TYPES: dict[MessageType, str] = { - MessageType.IMAGE: "image", - MessageType.VIDEO: "video", - MessageType.AUDIO: "audio", - MessageType.FILE: "file", - } - attachment_bridge_type = _ATTACHMENT_TYPES.get(response.message_type) + attachment_bridge_type = _ATTACHMENT_BRIDGE_TYPES.get(response.message_type) if attachment_bridge_type and response.attachments: + if len(response.attachments) > 1: + logger.warning( + f"[ZaloUser] Multiple attachments ({len(response.attachments)}) detected, only sending the first one." + ) att = response.attachments[0] payload["attachments"] = [ { @@ -231,18 +249,14 @@ async def send_audio( ) -> DeliveryResult: try: derived_name = file_name or _derive_file_name(audio_url, "audio", "audio.mp3") + payload: dict[str, Any] = { + "conversation_id": conversation_id, + "message_type": "voice", + "attachments": [{"type": "voice", "url": audio_url, "filename": derived_name}], + } if caption: - cap_result = await bridge.send_message( - { - "conversation_id": conversation_id, - "message_type": "text", - "text": caption, - } - ) - if not cap_result.success: - return cap_result - voice_result = await bridge.send_voice_attachment(conversation_id, audio_url, derived_name) - return voice_result + payload["text"] = caption + return await bridge.send_message(payload) except Exception as e: return DeliveryResult(success=False, error=str(e)) diff --git a/backend/package/yuxi/channels/adapters/zalo_user/status_issues.py b/backend/package/yuxi/channels/adapters/zalo_user/status_issues.py index 2d5f356c..4ab7ecaa 100644 --- a/backend/package/yuxi/channels/adapters/zalo_user/status_issues.py +++ b/backend/package/yuxi/channels/adapters/zalo_user/status_issues.py @@ -68,13 +68,18 @@ def collect_status_issues( ).to_dict() ) - if bridge_url and not bridge_url.startswith("https://"): + if ( + bridge_url + and not bridge_url.startswith("https://") + and "localhost" not in bridge_url + and "127.0.0.1" not in bridge_url + ): issues.append( StatusIssue( level="warn", - message="Bridge URL 未使用 HTTPS", + message="Bridge URL 未使用 HTTPS（非本地连接）", detail=( - f"当前 Bridge URL ({bridge_url}) 使用 HTTP 协议，" + f"当前 Bridge URL ({bridge_url}) 使用 HTTP 协议且非本地地址，" "凭据和会话令牌将以明文传输。生产环境建议使用 HTTPS。" ), ).to_dict()