feat(dingding): add dingtalk channel adapter implementation

实现钉钉官方的频道适配器，包含完整的消息收发、事件处理、媒体上传下载、流式响应以及webhook支持，覆盖了认证、签名校验、速率限制、健康检查等完整功能流程。

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

@@ -0,0 +1,3 @@
+from yuxi.channels.adapters.dingding.adapter import DingDingChannelAdapter
+
+__all__ = ["DingDingChannelAdapter"]

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

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},
+    }

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

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))

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,
+    )

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", ""),
+    )

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}"

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)

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)

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