Kris
0d067d61d3
本次提交包含多类代码优化与修复： 1. 重命名适配器协议类：WebhookTestable→WebhookTestAdapter、AttachmentUploadable→AttachmentUploadAdapter，并同步更新所有引用 2. 为飞书/企业微信插件添加凭证克隆能力开关配置 3. 修复飞书入站适配器URL解析错误，使用hostname替代host属性 4. 新增批量发送消息DTO与已读状态DTO 5. 新增插件目录列表接口与插件能力校验规则 6. 修复权限阶段配置，添加analytics权限映射 7. 优化健康检查、限流模块的环境变量配置支持 8. 修复配对过期扫描器参数名不匹配问题 9. 优化日志调用、文档注释与代码可读性 10. 移除废弃的AuditExportTask聚合根与相关导入
687 lines
21 KiB
Python
687 lines
21 KiB
Python
"""飞书 API HTTP 客户端。
|
||
|
||
通过 httpx.AsyncClient 调用飞书开放平台 API，封装 tenant_access_token
|
||
管理、通用请求与 IM / 通讯录 / OAuth API。HTTP 异常与飞书业务错误码
|
||
通过 error_translator 转换为契约层异常，不向外抛原生异常。token 通过
|
||
CachePort 缓存（key 为 feishu:token:{app_id}，TTL=expires_in-60s），
|
||
客户端本身不持有 mutable 状态。
|
||
|
||
依赖方向：仅 import yuxi.channels.contract.* 与标准库 / httpx，不污染
|
||
框架层。error_translator 为本插件内部模块（任务 2 产出）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import time
|
||
from typing import Any
|
||
|
||
import httpx
|
||
|
||
from yuxi.channels.contract.dtos.config import ConfigScope
|
||
from yuxi.channels.contract.errors import DependencyError
|
||
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
|
||
|
||
|
||
_TOKEN_CACHE_KEY_PREFIX = "feishu:token:"
|
||
_TOKEN_REFRESH_LEAD_TIME_SEC = 60
|
||
_CHANNEL_TYPE = "feishu"
|
||
|
||
|
||
class FeishuClient:
|
||
"""飞书 API HTTP 客户端。
|
||
|
||
通过 DI 接收 ConfigPort / CachePort / LoggerPort，不访问全局 settings
|
||
或 logger。token 通过 CachePort 缓存，客户端不持有 mutable 状态。所有
|
||
方法为 async，HTTP 异常通过 error_translator.wrap_http_exception 转换，
|
||
飞书业务错误码非 0 通过 error_translator.translate_feishu_error 转换。
|
||
"""
|
||
|
||
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
|
||
|
||
# ------------------------------------------------------------------
|
||
# 内部辅助
|
||
# ------------------------------------------------------------------
|
||
|
||
async def _get_account_config(self, account_id: str) -> dict[str, Any]:
|
||
"""读取账户级配置（app_id / app_secret）。"""
|
||
cv_app_id = await self._config.get(
|
||
"app_id", scope=ConfigScope.ACCOUNT, target=account_id,
|
||
)
|
||
cv_app_secret = await self._config.get(
|
||
"app_secret", scope=ConfigScope.ACCOUNT, target=account_id,
|
||
)
|
||
return {
|
||
"app_id": cv_app_id.value,
|
||
"app_secret": cv_app_secret.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_TYPE,
|
||
)
|
||
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, app_id: str) -> str:
|
||
return f"{_TOKEN_CACHE_KEY_PREFIX}{app_id}"
|
||
|
||
async def _get_token_for_account(self, account_id: str) -> str:
|
||
"""获取账户的 tenant_access_token。"""
|
||
account_config = await self._get_account_config(account_id)
|
||
token, _ = await self.get_tenant_access_token(
|
||
account_config["app_id"], account_config["app_secret"],
|
||
)
|
||
return token
|
||
|
||
async def _build_auth_header(self, account_id: str) -> dict[str, str]:
|
||
"""获取账户 token 并构造 Authorization 头。"""
|
||
token = await self._get_token_for_account(account_id)
|
||
return {"Authorization": f"Bearer {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 注入与业务错误码校验，由调用方处理。每次请求创建
|
||
独立 httpx.AsyncClient，不持有 mutable 连接池状态。HTTP 异常
|
||
经 ``error_translator.wrap_http_exception`` 转换，并传入实际配置
|
||
的 ``timeout_ms`` 保证 ``TimeoutError`` 携带准确超时值。
|
||
"""
|
||
timeout = await self._get_request_timeout_seconds()
|
||
try:
|
||
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(
|
||
"Feishu 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 响应，校验飞书业务错误码。
|
||
|
||
code 非 0 时通过 error_translator.translate_feishu_error 转换为
|
||
契约异常。code 为 0 时返回完整响应字典。
|
||
"""
|
||
data = resp.json()
|
||
code = data.get("code", 0)
|
||
if code != 0:
|
||
raise error_translator.translate_feishu_error(
|
||
code, data.get("msg", ""),
|
||
) 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:
|
||
"""发起带 tenant 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_tenant_access_token(
|
||
self,
|
||
app_id: str,
|
||
app_secret: str,
|
||
) -> tuple[str, int]:
|
||
"""获取 tenant_access_token，缓存命中返回剩余 TTL。
|
||
|
||
缓存写入时存储绝对过期时间戳 ``expires_at``，缓存命中时根据
|
||
``expires_at - now`` 计算剩余 TTL，避免长期持有 token 的调用方
|
||
误判有效期。``expires_in <= 0`` 表示已过期，调用方应触发重新获取。
|
||
|
||
@return (token, remaining_expires_in)
|
||
"""
|
||
cache_key = self._token_cache_key(app_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}/open-apis/auth/v3/tenant_access_token/internal"
|
||
body = {"app_id": app_id, "app_secret": app_secret}
|
||
|
||
resp = await self._execute_http(
|
||
"POST", url, headers={}, json_body=body,
|
||
)
|
||
data = self._parse_json_response(resp)
|
||
token = data["tenant_access_token"]
|
||
expires_in = int(data.get("expire", 0))
|
||
if expires_in <= 0:
|
||
raise DependencyError(
|
||
dep="feishu_api",
|
||
cause=ValueError(f"invalid expire 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(
|
||
"Feishu tenant_access_token refreshed",
|
||
app_id=app_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,
|
||
receive_id_type: str | None = None,
|
||
) -> dict:
|
||
"""通用请求方法。
|
||
|
||
自动获取 tenant_access_token，注入 Authorization 头，发起 HTTP
|
||
请求并校验业务错误码。receive_id_type 非空时作为查询参数注入。
|
||
"""
|
||
all_params: dict[str, Any] = dict(params) if params else {}
|
||
if receive_id_type is not None:
|
||
all_params["receive_id_type"] = receive_id_type
|
||
return await self._request_json(
|
||
method,
|
||
path,
|
||
account_id,
|
||
json_body=json,
|
||
params=all_params or None,
|
||
)
|
||
|
||
# ------------------------------------------------------------------
|
||
# IM API
|
||
# ------------------------------------------------------------------
|
||
|
||
async def send_message(
|
||
self,
|
||
account_id: str,
|
||
receive_id: str,
|
||
msg_type: str,
|
||
content: str,
|
||
receive_id_type: str = "open_id",
|
||
) -> dict:
|
||
"""发送消息。"""
|
||
return await self.request(
|
||
"POST",
|
||
"/open-apis/im/v1/messages",
|
||
account_id,
|
||
json={
|
||
"receive_id": receive_id,
|
||
"msg_type": msg_type,
|
||
"content": content,
|
||
},
|
||
receive_id_type=receive_id_type,
|
||
)
|
||
|
||
async def update_message(
|
||
self,
|
||
account_id: str,
|
||
message_id: str,
|
||
content: str,
|
||
) -> dict:
|
||
"""更新消息。"""
|
||
return await self.request(
|
||
"PATCH",
|
||
f"/open-apis/im/v1/messages/{message_id}",
|
||
account_id,
|
||
json={"content": content},
|
||
)
|
||
|
||
async def recall_message(
|
||
self,
|
||
account_id: str,
|
||
message_id: str,
|
||
) -> dict:
|
||
"""撤回消息。"""
|
||
return await self.request(
|
||
"DELETE",
|
||
f"/open-apis/im/v1/messages/{message_id}",
|
||
account_id,
|
||
)
|
||
|
||
async def get_message(
|
||
self,
|
||
account_id: str,
|
||
message_id: str,
|
||
) -> dict:
|
||
"""查询消息。"""
|
||
return await self.request(
|
||
"GET",
|
||
f"/open-apis/im/v1/messages/{message_id}",
|
||
account_id,
|
||
)
|
||
|
||
async def reply_message(
|
||
self,
|
||
account_id: str,
|
||
message_id: str,
|
||
msg_type: str,
|
||
content: str,
|
||
) -> dict:
|
||
"""回复消息。"""
|
||
return await self.request(
|
||
"POST",
|
||
f"/open-apis/im/v1/messages/{message_id}/reply",
|
||
account_id,
|
||
json={"msg_type": msg_type, "content": content},
|
||
)
|
||
|
||
async def batch_get_messages(
|
||
self,
|
||
account_id: str,
|
||
message_ids: list[str],
|
||
) -> dict:
|
||
"""批量查询消息。"""
|
||
return await self.request(
|
||
"GET",
|
||
"/open-apis/im/v1/messages/batch_get",
|
||
account_id,
|
||
params={"message_ids": ",".join(message_ids)},
|
||
)
|
||
|
||
async def update_card(
|
||
self,
|
||
account_id: str,
|
||
message_id: str,
|
||
card_content: dict,
|
||
) -> dict:
|
||
"""更新卡片消息。
|
||
|
||
飞书 PATCH 接口期望 ``content`` 为 JSON 字符串，对齐
|
||
``send_message`` 的 ``msg_type=interactive`` 投递约定。
|
||
"""
|
||
return await self.request(
|
||
"PATCH",
|
||
f"/open-apis/im/v1/messages/{message_id}",
|
||
account_id,
|
||
json={"content": json.dumps(card_content)},
|
||
)
|
||
|
||
async def streaming_start(
|
||
self,
|
||
account_id: str,
|
||
receive_id: str,
|
||
content: dict,
|
||
receive_id_type: str = "open_id",
|
||
) -> dict:
|
||
"""开启流式消息。
|
||
|
||
飞书流式接口期望 ``content`` 为 JSON 字符串，对齐
|
||
``send_message`` 与 ``update_card`` 的序列化约定。
|
||
"""
|
||
return await self.request(
|
||
"POST",
|
||
"/open-apis/im/v1/messages/streaming",
|
||
account_id,
|
||
json={
|
||
"receive_id": receive_id,
|
||
"content": json.dumps(content),
|
||
},
|
||
receive_id_type=receive_id_type,
|
||
)
|
||
|
||
async def streaming_update(
|
||
self,
|
||
account_id: str,
|
||
message_id: str,
|
||
content: dict,
|
||
) -> dict:
|
||
"""更新流式消息。
|
||
|
||
飞书流式接口期望 ``content`` 为 JSON 字符串，对齐
|
||
``send_message`` 与 ``update_card`` 的序列化约定。
|
||
"""
|
||
return await self.request(
|
||
"PATCH",
|
||
f"/open-apis/im/v1/messages/streaming/{message_id}",
|
||
account_id,
|
||
json={"content": json.dumps(content)},
|
||
)
|
||
|
||
async def streaming_stop(
|
||
self,
|
||
account_id: str,
|
||
message_id: str,
|
||
content: dict,
|
||
) -> dict:
|
||
"""结束流式消息。
|
||
|
||
飞书流式接口期望 ``content`` 为 JSON 字符串，对齐
|
||
``send_message`` 与 ``update_card`` 的序列化约定。
|
||
"""
|
||
return await self.request(
|
||
"POST",
|
||
f"/open-apis/im/v1/messages/streaming/{message_id}/finish",
|
||
account_id,
|
||
json={"content": json.dumps(content)},
|
||
)
|
||
|
||
async def add_reaction(
|
||
self,
|
||
account_id: str,
|
||
message_id: str,
|
||
emoji_type: str,
|
||
) -> dict:
|
||
"""添加表情回应。"""
|
||
return await self.request(
|
||
"POST",
|
||
f"/open-apis/im/v1/messages/{message_id}/reactions",
|
||
account_id,
|
||
json={"reaction_type": {"emoji_type": emoji_type}},
|
||
)
|
||
|
||
async def delete_reaction(
|
||
self,
|
||
account_id: str,
|
||
message_id: str,
|
||
reaction_id: str,
|
||
) -> dict:
|
||
"""删除表情回应。"""
|
||
return await self.request(
|
||
"DELETE",
|
||
f"/open-apis/im/v1/messages/{message_id}/reactions/{reaction_id}",
|
||
account_id,
|
||
)
|
||
|
||
async def pin_message(
|
||
self,
|
||
account_id: str,
|
||
message_id: str,
|
||
) -> dict:
|
||
"""Pin 消息。"""
|
||
return await self.request(
|
||
"POST",
|
||
"/open-apis/im/v1/pins",
|
||
account_id,
|
||
json={"message_id": message_id},
|
||
)
|
||
|
||
async def unpin_message(
|
||
self,
|
||
account_id: str,
|
||
message_id: str,
|
||
) -> dict:
|
||
"""取消 Pin 消息。"""
|
||
return await self.request(
|
||
"DELETE",
|
||
f"/open-apis/im/v1/pins/{message_id}",
|
||
account_id,
|
||
)
|
||
|
||
async def get_message_read_info(
|
||
self,
|
||
account_id: str,
|
||
message_id: str,
|
||
) -> dict:
|
||
"""查询消息已读信息。
|
||
|
||
飞书返回 ``data.read_users`` 列表，含 ``user_id`` / ``read_time``。
|
||
群消息已读人数 = ``data.read_count``，目标总数 = ``data.target_count``。
|
||
"""
|
||
return await self.request(
|
||
"GET",
|
||
f"/open-apis/im/v1/messages/{message_id}/read_info",
|
||
account_id,
|
||
)
|
||
|
||
async def upload_image(
|
||
self,
|
||
account_id: str,
|
||
image_data: bytes,
|
||
image_type: str = "message",
|
||
) -> dict:
|
||
"""上传图片（multipart/form-data）。"""
|
||
return await self._request_json(
|
||
"POST",
|
||
"/open-apis/im/v1/images",
|
||
account_id,
|
||
data={"image_type": image_type},
|
||
files={"image": ("image.png", image_data)},
|
||
)
|
||
|
||
async def upload_file(
|
||
self,
|
||
account_id: str,
|
||
file_data: bytes,
|
||
file_name: str,
|
||
file_type: str = "stream",
|
||
) -> dict:
|
||
"""上传文件（multipart/form-data）。"""
|
||
return await self._request_json(
|
||
"POST",
|
||
"/open-apis/im/v1/files",
|
||
account_id,
|
||
data={"file_type": file_type, "file_name": file_name},
|
||
files={"file": (file_name, file_data)},
|
||
)
|
||
|
||
async def download_resource(
|
||
self,
|
||
account_id: str,
|
||
message_id: str,
|
||
file_key: str,
|
||
file_type: str = "file",
|
||
) -> bytes:
|
||
"""下载资源文件，返回二进制内容。"""
|
||
headers = await self._build_auth_header(account_id)
|
||
api_base_url = await self._get_channel_config("api_base_url")
|
||
url = (
|
||
f"{api_base_url}/open-apis/im/v1/messages/{message_id}"
|
||
f"/resources/{file_key}"
|
||
)
|
||
resp = await self._execute_http(
|
||
"GET", url, headers, params={"type": file_type},
|
||
)
|
||
return resp.content
|
||
|
||
# ------------------------------------------------------------------
|
||
# 通讯录 API
|
||
# ------------------------------------------------------------------
|
||
|
||
async def get_user(
|
||
self,
|
||
account_id: str,
|
||
user_id: str,
|
||
user_id_type: str = "open_id",
|
||
) -> dict:
|
||
"""查询用户信息。"""
|
||
return await self.request(
|
||
"GET",
|
||
f"/open-apis/contact/v3/users/{user_id}",
|
||
account_id,
|
||
params={"user_id_type": user_id_type},
|
||
)
|
||
|
||
async def search_users(
|
||
self,
|
||
account_id: str,
|
||
query: str,
|
||
page_size: int = 50,
|
||
page_token: str | None = None,
|
||
) -> dict:
|
||
"""搜索用户。"""
|
||
params: dict[str, Any] = {"query": query, "page_size": page_size}
|
||
if page_token is not None:
|
||
params["page_token"] = page_token
|
||
return await self.request(
|
||
"GET",
|
||
"/open-apis/search/v1/user",
|
||
account_id,
|
||
params=params,
|
||
)
|
||
|
||
async def get_group_list(
|
||
self,
|
||
account_id: str,
|
||
page_size: int = 50,
|
||
page_token: str | None = None,
|
||
) -> dict:
|
||
"""查询群列表。"""
|
||
params: dict[str, Any] = {"page_size": page_size}
|
||
if page_token is not None:
|
||
params["page_token"] = page_token
|
||
return await self.request(
|
||
"GET",
|
||
"/open-apis/im/v1/chats",
|
||
account_id,
|
||
params=params,
|
||
)
|
||
|
||
async def get_group_members(
|
||
self,
|
||
account_id: str,
|
||
group_id: str,
|
||
page_size: int = 50,
|
||
page_token: str | None = None,
|
||
) -> dict:
|
||
"""查询群成员。"""
|
||
params: dict[str, Any] = {"page_size": page_size}
|
||
if page_token is not None:
|
||
params["page_token"] = page_token
|
||
return await self.request(
|
||
"GET",
|
||
f"/open-apis/im/v1/chats/{group_id}/members",
|
||
account_id,
|
||
params=params,
|
||
)
|
||
|
||
async def get_group_detail(
|
||
self,
|
||
account_id: str,
|
||
group_id: str,
|
||
) -> dict:
|
||
"""查询群详情。"""
|
||
return await self.request(
|
||
"GET",
|
||
f"/open-apis/im/v1/chats/{group_id}",
|
||
account_id,
|
||
)
|
||
|
||
# ------------------------------------------------------------------
|
||
# OAuth API
|
||
# ------------------------------------------------------------------
|
||
|
||
async def get_qr_login_url(
|
||
self,
|
||
account_id: str,
|
||
redirect_uri: str,
|
||
state: str,
|
||
) -> dict:
|
||
"""获取扫码登录 URL。"""
|
||
return await self.request(
|
||
"POST",
|
||
"/open-apis/authen/v1/index",
|
||
account_id,
|
||
json={"redirect_uri": redirect_uri, "state": state},
|
||
)
|
||
|
||
async def poll_qr_login_status(
|
||
self,
|
||
account_id: str,
|
||
login_id: str,
|
||
) -> dict:
|
||
"""轮询扫码登录状态。"""
|
||
return await self.request(
|
||
"GET",
|
||
f"/open-apis/authen/v1/login_state/{login_id}",
|
||
account_id,
|
||
)
|
||
|
||
async def revoke_token(
|
||
self,
|
||
account_id: str,
|
||
user_access_token: str,
|
||
) -> dict:
|
||
"""撤销用户 access token。
|
||
|
||
使用 user_access_token 而非 tenant_access_token 进行认证，
|
||
account_id 仅用于定位渠道级 api_base_url。请求体需携带被撤销的
|
||
``access_token``，对齐飞书 authen/v1/access_tokens/revoke 协议。
|
||
"""
|
||
api_base_url = await self._get_channel_config("api_base_url")
|
||
url = f"{api_base_url}/open-apis/authen/v1/access_tokens/revoke"
|
||
headers = {"Authorization": f"Bearer {user_access_token}"}
|
||
resp = await self._execute_http(
|
||
"POST",
|
||
url,
|
||
headers,
|
||
json_body={"access_token": user_access_token},
|
||
)
|
||
return self._parse_json_response(resp)
|
||
|
||
|
||
__all__ = ["FeishuClient"]
|