diff --git a/backend/package/yuxi/channels/adapters/wechat/adapter.py b/backend/package/yuxi/channels/adapters/wechat/adapter.py index abc8281d..5ee4f0c9 100644 --- a/backend/package/yuxi/channels/adapters/wechat/adapter.py +++ b/backend/package/yuxi/channels/adapters/wechat/adapter.py @@ -26,6 +26,7 @@ from yuxi.channels.models import ( EventType, HealthStatus, ) +from yuxi.channels.policy.dedup_policy import DedupPolicy from yuxi.channels.registry import register_builtin_adapter from yuxi.utils.datetime_utils import utc_now_naive from yuxi.utils.logging_config import logger @@ -103,11 +104,13 @@ from .selector import WeChatSelectorAdapter from .send_cache import WeChatSendCache from .sent_message_store import SentMessageStore from .template_adapter import WeChatTemplateAdapter +from .template_card import TemplateCard from .threading_adapter import WeChatThreadingAdapter from .voice import send_voice_bridge, send_voice_mp, send_voice_wecom from .wecom import ( WeComClient, WeComMonitor, + build_wecom_markdown_payload, build_wecom_text_payload, send_wecom_message, ) @@ -126,9 +129,10 @@ class WeChatAdapter(BaseChannelAdapter): meta = WECHAT_META text_chunk_limit = 2048 - supports_markdown = False + supports_markdown = True supports_streaming = False max_media_size_mb = 20 + webhook_path = "wechat" def __init__(self, config: dict[str, Any] | None = None): super().__init__(config) @@ -141,14 +145,23 @@ class WeChatAdapter(BaseChannelAdapter): self._polling_task: asyncio.Task | None = None self._wecom_client: WeComClient | None = None + self._wecom_client_lock = asyncio.Lock() self._mp_client: MPClient | None = None self._bridge_client: BridgeClient | None = None self._qr_login_mgr: QRLoginManager | None = None self._monitor: WeComMonitor | None = None + self._sub_monitors: dict[str, WeComMonitor] = {} + self._sub_clients: dict[str, WeComClient | MPClient | BridgeClient] = {} + self._sub_polling_tasks: dict[str, asyncio.Task] = {} self._rate_limiter = TokenBucketRateLimiter(rate=20, per=60.0) - self._dedup: dict[tuple, float] = {} - self._dedup_ttl = 1.0 + self._dedup_ttl = self.config.get("dedup_ttl_seconds", 1.0) + self._max_dedup_entries = self.config.get("max_dedup_entries", 10000) + self._send_dedup = DedupPolicy( + ttl=int(self._dedup_ttl), + maxsize=self._max_dedup_entries, + redis_url=os.environ.get("REDIS_URL"), + ) self._banned = False self._banned_reason: str | None = None @@ -175,7 +188,7 @@ class WeChatAdapter(BaseChannelAdapter): self._mention = WeChatMentionAdapter() self._router = WeChatMessagingRouter() self._heartbeat = WeChatHeartbeatAdapter(self.config) - self._message_actions = WeChatMessageActionAdapter() + self._message_actions = WeChatMessageActionAdapter(self) self._group_admin = WeChatGroupAdmin() self._attachment = WeChatAttachmentAdapter() @@ -249,6 +262,8 @@ class WeChatAdapter(BaseChannelAdapter): self._status = ChannelStatus.CONNECTED logger.info(f"[WeChat] Channel started ({self._mode} mode)") + await self._connect_sub_accounts() + except ChannelAuthenticationError: self._status = ChannelStatus.ERROR raise @@ -274,6 +289,22 @@ class WeChatAdapter(BaseChannelAdapter): await self._monitor.stop() self._monitor = None + for account_id, monitor in list(self._sub_monitors.items()): + try: + await monitor.stop() + except Exception as e: + logger.warning(f"[WeChat] Error stopping sub-account monitor {account_id}: {e}") + self._sub_monitors.clear() + + for account_id, task in list(self._sub_polling_tasks.items()): + if not task.done(): + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + self._sub_polling_tasks.clear() + if self._polling_task and not self._polling_task.done(): self._polling_task.cancel() try: @@ -288,6 +319,8 @@ class WeChatAdapter(BaseChannelAdapter): self._qr_login_mgr = None self._qr_url = None + await self._send_dedup.close() + if self._http_client: await self._http_client.aclose() self._http_client = None @@ -340,8 +373,14 @@ class WeChatAdapter(BaseChannelAdapter): error=f"Channel temporarily banned: {self._banned_reason or 'API unauthorized (48001)'}{remaining}", ) - if self._mode == "wecom" and self._is_wecom_duplicate(response): - return DeliveryResult(success=False, error="Duplicate message within dedup window") + if self._mode == "wecom": + import hashlib + + user_id = response.identity.channel_user_id + dedup_hash = hashlib.sha256(f"{user_id}:{response.content}".encode()).hexdigest() + dedup_key = f"send:{self.channel_id}:{dedup_hash}" + if await self._send_dedup.check_and_mark(dedup_key, ttl=int(self._dedup_ttl)): + return DeliveryResult(success=False, error="Duplicate message within dedup window") if not await self._rate_limiter.acquire(): return DeliveryResult(success=False, error="Rate limit exceeded, try again later") @@ -354,26 +393,79 @@ class WeChatAdapter(BaseChannelAdapter): except CircuitBreakerOpenError: return DeliveryResult(success=False, error="Circuit breaker is open, WeChat channel unavailable") - def _is_wecom_duplicate(self, response: ChannelResponse) -> bool: - import time + async def send_stream_chunk( + self, + chat_id: str, + msg_id: str, + chunk: str, + finished: bool = False, + ) -> DeliveryResult: + is_wecom = self._mode == "wecom" - user_id = response.identity.channel_user_id - key = (user_id, response.content) - now = time.monotonic() + if is_wecom and self._heartbeat.typing_enabled and not finished: + try: + await self._heartbeat.send_typing_via_response(self.send, chat_id, active=True) + except Exception: + pass - if key in self._dedup: - if now - self._dedup[key] < self._dedup_ttl: - return True - self._dedup[key] = now + from yuxi.channels.models import ChannelIdentity, ChannelResponse - stale_keys = [k for k, ts in self._dedup.items() if now - ts > self._dedup_ttl] - for k in stale_keys: - del self._dedup[k] - return False + identity = ChannelIdentity( + channel_id=self.channel_id, + channel_type=self.channel_type, + channel_chat_id=chat_id, + channel_user_id="", + ) + + if finished: + response = ChannelResponse(identity=identity, content=chunk) + return await self.send(response) + + stream_cfg = self.config.get("streaming", {}) if isinstance(self.config.get("streaming"), dict) else {} + fallback_cfg = stream_cfg.get("fallback", {}) if isinstance(stream_cfg, dict) else {} + chunk_size = fallback_cfg.get("chunk_size", self.text_chunk_limit) + + if len(chunk) <= chunk_size: + response = ChannelResponse(identity=identity, content=chunk) + return await self.send(response) + + from .streaming_fallback import ParagraphChunker, ProgressPrefixFormatter + + chunker = ParagraphChunker(chunk_size=chunk_size) + chunker.feed(chunk) + paragraphs = chunker.flush() + if not paragraphs: + paragraphs = [chunk] + + progress_enabled = stream_cfg.get("progress_indicator", True) + formatter = ProgressPrefixFormatter(enabled=progress_enabled) + total = len(paragraphs) + + for i, para in enumerate(paragraphs, 1): + prefix = formatter.format(i, total) + content = f"{prefix} {para}" if prefix else para + response = ChannelResponse(identity=identity, content=content) + await self.send(response) + + return DeliveryResult(success=True) + + async def send_typing_indicator(self, chat_id: str, active: bool) -> DeliveryResult: + is_wecom = self._mode == "wecom" + if not active or not is_wecom or not self._heartbeat.typing_enabled: + return DeliveryResult(success=True) + + try: + result = await self._heartbeat.send_typing_via_response(self.send, chat_id, active=True) + return result or DeliveryResult(success=True) + except Exception as e: + return DeliveryResult(success=False, error=str(e)) async def _do_send(self, response: ChannelResponse, disable_notification: bool = False) -> DeliveryResult: if self._mode == "wecom": - result = await self._send_wecom(response, disable_notification=disable_notification) + if response.metadata.get("template_card") is not None: + result = await self._send_wecom_template_card(response) + else: + result = await self._send_wecom(response, disable_notification=disable_notification) self._check_banned_response(result) return result elif self._mode == "mp": @@ -612,19 +704,151 @@ class WeChatAdapter(BaseChannelAdapter): "[mp] (app_id + app_secret), or [personal] (bridge_url)." ) + async def _connect_sub_accounts(self) -> None: + account_ids = self._account_mgr.list_account_ids(self.config) + if len(account_ids) <= 1: + return + + sub_limit = self.config.get("max_sub_accounts", 5) + sub_count = 0 + + for account_id in account_ids: + if account_id == "default": + continue + + if sub_count >= sub_limit: + logger.warning(f"[WeChat] Sub-account limit ({sub_limit}) reached, skipping remaining sub-accounts") + break + + account = self._account_mgr.resolve_account(self.config, account_id) + if not account.enabled: + continue + + sub_count += 1 + logger.info( + f"[WeChat] Connecting sub-account '{account_id}' (mode={account.mode}, {sub_count}/{sub_limit})" + ) + + if account.mode == "wecom" and self._http_client: + await self._connect_sub_wecom(account_id, account) + elif account.mode == "mp": + logger.info(f"[WeChat] Sub-account '{account_id}': MP mode uses webhook, no monitor needed") + elif account.mode == "personal" and self._http_client: + await self._connect_sub_personal(account_id, account) + + async def _connect_sub_wecom(self, account_id: str, account: Any) -> None: + sub_config = { + "corp_id": account.corp_id, + "corp_secret": account.corp_secret, + "agent_id": account.agent_id, + "proxy": account.proxy or self.config.get("proxy"), + "webhook_url": account.bridge_url or "", + } + sub_client = WeComClient(self._http_client, sub_config) + + token_data = await self.state_get(f"{account_id}_access_token", namespace="auth") + if token_data and isinstance(token_data, dict): + import time as _time + + token_str = token_data.get("access_token") + expires_at = token_data.get("expires_at", 0) + if token_str and expires_at > _time.time() + 300: + sub_client.set_token(token_str, expires_at) + logger.info(f"[WeChat] Sub-account '{account_id}' token restored from state_store") + + try: + await sub_client.get_access_token() + self._sub_clients[account_id] = sub_client + + import time as _time + + await self.state_set( + f"{account_id}_access_token", + { + "access_token": sub_client._access_token, + "expires_at": sub_client._token_expires_at, + }, + namespace="auth", + ) + + sub_monitor = WeComMonitor(self._http_client, sub_config, self._handle_message) + await sub_monitor.start(lambda: sub_client.get_access_token()) + self._sub_monitors[account_id] = sub_monitor + + logger.info(f"[WeChat] Sub-account '{account_id}' WeCom monitor started") + except Exception as e: + logger.error(f"[WeChat] Failed to connect sub-account '{account_id}': {e}") + + async def _connect_sub_personal(self, account_id: str, account: Any) -> None: + bridge_url = (account.bridge_url or "").rstrip("/") + if not bridge_url: + logger.warning(f"[WeChat] Sub-account '{account_id}' has no bridge_url") + return + + sub_bridge = BridgeClient(self._http_client, bridge_url) + healthy = await sub_bridge.health_check() + if not healthy: + logger.warning(f"[WeChat] Sub-account '{account_id}' bridge health check failed") + return + + self._sub_clients[account_id] = sub_bridge + + poll_interval = self.config.get("poll_interval", 1.0) + task = asyncio.create_task(self._sub_bridge_event_loop(account_id, sub_bridge, poll_interval)) + self._sub_polling_tasks[account_id] = task + logger.info(f"[WeChat] Sub-account '{account_id}' bridge polling started") + + async def _sub_bridge_event_loop(self, account_id: str, bridge_client: BridgeClient, poll_interval: float) -> None: + iteration = 0 + while True: + try: + messages = await bridge_client.fetch_events() + for raw_msg in messages or []: + channel_msg = self._normalize_bridge_message(raw_msg) + await self._handle_message(channel_msg) + except asyncio.CancelledError: + break + except Exception as e: + logger.error(f"[WeChat] Sub-account '{account_id}' bridge event loop error: {e}") + await asyncio.sleep(poll_interval) + iteration += 1 + # ---- WeCom Mode ---- async def _start_wecom(self) -> None: - corp_id = self.config.get("corp_id") - corp_secret = self.config.get("corp_secret") - agent_id = self.config.get("agent_id") + async with self._wecom_client_lock: + corp_id = self.config.get("corp_id") + corp_secret = self.config.get("corp_secret") + agent_id = self.config.get("agent_id") - if not all([corp_id, corp_secret, agent_id]): - raise ChannelException("WeCom mode requires corp_id, corp_secret, and agent_id") + if not all([corp_id, corp_secret, agent_id]): + raise ChannelException("WeCom mode requires corp_id, corp_secret, and agent_id") - self._wecom_client = WeComClient(self._http_client, self.config) - await self._wecom_client.get_access_token() - logger.info("[WeChat/WeCom] Access token obtained") + self._wecom_client = WeComClient(self._http_client, self.config) + + token_data = await self.state_get("access_token", namespace="auth") + if token_data and isinstance(token_data, dict): + import time as _time + + token_str = token_data.get("access_token") + expires_at = token_data.get("expires_at", 0) + if token_str and expires_at > _time.time() + 300: + self._wecom_client.set_token(token_str, expires_at) + logger.info("[WeChat/WeCom] Token restored from state_store") + + await self._wecom_client.get_access_token() + logger.info("[WeChat/WeCom] Access token obtained") + + import time as _time + + await self.state_set( + "access_token", + { + "access_token": self._wecom_client._access_token, + "expires_at": self._wecom_client._token_expires_at, + }, + namespace="auth", + ) self._monitor = WeComMonitor(self._http_client, self.config, self._handle_message) await self._monitor.start(lambda: self._wecom_client.get_access_token()) @@ -638,20 +862,54 @@ class WeChatAdapter(BaseChannelAdapter): await self._auto_join_groups() async def _send_wecom(self, response: ChannelResponse, disable_notification: bool = False) -> DeliveryResult: + if self._wecom_client is None: + return DeliveryResult(success=False, error="WeCom client not initialized") reply_to_user = None if response.reply_to_message_id: reply_to_user = response.metadata.get( "reply_to_channel_user_id", response.metadata.get("sender_wxid"), ) - payload = build_wecom_text_payload( + + use_markdown = ( + self.config.get("enable_markdown", True) + and response.metadata.get("markdown", False) + ) + + if use_markdown: + payload = build_wecom_markdown_payload( + agent_id=self.config["agent_id"], + to_user=response.identity.channel_user_id, + content=response.content, + chat_type=response.metadata.get("chat_type", "direct"), + reply_to_msg_id=response.reply_to_message_id, + reply_to_user=reply_to_user, + safe=1 if disable_notification else 0, + ) + else: + payload = build_wecom_text_payload( + agent_id=self.config["agent_id"], + to_user=response.identity.channel_user_id, + content=response.content, + chat_type=response.metadata.get("chat_type", "direct"), + reply_to_msg_id=response.reply_to_message_id, + reply_to_user=reply_to_user, + safe=1 if disable_notification else 0, + ) + return await send_wecom_message(self._wecom_client, self._http_client, payload) + + async def _send_wecom_template_card(self, response: ChannelResponse) -> DeliveryResult: + if self._wecom_client is None: + return DeliveryResult(success=False, error="WeCom client not initialized") + + card = response.metadata.get("template_card") + if not isinstance(card, TemplateCard): + return DeliveryResult(success=False, error="Invalid template_card in response metadata") + + payload = build_wecom_template_card_payload( agent_id=self.config["agent_id"], to_user=response.identity.channel_user_id, - content=response.content, - chat_type=response.metadata.get("chat_type", "direct"), - reply_to_msg_id=response.reply_to_message_id, - reply_to_user=reply_to_user, - safe=1 if disable_notification else 0, + card=card, ) return await send_wecom_message(self._wecom_client, self._http_client, payload) @@ -709,9 +967,31 @@ class WeChatAdapter(BaseChannelAdapter): raise ChannelException("MP mode requires app_id and app_secret") self._mp_client = MPClient(self._http_client, self.config) + + token_data = await self.state_get("access_token", namespace="auth") + if token_data and isinstance(token_data, dict): + import time as _time + + token_str = token_data.get("access_token") + expires_at = token_data.get("expires_at", 0) + if token_str and expires_at > _time.time() + 300: + self._mp_client.set_token(token_str, expires_at) + logger.info("[WeChat/MP] Token restored from state_store") + await self._mp_client.get_access_token() logger.info("[WeChat/MP] Access token obtained") + import time as _time + + await self.state_set( + "access_token", + { + "access_token": self._mp_client._access_token, + "expires_at": self._mp_client._token_expires_at, + }, + namespace="auth", + ) + webhook_url = self.config.get("webhook_url") if webhook_url: logger.info(f"[WeChat/MP] Webhook mode ready: {webhook_url}") @@ -727,6 +1007,13 @@ class WeChatAdapter(BaseChannelAdapter): "reply_to_channel_user_id", response.metadata.get("sender_wxid"), ) + + if self.config.get("passive_reply_mode", "auto") != "disabled": + await self._set_passive_cache( + response.identity.channel_user_id, + response.content, + ) + return await send_mp_custom_message( self._mp_client, self._http_client, @@ -736,6 +1023,48 @@ class WeChatAdapter(BaseChannelAdapter): reply_to_user=reply_to_user, ) + async def _set_passive_cache(self, openid: str, content: str, ttl: int = 30) -> None: + key = f"passive_reply:{openid}" + value = {"content": content, "expires_at": utc_now_naive().timestamp() + ttl} + await self.state_set(key, value, namespace="passive_reply") + + async def _pop_passive_cache(self, openid: str) -> dict | None: + key = f"passive_reply:{openid}" + data = await self.state_get(key, namespace="passive_reply") + if data: + await self.state_delete(key, namespace="passive_reply") + return data + + async def try_passive_reply(self, webhook_body: dict) -> str | None: + if self._mode != "mp": + return None + + passive_mode = self.config.get("passive_reply_mode", "auto") + + from_user = webhook_body.get("FromUserName", "") + to_user = webhook_body.get("ToUserName", "") + + if passive_mode == "disabled": + return None + + if passive_mode == "transfer_cs": + from .mp import build_transfer_customer_service_reply + + return build_transfer_customer_service_reply(from_user, to_user) + + if passive_mode == "auto": + cached = await self._pop_passive_cache(from_user) + if cached: + from .mp import build_text_passive_reply + + return build_text_passive_reply(from_user, to_user, cached["content"]) + + from .mp import build_empty_passive_reply + + return build_empty_passive_reply() + + return None + def _normalize_mp_message(self, payload: dict) -> ChannelMessage: from_user = payload.get("FromUserName", "") msg_id = str(payload.get("MsgId", "")) @@ -990,15 +1319,22 @@ class WeChatAdapter(BaseChannelAdapter): return await send_wecom_message(self._wecom_client, self._http_client, payload) async def _send_mp_media(self, media_type: str, data: Any, chat_id: str) -> DeliveryResult: - from .mp.send import send_mp_image + from .mp.send import send_mp_image, send_mp_video, send_mp_voice - if media_type != "image": - return DeliveryResult(success=False, error=f"MP only supports image media, got: {media_type}") + if media_type not in ("image", "voice", "video"): + return DeliveryResult(success=False, error=f"MP only supports image/voice/video media, got: {media_type}") if not isinstance(data, bytes): return DeliveryResult(success=False, error="send_media expects raw bytes for MP mode") - return await send_mp_image(self._mp_client, self._http_client, chat_id, data) + if media_type == "image": + return await send_mp_image(self._mp_client, self._http_client, chat_id, data) + elif media_type == "voice": + return await send_mp_voice(self._mp_client, self._http_client, chat_id, data) + elif media_type == "video": + return await send_mp_video(self._mp_client, self._http_client, chat_id, data) + + return DeliveryResult(success=False, error=f"Unsupported MP media type: {media_type}") async def _send_bridge_media( self, media_type: str, data: Any, chat_id: str, disable_notification: bool = False @@ -1036,6 +1372,30 @@ class WeChatAdapter(BaseChannelAdapter): async def send_reaction(self, chat_id: str, msg_id: str, emoji: str) -> DeliveryResult: return DeliveryResult(success=False, error="WeChat does not support reactions") + def supports_action(self, action: str) -> bool: + return self._message_actions.is_supported(action) + + def resolve_execution_mode(self, action: str) -> str: + return "direct" + + async def handle_action(self, ctx) -> DeliveryResult: + handler = self._message_actions.get_handler(ctx.action) + if handler is None: + return DeliveryResult(success=False, error=f"Action '{ctx.action}' not supported by WeChat") + try: + result = await handler(**ctx.args, chat_id=ctx.chat_id, msg_id=ctx.msg_id) + if isinstance(result, DeliveryResult): + return result + if isinstance(result, dict): + return DeliveryResult( + success=result.get("success", False), + message_id=result.get("message_id"), + error=result.get("error"), + ) + return DeliveryResult(success=True) + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + async def get_user_info(self, channel_user_id: str) -> dict[str, Any]: self._check_ban_cooldown() if self._banned: @@ -1159,7 +1519,7 @@ class WeChatAdapter(BaseChannelAdapter): @property def is_banned(self) -> bool: - return self._banned + return self._banned or self._ban_permanent @property def banned_reason(self) -> str | None: @@ -1391,9 +1751,20 @@ class WeChatAdapter(BaseChannelAdapter): "banned": self._banned, "ban_permanent": self._ban_permanent, "ban_attempts": self._ban_attempts, + "dedup_ttl": self._dedup_ttl, }, ) + def export_dedup_state(self) -> dict[str, Any]: + return { + "dedup_ttl": self._dedup_ttl, + "max_entries": self._max_dedup_entries, + "backend": "DedupPolicy (TTLCache + optional Redis)", + } + + def import_dedup_state(self, state: dict[str, Any]) -> None: + logger.debug("[WeChat] Dedup state import skipped: DedupPolicy handles persistence internally") + @property def selector(self) -> WeChatSelectorAdapter: return self._selector @@ -1477,28 +1848,32 @@ class WeChatAdapter(BaseChannelAdapter): if self._mode == "wecom": token = self.config.get("token", "") if not token: - return True + logger.warning("[WeChat] Webhook signature verification skipped: token not configured for wecom mode") + return False signature = headers.get("x-wx-signature", headers.get("signature", "")) timestamp = headers.get("x-wx-timestamp", headers.get("timestamp", "")) nonce = headers.get("x-wx-nonce", headers.get("nonce", "")) if not all([signature, timestamp, nonce]): - return True + logger.warning("[WeChat] Webhook signature verification failed: missing signature/timestamp/nonce") + return False return wecom_verify_sig(token, timestamp, nonce, signature) if self._mode == "mp": token = self.config.get("token", "") if not token: - return True + logger.warning("[WeChat] Webhook signature verification skipped: token not configured for mp mode") + return False signature = headers.get("x-wx-signature", headers.get("signature", "")) timestamp = headers.get("x-wx-timestamp", headers.get("timestamp", "")) nonce = headers.get("x-wx-nonce", headers.get("nonce", "")) if not all([signature, timestamp, nonce]): - return True + logger.warning("[WeChat] Webhook signature verification failed: missing signature/timestamp/nonce") + return False return mp_verify_sig(token, timestamp, nonce, signature) diff --git a/backend/package/yuxi/channels/adapters/wechat/agent_prompt.py b/backend/package/yuxi/channels/adapters/wechat/agent_prompt.py index 405266fd..886d5a30 100644 --- a/backend/package/yuxi/channels/adapters/wechat/agent_prompt.py +++ b/backend/package/yuxi/channels/adapters/wechat/agent_prompt.py @@ -2,7 +2,6 @@ from __future__ import annotations from typing import Any - PROMPT_PREFIXES = { "wecom": "你是一个企业微信 AI 助手，通过企业微信与用户沟通。请保持专业、简洁，使用中文回复。", "mp": "你是一个微信公众号 AI 助手，通过公众号消息与用户沟通。请保持友好、专业，使用中文回复。", diff --git a/backend/package/yuxi/channels/adapters/wechat/auth_adapter.py b/backend/package/yuxi/channels/adapters/wechat/auth_adapter.py index 273863a3..c3946eeb 100644 --- a/backend/package/yuxi/channels/adapters/wechat/auth_adapter.py +++ b/backend/package/yuxi/channels/adapters/wechat/auth_adapter.py @@ -4,12 +4,115 @@ from typing import Any class WeChatAuthAdapter: - @staticmethod - def get_dm_exposure(config: dict[str, Any]) -> str: + def get_dm_exposure(self, config: dict[str, Any]) -> str: return config.get("dm_policy", "pairing") - @staticmethod - def get_authorization_url(config: dict[str, Any]) -> str | None: + def get_authorization_url(self, config: dict[str, Any]) -> str | None: if config.get("wecom_auth_url"): return config["wecom_auth_url"] return None + + def build_wecom_oauth_url( + self, + corp_id: str, + redirect_uri: str, + state: str = "", + scope: str = "snsapi_base", + ) -> str: + params = { + "appid": corp_id, + "redirect_uri": redirect_uri, + "response_type": "code", + "scope": scope, + } + query = "&".join(f"{k}={v}" for k, v in params.items()) + url = f"https://open.weixin.qq.com/connect/oauth2/authorize?{query}" + if state: + url += f"&state={state}" + url += "#wechat_redirect" + return url + + def build_mp_oauth_url( + self, + app_id: str, + redirect_uri: str, + state: str = "", + scope: str = "snsapi_userinfo", + ) -> str: + params = { + "appid": app_id, + "redirect_uri": redirect_uri, + "response_type": "code", + "scope": scope, + } + query = "&".join(f"{k}={v}" for k, v in params.items()) + url = f"https://open.weixin.qq.com/connect/oauth2/authorize?{query}" + if state: + url += f"&state={state}" + url += "#wechat_redirect" + return url + + async def exchange_wecom_code(self, code: str, corp_id: str, corp_secret: str) -> dict[str, Any]: + import httpx + + access_token_url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken" + async with httpx.AsyncClient() as client: + token_resp = await client.get( + access_token_url, + params={"corpid": corp_id, "corpsecret": corp_secret}, + ) + token_data = token_resp.json() + access_token = token_data.get("access_token", "") + + if not access_token: + return {"success": False, "error": f"Failed to get access token: {token_data}"} + + userinfo_url = "https://qyapi.weixin.qq.com/cgi-bin/auth/getuserinfo" + user_resp = await client.get( + userinfo_url, + params={"access_token": access_token, "code": code}, + ) + user_data = user_resp.json() + + if user_data.get("errcode") == 0: + return {"success": True, "user_info": user_data, "access_token": access_token} + return {"success": False, "error": str(user_data)} + + async def exchange_mp_code(self, code: str, app_id: str, app_secret: str) -> dict[str, Any]: + import httpx + + url = "https://api.weixin.qq.com/sns/oauth2/access_token" + async with httpx.AsyncClient() as client: + resp = await client.get( + url, + params={ + "appid": app_id, + "secret": app_secret, + "code": code, + "grant_type": "authorization_code", + }, + ) + data = resp.json() + + if "access_token" in data and "openid" in data: + userinfo_url = "https://api.weixin.qq.com/sns/userinfo" + user_resp = await client.get( + userinfo_url, + params={ + "access_token": data["access_token"], + "openid": data["openid"], + "lang": "zh_CN", + }, + ) + user_data = user_resp.json() + return { + "success": True, + "user_info": user_data, + "access_token": data["access_token"], + "refresh_token": data.get("refresh_token"), + "openid": data["openid"], + } + return {"success": False, "error": str(data)} + + def get_oauth_enabled(self, config: dict[str, Any]) -> bool: + return bool(config.get("wecom_oauth_enabled", False) or config.get("mp_oauth_enabled", False)) diff --git a/backend/package/yuxi/channels/adapters/wechat/bridge.py b/backend/package/yuxi/channels/adapters/wechat/bridge.py index 0b951b7c..3edfa4a8 100644 --- a/backend/package/yuxi/channels/adapters/wechat/bridge.py +++ b/backend/package/yuxi/channels/adapters/wechat/bridge.py @@ -5,8 +5,8 @@ from typing import Any import httpx -from yuxi.channels.models import DeliveryResult from yuxi.channels.adapters.wechat.format import truncate_text +from yuxi.channels.models import DeliveryResult class BridgeClient: diff --git a/backend/package/yuxi/channels/adapters/wechat/capabilities.py b/backend/package/yuxi/channels/adapters/wechat/capabilities.py index ecb86dd3..47b3b559 100644 --- a/backend/package/yuxi/channels/adapters/wechat/capabilities.py +++ b/backend/package/yuxi/channels/adapters/wechat/capabilities.py @@ -15,7 +15,7 @@ WECHAT_META = ChannelMeta( aliases=["weixin", "WeChat"], detail_label="WeChat 微信", system_image="wechat", - markdown_capable=False, + markdown_capable=True, exposure="configured", show_configured=True, show_in_setup=True, diff --git a/backend/package/yuxi/channels/adapters/wechat/config_schema.py b/backend/package/yuxi/channels/adapters/wechat/config_schema.py index 4c757463..6c431c88 100644 --- a/backend/package/yuxi/channels/adapters/wechat/config_schema.py +++ b/backend/package/yuxi/channels/adapters/wechat/config_schema.py @@ -51,6 +51,17 @@ WECHAT_CONFIG_SCHEMA: dict[str, Any] = { "minItems": 1, "maxItems": 5, }, + "passive_reply_mode": { + "type": "string", + "enum": ["auto", "transfer_cs", "disabled"], + "default": "auto", + "description": ( + "MP 模式被动回复策略：" + "auto - 自动尝试被动回复（优先缓存 → 空回复 + 客服消息兜底）；" + "transfer_cs - 始终转发到客服；" + "disabled - 仅使用客服消息（当前行为）" + ), + }, }, } diff --git a/backend/package/yuxi/channels/adapters/wechat/config_ui_hints.py b/backend/package/yuxi/channels/adapters/wechat/config_ui_hints.py index 520cfb27..88826208 100644 --- a/backend/package/yuxi/channels/adapters/wechat/config_ui_hints.py +++ b/backend/package/yuxi/channels/adapters/wechat/config_ui_hints.py @@ -133,6 +133,22 @@ def get_config_ui_hints() -> dict[str, Any]: "min": 0.5, "max": 10.0, }, + "dedup_ttl_seconds": { + "label": "去重 TTL (秒)", + "description": "消息去重的有效时间窗口，相同 user+content 在此时间内的重复消息将被过滤", + "type": "number", + "default": 1.0, + "min": 0.1, + "max": 60.0, + }, + "max_dedup_entries": { + "label": "最大去重条目", + "description": "去重缓存的条目上限，超出后自动淘汰旧条目", + "type": "number", + "default": 10000, + "min": 100, + "max": 1000000, + }, "heartbeat_interval": { "label": "心跳间隔 (秒)", "description": "企业微信模式下的 token 续期间隔", diff --git a/backend/package/yuxi/channels/adapters/wechat/debounce.py b/backend/package/yuxi/channels/adapters/wechat/debounce.py index 32a93269..26822ff2 100644 --- a/backend/package/yuxi/channels/adapters/wechat/debounce.py +++ b/backend/package/yuxi/channels/adapters/wechat/debounce.py @@ -1,7 +1,7 @@ from __future__ import annotations import asyncio -from collections.abc import Callable, Awaitable +from collections.abc import Awaitable, Callable from typing import Any from yuxi.channels.models import ChannelMessage diff --git a/backend/package/yuxi/channels/adapters/wechat/diagnose.py b/backend/package/yuxi/channels/adapters/wechat/diagnose.py index 9ccec697..ccd062b9 100644 --- a/backend/package/yuxi/channels/adapters/wechat/diagnose.py +++ b/backend/package/yuxi/channels/adapters/wechat/diagnose.py @@ -1,10 +1,9 @@ from __future__ import annotations -from typing import Any, TYPE_CHECKING +from typing import TYPE_CHECKING, Any import httpx - if TYPE_CHECKING: from .adapter import WeChatAdapter diff --git a/backend/package/yuxi/channels/adapters/wechat/directory_contract.py b/backend/package/yuxi/channels/adapters/wechat/directory_contract.py index 6d7a3987..5799a528 100644 --- a/backend/package/yuxi/channels/adapters/wechat/directory_contract.py +++ b/backend/package/yuxi/channels/adapters/wechat/directory_contract.py @@ -2,7 +2,6 @@ from __future__ import annotations from typing import Any - WECHAT_DIRECTORY_CONTRACT: dict[str, Any] = { "channel_id": "wechat", "listing_methods": { diff --git a/backend/package/yuxi/channels/adapters/wechat/errors.py b/backend/package/yuxi/channels/adapters/wechat/errors.py index ff734e3f..ce4a7e56 100644 --- a/backend/package/yuxi/channels/adapters/wechat/errors.py +++ b/backend/package/yuxi/channels/adapters/wechat/errors.py @@ -2,7 +2,6 @@ from __future__ import annotations from typing import Any - WECOM_ERROR_CODES: dict[int, str] = { -1: "系统繁忙，请稍后重试", 0: "请求成功", diff --git a/backend/package/yuxi/channels/adapters/wechat/event_subscribe_adapter.py b/backend/package/yuxi/channels/adapters/wechat/event_subscribe_adapter.py index 810eef6c..a3aa5c61 100644 --- a/backend/package/yuxi/channels/adapters/wechat/event_subscribe_adapter.py +++ b/backend/package/yuxi/channels/adapters/wechat/event_subscribe_adapter.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.utils.logging_config import logger diff --git a/backend/package/yuxi/channels/adapters/wechat/format.py b/backend/package/yuxi/channels/adapters/wechat/format.py index 598586bc..76883c94 100644 --- a/backend/package/yuxi/channels/adapters/wechat/format.py +++ b/backend/package/yuxi/channels/adapters/wechat/format.py @@ -1,14 +1,33 @@ from __future__ import annotations +import re + from yuxi.channels.models import MessageType TRUNCATION_MARKER = "\n...(内容过长已截断)" +_MARKDOWN_BOUNDARY_PATTERN = re.compile( + r'(\n\n|\n(?=#{1,6}\s)|\n(?=>\s)|\n(?=-\s)|\n(?=\d+\.\s))' +) + def truncate_text(text: str, limit: int) -> str: if len(text) <= limit: return text - return text[: limit - len(TRUNCATION_MARKER)] + TRUNCATION_MARKER + cut = max(0, limit - len(TRUNCATION_MARKER)) + return text[:cut] + TRUNCATION_MARKER + + +def truncate_markdown(text: str, limit: int) -> str: + if len(text) <= limit: + return text + cut = max(0, limit - len(TRUNCATION_MARKER)) + candidates = list(_MARKDOWN_BOUNDARY_PATTERN.finditer(text[: cut + 1])) + if candidates: + boundary = candidates[-1].start() + if boundary > cut * 0.7: + return text[:boundary] + TRUNCATION_MARKER + return text[:cut] + TRUNCATION_MARKER def map_wecom_msg_type(msg_type: str) -> MessageType: diff --git a/backend/package/yuxi/channels/adapters/wechat/heartbeat_adapter.py b/backend/package/yuxi/channels/adapters/wechat/heartbeat_adapter.py index 21cd31ae..39c07ef2 100644 --- a/backend/package/yuxi/channels/adapters/wechat/heartbeat_adapter.py +++ b/backend/package/yuxi/channels/adapters/wechat/heartbeat_adapter.py @@ -1,8 +1,10 @@ from __future__ import annotations -from typing import Any, TYPE_CHECKING +from typing import TYPE_CHECKING, Any if TYPE_CHECKING: + from yuxi.channels.models import DeliveryResult + from .adapter import WeChatAdapter @@ -11,14 +13,44 @@ class WeChatHeartbeatAdapter: self._typing_enabled = config.get("typing_indicator", False) self._heartbeat_interval = config.get("heartbeat_interval", 300.0) - async def send_typing_indicator(self, send_fn, chat_id: str) -> None: - if not self._typing_enabled: + @property + def typing_enabled(self) -> bool: + return self._typing_enabled + + async def send_typing_indicator(self, send_fn, chat_id: str, active: bool = True) -> None: + if not self._typing_enabled or not active: return try: await send_fn(chat_id, "...") except Exception: pass + async def send_typing_via_response( + self, + send_fn, + chat_id: str, + active: bool = True, + ) -> DeliveryResult | None: + if not self._typing_enabled or not active: + return None + try: + from yuxi.channels.models import ( + ChannelIdentity, + ChannelResponse, + ChannelType, + ) + + identity = ChannelIdentity( + channel_id="wechat", + channel_type=ChannelType.WECHAT, + channel_user_id="", + channel_chat_id=chat_id, + ) + response = ChannelResponse(identity=identity, content="...") + return await send_fn(response) + except Exception: + return None + async def heartbeat_check(self, adapter: WeChatAdapter) -> dict[str, Any]: import time diff --git a/backend/package/yuxi/channels/adapters/wechat/message_actions.py b/backend/package/yuxi/channels/adapters/wechat/message_actions.py index 63a4f32b..0e5af976 100644 --- a/backend/package/yuxi/channels/adapters/wechat/message_actions.py +++ b/backend/package/yuxi/channels/adapters/wechat/message_actions.py @@ -1,7 +1,7 @@ from __future__ import annotations from collections.abc import Awaitable, Callable -from typing import Any, TYPE_CHECKING +from typing import TYPE_CHECKING, Any if TYPE_CHECKING: from yuxi.channels.message_actions import ActionStatus, MessageAction @@ -63,18 +63,152 @@ WECHAT_MESSAGE_ACTIONS: dict[str, dict[str, Any]] = { "search": {"status": "unsupported", "reason": "WeChat doesn't support message search API"}, "set-profile": {"status": "unsupported", "reason": "WeChat doesn't support profile API"}, "set-presence": {"status": "unsupported", "reason": "WeChat doesn't support presence API"}, + "template_card": {"status": "implemented", "impl": "template_card.py"}, } class WeChatMessageActionAdapter: + def __init__(self, adapter: Any = None): + self._adapter = adapter + @staticmethod def is_supported(action_name: str) -> bool: info = WECHAT_MESSAGE_ACTIONS.get(action_name, {}) return info.get("status") == "implemented" - @staticmethod - def get_handler(action_name: str) -> Callable[..., Awaitable[Any]] | None: - return None + def get_handler(self, action_name: str) -> Callable[..., Awaitable[Any]] | None: + if self._adapter is None: + return None + return self._build_handler(action_name) + + def _build_handler(self, action_name: str) -> Callable[..., Awaitable[Any]] | None: + adapter = self._adapter + + async def _send(**kwargs: Any) -> Any: + chat_id = kwargs.get("chat_id", "") + content = kwargs.get("content", "") + if not chat_id or not content: + return {"success": False, "error": "missing chat_id or content"} + from yuxi.channels.models import ChannelIdentity, ChannelResponse + + identity = ChannelIdentity( + channel_id=adapter.channel_id, + channel_type=adapter.channel_type, + channel_chat_id=chat_id, + channel_user_id=kwargs.get("channel_user_id", ""), + ) + response = ChannelResponse(identity=identity, content=content) + result = await adapter.send(response) + return {"success": result.success, "error": result.error} + + async def _reply(**kwargs: Any) -> Any: + chat_id = kwargs.get("chat_id", "") + content = kwargs.get("content", "") + reply_to_msg_id = kwargs.get("reply_to_msg_id", "") + if not chat_id or not content: + return {"success": False, "error": "missing chat_id or content"} + from yuxi.channels.models import ChannelIdentity, ChannelResponse + + identity = ChannelIdentity( + channel_id=adapter.channel_id, + channel_type=adapter.channel_type, + channel_chat_id=chat_id, + channel_user_id=kwargs.get("channel_user_id", ""), + ) + response = ChannelResponse( + identity=identity, + content=content, + reply_to_message_id=reply_to_msg_id, + metadata={ + "reply_to_channel_user_id": kwargs.get("reply_to_channel_user_id", ""), + "sender_wxid": kwargs.get("sender_wxid"), + "chat_type": kwargs.get("chat_type", "direct"), + }, + ) + result = await adapter.send(response) + return {"success": result.success, "error": result.error} + + async def _send_attachment(**kwargs: Any) -> Any: + chat_id = kwargs.get("chat_id", "") + media_type = kwargs.get("media_type", "file") + data = kwargs.get("data") + if not chat_id or data is None: + return {"success": False, "error": "missing chat_id or data"} + result = await adapter.send_media(chat_id, media_type, data) + return {"success": result.success, "error": result.error} + + async def _download_file(**kwargs: Any) -> Any: + file_id = kwargs.get("file_id", "") + if not file_id: + return {"success": False, "error": "missing file_id"} + try: + data = await adapter.download_media(file_id) + return {"success": True, "data": data} + except Exception as e: + return {"success": False, "error": str(e)} + + async def _upload_file(**kwargs: Any) -> Any: + chat_id = kwargs.get("chat_id", "") + data = kwargs.get("data") + media_type = kwargs.get("media_type", "file") + if not chat_id or data is None: + return {"success": False, "error": "missing chat_id or data"} + result = await adapter.send_media(chat_id, media_type, data) + return {"success": result.success, "error": result.error} + + async def _read(**kwargs: Any) -> Any: + msg_id = kwargs.get("msg_id", "") + if not msg_id: + return {"success": False, "error": "missing msg_id"} + result = await adapter.read_message(msg_id) + return result + + async def _rename_group(**kwargs: Any) -> Any: + chat_id = kwargs.get("chat_id", "") + name = kwargs.get("name", "") + if not chat_id or not name: + return {"success": False, "error": "missing chat_id or name"} + if adapter._mode != "wecom" or not adapter._wecom_client or not adapter._http_client: + return {"success": False, "error": "rename group only supported in WeCom mode"} + result = await adapter._group_admin.rename_group(adapter._wecom_client, adapter._http_client, chat_id, name) + return {"success": result.success, "error": result.error} + + async def _add_participant(**kwargs: Any) -> Any: + chat_id = kwargs.get("chat_id", "") + user_id = kwargs.get("user_id", "") + if not chat_id or not user_id: + return {"success": False, "error": "missing chat_id or user_id"} + if adapter._mode != "wecom" or not adapter._wecom_client or not adapter._http_client: + return {"success": False, "error": "add participant only supported in WeCom mode"} + result = await adapter._group_admin.add_participant( + adapter._wecom_client, adapter._http_client, chat_id, user_id + ) + return {"success": result.success, "error": result.error} + + async def _remove_participant(**kwargs: Any) -> Any: + chat_id = kwargs.get("chat_id", "") + user_id = kwargs.get("user_id", "") + if not chat_id or not user_id: + return {"success": False, "error": "missing chat_id or user_id"} + if adapter._mode != "wecom" or not adapter._wecom_client or not adapter._http_client: + return {"success": False, "error": "remove participant only supported in WeCom mode"} + result = await adapter._group_admin.remove_participant( + adapter._wecom_client, adapter._http_client, chat_id, user_id + ) + return {"success": result.success, "error": result.error} + + handler_map: dict[str, Callable[..., Awaitable[Any]]] = { + "send": _send, + "reply": _reply, + "sendAttachment": _send_attachment, + "download-file": _download_file, + "upload-file": _upload_file, + "read": _read, + "renameGroup": _rename_group, + "addParticipant": _add_participant, + "removeParticipant": _remove_participant, + } + return handler_map.get(action_name) @staticmethod def list_supported_actions() -> list[str]: diff --git a/backend/package/yuxi/channels/adapters/wechat/monitoring_adapter.py b/backend/package/yuxi/channels/adapters/wechat/monitoring_adapter.py index cff142bd..ac6fcf5f 100644 --- a/backend/package/yuxi/channels/adapters/wechat/monitoring_adapter.py +++ b/backend/package/yuxi/channels/adapters/wechat/monitoring_adapter.py @@ -112,4 +112,4 @@ class WeChatMonitoringAdapter: return issues def reset_metrics(self) -> None: - self._metrics.clear() \ No newline at end of file + self._metrics.clear() diff --git a/backend/package/yuxi/channels/adapters/wechat/mp/__init__.py b/backend/package/yuxi/channels/adapters/wechat/mp/__init__.py index 952bbc81..a71322e1 100644 --- a/backend/package/yuxi/channels/adapters/wechat/mp/__init__.py +++ b/backend/package/yuxi/channels/adapters/wechat/mp/__init__.py @@ -7,6 +7,12 @@ from yuxi.channels.adapters.wechat.mp.crypto import ( verify_url_echostr, ) from yuxi.channels.adapters.wechat.mp.send import ( + build_empty_passive_reply, + build_image_passive_reply, + build_text_passive_reply, + build_transfer_customer_service_reply, + build_video_passive_reply, + build_voice_passive_reply, send_mp_custom_message, send_mp_template_message, ) @@ -18,4 +24,10 @@ __all__ = [ "verify_url_echostr", "send_mp_custom_message", "send_mp_template_message", + "build_text_passive_reply", + "build_image_passive_reply", + "build_voice_passive_reply", + "build_video_passive_reply", + "build_transfer_customer_service_reply", + "build_empty_passive_reply", ] diff --git a/backend/package/yuxi/channels/adapters/wechat/mp/client.py b/backend/package/yuxi/channels/adapters/wechat/mp/client.py index 600be06d..599df39f 100644 --- a/backend/package/yuxi/channels/adapters/wechat/mp/client.py +++ b/backend/package/yuxi/channels/adapters/wechat/mp/client.py @@ -58,6 +58,10 @@ class MPClient: self._access_token = None self._token_expires_at = 0.0 + def set_token(self, access_token: str, expires_at: float) -> None: + self._access_token = access_token + self._token_expires_at = expires_at + async def upload_temp_media(self, file_data: bytes, filename: str, media_type: str) -> str: token = await self.get_access_token() url = f"https://api.weixin.qq.com/cgi-bin/media/upload?access_token={token}&type={media_type}" diff --git a/backend/package/yuxi/channels/adapters/wechat/mp/send.py b/backend/package/yuxi/channels/adapters/wechat/mp/send.py index 51096c43..06989af2 100644 --- a/backend/package/yuxi/channels/adapters/wechat/mp/send.py +++ b/backend/package/yuxi/channels/adapters/wechat/mp/send.py @@ -1,16 +1,19 @@ from __future__ import annotations +import time from typing import Any import httpx -from yuxi.channels.models import DeliveryResult -from yuxi.channels.adapters.wechat.format import truncate_text from yuxi.channels.adapters.wechat.errors import is_token_expired, parse_mp_error +from yuxi.channels.adapters.wechat.format import truncate_text from yuxi.channels.adapters.wechat.retry import retry_with_backoff +from yuxi.channels.models import DeliveryResult from .client import MPClient +XML_HEADER = '' + async def send_mp_custom_message( client: MPClient, @@ -203,3 +206,105 @@ async def send_mp_news( return await _post_with_token_retry(client, http_client, token, payload) except httpx.HTTPError as e: return DeliveryResult(success=False, error=str(e)) + + +def _escape_cdata(text: str) -> str: + return f"', ']]]]>')}]]>" + + +def build_text_passive_reply( + to_user: str, + from_user: str, + content: str, +) -> str: + timestamp = int(time.time()) + return ( + f'{XML_HEADER}\n' + f'\n' + f' {_escape_cdata(to_user)}\n' + f' {_escape_cdata(from_user)}\n' + f' {timestamp}\n' + f' {_escape_cdata("text")}\n' + f' {_escape_cdata(content[:2048])}\n' + f'' + ) + + +def build_image_passive_reply( + to_user: str, + from_user: str, + media_id: str, +) -> str: + timestamp = int(time.time()) + return ( + f'{XML_HEADER}\n' + f'\n' + f' {_escape_cdata(to_user)}\n' + f' {_escape_cdata(from_user)}\n' + f' {timestamp}\n' + f' {_escape_cdata("image")}\n' + f' {_escape_cdata(media_id)}\n' + f'' + ) + + +def build_voice_passive_reply( + to_user: str, + from_user: str, + media_id: str, +) -> str: + timestamp = int(time.time()) + return ( + f'{XML_HEADER}\n' + f'\n' + f' {_escape_cdata(to_user)}\n' + f' {_escape_cdata(from_user)}\n' + f' {timestamp}\n' + f' {_escape_cdata("voice")}\n' + f' {_escape_cdata(media_id)}\n' + f'' + ) + + +def build_video_passive_reply( + to_user: str, + from_user: str, + media_id: str, + title: str = "", + description: str = "", +) -> str: + timestamp = int(time.time()) + return ( + f'{XML_HEADER}\n' + f'\n' + f' {_escape_cdata(to_user)}\n' + f' {_escape_cdata(from_user)}\n' + f' {timestamp}\n' + f' {_escape_cdata("video")}\n' + f' \n' + f'' + ) + + +def build_transfer_customer_service_reply( + to_user: str, + from_user: str, +) -> str: + timestamp = int(time.time()) + return ( + f'{XML_HEADER}\n' + f'\n' + f' {_escape_cdata(to_user)}\n' + f' {_escape_cdata(from_user)}\n' + f' {timestamp}\n' + f' {_escape_cdata("transfer_customer_service")}\n' + f'' + ) + + +def build_empty_passive_reply() -> str: + return "" diff --git a/backend/package/yuxi/channels/adapters/wechat/pairing.py b/backend/package/yuxi/channels/adapters/wechat/pairing.py index 055b46f5..7322f805 100644 --- a/backend/package/yuxi/channels/adapters/wechat/pairing.py +++ b/backend/package/yuxi/channels/adapters/wechat/pairing.py @@ -3,7 +3,7 @@ from __future__ import annotations import json import secrets import time -from collections.abc import Callable, Awaitable +from collections.abc import Awaitable, Callable from dataclasses import dataclass from pathlib import Path from typing import Any diff --git a/backend/package/yuxi/channels/adapters/wechat/polling_lease.py b/backend/package/yuxi/channels/adapters/wechat/polling_lease.py index a8fcd379..d33ae9c2 100644 --- a/backend/package/yuxi/channels/adapters/wechat/polling_lease.py +++ b/backend/package/yuxi/channels/adapters/wechat/polling_lease.py @@ -64,15 +64,10 @@ class PollingLease: px=int(self._lease_ttl * 1000), ) if not acquired: - logger.warning( - f"[PollingLease/{self._channel_id}] Lease held by another instance" - ) + logger.warning(f"[PollingLease/{self._channel_id}] Lease held by another instance") return False except Exception as e: - logger.warning( - f"[PollingLease/{self._channel_id}] Redis unavailable, " - f"falling back to local mode: {e}" - ) + logger.warning(f"[PollingLease/{self._channel_id}] Redis unavailable, falling back to local mode: {e}") self._acquired_at = time.monotonic() self._last_renew = self._acquired_at @@ -108,9 +103,7 @@ class PollingLease: int(self._lease_ttl * 1000), ) except Exception as e: - logger.error( - f"[PollingLease/{self._channel_id}] Renew failed: {e}" - ) + logger.error(f"[PollingLease/{self._channel_id}] Renew failed: {e}") self._last_renew = time.monotonic() except asyncio.CancelledError: break @@ -144,4 +137,4 @@ class PollingLease: "acquired_at": self._acquired_at, "last_renew": self._last_renew, "lease_ttl": self._lease_ttl, - } \ No newline at end of file + } diff --git a/backend/package/yuxi/channels/adapters/wechat/setup_contract.py b/backend/package/yuxi/channels/adapters/wechat/setup_contract.py index 99a0306e..b3f00314 100644 --- a/backend/package/yuxi/channels/adapters/wechat/setup_contract.py +++ b/backend/package/yuxi/channels/adapters/wechat/setup_contract.py @@ -2,7 +2,6 @@ from __future__ import annotations from typing import Any - WECHAT_SETUP_CONTRACT: dict[str, Any] = { "channel_id": "wechat", "setup_steps": [ diff --git a/backend/package/yuxi/channels/adapters/wechat/setup_wizard.py b/backend/package/yuxi/channels/adapters/wechat/setup_wizard.py index 1e53d72b..4eeaaee0 100644 --- a/backend/package/yuxi/channels/adapters/wechat/setup_wizard.py +++ b/backend/package/yuxi/channels/adapters/wechat/setup_wizard.py @@ -3,6 +3,9 @@ from __future__ import annotations from enum import Enum from typing import Any +import httpx + +from yuxi.channels.adapters.wechat.probe import probe_bridge, probe_mp, probe_wecom from yuxi.utils.logging_config import logger @@ -86,13 +89,69 @@ class WeChatSetupWizard: else: valid = False - self._current_step = WizardStep.WEBHOOK if valid else WizardStep.CREDENTIALS - return { - "success": valid, - "mode": self._selected_mode, - "message": "Credentials validated" if valid else "Invalid credentials", - "next_step": "webhook" if valid else "credentials", - } + if not valid: + self._current_step = WizardStep.CREDENTIALS + return { + "success": False, + "mode": self._selected_mode, + "error": "Invalid credentials", + "next_step": "credentials", + } + + if http_client_factory is None: + proxy = self._config_snapshot.get("proxy") + timeout_s = self._config_snapshot.get("timeout_seconds", 15.0) + async with httpx.AsyncClient(proxy=proxy, timeout=httpx.Timeout(timeout_s)) as http_client: + return await self._probe_connection(http_client) + else: + http_client = http_client_factory() + return await self._probe_connection(http_client) + + async def _probe_connection(self, http_client: httpx.AsyncClient) -> dict[str, Any]: + try: + status = None + if self._selected_mode == "wecom": + status = await probe_wecom(http_client, self._config_snapshot) + elif self._selected_mode == "mp": + status = await probe_mp(http_client, self._config_snapshot) + elif self._selected_mode == "personal": + bridge_url = self._config_snapshot.get("bridge_url", "") + status = await probe_bridge(http_client, bridge_url) + + if status is None: + self._current_step = WizardStep.CREDENTIALS + return { + "success": False, + "error": "Unknown probe mode", + "next_step": "credentials", + } + + if status.status == "healthy": + self._current_step = WizardStep.WEBHOOK + return { + "success": True, + "mode": self._selected_mode, + "message": f"{self._selected_mode} API 连接验证成功", + "metadata": status.metadata, + "next_step": "webhook", + } + + self._current_step = WizardStep.CREDENTIALS + return { + "success": False, + "mode": self._selected_mode, + "error": status.last_error or f"{self._selected_mode} API 连接失败", + "next_step": "credentials", + } + except Exception as e: + logger.error(f"[WeChat/SetupWizard] Probe failed for mode={self._selected_mode}: {e}") + self._current_step = WizardStep.CREDENTIALS + return { + "success": False, + "mode": self._selected_mode, + "error": str(e), + "next_step": "credentials", + } async def configure_webhook(self, webhook_url: str) -> dict[str, Any]: if not self._config_snapshot: diff --git a/backend/package/yuxi/channels/adapters/wechat/voice.py b/backend/package/yuxi/channels/adapters/wechat/voice.py index 8b6b02e6..0dc5e38c 100644 --- a/backend/package/yuxi/channels/adapters/wechat/voice.py +++ b/backend/package/yuxi/channels/adapters/wechat/voice.py @@ -74,9 +74,7 @@ def ensure_format( return voice_data if target_format not in VOICE_FORMATS: - raise ValueError( - f"Unsupported target format: {target_format}, must be one of {VOICE_FORMATS}" - ) + raise ValueError(f"Unsupported target format: {target_format}, must be one of {VOICE_FORMATS}") raise ValueError( f"Voice format conversion from {source_format} to {target_format} is not supported. " diff --git a/backend/package/yuxi/channels/adapters/wechat/wecom/__init__.py b/backend/package/yuxi/channels/adapters/wechat/wecom/__init__.py index 7d7316e2..b30ff16b 100644 --- a/backend/package/yuxi/channels/adapters/wechat/wecom/__init__.py +++ b/backend/package/yuxi/channels/adapters/wechat/wecom/__init__.py @@ -9,6 +9,7 @@ from yuxi.channels.adapters.wechat.wecom.crypto import ( ) from yuxi.channels.adapters.wechat.wecom.monitor import WeComMonitor from yuxi.channels.adapters.wechat.wecom.send import ( + build_wecom_markdown_payload, build_wecom_text_payload, send_wecom_message, ) @@ -20,6 +21,7 @@ __all__ = [ "verify_signature", "verify_url_signature", "WeComMonitor", + "build_wecom_markdown_payload", "build_wecom_text_payload", "send_wecom_message", ] diff --git a/backend/package/yuxi/channels/adapters/wechat/wecom/client.py b/backend/package/yuxi/channels/adapters/wechat/wecom/client.py index e21f288d..6a9ae5b3 100644 --- a/backend/package/yuxi/channels/adapters/wechat/wecom/client.py +++ b/backend/package/yuxi/channels/adapters/wechat/wecom/client.py @@ -58,6 +58,10 @@ class WeComClient: self._access_token = None self._token_expires_at = 0.0 + def set_token(self, access_token: str, expires_at: float) -> None: + self._access_token = access_token + self._token_expires_at = expires_at + async def upload_media(self, file_data: bytes, filename: str, media_type: str) -> str: token = await self.get_access_token() url = f"https://qyapi.weixin.qq.com/cgi-bin/media/upload?access_token={token}&type={media_type}" diff --git a/backend/package/yuxi/channels/adapters/wechat/wecom/crypto.py b/backend/package/yuxi/channels/adapters/wechat/wecom/crypto.py index fe749477..4a6f5d5d 100644 --- a/backend/package/yuxi/channels/adapters/wechat/wecom/crypto.py +++ b/backend/package/yuxi/channels/adapters/wechat/wecom/crypto.py @@ -2,12 +2,12 @@ from __future__ import annotations import base64 import hashlib -import struct +import os import socket -import time +import struct -from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from cryptography.hazmat.primitives import padding as sym_padding +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes def verify_signature(token: str, timestamp: str, nonce: str, signature: str) -> bool: @@ -44,7 +44,7 @@ def encrypt_message( ) -> str: key = base64.b64decode(encoding_aes_key + "=") - random_bytes = struct.pack("I", int(time.time())) + random_bytes = os.urandom(16) content_bytes = content.encode("utf-8") app_id_bytes = app_id.encode("utf-8") diff --git a/backend/package/yuxi/channels/adapters/wechat/wecom/send.py b/backend/package/yuxi/channels/adapters/wechat/wecom/send.py index f1c5e8b1..eb125bbc 100644 --- a/backend/package/yuxi/channels/adapters/wechat/wecom/send.py +++ b/backend/package/yuxi/channels/adapters/wechat/wecom/send.py @@ -4,11 +4,11 @@ from typing import Any import httpx +from yuxi.channels.adapters.wechat.errors import is_token_expired, parse_wecom_error +from yuxi.channels.adapters.wechat.format import truncate_markdown, truncate_text +from yuxi.channels.adapters.wechat.retry import retry_with_backoff from yuxi.channels.exceptions import ChannelRateLimitError from yuxi.channels.models import DeliveryResult -from yuxi.channels.adapters.wechat.format import truncate_text -from yuxi.channels.adapters.wechat.errors import is_token_expired, parse_wecom_error -from yuxi.channels.adapters.wechat.retry import retry_with_backoff from .client import WeComClient @@ -79,6 +79,31 @@ def build_wecom_text_payload( return payload +def build_wecom_markdown_payload( + agent_id: str, + to_user: str, + content: str, + chat_type: str = "direct", + reply_to_msg_id: str | None = None, + reply_to_user: str | None = None, + safe: int = 0, +) -> dict[str, Any]: + payload: dict[str, Any] = { + "msgtype": "markdown", + "agentid": agent_id, + "markdown": {"content": truncate_markdown(content, 2048)}, + "safe": safe, + } + payload["touser"] = to_user + if reply_to_msg_id and reply_to_user: + payload["_reply_to_msg_id"] = reply_to_msg_id + payload["_reply_to_user"] = reply_to_user + quoted_prefix = f"> 回复 @{reply_to_user}\n\n" + if len(quoted_prefix + content) <= 2048: + payload["markdown"]["content"] = quoted_prefix + content + return payload + + def build_wecom_image_payload(agent_id: str, to_user: str, media_id: str) -> dict[str, Any]: return { "touser": to_user, @@ -180,3 +205,21 @@ async def send_wecom_video( payload = build_wecom_video_payload(agent_id, to_user, media_id, title, description) return await send_wecom_message(client, http_client, payload) + + +def build_wecom_template_card_payload( + agent_id: str, + to_user: str, + card: "TemplateCard", + enable_id_trans: bool = False, + enable_duplicate_check: bool = False, +) -> dict[str, Any]: + from yuxi.channels.adapters.wechat.template_card_render import render_template_card_message + + return render_template_card_message( + to_user=to_user, + agent_id=agent_id, + card=card, + enable_id_trans=enable_id_trans, + enable_duplicate_check=enable_duplicate_check, + )