Kris
ce0a7e7232
本次提交对钉钉插件进行了多维度改进： 1. 代码整洁优化：移除多余空行、调整导入顺序 2. 签名逻辑修正：修复hmac签名参数顺序，对齐官方文档 3. 消息撤回增强：使用枚举替代硬编码字符串，添加缓存未命中日志 4. 配置与能力扩展：新增互动卡片模板ID、Bot命令配置项，重构机器人编码获取逻辑 5. 流式会话优化：移除硬编码默认卡片ID，动态读取账户配置 6. 健康检查增强：新增流式端点连通性校验，替代原有简化校验逻辑 7. 附件下载升级：适配钉钉图片下载API，新增图片附件下载能力 8. 封装改进：将私有配置读取方法封装为公开接口，避免破坏封装性
741 lines
25 KiB
Python
741 lines
25 KiB
Python
"""钉钉 API HTTP 客户端。
|
||
|
||
通过 httpx.AsyncClient 调用钉钉开放平台 API，封装 access_token 管理、
|
||
通用请求与 IM / 通讯录 / 互动卡片 / Stream 接入 API。HTTP 异常与钉钉
|
||
业务错误码通过 error_translator 转换为契约层异常，不向外抛原生异常
|
||
（INV-7）。token 通过 CachePort 缓存（key 为
|
||
``dingtalk:token:{client_id}``，TTL=expires_in-300s，对齐设计方案 §5.4），
|
||
客户端本身不持有 mutable 状态。
|
||
|
||
连接池由 lifecycle 统一管理（onInit 创建 / onUnload 关闭），通过
|
||
``attach_http_client`` 注入；未注入时降级为每请求独立 client（兼容
|
||
测试与未启用 lifecycle 的场景）。
|
||
|
||
依赖方向：仅 import yuxi.channels.contract.* 与标准库 / httpx，不污染
|
||
框架层。error_translator 为本插件内部模块。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import time
|
||
from typing import Any
|
||
|
||
import httpx
|
||
|
||
from yuxi.channels.contract.dtos.config import ConfigScope
|
||
from yuxi.channels.contract.errors import DependencyError, ValidationError
|
||
from yuxi.channels.contract.ports.driven.cache_port import CachePort
|
||
from yuxi.channels.contract.ports.driven.config_port import ConfigPort
|
||
from yuxi.channels.contract.ports.driven.logger_port import LoggerPort
|
||
|
||
from . import error_translator
|
||
from ._constants import (
|
||
CHANNEL_TARGET,
|
||
DINGTALK_API_DEP,
|
||
TOKEN_CACHE_KEY_PREFIX,
|
||
TOKEN_REFRESH_LEAD_TIME_SEC,
|
||
)
|
||
|
||
|
||
class DingTalkClient:
|
||
"""钉钉 API HTTP 客户端。
|
||
|
||
通过 DI 接收 ConfigPort / CachePort / LoggerPort，不访问全局 settings
|
||
或 logger。token 通过 CachePort 缓存，客户端不持有 mutable 状态。所有
|
||
方法为 async，HTTP 异常通过 error_translator.wrap_http_exception 转换，
|
||
钉钉业务错误码非 0 通过 error_translator.translate_dingtalk_error 转换。
|
||
|
||
连接池由 lifecycle 统一管理（onInit 创建 / onUnload 关闭），通过
|
||
``attach_http_client`` 注入；未注入时降级为每请求独立 client（兼容
|
||
测试与未启用 lifecycle 的场景）。
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
config_port: ConfigPort,
|
||
cache_port: CachePort,
|
||
logger_port: LoggerPort,
|
||
) -> None:
|
||
self._config = config_port
|
||
self._cache = cache_port
|
||
self._logger = logger_port
|
||
# 由 lifecycle 注入的共享连接池；为 None 时降级为每请求独立 client
|
||
self._http_client: httpx.AsyncClient | None = None
|
||
|
||
# ------------------------------------------------------------------
|
||
# 连接池管理
|
||
# ------------------------------------------------------------------
|
||
|
||
def attach_http_client(self, client: httpx.AsyncClient) -> None:
|
||
"""注入 lifecycle 管理的共享 httpx 连接池。
|
||
|
||
由 ``DingTalkLifecycleHandler.onInit`` 调用，将创建好的连接池注入
|
||
客户端；``onUnload`` 关闭前由 lifecycle 置空以避免悬挂引用。
|
||
|
||
@consistency
|
||
- 幂等：重复调用以最后一次注入的 client 为准
|
||
"""
|
||
self._http_client = client
|
||
|
||
def detach_http_client(self) -> None:
|
||
"""解除共享连接池引用（由 lifecycle 在 onUnload 关闭前调用）。
|
||
|
||
@consistency
|
||
- 幂等：重复调用安全
|
||
"""
|
||
self._http_client = None
|
||
|
||
# ------------------------------------------------------------------
|
||
# 公开配置读取（供适配器调用，避免访问私有方法）
|
||
# ------------------------------------------------------------------
|
||
|
||
async def get_robot_code(self, account_id: str) -> str:
|
||
"""读取账户级 ``robot_code``。
|
||
|
||
供适配器（streaming / message_ops 等）调用，避免直接访问
|
||
``_get_account_config`` 私有方法。
|
||
|
||
@failure ``robot_code`` 为空时抛 ``ValidationError``。
|
||
"""
|
||
account_config = await self._get_account_config(account_id)
|
||
robot_code = account_config.get("robot_code", "") or ""
|
||
if not robot_code:
|
||
raise ValidationError(
|
||
field="robot_code",
|
||
message="robot_code not configured for account",
|
||
)
|
||
return robot_code
|
||
|
||
async def get_card_template_id(self, account_id: str) -> str:
|
||
"""读取账户级 ``card_template_id``（互动卡片模板 ID）。
|
||
|
||
供 streaming / rich_message 适配器调用。未配置时抛
|
||
``ValidationError``，不使用占位值掩盖配置缺失问题。
|
||
|
||
@failure ``card_template_id`` 为空时抛 ``ValidationError``。
|
||
"""
|
||
cv = await self._config.get(
|
||
"card_template_id",
|
||
scope=ConfigScope.ACCOUNT,
|
||
target=account_id,
|
||
)
|
||
template_id = cv.value or ""
|
||
if not template_id:
|
||
raise ValidationError(
|
||
field="card_template_id",
|
||
message="card_template_id not configured for account",
|
||
)
|
||
return template_id
|
||
|
||
# ------------------------------------------------------------------
|
||
# 内部辅助
|
||
# ------------------------------------------------------------------
|
||
|
||
async def _get_account_config(self, account_id: str) -> dict[str, Any]:
|
||
"""读取账户级配置（client_id / client_secret / robot_code）。"""
|
||
cv_client_id = await self._config.get(
|
||
"client_id",
|
||
scope=ConfigScope.ACCOUNT,
|
||
target=account_id,
|
||
)
|
||
cv_client_secret = await self._config.get(
|
||
"client_secret",
|
||
scope=ConfigScope.ACCOUNT,
|
||
target=account_id,
|
||
)
|
||
cv_robot_code = await self._config.get(
|
||
"robot_code",
|
||
scope=ConfigScope.ACCOUNT,
|
||
target=account_id,
|
||
)
|
||
return {
|
||
"client_id": cv_client_id.value,
|
||
"client_secret": cv_client_secret.value,
|
||
"robot_code": cv_robot_code.value,
|
||
}
|
||
|
||
async def _get_channel_config(self, key: str) -> Any:
|
||
"""读取渠道级配置（api_base_url / request_timeout_ms 等）。"""
|
||
cv = await self._config.get(
|
||
key,
|
||
scope=ConfigScope.CHANNEL,
|
||
target=CHANNEL_TARGET,
|
||
)
|
||
return cv.value
|
||
|
||
async def _get_request_timeout_seconds(self) -> float:
|
||
"""从配置读取请求超时（秒）。"""
|
||
timeout_ms = await self._get_channel_config("request_timeout_ms")
|
||
return float(timeout_ms) / 1000.0
|
||
|
||
def _token_cache_key(self, client_id: str) -> str:
|
||
return f"{TOKEN_CACHE_KEY_PREFIX}{client_id}"
|
||
|
||
async def _get_token_for_account(self, account_id: str) -> str:
|
||
"""获取账户的 access_token。"""
|
||
account_config = await self._get_account_config(account_id)
|
||
token, _ = await self.get_access_token(
|
||
account_config["client_id"],
|
||
account_config["client_secret"],
|
||
)
|
||
return token
|
||
|
||
async def _build_auth_header(self, account_id: str) -> dict[str, str]:
|
||
"""获取账户 token 并构造 x-acs-dingtalk-access-token 头。
|
||
|
||
钉钉 API 通过 ``x-acs-dingtalk-access-token`` 请求头携带
|
||
``access_token``，与飞书的 ``Authorization: Bearer xxx`` 不同。
|
||
"""
|
||
token = await self._get_token_for_account(account_id)
|
||
return {"x-acs-dingtalk-access-token": token}
|
||
|
||
async def _execute_http(
|
||
self,
|
||
method: str,
|
||
url: str,
|
||
headers: dict[str, str],
|
||
*,
|
||
json_body: dict | None = None,
|
||
params: dict | None = None,
|
||
data: dict | None = None,
|
||
files: dict | None = None,
|
||
) -> httpx.Response:
|
||
"""执行 HTTP 请求，HTTP 异常转换为契约异常。
|
||
|
||
不负责 token 注入与业务错误码校验，由调用方处理。优先复用 lifecycle
|
||
注入的共享连接池；未注入时降级为每请求独立 ``httpx.AsyncClient``
|
||
（兼容测试与未启用 lifecycle 的场景）。HTTP 异常经
|
||
``error_translator.wrap_http_exception`` 转换，并传入实际配置
|
||
的 ``timeout_ms`` 保证 ``OperationTimeoutError`` 携带准确超时值。
|
||
"""
|
||
timeout = await self._get_request_timeout_seconds()
|
||
try:
|
||
if self._http_client is not None:
|
||
resp = await self._http_client.request(
|
||
method,
|
||
url,
|
||
json=json_body,
|
||
params=params,
|
||
data=data,
|
||
files=files,
|
||
headers=headers,
|
||
timeout=timeout,
|
||
)
|
||
else:
|
||
# 连接池未注入时降级为每请求独立 client（兼容测试场景）
|
||
async with httpx.AsyncClient(timeout=timeout) as client:
|
||
resp = await client.request(
|
||
method,
|
||
url,
|
||
json=json_body,
|
||
params=params,
|
||
data=data,
|
||
files=files,
|
||
headers=headers,
|
||
)
|
||
resp.raise_for_status()
|
||
except httpx.HTTPError as exc:
|
||
await self._logger.exception(
|
||
"DingTalk HTTP request failed",
|
||
exc_info=exc,
|
||
method=method,
|
||
url=url,
|
||
)
|
||
raise error_translator.wrap_http_exception(
|
||
exc,
|
||
timeout_ms=int(timeout * 1000),
|
||
) from exc
|
||
return resp
|
||
|
||
@staticmethod
|
||
def _parse_json_response(resp: httpx.Response) -> dict:
|
||
"""解析 JSON 响应，校验钉钉业务错误码。
|
||
|
||
``errcode`` 非 0 时通过 error_translator.translate_dingtalk_error
|
||
转换为契约异常。``errcode`` 为 0 或缺失时返回完整响应字典。
|
||
部分钉钉端点（如 ``/v1.0/oauth2/accessToken``）将错误码放在
|
||
顶层而非 ``errcode`` 字段，由调用方在专用方法中处理。
|
||
"""
|
||
data = resp.json()
|
||
code = data.get("errcode", 0)
|
||
if code != 0:
|
||
raise error_translator.translate_dingtalk_error(
|
||
code,
|
||
data.get("errmsg", ""),
|
||
) from None
|
||
return data
|
||
|
||
async def _request_json(
|
||
self,
|
||
method: str,
|
||
path: str,
|
||
account_id: str,
|
||
*,
|
||
json_body: dict | None = None,
|
||
params: dict | None = None,
|
||
data: dict | None = None,
|
||
files: dict | None = None,
|
||
) -> dict:
|
||
"""发起带 access_token 的请求并返回校验后的 JSON。"""
|
||
headers = await self._build_auth_header(account_id)
|
||
api_base_url = await self._get_channel_config("api_base_url")
|
||
url = f"{api_base_url}{path}"
|
||
resp = await self._execute_http(
|
||
method,
|
||
url,
|
||
headers,
|
||
json_body=json_body,
|
||
params=params,
|
||
data=data,
|
||
files=files,
|
||
)
|
||
return self._parse_json_response(resp)
|
||
|
||
# ------------------------------------------------------------------
|
||
# Token 管理
|
||
# ------------------------------------------------------------------
|
||
|
||
async def get_access_token(
|
||
self,
|
||
client_id: str,
|
||
client_secret: str,
|
||
) -> tuple[str, int]:
|
||
"""获取 access_token，缓存命中返回剩余 TTL。
|
||
|
||
钉钉 access_token 有效期 7200 秒（2 小时），缓存存储绝对过期时间戳
|
||
``expires_at``，缓存命中时根据 ``expires_at - now`` 计算剩余 TTL，
|
||
避免长期持有 token 的调用方误判有效期。``expires_in <= 0`` 表示已
|
||
过期，调用方应触发重新获取。
|
||
|
||
钉钉 token 端点 ``/v1.0/oauth2/accessToken`` 响应不含 ``errcode``
|
||
字段（成功响应直接含 ``accessToken`` / ``expireIn``），失败时返回
|
||
HTTP 4xx 与 ``code`` / ``message`` 字段，由 ``raise_for_status``
|
||
触发的 ``HTTPStatusError`` 经 ``wrap_http_exception`` 转换。
|
||
|
||
@return (token, remaining_expires_in)
|
||
"""
|
||
cache_key = self._token_cache_key(client_id)
|
||
cached = await self._cache.get(cache_key)
|
||
if cached.is_some():
|
||
token, expires_at = cached.unwrap()
|
||
remaining = max(int(expires_at - time.time()), 0)
|
||
return token, remaining
|
||
|
||
api_base_url = await self._get_channel_config("api_base_url")
|
||
url = f"{api_base_url}/v1.0/oauth2/accessToken"
|
||
body = {"appKey": client_id, "appSecret": client_secret}
|
||
|
||
resp = await self._execute_http(
|
||
"POST",
|
||
url,
|
||
headers={},
|
||
json_body=body,
|
||
)
|
||
data = resp.json()
|
||
# 钉钉 token 端点失败响应含 code/message，需显式校验
|
||
if "code" in data and data.get("code") != 0:
|
||
raise error_translator.translate_dingtalk_error(
|
||
int(data.get("code", 0)),
|
||
data.get("message", ""),
|
||
) from None
|
||
token = data["accessToken"]
|
||
expires_in = int(data.get("expireIn", 0))
|
||
if expires_in <= 0:
|
||
raise DependencyError(
|
||
dep=DINGTALK_API_DEP,
|
||
cause=ValueError(f"invalid expireIn value: {expires_in}"),
|
||
)
|
||
expires_at = time.time() + expires_in
|
||
ttl = max(expires_in - TOKEN_REFRESH_LEAD_TIME_SEC, 1)
|
||
await self._cache.set(
|
||
cache_key,
|
||
(token, expires_at),
|
||
ttl_seconds=ttl,
|
||
)
|
||
await self._logger.debug(
|
||
"DingTalk access_token refreshed",
|
||
client_id=client_id,
|
||
expires_in=expires_in,
|
||
)
|
||
return token, expires_in
|
||
|
||
# ------------------------------------------------------------------
|
||
# 通用请求
|
||
# ------------------------------------------------------------------
|
||
|
||
async def request(
|
||
self,
|
||
method: str,
|
||
path: str,
|
||
account_id: str,
|
||
*,
|
||
json: dict | None = None,
|
||
params: dict | None = None,
|
||
) -> dict:
|
||
"""通用请求方法。
|
||
|
||
自动获取 access_token，注入 ``x-acs-dingtalk-access-token`` 头，
|
||
发起 HTTP 请求并校验业务错误码。
|
||
"""
|
||
return await self._request_json(
|
||
method,
|
||
path,
|
||
account_id,
|
||
json_body=json,
|
||
params=params,
|
||
)
|
||
|
||
# ------------------------------------------------------------------
|
||
# 机器人消息 API（设计方案 §4.1）
|
||
# ------------------------------------------------------------------
|
||
|
||
async def send_oto_message(
|
||
self,
|
||
account_id: str,
|
||
robot_code: str,
|
||
user_ids: list[str],
|
||
msg_key: str,
|
||
msg_param: dict,
|
||
) -> dict:
|
||
"""单聊消息发送（``batchSendUserMessageByRobot``）。
|
||
|
||
@return 钉钉响应字典，含 ``processQueryKey`` 用于后续撤回。
|
||
"""
|
||
return await self.request(
|
||
"POST",
|
||
"/v1.0/robot/oToMessages/batchSend",
|
||
account_id,
|
||
json={
|
||
"robotCode": robot_code,
|
||
"userIds": user_ids,
|
||
"msgKey": msg_key,
|
||
"msgParam": _to_json_str(msg_param),
|
||
},
|
||
)
|
||
|
||
async def send_group_message(
|
||
self,
|
||
account_id: str,
|
||
robot_code: str,
|
||
conversation_id: str,
|
||
msg_key: str,
|
||
msg_param: dict,
|
||
) -> dict:
|
||
"""群聊消息发送（``sendGroupMessageByRobot``）。
|
||
|
||
@return 钉钉响应字典，含 ``messageId`` / ``processQueryKey``。
|
||
"""
|
||
return await self.request(
|
||
"POST",
|
||
"/v1.0/robot/groupMessages/send",
|
||
account_id,
|
||
json={
|
||
"robotCode": robot_code,
|
||
"openConversationId": conversation_id,
|
||
"msgKey": msg_key,
|
||
"msgParam": _to_json_str(msg_param),
|
||
},
|
||
)
|
||
|
||
async def recall_oto_message(
|
||
self,
|
||
account_id: str,
|
||
robot_code: str,
|
||
process_query_key: str,
|
||
) -> dict:
|
||
"""单聊消息撤回（``batchRecallUserMessageByRobot``）。"""
|
||
return await self.request(
|
||
"POST",
|
||
"/v1.0/robot/oToMessages/batchRecall",
|
||
account_id,
|
||
json={
|
||
"robotCode": robot_code,
|
||
"processQueryKeys": [process_query_key],
|
||
},
|
||
)
|
||
|
||
async def recall_group_message(
|
||
self,
|
||
account_id: str,
|
||
robot_code: str,
|
||
message_id: str,
|
||
) -> dict:
|
||
"""群聊消息撤回（``recallGroupMessageByRobot``）。"""
|
||
return await self.request(
|
||
"POST",
|
||
"/v1.0/robot/groupMessages/recall",
|
||
account_id,
|
||
json={
|
||
"robotCode": robot_code,
|
||
"openConversationId": "",
|
||
"msgId": message_id,
|
||
},
|
||
)
|
||
|
||
# ------------------------------------------------------------------
|
||
# HTTP 模式 sessionWebhook 回复（设计方案 §4.1）
|
||
# ------------------------------------------------------------------
|
||
|
||
async def post_session_webhook(
|
||
self,
|
||
account_id: str,
|
||
session_webhook: str,
|
||
msg: dict,
|
||
) -> dict:
|
||
"""通过 HTTP 模式回调中的 sessionWebhook 临时地址直接 POST 回复。
|
||
|
||
sessionWebhook 由钉钉在 HTTP 模式回调事件中提供，含过期时间
|
||
（``sessionWebhookExpiredTime``）。本方法不携带 access_token，
|
||
仅通过 sessionWebhook 自带的临时鉴权信息投递回复。
|
||
|
||
@return 钉钉响应字典（含 ``errcode`` / ``errmsg``）。
|
||
"""
|
||
resp = await self._execute_http(
|
||
"POST",
|
||
session_webhook,
|
||
headers={"Content-Type": "application/json"},
|
||
json_body=msg,
|
||
)
|
||
data = resp.json()
|
||
code = data.get("errcode", 0)
|
||
if code != 0:
|
||
raise error_translator.translate_dingtalk_error(
|
||
code,
|
||
data.get("errmsg", ""),
|
||
) from None
|
||
return data
|
||
|
||
# ------------------------------------------------------------------
|
||
# 互动卡片 API（设计方案 §4.2）
|
||
# ------------------------------------------------------------------
|
||
|
||
async def send_interactive_card(
|
||
self,
|
||
account_id: str,
|
||
robot_code: str,
|
||
conversation_id: str,
|
||
user_id_list: list[str],
|
||
card_template_id: str,
|
||
card_data: dict,
|
||
out_track_id: str,
|
||
) -> dict:
|
||
"""发送互动卡片（``/v1.0/im/interactiveCards/send``）。"""
|
||
return await self.request(
|
||
"POST",
|
||
"/v1.0/im/interactiveCards/send",
|
||
account_id,
|
||
json={
|
||
"cardTemplateId": card_template_id,
|
||
"openConversationId": conversation_id,
|
||
"receiverUserIdList": user_id_list,
|
||
"outTrackId": out_track_id,
|
||
"robotCode": robot_code,
|
||
"cardData": {"cardParamMap": card_data},
|
||
},
|
||
)
|
||
|
||
async def update_interactive_card(
|
||
self,
|
||
account_id: str,
|
||
out_track_id: str,
|
||
card_data: dict,
|
||
) -> dict:
|
||
"""更新互动卡片（``/v1.0/im/interactiveCards``）。"""
|
||
return await self.request(
|
||
"PUT",
|
||
"/v1.0/im/interactiveCards",
|
||
account_id,
|
||
json={
|
||
"outTrackId": out_track_id,
|
||
"cardData": {"cardParamMap": card_data},
|
||
},
|
||
)
|
||
|
||
# ------------------------------------------------------------------
|
||
# 媒体上传/下载 API（设计方案 §3.5）
|
||
# ------------------------------------------------------------------
|
||
|
||
async def download_file(
|
||
self,
|
||
account_id: str,
|
||
download_url: str,
|
||
) -> bytes:
|
||
"""下载文件，返回二进制内容。
|
||
|
||
钉钉文件下载使用临时 ``downloadUrl``（事件 payload 中提供），不需
|
||
额外鉴权头（URL 自带签名），但走统一连接池与超时控制。
|
||
"""
|
||
resp = await self._execute_http(
|
||
"GET",
|
||
download_url,
|
||
headers={},
|
||
)
|
||
return resp.content
|
||
|
||
async def download_robot_image(
|
||
self,
|
||
account_id: str,
|
||
robot_code: str,
|
||
download_code: str,
|
||
) -> bytes:
|
||
"""下载机器人接收的图片（钉钉 ``/v1.0/robot/messageFiles/download``）。
|
||
|
||
钉钉机器人接收的图片消息含 ``imageCode``（下载码），非直接 URL。
|
||
需先调用 ``/v1.0/robot/messageFiles/download`` 用 ``downloadCode`` +
|
||
``robotCode`` 换取临时 ``downloadUrl``，再 GET 下载二进制内容。
|
||
|
||
@failure 钉钉 API 返回非 0 errcode 抛契约异常；HTTP 故障抛
|
||
``DependencyError``。
|
||
"""
|
||
result = await self.request(
|
||
"POST",
|
||
"/v1.0/robot/messageFiles/download",
|
||
account_id,
|
||
json={
|
||
"downloadCode": download_code,
|
||
"robotCode": robot_code,
|
||
},
|
||
)
|
||
download_url = result.get("downloadUrl", "")
|
||
if not download_url:
|
||
raise DependencyError(
|
||
dep=DINGTALK_API_DEP,
|
||
cause=ValueError("dingtalk messageFiles/download returned empty downloadUrl"),
|
||
)
|
||
return await self.download_file(account_id, download_url)
|
||
|
||
# ------------------------------------------------------------------
|
||
# 通讯录 API（设计方案 §3.6）
|
||
# ------------------------------------------------------------------
|
||
|
||
async def get_userid_by_unionid(
|
||
self,
|
||
account_id: str,
|
||
unionid: str,
|
||
) -> dict:
|
||
"""通过 unionId 查询 userId（``/topapi/user/getUseridByUnionid``）。"""
|
||
return await self.request(
|
||
"POST",
|
||
f"/topapi/user/getUseridByUnionid?unionid={unionid}",
|
||
account_id,
|
||
)
|
||
|
||
async def get_user_detail_by_userid(
|
||
self,
|
||
account_id: str,
|
||
userid: str,
|
||
) -> dict:
|
||
"""通过 userId 查询用户详情（``/topapi/v2/user/get``）。"""
|
||
return await self.request(
|
||
"POST",
|
||
"/topapi/v2/user/get",
|
||
account_id,
|
||
json={"userid": userid},
|
||
)
|
||
|
||
async def search_users(
|
||
self,
|
||
account_id: str,
|
||
keyword: str,
|
||
) -> dict:
|
||
"""搜索用户（钉钉通讯录搜索 API）。"""
|
||
return await self.request(
|
||
"POST",
|
||
"/topapi/contact/listserver",
|
||
account_id,
|
||
json={"keyword": keyword},
|
||
)
|
||
|
||
async def get_group_members(
|
||
self,
|
||
account_id: str,
|
||
conversation_id: str,
|
||
cursor: int = 0,
|
||
size: int = 100,
|
||
) -> dict:
|
||
"""查询群成员（``/topapi/im/chat/photocell/get``）。"""
|
||
return await self.request(
|
||
"POST",
|
||
"/topapi/im/chat/photocell/get",
|
||
account_id,
|
||
json={
|
||
"chatId": conversation_id,
|
||
"cursor": cursor,
|
||
"size": size,
|
||
},
|
||
)
|
||
|
||
async def get_group_detail(
|
||
self,
|
||
account_id: str,
|
||
conversation_id: str,
|
||
) -> dict:
|
||
"""查询群详情（``/topapi/im/chat/get``）。"""
|
||
return await self.request(
|
||
"POST",
|
||
"/topapi/im/chat/get",
|
||
account_id,
|
||
json={"chatId": conversation_id},
|
||
)
|
||
|
||
async def get_group_list(
|
||
self,
|
||
account_id: str,
|
||
cursor: int = 0,
|
||
size: int = 100,
|
||
) -> dict:
|
||
"""查询机器人所在群列表。"""
|
||
return await self.request(
|
||
"POST",
|
||
"/topapi/im/chat/list",
|
||
account_id,
|
||
json={"cursor": cursor, "size": size},
|
||
)
|
||
|
||
# ------------------------------------------------------------------
|
||
# Stream 接入 API（设计方案 §3.1）
|
||
# ------------------------------------------------------------------
|
||
|
||
async def get_stream_endpoint(
|
||
self,
|
||
client_id: str,
|
||
client_secret: str,
|
||
) -> dict:
|
||
"""获取 Stream 接入端点（connectionId + endpoint）。
|
||
|
||
钉钉 Stream 模式通过 POST ``/v1.0/gateway/connections/open`` 获取
|
||
WebSocket 接入端点与 connectionId，随后由 SDK 建立 WS 长连接。
|
||
本方法不携带 access_token（端点仅校验 ClientId/ClientSecret）。
|
||
"""
|
||
api_base_url = await self._get_channel_config("api_base_url")
|
||
url = f"{api_base_url}/v1.0/gateway/connections/open"
|
||
body = {"clientId": client_id, "clientSecret": client_secret}
|
||
resp = await self._execute_http(
|
||
"POST",
|
||
url,
|
||
headers={},
|
||
json_body=body,
|
||
)
|
||
data = resp.json()
|
||
code = data.get("code", 0)
|
||
if code != 0:
|
||
raise error_translator.translate_dingtalk_error(
|
||
int(code),
|
||
data.get("message", ""),
|
||
) from None
|
||
return data
|
||
|
||
|
||
def _to_json_str(value: dict) -> str:
|
||
"""将字典序列化为 JSON 字符串。
|
||
|
||
钉钉机器人消息 API 的 ``msgParam`` 字段需为 JSON 字符串（非嵌套对象），
|
||
与飞书 ``content`` 字段约定一致。
|
||
"""
|
||
import json
|
||
|
||
return json.dumps(value, ensure_ascii=False)
|
||
|
||
|
||
__all__ = ["DingTalkClient"]
|