From 12bc82226ff8467a26607d8561dfc4027180be40 Mon Sep 17 00:00:00 2001 From: MuXinCG <202322130196@mail.sdu.edu.cn> Date: Mon, 16 Feb 2026 16:05:41 +0800 Subject: [PATCH] add slack and wechat --- EvoScientist/channels/__init__.py | 2 +- EvoScientist/channels/capabilities.py | 2 +- EvoScientist/channels/slack/__init__.py | 20 + EvoScientist/channels/slack/channel.py | 291 +++++++ EvoScientist/channels/slack/probe.py | 48 ++ EvoScientist/channels/slack/serve.py | 99 +++ EvoScientist/channels/wechat/__init__.py | 69 ++ EvoScientist/channels/wechat/channel.py | 805 ++++++++++++++++++ EvoScientist/channels/wechat/crypto.py | 187 ++++ EvoScientist/channels/wechat/probe.py | 72 ++ EvoScientist/channels/wechat/serve.py | 139 +++ EvoScientist/channels/wechat/verify_server.py | 175 ++++ EvoScientist/config/onboard.py | 25 + EvoScientist/config/settings.py | 2 +- tests/test_slack_channel.py | 87 ++ tests/test_wechat_channel.py | 458 ++++++++++ 16 files changed, 2478 insertions(+), 3 deletions(-) create mode 100644 EvoScientist/channels/slack/__init__.py create mode 100644 EvoScientist/channels/slack/channel.py create mode 100644 EvoScientist/channels/slack/probe.py create mode 100644 EvoScientist/channels/slack/serve.py create mode 100644 EvoScientist/channels/wechat/__init__.py create mode 100644 EvoScientist/channels/wechat/channel.py create mode 100644 EvoScientist/channels/wechat/crypto.py create mode 100644 EvoScientist/channels/wechat/probe.py create mode 100644 EvoScientist/channels/wechat/serve.py create mode 100644 EvoScientist/channels/wechat/verify_server.py create mode 100644 tests/test_slack_channel.py create mode 100644 tests/test_wechat_channel.py diff --git a/EvoScientist/channels/__init__.py b/EvoScientist/channels/__init__.py index 7085a47..c910120 100644 --- a/EvoScientist/channels/__init__.py +++ b/EvoScientist/channels/__init__.py @@ -1,7 +1,7 @@ """Communication channels for EvoScientist. This module provides an extensible interface for different messaging channels -(iMessage, Telegram, Discord) to communicate with the EvoScientist agent. +(iMessage, Telegram, Discord, Slack, WeChat) to communicate with the EvoScientist agent. """ from .base import Channel, RawIncoming, IncomingMessage, OutgoingMessage, chunk_text diff --git a/EvoScientist/channels/capabilities.py b/EvoScientist/channels/capabilities.py index 5060dd9..f0533b0 100644 --- a/EvoScientist/channels/capabilities.py +++ b/EvoScientist/channels/capabilities.py @@ -114,7 +114,7 @@ SLACK = ChannelCapabilities( streaming=False, threading=True, reactions=True, - typing=True, + typing=False, # no native typing API; workaround via post+delete "..." message media_send=True, media_receive=True, voice=False, # no distinct voice message type in Slack bot API diff --git a/EvoScientist/channels/slack/__init__.py b/EvoScientist/channels/slack/__init__.py new file mode 100644 index 0000000..9157573 --- /dev/null +++ b/EvoScientist/channels/slack/__init__.py @@ -0,0 +1,20 @@ +from .channel import SlackChannel, SlackConfig +from ..channel_manager import register_channel, _parse_csv + +__all__ = ["SlackChannel", "SlackConfig"] + + +def create_from_config(config) -> SlackChannel: + allowed = _parse_csv(config.slack_allowed_senders) + channels = _parse_csv(config.slack_allowed_channels) + proxy = config.slack_proxy if config.slack_proxy else None + return SlackChannel(SlackConfig( + bot_token=config.slack_bot_token, + app_token=config.slack_app_token, + allowed_senders=allowed, + allowed_channels=channels, + proxy=proxy, + )) + + +register_channel("slack", create_from_config) diff --git a/EvoScientist/channels/slack/channel.py b/EvoScientist/channels/slack/channel.py new file mode 100644 index 0000000..21237f7 --- /dev/null +++ b/EvoScientist/channels/slack/channel.py @@ -0,0 +1,291 @@ +"""Slack channel implementation using slack-sdk Socket Mode.""" + +import asyncio +import logging +from dataclasses import dataclass +from datetime import datetime + +from ..base import Channel, RawIncoming, ChannelError +from ..capabilities import SLACK as SLACK_CAPS +from ..config import BaseChannelConfig + +logger = logging.getLogger(__name__) + + +@dataclass +class SlackConfig(BaseChannelConfig): + bot_token: str = "" + app_token: str = "" + text_chunk_limit: int = 4096 + + +class SlackChannel(Channel): + """Slack channel using slack-sdk Socket Mode.""" + + name = "slack" + + capabilities = SLACK_CAPS + _ready_attrs = ("_web_client",) + _mention_pattern = r"<@{bot_id}>\s*" + + def __init__(self, config: SlackConfig): + super().__init__(config) + self._socket_client = None + self._web_client = None + self._typing_message_ts: dict[str, str] = {} + + async def start(self) -> None: + try: + from slack_sdk.web.async_client import AsyncWebClient + from slack_sdk.socket_mode.aiohttp import SocketModeClient + from slack_sdk.socket_mode.request import SocketModeRequest + from slack_sdk.socket_mode.response import SocketModeResponse + except ImportError: + raise ChannelError( + "slack-sdk or aiohttp not installed. " + "Install with: pip install evoscientist[slack]" + ) + + if not self.config.bot_token: + raise ChannelError("Slack bot token is required") + if not self.config.app_token: + raise ChannelError( + "Slack app token is required for Socket Mode " + "(starts with xapp-)" + ) + + self._web_client = AsyncWebClient( + token=self.config.bot_token, + proxy=self._get_proxy(), + ) + + # Get bot user ID for filtering own messages + try: + auth = await asyncio.wait_for( + self._web_client.auth_test(), timeout=15, + ) + self._bot_user_id = auth["user_id"] + except asyncio.TimeoutError: + raise ChannelError( + "Slack auth_test timed out — check network and bot token" + ) + except Exception as e: + raise ChannelError(f"Failed to authenticate Slack bot: {e}") + + self._socket_client = SocketModeClient( + app_token=self.config.app_token, + web_client=self._web_client, + ) + + async def _event_handler( + client: SocketModeClient, + req: SocketModeRequest, + ) -> None: + # Acknowledge immediately + resp = SocketModeResponse(envelope_id=req.envelope_id) + await client.send_socket_mode_response(resp) + + logger.debug(f"Slack socket event: type={req.type}") + + if req.type == "events_api": + event = req.payload.get("event", {}) + event_type = event.get("type", "") + if event_type == "message" and "subtype" not in event: + is_dm = event.get("channel_type") == "im" + await self._on_message( + event, is_group=not is_dm, was_mentioned=is_dm, + ) + elif event_type == "app_mention": + await self._on_message( + event, is_group=True, was_mentioned=True, + ) + + self._socket_client.socket_mode_request_listeners.append( + _event_handler + ) + try: + await asyncio.wait_for( + self._socket_client.connect(), timeout=30, + ) + except asyncio.TimeoutError: + raise ChannelError( + "Slack Socket Mode connection timed out — " + "check app token (must start with xapp-) and " + "ensure Socket Mode is enabled in your Slack app settings" + ) + self._running = True + logger.info("Slack channel started (Socket Mode)") + + async def _cleanup(self) -> None: + if self._socket_client: + await self._socket_client.close() + logger.info("Slack channel stopped") + + # ── Typing indicator (override base) ──────────────────────────── + + async def _send_typing_action(self, chat_id: str) -> None: + """Send typing indicator via Slack. + + Slack's Web API and Socket Mode do not expose a dedicated + typing-indicator endpoint for bot tokens. We approximate + the experience by posting a short-lived status message that + is deleted once the real reply is sent (handled by + ``stop_typing``). When the status post fails we silently + fall back to no indicator. + """ + if not self._web_client: + return + try: + resp = await self._web_client.chat_postMessage( + channel=chat_id, + text="\u2026", # "…" ellipsis as minimal typing hint + ) + ts = resp.get("ts") + if ts: + self._typing_message_ts[chat_id] = ts + except Exception: + pass + + async def stop_typing(self, chat_id: str) -> None: + """Cancel typing loop and clean up the status message.""" + # Delete the ephemeral "…" message if we posted one + ts = self._typing_message_ts.pop(chat_id, None) + if ts and self._web_client: + try: + await self._web_client.chat_delete(channel=chat_id, ts=ts) + except Exception: + pass + await super().stop_typing(chat_id) + + # ── Send (template method overrides) ────────────────────────── + + + async def _send_chunk(self, chat_id, formatted_text, raw_text, reply_to, metadata): + kwargs = dict(channel=chat_id) + # Always route to thread if thread_ts is present in metadata, + # not just for the first chunk (reply_to is only set for chunk 0). + if metadata: + thread_ts = metadata.get("thread_ts") + if thread_ts: + kwargs["thread_ts"] = thread_ts + + async def _send(text): + await self._web_client.chat_postMessage(text=text, **kwargs) + + await self._send_with_format_fallback(_send, formatted_text, raw_text) + + async def _send_media_impl( + self, + recipient: str, + file_path: str, + caption: str = "", + metadata: dict | None = None, + ) -> bool: + """Send a media file through Slack.""" + channel_id = self._resolve_media_chat_id(recipient, metadata) + await self._web_client.files_upload_v2( + channel=channel_id, + file=file_path, + initial_comment=caption or None, + ) + return True + + def _get_bot_identifier(self) -> str | None: + return getattr(self, "_bot_user_id", None) + + # ── ACK Reactions ─────────────────────────────────────────────── + + async def _send_ack_reaction(self, chat_id: str, message_id: str, emoji: str = "eyes") -> None: + """Add an emoji reaction to acknowledge receipt.""" + if self._web_client and message_id: + try: + await self._web_client.reactions_add( + channel=chat_id, timestamp=message_id, name=emoji, + ) + except Exception as e: + logger.debug(f"Slack ACK reaction failed: {e}") + + async def _remove_ack_reaction(self, chat_id: str, message_id: str, emoji: str = "eyes") -> None: + """Remove the ACK reaction after replying.""" + if self._web_client and message_id: + try: + await self._web_client.reactions_remove( + channel=chat_id, timestamp=message_id, name=emoji, + ) + except Exception as e: + logger.debug(f"Slack remove ACK reaction failed: {e}") + + async def _on_message( + self, + event: dict, + *, + is_group: bool = False, + was_mentioned: bool = True, + ) -> None: + """Handle an incoming Slack message event.""" + user_id = event.get("user", "") + + # Skip bot's own messages + if user_id == getattr(self, "_bot_user_id", None): + logger.debug("Skipping own bot message") + return + + # Skip bot messages (e.g. from other bots) + if event.get("bot_id"): + logger.debug(f"Skipping bot message from bot_id={event.get('bot_id')}") + return + + channel_id = event.get("channel", "") + + text = event.get("text", "") + + annotations: list[str] = [] + media_paths: list[str] = [] + + # Handle file attachments + if self.config.include_attachments: + files = event.get("files", []) + for file_info in files: + file_size = file_info.get("size", 0) + filename = file_info.get("name", "unknown") + + url = file_info.get("url_private_download") or file_info.get( + "url_private" + ) + if url and self._web_client: + headers = { + "Authorization": f"Bearer {self.config.bot_token}" + } + local_path, annotation = await self._download_attachment( + url, f"{file_info.get('id', 'unknown')}_{filename}", + headers=headers, + file_size=file_size, + ) + if local_path: + media_paths.append(local_path) + if annotation: + annotations.append(annotation) + + ts = event.get("ts", "") + thread_ts = event.get("thread_ts") or ts + try: + timestamp = datetime.fromtimestamp(float(ts)) if ts else datetime.now() + except (ValueError, TypeError): + timestamp = datetime.now() + + await self._enqueue_raw(RawIncoming( + sender_id=user_id, + chat_id=channel_id, + text=text, + media_files=media_paths, + content_annotations=annotations, + timestamp=timestamp, + message_id=ts, + metadata={"chat_id": channel_id, "thread_ts": thread_ts}, + is_group=is_group, + was_mentioned=was_mentioned, + )) + logger.info( + f"Slack message queued: sender={user_id}, " + f"channel={channel_id}, content={text[:50]}" + ) diff --git a/EvoScientist/channels/slack/probe.py b/EvoScientist/channels/slack/probe.py new file mode 100644 index 0000000..fa6d260 --- /dev/null +++ b/EvoScientist/channels/slack/probe.py @@ -0,0 +1,48 @@ +"""Slack bot token validation.""" + +import logging + +logger = logging.getLogger(__name__) + + +async def validate_slack_tokens( + bot_token: str, + app_token: str | None = None, + proxy: str | None = None, +) -> tuple[bool, str]: + """Validate Slack bot token via the auth.test API. + + Optionally checks the app-level token format (must start with ``xapp-``). + + Returns: + Tuple of (is_valid, message). + """ + if not bot_token: + return False, "No bot token provided" + + try: + import httpx + except ImportError: + return False, "httpx not installed" + + # Validate bot token via auth.test + url = "https://slack.com/api/auth.test" + headers = {"Authorization": f"Bearer {bot_token}"} + try: + async with httpx.AsyncClient(proxy=proxy) as client: + resp = await client.post(url, headers=headers, timeout=10) + data = resp.json() + if not data.get("ok"): + error = data.get("error", "unknown error") + return False, f"Invalid bot token: {error}" + bot_name = data.get("user", "unknown") + team = data.get("team", "unknown") + except Exception as e: + return False, f"Error: {e}" + + # Optionally validate app token format + if app_token: + if not app_token.startswith("xapp-"): + return False, "App token must start with 'xapp-'" + + return True, f"Bot: {bot_name} (team: {team})" diff --git a/EvoScientist/channels/slack/serve.py b/EvoScientist/channels/slack/serve.py new file mode 100644 index 0000000..8bcae33 --- /dev/null +++ b/EvoScientist/channels/slack/serve.py @@ -0,0 +1,99 @@ +"""Slack channel server. + +Standalone script to run the Slack channel with CLI options. + +Usage: + python -m EvoScientist.channels.slack.serve --bot-token TOKEN --app-token TOKEN [OPTIONS] + +Examples: + # Allow all senders (default) + python -m EvoScientist.channels.slack.serve --bot-token xoxb-... --app-token xapp-... + + # Only allow specific senders and channels + python -m EvoScientist.channels.slack.serve --bot-token xoxb-... --app-token xapp-... --allow U123 --allow-channel C456 + + # With proxy, agent and thinking + python -m EvoScientist.channels.slack.serve --bot-token xoxb-... --app-token xapp-... --proxy http://proxy:8080 --agent --thinking +""" + +import argparse +import logging + +from .channel import SlackChannel, SlackConfig +from ..bus import MessageBus +from ..standalone import run_standalone + +logging.basicConfig( + level=logging.DEBUG, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + datefmt="%H:%M:%S", +) +logger = logging.getLogger(__name__) + + +def parse_args(): + """Parse command line arguments.""" + parser = argparse.ArgumentParser( + description="Slack channel server", + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument( + "--bot-token", + required=True, + help="Slack bot token (xoxb-...)", + ) + parser.add_argument( + "--app-token", + required=True, + help="Slack app-level token for Socket Mode (xapp-...)", + ) + parser.add_argument( + "--allow", + action="append", + dest="allowed_senders", + help="Allowed sender (Slack user ID). Can be used multiple times.", + ) + parser.add_argument( + "--allow-channel", + action="append", + dest="allowed_channels", + help="Allowed channel ID. Can be used multiple times.", + ) + parser.add_argument( + "--proxy", + help="HTTP proxy URL for Slack API requests", + ) + parser.add_argument( + "--agent", + action="store_true", + help="Use EvoScientist agent as handler (default: echo)", + ) + parser.add_argument( + "--thinking", + action="store_true", + help="Send thinking content as intermediate messages (requires --agent)", + ) + return parser.parse_args() + + +def main(): + """Entry point.""" + args = parse_args() + + config = SlackConfig( + bot_token=args.bot_token, + app_token=args.app_token, + allowed_senders=set(args.allowed_senders) if args.allowed_senders else None, + allowed_channels=set(args.allowed_channels) if args.allowed_channels else None, + proxy=args.proxy, + ) + + send_thinking = args.thinking and args.agent + bus = MessageBus() + channel = SlackChannel(config) + + run_standalone(channel, bus, use_agent=args.agent, send_thinking=send_thinking) + + +if __name__ == "__main__": + main() diff --git a/EvoScientist/channels/wechat/__init__.py b/EvoScientist/channels/wechat/__init__.py new file mode 100644 index 0000000..539261e --- /dev/null +++ b/EvoScientist/channels/wechat/__init__.py @@ -0,0 +1,69 @@ +"""WeChat channel implementations for EvoScientist. + +Supports multiple WeChat backends: + - **wecom**: 企业微信应用 (WeCom / WeChat Work) via official API + — Most stable, pure HTTP, no third-party dependencies + - **wechatmp**: 微信公众号 (WeChat Official Account) via official API + — Pure HTTP webhook, suitable for public-facing bots + +Both backends use httpx (already a core dependency) and receive messages +via HTTP webhook, send replies via REST API. + +Usage in config: + channel_enabled = "wechat" + wechat_backend = "wecom" # or "wechatmp" + + # WeCom settings + wechat_wecom_corp_id = "..." + wechat_wecom_agent_id = "..." + wechat_wecom_secret = "..." + wechat_wecom_token = "..." + wechat_wecom_encoding_aes_key = "..." + wechat_webhook_port = 9001 + + # OR: Official Account settings + wechat_mp_app_id = "..." + wechat_mp_app_secret = "..." + wechat_mp_token = "..." + wechat_mp_encoding_aes_key = "..." + wechat_webhook_port = 9001 +""" + +from .channel import WeChatChannel, WeComConfig, WeChatMPConfig +from ..channel_manager import register_channel, _parse_csv + +__all__ = ["WeChatChannel", "WeComConfig", "WeChatMPConfig"] + + +def create_from_config(config) -> WeChatChannel: + backend = config.wechat_backend or "wecom" + allowed = _parse_csv(config.wechat_allowed_senders) + proxy = config.wechat_proxy if config.wechat_proxy else None + port = int(config.wechat_webhook_port or 9001) + + if backend == "wechatmp": + mp_config = WeChatMPConfig( + app_id=config.wechat_mp_app_id, + app_secret=config.wechat_mp_app_secret, + token=config.wechat_mp_token, + encoding_aes_key=config.wechat_mp_encoding_aes_key, + webhook_port=port, + allowed_senders=allowed, + proxy=proxy, + ) + return WeChatChannel(mp_config, backend="wechatmp") + else: + wecom_config = WeComConfig( + corp_id=config.wechat_wecom_corp_id, + agent_id=config.wechat_wecom_agent_id, + secret=config.wechat_wecom_secret, + token=config.wechat_wecom_token, + encoding_aes_key=config.wechat_wecom_encoding_aes_key, + webhook_port=port, + allowed_senders=allowed, + proxy=proxy, + ) + return WeChatChannel(wecom_config, backend="wecom") + + +register_channel("wechat", create_from_config) diff --git a/EvoScientist/channels/wechat/channel.py b/EvoScientist/channels/wechat/channel.py new file mode 100644 index 0000000..c8929c2 --- /dev/null +++ b/EvoScientist/channels/wechat/channel.py @@ -0,0 +1,805 @@ +"""WeChat channel implementation. + +Supports two backends via a unified Channel interface: + +1. **wecom** (企业微信应用): Corporate WeChat official API + - Receives messages via HTTP callback (XML + optional AES encryption) + - Sends replies via REST API (POST /cgi-bin/message/send) + - Supports text, image, file, markdown messages + - Token auto-refresh with 2-hour TTL + +2. **wechatmp** (微信公众号): WeChat Official Account API + - Receives messages via HTTP callback (XML + optional AES encryption) + - Sends replies via REST API (POST /cgi-bin/message/custom/send) + - Supports text, image, news messages + +Both backends use httpx (already a core dependency) and aiohttp for +webhook server — matching the Feishu channel pattern. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import logging +import re +import time +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import Any, TYPE_CHECKING + +if TYPE_CHECKING: + from aiohttp import web + +from ..mixins import WebhookMixin, TokenMixin +from ..base import Channel, RawIncoming, ChannelError +from ..capabilities import WECHAT as WECHAT_CAPS +from ..config import BaseChannelConfig + +logger = logging.getLogger(__name__) + + +# ── Markdown → plain text (fallback for WeChat text messages) ──── + +def _strip_markdown(text: str) -> str: + """Strip Markdown formatting for plain-text WeChat messages.""" + # Remove code blocks + text = re.sub(r"```[\s\S]*?```", lambda m: m.group(0).strip("`").strip(), text) + # Remove inline code + text = re.sub(r"`([^`]+)`", r"\1", text) + # Remove bold + text = re.sub(r"\*\*(.+?)\*\*", r"\1", text) + # Remove italic + text = re.sub(r"(? list[tuple[str, str, Any]]: + """Return HTTP routes for the shared webhook server.""" + return [ + ("GET", "/wechat/callback", self._handle_verify), + ("POST", "/wechat/callback", self._handle_message), + ] + + async def start(self) -> None: + try: + from aiohttp import web + import httpx # noqa: F401 + except ImportError: + raise ChannelError( + "aiohttp or httpx not installed. " + "Install with: pip install aiohttp httpx" + ) + + self._validate_config() + + import httpx + self._http_client = httpx.AsyncClient( + timeout=15, + proxy=self._get_proxy(), + ) + + # Set up message encryption if configured + if self.config.encoding_aes_key and self.config.token: + from .crypto import WeChatCrypto + app_id = self._get_app_id() + self._crypto = WeChatCrypto( + token=self.config.token, + encoding_aes_key=self.config.encoding_aes_key, + app_id=app_id, + ) + + # Verify credentials by fetching initial token + await self._refresh_token() + + if not getattr(self, "_shared_webhook_server", None): + app = web.Application() + app.router.add_get("/wechat/callback", self._handle_verify) + app.router.add_post("/wechat/callback", self._handle_message) + + self._runner = web.AppRunner(app) + await self._runner.setup() + self._site = web.TCPSite( + self._runner, "0.0.0.0", self.config.webhook_port, + ) + await self._site.start() + + self._running = True + logger.info( + f"WeChat channel started " + f"(backend={self._backend}, " + f"webhook on port {self.config.webhook_port})" + ) + + async def _cleanup(self) -> None: + if self._site: + await self._site.stop() + if self._runner: + await self._runner.cleanup() + if self._http_client: + await self._http_client.aclose() + self._http_client = None + self._access_token = None + logger.info("WeChat channel stopped") + + def _validate_config(self) -> None: + """Validate required config fields based on backend.""" + if self._backend == "wecom": + cfg = self.config + if not cfg.corp_id: + raise ChannelError("WeCom corp_id is required") + if not cfg.secret: + raise ChannelError("WeCom secret is required") + if not cfg.agent_id: + raise ChannelError("WeCom agent_id is required") + elif self._backend == "wechatmp": + cfg = self.config + if not cfg.app_id: + raise ChannelError("WeChat MP app_id is required") + if not cfg.app_secret: + raise ChannelError("WeChat MP app_secret is required") + + def _get_app_id(self) -> str: + """Return the app identifier for crypto operations.""" + if self._backend == "wecom": + return self.config.corp_id + return self.config.app_id + + # ── Token management ────────────────────────────────────────── + + async def _refresh_token(self) -> None: + """Fetch or refresh the access_token.""" + if self._backend == "wecom": + url = ( + f"https://qyapi.weixin.qq.com/cgi-bin/gettoken" + f"?corpid={self.config.corp_id}" + f"&corpsecret={self.config.secret}" + ) + else: + url = ( + f"https://api.weixin.qq.com/cgi-bin/token" + f"?grant_type=client_credential" + f"&appid={self.config.app_id}" + f"&secret={self.config.app_secret}" + ) + + try: + resp = await self._http_client.get(url) + data = resp.json() + except Exception as e: + if not self._running: + raise ChannelError(f"Failed to get WeChat access token: {e}") + raise RuntimeError(f"Failed to get WeChat access token: {e}") + + if data.get("errcode", 0) != 0: + err_msg = ( + f"WeChat auth error ({data.get('errcode')}): " + f"{data.get('errmsg', 'unknown')}" + ) + if not self._running: + raise ChannelError(err_msg) + raise RuntimeError(err_msg) + + self._access_token = data["access_token"] + expire = data.get("expires_in", 7200) + # Refresh 5 minutes before expiry + self._token_expires = time.monotonic() + expire - 300 + logger.debug(f"WeChat token refreshed, expires in {expire}s") + + async def _ensure_token(self) -> str: + """Return a valid access token, refreshing if needed.""" + if not self._access_token or time.monotonic() >= self._token_expires: + await self._refresh_token() + return self._access_token + + # ── Signature verification (GET callback) ───────────────────── + + async def _handle_verify(self, request) -> "web.Response": + """Handle GET /wechat/callback for URL verification. + + WeChat/WeCom sends: msg_signature, timestamp, nonce, echostr + We decrypt echostr (encrypted mode) or verify signature (plain mode) + and return the plain echostr. + """ + from aiohttp import web + + signature = request.query.get("msg_signature") or request.query.get("signature", "") + timestamp = request.query.get("timestamp", "") + nonce = request.query.get("nonce", "") + echostr = request.query.get("echostr", "") + + logger.info(f"Verify request received: timestamp={timestamp}") + + if not echostr: + return web.Response(status=400, text="missing echostr") + + # Encrypted mode: WeCom sends msg_signature and encrypted echostr + if self._crypto and request.query.get("msg_signature"): + # Verify signature first + sig_ok = self._crypto.verify_signature(signature, timestamp, nonce, echostr) + if not sig_ok: + logger.warning("WeChat verify: signature mismatch") + # Try to decrypt regardless — the decrypted echostr must be returned + try: + plain_echostr, _ = self._crypto.decrypt(echostr) + logger.info("WeChat verify: echostr decrypted successfully") + return web.Response(text=plain_echostr) + except Exception as e: + logger.error(f"WeChat verify: echostr decrypt failed: {e}") + return web.Response(status=500) + else: + # Plain mode verification + token = self.config.token + if token: + parts = sorted([token, timestamp, nonce]) + expected = hashlib.sha1("".join(parts).encode()).hexdigest() + if expected != signature: + logger.warning("WeChat verify: signature mismatch (plain)") + return web.Response(status=403) + return web.Response(text=echostr) + + # ── Inbound message handling (POST callback) ────────────────── + + async def _handle_message(self, request) -> "web.Response": + """Handle POST /wechat/callback for incoming messages.""" + from aiohttp import web + from .crypto import parse_xml + + try: + body = await request.text() + except Exception: + return web.Response(status=400) + + logger.info(f"WeChat callback POST received, body length={len(body)}") + xml_data = parse_xml(body) + + # If encrypted, decrypt first + encrypt = xml_data.get("Encrypt", "") + if encrypt and self._crypto: + signature = request.query.get("msg_signature", "") + timestamp = request.query.get("timestamp", "") + nonce = request.query.get("nonce", "") + + if not self._crypto.verify_signature(signature, timestamp, nonce, encrypt): + logger.warning("WeChat message signature mismatch") + return web.Response(status=403) + + try: + decrypted_xml, from_id = self._crypto.decrypt(encrypt) + xml_data = parse_xml(decrypted_xml) + except Exception as e: + logger.error(f"WeChat decrypt failed: {e}") + return web.Response(status=500) + + # Process message asynchronously — WeCom requires a response within + # 5 seconds, but media downloads can take much longer. Return + # "success" immediately and handle the message in the background. + asyncio.create_task(self._safe_process_message(xml_data)) + + return web.Response(text="success") + + async def _safe_process_message(self, xml_data: dict[str, str]) -> None: + """Wrapper that catches exceptions so fire-and-forget tasks don't leak.""" + try: + await self._process_message(xml_data) + except Exception: + logger.exception("Error processing WeChat message") + + async def _process_message(self, xml_data: dict[str, str]) -> None: + """Process a parsed XML message from WeChat/WeCom callback.""" + msg_type = xml_data.get("MsgType", "") + from_user = xml_data.get("FromUserName", "") + to_user = xml_data.get("ToUserName", "") + content = xml_data.get("Content", "") + msg_id = xml_data.get("MsgId", "") + create_time = xml_data.get("CreateTime", "") + + logger.info(f"WeChat message received: type={msg_type}, from={from_user}, id={msg_id}, keys={list(xml_data.keys())}") + + if not from_user: + return + + # Determine chat_id + # For WeCom: FromUserName is the user's UserID + # For MP: FromUserName is the user's OpenID + chat_id = from_user + + # Group chat detection + is_group = False + was_mentioned = True # Default: treat as mentioned (DMs) + + # WeCom group detection: ChatId field indicates a group message + if self._backend == "wecom": + group_chat_id = xml_data.get("ChatId", "") + if group_chat_id: + is_group = True + chat_id = group_chat_id + # WeCom sets MsgType=event with Event=sys when bot is @mentioned, + # but for text messages we check the XML AtUserList field + at_user_list = xml_data.get("AtUserList", "") + was_mentioned = bool(at_user_list) + + # Handle different message types + text = "" + annotations: list[str] = [] + media_paths: list[str] = [] + + if msg_type == "text": + text = content + elif msg_type == "image": + pic_url = xml_data.get("PicUrl", "") + media_id = xml_data.get("MediaId", "") + if pic_url: + local, ann = await self._download_attachment( + pic_url, f"wechat_{msg_id}.jpg", + ) + if local: + media_paths.append(local) + if ann: + annotations.append(ann) + elif media_id: + local, ann = await self._download_wechat_media( + media_id, f"wechat_image_{msg_id}", + ) + if local: + media_paths.append(local) + if ann: + annotations.append(ann) + else: + annotations.append("[image: no download source]") + elif msg_type == "voice": + recognition = xml_data.get("Recognition", "") + media_id = xml_data.get("MediaId", "") + if media_id: + local, ann = await self._download_wechat_media(media_id, f"wechat_voice_{msg_id}") + if local: + media_paths.append(local) + if ann: + ann = ann.replace("[attachment:", "[voice:") + annotations.append(ann) + if recognition: + text = f"[语音识别] {recognition}" + elif not media_paths: + annotations.append("[voice message]") + elif msg_type in ("video", "shortvideo"): + media_id = xml_data.get("MediaId", "") + if media_id: + local, ann = await self._download_wechat_media(media_id, f"wechat_{msg_type}_{msg_id}") + if local: + media_paths.append(local) + if ann: + annotations.append(ann) + if not media_paths: + annotations.append(f"[{msg_type} message]") + elif msg_type == "location": + label = xml_data.get("Label", "") + lat = xml_data.get("Location_X", "") + lon = xml_data.get("Location_Y", "") + text = f"[位置] {label} ({lat}, {lon})" + elif msg_type == "file": + media_id = xml_data.get("MediaId", "") + file_name = xml_data.get("FileName", "") or xml_data.get("Title", f"wechat_file_{msg_id}") + logger.info(f"WeChat file message: name={file_name}, media_id={media_id!r}, keys={list(xml_data.keys())}") + if media_id: + local, ann = await self._download_wechat_media( + media_id, f"wechat_file_{msg_id}_{file_name}", + ) + logger.info(f"WeChat file download result: local={local}, ann={ann}") + if local: + media_paths.append(local) + if ann: + annotations.append(ann) + if not media_paths: + annotations.append(f"[file: {file_name}]") + elif msg_type == "link": + title = xml_data.get("Title", "") + description = xml_data.get("Description", "") + url = xml_data.get("Url", "") + text = f"[链接] {title}\n{description}\n{url}" + elif msg_type == "event": + event_type = xml_data.get("Event", "") + if event_type == "subscribe": + text = "[用户关注]" + elif event_type == "unsubscribe": + logger.info(f"User {from_user} unsubscribed") + return # Don't process + elif event_type == "CLICK": + event_key = xml_data.get("EventKey", "") + text = f"[菜单点击] {event_key}" + elif event_type in ("LOCATION", "VIEW"): + # Periodic location reports and menu-link clicks — ignore + return + else: + logger.debug(f"Ignoring WeChat event: {event_type}") + return + else: + text = f"[{msg_type} message]" + + if not text and not media_paths and not annotations: + return + + # Parse timestamp + try: + timestamp = datetime.fromtimestamp( + int(create_time) + ) if create_time else datetime.now() + except (ValueError, TypeError, OSError): + timestamp = datetime.now() + + await self._enqueue_raw(RawIncoming( + sender_id=from_user, + chat_id=chat_id, + text=text, + media_files=media_paths, + content_annotations=annotations, + timestamp=timestamp, + message_id=msg_id, + is_group=is_group, + was_mentioned=was_mentioned, + metadata={ + "chat_id": chat_id, + "to_user": to_user, + "backend": self._backend, + }, + )) + + # ── Send (template method overrides) ────────────────────────── + + def _format_chunk(self, text: str) -> str: + """WeCom uses markdown formatter; MP uses plain text.""" + if self._backend == "wecom": + return self._formatter.format(text) # markdown profile + return _strip_markdown(text) + + async def _send_chunk( + self, chat_id, formatted_text, raw_text, reply_to, metadata, + ): + token = await self._ensure_token() + + if self._backend == "wecom": + # Group chat: use appchat/send endpoint + if chat_id.startswith("wr"): + try: + await self._wecom_send_group_markdown(token, chat_id, raw_text) + return + except Exception: + pass + await self._wecom_send_group_text(token, chat_id, raw_text) + else: + # DM: Try markdown first, fall back to plain text + try: + await self._wecom_send_markdown(token, chat_id, raw_text) + return + except Exception: + pass + await self._wecom_send_text(token, chat_id, raw_text) + else: + await self._mp_send_text(token, chat_id, raw_text) + + # ── WeCom send ──────────────────────────────────────────────── + + async def _wecom_send_text( + self, token: str, user_id: str, text: str, + ) -> None: + """Send a text message via WeCom API.""" + url = ( + f"https://qyapi.weixin.qq.com/cgi-bin/message/send" + f"?access_token={token}" + ) + body = { + "touser": user_id, + "msgtype": "text", + "agentid": int(self.config.agent_id), + "text": {"content": _strip_markdown(text)}, + } + await self._post_api(url, body) + + async def _wecom_send_markdown( + self, token: str, user_id: str, text: str, + ) -> None: + """Send a markdown message via WeCom API. + + Note: WeCom markdown only supports a subset of Markdown + (no code blocks, no images). Falls back to text if the + message is too complex. + """ + url = ( + f"https://qyapi.weixin.qq.com/cgi-bin/message/send" + f"?access_token={token}" + ) + body = { + "touser": user_id, + "msgtype": "markdown", + "agentid": int(self.config.agent_id), + "markdown": {"content": text}, + } + await self._post_api(url, body) + + # ── WeCom group send ──────────────────────────────────────────── + + async def _wecom_send_group_text( + self, token: str, chatid: str, text: str, + ) -> None: + """Send a text message to a WeCom group chat.""" + url = ( + f"https://qyapi.weixin.qq.com/cgi-bin/appchat/send" + f"?access_token={token}" + ) + body = { + "chatid": chatid, + "msgtype": "text", + "text": {"content": _strip_markdown(text)}, + } + await self._post_api(url, body) + + async def _wecom_send_group_markdown( + self, token: str, chatid: str, text: str, + ) -> None: + """Send a markdown message to a WeCom group chat.""" + url = ( + f"https://qyapi.weixin.qq.com/cgi-bin/appchat/send" + f"?access_token={token}" + ) + body = { + "chatid": chatid, + "msgtype": "markdown", + "markdown": {"content": text}, + } + await self._post_api(url, body) + + # ── MP send ─────────────────────────────────────────────────── + + async def _mp_send_text( + self, token: str, openid: str, text: str, + ) -> None: + """Send a text message via WeChat MP customer service API.""" + url = ( + f"https://api.weixin.qq.com/cgi-bin/message/custom/send" + f"?access_token={token}" + ) + body = { + "touser": openid, + "msgtype": "text", + "text": {"content": _strip_markdown(text)}, + } + await self._post_api(url, body) + + # ── Media send ──────────────────────────────────────────────── + + async def _send_media_impl( + self, + recipient: str, + file_path: str, + caption: str = "", + metadata: dict | None = None, + ) -> bool: + """Send a media file via WeChat/WeCom.""" + token = await self._ensure_token() + chat_id = self._resolve_media_chat_id(recipient, metadata) + + # Upload media to get media_id + media_id = await self._upload_media(token, file_path) + if not media_id: + return False + + path = Path(file_path) + ext = path.suffix.lower() + is_image = ext in {".jpg", ".jpeg", ".png", ".gif", ".bmp"} + + if self._backend == "wecom": + msg_type = "image" if is_image else "file" + # Group chat: use appchat/send endpoint + if chat_id.startswith("wr"): + url = ( + f"https://qyapi.weixin.qq.com/cgi-bin/appchat/send" + f"?access_token={token}" + ) + body = { + "chatid": chat_id, + "msgtype": msg_type, + msg_type: {"media_id": media_id}, + } + else: + url = ( + f"https://qyapi.weixin.qq.com/cgi-bin/message/send" + f"?access_token={token}" + ) + body = { + "touser": chat_id, + "msgtype": msg_type, + "agentid": int(self.config.agent_id), + msg_type: {"media_id": media_id}, + } + else: + url = ( + f"https://api.weixin.qq.com/cgi-bin/message/custom/send" + f"?access_token={token}" + ) + msg_type = "image" if is_image else "file" # MP only supports image + if not is_image: + # MP doesn't support file via customer service API; + # send caption as text instead + if caption: + await self._mp_send_text(token, chat_id, f"[文件] {path.name}\n{caption}") + return True + body = { + "touser": chat_id, + "msgtype": "image", + "image": {"media_id": media_id}, + } + + await self._post_api(url, body) + + # Send caption separately if provided + if caption: + if self._backend == "wecom": + if chat_id.startswith("wr"): + await self._wecom_send_group_text(token, chat_id, caption) + else: + await self._wecom_send_text(token, chat_id, caption) + else: + await self._mp_send_text(token, chat_id, caption) + + return True + + async def _upload_media( + self, token: str, file_path: str, + ) -> str | None: + """Upload a media file and return the media_id.""" + path = Path(file_path) + ext = path.suffix.lower() + is_image = ext in {".jpg", ".jpeg", ".png", ".gif", ".bmp"} + media_type = "image" if is_image else "file" + + if self._backend == "wecom": + url = ( + f"https://qyapi.weixin.qq.com/cgi-bin/media/upload" + f"?access_token={token}&type={media_type}" + ) + else: + url = ( + f"https://api.weixin.qq.com/cgi-bin/media/upload" + f"?access_token={token}&type={media_type}" + ) + + try: + with open(file_path, "rb") as f: + resp = await self._http_client.post( + url, + files={"media": (path.name, f)}, + ) + data = resp.json() + if data.get("errcode", 0) != 0 and "media_id" not in data: + logger.error( + f"WeChat media upload failed: {data.get('errmsg')}" + ) + return None + return data.get("media_id") + except Exception as e: + logger.error(f"WeChat media upload error: {e}") + return None + + # ── Media download helper ──────────────────────────────────── + + async def _download_wechat_media( + self, media_id: str, filename: str, + ) -> tuple[str | None, str | None]: + """Download media by media_id via WeChat/WeCom media API.""" + token = await self._ensure_token() + if self._backend == "wecom": + url = f"https://qyapi.weixin.qq.com/cgi-bin/media/get?access_token={token}&media_id={media_id}" + else: + url = f"https://api.weixin.qq.com/cgi-bin/media/get?access_token={token}&media_id={media_id}" + return await self._download_attachment(url, filename) + + # ── Shared API helper ───────────────────────────────────────── + + async def _post_api(self, url: str, body: dict) -> dict: + """POST to WeChat/WeCom API, check errcode, return response.""" + try: + resp = await self._http_client.post(url, json=body) + data = resp.json() + except Exception as e: + raise RuntimeError(f"WeChat API error: {e}") + + errcode = data.get("errcode", 0) + if errcode != 0: + errmsg = data.get("errmsg", "unknown") + # Token expired — refresh and retry once + if errcode in (40014, 42001): + logger.warning("WeChat token expired, refreshing...") + await self._refresh_token() + token = self._access_token + # Replace token in URL + if "access_token=" in url: + url = re.sub( + r"access_token=[^&]+", + f"access_token={token}", + url, + ) + resp = await self._http_client.post(url, json=body) + data = resp.json() + if data.get("errcode", 0) != 0: + raise RuntimeError( + f"WeChat API error after retry: " + f"{data.get('errmsg')}" + ) + return data + else: + raise RuntimeError( + f"WeChat API error ({errcode}): {errmsg}" + ) + + return data + + # _send_typing_action: inherited no-op (WeChat has no typing API) + + diff --git a/EvoScientist/channels/wechat/crypto.py b/EvoScientist/channels/wechat/crypto.py new file mode 100644 index 0000000..3652a34 --- /dev/null +++ b/EvoScientist/channels/wechat/crypto.py @@ -0,0 +1,187 @@ +"""WeChat / WeCom crypto helpers. + +Implements the message encryption/decryption protocol used by both +WeCom (企业微信) and WeChat Official Account (公众号) callback APIs. + +The protocol uses AES-256-CBC with a key derived from the EncodingAESKey +(base64-encoded 43-char string → 32-byte AES key). + +References: + - WeCom: https://developer.work.weixin.qq.com/document/path/90930 + - MP: https://developers.weixin.qq.com/doc/offiaccount/Message_Management/Message_Encryption_and_Decryption_Instructions.html +""" + +import base64 +import hashlib +import struct +import time +import xml.etree.ElementTree as ET + +# Crypto imports — all from the Python standard library + pycryptodome +# (but we'll use a pure-Python fallback if not available) +try: + from Crypto.Cipher import AES + _HAS_PYCRYPTO = True +except ImportError: + _HAS_PYCRYPTO = False + + +def _pkcs7_pad(data: bytes, block_size: int = 32) -> bytes: + """PKCS#7 padding.""" + pad_len = block_size - (len(data) % block_size) + return data + bytes([pad_len]) * pad_len + + +def _pkcs7_unpad(data: bytes) -> bytes: + """PKCS#7 unpadding.""" + pad_len = data[-1] + if pad_len < 1 or pad_len > 32: + return data + return data[:-pad_len] + + +def _aes_decrypt(key: bytes, iv: bytes, ciphertext: bytes) -> bytes: + """AES-256-CBC decryption.""" + if _HAS_PYCRYPTO: + cipher = AES.new(key, AES.MODE_CBC, iv) + return cipher.decrypt(ciphertext) + else: + # Pure-Python AES fallback (slower but no C deps) + # We'll try pyaes as a fallback + try: + import pyaes + decrypter = pyaes.Decrypter( + pyaes.AESModeOfOperationCBC(key, iv=iv) + ) + decrypted = decrypter.feed(ciphertext) + decrypted += decrypter.feed() + return decrypted + except ImportError: + raise ImportError( + "WeChat message decryption requires pycryptodome or pyaes. " + "Install with: pip install pycryptodome" + ) + + +def _aes_encrypt(key: bytes, iv: bytes, plaintext: bytes) -> bytes: + """AES-256-CBC encryption.""" + if _HAS_PYCRYPTO: + cipher = AES.new(key, AES.MODE_CBC, iv) + return cipher.encrypt(plaintext) + else: + try: + import pyaes + encrypter = pyaes.Encrypter( + pyaes.AESModeOfOperationCBC(key, iv=iv) + ) + encrypted = encrypter.feed(plaintext) + encrypted += encrypter.feed() + return encrypted + except ImportError: + raise ImportError( + "WeChat message encryption requires pycryptodome or pyaes. " + "Install with: pip install pycryptodome" + ) + + +class WeChatCrypto: + """Handles WeChat/WeCom message encryption and decryption. + + Parameters + ---------- + token: + The Token configured in the WeChat/WeCom callback URL settings. + encoding_aes_key: + The 43-character EncodingAESKey (base64-encoded). + app_id: + The AppID (for MP) or CorpID (for WeCom). + """ + + def __init__(self, token: str, encoding_aes_key: str, app_id: str): + self.token = token + self.app_id = app_id + # Decode the AES key: EncodingAESKey + "=" → base64 decode → 32 bytes + self.aes_key = base64.b64decode(encoding_aes_key + "=") + # IV is the first 16 bytes of the key + self.iv = self.aes_key[:16] + + def verify_signature( + self, signature: str, timestamp: str, nonce: str, + encrypt: str = "", + ) -> bool: + """Verify the callback signature. + + For plain-mode verification (no encryption), *encrypt* can be empty. + """ + parts = sorted([self.token, timestamp, nonce] + ([encrypt] if encrypt else [])) + sha1 = hashlib.sha1("".join(parts).encode()).hexdigest() + return sha1 == signature + + def decrypt(self, encrypt: str) -> tuple[str, str]: + """Decrypt an encrypted message. + + Returns ``(xml_content, from_app_id)`` tuple. + """ + ciphertext = base64.b64decode(encrypt) + plaintext = _aes_decrypt(self.aes_key, self.iv, ciphertext) + plaintext = _pkcs7_unpad(plaintext) + + # plaintext layout: + # 16 bytes random + 4 bytes msg_len (big-endian) + msg + app_id + msg_len = struct.unpack("!I", plaintext[16:20])[0] + msg = plaintext[20:20 + msg_len].decode("utf-8") + from_app_id = plaintext[20 + msg_len:].decode("utf-8") + return msg, from_app_id + + def encrypt(self, reply_msg: str) -> str: + """Encrypt a reply message. + + Returns the base64-encoded ciphertext. + """ + msg_bytes = reply_msg.encode("utf-8") + app_id_bytes = self.app_id.encode("utf-8") + + # Random 16 bytes + msg_len (4 bytes big-endian) + msg + app_id + import os + random_bytes = os.urandom(16) + msg_len = struct.pack("!I", len(msg_bytes)) + plaintext = random_bytes + msg_len + msg_bytes + app_id_bytes + plaintext = _pkcs7_pad(plaintext) + + ciphertext = _aes_encrypt(self.aes_key, self.iv, plaintext) + return base64.b64encode(ciphertext).decode("utf-8") + + def generate_signature( + self, encrypt: str, timestamp: str, nonce: str, + ) -> str: + """Generate the msg_signature for an encrypted reply.""" + parts = sorted([self.token, timestamp, nonce, encrypt]) + return hashlib.sha1("".join(parts).encode()).hexdigest() + + def wrap_encrypted_reply(self, reply_msg: str) -> str: + """Encrypt a reply and wrap it in the XML envelope. + + Returns the full XML string to return in the HTTP response. + """ + encrypt = self.encrypt(reply_msg) + timestamp = str(int(time.time())) + nonce = hashlib.md5(str(time.time()).encode()).hexdigest()[:10] + signature = self.generate_signature(encrypt, timestamp, nonce) + + return ( + f"" + f"" + f"" + f"{timestamp}" + f"" + f"" + ) + + +def parse_xml(xml_str: str) -> dict[str, str]: + """Parse a WeChat callback XML into a flat dict.""" + root = ET.fromstring(xml_str) + result = {} + for child in root: + result[child.tag] = child.text or "" + return result diff --git a/EvoScientist/channels/wechat/probe.py b/EvoScientist/channels/wechat/probe.py new file mode 100644 index 0000000..ffa26c2 --- /dev/null +++ b/EvoScientist/channels/wechat/probe.py @@ -0,0 +1,72 @@ +"""WeChat/WeCom credential validation.""" + +import logging + +logger = logging.getLogger(__name__) + + +async def validate_wecom( + corp_id: str, + secret: str, + proxy: str | None = None, +) -> tuple[bool, str]: + """Validate WeCom credentials by fetching an access token. + + Returns: + Tuple of (is_valid, message). + """ + if not corp_id or not secret: + return False, "corp_id and secret are required" + + try: + import httpx + except ImportError: + return False, "httpx not installed" + + url = ( + f"https://qyapi.weixin.qq.com/cgi-bin/gettoken" + f"?corpid={corp_id}&corpsecret={secret}" + ) + try: + async with httpx.AsyncClient(proxy=proxy) as client: + resp = await client.get(url, timeout=10) + data = resp.json() + if data.get("errcode", 0) == 0: + return True, "WeCom credentials valid" + return False, f"Error ({data.get('errcode')}): {data.get('errmsg')}" + except Exception as e: + return False, f"Error: {e}" + + +async def validate_wechat_mp( + app_id: str, + app_secret: str, + proxy: str | None = None, +) -> tuple[bool, str]: + """Validate WeChat Official Account credentials. + + Returns: + Tuple of (is_valid, message). + """ + if not app_id or not app_secret: + return False, "app_id and app_secret are required" + + try: + import httpx + except ImportError: + return False, "httpx not installed" + + url = ( + f"https://api.weixin.qq.com/cgi-bin/token" + f"?grant_type=client_credential" + f"&appid={app_id}&secret={app_secret}" + ) + try: + async with httpx.AsyncClient(proxy=proxy) as client: + resp = await client.get(url, timeout=10) + data = resp.json() + if "access_token" in data: + return True, "WeChat MP credentials valid" + return False, f"Error ({data.get('errcode')}): {data.get('errmsg')}" + except Exception as e: + return False, f"Error: {e}" diff --git a/EvoScientist/channels/wechat/serve.py b/EvoScientist/channels/wechat/serve.py new file mode 100644 index 0000000..1a518bf --- /dev/null +++ b/EvoScientist/channels/wechat/serve.py @@ -0,0 +1,139 @@ +"""WeChat channel server. + +Standalone script to run the WeChat channel with CLI options. + +Usage: + # WeCom (企业微信应用) + python -m EvoScientist.channels.wechat.serve \\ + --backend wecom \\ + --corp-id CORP_ID \\ + --agent-id AGENT_ID \\ + --secret SECRET \\ + --token TOKEN \\ + --aes-key AES_KEY + + # WeChat Official Account (公众号) + python -m EvoScientist.channels.wechat.serve \\ + --backend wechatmp \\ + --app-id APP_ID \\ + --app-secret APP_SECRET \\ + --token TOKEN \\ + --aes-key AES_KEY + +Options: + --port PORT Webhook listen port (default: 9001) + --allow USER_ID Allowed sender (repeatable) + --agent Use EvoScientist agent as handler + --thinking Send thinking content as intermediate messages +""" + +import argparse +import logging + +from .channel import WeChatChannel, WeComConfig, WeChatMPConfig +from ..bus import MessageBus +from ..standalone import run_standalone + +logging.basicConfig( + level=logging.DEBUG, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + datefmt="%H:%M:%S", +) +logger = logging.getLogger(__name__) + + +def parse_args(): + """Parse command line arguments.""" + parser = argparse.ArgumentParser( + description="WeChat channel server", + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument( + "--backend", + choices=["wecom", "wechatmp"], + default="wecom", + help="WeChat backend type (default: wecom)", + ) + parser.add_argument("--port", type=int, default=9001, help="Webhook port") + parser.add_argument( + "--allow", + action="append", + dest="allowed_senders", + help="Allowed sender ID (repeatable)", + ) + parser.add_argument( + "--allow-channel", + action="append", + dest="allowed_channels", + help="Allowed channel ID. Can be used multiple times.", + ) + parser.add_argument( + "--agent", + action="store_true", + help="Use EvoScientist agent as handler", + ) + parser.add_argument( + "--thinking", + action="store_true", + help="Send thinking content (requires --agent)", + ) + + # WeCom settings + wecom = parser.add_argument_group("WeCom (企业微信)") + wecom.add_argument("--corp-id", default="", help="WeCom Corp ID") + wecom.add_argument("--agent-id", default="", help="WeCom Agent ID") + wecom.add_argument("--secret", default="", help="WeCom Secret") + + # MP settings + mp = parser.add_argument_group("WeChat Official Account (公众号)") + mp.add_argument("--app-id", default="", help="MP App ID") + mp.add_argument("--app-secret", default="", help="MP App Secret") + + # Shared settings + parser.add_argument("--token", default="", help="Callback verification token") + parser.add_argument("--aes-key", default="", help="EncodingAESKey") + parser.add_argument("--proxy", default="", help="HTTP proxy URL") + + return parser.parse_args() + + +def main(): + """Entry point.""" + args = parse_args() + allowed = set(args.allowed_senders) if args.allowed_senders else None + allowed_channels = set(args.allowed_channels) if args.allowed_channels else None + proxy = args.proxy or None + + if args.backend == "wecom": + config = WeComConfig( + corp_id=args.corp_id, + agent_id=args.agent_id, + secret=args.secret, + token=args.token, + encoding_aes_key=args.aes_key, + webhook_port=args.port, + allowed_senders=allowed, + allowed_channels=allowed_channels, + proxy=proxy, + ) + else: + config = WeChatMPConfig( + app_id=args.app_id, + app_secret=args.app_secret, + token=args.token, + encoding_aes_key=args.aes_key, + webhook_port=args.port, + allowed_senders=allowed, + allowed_channels=allowed_channels, + proxy=proxy, + ) + + send_thinking = args.thinking and args.agent + bus = MessageBus() + channel = WeChatChannel(config, backend=args.backend) + + run_standalone(channel, bus, use_agent=args.agent, send_thinking=send_thinking) + + +if __name__ == "__main__": + main() diff --git a/EvoScientist/channels/wechat/verify_server.py b/EvoScientist/channels/wechat/verify_server.py new file mode 100644 index 0000000..bf00c66 --- /dev/null +++ b/EvoScientist/channels/wechat/verify_server.py @@ -0,0 +1,175 @@ +"""WeChat callback verification server. + +Provides a lightweight temporary HTTP server that handles the WeChat/WeCom +URL verification handshake during onboarding. This solves the chicken-and-egg +problem: WeChat requires a live server to verify the callback URL before +saving, but the main EvoScientist service isn't running during onboard. + +Usage: + server = VerifyServer(port, token, encoding_aes_key, corp_id) + await server.start() + # ... user clicks "Save" in WeCom admin console ... + # ... server auto-responds to the verification GET request ... + await server.wait_for_verify(timeout=120) + await server.stop() +""" + +from __future__ import annotations + +import asyncio +import hashlib +import logging +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from aiohttp import web + +logger = logging.getLogger(__name__) + + +class VerifyServer: + """Temporary HTTP server for WeChat/WeCom callback URL verification. + + Handles the GET verification request (signature + echostr) and + signals when verification succeeds. + """ + + def __init__( + self, + port: int, + token: str, + encoding_aes_key: str = "", + app_id: str = "", + ): + self.port = port + self.token = token + self._crypto = None + self._runner = None + self._site = None + self._verified = asyncio.Event() + + if encoding_aes_key and token and app_id: + from .crypto import WeChatCrypto + self._crypto = WeChatCrypto( + token=token, + encoding_aes_key=encoding_aes_key, + app_id=app_id, + ) + + async def start(self) -> None: + """Start the verification server.""" + from aiohttp import web + + app = web.Application() + app.router.add_get("/wechat/callback", self._handle) + # Also handle POST in case WeCom sends a POST for some reason + app.router.add_post("/wechat/callback", self._handle_post) + + self._runner = web.AppRunner(app) + await self._runner.setup() + self._site = web.TCPSite(self._runner, "0.0.0.0", self.port) + await self._site.start() + logger.info(f"Verify server listening on port {self.port}") + + async def stop(self) -> None: + """Stop the verification server.""" + if self._site: + await self._site.stop() + if self._runner: + await self._runner.cleanup() + self._site = None + self._runner = None + + async def wait_for_verify(self, timeout: float = 120) -> bool: + """Wait for verification to succeed. + + Returns True if verified within timeout, False otherwise. + """ + try: + await asyncio.wait_for(self._verified.wait(), timeout=timeout) + return True + except asyncio.TimeoutError: + return False + + @property + def is_verified(self) -> bool: + return self._verified.is_set() + + async def _handle(self, request) -> web.Response: + """Handle GET verification request. + + During onboarding we use a lenient approach: + 1. Try strict crypto verification (encrypted mode) + 2. Try strict plain-mode signature check + 3. If both fail, fall back to decrypting echostr without + signature check (WeCom requires the decrypted echostr) + 4. Last resort: echo back raw echostr + + This ensures the callback URL can be saved even if Token/AESKey + have minor issues, while still attempting proper verification. + """ + from aiohttp import web + + signature = ( + request.query.get("msg_signature") + or request.query.get("signature", "") + ) + timestamp = request.query.get("timestamp", "") + nonce = request.query.get("nonce", "") + echostr = request.query.get("echostr", "") + + logger.info( + f"Verify request: msg_signature={signature[:16]}... " + f"timestamp={timestamp} nonce={nonce} " + f"echostr={echostr[:32]}..." + ) + + if not echostr: + return web.Response(status=400, text="missing echostr") + + # Attempt 1: Encrypted mode with full signature verification + if self._crypto and request.query.get("msg_signature"): + sig_ok = self._crypto.verify_signature( + signature, timestamp, nonce, echostr, + ) + if sig_ok: + try: + plain_echostr, _ = self._crypto.decrypt(echostr) + self._verified.set() + logger.info("✓ Verified (encrypted, signature OK)") + return web.Response(text=plain_echostr) + except Exception as e: + logger.warning(f"Signature OK but decrypt failed: {e}") + else: + logger.warning("Signature mismatch, trying decrypt anyway...") + + # Attempt 2: Try decrypt without signature check + # (WeCom requires the decrypted echostr to be returned) + try: + plain_echostr, _ = self._crypto.decrypt(echostr) + self._verified.set() + logger.info("✓ Verified (decrypted, signature skipped)") + return web.Response(text=plain_echostr) + except Exception as e: + logger.warning(f"Decrypt also failed: {e}") + + # Attempt 3: Plain mode signature check + if self.token: + parts = sorted([self.token, timestamp, nonce]) + expected = hashlib.sha1("".join(parts).encode()).hexdigest() + if expected == signature: + self._verified.set() + logger.info("✓ Verified (plain mode)") + return web.Response(text=echostr) + + # Attempt 4: Last resort — just echo back the echostr + # This won't work for encrypted mode (WeCom expects decrypted), + # but works for plain mode with wrong token. + logger.warning("All verification methods failed, echoing raw echostr") + self._verified.set() + return web.Response(text=echostr) + + async def _handle_post(self, request) -> web.Response: + """Handle POST — just acknowledge during verification phase.""" + from aiohttp import web + return web.Response(text="success") diff --git a/EvoScientist/config/onboard.py b/EvoScientist/config/onboard.py index 7335b4d..6b5bbbe 100644 --- a/EvoScientist/config/onboard.py +++ b/EvoScientist/config/onboard.py @@ -1317,6 +1317,8 @@ def _step_channels(config: EvoScientistConfig) -> dict[str, object]: _CHANNELS = [ ("telegram", "Telegram", [("telegram_bot_token", "Bot token (from @BotFather)")]), ("discord", "Discord", [("discord_bot_token", "Bot token")]), + ("slack", "Slack", [("slack_bot_token", "Bot token (xoxb-...)"), ("slack_app_token", "App token for Socket Mode (xapp-...)")]), + ("wechat", "WeChat", [("wechat_wecom_corp_id", "WeCom Corp ID"), ("wechat_wecom_agent_id", "WeCom Agent ID"), ("wechat_wecom_secret", "WeCom Secret")]), ("imessage", "iMessage", []), # handled via _setup_imessage() ] @@ -1475,6 +1477,29 @@ def _probe_channel( _val("discord_bot_token"), _val("discord_proxy") or None, ) + elif ch_name == "slack": + from ..channels.slack.probe import validate_slack_tokens + return await validate_slack_tokens( + _val("slack_bot_token"), + _val("slack_app_token") or None, + _val("slack_proxy") or None, + ) + elif ch_name == "wechat": + backend = _val("wechat_backend", "wecom") + if backend == "wechatmp": + from ..channels.wechat.probe import validate_wechat_mp + return await validate_wechat_mp( + _val("wechat_mp_app_id"), + _val("wechat_mp_app_secret"), + _val("wechat_proxy") or None, + ) + else: + from ..channels.wechat.probe import validate_wecom + return await validate_wecom( + _val("wechat_wecom_corp_id"), + _val("wechat_wecom_secret"), + _val("wechat_proxy") or None, + ) else: return True, "No probe available" diff --git a/EvoScientist/config/settings.py b/EvoScientist/config/settings.py index cc081ba..4a40e56 100644 --- a/EvoScientist/config/settings.py +++ b/EvoScientist/config/settings.py @@ -85,7 +85,7 @@ class EvoScientistConfig: show_thinking: bool = True # Channel Settings - channel_enabled: str = "" # "imessage" | "telegram" | "discord" | "" (comma-separated for multiple) + channel_enabled: str = "" # "imessage" | "telegram" | "discord" | "slack" | "wechat" | "" (comma-separated for multiple) channel_send_thinking: bool = True # forward thinking to any channel require_mention: str = "group" # "always" | "group" | "off" text_chunk_limit: int = 0 # 0 = use capability default diff --git a/tests/test_slack_channel.py b/tests/test_slack_channel.py new file mode 100644 index 0000000..f33bb95 --- /dev/null +++ b/tests/test_slack_channel.py @@ -0,0 +1,87 @@ +"""Tests for Slack channel implementation.""" + +import asyncio + +import pytest + +from EvoScientist.channels.slack.channel import SlackChannel, SlackConfig +from EvoScientist.channels.base import ChannelError + + +def _run(coro): + """Run an async coroutine safely, creating a fresh event loop.""" + loop = asyncio.new_event_loop() + try: + return loop.run_until_complete(coro) + finally: + loop.close() + + +class TestSlackConfig: + def test_default_values(self): + config = SlackConfig() + assert config.bot_token == "" + assert config.app_token == "" + assert config.allowed_senders is None + assert config.allowed_channels is None + assert config.text_chunk_limit == 4096 + + def test_custom_values(self): + config = SlackConfig( + bot_token="xoxb-test", + app_token="xapp-test", + allowed_senders={"U123"}, + allowed_channels={"C456"}, + text_chunk_limit=2000, + ) + assert config.bot_token == "xoxb-test" + assert config.app_token == "xapp-test" + assert config.allowed_senders == {"U123"} + assert config.allowed_channels == {"C456"} + assert config.text_chunk_limit == 2000 + + +class TestSlackChannel: + def test_init(self): + config = SlackConfig(bot_token="xoxb-test", app_token="xapp-test") + channel = SlackChannel(config) + assert channel.config is config + assert channel._running is False + + def test_start_raises_without_bot_token(self): + config = SlackConfig(bot_token="", app_token="xapp-test") + channel = SlackChannel(config) + with pytest.raises(ChannelError, match="bot token"): + _run(channel.start()) + + def test_start_raises_without_app_token(self): + config = SlackConfig(bot_token="xoxb-test", app_token="") + channel = SlackChannel(config) + with pytest.raises(ChannelError, match="app token"): + _run(channel.start()) + + def test_stop_when_not_running(self): + config = SlackConfig(bot_token="xoxb-test", app_token="xapp-test") + channel = SlackChannel(config) + _run(channel.stop()) + + def test_send_returns_false_without_client(self): + from EvoScientist.channels.base import OutboundMessage + + config = SlackConfig(bot_token="xoxb-test", app_token="xapp-test") + channel = SlackChannel(config) + msg = OutboundMessage( + channel="slack", + chat_id="C123", + content="hello", + metadata={"chat_id": "C123"}, + ) + result = _run(channel.send(msg)) + assert result is False + + +class TestSlackChannelRegistration: + def test_slack_registered(self): + from EvoScientist.channels.channel_manager import available_channels + channels = available_channels() + assert "slack" in channels diff --git a/tests/test_wechat_channel.py b/tests/test_wechat_channel.py new file mode 100644 index 0000000..e0ae457 --- /dev/null +++ b/tests/test_wechat_channel.py @@ -0,0 +1,458 @@ +"""Tests for WeChat channel implementation.""" + +import asyncio +import hashlib +import time +import xml.etree.ElementTree as ET + +import pytest + +from EvoScientist.channels.wechat.channel import ( + WeChatChannel, + WeComConfig, + WeChatMPConfig, + _strip_markdown, +) +from EvoScientist.channels.wechat.crypto import ( + WeChatCrypto, + parse_xml, + _pkcs7_pad, + _pkcs7_unpad, +) +from EvoScientist.channels.base import ChannelError + + +def _run(coro): + """Run an async coroutine safely, creating a fresh event loop.""" + loop = asyncio.new_event_loop() + try: + return loop.run_until_complete(coro) + finally: + loop.close() + + +# ── Config tests ────────────────────────────────────────────────── + +class TestWeComConfig: + def test_default_values(self): + config = WeComConfig() + assert config.corp_id == "" + assert config.agent_id == "" + assert config.secret == "" + assert config.webhook_port == 9001 + assert config.allowed_senders is None + assert config.text_chunk_limit == 4096 + + def test_custom_values(self): + config = WeComConfig( + corp_id="corp123", + agent_id="1000001", + secret="my-secret", + token="my-token", + encoding_aes_key="a" * 43, + webhook_port=8080, + allowed_senders={"user1", "user2"}, + ) + assert config.corp_id == "corp123" + assert config.agent_id == "1000001" + assert config.allowed_senders == {"user1", "user2"} + assert config.webhook_port == 8080 + + +class TestWeChatMPConfig: + def test_default_values(self): + config = WeChatMPConfig() + assert config.app_id == "" + assert config.app_secret == "" + assert config.webhook_port == 9001 + + def test_custom_values(self): + config = WeChatMPConfig( + app_id="wx1234", + app_secret="secret", + token="mp-token", + ) + assert config.app_id == "wx1234" + + +# ── Channel init / lifecycle tests ──────────────────────────────── + +class TestWeChatChannelInit: + def test_wecom_init(self): + config = WeComConfig(corp_id="corp", agent_id="1", secret="s") + channel = WeChatChannel(config, backend="wecom") + assert channel.name == "wechat" + assert channel._backend == "wecom" + assert channel._running is False + + def test_mp_init(self): + config = WeChatMPConfig(app_id="wx", app_secret="s") + channel = WeChatChannel(config, backend="wechatmp") + assert channel._backend == "wechatmp" + + def test_start_raises_without_corp_id(self): + config = WeComConfig(corp_id="", agent_id="1", secret="s") + channel = WeChatChannel(config, backend="wecom") + with pytest.raises(ChannelError, match="corp_id"): + _run(channel.start()) + + def test_start_raises_without_secret(self): + config = WeComConfig(corp_id="corp", agent_id="1", secret="") + channel = WeChatChannel(config, backend="wecom") + with pytest.raises(ChannelError, match="secret"): + _run(channel.start()) + + def test_start_raises_without_agent_id(self): + config = WeComConfig(corp_id="corp", agent_id="", secret="s") + channel = WeChatChannel(config, backend="wecom") + with pytest.raises(ChannelError, match="agent_id"): + _run(channel.start()) + + def test_start_raises_mp_without_app_id(self): + config = WeChatMPConfig(app_id="", app_secret="s") + channel = WeChatChannel(config, backend="wechatmp") + with pytest.raises(ChannelError, match="app_id"): + _run(channel.start()) + + def test_stop_when_not_running(self): + config = WeComConfig(corp_id="c", agent_id="1", secret="s") + channel = WeChatChannel(config, backend="wecom") + _run(channel.stop()) # Should not raise + + def test_send_returns_false_without_client(self): + from EvoScientist.channels.base import OutboundMessage + + config = WeComConfig(corp_id="c", agent_id="1", secret="s") + channel = WeChatChannel(config, backend="wecom") + msg = OutboundMessage( + channel="wechat", + chat_id="user1", + content="hello", + metadata={"chat_id": "user1"}, + ) + result = _run(channel.send(msg)) + assert result is False + + +# ── Markdown stripping tests ────────────────────────────────────── + +class TestStripMarkdown: + def test_plain_text(self): + assert _strip_markdown("hello world") == "hello world" + + def test_bold(self): + assert _strip_markdown("**bold**") == "bold" + + def test_italic(self): + assert _strip_markdown("_italic_") == "italic" + + def test_code(self): + assert _strip_markdown("`code`") == "code" + + def test_link(self): + result = _strip_markdown("[text](https://example.com)") + assert "text" in result + assert "https://example.com" in result + + def test_heading(self): + assert _strip_markdown("## Title").strip() == "Title" + + def test_list_items(self): + result = _strip_markdown("- item1\n- item2") + assert "• item1" in result + assert "• item2" in result + + def test_strikethrough(self): + assert _strip_markdown("~~deleted~~") == "deleted" + + def test_code_block(self): + text = "```python\nprint('hi')\n```" + result = _strip_markdown(text) + assert "print('hi')" in result + + +# ── XML parsing tests ───────────────────────────────────────────── + +class TestParseXml: + def test_basic_text_message(self): + xml = ( + "" + "" + "" + "" + "" + "1234" + "1700000000" + "" + ) + data = parse_xml(xml) + assert data["MsgType"] == "text" + assert data["Content"] == "hello" + assert data["FromUserName"] == "user123" + assert data["MsgId"] == "1234" + + def test_image_message(self): + xml = ( + "" + "" + "" + "" + "" + "" + ) + data = parse_xml(xml) + assert data["MsgType"] == "image" + assert data["PicUrl"] == "https://example.com/img.jpg" + + def test_event_message(self): + xml = ( + "" + "" + "" + "" + "" + ) + data = parse_xml(xml) + assert data["MsgType"] == "event" + assert data["Event"] == "subscribe" + + +# ── Crypto tests ────────────────────────────────────────────────── + +class TestPKCS7: + def test_pad_unpad_roundtrip(self): + data = b"hello" + padded = _pkcs7_pad(data) + assert len(padded) % 32 == 0 + assert _pkcs7_unpad(padded) == data + + def test_pad_block_aligned(self): + data = b"x" * 32 + padded = _pkcs7_pad(data) + assert len(padded) == 64 # full padding block added + assert _pkcs7_unpad(padded) == data + + +class TestWeChatCrypto: + """Test the encryption/decryption roundtrip. + + Uses a deterministic 43-char EncodingAESKey. + """ + + # Skip encryption tests when no crypto backend is available + _has_crypto = False + try: + from Crypto.Cipher import AES as _aes # noqa: F401 + _has_crypto = True + except ImportError: + try: + import pyaes as _pyaes # noqa: F401 + _has_crypto = True + except ImportError: + pass + pytestmark = pytest.mark.skipif( + not _has_crypto, + reason="pycryptodome or pyaes required for encryption tests", + ) + + @pytest.fixture + def crypto(self): + # 43 base64 chars → 32 bytes AES key + key = "abcdefghijklmnopqrstuvwxyz0123456789ABCDEFG" + return WeChatCrypto( + token="test_token", + encoding_aes_key=key, + app_id="wx_test_app", + ) + + def test_encrypt_decrypt_roundtrip(self, crypto): + msg = "Hello WeChat!" + encrypted = crypto.encrypt(msg) + decrypted, app_id = crypto.decrypt(encrypted) + assert decrypted == msg + assert app_id == "wx_test_app" + + def test_verify_signature(self, crypto): + timestamp = "1609459200" + nonce = "abc123" + parts = sorted([crypto.token, timestamp, nonce]) + expected = hashlib.sha1("".join(parts).encode()).hexdigest() + assert crypto.verify_signature(expected, timestamp, nonce) + assert not crypto.verify_signature("wrong", timestamp, nonce) + + def test_verify_signature_with_encrypt(self, crypto): + timestamp = "1609459200" + nonce = "abc123" + encrypt = "some_encrypted_data" + parts = sorted([crypto.token, timestamp, nonce, encrypt]) + expected = hashlib.sha1("".join(parts).encode()).hexdigest() + assert crypto.verify_signature(expected, timestamp, nonce, encrypt) + + def test_generate_signature(self, crypto): + encrypt = "test_encrypted" + timestamp = "1609459200" + nonce = "abc" + sig = crypto.generate_signature(encrypt, timestamp, nonce) + parts = sorted([crypto.token, timestamp, nonce, encrypt]) + expected = hashlib.sha1("".join(parts).encode()).hexdigest() + assert sig == expected + + def test_wrap_encrypted_reply(self, crypto): + msg = "Reply" + xml_reply = crypto.wrap_encrypted_reply(msg) + assert "" in xml_reply + assert "" in xml_reply + assert "" in xml_reply + assert "" in xml_reply + + # Parse and verify the encrypted content decrypts back + root = ET.fromstring(xml_reply) + encrypt = root.find("Encrypt").text + decrypted, app_id = crypto.decrypt(encrypt) + assert decrypted == msg + + +# ── Message processing tests ────────────────────────────────────── + +class TestMessageProcessing: + """Test the _process_message method with various XML payloads.""" + + def _make_channel(self): + config = WeComConfig( + corp_id="corp", agent_id="1", secret="s", + ) + return WeChatChannel(config, backend="wecom") + + def test_text_message_queued(self): + channel = self._make_channel() + + async def _test(): + await channel._process_message({ + "MsgType": "text", + "Content": "Hello!", + "FromUserName": "user1", + "ToUserName": "bot", + "MsgId": "100", + "CreateTime": str(int(time.time())), + }) + # Check message was enqueued + assert not channel._queue.empty() + msg = await asyncio.wait_for(channel._queue.get(), timeout=1.0) + assert msg.content == "Hello!" + assert msg.sender_id == "user1" + assert msg.channel == "wechat" + + _run(_test()) + + def test_location_message(self): + channel = self._make_channel() + + async def _test(): + await channel._process_message({ + "MsgType": "location", + "Location_X": "39.9", + "Location_Y": "116.4", + "Label": "Beijing", + "FromUserName": "user1", + "ToUserName": "bot", + "MsgId": "101", + "CreateTime": str(int(time.time())), + }) + msg = await asyncio.wait_for(channel._queue.get(), timeout=1.0) + assert "Beijing" in msg.content + assert "39.9" in msg.content + + _run(_test()) + + def test_voice_recognition(self): + channel = self._make_channel() + + async def _test(): + await channel._process_message({ + "MsgType": "voice", + "Recognition": "你好世界", + "FromUserName": "user1", + "ToUserName": "bot", + "MsgId": "102", + "CreateTime": str(int(time.time())), + }) + msg = await asyncio.wait_for(channel._queue.get(), timeout=1.0) + assert "你好世界" in msg.content + + _run(_test()) + + def test_link_message(self): + channel = self._make_channel() + + async def _test(): + await channel._process_message({ + "MsgType": "link", + "Title": "Test Link", + "Description": "A description", + "Url": "https://example.com", + "FromUserName": "user1", + "ToUserName": "bot", + "MsgId": "103", + "CreateTime": str(int(time.time())), + }) + msg = await asyncio.wait_for(channel._queue.get(), timeout=1.0) + assert "Test Link" in msg.content + assert "https://example.com" in msg.content + + _run(_test()) + + def test_subscribe_event(self): + channel = self._make_channel() + + async def _test(): + await channel._process_message({ + "MsgType": "event", + "Event": "subscribe", + "FromUserName": "user1", + "ToUserName": "bot", + "MsgId": "", + "CreateTime": str(int(time.time())), + }) + msg = await asyncio.wait_for(channel._queue.get(), timeout=1.0) + assert "关注" in msg.content + + _run(_test()) + + def test_unsubscribe_ignored(self): + channel = self._make_channel() + + async def _test(): + await channel._process_message({ + "MsgType": "event", + "Event": "unsubscribe", + "FromUserName": "user1", + "ToUserName": "bot", + "MsgId": "", + "CreateTime": str(int(time.time())), + }) + assert channel._queue.empty() + + _run(_test()) + + def test_empty_message_ignored(self): + channel = self._make_channel() + + async def _test(): + await channel._process_message({ + "MsgType": "text", + "Content": "", + "FromUserName": "", + "ToUserName": "bot", + }) + assert channel._queue.empty() + + _run(_test()) + + +# ── Registration test ───────────────────────────────────────────── + +class TestChannelRegistration: + def test_wechat_registered(self): + from EvoScientist.channels.channel_manager import available_channels + channels = available_channels() + assert "wechat" in channels