Codex
242 lines
8.0 KiB
Python
242 lines
8.0 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
钉钉通讯录 API 客户端
|
|
封装钉钉开放平台部门和用户相关接口
|
|
"""
|
|
import logging
|
|
import time
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
import httpx
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
DINGTALK_API_BASE = "https://oapi.dingtalk.com"
|
|
|
|
|
|
class DingtalkClient:
|
|
"""钉钉通讯录 API 客户端"""
|
|
|
|
def __init__(self, app_key: str, app_secret: str):
|
|
self.app_key = app_key
|
|
self.app_secret = app_secret
|
|
self._access_token: Optional[str] = None
|
|
self._token_expires_at: float = 0
|
|
|
|
async def get_access_token(self) -> str:
|
|
"""获取 access_token（带内存缓存，提前 5 分钟过期）"""
|
|
now = time.time()
|
|
if self._access_token and now < self._token_expires_at:
|
|
return self._access_token
|
|
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
resp = await client.get(
|
|
f"{DINGTALK_API_BASE}/gettoken",
|
|
params={"appkey": self.app_key, "appsecret": self.app_secret},
|
|
)
|
|
resp.raise_for_status()
|
|
result = resp.json()
|
|
|
|
if result.get("errcode") != 0:
|
|
raise Exception(f"获取钉钉 access_token 失败: {result.get('errmsg', result)}")
|
|
|
|
self._access_token = result["access_token"]
|
|
self._token_expires_at = now + result.get("expires_in", 7200) - 300
|
|
return self._access_token
|
|
|
|
async def _post(self, path: str, body: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""带 access_token 的 POST 请求"""
|
|
token = await self.get_access_token()
|
|
async with httpx.AsyncClient(timeout=15.0) as client:
|
|
resp = await client.post(
|
|
f"{DINGTALK_API_BASE}{path}",
|
|
params={"access_token": token},
|
|
json=body,
|
|
)
|
|
resp.raise_for_status()
|
|
result = resp.json()
|
|
|
|
if result.get("errcode") != 0:
|
|
raise Exception(f"钉钉 API 调用失败 [{path}]: {result.get('errmsg', result)}")
|
|
return result
|
|
|
|
# ==================== 连接测试 ====================
|
|
|
|
async def test_connection(self) -> Dict[str, Any]:
|
|
"""
|
|
测试连接：获取 token + 拉取根部门信息验证
|
|
返回根部门名称等基本信息
|
|
"""
|
|
await self.get_access_token()
|
|
result = await self._post("/topapi/v2/department/get", {"dept_id": 1})
|
|
dept_info = result.get("result", {})
|
|
return {
|
|
"success": True,
|
|
"corp_name": dept_info.get("name", ""),
|
|
"dept_id": dept_info.get("dept_id"),
|
|
}
|
|
|
|
# ==================== 部门 API ====================
|
|
|
|
async def get_dept_list(self, dept_id: int = 1) -> List[Dict[str, Any]]:
|
|
"""
|
|
获取子部门列表
|
|
https://open.dingtalk.com/document/orgapp/obtain-the-department-list-v2
|
|
"""
|
|
result = await self._post(
|
|
"/topapi/v2/department/listsub",
|
|
{"dept_id": dept_id},
|
|
)
|
|
return result.get("result", [])
|
|
|
|
async def get_dept_detail(self, dept_id: int) -> Dict[str, Any]:
|
|
"""获取部门详情"""
|
|
result = await self._post(
|
|
"/topapi/v2/department/get",
|
|
{"dept_id": dept_id},
|
|
)
|
|
return result.get("result", {})
|
|
|
|
async def get_all_depts(self, root_dept_id: int = 1) -> List[Dict[str, Any]]:
|
|
"""
|
|
递归获取指定根部门下的全量部门列表（广度优先）
|
|
返回扁平列表，包含 parent_id 信息
|
|
"""
|
|
all_depts = []
|
|
queue = [root_dept_id]
|
|
|
|
while queue:
|
|
current_dept_id = queue.pop(0)
|
|
|
|
if current_dept_id != root_dept_id:
|
|
detail = await self.get_dept_detail(current_dept_id)
|
|
detail["parent_dept_id"] = detail.get("parent_id", root_dept_id)
|
|
all_depts.append(detail)
|
|
|
|
children = await self.get_dept_list(current_dept_id)
|
|
for child in children:
|
|
child_id = child.get("dept_id")
|
|
if child_id:
|
|
queue.append(child_id)
|
|
|
|
return all_depts
|
|
|
|
async def get_dept_tree(self, root_dept_id: int = 1) -> List[Dict[str, Any]]:
|
|
"""
|
|
获取部门树形结构（供前端选择同步范围）
|
|
"""
|
|
children = await self.get_dept_list(root_dept_id)
|
|
tree = []
|
|
for child in children:
|
|
node = {
|
|
"dept_id": child.get("dept_id"),
|
|
"name": child.get("name"),
|
|
"children": await self.get_dept_tree(child.get("dept_id")),
|
|
}
|
|
tree.append(node)
|
|
return tree
|
|
|
|
# ==================== 用户 API ====================
|
|
|
|
async def get_user_list(self, dept_id: int, cursor: int = 0, size: int = 100) -> Dict[str, Any]:
|
|
"""
|
|
获取部门用户列表（分页）
|
|
https://open.dingtalk.com/document/orgapp/queries-the-complete-information-of-a-department-user
|
|
"""
|
|
result = await self._post(
|
|
"/topapi/v2/user/list",
|
|
{"dept_id": dept_id, "cursor": cursor, "size": size},
|
|
)
|
|
return result.get("result", {})
|
|
|
|
async def get_all_users_in_dept(self, dept_id: int) -> List[Dict[str, Any]]:
|
|
"""获取部门下所有用户（自动翻页）"""
|
|
users = []
|
|
cursor = 0
|
|
|
|
while True:
|
|
page = await self.get_user_list(dept_id, cursor=cursor)
|
|
page_list = page.get("list", [])
|
|
users.extend(page_list)
|
|
|
|
if not page.get("has_more", False):
|
|
break
|
|
cursor = page.get("next_cursor", 0)
|
|
|
|
return users
|
|
|
|
async def get_user_detail(self, userid: str) -> Dict[str, Any]:
|
|
"""获取用户详情"""
|
|
result = await self._post(
|
|
"/topapi/v2/user/get",
|
|
{"userid": userid},
|
|
)
|
|
return result.get("result", {})
|
|
|
|
# ==================== 回调注册 API ====================
|
|
|
|
async def register_callback(
|
|
self,
|
|
callback_url: str,
|
|
callback_tag: List[str],
|
|
token: str,
|
|
aes_key: str,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
注册事件回调
|
|
https://open.dingtalk.com/document/orgapp/registers-an-event-callback-interface
|
|
"""
|
|
return await self._post(
|
|
"/call_back/register_call_back",
|
|
{
|
|
"call_back_tag": callback_tag,
|
|
"token": token,
|
|
"aes_key": aes_key,
|
|
"url": callback_url,
|
|
},
|
|
)
|
|
|
|
async def update_callback(
|
|
self,
|
|
callback_url: str,
|
|
callback_tag: List[str],
|
|
token: str,
|
|
aes_key: str,
|
|
) -> Dict[str, Any]:
|
|
"""更新事件回调"""
|
|
return await self._post(
|
|
"/call_back/update_call_back",
|
|
{
|
|
"call_back_tag": callback_tag,
|
|
"token": token,
|
|
"aes_key": aes_key,
|
|
"url": callback_url,
|
|
},
|
|
)
|
|
|
|
async def delete_callback(self) -> Dict[str, Any]:
|
|
"""删除事件回调"""
|
|
token = await self.get_access_token()
|
|
async with httpx.AsyncClient(timeout=15.0) as client:
|
|
resp = await client.get(
|
|
f"{DINGTALK_API_BASE}/call_back/delete_call_back",
|
|
params={"access_token": token},
|
|
)
|
|
resp.raise_for_status()
|
|
result = resp.json()
|
|
if result.get("errcode") != 0:
|
|
raise Exception(f"删除回调失败: {result.get('errmsg', result)}")
|
|
return result
|
|
|
|
async def get_callback(self) -> Dict[str, Any]:
|
|
"""查询已注册的事件回调"""
|
|
token = await self.get_access_token()
|
|
async with httpx.AsyncClient(timeout=15.0) as client:
|
|
resp = await client.get(
|
|
f"{DINGTALK_API_BASE}/call_back/get_call_back",
|
|
params={"access_token": token},
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|