813 lines
33 KiB
Python
813 lines
33 KiB
Python
"""DingTalk/DingDing channel implementation using Stream Mode."""
|
|
|
|
import asyncio
|
|
import json
|
|
import mimetypes
|
|
import os
|
|
import time
|
|
import zipfile
|
|
from contextlib import suppress
|
|
from inspect import isawaitable
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.parse import unquote, urljoin, urlparse
|
|
|
|
import httpx
|
|
from pydantic import Field
|
|
|
|
from nanobot.bus.events import OutboundMessage
|
|
from nanobot.bus.queue import MessageBus
|
|
from nanobot.channels.base import BaseChannel
|
|
from nanobot.config.schema import Base
|
|
from nanobot.security.network import validate_resolved_url, validate_url_target
|
|
|
|
DINGTALK_MAX_REMOTE_MEDIA_BYTES = 20 * 1024 * 1024
|
|
DINGTALK_MAX_REMOTE_MEDIA_REDIRECTS = 3
|
|
|
|
try:
|
|
from dingtalk_stream import (
|
|
AckMessage,
|
|
CallbackHandler,
|
|
CallbackMessage,
|
|
Credential,
|
|
DingTalkStreamClient,
|
|
)
|
|
from dingtalk_stream.chatbot import ChatbotMessage
|
|
|
|
DINGTALK_AVAILABLE = True
|
|
except ImportError:
|
|
DINGTALK_AVAILABLE = False
|
|
# Fallback so class definitions don't crash at module level
|
|
CallbackHandler = object # type: ignore[assignment,misc]
|
|
CallbackMessage = None # type: ignore[assignment,misc]
|
|
AckMessage = None # type: ignore[assignment,misc]
|
|
ChatbotMessage = None # type: ignore[assignment,misc]
|
|
|
|
|
|
class NanobotDingTalkHandler(CallbackHandler):
|
|
"""
|
|
Standard DingTalk Stream SDK Callback Handler.
|
|
Parses incoming messages and forwards them to the Nanobot channel.
|
|
"""
|
|
|
|
def __init__(self, channel: "DingTalkChannel"):
|
|
super().__init__()
|
|
self.channel = channel
|
|
|
|
async def process(self, message: CallbackMessage):
|
|
"""Process incoming stream message."""
|
|
try:
|
|
# Parse using SDK's ChatbotMessage for robust handling
|
|
chatbot_msg = ChatbotMessage.from_dict(message.data)
|
|
|
|
# Extract text content; fall back to raw dict if SDK object is empty
|
|
content = ""
|
|
if chatbot_msg.text:
|
|
content = chatbot_msg.text.content.strip()
|
|
elif chatbot_msg.extensions.get("content", {}).get("recognition"):
|
|
content = chatbot_msg.extensions["content"]["recognition"].strip()
|
|
if not content:
|
|
content = message.data.get("text", {}).get("content", "").strip()
|
|
|
|
# Handle file/image messages
|
|
file_paths = []
|
|
if chatbot_msg.message_type == "picture" and chatbot_msg.image_content:
|
|
download_code = chatbot_msg.image_content.download_code
|
|
if download_code:
|
|
sender_uid = chatbot_msg.sender_staff_id or chatbot_msg.sender_id or "unknown"
|
|
fp = await self.channel._download_dingtalk_file(download_code, "image.jpg", sender_uid)
|
|
if fp:
|
|
file_paths.append(fp)
|
|
content = content or "[Image]"
|
|
|
|
elif chatbot_msg.message_type == "file":
|
|
download_code = message.data.get("content", {}).get("downloadCode") or message.data.get("downloadCode")
|
|
fname = message.data.get("content", {}).get("fileName") or message.data.get("fileName") or "file"
|
|
if download_code:
|
|
sender_uid = chatbot_msg.sender_staff_id or chatbot_msg.sender_id or "unknown"
|
|
fp = await self.channel._download_dingtalk_file(download_code, fname, sender_uid)
|
|
if fp:
|
|
file_paths.append(fp)
|
|
content = content or "[File]"
|
|
|
|
elif chatbot_msg.message_type == "richText" and chatbot_msg.rich_text_content:
|
|
rich_list = chatbot_msg.rich_text_content.rich_text_list or []
|
|
for item in rich_list:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
# A rich-text item may carry text and/or a downloadCode; the
|
|
# DingTalk SDK treats them independently, so handle both.
|
|
t = item.get("text", "").strip()
|
|
if t:
|
|
fmt = item.get("type", "")
|
|
if fmt == "bold":
|
|
formatted = f"**{t}**"
|
|
elif fmt == "italic":
|
|
formatted = f"*{t}*"
|
|
elif fmt == "inlineCode":
|
|
formatted = f"`{t}`"
|
|
elif fmt == "pre":
|
|
formatted = f"```\n{t}\n```"
|
|
else:
|
|
formatted = t
|
|
content = (content + " " + formatted).strip() if content else formatted
|
|
if item.get("downloadCode"):
|
|
dc = item["downloadCode"]
|
|
fname = item.get("fileName") or "file"
|
|
sender_uid = chatbot_msg.sender_staff_id or chatbot_msg.sender_id or "unknown"
|
|
fp = await self.channel._download_dingtalk_file(dc, fname, sender_uid)
|
|
if fp:
|
|
file_paths.append(fp)
|
|
content = content or "[File]"
|
|
|
|
if file_paths:
|
|
file_list = "\n".join("- " + p for p in file_paths)
|
|
content = content + "\n\nReceived files:\n" + file_list
|
|
|
|
if not content:
|
|
self.channel.logger.warning(
|
|
"Received empty or unsupported message type: {}",
|
|
chatbot_msg.message_type,
|
|
)
|
|
return AckMessage.STATUS_OK, "OK"
|
|
|
|
sender_id = chatbot_msg.sender_staff_id or chatbot_msg.sender_id
|
|
sender_name = chatbot_msg.sender_nick or "Unknown"
|
|
|
|
conversation_type = message.data.get("conversationType")
|
|
conversation_id = (
|
|
message.data.get("conversationId")
|
|
or message.data.get("openConversationId")
|
|
)
|
|
|
|
self.channel.logger.info("Received message from {} ({}): {}", sender_name, sender_id, content)
|
|
|
|
# Forward to Nanobot via _on_message (non-blocking).
|
|
# Store reference to prevent GC before task completes.
|
|
task = asyncio.create_task(
|
|
self.channel._on_message(
|
|
content,
|
|
sender_id,
|
|
sender_name,
|
|
conversation_type,
|
|
conversation_id,
|
|
)
|
|
)
|
|
self.channel._background_tasks.add(task)
|
|
task.add_done_callback(self.channel._background_tasks.discard)
|
|
|
|
return AckMessage.STATUS_OK, "OK"
|
|
|
|
except Exception:
|
|
self.channel.logger.exception("Error processing message")
|
|
# Return OK to avoid retry loop from DingTalk server
|
|
return AckMessage.STATUS_OK, "Error"
|
|
|
|
|
|
class DingTalkConfig(Base):
|
|
"""DingTalk channel configuration using Stream mode."""
|
|
|
|
enabled: bool = False
|
|
client_id: str = ""
|
|
client_secret: str = ""
|
|
allow_from: list[str] = Field(default_factory=list)
|
|
allow_remote_media_redirects: bool = False
|
|
remote_media_redirect_allowed_hosts: list[str] = Field(default_factory=list)
|
|
group_user_isolation: bool = False # If True, each user in group chat gets their own session
|
|
|
|
|
|
class DingTalkChannel(BaseChannel):
|
|
"""
|
|
DingTalk channel using Stream Mode.
|
|
|
|
Uses WebSocket to receive events via `dingtalk-stream` SDK.
|
|
Uses direct HTTP API to send messages (SDK is mainly for receiving).
|
|
|
|
Supports both private (1:1) and group chats.
|
|
Group chat_id is stored with a "group:" prefix to route replies back.
|
|
"""
|
|
|
|
name = "dingtalk"
|
|
display_name = "DingTalk"
|
|
_IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".gif", ".bmp", ".webp"}
|
|
_AUDIO_EXTS = {".amr", ".mp3", ".wav", ".ogg", ".m4a", ".aac"}
|
|
_VIDEO_EXTS = {".mp4", ".mov", ".avi", ".mkv", ".webm"}
|
|
_ZIP_BEFORE_UPLOAD_EXTS = {".htm", ".html"}
|
|
|
|
@classmethod
|
|
def default_config(cls) -> dict[str, Any]:
|
|
return DingTalkConfig().model_dump(by_alias=True)
|
|
|
|
def __init__(self, config: Any, bus: MessageBus):
|
|
if isinstance(config, dict):
|
|
config = DingTalkConfig.model_validate(config)
|
|
super().__init__(config, bus)
|
|
self.config: DingTalkConfig = config
|
|
self._client: Any = None
|
|
self._http: httpx.AsyncClient | None = None
|
|
self._start_task: asyncio.Task | None = None
|
|
|
|
# Access Token management for sending messages
|
|
self._access_token: str | None = None
|
|
self._token_expiry: float = 0
|
|
|
|
# Hold references to background tasks to prevent GC
|
|
self._background_tasks: set[asyncio.Task] = set()
|
|
|
|
async def start(self) -> None:
|
|
"""Start the DingTalk bot with Stream Mode."""
|
|
current_task = asyncio.current_task()
|
|
self._start_task = current_task
|
|
try:
|
|
if not DINGTALK_AVAILABLE:
|
|
self.logger.error(
|
|
"Stream SDK not installed. Run: nanobot plugins enable dingtalk"
|
|
)
|
|
return
|
|
|
|
if not self.config.client_id or not self.config.client_secret:
|
|
self.logger.error("client_id and client_secret not configured")
|
|
return
|
|
|
|
self._running = True
|
|
self._http = httpx.AsyncClient(
|
|
timeout=httpx.Timeout(10.0, connect=10.0, read=30.0, write=30.0, pool=10.0)
|
|
)
|
|
|
|
self.logger.info(
|
|
"Initializing Stream Client with Client ID: {}...",
|
|
self.config.client_id,
|
|
)
|
|
credential = Credential(self.config.client_id, self.config.client_secret)
|
|
self._client = DingTalkStreamClient(credential)
|
|
|
|
# Register standard handler
|
|
handler = NanobotDingTalkHandler(self)
|
|
self._client.register_callback_handler(ChatbotMessage.TOPIC, handler)
|
|
|
|
self.logger.info("bot started with Stream Mode")
|
|
|
|
# Reconnect loop: restart stream if SDK exits or crashes
|
|
while self._running:
|
|
try:
|
|
await self._client.start()
|
|
except Exception as e:
|
|
self.logger.warning("stream error: {}", e)
|
|
if self._running:
|
|
self.logger.info("Reconnecting stream in 5 seconds...")
|
|
await asyncio.sleep(5)
|
|
|
|
except Exception:
|
|
self.logger.exception("Failed to start channel")
|
|
finally:
|
|
self._running = False
|
|
if self._start_task is current_task:
|
|
self._start_task = None
|
|
|
|
async def stop(self) -> None:
|
|
"""Stop the DingTalk bot."""
|
|
self._running = False
|
|
await self._close_stream_client()
|
|
start_task = self._start_task
|
|
if start_task and start_task is not asyncio.current_task() and not start_task.done():
|
|
start_task.cancel()
|
|
await asyncio.sleep(0)
|
|
if not start_task.done():
|
|
start_task.cancel()
|
|
with suppress(asyncio.CancelledError):
|
|
await start_task
|
|
self._client = None
|
|
|
|
# Close the shared HTTP client
|
|
if self._http:
|
|
await self._http.aclose()
|
|
self._http = None
|
|
# Cancel outstanding background tasks
|
|
for task in self._background_tasks:
|
|
task.cancel()
|
|
self._background_tasks.clear()
|
|
|
|
async def _close_stream_client(self) -> None:
|
|
client = self._client
|
|
if client is None:
|
|
return
|
|
close = getattr(client, "close", None)
|
|
if close is None:
|
|
websocket = getattr(client, "websocket", None)
|
|
close = getattr(websocket, "close", None)
|
|
if close is None:
|
|
return
|
|
try:
|
|
result = close()
|
|
if isawaitable(result):
|
|
await result
|
|
except Exception:
|
|
self.logger.debug("DingTalk stream client close failed", exc_info=True)
|
|
|
|
async def _get_access_token(self) -> str | None:
|
|
"""Get or refresh Access Token."""
|
|
if self._access_token and time.time() < self._token_expiry:
|
|
return self._access_token
|
|
|
|
url = "https://api.dingtalk.com/v1.0/oauth2/accessToken"
|
|
data = {
|
|
"appKey": self.config.client_id,
|
|
"appSecret": self.config.client_secret,
|
|
}
|
|
|
|
if not self._http:
|
|
self.logger.warning("HTTP client not initialized, cannot refresh token")
|
|
return None
|
|
|
|
try:
|
|
resp = await self._http.post(url, json=data)
|
|
resp.raise_for_status()
|
|
res_data = resp.json()
|
|
self._access_token = res_data.get("accessToken")
|
|
# Expire 60s early to be safe
|
|
self._token_expiry = time.time() + int(res_data.get("expireIn", 7200)) - 60
|
|
return self._access_token
|
|
except Exception:
|
|
self.logger.exception("Failed to get access token")
|
|
return None
|
|
|
|
@staticmethod
|
|
def _is_http_url(value: str) -> bool:
|
|
return urlparse(value).scheme in ("http", "https")
|
|
|
|
def _guess_upload_type(self, media_ref: str) -> str:
|
|
ext = Path(urlparse(media_ref).path).suffix.lower()
|
|
if ext in self._IMAGE_EXTS:
|
|
return "image"
|
|
if ext in self._AUDIO_EXTS:
|
|
return "voice"
|
|
if ext in self._VIDEO_EXTS:
|
|
return "video"
|
|
return "file"
|
|
|
|
def _guess_filename(self, media_ref: str, upload_type: str) -> str:
|
|
name = os.path.basename(urlparse(media_ref).path)
|
|
return name or {"image": "image.jpg", "voice": "audio.amr", "video": "video.mp4"}.get(upload_type, "file.bin")
|
|
|
|
@staticmethod
|
|
def _zip_bytes(filename: str, data: bytes) -> tuple[bytes, str, str]:
|
|
stem = Path(filename).stem or "attachment"
|
|
safe_name = filename or "attachment.bin"
|
|
zip_name = f"{stem}.zip"
|
|
buffer = BytesIO()
|
|
with zipfile.ZipFile(buffer, mode="w", compression=zipfile.ZIP_DEFLATED) as archive:
|
|
archive.writestr(safe_name, data)
|
|
return buffer.getvalue(), zip_name, "application/zip"
|
|
|
|
def _normalize_upload_payload(
|
|
self,
|
|
filename: str,
|
|
data: bytes,
|
|
content_type: str | None,
|
|
) -> tuple[bytes, str, str | None]:
|
|
ext = Path(filename).suffix.lower()
|
|
if ext in self._ZIP_BEFORE_UPLOAD_EXTS or content_type == "text/html":
|
|
self.logger.info(
|
|
"does not accept raw HTML attachments, zipping {} before upload",
|
|
filename,
|
|
)
|
|
return self._zip_bytes(filename, data)
|
|
return data, filename, content_type
|
|
|
|
def _validate_remote_media_url(self, media_ref: str) -> bool:
|
|
ok, err = validate_url_target(media_ref)
|
|
if not ok:
|
|
self.logger.warning("remote media URL blocked ref={} reason={}", media_ref, err)
|
|
return False
|
|
return True
|
|
|
|
def _redirect_host_allowed(self, current_url: str, next_url: str) -> bool:
|
|
current_host = (urlparse(current_url).hostname or "").lower()
|
|
next_host = (urlparse(next_url).hostname or "").lower()
|
|
if not next_host:
|
|
return False
|
|
if next_host == current_host:
|
|
return True
|
|
allowed_hosts = {host.lower() for host in self.config.remote_media_redirect_allowed_hosts}
|
|
return next_host in allowed_hosts
|
|
|
|
def _next_remote_media_url(self, current_url: str, location: str | None) -> str | None:
|
|
if not self.config.allow_remote_media_redirects:
|
|
self.logger.warning("media download redirect refused ref={}", current_url)
|
|
return None
|
|
if not location:
|
|
self.logger.warning("media download redirect without Location ref={}", current_url)
|
|
return None
|
|
next_url = urljoin(current_url, location)
|
|
if not self._redirect_host_allowed(current_url, next_url):
|
|
self.logger.warning(
|
|
"media download cross-host redirect refused ref={} next={}",
|
|
current_url,
|
|
next_url,
|
|
)
|
|
return None
|
|
if not self._validate_remote_media_url(next_url):
|
|
return None
|
|
return next_url
|
|
|
|
async def _fetch_remote_media_bytes(
|
|
self,
|
|
media_ref: str,
|
|
) -> tuple[bytes | None, str | None]:
|
|
"""Fetch a remote media URL with SSRF, redirect, and size checks."""
|
|
if not self._http:
|
|
return None, None
|
|
|
|
if not self._validate_remote_media_url(media_ref):
|
|
return None, None
|
|
|
|
try:
|
|
# Prefer streaming with a running byte cap so large responses are not
|
|
# materialized before the limit is enforced. Test fakes may only
|
|
# implement get(), so keep a small compatibility fallback below.
|
|
stream = getattr(self._http, "stream", None)
|
|
if stream is not None:
|
|
current_url = media_ref
|
|
for _ in range(DINGTALK_MAX_REMOTE_MEDIA_REDIRECTS + 1):
|
|
async with stream("GET", current_url, follow_redirects=False) as resp:
|
|
final_ok, final_err = validate_resolved_url(str(resp.url))
|
|
if not final_ok:
|
|
self.logger.warning(
|
|
"remote media redirect blocked ref={} final={} reason={}",
|
|
media_ref,
|
|
resp.url,
|
|
final_err,
|
|
)
|
|
return None, None
|
|
if 300 <= resp.status_code < 400:
|
|
next_url = self._next_remote_media_url(
|
|
str(resp.url), resp.headers.get("location")
|
|
)
|
|
if not next_url:
|
|
return None, None
|
|
current_url = next_url
|
|
continue
|
|
if resp.status_code >= 400:
|
|
self.logger.warning(
|
|
"media download failed status={} ref={}",
|
|
resp.status_code,
|
|
current_url,
|
|
)
|
|
return None, None
|
|
chunks: list[bytes] = []
|
|
total = 0
|
|
async for chunk in resp.aiter_bytes():
|
|
total += len(chunk)
|
|
if total > DINGTALK_MAX_REMOTE_MEDIA_BYTES:
|
|
self.logger.warning(
|
|
"media download too large ref={} bytes>{}",
|
|
current_url,
|
|
DINGTALK_MAX_REMOTE_MEDIA_BYTES,
|
|
)
|
|
return None, None
|
|
chunks.append(chunk)
|
|
return b"".join(chunks), (resp.headers.get("content-type") or "")
|
|
self.logger.warning("media download exceeded redirect limit ref={}", media_ref)
|
|
return None, None
|
|
|
|
current_url = media_ref
|
|
for _ in range(DINGTALK_MAX_REMOTE_MEDIA_REDIRECTS + 1):
|
|
resp = await self._http.get(current_url, follow_redirects=False)
|
|
final_ok, final_err = validate_resolved_url(str(getattr(resp, "url", current_url)))
|
|
if not final_ok:
|
|
self.logger.warning(
|
|
"remote media redirect blocked ref={} final={} reason={}",
|
|
media_ref,
|
|
getattr(resp, "url", current_url),
|
|
final_err,
|
|
)
|
|
return None, None
|
|
if 300 <= resp.status_code < 400:
|
|
next_url = self._next_remote_media_url(
|
|
str(getattr(resp, "url", current_url)), resp.headers.get("location")
|
|
)
|
|
if not next_url:
|
|
return None, None
|
|
current_url = next_url
|
|
continue
|
|
if resp.status_code >= 400:
|
|
self.logger.warning(
|
|
"media download failed status={} ref={}",
|
|
resp.status_code,
|
|
current_url,
|
|
)
|
|
return None, None
|
|
if len(resp.content) > DINGTALK_MAX_REMOTE_MEDIA_BYTES:
|
|
self.logger.warning(
|
|
"media download too large ref={} bytes>{}",
|
|
current_url,
|
|
DINGTALK_MAX_REMOTE_MEDIA_BYTES,
|
|
)
|
|
return None, None
|
|
return resp.content, (resp.headers.get("content-type") or "")
|
|
self.logger.warning("media download exceeded redirect limit ref={}", media_ref)
|
|
return None, None
|
|
except httpx.TransportError:
|
|
self.logger.exception("media download network error ref={}", media_ref)
|
|
raise
|
|
except Exception:
|
|
self.logger.exception("media download error ref={}", media_ref)
|
|
return None, None
|
|
|
|
async def _read_media_bytes(
|
|
self,
|
|
media_ref: str,
|
|
) -> tuple[bytes | None, str | None, str | None]:
|
|
if not media_ref:
|
|
return None, None, None
|
|
|
|
if self._is_http_url(media_ref):
|
|
data, raw_content_type = await self._fetch_remote_media_bytes(media_ref)
|
|
if data is None:
|
|
return None, None, None
|
|
content_type = (raw_content_type or "").split(";")[0].strip()
|
|
filename = self._guess_filename(media_ref, self._guess_upload_type(media_ref))
|
|
return data, filename, content_type or None
|
|
|
|
try:
|
|
if media_ref.startswith("file://"):
|
|
parsed = urlparse(media_ref)
|
|
local_path = Path(unquote(parsed.path))
|
|
else:
|
|
local_path = Path(os.path.expanduser(media_ref))
|
|
if not local_path.is_file():
|
|
self.logger.warning("media file not found: {}", local_path)
|
|
return None, None, None
|
|
data = await asyncio.to_thread(local_path.read_bytes)
|
|
content_type = mimetypes.guess_type(local_path.name)[0]
|
|
return data, local_path.name, content_type
|
|
except Exception:
|
|
self.logger.exception("media read error ref={}", media_ref)
|
|
return None, None, None
|
|
|
|
async def _upload_media(
|
|
self,
|
|
token: str,
|
|
data: bytes,
|
|
media_type: str,
|
|
filename: str,
|
|
content_type: str | None,
|
|
) -> str | None:
|
|
if not self._http:
|
|
return None
|
|
url = f"https://oapi.dingtalk.com/media/upload?access_token={token}&type={media_type}"
|
|
mime = content_type or mimetypes.guess_type(filename)[0] or "application/octet-stream"
|
|
files = {"media": (filename, data, mime)}
|
|
|
|
try:
|
|
resp = await self._http.post(url, files=files)
|
|
text = resp.text
|
|
result = resp.json() if resp.headers.get("content-type", "").startswith("application/json") else {}
|
|
if resp.status_code >= 400:
|
|
self.logger.error("media upload failed status={} type={} body={}", resp.status_code, media_type, text[:500])
|
|
return None
|
|
errcode = result.get("errcode", 0)
|
|
if errcode != 0:
|
|
self.logger.error("media upload api error type={} errcode={} body={}", media_type, errcode, text[:500])
|
|
return None
|
|
sub = result.get("result") or {}
|
|
media_id = result.get("media_id") or result.get("mediaId") or sub.get("media_id") or sub.get("mediaId")
|
|
if not media_id:
|
|
self.logger.error("media upload missing media_id body={}", text[:500])
|
|
return None
|
|
return str(media_id)
|
|
except httpx.TransportError:
|
|
self.logger.exception("media upload network error type={}", media_type)
|
|
raise
|
|
except Exception:
|
|
self.logger.exception("media upload error type={}", media_type)
|
|
return None
|
|
|
|
async def _send_batch_message(
|
|
self,
|
|
token: str,
|
|
chat_id: str,
|
|
msg_key: str,
|
|
msg_param: dict[str, Any],
|
|
) -> bool:
|
|
if not self._http:
|
|
self.logger.warning("HTTP client not initialized, cannot send")
|
|
return False
|
|
|
|
headers = {"x-acs-dingtalk-access-token": token}
|
|
if chat_id.startswith("group:"):
|
|
# Group chat
|
|
url = "https://api.dingtalk.com/v1.0/robot/groupMessages/send"
|
|
payload = {
|
|
"robotCode": self.config.client_id,
|
|
"openConversationId": chat_id[6:], # Remove "group:" prefix,
|
|
"msgKey": msg_key,
|
|
"msgParam": json.dumps(msg_param, ensure_ascii=False),
|
|
}
|
|
else:
|
|
# Private chat
|
|
url = "https://api.dingtalk.com/v1.0/robot/oToMessages/batchSend"
|
|
payload = {
|
|
"robotCode": self.config.client_id,
|
|
"userIds": [chat_id],
|
|
"msgKey": msg_key,
|
|
"msgParam": json.dumps(msg_param, ensure_ascii=False),
|
|
}
|
|
|
|
try:
|
|
resp = await self._http.post(url, json=payload, headers=headers)
|
|
body = resp.text
|
|
if resp.status_code != 200:
|
|
self.logger.error("send failed msgKey={} status={} body={}", msg_key, resp.status_code, body[:500])
|
|
return False
|
|
try:
|
|
result = resp.json()
|
|
except Exception:
|
|
result = {}
|
|
errcode = result.get("errcode")
|
|
if errcode not in (None, 0):
|
|
self.logger.error("send api error msgKey={} errcode={} body={}", msg_key, errcode, body[:500])
|
|
return False
|
|
self.logger.debug("message sent to {} with msgKey={}", chat_id, msg_key)
|
|
return True
|
|
except httpx.TransportError:
|
|
self.logger.exception("network error sending message msgKey={}", msg_key)
|
|
raise
|
|
except Exception:
|
|
self.logger.exception("Error sending message msgKey={}", msg_key)
|
|
return False
|
|
|
|
async def _send_markdown_text(self, token: str, chat_id: str, content: str) -> bool:
|
|
return await self._send_batch_message(
|
|
token,
|
|
chat_id,
|
|
"sampleMarkdown",
|
|
{"text": content, "title": "Nanobot Reply"},
|
|
)
|
|
|
|
async def _send_media_ref(self, token: str, chat_id: str, media_ref: str) -> bool:
|
|
media_ref = (media_ref or "").strip()
|
|
if not media_ref:
|
|
return True
|
|
|
|
upload_type = self._guess_upload_type(media_ref)
|
|
if upload_type == "image" and self._is_http_url(media_ref):
|
|
ok = await self._send_batch_message(
|
|
token,
|
|
chat_id,
|
|
"sampleImageMsg",
|
|
{"photoURL": media_ref},
|
|
)
|
|
if ok:
|
|
return True
|
|
self.logger.warning("image url send failed, trying upload fallback: {}", media_ref)
|
|
|
|
data, filename, content_type = await self._read_media_bytes(media_ref)
|
|
if not data:
|
|
self.logger.error("media read failed: {}", media_ref)
|
|
return False
|
|
|
|
filename = filename or self._guess_filename(media_ref, upload_type)
|
|
data, filename, content_type = self._normalize_upload_payload(filename, data, content_type)
|
|
file_type = Path(filename).suffix.lower().lstrip(".")
|
|
if not file_type:
|
|
guessed = mimetypes.guess_extension(content_type or "")
|
|
file_type = (guessed or ".bin").lstrip(".")
|
|
if file_type == "jpeg":
|
|
file_type = "jpg"
|
|
|
|
media_id = await self._upload_media(
|
|
token=token,
|
|
data=data,
|
|
media_type=upload_type,
|
|
filename=filename,
|
|
content_type=content_type,
|
|
)
|
|
if not media_id:
|
|
return False
|
|
|
|
if upload_type == "image":
|
|
# Verified in production: sampleImageMsg accepts media_id in photoURL.
|
|
ok = await self._send_batch_message(
|
|
token,
|
|
chat_id,
|
|
"sampleImageMsg",
|
|
{"photoURL": media_id},
|
|
)
|
|
if ok:
|
|
return True
|
|
self.logger.warning("image media_id send failed, falling back to file: {}", media_ref)
|
|
|
|
return await self._send_batch_message(
|
|
token,
|
|
chat_id,
|
|
"sampleFile",
|
|
{"mediaId": media_id, "fileName": filename, "fileType": file_type},
|
|
)
|
|
|
|
async def send(self, msg: OutboundMessage) -> None:
|
|
"""Send a message through DingTalk."""
|
|
token = await self._get_access_token()
|
|
if not token:
|
|
raise RuntimeError("DingTalk access token unavailable")
|
|
|
|
if msg.content and msg.content.strip():
|
|
if not await self._send_markdown_text(token, msg.chat_id, msg.content.strip()):
|
|
raise RuntimeError("DingTalk text message was not delivered")
|
|
|
|
for media_ref in msg.media or []:
|
|
ok = await self._send_media_ref(token, msg.chat_id, media_ref)
|
|
if ok:
|
|
continue
|
|
self.logger.error("media send failed for {}", media_ref)
|
|
# Send visible fallback so failures are observable by the user.
|
|
filename = self._guess_filename(media_ref, self._guess_upload_type(media_ref))
|
|
if not await self._send_markdown_text(
|
|
token,
|
|
msg.chat_id,
|
|
f"[Attachment send failed: {filename}]",
|
|
):
|
|
raise RuntimeError("DingTalk attachment fallback was not delivered")
|
|
|
|
async def _on_message(
|
|
self,
|
|
content: str,
|
|
sender_id: str,
|
|
sender_name: str,
|
|
conversation_type: str | None = None,
|
|
conversation_id: str | None = None,
|
|
) -> None:
|
|
"""Handle incoming message (called by NanobotDingTalkHandler).
|
|
|
|
Delegates to BaseChannel._handle_message() which enforces allow_from
|
|
permission checks before publishing to the bus.
|
|
"""
|
|
try:
|
|
self.logger.info("inbound: {} from {}", content, sender_name)
|
|
is_group = conversation_type == "2" and conversation_id
|
|
chat_id = f"group:{conversation_id}" if is_group else sender_id
|
|
session_key = None
|
|
if is_group and self.config.group_user_isolation:
|
|
session_key = f"{self.name}:group:{conversation_id}:{sender_id}"
|
|
await self._handle_message(
|
|
sender_id=sender_id,
|
|
chat_id=chat_id,
|
|
content=str(content),
|
|
metadata={
|
|
"sender_name": sender_name,
|
|
"platform": "dingtalk",
|
|
"conversation_type": conversation_type,
|
|
},
|
|
session_key=session_key,
|
|
)
|
|
except Exception:
|
|
self.logger.exception("Error publishing message")
|
|
|
|
async def _download_dingtalk_file(
|
|
self,
|
|
download_code: str,
|
|
filename: str,
|
|
sender_id: str,
|
|
) -> str | None:
|
|
"""Download a DingTalk file to the media directory, return local path."""
|
|
from nanobot.config.paths import get_media_dir
|
|
|
|
try:
|
|
token = await self._get_access_token()
|
|
if not token or not self._http:
|
|
self.logger.error("file download: no token or http client")
|
|
return None
|
|
|
|
# Step 1: Exchange downloadCode for a temporary download URL
|
|
api_url = "https://api.dingtalk.com/v1.0/robot/messageFiles/download"
|
|
headers = {"x-acs-dingtalk-access-token": token, "Content-Type": "application/json"}
|
|
payload = {"downloadCode": download_code, "robotCode": self.config.client_id}
|
|
resp = await self._http.post(api_url, json=payload, headers=headers)
|
|
if resp.status_code != 200:
|
|
self.logger.error("get download URL failed: status={}, body={}", resp.status_code, resp.text)
|
|
return None
|
|
|
|
result = resp.json()
|
|
download_url = result.get("downloadUrl")
|
|
if not download_url:
|
|
self.logger.error("download URL not found in response: {}", result)
|
|
return None
|
|
|
|
# Step 2: Download the file content
|
|
file_resp = await self._http.get(download_url, follow_redirects=True)
|
|
if file_resp.status_code != 200:
|
|
self.logger.error("file download failed: status={}", file_resp.status_code)
|
|
return None
|
|
|
|
# Save to media directory (accessible under workspace)
|
|
download_dir = get_media_dir("dingtalk") / sender_id
|
|
download_dir.mkdir(parents=True, exist_ok=True)
|
|
file_path = download_dir / filename
|
|
await asyncio.to_thread(file_path.write_bytes, file_resp.content)
|
|
self.logger.info("file saved: {}", file_path)
|
|
return str(file_path)
|
|
except Exception:
|
|
self.logger.exception("file download error")
|
|
return None
|