diff --git a/backend/package/yuxi/channels/adapters/zalo_oa/__init__.py b/backend/package/yuxi/channels/adapters/zalo_oa/__init__.py index fcd64b84..585788be 100644 --- a/backend/package/yuxi/channels/adapters/zalo_oa/__init__.py +++ b/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.pairing import PairingStore -from yuxi.channels.adapters.zalo_oa.dedup import EventDeduplicator -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.approval import normalize_approver_id, resolve_approvers +from yuxi.channels.adapters.zalo_oa.audit import AuditEventType, AuditLogger +from yuxi.channels.adapters.zalo_oa.cache import SentMessageCache 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.cache import SentMessageCache -from yuxi.channels.adapters.zalo_oa.message_actions import describe_actions, is_action_supported -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.directory import ZaloOADirectory +from yuxi.channels.adapters.zalo_oa.doctor import run_doctor 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", diff --git a/backend/package/yuxi/channels/adapters/zalo_oa/adapter.py b/backend/package/yuxi/channels/adapters/zalo_oa/adapter.py index 9cf5e7ff..4693e477 100644 --- a/backend/package/yuxi/channels/adapters/zalo_oa/adapter.py +++ b/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) - self._last_outbound_at = time.time() - self._outbound_count += 1 - self._audit.record( - AuditEventType.MESSAGE_SENT if result.success else AuditEventType.MESSAGE_FAILED, - { - "recipient": response.identity.channel_user_id, - "message_id": result.message_id, - "success": result.success, - }, - ) - return result + async def _do_send() -> DeliveryResult: + result = await self._sender.send(response) + self._last_outbound_at = time.time() + self._outbound_count += 1 + self._audit.record( + AuditEventType.MESSAGE_SENT if result.success else AuditEventType.MESSAGE_FAILED, + { + "recipient": response.identity.channel_user_id, + "message_id": result.message_id, + "success": result.success, + }, + ) + 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: diff --git a/backend/package/yuxi/channels/adapters/zalo_oa/commands.py b/backend/package/yuxi/channels/adapters/zalo_oa/commands.py index 4a6a5bf5..cad91f8e 100644 --- a/backend/package/yuxi/channels/adapters/zalo_oa/commands.py +++ b/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 "" diff --git a/backend/package/yuxi/channels/adapters/zalo_oa/normalizer.py b/backend/package/yuxi/channels/adapters/zalo_oa/normalizer.py index 4e54bec0..f97dfdb8 100644 --- a/backend/package/yuxi/channels/adapters/zalo_oa/normalizer.py +++ b/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] diff --git a/backend/package/yuxi/channels/adapters/zalo_oa/polling.py b/backend/package/yuxi/channels/adapters/zalo_oa/polling.py index 07af55af..b2e52d8e 100644 --- a/backend/package/yuxi/channels/adapters/zalo_oa/polling.py +++ b/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 diff --git a/backend/package/yuxi/channels/adapters/zalo_oa/send.py b/backend/package/yuxi/channels/adapters/zalo_oa/send.py index 493dc08b..c48cc9e0 100644 --- a/backend/package/yuxi/channels/adapters/zalo_oa/send.py +++ b/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 diff --git a/backend/package/yuxi/channels/adapters/zalo_oa/status_issues.py b/backend/package/yuxi/channels/adapters/zalo_oa/status_issues.py index 02850d63..0a1cee08 100644 --- a/backend/package/yuxi/channels/adapters/zalo_oa/status_issues.py +++ b/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]]: