kris/ForcePilot
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(dingtalk): 新增钉钉渠道插件完整实现

Browse Source 
该提交实现了完整的钉钉聊天渠道插件，包含：
1. 基础配置、账号管理与凭证校验
2. WebSocket长连接网关与消息去重
3. 消息接收/解析/分发与安全校验
4. 媒体文件上传下载与缓存
5. 互动卡片流式更新与回调处理
6. 群管理、命令支持与诊断工具
7. 完整的插件元数据与依赖声明
This commit is contained in:
Kris 2026-05-21 10:44:03 +08:00
parent f943c31ce4
commit 472afc4d1e
15 changed files with 2623 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

392
backend/package/yuxi/channel/extensions/dingtalk/__init__.py Normal file
View File

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

173
backend/package/yuxi/channel/extensions/dingtalk/card.py Normal file
View File

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

187
backend/package/yuxi/channel/extensions/dingtalk/config.py Normal file
View File

@ -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", {})

50
backend/package/yuxi/channel/extensions/dingtalk/dedupe.py Normal file
View File

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

215
backend/package/yuxi/channel/extensions/dingtalk/doctor.py Normal file
View File

@ -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 []

40
backend/package/yuxi/channel/extensions/dingtalk/file_cache.py Normal file
View File

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

207
backend/package/yuxi/channel/extensions/dingtalk/gateway.py Normal file
View File

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

114
backend/package/yuxi/channel/extensions/dingtalk/media.py Normal file
View File

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

397
backend/package/yuxi/channel/extensions/dingtalk/monitor.py Normal file
View File

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

666
backend/package/yuxi/channel/extensions/dingtalk/outbound.py Normal file
View File

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

37
backend/package/yuxi/channel/extensions/dingtalk/pairing.py Normal file
View File

@ -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]

32
backend/package/yuxi/channel/extensions/dingtalk/plugin.json Normal file
View File

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

64
backend/package/yuxi/channel/extensions/dingtalk/security.py Normal file
View File

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

15
backend/package/yuxi/channel/extensions/dingtalk/streaming.py Normal file
View File

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

34
backend/package/yuxi/channel/extensions/dingtalk/types.py Normal file
View File

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