refactor(zalo_oa): 整理导入顺序并修复部分代码逻辑

1. 调整多个文件的导入排序与导入项
2. 为命令注册添加名称小写处理与空命令拦截
3. 新增link消息、reaction和已读回执的事件映射
4. 实现消息发送熔断器与流式发送支持
5. 修复导入顺序与异常捕获细节

backend/package/yuxi/channels/adapters/zalo_oa/__init__.py

@@ -1,15 +1,9 @@
+from yuxi.channels.adapters.zalo_oa.accounts import get_default_account_id, list_account_ids, resolve_account
 from yuxi.channels.adapters.zalo_oa.adapter import ZaloOAAdapter, get_webhook_path_refs
-from yuxi.channels.adapters.zalo_oa.security import DMPolicy, check_dm_allowed, load_allowlist, resolve_dm_policy
+from yuxi.channels.adapters.zalo_oa.approval import normalize_approver_id, resolve_approvers
-from yuxi.channels.adapters.zalo_oa.pairing import PairingStore
+from yuxi.channels.adapters.zalo_oa.audit import AuditEventType, AuditLogger
-from yuxi.channels.adapters.zalo_oa.dedup import EventDeduplicator
+from yuxi.channels.adapters.zalo_oa.cache import SentMessageCache
-from yuxi.channels.adapters.zalo_oa.webhook import register_webhook, unregister_webhook, get_webhook_info
-from yuxi.channels.adapters.zalo_oa.webhook_ratelimit import WebhookRateLimiter, build_rate_limit_key, resolve_client_ip
-from yuxi.channels.adapters.zalo_oa.webhook_anomaly import WebhookAnomalyTracker
-from yuxi.channels.adapters.zalo_oa.directory import ZaloOADirectory
-from yuxi.channels.adapters.zalo_oa.status_issues import collect_status_issues
 from yuxi.channels.adapters.zalo_oa.chunking import chunk_text
