diff --git a/backend/package/yuxi/channel/extensions/dingtalk/__init__.py b/backend/package/yuxi/channel/extensions/dingtalk/__init__.py new file mode 100644 index 00000000..43fe2793 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/dingtalk/__init__.py @@ -0,0 +1,392 @@ +from __future__ import annotations + +import logging + +from yuxi.channel.capabilities import ChannelCapabilities +from yuxi.channel.extensions.base import BaseChannelPlugin +from yuxi.channel.extensions.dingtalk.config import DingTalkConfig +from yuxi.channel.extensions.dingtalk.dedupe import DingTalkDeduplicator +from yuxi.channel.extensions.dingtalk.doctor import DingTalkDoctor +from yuxi.channel.extensions.dingtalk.file_cache import FileCache +from yuxi.channel.extensions.dingtalk.gateway import DingTalkGateway, DingTalkTokenManager +from yuxi.channel.extensions.dingtalk.monitor import DingTalkMonitor +from yuxi.channel.extensions.dingtalk.outbound import DingTalkOutbound +from yuxi.channel.extensions.dingtalk.pairing import DingTalkPairing +from yuxi.channel.extensions.dingtalk.security import DingTalkSecurity +from yuxi.channel.extensions.dingtalk.streaming import DingTalkStreamingAdapter +from yuxi.channel.plugins.registry import ChannelPluginRegistry +from yuxi.channel.sdk.actions import MessageAction, MessageActionRegistry + +logger = logging.getLogger(__name__) + + +class DingTalkPlugin(BaseChannelPlugin): + id = "dingtalk" + name = "钉钉" + order = 20 + label = "DingTalk" + aliases = ["ding", "dd"] + + def __init__(self): + self._config = DingTalkConfig() + self._gateway: DingTalkGateway | None = None + self._outbound: DingTalkOutbound | None = None + self._monitor = DingTalkMonitor() + self._security: DingTalkSecurity | None = None + self._pairing = DingTalkPairing() + self._deduplicator: DingTalkDeduplicator | None = None + self._file_cache = FileCache() + self._token_manager: DingTalkTokenManager | None = None + self._http = None + self._card_manager = None + self._doctor = DingTalkDoctor() + self._streaming = DingTalkStreamingAdapter() + self._actions = MessageActionRegistry() + self._actions.register(MessageAction.SEND, self._handle_send, description="发送文本消息到钉钉") + self._actions.register( + MessageAction.SEND_MEDIA, self._handle_send_media, description="发送图片/文件/音频/视频到钉钉" + ) + self._actions.register(MessageAction.REPLY, self._handle_reply, description="回复某条消息") + self._actions.register(MessageAction.ADD_PARTICIPANT, self._handle_add_member, description="添加成员到群聊") + self._actions.register( + MessageAction.REMOVE_PARTICIPANT, self._handle_remove_member, description="从群聊移除成员" + ) + + @property + def capabilities(self) -> ChannelCapabilities: + return ChannelCapabilities( + chat_types=["direct", "group"], + message_types=["text", "image", "file", "audio", "video"], + reactions=False, + reply=True, + media=True, + native_commands=True, + streaming=True, + streaming_mode="block", + block_streaming=True, + group_management=True, + adaptive_cards=True, + ) + + @property + def actions(self) -> MessageActionRegistry: + return self._actions + + def list_account_ids(self, config: dict) -> list[str]: + return self._config.list_account_ids(config) + + async def resolve_account(self, account_id: str) -> dict: + account = await self._config.resolve_account(account_id) + return { + "account_id": account.account_id, + "app_key": account.app_key, + "app_secret": account.app_secret, + "robot_code": account.robot_code, + "mode": account.mode, + "streaming": account.streaming, + "dingtalk_card_enabled": account.dingtalk_card_enabled, + "card_template_id": account.card_template_id, + "dm_policy": account.dm_policy, + "group_policy": account.group_policy, + "group_shared_session": account.group_shared_session, + "expires_in_seconds": account.expires_in_seconds, + "hot_reload": account.hot_reload, + "chat_time_module": account.chat_time_module, + "chat_start_time": account.chat_start_time, + "chat_stop_time": account.chat_stop_time, + "agent": account.agent, + "agent_workspace": account.agent_workspace, + "render_mode": account.render_mode, + "subscribe_topics": account.subscribe_topics, + "enabled": account.enabled, + } + + def is_configured(self, account: dict) -> bool: + return self._config.is_configured(account) + + def is_enabled(self, account: dict) -> bool: + return self._config.is_enabled(account) + + def disabled_reason(self, account: dict) -> str: + return self._config.disabled_reason(account) + + def describe_account(self, account: dict) -> dict: + return self._config.describe_account(account) + + async def start(self, ctx) -> object: + account = await self._config.resolve_account("default") + try: + config = getattr(ctx, "config", {}) or {} + except Exception: + config = {} + account = await self._config.resolve_account("default", config) + + self._security = DingTalkSecurity( + dm_policy=account.dm_policy, + group_policy=account.group_policy, + ) + self._deduplicator = DingTalkDeduplicator( + ttl=account.expires_in_seconds, + hot_reload=account.hot_reload, + ) + + import httpx + + self._http = httpx.AsyncClient(timeout=30.0) + + self._token_manager = DingTalkTokenManager(account.app_key, account.app_secret) + self._token_manager.set_http(self._http) + + self._gateway = DingTalkGateway(account, self._token_manager) + self._outbound = DingTalkOutbound(self._gateway) + self._outbound.set_http(self._http) + + if account.dingtalk_card_enabled and account.card_template_id: + from yuxi.channel.extensions.dingtalk.card import DingTalkCardManager + + self._card_manager = DingTalkCardManager(self._gateway, http=self._http) + self._outbound.set_card_manager(self._card_manager) + logger.info("DingTalk card manager initialized") + + self._monitor.inject_dependencies( + gateway=self._gateway, + outbound=self._outbound, + security=self._security, + pairing=self._pairing, + deduplicator=self._deduplicator, + file_cache=self._file_cache, + ) + self._gateway.set_handler(self._monitor) + + try: + from yuxi.channel.extensions.dingtalk.card import apply_ai_card_monkey_patch + + apply_ai_card_monkey_patch() + except Exception: + logger.debug("Failed to apply DingTalk AICard monkey patch", exc_info=True) + + dingtalk_stream_logger = logging.getLogger("dingtalk_stream") + dingtalk_stream_logger.setLevel(logging.WARNING) + + return await self._gateway.start(ctx) + + async def stop(self, ctx) -> None: + if self._gateway: + await self._gateway.stop(ctx) + if self._http: + await self._http.aclose() + self._http = None + + async def send_text( + self, + target_id: str, + content: str, + *, + reply_to_id: str | None = None, + thread_id: str | None = None, + account_id: str | None = None, + ) -> None: + if self._outbound: + await self._outbound.send_text(target_id, content, reply_to_id=reply_to_id) + + async def send_media( + self, + target_id: str, + media_url: str, + media_type: str, + reply_to_id: str | None = None, + thread_id: str | None = None, + ) -> None: + if self._outbound: + await self._outbound.send_media(target_id, media_url, media_type) + + async def probe(self, account: dict) -> bool: + return bool(account.get("app_key") and account.get("app_secret")) + + async def check_allowlist(self, peer_id: str, channel_type: str) -> bool: + if self._security: + return self._security.check_allowlist(peer_id) + return False + + def resolve_dm_policy(self) -> dict: + if self._security: + return {"mode": self._security.resolve_dm_policy(), "allow_from": []} + return {"mode": "open", "allow_from": []} + + async def generate_code(self, peer_id: str) -> str: + return await self._pairing.generate_code(peer_id) + + async def verify_code(self, peer_id: str, code: str) -> bool: + return await self._pairing.verify_code(peer_id, code) + + # ── DedupeProtocol ──────────────────────────────── + + def is_duplicate(self, key: str) -> bool: + if self._deduplicator: + return self._deduplicator.is_duplicate(key) + return False + + def mark_seen(self, key: str) -> None: + if self._deduplicator: + self._deduplicator._received.add(key) + + # ── CardProtocol ────────────────────────────────── + + async def send_card( + self, + target_id: str, + card_content: dict, + *, + reply_to_id: str | None = None, + account_id: str | None = None, + ) -> str | None: + if self._outbound: + result = await self._outbound.send_action_card(target_id, card_content) + if result.get("success"): + return result.get("response", {}).get("processQueryKey") + return None + + async def update_card( + self, + message_id: str, + card_content: dict, + *, + account_id: str | None = None, + ) -> bool: + return False + + # ── StreamingProtocol ───────────────────────────── + + @property + def streaming_mode(self) -> str: + return self._streaming.streaming_mode + + @property + def block_streaming_config(self) -> dict: + return self._streaming.block_streaming_config + + # ── DoctorProtocol ──────────────────────────────── + + @property + def dm_allow_from_mode(self) -> str: + return self._doctor.dm_allow_from_mode + + @property + def group_model(self) -> str: + return self._doctor.group_model + + group_allow_from_fallback_to_allow_from = False + warn_on_empty_group_sender_allowlist = True + + async def check_credentials(self, config: dict): + return await self._doctor.check_credentials(config) + + async def check_permissions(self, config: dict): + return await self._doctor.check_permissions(config) + + async def check_connectivity(self, config: dict): + return await self._doctor.check_connectivity(config) + + def normalize_compatibility_config(self, config: dict) -> dict: + return self._doctor.normalize_compatibility_config(config) + + def collect_allowlist_warnings(self, allowlist_config: dict): + return self._doctor.collect_allowlist_warnings(allowlist_config) + + def generate_repair_plan(self, diagnosis): + return self._doctor.generate_repair_plan(diagnosis) + + async def execute_repair(self, step, config: dict): + return await self._doctor.execute_repair(step, config) + + def legacy_config_rules(self): + return self._doctor.legacy_config_rules() + + # ── ConfigSchemaProtocol ────────────────────────── + + def config_schema(self) -> dict: + return { + "type": "object", + "properties": { + "app_key": { + "type": "string", + "label": "AppKey", + "description": "钉钉应用 AppKey", + "required": True, + }, + "app_secret": { + "type": "string", + "label": "AppSecret", + "description": "钉钉应用 AppSecret", + "required": True, + "secret": True, + }, + "robot_code": { + "type": "string", + "label": "Robot Code", + "description": "钉钉机器人 Code", + }, + "mode": { + "type": "string", + "label": "连接模式", + "description": "stream 或 webhook", + "default": "stream", + }, + "streaming": { + "type": "string", + "label": "流式模式", + "description": "block 或 card_kit", + "default": "block", + }, + "dingtalk_card_enabled": { + "type": "boolean", + "label": "启用卡片流式", + "default": False, + }, + "card_template_id": { + "type": "string", + "label": "卡片模板 ID", + "description": "钉钉互动卡片模板 ID", + }, + "dm_policy": { + "type": "string", + "label": "DM 策略", + "enum": ["open", "allowlist", "pairing", "disabled"], + "default": "open", + }, + "group_policy": { + "type": "string", + "label": "群聊策略", + "enum": ["open", "allowlist", "disabled"], + "default": "open", + }, + }, + } + + # ── MessageAction handlers ──────────────────────── + + async def _handle_send(self, *, content: str, target_id: str, reply_to_id: str | None = None, **kwargs) -> dict: + await self.send_text(target_id, content, reply_to_id=reply_to_id) + return {"success": True} + + async def _handle_send_media(self, *, media_url: str, media_type: str, target_id: str, **kwargs) -> dict: + await self.send_media(target_id, media_url, media_type) + return {"success": True} + + async def _handle_reply(self, *, content: str, target_id: str, reply_to_id: str, **kwargs) -> dict: + await self.send_text(target_id, content, reply_to_id=reply_to_id) + return {"success": True} + + async def _handle_add_member(self, *, group_id: str, user_ids: list[str], **kwargs) -> dict: + if self._outbound: + return await self._outbound.add_group_members(group_id, user_ids) + return {"success": False, "error": "Outbound not initialized"} + + async def _handle_remove_member(self, *, group_id: str, user_ids: list[str], **kwargs) -> dict: + if self._outbound: + return await self._outbound.remove_group_members(group_id, user_ids) + return {"success": False, "error": "Outbound not initialized"} + + +dingtalk_plugin = DingTalkPlugin() +ChannelPluginRegistry.register(dingtalk_plugin) diff --git a/backend/package/yuxi/channel/extensions/dingtalk/card.py b/backend/package/yuxi/channel/extensions/dingtalk/card.py new file mode 100644 index 00000000..a9ead937 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/dingtalk/card.py @@ -0,0 +1,173 @@ +from __future__ import annotations + +import json +import logging +import uuid + +import httpx + +logger = logging.getLogger(__name__) + +CREATE_AND_DELIVER_URL = "https://api.dingtalk.com/v1.0/card/instances/createAndDeliver" +STREAMING_UPDATE_URL = "https://api.dingtalk.com/v1.0/card/streamingUpdate" + + +class DingTalkCardManager: + def __init__(self, gateway, http: httpx.AsyncClient | None = None): + self._gateway = gateway + self._http = http + self._card_instance_id: str | None = None + self._out_track_id: str | None = None + + def set_http(self, http: httpx.AsyncClient) -> None: + self._http = http + + async def _get_http(self) -> httpx.AsyncClient: + if self._http is not None: + return self._http + client = httpx.AsyncClient(timeout=15.0) + self._http = client + return client + + async def create_and_deliver( + self, + open_conversation_id: str, + robot_code: str, + card_template_id: str, + title: str = "AI 正在思考...", + ) -> bool: + token = await self._gateway.token_manager.get_access_token() + if not token: + return False + + self._out_track_id = str(uuid.uuid4()) + + body = { + "cardTemplateId": card_template_id, + "outTrackId": self._out_track_id, + "robotCode": robot_code, + "openConversationId": open_conversation_id, + "cardData": json.dumps({"title": title, "content": ""}), + "imGroupOpenDeliverModel": {"robotCode": robot_code}, + "callbackType": "STREAM", + } + + http = await self._get_http() + resp = await http.post( + CREATE_AND_DELIVER_URL, + headers={ + "x-acs-dingtalk-access-token": token, + "Content-Type": "application/json", + }, + json=body, + ) + data = resp.json() + self._card_instance_id = data.get("cardInstanceId") + return bool(self._card_instance_id) + + async def streaming_update(self, content: str, status: str = "PROCESSING") -> bool: + if not self._card_instance_id: + return False + + token = await self._gateway.token_manager.get_access_token() + if not token: + return False + + body = { + "outTrackId": self._out_track_id, + "cardInstanceId": self._card_instance_id, + "content": content, + "fullContent": False, + "flowStatus": status, + } + + http = await self._get_http() + resp = await http.put( + STREAMING_UPDATE_URL, + headers={ + "x-acs-dingtalk-access-token": token, + "Content-Type": "application/json", + }, + json=body, + ) + return resp.status_code == 200 + + +def build_approval_card_buttons(request_id: str) -> list[dict]: + return [ + { + "title": "批准", + "actionURL": f"dingtalk://yuxi/approval/approve?req={request_id}", + }, + { + "title": "拒绝", + "actionURL": f"dingtalk://yuxi/approval/deny?req={request_id}", + }, + ] + + +def build_action_card(title: str, text: str, buttons: list[dict], btn_orientation: str = "1") -> dict: + return { + "msgtype": "actionCard", + "actionCard": { + "title": title, + "text": text, + "btnOrientation": btn_orientation, + "btns": buttons, + }, + } + + +def build_markdown_message(title: str, text: str) -> dict: + return { + "msgtype": "markdown", + "markdown": { + "title": title, + "text": text, + }, + } + + +def build_image_reply_button(image_url: str, prompt_en: str = "") -> dict | None: + if not image_url or not prompt_en: + return None + return { + "title": "查看原图", + "actionURL": image_url, + } + + +def apply_ai_card_monkey_patch(): + try: + from dingtalk_stream import AICardReplier, AICardStatus, CardReplier + + class CustomAICardReplier(CardReplier): + async def start(self, card_instance_id: str) -> None: + self._card_instance_id = card_instance_id + await self._update_card( + card_instance_id, + { + "flowStatus": AICardStatus.PROCESSING, + "content": "", + }, + ) + + async def update_content(self, content: str) -> None: + await self._update_card( + self._card_instance_id, + { + "content": content, + "flowStatus": AICardStatus.PROCESSING, + }, + ) + + async def finish(self) -> None: + await self._update_card( + self._card_instance_id, + {"flowStatus": AICardStatus.FINISHED}, + ) + + AICardReplier.start = CustomAICardReplier.start + logger.info("DingTalk AICardReplier monkey patch applied") + except ImportError: + logger.debug("dingtalk_stream SDK not available, skipping monkey patch") diff --git a/backend/package/yuxi/channel/extensions/dingtalk/config.py b/backend/package/yuxi/channel/extensions/dingtalk/config.py new file mode 100644 index 00000000..369772f6 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/dingtalk/config.py @@ -0,0 +1,187 @@ +from __future__ import annotations + +import logging +import os + +from yuxi.channel.extensions.dingtalk.types import DingTalkAccount + +logger = logging.getLogger(__name__) + + +class DingTalkConfig: + def list_account_ids(self, config: dict) -> list[str]: + dingtalk_cfg = _get_dingtalk_config(config) + accounts = dingtalk_cfg.get("accounts", {}) + if accounts: + return list(accounts.keys()) + if dingtalk_cfg.get("appKey") or dingtalk_cfg.get("app_key"): + return ["default"] + if os.environ.get("DINGTALK_APP_KEY"): + return ["default"] + return [] + + def default_account_id(self, config: dict) -> str: + dingtalk_cfg = _get_dingtalk_config(config) + return dingtalk_cfg.get("defaultAccount") or dingtalk_cfg.get("default_account", "default") + + async def resolve_account(self, account_id: str, config: dict | None = None) -> DingTalkAccount: + dingtalk_cfg = _get_dingtalk_config(config) if config else {} + accounts = dingtalk_cfg.get("accounts", {}) + account = accounts.get(account_id, {}) if accounts else {} + + app_key = ( + account.get("appKey") + or account.get("app_key") + or dingtalk_cfg.get("appKey") + or dingtalk_cfg.get("app_key") + or os.environ.get("DINGTALK_APP_KEY", "") + ) + app_secret = ( + account.get("appSecret") + or account.get("app_secret") + or dingtalk_cfg.get("appSecret") + or dingtalk_cfg.get("app_secret") + or os.environ.get("DINGTALK_APP_SECRET", "") + ) + robot_code = ( + account.get("robotCode") + or account.get("robot_code") + or dingtalk_cfg.get("robotCode") + or dingtalk_cfg.get("robot_code") + or os.environ.get("DINGTALK_ROBOT_CODE", "") + ) + + mode = account.get("mode") or dingtalk_cfg.get("mode", "stream") + streaming = account.get("streaming") or dingtalk_cfg.get("streaming", "block") + dingtalk_card_enabled = ( + account.get("dingtalkCardEnabled") + if "dingtalkCardEnabled" in account + else dingtalk_cfg.get("dingtalkCardEnabled", dingtalk_cfg.get("dingtalk_card_enabled", False)) + ) + card_template_id = ( + account.get("cardTemplateId") + or account.get("card_template_id") + or dingtalk_cfg.get("cardTemplateId") + or dingtalk_cfg.get("card_template_id", "") + ) + dm_policy = ( + account.get("dmPolicy") + or account.get("dm_policy") + or dingtalk_cfg.get("dmPolicy") + or dingtalk_cfg.get("dm_policy", "open") + ) + group_policy = ( + account.get("groupPolicy") + or account.get("group_policy") + or dingtalk_cfg.get("groupPolicy") + or dingtalk_cfg.get("group_policy", "open") + ) + group_shared_session = ( + account.get("groupSharedSession") + if "groupSharedSession" in account + else dingtalk_cfg.get("groupSharedSession", dingtalk_cfg.get("group_shared_session", True)) + ) + expires_in_seconds = ( + account.get("expiresInSeconds") + or dingtalk_cfg.get("expiresInSeconds") + or dingtalk_cfg.get("expires_in_seconds", 3600) + ) + hot_reload = ( + account.get("hotReload") + if "hotReload" in account + else dingtalk_cfg.get("hotReload", dingtalk_cfg.get("hot_reload", False)) + ) + chat_time_module = ( + account.get("chatTimeModule") + if "chatTimeModule" in account + else dingtalk_cfg.get("chatTimeModule", dingtalk_cfg.get("chat_time_module", False)) + ) + chat_start_time = ( + account.get("chatStartTime") + or dingtalk_cfg.get("chatStartTime") + or dingtalk_cfg.get("chat_start_time", "00:00") + ) + chat_stop_time = ( + account.get("chatStopTime") + or dingtalk_cfg.get("chatStopTime") + or dingtalk_cfg.get("chat_stop_time", "24:00") + ) + agent = account.get("agent") if "agent" in account else dingtalk_cfg.get("agent", False) + agent_workspace = ( + account.get("agentWorkspace") + or dingtalk_cfg.get("agentWorkspace") + or dingtalk_cfg.get("agent_workspace", "~/cow") + ) + render_mode = ( + account.get("renderMode") + or account.get("render_mode") + or dingtalk_cfg.get("renderMode") + or dingtalk_cfg.get("render_mode", "auto") + ) + subscribe_topics = ( + account.get("subscribeTopics") + or account.get("subscribe_topics") + or dingtalk_cfg.get("subscribeTopics") + or dingtalk_cfg.get("subscribe_topics", []) + ) + enabled = account.get("enabled") if "enabled" in account else dingtalk_cfg.get("enabled", True) + + return DingTalkAccount( + account_id=account_id, + app_key=app_key, + app_secret=app_secret, + robot_code=robot_code, + mode=mode, + streaming=streaming, + dingtalk_card_enabled=dingtalk_card_enabled, + card_template_id=card_template_id, + dm_policy=dm_policy, + group_policy=group_policy, + group_shared_session=group_shared_session, + expires_in_seconds=expires_in_seconds, + hot_reload=hot_reload, + chat_time_module=chat_time_module, + chat_start_time=chat_start_time, + chat_stop_time=chat_stop_time, + agent=agent, + agent_workspace=agent_workspace, + render_mode=render_mode, + subscribe_topics=subscribe_topics if isinstance(subscribe_topics, list) else [], + enabled=enabled, + ) + + def is_configured(self, account: DingTalkAccount | dict) -> bool: + if isinstance(account, DingTalkAccount): + return account.is_configured() + return bool(account.get("app_key") or account.get("appKey")) and bool( + account.get("app_secret") or account.get("appSecret") + ) + + def is_enabled(self, account: DingTalkAccount | dict) -> bool: + if isinstance(account, DingTalkAccount): + return account.enabled + return account.get("enabled", True) + + def disabled_reason(self, account: DingTalkAccount | dict) -> str: + if not self.is_configured(account): + return "AppKey or AppSecret not configured" + return "" + + def describe_account(self, account: DingTalkAccount | dict) -> dict: + if isinstance(account, DingTalkAccount): + return { + "account_id": account.account_id, + "configured": account.is_configured(), + "mode": account.mode, + } + return { + "account_id": account.get("account_id", ""), + "configured": self.is_configured(account), + "mode": account.get("mode", "stream"), + } + + +def _get_dingtalk_config(config: dict | None) -> dict: + if not config: + return {} + return config.get("channels", {}).get("dingtalk", {}) diff --git a/backend/package/yuxi/channel/extensions/dingtalk/dedupe.py b/backend/package/yuxi/channel/extensions/dingtalk/dedupe.py new file mode 100644 index 00000000..0d1f4727 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/dingtalk/dedupe.py @@ -0,0 +1,50 @@ +from __future__ import annotations + +import time +from collections import OrderedDict + +DEFAULT_TTL = 3600 + + +class ExpiredDict: + def __init__(self, ttl: int = DEFAULT_TTL): + self._ttl = ttl + self._data: OrderedDict[str, tuple[object, float]] = OrderedDict() + + def __contains__(self, key: str) -> bool: + self._cleanup_expired() + return key in self._data + + def add(self, key: str, value: object = True) -> None: + self._cleanup_expired() + self._data[key] = (value, time.time() + self._ttl) + self._data.move_to_end(key) + + def _cleanup_expired(self) -> None: + now = time.time() + while self._data: + _, (_, expires) = next(iter(self._data.items())) + if expires > now: + break + self._data.popitem(last=False) + + +class DingTalkDeduplicator: + def __init__(self, ttl: int = DEFAULT_TTL, hot_reload: bool = False): + self._received = ExpiredDict(ttl) + self._hot_reload = hot_reload + self._start_time = time.time() + + def is_duplicate(self, msg_id: str, create_time: float | None = None) -> bool: + if msg_id in self._received: + return True + + if self._hot_reload and create_time: + if create_time < self._start_time - 60: + return True + + self._received.add(msg_id) + return False + + def is_my_msg(self, msg) -> bool: + return getattr(msg, "my_msg", False) diff --git a/backend/package/yuxi/channel/extensions/dingtalk/doctor.py b/backend/package/yuxi/channel/extensions/dingtalk/doctor.py new file mode 100644 index 00000000..b2ec54b9 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/dingtalk/doctor.py @@ -0,0 +1,215 @@ +from __future__ import annotations + +import logging +import time + +import httpx + +from yuxi.channel.doctor.models import ( + CheckStatus, + ConnectivityCheckResult, + CredentialCheckResult, + DiagnosisResult, + DiagnosisWarning, + LegacyConfigRule, + PermissionCheckResult, + RepairResult, + RepairStep, + Severity, +) + +logger = logging.getLogger(__name__) + +DINGTALK_TOKEN_URL = "https://api.dingtalk.com/v1.0/oauth2/accessToken" +DINGTALK_CONNECTIVITY_URL = "https://api.dingtalk.com" + + +def _extract_credential(config: dict) -> tuple[str, str]: + channels = config.get("channels", {}) + dingtalk = channels.get("dingtalk", {}) + app_key = ( + dingtalk.get("appKey") + or dingtalk.get("app_key") + or "" + ) + app_secret = ( + dingtalk.get("appSecret") + or dingtalk.get("app_secret") + or "" + ) + return app_key, app_secret + + +class DingTalkDoctor: + dm_allow_from_mode = "topOnly" + group_model = "route" + group_allow_from_fallback_to_allow_from = False + warn_on_empty_group_sender_allowlist = True + + async def check_credentials(self, config: dict) -> CredentialCheckResult: + app_key, app_secret = _extract_credential(config) + if not app_key or not app_secret: + return CredentialCheckResult( + status=CheckStatus.FAIL, + token_obtained=False, + error="AppKey 或 AppSecret 未配置", + ) + + try: + async with httpx.AsyncClient(timeout=15.0) as client: + resp = await client.post( + DINGTALK_TOKEN_URL, + json={"appKey": app_key, "appSecret": app_secret}, + ) + data = resp.json() + if "accessToken" in data: + expires_in = data.get("expireIn", 7200) + return CredentialCheckResult( + status=CheckStatus.PASS, + token_obtained=True, + expires_in=expires_in, + ) + error_msg = data.get("message", str(data)) + return CredentialCheckResult( + status=CheckStatus.FAIL, + token_obtained=False, + error=error_msg, + ) + except httpx.HTTPStatusError as e: + return CredentialCheckResult( + status=CheckStatus.FAIL, + token_obtained=False, + error=f"HTTP {e.response.status_code}: {e.response.text[:200]}", + ) + except Exception as e: + logger.warning("DingTalk credential check error: %s", e) + return CredentialCheckResult( + status=CheckStatus.FAIL, + token_obtained=False, + error=str(e), + ) + + async def check_permissions(self, config: dict) -> PermissionCheckResult: + app_key, app_secret = _extract_credential(config) + if not app_key or not app_secret: + return PermissionCheckResult( + status=CheckStatus.FAIL, + error="AppKey 或 AppSecret 未配置，无法检查权限", + ) + + try: + async with httpx.AsyncClient(timeout=15.0) as client: + token_resp = await client.post( + DINGTALK_TOKEN_URL, + json={"appKey": app_key, "appSecret": app_secret}, + ) + token_data = token_resp.json() + access_token = token_data.get("accessToken") + if not access_token: + return PermissionCheckResult( + status=CheckStatus.FAIL, + error="无法获取 access_token，权限检查中止", + ) + + resp = await client.get( + "https://api.dingtalk.com/v1.0/robot/organizations/briefManage", + headers={"x-acs-dingtalk-access-token": access_token}, + ) + if resp.status_code == 200: + return PermissionCheckResult( + status=CheckStatus.PASS, + can_send_message=True, + can_read_message=True, + ) + return PermissionCheckResult( + status=CheckStatus.WARN, + can_send_message=True, + can_read_message=True, + error=f"权限查询异常: HTTP {resp.status_code}", + ) + except Exception as e: + logger.warning("DingTalk permission check error: %s", e) + return PermissionCheckResult( + status=CheckStatus.FAIL, + error=str(e), + ) + + async def check_connectivity(self, config: dict) -> ConnectivityCheckResult: + try: + start = time.monotonic() + async with httpx.AsyncClient(timeout=10.0) as client: + resp = await client.get(DINGTALK_CONNECTIVITY_URL) + latency_ms = (time.monotonic() - start) * 1000 + if resp.status_code < 500: + return ConnectivityCheckResult( + status=CheckStatus.PASS, + latency_ms=round(latency_ms, 2), + endpoint=DINGTALK_CONNECTIVITY_URL, + ) + return ConnectivityCheckResult( + status=CheckStatus.FAIL, + latency_ms=round(latency_ms, 2), + endpoint=DINGTALK_CONNECTIVITY_URL, + error=f"HTTP {resp.status_code}", + ) + except Exception as e: + logger.warning("DingTalk connectivity check error: %s", e) + return ConnectivityCheckResult( + status=CheckStatus.FAIL, + endpoint=DINGTALK_CONNECTIVITY_URL, + error=str(e), + ) + + def normalize_compatibility_config(self, config: dict) -> dict: + return config + + def collect_allowlist_warnings(self, allowlist_config: dict) -> list[DiagnosisWarning]: + warnings: list[DiagnosisWarning] = [] + dm_allowlist = allowlist_config.get("dm_allowlist", []) + if not dm_allowlist: + warnings.append( + DiagnosisWarning( + severity=Severity.WARNING, + code="dm_allowlist_empty", + message="DM 白名单为空，任何用户都可以直接向机器人发消息", + suggestion="建议配置 dm_allowlist 限制可访问用户", + ) + ) + return warnings + + def generate_repair_plan(self, diagnosis: DiagnosisResult) -> list[RepairStep]: + plan: list[RepairStep] = [] + if diagnosis.credential and diagnosis.credential.status == CheckStatus.FAIL: + plan.append( + RepairStep( + id="fix_dingtalk_credential", + description="检查钉钉 AppKey/AppSecret 配置", + action="update_config", + params={"field": "appKey/appSecret"}, + ) + ) + if diagnosis.connectivity and diagnosis.connectivity.status == CheckStatus.FAIL: + plan.append( + RepairStep( + id="fix_dingtalk_connectivity", + description="检查网络连通性和钉钉 API 可达性", + action="check_network", + ) + ) + return plan + + async def execute_repair(self, step: RepairStep, config: dict) -> RepairResult: + if step.id == "fix_dingtalk_credential": + return RepairResult( + step_id=step.id, + success=False, + message="凭证修复需要手动更新 AppKey/AppSecret 配置", + ) + return RepairResult( + step_id=step.id, + success=False, + message=f"未知修复步骤: {step.id}", + ) + + def legacy_config_rules(self) -> list[LegacyConfigRule]: + return [] diff --git a/backend/package/yuxi/channel/extensions/dingtalk/file_cache.py b/backend/package/yuxi/channel/extensions/dingtalk/file_cache.py new file mode 100644 index 00000000..ba1ac93d --- /dev/null +++ b/backend/package/yuxi/channel/extensions/dingtalk/file_cache.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +import logging +import time +from collections import OrderedDict + +logger = logging.getLogger(__name__) + +DEFAULT_TTL_SECONDS = 120 + + +class FileCacheEntry: + __slots__ = ("file_path", "cached_at") + + def __init__(self, file_path: str): + self.file_path = file_path + self.cached_at = time.monotonic() + + +class FileCache: + def __init__(self, ttl_seconds: int = DEFAULT_TTL_SECONDS): + self._ttl = ttl_seconds + self._entries: OrderedDict[str, FileCacheEntry] = OrderedDict() + + def add(self, session_key: str, file_path: str) -> None: + self._evict_expired() + self._entries[session_key] = FileCacheEntry(file_path) + + def get(self, session_key: str) -> FileCacheEntry | None: + self._evict_expired() + return self._entries.get(session_key) + + def clear(self, session_key: str) -> None: + self._entries.pop(session_key, None) + + def _evict_expired(self) -> None: + now = time.monotonic() + expired = [k for k, v in self._entries.items() if now - v.cached_at > self._ttl] + for k in expired: + self._entries.pop(k, None) diff --git a/backend/package/yuxi/channel/extensions/dingtalk/gateway.py b/backend/package/yuxi/channel/extensions/dingtalk/gateway.py new file mode 100644 index 00000000..c0f6d2e4 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/dingtalk/gateway.py @@ -0,0 +1,207 @@ +from __future__ import annotations + +import asyncio +import json +import logging +import time +from typing import TYPE_CHECKING + +import httpx + +from yuxi.channel.extensions.dingtalk.types import DingTalkAccount + +if TYPE_CHECKING: + from yuxi.channel.extensions.dingtalk.monitor import DingTalkMonitor + +logger = logging.getLogger(__name__) + +TOKEN_URL = "https://api.dingtalk.com/v1.0/oauth2/accessToken" +TOKEN_REFRESH_MARGIN = 300 +RECONNECT_WAIT_MS = 100 +RECONNECT_MAX_WAITS = 100 +OPEN_CONNECTION_URL = "https://api.dingtalk.com/v1.0/gateway/connections/open" + + +class DingTalkTokenManager: + def __init__(self, app_key: str, app_secret: str): + self._app_key = app_key + self._app_secret = app_secret + self._token: str | None = None + self._expires_at: float = 0.0 + self._lock = asyncio.Lock() + self._http: httpx.AsyncClient | None = None + + def set_http(self, http: httpx.AsyncClient) -> None: + self._http = http + + @property + def token(self) -> str | None: + if self._token and time.time() < self._expires_at: + return self._token + return None + + async def get_access_token(self) -> str | None: + if self._token and time.time() < self._expires_at: + return self._token + + async with self._lock: + if self._token and time.time() < self._expires_at: + return self._token + + try: + resp = await self._http.post( + TOKEN_URL, + json={ + "appKey": self._app_key, + "appSecret": self._app_secret, + }, + timeout=10.0, + ) + data = resp.json() + self._token = data["accessToken"] + expire_in = data.get("expireIn", 7200) + self._expires_at = time.time() + expire_in - TOKEN_REFRESH_MARGIN + logger.info("DingTalk access_token refreshed, expires_in=%ds", expire_in) + return self._token + except Exception: + logger.exception("Failed to get DingTalk access_token") + return None + + +class DingTalkGateway: + def __init__(self, account: DingTalkAccount, token_manager: DingTalkTokenManager): + self._account = account + self._token_manager = token_manager + self._running = False + self._event_loop: asyncio.AbstractEventLoop | None = None + self._robot_code_cache: str | None = None + self._handler: DingTalkMonitor | None = None + self._stream_client = None + self._credential = None + + def set_handler(self, handler: DingTalkMonitor) -> None: + self._handler = handler + + @property + def robot_code(self) -> str | None: + return self._robot_code_cache or self._account.robot_code + + @property + def app_key(self) -> str: + return self._account.app_key + + @property + def app_secret(self) -> str: + return self._account.app_secret + + @property + def token_manager(self) -> DingTalkTokenManager: + return self._token_manager + + async def start(self, ctx) -> object: + if not self._account.is_configured(): + logger.warning("DingTalk account not configured, skipping start") + return {"running": False, "reason": "not-configured"} + + self._running = True + self._event_loop = asyncio.get_running_loop() + + try: + from dingtalk_stream import ChatbotMessage, Credential, DingTalkStreamClient + + self._credential = Credential(self._account.app_key, self._account.app_secret) + + self._stream_client = DingTalkStreamClient( + credential=self._credential, + ) + self._stream_client.register_callback_handler(ChatbotMessage.TOPIC, self._handler) + + extra_topics = getattr(self._account, "subscribe_topics", []) + for topic in extra_topics: + try: + self._stream_client.register_callback_handler(topic, self._handler) + logger.info("DingTalk extra topic registered: %s", topic) + except Exception: + logger.warning("Failed to register topic: %s", topic, exc_info=True) + + try: + self._stream_client.register_callback_handler("/v1.0/card/instances/callback", self._handler) + logger.info("DingTalk card callback registered") + except Exception: + logger.debug("Failed to register card callback", exc_info=True) + + self._stream_client.pre_start() + except ImportError: + logger.error("dingtalk-stream-sdk-python not installed") + return {"running": False, "reason": "sdk-missing"} + except Exception: + logger.exception("Failed to initialize DingTalk stream client") + return {"running": False, "reason": "init-failed"} + + asyncio.create_task(self._connection_loop(), name="dingtalk-stream-loop") + logger.info("DingTalk gateway started for account %s", self._account.account_id) + return {"running": True, "account_id": self._account.account_id} + + async def stop(self, ctx) -> None: + self._running = False + self._stream_client = None + logger.info("DingTalk gateway stopped") + + async def _open_connection(self) -> tuple[str, str]: + async with httpx.AsyncClient(timeout=10.0) as http: + resp = await http.post( + OPEN_CONNECTION_URL, + headers={ + "Content-Type": "application/json", + "clientId": self._account.app_key, + "dingtalk-accept-encoding": "base64", + }, + ) + resp.raise_for_status() + data = resp.json() + return data["endpoint"], data["ticket"] + + async def _connection_loop(self): + first_attempt = True + + while self._running: + try: + endpoint, ticket = await self._open_connection() + except Exception as e: + if first_attempt: + logger.error("DingTalk first connection failed: %s", e) + first_attempt = False + await self._wait_with_check(RECONNECT_MAX_WAITS) + continue + + try: + import websockets + + async with websockets.connect(f"{endpoint}?ticket={ticket}") as ws: + if first_attempt: + logger.info("DingTalk stream connected") + first_attempt = False + + async for raw_message in ws: + if not self._running: + break + try: + json_message = json.loads(raw_message) + result = self._stream_client.route_message(json_message) + if result == "TAG_DISCONNECT": + logger.warning("DingTalk stream: TAG_DISCONNECT received") + break + except Exception: + logger.exception("DingTalk stream message routing error") + except Exception: + logger.exception("DingTalk stream connection error") + await self._wait_with_check(RECONNECT_MAX_WAITS) + + async def _wait_with_check(self, max_waits: int): + for _ in range(max_waits): + if not self._running: + break + await asyncio.sleep(RECONNECT_WAIT_MS / 1000) + + async def probe(self, account: dict) -> bool: + return account.get("app_key") and account.get("app_secret") diff --git a/backend/package/yuxi/channel/extensions/dingtalk/media.py b/backend/package/yuxi/channel/extensions/dingtalk/media.py new file mode 100644 index 00000000..43c87f15 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/dingtalk/media.py @@ -0,0 +1,114 @@ +from __future__ import annotations + +import hashlib +import logging +import os +import tempfile + +import httpx + +logger = logging.getLogger(__name__) + +MEDIA_UPLOAD_URL = "https://oapi.dingtalk.com/media/upload" +MESSAGE_FILES_DOWNLOAD_URL = "https://api.dingtalk.com/v1.0/robot/messageFiles/download" + + +async def upload_media( + http: httpx.AsyncClient, + token: str, + file_path: str, + media_type: str, +) -> str | None: + if file_path.startswith("file://"): + file_path = file_path[7:] + + if file_path.startswith(("http://", "https://")): + file_path = await _download_to_tmp(http, file_path) + + if not os.path.exists(file_path): + logger.error("Media file not found: %s", file_path) + return None + + with open(file_path, "rb") as f: + resp = await http.post( + MEDIA_UPLOAD_URL, + params={"access_token": token, "type": media_type}, + files={"media": (os.path.basename(file_path), f)}, + timeout=60.0, + ) + data = resp.json() + if data.get("errcode") == 0: + return data.get("media_id") + logger.error("Media upload failed: %s", data) + return None + + +async def download_image( + http: httpx.AsyncClient, + token: str, + download_code: str, + robot_code: str, + save_dir: str | None = None, +) -> str | None: + resp = await http.post( + MESSAGE_FILES_DOWNLOAD_URL, + headers={ + "x-acs-dingtalk-access-token": token, + "Content-Type": "application/json", + }, + json={ + "downloadCode": download_code, + "robotCode": robot_code, + }, + timeout=30.0, + ) + data = resp.json() + download_url = data.get("downloadUrl") + if not download_url: + logger.error("Failed to get download URL: %s", data) + return None + + img_resp = await http.get(download_url, timeout=30.0) + img_resp.raise_for_status() + + img_data = img_resp.content + ext = _guess_extension(img_data) + filename = f"{hashlib.md5(img_data).hexdigest()}{ext}" + + if save_dir: + os.makedirs(save_dir, exist_ok=True) + save_path = os.path.join(save_dir, filename) + else: + tmp = tempfile.NamedTemporaryFile(delete=False, suffix=ext) + save_path = tmp.name + tmp.close() + + with open(save_path, "wb") as f: + f.write(img_data) + + logger.debug("Image downloaded: %s -> %s", download_code, save_path) + return save_path + + +async def _download_to_tmp(http: httpx.AsyncClient, url: str) -> str: + resp = await http.get(url, timeout=60.0) + resp.raise_for_status() + data = resp.content + ext = _guess_extension(data) + + tmp = tempfile.NamedTemporaryFile(delete=False, suffix=ext) + tmp.write(data) + tmp.close() + return tmp.name + + +def _guess_extension(data: bytes) -> str: + if data[:4] == b"\x89PNG": + return ".png" + if data[:2] == b"\xff\xd8": + return ".jpg" + if data[:4] == b"GIF8": + return ".gif" + if data[:4] == b"RIFF" and data[8:12] == b"WEBP": + return ".webp" + return ".bin" diff --git a/backend/package/yuxi/channel/extensions/dingtalk/monitor.py b/backend/package/yuxi/channel/extensions/dingtalk/monitor.py new file mode 100644 index 00000000..dfad7bd1 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/dingtalk/monitor.py @@ -0,0 +1,397 @@ +from __future__ import annotations + +import asyncio +import json +import logging +import time +from datetime import UTC, datetime + +try: + from dingtalk_stream import AckMessage, ChatbotMessage +except ImportError: + AckMessage = None + ChatbotMessage = None + +from yuxi.channel.extensions.dingtalk.dedupe import DingTalkDeduplicator +from yuxi.channel.extensions.dingtalk.file_cache import FileCache + +logger = logging.getLogger(__name__) + +STALE_MSG_AGE_S = 60 + +BUILTIN_COMMANDS = { + "/help": "查看帮助信息", + "/status": "查看 Bot 运行状态", + "/ping": "测试连通性", +} + + +class CommandResult: + __slots__ = ("handled", "response") + + def __init__(self, handled: bool = False, response: str = ""): + self.handled = handled + self.response = response + + +class DingTalkMonitor: + def __init__(self): + self._robot_code_cache: str | None = None + self._deduplicator: DingTalkDeduplicator | None = None + self._file_cache: FileCache | None = None + self._security = None + self._pairing = None + self._outbound = None + self._gateway = None + self._dispatch_callback = None + + def inject_dependencies( + self, + *, + gateway=None, + outbound=None, + security=None, + pairing=None, + deduplicator: DingTalkDeduplicator | None = None, + file_cache: FileCache | None = None, + ) -> None: + self._gateway = gateway + self._outbound = outbound + self._security = security + self._pairing = pairing + self._deduplicator = deduplicator + self._file_cache = file_cache + + def set_dispatch_callback(self, callback) -> None: + self._dispatch_callback = callback + + async def process(self, callback): + topic = getattr(callback, "topic", "") + if topic and "/card/instances/callback" in topic: + return await self._handle_card_callback(callback) + + try: + incoming = ChatbotMessage.from_dict(callback.data) + except Exception: + logger.exception("Failed to parse DingTalk ChatbotMessage") + return AckMessage.STATUS_SYSTEM_EXCEPTION, "ERROR" + + self._robot_code_cache = incoming.robot_code + + msg_age_s = (time.time() * 1000 - incoming.create_at) / 1000 + if msg_age_s > STALE_MSG_AGE_S: + logger.debug("DingTalk stale message ignored, age=%.1fs", msg_age_s) + return AckMessage.STATUS_OK, "OK" + + is_group = incoming.conversation_type != "1" + + if self._deduplicator: + create_time_s = incoming.create_at / 1000 if incoming.create_at else None + if self._deduplicator.is_duplicate(incoming.message_id, create_time_s): + logger.debug("Duplicate DingTalk message: %s", incoming.message_id) + return AckMessage.STATUS_OK, "OK" + + if not is_group and self._deduplicator and self._deduplicator.is_my_msg(incoming): + logger.debug("Filtered own message in direct chat: %s", incoming.message_id) + return AckMessage.STATUS_OK, "OK" + + if incoming.message_type == "picture": + image_path = await self._handle_picture(incoming, is_group) + if image_path: + content = f"[图片: {image_path}]" + unified = self._build_unified_message(incoming, content, "IMAGE", is_group) + if self._dispatch_callback: + asyncio.create_task( + self._dispatch_callback(unified), + name=f"dingtalk-dispatch-{incoming.message_id[:8]}", + ) + return AckMessage.STATUS_OK, "OK" + + content, msg_type = await self._parse_message_content(incoming, is_group) + if content is None: + return AckMessage.STATUS_OK, "OK" + + if msg_type == "TEXT" and content: + cmd_result = await self._handle_command(content, incoming, is_group) + if cmd_result.handled: + return AckMessage.STATUS_OK, "OK" + + if is_group: + from_user_id = incoming.conversation_id + other_user_id = incoming.conversation_id + actual_user_id = incoming.sender_id + else: + from_user_id = incoming.sender_id + other_user_id = incoming.sender_id + actual_user_id = incoming.sender_id + + unified = { + "msg_id": incoming.message_id, + "channel_type": "dingtalk", + "account_id": "default", + "content": content, + "message_type": msg_type, + "sender": { + "id": from_user_id, + "display_name": getattr(incoming, "sender_nick", None) or from_user_id, + "kind": "GROUP" if is_group else "DIRECT", + }, + "group": { + "id": incoming.conversation_id, + "type": "group" if is_group else "direct", + } + if is_group + else None, + "timestamp": datetime.fromtimestamp(incoming.create_at / 1000, tz=UTC), + "mentioned_ids": self._extract_mentioned_ids(incoming), + "reply_to_id": None, + "thread_id": None, + "session_key": incoming.conversation_id if is_group else incoming.sender_id, + "raw_payload": incoming, + "metadata": { + "conversation_id": incoming.conversation_id, + "conversation_type": "2" if is_group else "1", + "sender_staff_id": getattr(incoming, "sender_staff_id", ""), + "robot_code": self._robot_code_cache, + "is_group": is_group, + "actual_user_id": actual_user_id, + "other_user_id": other_user_id, + }, + } + + if self._dispatch_callback: + asyncio.create_task( + self._dispatch_callback(unified), + name=f"dingtalk-dispatch-{incoming.message_id[:8]}", + ) + else: + logger.warning("DingTalk: no dispatch callback set, message dropped: %s", incoming.message_id) + + return AckMessage.STATUS_OK, "OK" + + async def _handle_picture(self, incoming, is_group: bool) -> str | None: + image_list = incoming.get_image_list() if hasattr(incoming, "get_image_list") else [] + if not image_list: + return None + download_code = image_list[0] + image_path = await self._download_image(incoming, download_code) + if image_path and self._file_cache: + cache_key = incoming.conversation_id if is_group else incoming.sender_id + self._file_cache.add(cache_key, image_path) + logger.debug("DingTalk image cached: %s -> %s", download_code, image_path) + return image_path + + async def _parse_message_content(self, incoming, is_group: bool) -> tuple[str | None, str | None]: + msg_type = incoming.message_type + + if msg_type == "text": + text = getattr(incoming, "text", None) + content = text.content.strip() if text else "" + + if self._file_cache: + cache_key = incoming.conversation_id if is_group else incoming.sender_id + cached = self._file_cache.get(cache_key) + if cached: + content = f"{content}\n[图片: {cached.file_path}]" if content else f"[图片: {cached.file_path}]" + self._file_cache.clear(cache_key) + + return content, "TEXT" + + elif msg_type == "audio": + extensions = getattr(incoming, "extensions", {}) + recognition = extensions.get("content", {}).get("recognition", "") + return recognition, "TEXT" + + elif msg_type == "richText": + text_list = incoming.get_text_list() if hasattr(incoming, "get_text_list") else [] + image_list = incoming.get_image_list() if hasattr(incoming, "get_image_list") else [] + text_content = "".join(text_list).strip() + image_paths = [] + for download_code in image_list: + path = await self._download_image(incoming, download_code) + if path: + image_paths.append(path) + if image_paths: + image_lines = "\n".join(f"[图片: {p}]" for p in image_paths) + text_content = f"{text_content}\n{image_lines}" if text_content else image_lines + return text_content or None, "TEXT" + + elif msg_type == "video": + download_code = getattr(incoming, "download_code", None) or getattr(incoming, "video_download_code", None) + if download_code: + video_path = await self._download_image(incoming, download_code) + if video_path: + return f"[视频: {video_path}]", "VIDEO" + return "[视频消息]", "VIDEO" + + elif msg_type == "file": + file_name = getattr(incoming, "file_name", "") or "未知文件" + download_code = getattr(incoming, "download_code", None) or getattr(incoming, "file_download_code", None) + if download_code: + file_path = await self._download_image(incoming, download_code) + if file_path: + return f"[文件: {file_name} -> {file_path}]", "FILE" + return f"[文件: {file_name}]", "FILE" + + return None, None + + async def _download_image(self, incoming, download_code: str) -> str | None: + if not self._gateway or not self._gateway.token_manager: + return None + + http = self._gateway.token_manager._http + if http is None: + async with __import__("httpx").AsyncClient(timeout=30.0) as http: + return await self._do_download(http, download_code) + return await self._do_download(http, download_code) + + def _extract_mentioned_ids(self, incoming) -> list[str]: + at_users = getattr(incoming, "at_users", None) + if at_users: + return [u.get("dingtalkId", "") for u in at_users if u.get("dingtalkId")] + raw = getattr(incoming, "raw_payload", None) + if raw and isinstance(raw, dict): + at_users_raw = raw.get("atUsers", []) + return [u.get("dingtalkId", "") for u in at_users_raw if u.get("dingtalkId")] + return [] + + def _build_unified_message(self, incoming, content: str, msg_type: str, is_group: bool) -> dict: + if is_group: + from_user_id = incoming.conversation_id + other_user_id = incoming.conversation_id + actual_user_id = incoming.sender_id + else: + from_user_id = incoming.sender_id + other_user_id = incoming.sender_id + actual_user_id = incoming.sender_id + + return { + "msg_id": incoming.message_id, + "channel_type": "dingtalk", + "account_id": "default", + "content": content, + "message_type": msg_type, + "sender": { + "id": from_user_id, + "display_name": getattr(incoming, "sender_nick", None) or from_user_id, + "kind": "GROUP" if is_group else "DIRECT", + }, + "group": { + "id": incoming.conversation_id, + "type": "group" if is_group else "direct", + } + if is_group + else None, + "timestamp": datetime.fromtimestamp(incoming.create_at / 1000, tz=UTC), + "mentioned_ids": self._extract_mentioned_ids(incoming), + "reply_to_id": None, + "thread_id": None, + "session_key": incoming.conversation_id if is_group else incoming.sender_id, + "raw_payload": incoming, + "metadata": { + "conversation_id": incoming.conversation_id, + "conversation_type": "2" if is_group else "1", + "sender_staff_id": getattr(incoming, "sender_staff_id", ""), + "robot_code": self._robot_code_cache, + "is_group": is_group, + "actual_user_id": actual_user_id, + "other_user_id": other_user_id, + }, + } + + async def _handle_card_callback(self, callback) -> tuple[str, str]: + data = callback.data if hasattr(callback, "data") else {} + if isinstance(data, str): + try: + data = json.loads(data) + except Exception: + data = {} + + card_instance_id = data.get("cardInstanceId", "") + out_track_id = data.get("outTrackId", "") + button_key = data.get("buttonKey", "") + button_value = data.get("buttonValue", "") + user_id = data.get("userId", "") + conversation_id = data.get("conversationId", "") + + logger.info( + "DingTalk card callback: card=%s button=%s value=%s user=%s", + card_instance_id, + button_key, + button_value, + user_id, + ) + + unified = { + "msg_id": out_track_id or card_instance_id, + "channel_type": "dingtalk", + "account_id": "default", + "content": f"[卡片回调] buttonKey={button_key} buttonValue={button_value}", + "message_type": "CARD_CALLBACK", + "sender": {"id": user_id, "display_name": user_id, "kind": "DIRECT"}, + "group": {"id": conversation_id, "type": "group"} if conversation_id else None, + "timestamp": datetime.now(UTC), + "mentioned_ids": [], + "reply_to_id": None, + "thread_id": None, + "session_key": conversation_id or user_id, + "raw_payload": data, + "metadata": { + "card_instance_id": card_instance_id, + "out_track_id": out_track_id, + "button_key": button_key, + "button_value": button_value, + "conversation_id": conversation_id, + "user_id": user_id, + "is_card_callback": True, + }, + } + + if self._dispatch_callback: + asyncio.create_task( + self._dispatch_callback(unified), + name=f"dingtalk-card-{out_track_id[:8] if out_track_id else 'unknown'}", + ) + + return AckMessage.STATUS_OK, "OK" + + async def _handle_command(self, text: str, incoming, is_group: bool) -> CommandResult: + if not text.startswith("/"): + return CommandResult() + + parts = text.strip().split(maxsplit=1) + cmd = parts[0].lower() + + if cmd == "/help": + help_lines = ["**可用命令**:"] + [f"- `{k}`: {v}" for k, v in BUILTIN_COMMANDS.items()] + response = "\n".join(help_lines) + elif cmd == "/ping": + response = "pong 🏓" + elif cmd == "/status": + response = "运行正常 ✅" + else: + response = f"未知命令: `{cmd}`，输入 `/help` 查看帮助" + + if self._outbound: + try: + await self._outbound.send_text( + incoming.sender_id, + response, + is_group=is_group, + conversation_id=incoming.conversation_id if is_group else "", + sender_staff_id=getattr(incoming, "sender_staff_id", ""), + ) + except Exception: + logger.exception("DingTalk command reply failed") + + return CommandResult(handled=True, response=response) + + async def _do_download(self, http, download_code: str) -> str | None: + token = await self._gateway.token_manager.get_access_token() + if not token: + return None + + from yuxi.channel.extensions.dingtalk.media import download_image + + robot_code = self._robot_code_cache or self._gateway.robot_code or "" + return await download_image(http, token, download_code, robot_code) diff --git a/backend/package/yuxi/channel/extensions/dingtalk/outbound.py b/backend/package/yuxi/channel/extensions/dingtalk/outbound.py new file mode 100644 index 00000000..44a8ce10 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/dingtalk/outbound.py @@ -0,0 +1,666 @@ +from __future__ import annotations + +import asyncio +import json +import logging + +import httpx + +from yuxi.channel.extensions.dingtalk.media import upload_media + +logger = logging.getLogger(__name__) + +SINGLE_API = "https://api.dingtalk.com/v1.0/robot/oToMessages/batchSend" +GROUP_API = "https://api.dingtalk.com/v1.0/robot/groupMessages/send" +RECALL_OTO_URL = "https://api.dingtalk.com/v1.0/robot/otoMessages/batchRecall" +RECALL_GROUP_URL = "https://api.dingtalk.com/v1.0/robot/groupMessages/recall" +QUERY_OTO_URL = "https://api.dingtalk.com/v1.0/robot/otoMessages/query" +QUERY_GROUP_URL = "https://api.dingtalk.com/v1.0/robot/groupMessages/query" +PLUGIN_SET_URL = "https://api.dingtalk.com/v1.0/robot/plugin/set" +CREATE_GROUP_URL = "https://api.dingtalk.com/v1.0/im/chat/scenegroup/create" +GROUP_MEMBERS_ADD_URL = "https://api.dingtalk.com/v1.0/im/sceneGroup/members/batchAdd" +GROUP_MEMBERS_REMOVE_URL = "https://api.dingtalk.com/v1.0/im/sceneGroup/members/batchRemove" +SEND_DING_URL = "https://api.dingtalk.com/v1.0/robot/ding/send" +BOT_LIST_IN_GROUP_URL = "https://api.dingtalk.com/v1.0/im/sceneGroup/robots/query" +BOT_GROUP_INFO_URL = "https://api.dingtalk.com/v1.0/robot/groupInfos/query" +UPDATE_ROBOT_URL = "https://api.dingtalk.com/v1.0/robot/info/update" +EXECUTE_AI_SKILL_URL = "https://api.dingtalk.com/v1.0/robot/aiSkills/execute" + + +class DingTalkOutbound: + def __init__(self, gateway): + self._gateway = gateway + self._http: httpx.AsyncClient | None = None + self._card_manager = None + + def set_http(self, http: httpx.AsyncClient) -> None: + self._http = http + + def set_card_manager(self, card_manager) -> None: + self._card_manager = card_manager + + async def _ensure_http(self) -> httpx.AsyncClient: + if self._http is None: + self._http = httpx.AsyncClient(timeout=30.0) + return self._http + + async def _get_token(self) -> str | None: + return await self._gateway.token_manager.get_access_token() + + async def send_text(self, target_id: str, content: str, **kwargs) -> dict: + reply_to_id = kwargs.pop("reply_to_id", None) + msg_content = content + if reply_to_id: + msg_content = f"> 回复消息\n\n{content}" + + return await self._send_with_msg_key( + target_id, + msg_content, + "sampleText", + json.dumps({"content": msg_content}), + **kwargs, + ) + + async def send_markdown(self, target_id: str, title: str, text: str, **kwargs) -> dict: + reply_to_id = kwargs.pop("reply_to_id", None) + msg_text = text + if reply_to_id: + msg_text = f"> 回复消息\n\n{text}" + + return await self._send_with_msg_key( + target_id, + msg_text, + "sampleMarkdown", + json.dumps({"title": title, "text": msg_text}), + **kwargs, + ) + + async def send_link( + self, + target_id: str, + title: str, + text: str, + message_url: str, + pic_url: str = "", + **kwargs, + ) -> dict: + msg_param = json.dumps( + { + "title": title, + "text": text, + "messageUrl": message_url, + "picUrl": pic_url, + } + ) + return await self._send_with_msg_key( + target_id, + text, + "sampleLink", + msg_param, + **kwargs, + ) + + async def send_image(self, target_id: str, image_path: str, **kwargs) -> dict: + http = await self._ensure_http() + token = await self._get_token() + if not token: + return {"success": False, "error": "no token"} + + media_id = await upload_media(http, token, image_path, "image") + if not media_id: + return {"success": False, "error": "upload failed"} + + text_content = kwargs.pop("text_content", None) + if text_content: + await self.send_text(target_id, text_content, **kwargs) + await asyncio.sleep(0.3) + + return await self._send_with_msg_key( + target_id, + image_path, + "sampleImageMsg", + json.dumps({"photoURL": media_id}), + **kwargs, + ) + + async def send_file(self, target_id: str, file_path: str, **kwargs) -> dict: + http = await self._ensure_http() + token = await self._get_token() + if not token: + return {"success": False, "error": "no token"} + + ext = file_path.rsplit(".", 1)[-1].lower() if "." in file_path else "" + media_id = await upload_media(http, token, file_path, "file") + if not media_id: + return {"success": False, "error": "upload failed"} + + return await self._send_with_msg_key( + target_id, + file_path, + "sampleFile", + json.dumps( + { + "mediaId": media_id, + "fileName": file_path.rsplit("/", 1)[-1].rsplit("\\", 1)[-1], + "fileType": ext, + } + ), + **kwargs, + ) + + async def send_video(self, target_id: str, file_path: str, **kwargs) -> dict: + http = await self._ensure_http() + token = await self._get_token() + if not token: + return {"success": False, "error": "no token"} + + ext = file_path.rsplit(".", 1)[-1].lower() if "." in file_path else "" + media_id = await upload_media(http, token, file_path, "video") + if not media_id: + return {"success": False, "error": "upload failed"} + + return await self._send_with_msg_key( + target_id, + file_path, + "sampleVideo", + json.dumps( + { + "duration": "30", + "videoMediaId": media_id, + "videoType": ext, + "height": "400", + "width": "600", + } + ), + **kwargs, + ) + + async def send_action_card(self, target_id: str, card: dict, **kwargs) -> dict: + text = card.get("text", "") + title = card.get("title", "通知") + buttons = card.get("btns", []) + single_title = card.get("singleTitle", "") + single_url = card.get("singleURL", "") + + btn_count = len(buttons) + + if btn_count == 0 and single_title and single_url: + msg_key = "sampleActionCard" + msg_param = json.dumps( + { + "title": title, + "text": text, + "singleTitle": single_title, + "singleURL": single_url, + } + ) + elif 1 <= btn_count <= 2: + msg_key = "sampleActionCard2" + msg_param = json.dumps( + { + "title": title, + "text": text, + "btnOrientation": card.get("btnOrientation", "1"), + "actionButtons": buttons, + } + ) + elif btn_count == 3: + msg_key = "sampleActionCard3" + msg_param = json.dumps( + { + "title": title, + "text": text, + "btnOrientation": card.get("btnOrientation", "1"), + "actionButtons": buttons, + } + ) + elif btn_count == 4: + msg_key = "sampleActionCard4" + msg_param = json.dumps( + { + "title": title, + "text": text, + "btnOrientation": card.get("btnOrientation", "1"), + "actionButtons": buttons, + } + ) + else: + return {"success": False, "error": f"unsupported button count: {btn_count}"} + + return await self._send_with_msg_key( + target_id, + text, + msg_key, + msg_param, + **kwargs, + ) + + async def send_media(self, target_id: str, media_url: str, media_type: str, **kwargs) -> dict: + if media_type.startswith("image"): + return await self.send_image(target_id, media_url, **kwargs) + elif media_type.startswith("video"): + return await self.send_video(target_id, media_url, **kwargs) + elif media_type.startswith("file"): + return await self.send_file(target_id, media_url, **kwargs) + return {"success": False, "error": f"unsupported media type: {media_type}"} + + async def recall_message( + self, + target_id: str, + process_query_keys: list[str], + *, + is_group: bool = False, + ) -> dict: + http = await self._ensure_http() + token = await self._get_token() + if not token: + return {"success": False, "error": "no token"} + + robot_code = self._gateway.robot_code + + if is_group: + url = RECALL_GROUP_URL + body = { + "robotCode": robot_code, + "openConversationId": target_id, + "processQueryKeys": process_query_keys, + } + else: + url = RECALL_OTO_URL + body = { + "robotCode": robot_code, + "processQueryKeys": process_query_keys, + } + + try: + resp = await http.post( + url, + headers={ + "x-acs-dingtalk-access-token": token, + "Content-Type": "application/json", + }, + json=body, + timeout=10.0, + ) + data = resp.json() + return {"success": True, "response": data} + except Exception: + logger.exception("DingTalk recall failed") + return {"success": False, "error": "recall failed"} + + async def query_message_status( + self, + process_query_key: str, + *, + is_group: bool = False, + open_conversation_id: str = "", + ) -> dict: + http = await self._ensure_http() + token = await self._get_token() + if not token: + return {"success": False, "error": "no token"} + + robot_code = self._gateway.robot_code + + if is_group: + url = QUERY_GROUP_URL + body = { + "robotCode": robot_code, + "processQueryKey": process_query_key, + "openConversationId": open_conversation_id, + } + else: + url = QUERY_OTO_URL + body = { + "robotCode": robot_code, + "processQueryKey": process_query_key, + } + + try: + resp = await http.post( + url, + headers={ + "x-acs-dingtalk-access-token": token, + "Content-Type": "application/json", + }, + json=body, + timeout=10.0, + ) + data = resp.json() + return {"success": True, "response": data} + except Exception: + logger.exception("DingTalk query message status failed") + return {"success": False, "error": "query failed"} + + async def set_robot_plugin(self, plugin_config: dict) -> dict: + http = await self._ensure_http() + token = await self._get_token() + if not token: + return {"success": False, "error": "no token"} + + robot_code = self._gateway.robot_code + body = {"robotCode": robot_code, **plugin_config} + + try: + resp = await http.post( + PLUGIN_SET_URL, + headers={ + "x-acs-dingtalk-access-token": token, + "Content-Type": "application/json", + }, + json=body, + timeout=10.0, + ) + data = resp.json() + return {"success": True, "response": data} + except Exception: + logger.exception("DingTalk set robot plugin failed") + return {"success": False, "error": "plugin set failed"} + + async def create_group( + self, + title: str, + owner_user_id: str, + *, + user_ids: list[str] | None = None, + ) -> dict: + http = await self._ensure_http() + token = await self._get_token() + if not token: + return {"success": False, "error": "no token"} + + body = { + "title": title, + "ownerUserId": owner_user_id, + } + if user_ids: + body["userIds"] = user_ids + + try: + resp = await http.post( + CREATE_GROUP_URL, + headers={ + "x-acs-dingtalk-access-token": token, + "Content-Type": "application/json", + }, + json=body, + timeout=10.0, + ) + data = resp.json() + return {"success": True, "response": data} + except Exception: + logger.exception("DingTalk create group failed") + return {"success": False, "error": "create failed"} + + async def add_group_members(self, open_conversation_id: str, user_ids: list[str]) -> dict: + http = await self._ensure_http() + token = await self._get_token() + if not token: + return {"success": False, "error": "no token"} + + try: + resp = await http.post( + GROUP_MEMBERS_ADD_URL, + headers={ + "x-acs-dingtalk-access-token": token, + "Content-Type": "application/json", + }, + json={ + "openConversationId": open_conversation_id, + "userIds": user_ids, + }, + timeout=10.0, + ) + data = resp.json() + return {"success": True, "response": data} + except Exception: + logger.exception("DingTalk add group members failed") + return {"success": False, "error": "add members failed"} + + async def remove_group_members(self, open_conversation_id: str, user_ids: list[str]) -> dict: + http = await self._ensure_http() + token = await self._get_token() + if not token: + return {"success": False, "error": "no token"} + + try: + resp = await http.post( + GROUP_MEMBERS_REMOVE_URL, + headers={ + "x-acs-dingtalk-access-token": token, + "Content-Type": "application/json", + }, + json={ + "openConversationId": open_conversation_id, + "userIds": user_ids, + }, + timeout=10.0, + ) + data = resp.json() + return {"success": True, "response": data} + except Exception: + logger.exception("DingTalk remove group members failed") + return {"success": False, "error": "remove members failed"} + + async def send_ding( + self, + user_ids: list[str], + content: str, + *, + ding_type: str = "APP", + ) -> dict: + http = await self._ensure_http() + token = await self._get_token() + if not token: + return {"success": False, "error": "no token"} + + robot_code = self._gateway.robot_code + body = { + "robotCode": robot_code, + "userIds": user_ids, + "content": content, + "type": ding_type, + } + + try: + resp = await http.post( + SEND_DING_URL, + headers={ + "x-acs-dingtalk-access-token": token, + "Content-Type": "application/json", + }, + json=body, + timeout=10.0, + ) + data = resp.json() + return {"success": True, "response": data} + except Exception: + logger.exception("DingTalk send DING failed") + return {"success": False, "error": "send failed"} + + async def get_bot_list_in_group(self, open_conversation_id: str) -> dict: + http = await self._ensure_http() + token = await self._get_token() + if not token: + return {"success": False, "error": "no token"} + + try: + resp = await http.post( + BOT_LIST_IN_GROUP_URL, + headers={ + "x-acs-dingtalk-access-token": token, + "Content-Type": "application/json", + }, + json={"openConversationId": open_conversation_id}, + timeout=10.0, + ) + data = resp.json() + return {"success": True, "response": data} + except Exception: + logger.exception("DingTalk get bot list failed") + return {"success": False, "error": "query failed"} + + async def query_bot_group_info(self, open_conversation_id: str) -> dict: + http = await self._ensure_http() + token = await self._get_token() + if not token: + return {"success": False, "error": "no token"} + + robot_code = self._gateway.robot_code + try: + resp = await http.post( + BOT_GROUP_INFO_URL, + headers={ + "x-acs-dingtalk-access-token": token, + "Content-Type": "application/json", + }, + json={ + "robotCode": robot_code, + "openConversationId": open_conversation_id, + }, + timeout=10.0, + ) + data = resp.json() + return {"success": True, "response": data} + except Exception: + logger.exception("DingTalk query bot group info failed") + return {"success": False, "error": "query failed"} + + async def update_robot_info(self, robot_info: dict) -> dict: + http = await self._ensure_http() + token = await self._get_token() + if not token: + return {"success": False, "error": "no token"} + + robot_code = self._gateway.robot_code + body = {"robotCode": robot_code, **robot_info} + + try: + resp = await http.post( + UPDATE_ROBOT_URL, + headers={ + "x-acs-dingtalk-access-token": token, + "Content-Type": "application/json", + }, + json=body, + timeout=10.0, + ) + data = resp.json() + return {"success": True, "response": data} + except Exception: + logger.exception("DingTalk update robot info failed") + return {"success": False, "error": "update failed"} + + async def execute_ai_skill( + self, + skill_id: str, + prompt: str, + *, + user_id: str = "", + conversation_id: str = "", + ) -> dict: + http = await self._ensure_http() + token = await self._get_token() + if not token: + return {"success": False, "error": "no token"} + + robot_code = self._gateway.robot_code + body = { + "robotCode": robot_code, + "skillId": skill_id, + "prompt": prompt, + } + if user_id: + body["userId"] = user_id + if conversation_id: + body["openConversationId"] = conversation_id + + try: + resp = await http.post( + EXECUTE_AI_SKILL_URL, + headers={ + "x-acs-dingtalk-access-token": token, + "Content-Type": "application/json", + }, + json=body, + timeout=30.0, + ) + data = resp.json() + return {"success": True, "response": data} + except Exception: + logger.exception("DingTalk execute AI skill failed") + return {"success": False, "error": "execute failed"} + + async def streaming_send( + self, + target_id: str, + content: str, + *, + is_group: bool = False, + conversation_id: str = "", + status: str = "PROCESSING", + ) -> dict: + if not self._card_manager: + return {"success": False, "error": "card manager not initialized"} + + success = await self._card_manager.streaming_update(content, status) + return {"success": success} + + async def _send_with_msg_key( + self, + target_id: str, + content: str, + msg_key: str, + msg_param: str, + **kwargs, + ) -> dict: + http = await self._ensure_http() + token = await self._get_token() + if not token: + return {"success": False, "error": "no token"} + + is_group = kwargs.get("is_group", False) + conversation_id = kwargs.get("conversation_id", "") + sender_staff_id = kwargs.get("sender_staff_id", "") + + robot_code = self._gateway.robot_code + + if is_group: + url = GROUP_API + body = { + "robotCode": robot_code, + "openConversationId": conversation_id or target_id, + "msgKey": msg_key, + "msgParam": msg_param, + } + else: + url = SINGLE_API + user_ids = kwargs.get("user_ids") + if not user_ids and sender_staff_id: + user_ids = [sender_staff_id] + if not user_ids: + user_ids = [target_id] + + body = { + "robotCode": robot_code, + "userIds": user_ids, + "msgKey": msg_key, + "msgParam": msg_param, + } + + try: + resp = await http.post( + url, + headers={ + "x-acs-dingtalk-access-token": token, + "Content-Type": "application/json", + }, + json=body, + timeout=30.0, + ) + data = resp.json() + logger.debug("DingTalk send result: %s", data) + return {"success": True, "response": data} + except Exception: + logger.exception("DingTalk send failed") + return {"success": False, "error": "send failed"} diff --git a/backend/package/yuxi/channel/extensions/dingtalk/pairing.py b/backend/package/yuxi/channel/extensions/dingtalk/pairing.py new file mode 100644 index 00000000..3825c38d --- /dev/null +++ b/backend/package/yuxi/channel/extensions/dingtalk/pairing.py @@ -0,0 +1,37 @@ +from __future__ import annotations + +import logging +import secrets +import time + +logger = logging.getLogger(__name__) + +CODE_TTL = 300 + + +class DingTalkPairing: + def __init__(self): + self._codes: dict[str, tuple[str, float]] = {} + + async def generate_code(self, peer_id: str) -> str: + code = f"{secrets.randbelow(1_000_000):06d}" + self._codes[peer_id] = (code, time.monotonic()) + self._evict_expired() + return code + + async def verify_code(self, peer_id: str, code: str) -> bool: + self._evict_expired() + entry = self._codes.get(peer_id) + if entry is None: + return False + stored_code, _ts = entry + if stored_code == code: + del self._codes[peer_id] + return True + return False + + def _evict_expired(self) -> None: + now = time.monotonic() + expired = [pid for pid, (_code, ts) in self._codes.items() if now - ts > CODE_TTL] + for pid in expired: + del self._codes[pid] diff --git a/backend/package/yuxi/channel/extensions/dingtalk/plugin.json b/backend/package/yuxi/channel/extensions/dingtalk/plugin.json new file mode 100644 index 00000000..0e45d042 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/dingtalk/plugin.json @@ -0,0 +1,32 @@ +{ + "id": "dingtalk", + "name": "钉钉", + "version": "1.1.0", + "label": "DingTalk", + "aliases": ["ding", "dd"], + "description": "钉钉渠道插件 — 基于 Stream 模式 WebSocket 长连接，支持单聊/群聊/卡片流式/审批通知/消息撤回/群管理", + "author": "ForcePilot Team", + "order": 20, + "dependencies": ["dingtalk-stream", "httpx", "websockets"], + "capabilities": { + "chat_types": ["direct", "group"], + "message_types": ["text", "image", "file", "audio", "video", "markdown"], + "reactions": false, + "typing_indicator": false, + "threads": false, + "edit": false, + "unsend": true, + "reply": true, + "media": true, + "effects": false, + "native_commands": true, + "polls": false, + "group_management": true, + "streaming": true, + "streaming_mode": "block", + "block_streaming": true, + "adaptive_cards": true + }, + "enabled": true, + "python_requires": ">=3.12" +} diff --git a/backend/package/yuxi/channel/extensions/dingtalk/security.py b/backend/package/yuxi/channel/extensions/dingtalk/security.py new file mode 100644 index 00000000..332ddfae --- /dev/null +++ b/backend/package/yuxi/channel/extensions/dingtalk/security.py @@ -0,0 +1,64 @@ +from __future__ import annotations + +import logging + +logger = logging.getLogger(__name__) + + +class DingTalkSecurity: + def __init__(self, dm_policy: str = "open", group_policy: str = "open"): + self._dm_policy = dm_policy + self._group_policy = group_policy + self._allowlist: set[str] = set() + self._group_allowlist: set[str] = set() + + @property + def dm_policy(self) -> str: + return self._dm_policy + + def resolve_dm_policy(self) -> str: + return self._dm_policy + + def check_allowlist(self, user_id: str) -> bool: + if self._dm_policy == "open": + return True + return user_id in self._allowlist + + def check_dm_access(self, sender_id: str) -> tuple[bool, str]: + if self._dm_policy == "disabled": + return False, "DM is disabled" + if self._dm_policy == "open": + return True, "open" + if self._dm_policy == "allowlist": + if sender_id in self._allowlist: + return True, "allowlist" + return False, f"sender {sender_id} not in allowlist" + if self._dm_policy == "pairing": + return True, "pairing" + return False, f"unknown dm_policy: {self._dm_policy}" + + def check_group_access(self, conversation_id: str) -> tuple[bool, str]: + if self._group_policy == "open": + return True, "open" + if self._group_policy == "disabled": + return False, "group access disabled" + if self._group_policy == "allowlist": + if conversation_id in self._group_allowlist: + return True, "allowlist" + return False, f"group {conversation_id} not in allowlist" + return False, f"unknown group_policy: {self._group_policy}" + + def add_to_allowlist(self, user_id: str) -> None: + self._allowlist.add(user_id) + + def remove_from_allowlist(self, user_id: str) -> None: + self._allowlist.discard(user_id) + + def load_allowlist(self, allow_from: list[str]) -> None: + self._allowlist = set(allow_from) + + def add_group_to_allowlist(self, conversation_id: str) -> None: + self._group_allowlist.add(conversation_id) + + def remove_group_from_allowlist(self, conversation_id: str) -> None: + self._group_allowlist.discard(conversation_id) diff --git a/backend/package/yuxi/channel/extensions/dingtalk/streaming.py b/backend/package/yuxi/channel/extensions/dingtalk/streaming.py new file mode 100644 index 00000000..b79154f1 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/dingtalk/streaming.py @@ -0,0 +1,15 @@ +from __future__ import annotations + + +class DingTalkStreamingAdapter: + @property + def streaming_mode(self) -> str: + return "block" + + @property + def block_streaming_config(self) -> dict: + return { + "chunk_size": 1024, + "max_chunks": 20, + "flush_interval_ms": 2000, + } diff --git a/backend/package/yuxi/channel/extensions/dingtalk/types.py b/backend/package/yuxi/channel/extensions/dingtalk/types.py new file mode 100644 index 00000000..319b9be2 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/dingtalk/types.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +from dataclasses import dataclass, field + + +@dataclass +class DingTalkAccount: + account_id: str = "default" + app_key: str = "" + app_secret: str = "" + robot_code: str = "" + mode: str = "stream" + streaming: str = "block" + dingtalk_card_enabled: bool = False + card_template_id: str = "" + dm_policy: str = "open" + group_policy: str = "open" + group_shared_session: bool = True + expires_in_seconds: int = 3600 + hot_reload: bool = False + chat_time_module: bool = False + chat_start_time: str = "00:00" + chat_stop_time: str = "24:00" + agent: bool = False + agent_workspace: str = "~/cow" + render_mode: str = "auto" + subscribe_topics: list[str] = field(default_factory=list) + enabled: bool = True + + def is_configured(self) -> bool: + return bool(self.app_key and self.app_secret) + + def is_enabled(self) -> bool: + return self.enabled