| 1 | """DingTalk/DingDing channel implementation using Stream Mode.""" |
| 2 | |
| 3 | import asyncio |
| 4 | import json |
| 5 | import mimetypes |
| 6 | import os |
| 7 | import time |
| 8 | import zipfile |
| 9 | from io import BytesIO |
| 10 | from pathlib import Path |
| 11 | from typing import Any |
| 12 | from urllib.parse import unquote, urlparse |
| 13 | |
| 14 | import httpx |
| 15 | from loguru import logger |
| 16 | from pydantic import Field |
| 17 | |
| 18 | from nanobot.bus.events import OutboundMessage |
| 19 | from nanobot.bus.queue import MessageBus |
| 20 | from nanobot.channels.base import BaseChannel |
| 21 | from nanobot.config.schema import Base |
| 22 | |
| 23 | try: |
| 24 | from dingtalk_stream import ( |
| 25 | AckMessage, |
| 26 | CallbackHandler, |
| 27 | CallbackMessage, |
| 28 | Credential, |
| 29 | DingTalkStreamClient, |
| 30 | ) |
| 31 | from dingtalk_stream.chatbot import ChatbotMessage |
| 32 | |
| 33 | DINGTALK_AVAILABLE = True |
| 34 | except ImportError: |
| 35 | DINGTALK_AVAILABLE = False |
| 36 | # Fallback so class definitions don't crash at module level |
| 37 | CallbackHandler = object # type: ignore[assignment,misc] |
| 38 | CallbackMessage = None # type: ignore[assignment,misc] |
| 39 | AckMessage = None # type: ignore[assignment,misc] |
| 40 | ChatbotMessage = None # type: ignore[assignment,misc] |
| 41 | |
| 42 | |
| 43 | class NanobotDingTalkHandler(CallbackHandler): |
| 44 | """ |
| 45 | Standard DingTalk Stream SDK Callback Handler. |
| 46 | Parses incoming messages and forwards them to the Nanobot channel. |
| 47 | """ |
| 48 | |
| 49 | def __init__(self, channel: "DingTalkChannel"): |
| 50 | super().__init__() |
| 51 | self.channel = channel |
| 52 | |
| 53 | async def process(self, message: CallbackMessage): |
| 54 | """Process incoming stream message.""" |
| 55 | try: |
| 56 | # Parse using SDK's ChatbotMessage for robust handling |
| 57 | chatbot_msg = ChatbotMessage.from_dict(message.data) |
| 58 | |
| 59 | # Extract text content; fall back to raw dict if SDK object is empty |
| 60 | content = "" |
| 61 | if chatbot_msg.text: |
| 62 | content = chatbot_msg.text.content.strip() |
| 63 | elif chatbot_msg.extensions.get("content", {}).get("recognition"): |
| 64 | content = chatbot_msg.extensions["content"]["recognition"].strip() |
| 65 | if not content: |
| 66 | content = message.data.get("text", {}).get("content", "").strip() |
| 67 | |
| 68 | # Handle file/image messages |
| 69 | file_paths = [] |
| 70 | if chatbot_msg.message_type == "picture" and chatbot_msg.image_content: |
| 71 | download_code = chatbot_msg.image_content.download_code |
| 72 | if download_code: |
| 73 | sender_uid = chatbot_msg.sender_staff_id or chatbot_msg.sender_id or "unknown" |
| 74 | fp = await self.channel._download_dingtalk_file(download_code, "image.jpg", sender_uid) |
| 75 | if fp: |
| 76 | file_paths.append(fp) |
| 77 | content = content or "[Image]" |
| 78 | |
| 79 | elif chatbot_msg.message_type == "file": |
| 80 | download_code = message.data.get("content", {}).get("downloadCode") or message.data.get("downloadCode") |
| 81 | fname = message.data.get("content", {}).get("fileName") or message.data.get("fileName") or "file" |
| 82 | if download_code: |
| 83 | sender_uid = chatbot_msg.sender_staff_id or chatbot_msg.sender_id or "unknown" |
| 84 | fp = await self.channel._download_dingtalk_file(download_code, fname, sender_uid) |
| 85 | if fp: |
| 86 | file_paths.append(fp) |
| 87 | content = content or "[File]" |
| 88 | |
| 89 | elif chatbot_msg.message_type == "richText" and chatbot_msg.rich_text_content: |
| 90 | rich_list = chatbot_msg.rich_text_content.rich_text_list or [] |
| 91 | for item in rich_list: |
| 92 | if not isinstance(item, dict): |
| 93 | continue |
| 94 | if item.get("type") == "text": |
| 95 | t = item.get("text", "").strip() |
| 96 | if t: |
| 97 | content = (content + " " + t).strip() if content else t |
| 98 | elif item.get("downloadCode"): |
| 99 | dc = item["downloadCode"] |
| 100 | fname = item.get("fileName") or "file" |
| 101 | sender_uid = chatbot_msg.sender_staff_id or chatbot_msg.sender_id or "unknown" |
| 102 | fp = await self.channel._download_dingtalk_file(dc, fname, sender_uid) |
| 103 | if fp: |
| 104 | file_paths.append(fp) |
| 105 | content = content or "[File]" |
| 106 | |
| 107 | if file_paths: |
| 108 | file_list = "\n".join("- " + p for p in file_paths) |
| 109 | content = content + "\n\nReceived files:\n" + file_list |
| 110 | |
| 111 | if not content: |
| 112 | logger.warning( |
| 113 | "Received empty or unsupported message type: {}", |
| 114 | chatbot_msg.message_type, |
| 115 | ) |
| 116 | return AckMessage.STATUS_OK, "OK" |
| 117 | |
| 118 | sender_id = chatbot_msg.sender_staff_id or chatbot_msg.sender_id |
| 119 | sender_name = chatbot_msg.sender_nick or "Unknown" |
| 120 | |
| 121 | conversation_type = message.data.get("conversationType") |
| 122 | conversation_id = ( |
| 123 | message.data.get("conversationId") |
| 124 | or message.data.get("openConversationId") |
| 125 | ) |
| 126 | |
| 127 | logger.info("Received DingTalk message from {} ({}): {}", sender_name, sender_id, content) |
| 128 | |
| 129 | # Forward to Nanobot via _on_message (non-blocking). |
| 130 | # Store reference to prevent GC before task completes. |
| 131 | task = asyncio.create_task( |
| 132 | self.channel._on_message( |
| 133 | content, |
| 134 | sender_id, |
| 135 | sender_name, |
| 136 | conversation_type, |
| 137 | conversation_id, |
| 138 | ) |
| 139 | ) |
| 140 | self.channel._background_tasks.add(task) |
| 141 | task.add_done_callback(self.channel._background_tasks.discard) |
| 142 | |
| 143 | return AckMessage.STATUS_OK, "OK" |
| 144 | |
| 145 | except Exception as e: |
| 146 | logger.error("Error processing DingTalk message: {}", e) |
| 147 | # Return OK to avoid retry loop from DingTalk server |
| 148 | return AckMessage.STATUS_OK, "Error" |
| 149 | |
| 150 | |
| 151 | class DingTalkConfig(Base): |
| 152 | """DingTalk channel configuration using Stream mode.""" |
| 153 | |
| 154 | enabled: bool = False |
| 155 | client_id: str = "" |
| 156 | client_secret: str = "" |
| 157 | allow_from: list[str] = Field(default_factory=list) |
| 158 | |
| 159 | |
| 160 | class DingTalkChannel(BaseChannel): |
| 161 | """ |
| 162 | DingTalk channel using Stream Mode. |
| 163 | |
| 164 | Uses WebSocket to receive events via `dingtalk-stream` SDK. |
| 165 | Uses direct HTTP API to send messages (SDK is mainly for receiving). |
| 166 | |
| 167 | Supports both private (1:1) and group chats. |
| 168 | Group chat_id is stored with a "group:" prefix to route replies back. |
| 169 | """ |
| 170 | |
| 171 | name = "dingtalk" |
| 172 | display_name = "DingTalk" |
| 173 | _IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".gif", ".bmp", ".webp"} |
| 174 | _AUDIO_EXTS = {".amr", ".mp3", ".wav", ".ogg", ".m4a", ".aac"} |
| 175 | _VIDEO_EXTS = {".mp4", ".mov", ".avi", ".mkv", ".webm"} |
| 176 | _ZIP_BEFORE_UPLOAD_EXTS = {".htm", ".html"} |
| 177 | |
| 178 | @classmethod |
| 179 | def default_config(cls) -> dict[str, Any]: |
| 180 | return DingTalkConfig().model_dump(by_alias=True) |
| 181 | |
| 182 | def __init__(self, config: Any, bus: MessageBus): |
| 183 | if isinstance(config, dict): |
| 184 | config = DingTalkConfig.model_validate(config) |
| 185 | super().__init__(config, bus) |
| 186 | self.config: DingTalkConfig = config |
| 187 | self._client: Any = None |
| 188 | self._http: httpx.AsyncClient | None = None |
| 189 | |
| 190 | # Access Token management for sending messages |
| 191 | self._access_token: str | None = None |
| 192 | self._token_expiry: float = 0 |
| 193 | |
| 194 | # Hold references to background tasks to prevent GC |
| 195 | self._background_tasks: set[asyncio.Task] = set() |
| 196 | |
| 197 | async def start(self) -> None: |
| 198 | """Start the DingTalk bot with Stream Mode.""" |
| 199 | try: |
| 200 | if not DINGTALK_AVAILABLE: |
| 201 | logger.error( |
| 202 | "DingTalk Stream SDK not installed. Run: pip install dingtalk-stream" |
| 203 | ) |
| 204 | return |
| 205 | |
| 206 | if not self.config.client_id or not self.config.client_secret: |
| 207 | logger.error("DingTalk client_id and client_secret not configured") |
| 208 | return |
| 209 | |
| 210 | self._running = True |
| 211 | self._http = httpx.AsyncClient() |
| 212 | |
| 213 | logger.info( |
| 214 | "Initializing DingTalk Stream Client with Client ID: {}...", |
| 215 | self.config.client_id, |
| 216 | ) |
| 217 | credential = Credential(self.config.client_id, self.config.client_secret) |
| 218 | self._client = DingTalkStreamClient(credential) |
| 219 | |
| 220 | # Register standard handler |
| 221 | handler = NanobotDingTalkHandler(self) |
| 222 | self._client.register_callback_handler(ChatbotMessage.TOPIC, handler) |
| 223 | |
| 224 | logger.info("DingTalk bot started with Stream Mode") |
| 225 | |
| 226 | # Reconnect loop: restart stream if SDK exits or crashes |
| 227 | while self._running: |
| 228 | try: |
| 229 | await self._client.start() |
| 230 | except Exception as e: |
| 231 | logger.warning("DingTalk stream error: {}", e) |
| 232 | if self._running: |
| 233 | logger.info("Reconnecting DingTalk stream in 5 seconds...") |
| 234 | await asyncio.sleep(5) |
| 235 | |
| 236 | except Exception as e: |
| 237 | logger.exception("Failed to start DingTalk channel: {}", e) |
| 238 | |
| 239 | async def stop(self) -> None: |
| 240 | """Stop the DingTalk bot.""" |
| 241 | self._running = False |
| 242 | # Close the shared HTTP client |
| 243 | if self._http: |
| 244 | await self._http.aclose() |
| 245 | self._http = None |
| 246 | # Cancel outstanding background tasks |
| 247 | for task in self._background_tasks: |
| 248 | task.cancel() |
| 249 | self._background_tasks.clear() |
| 250 | |
| 251 | async def _get_access_token(self) -> str | None: |
| 252 | """Get or refresh Access Token.""" |
| 253 | if self._access_token and time.time() < self._token_expiry: |
| 254 | return self._access_token |
| 255 | |
| 256 | url = "https://api.dingtalk.com/v1.0/oauth2/accessToken" |
| 257 | data = { |
| 258 | "appKey": self.config.client_id, |
| 259 | "appSecret": self.config.client_secret, |
| 260 | } |
| 261 | |
| 262 | if not self._http: |
| 263 | logger.warning("DingTalk HTTP client not initialized, cannot refresh token") |
| 264 | return None |
| 265 | |
| 266 | try: |
| 267 | resp = await self._http.post(url, json=data) |
| 268 | resp.raise_for_status() |
| 269 | res_data = resp.json() |
| 270 | self._access_token = res_data.get("accessToken") |
| 271 | # Expire 60s early to be safe |
| 272 | self._token_expiry = time.time() + int(res_data.get("expireIn", 7200)) - 60 |
| 273 | return self._access_token |
| 274 | except Exception as e: |
| 275 | logger.error("Failed to get DingTalk access token: {}", e) |
| 276 | return None |
| 277 | |
| 278 | @staticmethod |
| 279 | def _is_http_url(value: str) -> bool: |
| 280 | return urlparse(value).scheme in ("http", "https") |
| 281 | |
| 282 | def _guess_upload_type(self, media_ref: str) -> str: |
| 283 | ext = Path(urlparse(media_ref).path).suffix.lower() |
| 284 | if ext in self._IMAGE_EXTS: return "image" |
| 285 | if ext in self._AUDIO_EXTS: return "voice" |
| 286 | if ext in self._VIDEO_EXTS: return "video" |
| 287 | return "file" |
| 288 | |
| 289 | def _guess_filename(self, media_ref: str, upload_type: str) -> str: |
| 290 | name = os.path.basename(urlparse(media_ref).path) |
| 291 | return name or {"image": "image.jpg", "voice": "audio.amr", "video": "video.mp4"}.get(upload_type, "file.bin") |
| 292 | |
| 293 | @staticmethod |
| 294 | def _zip_bytes(filename: str, data: bytes) -> tuple[bytes, str, str]: |
| 295 | stem = Path(filename).stem or "attachment" |
| 296 | safe_name = filename or "attachment.bin" |
| 297 | zip_name = f"{stem}.zip" |
| 298 | buffer = BytesIO() |
| 299 | with zipfile.ZipFile(buffer, mode="w", compression=zipfile.ZIP_DEFLATED) as archive: |
| 300 | archive.writestr(safe_name, data) |
| 301 | return buffer.getvalue(), zip_name, "application/zip" |
| 302 | |
| 303 | def _normalize_upload_payload( |
| 304 | self, |
| 305 | filename: str, |
| 306 | data: bytes, |
| 307 | content_type: str | None, |
| 308 | ) -> tuple[bytes, str, str | None]: |
| 309 | ext = Path(filename).suffix.lower() |
| 310 | if ext in self._ZIP_BEFORE_UPLOAD_EXTS or content_type == "text/html": |
| 311 | logger.info( |
| 312 | "DingTalk does not accept raw HTML attachments, zipping {} before upload", |
| 313 | filename, |
| 314 | ) |
| 315 | return self._zip_bytes(filename, data) |
| 316 | return data, filename, content_type |
| 317 | |
| 318 | async def _read_media_bytes( |
| 319 | self, |
| 320 | media_ref: str, |
| 321 | ) -> tuple[bytes | None, str | None, str | None]: |
| 322 | if not media_ref: |
| 323 | return None, None, None |
| 324 | |
| 325 | if self._is_http_url(media_ref): |
| 326 | if not self._http: |
| 327 | return None, None, None |
| 328 | try: |
| 329 | resp = await self._http.get(media_ref, follow_redirects=True) |
| 330 | if resp.status_code >= 400: |
| 331 | logger.warning( |
| 332 | "DingTalk media download failed status={} ref={}", |
| 333 | resp.status_code, |
| 334 | media_ref, |
| 335 | ) |
| 336 | return None, None, None |
| 337 | content_type = (resp.headers.get("content-type") or "").split(";")[0].strip() |
| 338 | filename = self._guess_filename(media_ref, self._guess_upload_type(media_ref)) |
| 339 | return resp.content, filename, content_type or None |
| 340 | except httpx.TransportError as e: |
| 341 | logger.error("DingTalk media download network error ref={} err={}", media_ref, e) |
| 342 | raise |
| 343 | except Exception as e: |
| 344 | logger.error("DingTalk media download error ref={} err={}", media_ref, e) |
| 345 | return None, None, None |
| 346 | |
| 347 | try: |
| 348 | if media_ref.startswith("file://"): |
| 349 | parsed = urlparse(media_ref) |
| 350 | local_path = Path(unquote(parsed.path)) |
| 351 | else: |
| 352 | local_path = Path(os.path.expanduser(media_ref)) |
| 353 | if not local_path.is_file(): |
| 354 | logger.warning("DingTalk media file not found: {}", local_path) |
| 355 | return None, None, None |
| 356 | data = await asyncio.to_thread(local_path.read_bytes) |
| 357 | content_type = mimetypes.guess_type(local_path.name)[0] |
| 358 | return data, local_path.name, content_type |
| 359 | except Exception as e: |
| 360 | logger.error("DingTalk media read error ref={} err={}", media_ref, e) |
| 361 | return None, None, None |
| 362 | |
| 363 | async def _upload_media( |
| 364 | self, |
| 365 | token: str, |
| 366 | data: bytes, |
| 367 | media_type: str, |
| 368 | filename: str, |
| 369 | content_type: str | None, |
| 370 | ) -> str | None: |
| 371 | if not self._http: |
| 372 | return None |
| 373 | url = f"https://oapi.dingtalk.com/media/upload?access_token={token}&type={media_type}" |
| 374 | mime = content_type or mimetypes.guess_type(filename)[0] or "application/octet-stream" |
| 375 | files = {"media": (filename, data, mime)} |
| 376 | |
| 377 | try: |
| 378 | resp = await self._http.post(url, files=files) |
| 379 | text = resp.text |
| 380 | result = resp.json() if resp.headers.get("content-type", "").startswith("application/json") else {} |
| 381 | if resp.status_code >= 400: |
| 382 | logger.error("DingTalk media upload failed status={} type={} body={}", resp.status_code, media_type, text[:500]) |
| 383 | return None |
| 384 | errcode = result.get("errcode", 0) |
| 385 | if errcode != 0: |
| 386 | logger.error("DingTalk media upload api error type={} errcode={} body={}", media_type, errcode, text[:500]) |
| 387 | return None |
| 388 | sub = result.get("result") or {} |
| 389 | media_id = result.get("media_id") or result.get("mediaId") or sub.get("media_id") or sub.get("mediaId") |
| 390 | if not media_id: |
| 391 | logger.error("DingTalk media upload missing media_id body={}", text[:500]) |
| 392 | return None |
| 393 | return str(media_id) |
| 394 | except httpx.TransportError as e: |
| 395 | logger.error("DingTalk media upload network error type={} err={}", media_type, e) |
| 396 | raise |
| 397 | except Exception as e: |
| 398 | logger.error("DingTalk media upload error type={} err={}", media_type, e) |
| 399 | return None |
| 400 | |
| 401 | async def _send_batch_message( |
| 402 | self, |
| 403 | token: str, |
| 404 | chat_id: str, |
| 405 | msg_key: str, |
| 406 | msg_param: dict[str, Any], |
| 407 | ) -> bool: |
| 408 | if not self._http: |
| 409 | logger.warning("DingTalk HTTP client not initialized, cannot send") |
| 410 | return False |
| 411 | |
| 412 | headers = {"x-acs-dingtalk-access-token": token} |
| 413 | if chat_id.startswith("group:"): |
| 414 | # Group chat |
| 415 | url = "https://api.dingtalk.com/v1.0/robot/groupMessages/send" |
| 416 | payload = { |
| 417 | "robotCode": self.config.client_id, |
| 418 | "openConversationId": chat_id[6:], # Remove "group:" prefix, |
| 419 | "msgKey": msg_key, |
| 420 | "msgParam": json.dumps(msg_param, ensure_ascii=False), |
| 421 | } |
| 422 | else: |
| 423 | # Private chat |
| 424 | url = "https://api.dingtalk.com/v1.0/robot/oToMessages/batchSend" |
| 425 | payload = { |
| 426 | "robotCode": self.config.client_id, |
| 427 | "userIds": [chat_id], |
| 428 | "msgKey": msg_key, |
| 429 | "msgParam": json.dumps(msg_param, ensure_ascii=False), |
| 430 | } |
| 431 | |
| 432 | try: |
| 433 | resp = await self._http.post(url, json=payload, headers=headers) |
| 434 | body = resp.text |
| 435 | if resp.status_code != 200: |
| 436 | logger.error("DingTalk send failed msgKey={} status={} body={}", msg_key, resp.status_code, body[:500]) |
| 437 | return False |
| 438 | try: result = resp.json() |
| 439 | except Exception: result = {} |
| 440 | errcode = result.get("errcode") |
| 441 | if errcode not in (None, 0): |
| 442 | logger.error("DingTalk send api error msgKey={} errcode={} body={}", msg_key, errcode, body[:500]) |
| 443 | return False |
| 444 | logger.debug("DingTalk message sent to {} with msgKey={}", chat_id, msg_key) |
| 445 | return True |
| 446 | except httpx.TransportError as e: |
| 447 | logger.error("DingTalk network error sending message msgKey={} err={}", msg_key, e) |
| 448 | raise |
| 449 | except Exception as e: |
| 450 | logger.error("Error sending DingTalk message msgKey={} err={}", msg_key, e) |
| 451 | return False |
| 452 | |
| 453 | async def _send_markdown_text(self, token: str, chat_id: str, content: str) -> bool: |
| 454 | return await self._send_batch_message( |
| 455 | token, |
| 456 | chat_id, |
| 457 | "sampleMarkdown", |
| 458 | {"text": content, "title": "Nanobot Reply"}, |
| 459 | ) |
| 460 | |
| 461 | async def _send_media_ref(self, token: str, chat_id: str, media_ref: str) -> bool: |
| 462 | media_ref = (media_ref or "").strip() |
| 463 | if not media_ref: |
| 464 | return True |
| 465 | |
| 466 | upload_type = self._guess_upload_type(media_ref) |
| 467 | if upload_type == "image" and self._is_http_url(media_ref): |
| 468 | ok = await self._send_batch_message( |
| 469 | token, |
| 470 | chat_id, |
| 471 | "sampleImageMsg", |
| 472 | {"photoURL": media_ref}, |
| 473 | ) |
| 474 | if ok: |
| 475 | return True |
| 476 | logger.warning("DingTalk image url send failed, trying upload fallback: {}", media_ref) |
| 477 | |
| 478 | data, filename, content_type = await self._read_media_bytes(media_ref) |
| 479 | if not data: |
| 480 | logger.error("DingTalk media read failed: {}", media_ref) |
| 481 | return False |
| 482 | |
| 483 | filename = filename or self._guess_filename(media_ref, upload_type) |
| 484 | data, filename, content_type = self._normalize_upload_payload(filename, data, content_type) |
| 485 | file_type = Path(filename).suffix.lower().lstrip(".") |
| 486 | if not file_type: |
| 487 | guessed = mimetypes.guess_extension(content_type or "") |
| 488 | file_type = (guessed or ".bin").lstrip(".") |
| 489 | if file_type == "jpeg": |
| 490 | file_type = "jpg" |
| 491 | |
| 492 | media_id = await self._upload_media( |
| 493 | token=token, |
| 494 | data=data, |
| 495 | media_type=upload_type, |
| 496 | filename=filename, |
| 497 | content_type=content_type, |
| 498 | ) |
| 499 | if not media_id: |
| 500 | return False |
| 501 | |
| 502 | if upload_type == "image": |
| 503 | # Verified in production: sampleImageMsg accepts media_id in photoURL. |
| 504 | ok = await self._send_batch_message( |
| 505 | token, |
| 506 | chat_id, |
| 507 | "sampleImageMsg", |
| 508 | {"photoURL": media_id}, |
| 509 | ) |
| 510 | if ok: |
| 511 | return True |
| 512 | logger.warning("DingTalk image media_id send failed, falling back to file: {}", media_ref) |
| 513 | |
| 514 | return await self._send_batch_message( |
| 515 | token, |
| 516 | chat_id, |
| 517 | "sampleFile", |
| 518 | {"mediaId": media_id, "fileName": filename, "fileType": file_type}, |
| 519 | ) |
| 520 | |
| 521 | async def send(self, msg: OutboundMessage) -> None: |
| 522 | """Send a message through DingTalk.""" |
| 523 | token = await self._get_access_token() |
| 524 | if not token: |
| 525 | return |
| 526 | |
| 527 | if msg.content and msg.content.strip(): |
| 528 | await self._send_markdown_text(token, msg.chat_id, msg.content.strip()) |
| 529 | |
| 530 | for media_ref in msg.media or []: |
| 531 | ok = await self._send_media_ref(token, msg.chat_id, media_ref) |
| 532 | if ok: |
| 533 | continue |
| 534 | logger.error("DingTalk media send failed for {}", media_ref) |
| 535 | # Send visible fallback so failures are observable by the user. |
| 536 | filename = self._guess_filename(media_ref, self._guess_upload_type(media_ref)) |
| 537 | await self._send_markdown_text( |
| 538 | token, |
| 539 | msg.chat_id, |
| 540 | f"[Attachment send failed: {filename}]", |
| 541 | ) |
| 542 | |
| 543 | async def _on_message( |
| 544 | self, |
| 545 | content: str, |
| 546 | sender_id: str, |
| 547 | sender_name: str, |
| 548 | conversation_type: str | None = None, |
| 549 | conversation_id: str | None = None, |
| 550 | ) -> None: |
| 551 | """Handle incoming message (called by NanobotDingTalkHandler). |
| 552 | |
| 553 | Delegates to BaseChannel._handle_message() which enforces allow_from |
| 554 | permission checks before publishing to the bus. |
| 555 | """ |
| 556 | try: |
| 557 | logger.info("DingTalk inbound: {} from {}", content, sender_name) |
| 558 | is_group = conversation_type == "2" and conversation_id |
| 559 | chat_id = f"group:{conversation_id}" if is_group else sender_id |
| 560 | await self._handle_message( |
| 561 | sender_id=sender_id, |
| 562 | chat_id=chat_id, |
| 563 | content=str(content), |
| 564 | metadata={ |
| 565 | "sender_name": sender_name, |
| 566 | "platform": "dingtalk", |
| 567 | "conversation_type": conversation_type, |
| 568 | }, |
| 569 | ) |
| 570 | except Exception as e: |
| 571 | logger.error("Error publishing DingTalk message: {}", e) |
| 572 | |
| 573 | async def _download_dingtalk_file( |
| 574 | self, |
| 575 | download_code: str, |
| 576 | filename: str, |
| 577 | sender_id: str, |
| 578 | ) -> str | None: |
| 579 | """Download a DingTalk file to the media directory, return local path.""" |
| 580 | from nanobot.config.paths import get_media_dir |
| 581 | |
| 582 | try: |
| 583 | token = await self._get_access_token() |
| 584 | if not token or not self._http: |
| 585 | logger.error("DingTalk file download: no token or http client") |
| 586 | return None |
| 587 | |
| 588 | # Step 1: Exchange downloadCode for a temporary download URL |
| 589 | api_url = "https://api.dingtalk.com/v1.0/robot/messageFiles/download" |
| 590 | headers = {"x-acs-dingtalk-access-token": token, "Content-Type": "application/json"} |
| 591 | payload = {"downloadCode": download_code, "robotCode": self.config.client_id} |
| 592 | resp = await self._http.post(api_url, json=payload, headers=headers) |
| 593 | if resp.status_code != 200: |
| 594 | logger.error("DingTalk get download URL failed: status={}, body={}", resp.status_code, resp.text) |
| 595 | return None |
| 596 | |
| 597 | result = resp.json() |
| 598 | download_url = result.get("downloadUrl") |
| 599 | if not download_url: |
| 600 | logger.error("DingTalk download URL not found in response: {}", result) |
| 601 | return None |
| 602 | |
| 603 | # Step 2: Download the file content |
| 604 | file_resp = await self._http.get(download_url, follow_redirects=True) |
| 605 | if file_resp.status_code != 200: |
| 606 | logger.error("DingTalk file download failed: status={}", file_resp.status_code) |
| 607 | return None |
| 608 | |
| 609 | # Save to media directory (accessible under workspace) |
| 610 | download_dir = get_media_dir("dingtalk") / sender_id |
| 611 | download_dir.mkdir(parents=True, exist_ok=True) |
| 612 | file_path = download_dir / filename |
| 613 | await asyncio.to_thread(file_path.write_bytes, file_resp.content) |
| 614 | logger.info("DingTalk file saved: {}", file_path) |
| 615 | return str(file_path) |
| 616 | except Exception as e: |
| 617 | logger.error("DingTalk file download error: {}", e) |
| 618 | return None |
| 619 |