-from yuxi.channels.adapters.zalo_oa.messaging import normalize_messaging_target, looks_like_user_id
-from yuxi.channels.adapters.zalo_oa.accounts import list_account_ids, resolve_account, get_default_account_id
 from yuxi.channels.adapters.zalo_oa.commands import (
     CommandHandler,
     CommandRegistry,
@@ -19,12 +13,12 @@ from yuxi.channels.adapters.zalo_oa.commands import (
     handle_command,
     register_command,
 )
-from yuxi.channels.adapters.zalo_oa.approval import resolve_approvers, normalize_approver_id
+from yuxi.channels.adapters.zalo_oa.dedup import EventDeduplicator
-from yuxi.channels.adapters.zalo_oa.cache import SentMessageCache
+from yuxi.channels.adapters.zalo_oa.directory import ZaloOADirectory
-from yuxi.channels.adapters.zalo_oa.message_actions import describe_actions, is_action_supported
+from yuxi.channels.adapters.zalo_oa.doctor import run_doctor
-from yuxi.channels.adapters.zalo_oa.audit import AuditEventType, AuditLogger
 from yuxi.channels.adapters.zalo_oa.media_vision import MediaVisionCache, MediaVisionProcessor
-from yuxi.channels.adapters.zalo_oa.setup_entry import ZaloOASetupPlugin, get_setup_plugin, verify_channel_setup
+from yuxi.channels.adapters.zalo_oa.message_actions import describe_actions, is_action_supported
+from yuxi.channels.adapters.zalo_oa.messaging import looks_like_user_id, normalize_messaging_target
 from yuxi.channels.adapters.zalo_oa.outbound_media import (
     OutboundMediaHost,
     cleanup_media_cache,
@@ -32,7 +26,13 @@ from yuxi.channels.adapters.zalo_oa.outbound_media import (
     resolve_attachment,
     store_media,
 )
-from yuxi.channels.adapters.zalo_oa.doctor import run_doctor
+from yuxi.channels.adapters.zalo_oa.pairing import PairingStore
+from yuxi.channels.adapters.zalo_oa.security import DMPolicy, check_dm_allowed, load_allowlist, resolve_dm_policy
+from yuxi.channels.adapters.zalo_oa.setup_entry import ZaloOASetupPlugin, get_setup_plugin, verify_channel_setup
+from yuxi.channels.adapters.zalo_oa.status_issues import collect_status_issues
+from yuxi.channels.adapters.zalo_oa.webhook import get_webhook_info, register_webhook, unregister_webhook
+from yuxi.channels.adapters.zalo_oa.webhook_anomaly import WebhookAnomalyTracker
+from yuxi.channels.adapters.zalo_oa.webhook_ratelimit import WebhookRateLimiter, build_rate_limit_key, resolve_client_ip
 
 __all__ = [
     "ZaloOAAdapter",

backend/package/yuxi/channels/adapters/zalo_oa/adapter.py

@@ -7,35 +7,36 @@ import time
 from hashlib import sha256
 from typing import Any, ClassVar
 
-from yuxi.channels.adapters.zalo_oa.approval import check_approval_required, build_approval_request
+from yuxi.channels.adapters.zalo_oa.approval import build_approval_request, check_approval_required
 from yuxi.channels.adapters.zalo_oa.audit import AuditEventType, AuditLogger
 from yuxi.channels.adapters.zalo_oa.client import ZaloOAClient
 from yuxi.channels.adapters.zalo_oa.dedup import EventDeduplicator
+from yuxi.channels.adapters.zalo_oa.directory import ZaloOADirectory
 from yuxi.channels.adapters.zalo_oa.formatter import ZaloOAMessageFormatter
+from yuxi.channels.adapters.zalo_oa.media_vision import MediaVisionProcessor
 from yuxi.channels.adapters.zalo_oa.normalizer import SkipMessageError, ZaloOAEventNormalizer
 from yuxi.channels.adapters.zalo_oa.pairing import PairingStore, send_pairing_message, send_pairing_success
+from yuxi.channels.adapters.zalo_oa.polling import ZaloOAPoller
 from yuxi.channels.adapters.zalo_oa.probe import probe_zalo_oa
 from yuxi.channels.adapters.zalo_oa.security import DMPolicy, check_dm_allowed, load_allowlist, resolve_dm_policy
 from yuxi.channels.adapters.zalo_oa.send import ZaloOASender
 from yuxi.channels.adapters.zalo_oa.signature import verify_zalo_oa_signature
+from yuxi.channels.adapters.zalo_oa.status_issues import collect_status_issues
+from yuxi.channels.adapters.zalo_oa.voice import ZaloOAVoice
 from yuxi.channels.adapters.zalo_oa.webhook import register_webhook, unregister_webhook
+from yuxi.channels.adapters.zalo_oa.webhook_anomaly import WebhookAnomalyTracker
 from yuxi.channels.adapters.zalo_oa.webhook_ratelimit import (
     WebhookRateLimiter,
     build_rate_limit_key,
     resolve_client_ip,
 )
-from yuxi.channels.adapters.zalo_oa.webhook_anomaly import WebhookAnomalyTracker
-from yuxi.channels.adapters.zalo_oa.directory import ZaloOADirectory
-from yuxi.channels.adapters.zalo_oa.polling import ZaloOAPoller
-from yuxi.channels.adapters.zalo_oa.voice import ZaloOAVoice
-from yuxi.channels.adapters.zalo_oa.media_vision import MediaVisionProcessor
-from yuxi.channels.adapters.zalo_oa.status_issues import collect_status_issues
 from yuxi.channels.base import BaseChannelAdapter
 from yuxi.channels.capabilities import ChannelCapabilities
 from yuxi.channels.exceptions import (
     ChannelAuthenticationError,
     MessageFormatError,
 )
+from yuxi.channels.infra.circuit_breaker import CircuitBreaker, CircuitBreakerOpenError
 from yuxi.channels.meta import ChannelMeta
 from yuxi.channels.models import (
     ChannelMessage,
@@ -92,7 +93,7 @@ class ZaloOAAdapter(BaseChannelAdapter):
     reply_to_mode: ClassVar[str] = "off"
 
     capabilities = ChannelCapabilities(
-        chat_types=["direct"],  # Zalo OA API 仅支持 direct（1对1私聊），不支持 group
+        chat_types=["direct"],
         reply=False,
         unsend=True,
         media=True,
@@ -109,6 +110,7 @@ class ZaloOAAdapter(BaseChannelAdapter):
         streaming_modes=["off"],
         text_chunk_limit=2000,
         max_media_size_mb=10,
+        typing=True,
     )
     meta = ChannelMeta(
         id="zalo_oa",
@@ -161,6 +163,7 @@ class ZaloOAAdapter(BaseChannelAdapter):
         self._heartbeat_count = 0
         self._heartbeat_failures = 0
         self._config_hash = self._compute_config_hash()
+        self._circuit_breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60, channel_id="zalo_oa")
 
     def _compute_config_hash(self) -> str:
         """计算当前配置的哈希值，用于检测配置变更."""
@@ -383,18 +386,85 @@ class ZaloOAAdapter(BaseChannelAdapter):
         if self._status != ChannelStatus.CONNECTED or not self._sender:
             return DeliveryResult(success=False, error="Zalo OA not connected")
 
-        result = await self._sender.send(response)
+        async def _do_send() -> DeliveryResult:
-        self._last_outbound_at = time.time()
+            result = await self._sender.send(response)
-        self._outbound_count += 1
+            self._last_outbound_at = time.time()
-        self._audit.record(
+            self._outbound_count += 1
-            AuditEventType.MESSAGE_SENT if result.success else AuditEventType.MESSAGE_FAILED,
+            self._audit.record(
-            {
+                AuditEventType.MESSAGE_SENT if result.success else AuditEventType.MESSAGE_FAILED,
-                "recipient": response.identity.channel_user_id,
+                {
-                "message_id": result.message_id,
+                    "recipient": response.identity.channel_user_id,
-                "success": result.success,
+                    "message_id": result.message_id,
-            },
+                    "success": result.success,
-        )
+                },
-        return result
+            )
+            return result
+
+        try:
+            return await self._circuit_breaker.call(_do_send)
+        except CircuitBreakerOpenError:
+            return DeliveryResult(success=False, error="Circuit breaker open")
+
+    async def send_stream_chunk(
+        self,
+        chat_id: str,
+        msg_id: str,
+        chunk: str,
+        finished: bool = False,
+    ) -> DeliveryResult:
+        if finished:
+            from yuxi.channels.models import ChannelIdentity, ChannelResponse
+
+            identity = ChannelIdentity(
+                channel_id=self.channel_id,
+                channel_type=self.channel_type,
+                channel_chat_id=chat_id,
+                channel_user_id=chat_id,
+            )
+            response = ChannelResponse(identity=identity, content=chunk)
+            return await self.send(response)
+
+        if not self._sender:
+            return DeliveryResult(success=False, error="Sender not initialized")
+
+        streaming_cfg = self.config.get("streaming", {})
+        if not isinstance(streaming_cfg, dict):
+            streaming_cfg = {}
+
+        if self.config.get("auto_typing", True):
+            await self._sender.send_typing_indicator(chat_id)
+
+        from yuxi.channels.adapters.zalo_oa.chunking import chunk_text
+        from yuxi.channels.models import ChannelIdentity, ChannelResponse
+
+        text_limit = self.text_chunk_limit
+        fallback_cfg = streaming_cfg.get("fallback", {})
+        if isinstance(fallback_cfg, dict):
+            text_limit = fallback_cfg.get("chunk_size", text_limit)
+
+        chunks = chunk_text(chunk, text_limit)
+        show_progress = streaming_cfg.get("progress_indicator", True)
+        total = len(chunks)
+
+        for i, ch in enumerate(chunks, 1):
+            content = ch
+            if show_progress and total > 1:
+                content = f"[{i}/{total}] {ch}"
+            identity = ChannelIdentity(
+                channel_id=self.channel_id,
+                channel_type=self.channel_type,
+                channel_chat_id=chat_id,
+                channel_user_id=chat_id,
+            )
+            response = ChannelResponse(identity=identity, content=content)
+            await self.send(response)
+
+            if i < total:
+                import asyncio
+
+                await asyncio.sleep(0.3)
+
+        return DeliveryResult(success=True)
 
     def validate_webhook_request(self, headers: dict, body: bytes) -> None:
         content_type = headers.get("content-type", "")
@@ -440,6 +510,7 @@ class ZaloOAAdapter(BaseChannelAdapter):
             "user_send_video",
             "user_send_audio",
             "user_forward_message",
+            "user_send_link",
         ):
             if not self._dedup.claim(data):
                 logger.debug("[ZaloOA] Duplicate event skipped")
@@ -583,7 +654,7 @@ class ZaloOAAdapter(BaseChannelAdapter):
             try:
                 try:
                     raw, headers, future = await asyncio.wait_for(queue.get(), timeout=idle_timeout)
-                except asyncio.TimeoutError:
+                except TimeoutError:
                     break
                 try:
                     if headers:

backend/package/yuxi/channels/adapters/zalo_oa/commands.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
-from typing import Any
 from collections.abc import Callable, Coroutine
+from typing import Any
 
 COMMAND_PREFIXES = ("/", "!")
 
@@ -22,6 +22,7 @@ class CommandRegistry:
         description: str = "",
         handler: CommandHandler | None = None,
     ):
+        name = name.lower()
         self._commands[name] = {
             "description": description,
             "handler": handler or self._make_default_handler(name),
@@ -91,7 +92,9 @@ def extract_command(text: str) -> tuple[str | None, str]:
     text = text.strip()
     for prefix in COMMAND_PREFIXES:
         if text.startswith(prefix):
-            cmd_text = text[len(prefix) :]
+            cmd_text = text[len(prefix) :].strip()
+            if not cmd_text:
+                return "", ""
             parts = cmd_text.split(None, 1)
             cmd = parts[0].lower()
             args = parts[1] if len(parts) > 1 else ""

backend/package/yuxi/channels/adapters/zalo_oa/normalizer.py

@@ -71,11 +71,14 @@ class ZaloOAEventNormalizer:
             "user_send_gif": EventType.MESSAGE_RECEIVED,
             "user_send_location": EventType.MESSAGE_RECEIVED,
             "user_send_business_card": EventType.MESSAGE_RECEIVED,
+            "user_send_link": EventType.MESSAGE_RECEIVED,
             "user_submit_form": EventType.CARD_ACTION,
             "user_click_button": EventType.CARD_ACTION,
             "user_send_video": EventType.MESSAGE_RECEIVED,
             "user_send_audio": EventType.MESSAGE_RECEIVED,
             "user_forward_message": EventType.MESSAGE_RECEIVED,
+            "reaction": EventType.REACTION_ADDED,
+            "read_receipt": EventType.READ_RECEIPT,
         }
         return mapping.get(event_name, EventType.MESSAGE_RECEIVED)
 
@@ -89,6 +92,7 @@ class ZaloOAEventNormalizer:
             "user_send_gif": MessageType.IMAGE,
             "user_send_location": MessageType.LOCATION,
             "user_send_business_card": MessageType.CARD,
+            "user_send_link": MessageType.TEXT,
             "user_send_video": MessageType.VIDEO,
             "user_send_audio": MessageType.AUDIO,
             "user_submit_form": MessageType.TEXT,
@@ -167,6 +171,22 @@ class ZaloOAEventNormalizer:
                 )
             elif att_type == "location":
                 text = f"Location: ({payload.get('latitude')}, {payload.get('longitude')})"
+            elif att_type == "link":
+                url = payload.get("url", "")
+                title = payload.get("title", "")
+                description = payload.get("description", "")
+                text_parts = [f"[Link] {url}"]
+                if title:
+                    text_parts.append(f"Title: {title}")
+                if description:
+                    text_parts.append(f"Description: {description}")
+                text = "\n".join(text_parts)
+            elif att_type == "business_card":
+                contact_name = payload.get("name", payload.get("contact_name", ""))
+                phone = payload.get("phone", "")
+                text = f"[Business Card] {contact_name}"
+                if phone:
+                    text += f" ({phone})"
             else:
                 text = text or str(payload)[:500]
 

backend/package/yuxi/channels/adapters/zalo_oa/polling.py

@@ -2,8 +2,8 @@ from __future__ import annotations
 
 import asyncio
 import time
-from typing import Any
 from collections.abc import Callable
+from typing import Any
 
 from yuxi.utils.logging_config import logger
 

backend/package/yuxi/channels/adapters/zalo_oa/send.py

@@ -4,8 +4,8 @@ import asyncio
 import json
 import os
 import random
-from typing import Any
 from collections.abc import Callable
+from typing import Any
 
 from yuxi.channels.adapters.zalo_oa.cache import SentMessageCache
 from yuxi.channels.adapters.zalo_oa.client import ZaloOAClient

backend/package/yuxi/channels/adapters/zalo_oa/status_issues.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from typing import Any
 
-from yuxi.channels.adapters.zalo_oa.security import DMPolicy, resolve_dm_policy, collect_security_warnings
+from yuxi.channels.adapters.zalo_oa.security import DMPolicy, collect_security_warnings, resolve_dm_policy
 
 
 def collect_status_issues(config: dict[str, Any], oa_info: dict[str, Any]) -> list[dict[str, Any]]: