From 86105d163d7648739c8a05cbf002303af36ddaff Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Tue, 12 May 2026 00:43:14 +0800 Subject: [PATCH] feat(dingding): add dingtalk channel adapter implementation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 实现钉钉官方的频道适配器，包含完整的消息收发、事件处理、媒体上传下载、流式响应以及webhook支持，覆盖了认证、签名校验、速率限制、健康检查等完整功能流程。 --- .../channels/adapters/dingding/__init__.py | 3 + .../channels/adapters/dingding/adapter.py | 567 ++++++++++++++++++ .../yuxi/channels/adapters/dingding/cards.py | 47 ++ .../channels/adapters/dingding/formatter.py | 39 ++ .../yuxi/channels/adapters/dingding/media.py | 100 +++ .../channels/adapters/dingding/normalizer.py | 212 +++++++ .../yuxi/channels/adapters/dingding/send.py | 189 ++++++ .../channels/adapters/dingding/session.py | 20 + .../yuxi/channels/adapters/dingding/sign.py | 32 + .../adapters/dingding/stream_handler.py | 67 +++ .../yuxi/channels/adapters/dingding/token.py | 79 +++ 11 files changed, 1355 insertions(+) create mode 100644 backend/package/yuxi/channels/adapters/dingding/__init__.py create mode 100644 backend/package/yuxi/channels/adapters/dingding/adapter.py create mode 100644 backend/package/yuxi/channels/adapters/dingding/cards.py create mode 100644 backend/package/yuxi/channels/adapters/dingding/formatter.py create mode 100644 backend/package/yuxi/channels/adapters/dingding/media.py create mode 100644 backend/package/yuxi/channels/adapters/dingding/normalizer.py create mode 100644 backend/package/yuxi/channels/adapters/dingding/send.py create mode 100644 backend/package/yuxi/channels/adapters/dingding/session.py create mode 100644 backend/package/yuxi/channels/adapters/dingding/sign.py create mode 100644 backend/package/yuxi/channels/adapters/dingding/stream_handler.py create mode 100644 backend/package/yuxi/channels/adapters/dingding/token.py diff --git a/backend/package/yuxi/channels/adapters/dingding/__init__.py b/backend/package/yuxi/channels/adapters/dingding/__init__.py new file mode 100644 index 00000000..e4c1832c --- /dev/null +++ b/backend/package/yuxi/channels/adapters/dingding/__init__.py @@ -0,0 +1,3 @@ +from yuxi.channels.adapters.dingding.adapter import DingDingChannelAdapter + +__all__ = ["DingDingChannelAdapter"] diff --git a/backend/package/yuxi/channels/adapters/dingding/adapter.py b/backend/package/yuxi/channels/adapters/dingding/adapter.py new file mode 100644 index 00000000..00cb825f --- /dev/null +++ b/backend/package/yuxi/channels/adapters/dingding/adapter.py @@ -0,0 +1,567 @@ +from __future__ import annotations + +import asyncio +import json +import logging +import time +from collections.abc import AsyncIterator +from typing import Any + +import httpx + +from yuxi.channels.base import BaseChannelAdapter +from yuxi.channels.capabilities import ChannelCapabilities +from yuxi.channels.meta import ChannelMeta +from yuxi.channels.exceptions import ChannelAuthenticationError +from yuxi.channels.models import ( + ChannelMessage, + ChannelResponse, + ChannelStatus, + ChannelType, + DeliveryResult, + HealthStatus, +) +from yuxi.channels.registry import register_builtin_adapter +from yuxi.utils.datetime_utils import utc_now_naive + +from .cards import TEXT_CHUNK_LIMIT +from .formatter import format_outbound +from .media import download_media as _download_media +from .media import send_media as _send_media +from .media import upload_media +from .normalizer import normalize_inbound +from .send import _send_via_webhook, send_card, send_markdown, send_text +from .sign import verify_webhook_signature as verify_sign +from .token import DingDingTokenManager + +logger = logging.getLogger(__name__) + +HAS_DINGTALK_STREAM = False +try: + import dingtalk_stream + + HAS_DINGTALK_STREAM = True +except ImportError: + dingtalk_stream = None # type: ignore[assignment] + +DINGDING_RATE_LIMIT_PER_MINUTE = 20 +STREAM_RECONNECT_DELAY_S = 5 +STREAM_MAX_RECONNECT_DELAY_S = 300 + + +@register_builtin_adapter +class DingDingChannelAdapter(BaseChannelAdapter): + channel_id = "dingding" + channel_type = ChannelType.DINGDING + + text_chunk_limit = TEXT_CHUNK_LIMIT + supports_markdown = True + supports_streaming = True + streaming_modes = ["off", "block"] + max_media_size_mb = 10 + + capabilities = ChannelCapabilities( + chat_types=["direct"], + reactions=True, + edit=True, + unsend=True, + reply=True, + media=True, + supports_markdown=True, + supports_streaming=True, + streaming_modes=["off", "block"], + text_chunk_limit=4096, + max_media_size_mb=10, + ) + meta = ChannelMeta(id="dingding", label="DingDing") + + webhook_path = "/api/channels/dingding/events" + + def __init__(self, config: dict[str, Any] | None = None): + super().__init__(config) + self._status = ChannelStatus.DISCONNECTED + self._token_manager: DingDingTokenManager | None = None + self._stream_client: object = None + self._stream_task: asyncio.Task | None = None + self._connected_at: float | None = None + self._app_key: str = "" + self._app_secret: str = "" + self._http_client: httpx.AsyncClient | None = None + self._send_semaphore = asyncio.Semaphore(DINGDING_RATE_LIMIT_PER_MINUTE) + self._rate_reset_task: asyncio.Task | None = None + self._pending_streams: dict[str, str] = {} + self._session_webhooks: dict[str, str] = {} + self._stream_lock = asyncio.Lock() + + async def connect(self) -> None: + if self._status == ChannelStatus.CONNECTED: + return + + self._status = ChannelStatus.CONNECTING + logger.info(f"[DingDing] Starting channel '{self.config.get('name', self.channel_id)}'") + + accounts = self.config.get("accounts", {}) + default_account = accounts.get("default", {}) + self._app_key = default_account.get("app_key", "") + self._app_secret = default_account.get("app_secret", "") + + if not self._app_key or not self._app_secret: + raise ChannelAuthenticationError() + + http_client = await self._get_http_client() + self._token_manager = DingDingTokenManager(self._app_key, self._app_secret, http_client=http_client) + + try: + await self._token_manager.get_token() + except Exception as e: + raise ChannelAuthenticationError() from e + + logger.info("[DingDing] Token acquired successfully") + + if self._rate_reset_task is None: + self._rate_reset_task = asyncio.create_task(self._reset_rate_limit_loop()) + + mode = self.config.get("mode", "stream") + if mode == "stream": + await self._connect_stream_mode() + elif mode == "webhook": + await self._connect_webhook_mode() + + self._status = ChannelStatus.CONNECTED + self._connected_at = time.time() + logger.info(f"[DingDing] Channel started, mode: {mode}") + + async def _connect_stream_mode(self) -> None: + if not HAS_DINGTALK_STREAM: + logger.warning("[DingDing] dingtalk-stream SDK not installed") + return + + credential = dingtalk_stream.Credential( + client_id=self._app_key, + client_secret=self._app_secret, + ) + + self._stream_client = dingtalk_stream.DingTalkStreamClient(credential) + + from .stream_handler import DingDingCallbackHandler, DingDingChatbotHandler + + chatbot_handler = DingDingChatbotHandler(self) + self._stream_client.register_callback_handler(dingtalk_stream.ChatbotMessage.TOPIC, chatbot_handler) + + callback_handler = DingDingCallbackHandler(self) + self._stream_client.register_callback_handler(dingtalk_stream.CallbackHandler.TOPIC, callback_handler) + + self._stream_task = asyncio.create_task(self._run_stream_with_reconnect()) + logger.info("[DingDing] Stream mode task started") + + async def _run_stream_with_reconnect(self) -> None: + reconnect_delay = STREAM_RECONNECT_DELAY_S + while self._status in (ChannelStatus.CONNECTED, ChannelStatus.CONNECTING): + try: + await self._stream_client.start_forever() + except asyncio.CancelledError: + break + except Exception as e: + logger.warning(f"[DingDing] Stream disconnected: {e}, reconnecting in {reconnect_delay}s") + await asyncio.sleep(reconnect_delay) + reconnect_delay = min(reconnect_delay * 2, STREAM_MAX_RECONNECT_DELAY_S) + else: + reconnect_delay = STREAM_RECONNECT_DELAY_S + + async def _connect_webhook_mode(self) -> None: + logger.info("[DingDing] Webhook mode ready") + + async def disconnect(self) -> None: + if self._status == ChannelStatus.DISCONNECTED: + return + + logger.info(f"[DingDing] Stopping channel '{self.config.get('name', self.channel_id)}'") + + if self._rate_reset_task and not self._rate_reset_task.done(): + self._rate_reset_task.cancel() + try: + await self._rate_reset_task + except asyncio.CancelledError: + pass + self._rate_reset_task = None + + if self._stream_task and not self._stream_task.done(): + self._stream_task.cancel() + try: + await self._stream_task + except asyncio.CancelledError: + pass + self._stream_task = None + + if self._stream_client and hasattr(self._stream_client, "stop"): + try: + self._stream_client.stop() + except Exception: + pass + self._stream_client = None + + if self._token_manager: + self._token_manager.invalidate() + self._token_manager = None + + if self._http_client and not self._http_client.is_closed: + await self._http_client.aclose() + self._http_client = None + + self._pending_streams.clear() + self._session_webhooks.clear() + self._connected_at = None + self._status = ChannelStatus.DISCONNECTED + + def normalize_inbound(self, raw: dict[str, Any]) -> ChannelMessage: + return normalize_inbound(self.channel_id, self.channel_type, raw) + + def format_outbound(self, response: ChannelResponse) -> dict[str, Any]: + metadata = response.metadata or {} + chat_type = "group" if response.identity.channel_chat_id.startswith("group_") else "direct" + + return format_outbound( + response.content, + chat_type=chat_type, + metadata={**metadata}, + ) + + async def send(self, response: ChannelResponse) -> DeliveryResult: + if not self._token_manager: + return DeliveryResult(success=False, error="Not connected") + + payload = self.format_outbound(response) + chat_id = response.identity.channel_chat_id + chat_type = "group" if chat_id.startswith("group_") else "direct" + + open_conversation_id = chat_id.replace("group_", "").replace("dm_", "") + accounts = self.config.get("accounts", {}) + robot_code = accounts.get("default", {}).get("robot_code", "") + + content = response.content + msg_key = payload.get("msgKey", "sampleText") + http_client = await self._get_http_client() + + thread_root_id = response.metadata.get("root_id") or response.metadata.get("thread_id") + + async with self._send_semaphore: + if msg_key == "sampleMarkdown": + return await send_markdown( + self._token_manager, + open_conversation_id, + robot_code, + "ForcePilot", + content, + chat_type=chat_type, + thread_root_id=thread_root_id, + http_client=http_client, + ) + elif msg_key not in ("sampleText",): + msg_param_raw = payload.get("msgParam", "{}") + msg_param = json.loads(msg_param_raw) if isinstance(msg_param_raw, str) else msg_param_raw + return await send_card( + self._token_manager, + open_conversation_id, + robot_code, + msg_key, + msg_param, + chat_type=chat_type, + thread_root_id=thread_root_id, + http_client=http_client, + ) + return await send_text( + self._token_manager, + open_conversation_id, + robot_code, + content, + chat_type=chat_type, + thread_root_id=thread_root_id, + http_client=http_client, + ) + + async def send_media(self, chat_id: str, media_type: str, data: Any) -> DeliveryResult: + if not self._token_manager: + return DeliveryResult(success=False, error="Not connected") + + accounts = self.config.get("accounts", {}) + robot_code = accounts.get("default", {}).get("robot_code", "") + open_conversation_id = chat_id.replace("group_", "").replace("dm_", "") + chat_type = "group" if chat_id.startswith("group_") else "direct" + http_client = await self._get_http_client() + + async with self._send_semaphore: + try: + filename = f"media_{int(time.time())}" + upload_resp = await upload_media( + self._token_manager, media_type, data, filename=filename, http_client=http_client + ) + media_id = upload_resp.get("media_id", upload_resp.get("mediaId", "")) + if not media_id: + return DeliveryResult(success=False, error="Failed to get media_id from upload") + return await _send_media( + self._token_manager, + open_conversation_id, + robot_code, + media_id, + media_type, + chat_type=chat_type, + http_client=http_client, + ) + except Exception as e: + logger.error(f"[DingDing] send_media failed: {e}") + return DeliveryResult(success=False, error=str(e)) + + async def send_stream_chunk(self, chat_id: str, msg_id: str, chunk: str, finished: bool) -> DeliveryResult: + if not self._token_manager: + return DeliveryResult(success=False, error="Not connected") + + async with self._stream_lock: + if not finished: + existing = self._pending_streams.get(chat_id, "") + self._pending_streams[chat_id] = existing + chunk + return DeliveryResult(success=True, message_id=msg_id or "pending") + + final_text = self._pending_streams.pop(chat_id, chunk) + + accounts = self.config.get("accounts", {}) + robot_code = accounts.get("default", {}).get("robot_code", "") + open_conversation_id = chat_id.replace("group_", "").replace("dm_", "") + chat_type = "group" if chat_id.startswith("group_") else "direct" + mode = self.config.get("mode", "stream") + + webhook_url = self._session_webhooks.pop(chat_id.replace("group_", "").replace("dm_", ""), "") + + if mode == "stream" and webhook_url: + try: + http_client = await self._get_http_client() + async with self._send_semaphore: + return await _send_via_webhook( + http_client, webhook_url, robot_code, final_text, open_conversation_id + ) + except Exception as e: + logger.warning(f"[DingDing] Webhook reply failed, falling back to REST API: {e}") + + http_client = await self._get_http_client() + async with self._send_semaphore: + return await send_markdown( + self._token_manager, + open_conversation_id, + robot_code, + "ForcePilot", + final_text, + chat_type=chat_type, + http_client=http_client, + ) + + async def edit_message(self, chat_id: str, msg_id: str, content: str) -> DeliveryResult: + if not self._token_manager: + return DeliveryResult(success=False, error="Not connected") + + try: + token = await self._token_manager.get_token() + http_client = await self._get_http_client() + url = f"https://api.dingtalk.com/v1.0/robot/messages/{msg_id}" + + headers = { + "x-acs-dingtalk-access-token": token, + "Content-Type": "application/json", + } + + accounts = self.config.get("accounts", {}) + default_account = accounts.get("default", {}) + + payload = { + "msgParam": json.dumps({"content": content}, ensure_ascii=False), + "msgKey": "sampleMarkdown", + "openConversationId": chat_id.replace("group_", "").replace("dm_", ""), + "robotCode": default_account.get("robot_code", ""), + } + + resp = await http_client.put(url, json=payload, headers=headers) + if resp.status_code != 200: + return DeliveryResult(success=False, error=f"DingDing API HTTP {resp.status_code}") + + data = resp.json() + return DeliveryResult(success=True, message_id=data.get("processQueryKey", msg_id)) + except Exception as e: + logger.error(f"[DingDing] edit_message failed: {e}") + return DeliveryResult(success=False, error=str(e)) + + async def delete_message(self, chat_id: str, msg_id: str) -> DeliveryResult: + if not self._token_manager: + return DeliveryResult(success=False, error="Not connected") + + try: + token = await self._token_manager.get_token() + http_client = await self._get_http_client() + + accounts = self.config.get("accounts", {}) + robot_code = accounts.get("default", {}).get("robot_code", "") + open_conversation_id = chat_id.replace("group_", "").replace("dm_", "") + + url = "https://api.dingtalk.com/v1.0/robot/groupMessages/recall" + + headers = { + "x-acs-dingtalk-access-token": token, + "Content-Type": "application/json", + } + payload = { + "openConversationId": open_conversation_id, + "robotCode": robot_code, + "processQueryKeys": [msg_id], + } + + resp = await http_client.post(url, json=payload, headers=headers) + if resp.status_code != 200: + return DeliveryResult(success=False, error=f"DingDing API HTTP {resp.status_code}: {resp.text[:200]}") + + return DeliveryResult(success=True) + except Exception as e: + logger.error(f"[DingDing] delete_message failed: {e}") + return DeliveryResult(success=False, error=str(e)) + + async def send_reaction(self, chat_id: str, msg_id: str, emoji: str) -> DeliveryResult: + if not self._token_manager: + return DeliveryResult(success=False, error="Not connected") + + emotion_map = { + "👍": ("101", "like"), + "👎": ("102", "dislike"), + "❤️": ("103", "heart"), + "😂": ("104", "laugh"), + "😮": ("105", "surprise"), + "😢": ("106", "sad"), + "👀": ("107", "looking"), + "✅": ("108", "done"), + "❌": ("109", "error"), + "🤔": ("110", "thinking"), + "🎉": ("111", "celebrate"), + "🔥": ("112", "fire"), + "💯": ("113", "hundred"), + "🙏": ("114", "pray"), + } + + emotion_info = emotion_map.get(emoji) + if not emotion_info: + return DeliveryResult(success=False, error=f"Unsupported emoji: {emoji}") + + emotion_type, emotion_name = emotion_info + + try: + token = await self._token_manager.get_token() + http_client = await self._get_http_client() + + accounts = self.config.get("accounts", {}) + robot_code = accounts.get("default", {}).get("robot_code", "") + open_conversation_id = chat_id.replace("group_", "").replace("dm_", "") + + url = "https://api.dingtalk.com/v1.0/robot/emotion/reply" + headers = { + "x-acs-dingtalk-access-token": token, + "Content-Type": "application/json", + } + payload = { + "robotCode": robot_code, + "openMsgId": msg_id, + "openConversationId": open_conversation_id, + "emotionType": int(emotion_type), + "emotionName": emotion_name, + } + + resp = await http_client.post(url, json=payload, headers=headers) + if resp.status_code != 200: + return DeliveryResult( + success=False, error=f"DingDing emotion API HTTP {resp.status_code}: {resp.text[:200]}" + ) + + return DeliveryResult(success=True) + except Exception as e: + logger.error(f"[DingDing] send_reaction failed: {e}") + return DeliveryResult(success=False, error=str(e)) + + async def download_media(self, file_id: str) -> bytes: + if not self._token_manager: + raise RuntimeError("Not connected") + + http_client = await self._get_http_client() + return await _download_media(self._token_manager, file_id, http_client=http_client) + + async def health_check(self) -> HealthStatus: + if not self._token_manager: + return HealthStatus(status="unhealthy", last_error="Not connected") + + try: + token = await self._token_manager.get_token() + return HealthStatus( + status="healthy" if token else "degraded", + metadata={ + "mode": self.config.get("mode", "stream"), + "adapter_status": self._status.value, + }, + last_connected_at=utc_now_naive() if token else None, + ) + except Exception as e: + return HealthStatus(status="unhealthy", last_error=str(e)) + + async def verify_webhook_signature(self, headers: dict, body: bytes) -> bool: + accounts = self.config.get("accounts", {}) + app_secret = accounts.get("default", {}).get("app_secret", "") + if not app_secret: + logger.error("[DingDing] Webhook signature verification failed: no app_secret configured") + return False + return verify_sign(headers, app_secret) + + async def get_user_info(self, channel_user_id: str) -> dict[str, Any]: + if not self._token_manager: + return {} + try: + token = await self._token_manager.get_token() + http_client = await self._get_http_client() + resp = await http_client.get( + f"https://api.dingtalk.com/v1.0/contact/users/{channel_user_id}", + headers={"x-acs-dingtalk-access-token": token}, + ) + if resp.status_code == 200: + data = resp.json() + return { + "user_id": data.get("openId", data.get("unionId", channel_user_id)), + "name": data.get("nick", data.get("name", "")), + "avatar_url": data.get("avatarUrl", data.get("avatar", "")), + "email": data.get("email", ""), + "mobile": data.get("mobile", ""), + "raw": data, + } + except Exception: + logger.warning(f"[DingDing] Failed to get user info for {channel_user_id}", exc_info=True) + return {} + + async def receive(self) -> AsyncIterator[ChannelMessage]: + if False: + yield + + async def _refresh_token_if_needed(self) -> bool: + if not self._token_manager: + return False + try: + await self._token_manager.get_token() + return True + except Exception: + return False + + async def _get_http_client(self) -> httpx.AsyncClient: + if self._http_client is None or self._http_client.is_closed: + self._http_client = httpx.AsyncClient( + timeout=httpx.Timeout(15), + limits=httpx.Limits(max_keepalive_connections=5, max_connections=20), + ) + return self._http_client + + async def _reset_rate_limit_loop(self) -> None: + while True: + await asyncio.sleep(60) + old_sem, self._send_semaphore = self._send_semaphore, asyncio.Semaphore(DINGDING_RATE_LIMIT_PER_MINUTE) + for _ in range(DINGDING_RATE_LIMIT_PER_MINUTE): + try: + old_sem.release() + except ValueError: + break diff --git a/backend/package/yuxi/channels/adapters/dingding/cards.py b/backend/package/yuxi/channels/adapters/dingding/cards.py new file mode 100644 index 00000000..24a5404e --- /dev/null +++ b/backend/package/yuxi/channels/adapters/dingding/cards.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +from typing import Any + +TEXT_CHUNK_LIMIT = 30000 + + +def build_dingding_text_payload(content: str) -> dict[str, Any]: + return {"content": str(content)[:TEXT_CHUNK_LIMIT]} + + +def build_dingding_markdown_payload(title: str, text: str) -> dict[str, Any]: + return { + "title": str(title), + "text": str(text), + } + + +def build_dingding_action_card( + title: str, + text: str, + *, + single_title: str | None = None, + single_url: str | None = None, + buttons: list[dict[str, str]] | None = None, + btn_orientation: str = "0", +) -> dict[str, Any]: + msg_key = "sampleActionCard1" if single_title else "sampleActionCard2" + params: dict[str, Any] = { + "title": str(title), + "text": str(text), + } + if single_title and single_url: + params["singleTitle"] = single_title + params["singleURL"] = single_url + if buttons: + params["btns"] = buttons + params["btnOrientation"] = btn_orientation + + return {"msg_key": msg_key, "msg_param": params} + + +def build_dingding_feed_card(links: list[dict[str, str]]) -> dict[str, Any]: + return { + "msg_key": "sampleFeedCard", + "msg_param": {"links": links}, + } diff --git a/backend/package/yuxi/channels/adapters/dingding/formatter.py b/backend/package/yuxi/channels/adapters/dingding/formatter.py new file mode 100644 index 00000000..b286a0b0 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/dingding/formatter.py @@ -0,0 +1,39 @@ +from __future__ import annotations + +import json +from typing import Any + + +def format_outbound( + content: str, + chat_type: str = "direct", + metadata: dict[str, Any] | None = None, +) -> dict[str, Any]: + meta = metadata or {} + + msg_key = "sampleText" + msg_param: dict[str, Any] = {"content": content} + + if meta.get("use_markdown"): + msg_key = "sampleMarkdown" + msg_param = { + "title": meta.get("title", "ForcePilot"), + "text": content, + } + + if meta.get("msg_key"): + msg_key = meta["msg_key"] + if meta.get("msg_param"): + msg_param = meta["msg_param"] + + result: dict[str, Any] = { + "msgKey": msg_key, + "msgParam": json.dumps(msg_param, ensure_ascii=False), + } + + if meta.get("openConversationId"): + result["openConversationId"] = meta["openConversationId"] + if meta.get("robotCode"): + result["robotCode"] = meta["robotCode"] + + return result diff --git a/backend/package/yuxi/channels/adapters/dingding/media.py b/backend/package/yuxi/channels/adapters/dingding/media.py new file mode 100644 index 00000000..ddcc3723 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/dingding/media.py @@ -0,0 +1,100 @@ +from __future__ import annotations + +import logging +from typing import Any + +import httpx + +from yuxi.channels.models import DeliveryResult + +from .token import DingDingTokenManager + +logger = logging.getLogger(__name__) + + +async def upload_media( + token_manager: DingDingTokenManager, + media_type: str, + media_data: bytes, + filename: str = "file", + *, + http_client: httpx.AsyncClient | None = None, +) -> dict[str, Any]: + token = await token_manager.get_token() + url = "https://api.dingtalk.com/v1.0/media/upload" + + headers = {"x-acs-dingtalk-access-token": token} + files = {"media": (filename, media_data)} + params = {"type": media_type} + + if http_client is not None: + resp = await http_client.post(url, headers=headers, files=files, params=params) + else: + async with httpx.AsyncClient(timeout=httpx.Timeout(30)) as client: + resp = await client.post(url, headers=headers, files=files, params=params) + + if resp.status_code != 200: + logger.error(f"upload_media HTTP {resp.status_code}: {resp.text[:300]}") + raise RuntimeError(f"Media upload failed: HTTP {resp.status_code}") + return resp.json() + + +async def download_media( + token_manager: DingDingTokenManager, + download_code: str, + *, + http_client: httpx.AsyncClient | None = None, +) -> bytes: + token = await token_manager.get_token() + url = "https://api.dingtalk.com/v1.0/media/download" + + headers = {"x-acs-dingtalk-access-token": token} + params = {"downloadCode": download_code} + + if http_client is not None: + resp = await http_client.get(url, headers=headers, params=params) + else: + async with httpx.AsyncClient(timeout=httpx.Timeout(30)) as client: + resp = await client.get(url, headers=headers, params=params) + + if resp.status_code != 200: + logger.error(f"download_media HTTP {resp.status_code}: {resp.text[:300]}") + raise RuntimeError(f"Media download failed: HTTP {resp.status_code}") + + return resp.content + + +async def send_media( + token_manager: DingDingTokenManager, + open_conversation_id: str, + robot_code: str, + media_id: str, + media_type: str, + *, + chat_type: str = "direct", + http_client: httpx.AsyncClient | None = None, +) -> DeliveryResult: + try: + msg_key_map = { + "image": "sampleImageMsg", + "voice": "sampleAudio", + "file": "sampleFile", + "video": "sampleVideo", + } + msg_key = msg_key_map.get(media_type, "sampleFile") + msg_param = {"mediaId": media_id} + + from .send import _send_msg + + return await _send_msg( + token_manager, + open_conversation_id, + robot_code, + msg_key, + msg_param, + chat_type=chat_type, + http_client=http_client, + ) + except Exception as e: + logger.error(f"send_media failed: {e}") + return DeliveryResult(success=False, error=str(e)) diff --git a/backend/package/yuxi/channels/adapters/dingding/normalizer.py b/backend/package/yuxi/channels/adapters/dingding/normalizer.py new file mode 100644 index 00000000..a7b4fa27 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/dingding/normalizer.py @@ -0,0 +1,212 @@ +from __future__ import annotations + +import json +from typing import Any + +from yuxi.channels.models import ( + Attachment, + ChannelIdentity, + ChannelMessage, + ChannelType, + ChatType, + EventType, + MentionsInfo, + MessageType, +) + +from .session import generate_dingding_chat_id, resolve_dingding_chat_type + + +def map_event_type(raw: dict) -> EventType: + if raw.get("event_type") == "card_action": + return EventType.CARD_ACTION + if raw.get("action_from") == "card_callback": + return EventType.CARD_ACTION + if raw.get("robotCode"): + return EventType.BOT_ADDED + return EventType.MESSAGE_RECEIVED + + +def map_msg_type(raw: dict) -> MessageType: + msg_type = raw.get("msgtype", "text") + _map: dict[str, MessageType] = { + "text": MessageType.TEXT, + "image": MessageType.IMAGE, + "file": MessageType.FILE, + "voice": MessageType.AUDIO, + "video": MessageType.VIDEO, + "markdown": MessageType.TEXT, + "link": MessageType.TEXT, + "actionCard": MessageType.CARD, + "feedCard": MessageType.CARD, + "oa": MessageType.TEXT, + "sticker": MessageType.STICKER, + } + return _map.get(msg_type, MessageType.TEXT) + + +def _parse_content(content: Any) -> dict: + if isinstance(content, dict): + return content + if isinstance(content, str): + try: + return json.loads(content) + except (json.JSONDecodeError, TypeError): + return {} + return {} + + +def _describe_media(msg_type: str, parsed: dict, raw: dict | None = None) -> str: + if msg_type == "image": + return "[图片]" + if msg_type == "file": + filename = parsed.get("fileName", parsed.get("file_name", "")) + if not filename and raw: + filename = raw.get("fileName", "") + return f"[文件: {filename}]" if filename else "[文件]" + if msg_type == "voice": + return "[语音]" + if msg_type == "video": + return "[视频]" + if msg_type == "sticker": + return "[贴纸]" + return "" + + +def extract_text(raw: dict) -> str: + msg_type = raw.get("msgtype", "text") + text_block = raw.get("text", {}) + + if msg_type in ("image", "file", "voice", "video", "sticker"): + parsed = _parse_content(raw.get("content", text_block)) + return _describe_media(msg_type, parsed, raw) + + if isinstance(text_block, dict): + return text_block.get("content", "") + if isinstance(text_block, str): + return text_block + try: + return str(text_block) if text_block else "" + except Exception: + return "" + + +def extract_attachments(raw: dict) -> list[Attachment]: + msg_type = raw.get("msgtype", "text") + content = raw.get("content", {}) + + if msg_type == "image": + download_code = raw.get("downloadCode", "") or raw.get("download_code", "") + if download_code: + return [Attachment(type="image", file_id=download_code)] + return [] + + if msg_type == "file": + download_code = raw.get("downloadCode", "") or raw.get("download_code", "") + if download_code: + return [ + Attachment( + type="file", + file_id=download_code, + filename=raw.get("fileName", ""), + ) + ] + return [] + + if msg_type == "voice": + download_code = raw.get("downloadCode", "") or raw.get("download_code", "") + if download_code: + return [Attachment(type="audio", file_id=download_code)] + return [] + + if msg_type == "video": + download_code = raw.get("downloadCode", "") or raw.get("download_code", "") + if download_code: + return [Attachment(type="video", file_id=download_code)] + return [] + + if msg_type == "sticker": + parsed = _parse_content(content) + download_code = parsed.get("downloadCode", "") or parsed.get("download_code", "") + if download_code: + return [Attachment(type="sticker", file_id=download_code)] + return [] + + return [] + + +def extract_mentions(raw: dict) -> MentionsInfo | None: + if not raw.get("isGroupChat"): + return None + + mentioned_ids: list[str] = [] + is_at_all = raw.get("isAtAll", False) + + at_users = raw.get("atUsers", []) + if isinstance(at_users, list): + for u in at_users: + if isinstance(u, dict): + uid = u.get("dingtalkId", "") + if uid: + mentioned_ids.append(uid) + + if is_at_all: + mentioned_ids.append("@all") + + raw_text = extract_text(raw).strip() + + return MentionsInfo( + mentioned_user_ids=mentioned_ids, + is_bot_mentioned=bool(mentioned_ids), + raw_text=raw_text, + ) + + +def normalize_inbound( + channel_id: str, + channel_type: ChannelType, + raw_payload: dict[str, Any], +) -> ChannelMessage: + chat_type_str = resolve_dingding_chat_type(raw_payload) + chat_type = ChatType(chat_type_str) + + chat_id = generate_dingding_chat_id(raw_payload, chat_type_str) + sender_id = ( + raw_payload.get("senderId", "") + or raw_payload.get("senderStaffId", "") + or raw_payload.get("senderNick", "") + or "unknown" + ) + message_id = raw_payload.get("msgId", "") or raw_payload.get("openMsgId", "") + + event_type = map_event_type(raw_payload) + msg_type = map_msg_type(raw_payload) + content = extract_text(raw_payload) + attachments = extract_attachments(raw_payload) + mentions = extract_mentions(raw_payload) + + metadata: dict[str, Any] = { + "msgtype": raw_payload.get("msgtype", "text"), + "isGroupChat": raw_payload.get("isGroupChat", False), + } + if raw_payload.get("rootId"): + metadata["root_id"] = raw_payload["rootId"] + if raw_payload.get("conversationId"): + metadata["conversation_id"] = raw_payload["conversationId"] + + return ChannelMessage( + identity=ChannelIdentity( + channel_id=channel_id, + channel_type=channel_type, + channel_user_id=sender_id, + channel_chat_id=chat_id, + channel_message_id=message_id, + ), + event_type=event_type, + message_type=msg_type, + chat_type=chat_type, + content=content, + attachments=attachments, + mentions=mentions, + metadata=metadata, + ) diff --git a/backend/package/yuxi/channels/adapters/dingding/send.py b/backend/package/yuxi/channels/adapters/dingding/send.py new file mode 100644 index 00000000..28b8df5d --- /dev/null +++ b/backend/package/yuxi/channels/adapters/dingding/send.py @@ -0,0 +1,189 @@ +from __future__ import annotations + +import json +import logging +import uuid +from typing import Any + +import httpx + +from yuxi.channels.models import DeliveryResult + +from .token import DingDingTokenManager + +logger = logging.getLogger(__name__) + + +async def send_text( + token_manager: DingDingTokenManager, + open_conversation_id: str, + robot_code: str, + content: str, + *, + chat_type: str = "direct", + msg_uuid: str | None = None, + thread_root_id: str | None = None, + http_client: httpx.AsyncClient | None = None, +) -> DeliveryResult: + try: + return await _send_msg( + token_manager, + open_conversation_id, + robot_code, + "sampleText", + {"content": content}, + chat_type=chat_type, + msg_uuid=msg_uuid, + thread_root_id=thread_root_id, + http_client=http_client, + ) + except Exception as e: + logger.error(f"send_text failed: {e}") + return DeliveryResult(success=False, error=str(e)) + + +async def send_markdown( + token_manager: DingDingTokenManager, + open_conversation_id: str, + robot_code: str, + title: str, + text: str, + *, + chat_type: str = "direct", + msg_uuid: str | None = None, + thread_root_id: str | None = None, + http_client: httpx.AsyncClient | None = None, +) -> DeliveryResult: + try: + return await _send_msg( + token_manager, + open_conversation_id, + robot_code, + "sampleMarkdown", + {"title": title, "text": text}, + chat_type=chat_type, + msg_uuid=msg_uuid, + thread_root_id=thread_root_id, + http_client=http_client, + ) + except Exception as e: + logger.error(f"send_markdown failed: {e}") + return DeliveryResult(success=False, error=str(e)) + + +async def send_card( + token_manager: DingDingTokenManager, + open_conversation_id: str, + robot_code: str, + msg_key: str, + msg_param: dict[str, Any], + *, + chat_type: str = "direct", + msg_uuid: str | None = None, + thread_root_id: str | None = None, + http_client: httpx.AsyncClient | None = None, +) -> DeliveryResult: + try: + return await _send_msg( + token_manager, + open_conversation_id, + robot_code, + msg_key, + msg_param, + chat_type=chat_type, + msg_uuid=msg_uuid, + thread_root_id=thread_root_id, + http_client=http_client, + ) + except Exception as e: + logger.error(f"send_card failed: {e}") + return DeliveryResult(success=False, error=str(e)) + + +async def _send_msg( + token_manager: DingDingTokenManager, + open_conversation_id: str, + robot_code: str, + msg_key: str, + msg_param: dict[str, Any], + *, + chat_type: str = "direct", + msg_uuid: str | None = None, + thread_root_id: str | None = None, + http_client: httpx.AsyncClient | None = None, +) -> DeliveryResult: + token = await token_manager.get_token() + + if chat_type == "group": + url = "https://api.dingtalk.com/v1.0/robot/groupMessages/send" + else: + url = "https://api.dingtalk.com/v1.0/robot/privateChatMessages/send" + + payload: dict[str, Any] = { + "msgParam": json.dumps(msg_param, ensure_ascii=False), + "msgKey": msg_key, + "openConversationId": open_conversation_id, + "robotCode": robot_code, + "msgUuid": msg_uuid or str(uuid.uuid4()), + } + + if thread_root_id: + payload["rootId"] = thread_root_id + + headers = { + "x-acs-dingtalk-access-token": token, + "Content-Type": "application/json", + } + + if http_client is not None: + resp = await http_client.post(url, json=payload, headers=headers) + else: + async with httpx.AsyncClient(timeout=httpx.Timeout(15)) as client: + resp = await client.post(url, json=payload, headers=headers) + + if resp.status_code != 200: + logger.error(f"_send_msg HTTP {resp.status_code}: {resp.text[:300]} url={url}") + return DeliveryResult( + success=False, + error=f"DingDing API HTTP {resp.status_code}: {resp.text[:200]}", + ) + + data = resp.json() + return DeliveryResult( + success=True, + message_id=data.get("processQueryKey") or data.get("messageId", ""), + ) + + +async def _send_via_webhook( + http_client: httpx.AsyncClient, + webhook_url: str, + robot_code: str, + content: str, + open_conversation_id: str, +) -> DeliveryResult: + payload = { + "msgKey": "sampleMarkdown", + "msgParam": json.dumps({"content": content}, ensure_ascii=False), + "robotCode": robot_code, + "openConversationId": open_conversation_id, + } + + resp = await http_client.post( + webhook_url, + json=payload, + headers={"Content-Type": "application/json"}, + timeout=15.0, + ) + + if resp.status_code != 200: + return DeliveryResult( + success=False, + error=f"Webhook HTTP {resp.status_code}: {resp.text[:200]}", + ) + + data = resp.json() + return DeliveryResult( + success=True, + message_id=data.get("processQueryKey") or data.get("messageId", ""), + ) diff --git a/backend/package/yuxi/channels/adapters/dingding/session.py b/backend/package/yuxi/channels/adapters/dingding/session.py new file mode 100644 index 00000000..81fedd5e --- /dev/null +++ b/backend/package/yuxi/channels/adapters/dingding/session.py @@ -0,0 +1,20 @@ +from __future__ import annotations + + +def generate_dingding_chat_id(raw: dict, chat_type: str) -> str: + if chat_type == "direct": + sender_id = raw.get("senderId", "") + return f"dm_{sender_id}" + conversation_id = raw.get("conversationId", "") + return f"group_{conversation_id}" + + +def resolve_dingding_chat_type(raw: dict) -> str: + is_group = raw.get("isGroupChat", False) + return "group" if is_group else "direct" + + +def generate_thread_key(channel_id: str, chat_id: str) -> str: + if chat_id.startswith("dm_"): + return f"{channel_id}:direct:{chat_id}" + return f"{channel_id}:group:{chat_id}" diff --git a/backend/package/yuxi/channels/adapters/dingding/sign.py b/backend/package/yuxi/channels/adapters/dingding/sign.py new file mode 100644 index 00000000..79071154 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/dingding/sign.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +import base64 +import hashlib +import hmac +import time +import urllib.parse + + +def compute_dingtalk_sign(timestamp_ms: str, app_secret: str) -> str: + secret_enc = app_secret.encode("utf-8") + string_to_sign = f"{timestamp_ms}\n{app_secret}".encode() + hmac_code = hmac.new(secret_enc, string_to_sign, digestmod=hashlib.sha256).digest() + return urllib.parse.quote_plus(base64.b64encode(hmac_code)) + + +def verify_webhook_signature(headers: dict, app_secret: str) -> bool: + request_timestamp = headers.get("timestamp", "") + request_sign = headers.get("sign", "") + + if not request_timestamp or not request_sign: + return True + + try: + ts = int(request_timestamp) / 1000 + if abs(time.time() - ts) > 3600: + return False + except (ValueError, TypeError): + return False + + expected_sign = compute_dingtalk_sign(request_timestamp, app_secret) + return hmac.compare_digest(request_sign, expected_sign) diff --git a/backend/package/yuxi/channels/adapters/dingding/stream_handler.py b/backend/package/yuxi/channels/adapters/dingding/stream_handler.py new file mode 100644 index 00000000..1124af03 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/dingding/stream_handler.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING + +logger = logging.getLogger(__name__) + +if TYPE_CHECKING: + from yuxi.channels.adapters.dingding.adapter import DingDingChannelAdapter + +HAS_DINGTALK_STREAM = False +try: + import dingtalk_stream # noqa: F401 + + HAS_DINGTALK_STREAM = True +except ImportError: + pass + + +class DingDingChatbotHandler: + def __init__(self, adapter: DingDingChannelAdapter): + self._adapter = adapter + + async def process(self, callback: object) -> tuple[int, str]: + if not HAS_DINGTALK_STREAM: + return 200, "OK" + + from dingtalk_stream import AckMessage + + try: + raw_data = getattr(callback, "data", {}) + session_webhook = getattr(callback, "sessionWebhook", None) or raw_data.get("sessionWebhook", "") + + if session_webhook: + self._adapter._session_webhooks[raw_data.get("conversationId", "")] = session_webhook + + msg = self._adapter.normalize_inbound(raw_data) + await self._adapter._handle_message(msg) + return AckMessage.STATUS_OK, "OK" + except Exception as e: + logger.error(f"[DingDing] ChatbotHandler error: {e}") + return AckMessage.STATUS_INTERNAL_SERVER_ERROR, str(e) + + +class DingDingCallbackHandler: + """互动卡片回传事件处理器""" + + def __init__(self, adapter: DingDingChannelAdapter): + self._adapter = adapter + + async def process(self, callback: object) -> tuple[int, str]: + if not HAS_DINGTALK_STREAM: + return 200, "OK" + + from dingtalk_stream import AckMessage + + try: + raw_data = getattr(callback, "data", {}) + from yuxi.channels.models import EventType + + msg = self._adapter.normalize_inbound(raw_data) + msg.event_type = EventType.CARD_ACTION + await self._adapter._handle_message(msg) + return AckMessage.STATUS_OK, "OK" + except Exception as e: + logger.error(f"[DingDing] CallbackHandler error: {e}") + return AckMessage.STATUS_INTERNAL_SERVER_ERROR, str(e) diff --git a/backend/package/yuxi/channels/adapters/dingding/token.py b/backend/package/yuxi/channels/adapters/dingding/token.py new file mode 100644 index 00000000..f923ee54 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/dingding/token.py @@ -0,0 +1,79 @@ +from __future__ import annotations + +import asyncio +import time +from typing import Any + +import httpx + +from yuxi.utils.logging_config import logger + +TOKEN_REFRESH_MARGIN_S = 200 +TOKEN_VALIDITY_S = 7200 + + +class DingDingTokenManager: + def __init__(self, app_key: str, app_secret: str, http_client: httpx.AsyncClient | None = None): + self._app_key = app_key + self._app_secret = app_secret + self._access_token: str | None = None + self._token_expires_at: float = 0 + self._lock = asyncio.Lock() + self._http_client = http_client + + async def get_token(self) -> str: + if self._is_valid(): + return self._access_token # type: ignore[return-value] + + async with self._lock: + if self._is_valid(): + return self._access_token # type: ignore[return-value] + await self._refresh() + + if not self._access_token: + raise RuntimeError("Failed to obtain DingDing access token") + return self._access_token + + async def _refresh(self) -> None: + refresh_start = time.monotonic() + url = "https://api.dingtalk.com/v1.0/oauth2/accessToken" + payload: dict[str, Any] = { + "appKey": self._app_key, + "appSecret": self._app_secret, + } + + if self._http_client is not None: + resp = await self._http_client.post(url, json=payload) + else: + async with httpx.AsyncClient(timeout=httpx.Timeout(15)) as client: + resp = await client.post(url, json=payload) + + if resp.status_code != 200: + logger.error(f"[DingDing] Token refresh HTTP {resp.status_code}: {resp.text[:200]}") + raise RuntimeError(f"Token refresh failed: HTTP {resp.status_code}") + + data = resp.json() + token = data.get("accessToken") + if not token: + logger.error(f"[DingDing] Token refresh response missing accessToken: {data}") + raise RuntimeError("Token refresh response missing accessToken") + + expire_in = data.get("expireIn", TOKEN_VALIDITY_S) + self._access_token = token + self._token_expires_at = time.time() + expire_in - TOKEN_REFRESH_MARGIN_S + + elapsed = time.monotonic() - refresh_start + if elapsed > 2.0: + logger.warning(f"[DingDing] Token refresh took {elapsed:.1f}s") + logger.info(f"[DingDing] Token refreshed, expires in {expire_in - TOKEN_REFRESH_MARGIN_S}s") + + def _is_valid(self) -> bool: + return self._access_token is not None and time.time() < self._token_expires_at + + def invalidate(self) -> None: + self._access_token = None + self._token_expires_at = 0 + + async def close(self) -> None: + self.invalidate() + self._http_client = None