204 lines
7.0 KiB
Python
204 lines
7.0 KiB
Python
# -*- coding: utf-8 -*-
|
||
import requests
|
||
import sys
|
||
from typing import List
|
||
from alibabacloud_dingtalk.wiki_2_0.client import Client as dingtalkwiki_2_0Client
|
||
from alibabacloud_tea_openapi import models as open_api_models
|
||
from alibabacloud_dingtalk.wiki_2_0 import models as wiki_models
|
||
from alibabacloud_tea_util import models as util_models
|
||
from dingding_calendars import DingTalkHelper
|
||
APP_KEY = "dinguetojbaxvvhzpk3d"
|
||
APP_SECRET = "lMFqns_ceLIcXvfLL8GKfa3ZiPKHcaZq0VbGtJXJlDuK8AEJ2WV3-PN8zv61ajm3"
|
||
|
||
|
||
def get_workspaces_details(client, token, operator_id, workspace_ids):
|
||
batch_size = 10
|
||
all_details = []
|
||
for i in range(0, len(workspace_ids), batch_size):
|
||
batch = workspace_ids[i:i+batch_size]
|
||
get_request = wiki_models.GetWorkspacesRequest(
|
||
operator_id=operator_id,
|
||
workspace_ids=batch,
|
||
option=wiki_models.GetWorkspacesRequestOption(with_permission_role=False)
|
||
)
|
||
get_headers = wiki_models.GetWorkspacesHeaders()
|
||
get_headers.x_acs_dingtalk_access_token = token
|
||
|
||
resp = client.get_workspaces_with_options(get_request, get_headers, util_models.RuntimeOptions())
|
||
details = getattr(resp.body, 'workspaces', [])
|
||
for d in details:
|
||
workspace_dict = {
|
||
"name": d.name,
|
||
"workspace_id": d.workspace_id,
|
||
"type": d.type,
|
||
"url": d.url,
|
||
"root_node_id": d.root_node_id
|
||
}
|
||
all_details.append(workspace_dict)
|
||
return all_details
|
||
|
||
def get_admin_userid(access_token: str) -> str:
|
||
url = "https://oapi.dingtalk.com/topapi/user/listadmin"
|
||
params = {"access_token": access_token}
|
||
resp = requests.get(url, params=params).json()
|
||
|
||
if resp.get("errcode") == 0:
|
||
admins = resp.get("result", [])
|
||
if admins:
|
||
# 返回第一个管理员的userid
|
||
return admins[0].get("userid", "")
|
||
else:
|
||
print("没有找到管理员")
|
||
return ""
|
||
else:
|
||
print(f"获取管理员失败: {resp}")
|
||
return ""
|
||
|
||
def get_userid_by_mobile(access_token: str, mobile: str) -> str:
|
||
url = "https://oapi.dingtalk.com/topapi/v2/user/getbymobile"
|
||
params = {"access_token": access_token}
|
||
data = {"mobile": mobile}
|
||
resp = requests.post(url, params=params, json=data)
|
||
r = resp.json()
|
||
if r.get("errcode") == 0:
|
||
print(f"✅ 获取 {mobile} 的 userId: {r['result']['userid']}")
|
||
return r["result"]["userid"]
|
||
print(f"❌ 获取 userId 失败: {r}")
|
||
return ""
|
||
|
||
|
||
|
||
def get_access_token(APP_KEY:str,APP_SECRET:str) -> str:
|
||
url = "https://oapi.dingtalk.com/gettoken"
|
||
params = {"appkey": APP_KEY, "appsecret": APP_SECRET}
|
||
resp = requests.get(url, params=params).json()
|
||
if resp.get("errcode") == 0:
|
||
print("✅ AccessToken 获取成功")
|
||
return resp["access_token"]
|
||
else:
|
||
raise Exception(f"获取AccessToken失败: {resp}")
|
||
|
||
|
||
def create_wiki_client() -> dingtalkwiki_2_0Client:
|
||
config = open_api_models.Config()
|
||
config.protocol = 'https'
|
||
config.region_id = 'central'
|
||
return dingtalkwiki_2_0Client(config)
|
||
|
||
|
||
def list_workspaces(client: dingtalkwiki_2_0Client, token: str, operator_id: str) -> List[str]:
|
||
next_token = None
|
||
all_workspace_ids = []
|
||
while True:
|
||
request = wiki_models.ListWorkspacesRequest(
|
||
operator_id=operator_id,
|
||
next_token=next_token,
|
||
max_results=30,
|
||
order_by='VIEW_TIME_DESC',
|
||
with_permission_role=False
|
||
)
|
||
headers = wiki_models.ListWorkspacesHeaders()
|
||
headers.x_acs_dingtalk_access_token = token
|
||
|
||
resp = client.list_workspaces_with_options(request, headers, util_models.RuntimeOptions())
|
||
workspaces = getattr(resp.body, 'workspaces', [])
|
||
all_workspace_ids.extend([ws.workspace_id for ws in workspaces])
|
||
|
||
next_token = getattr(resp.body, 'next_token', None)
|
||
if not next_token:
|
||
break
|
||
return all_workspace_ids
|
||
|
||
def list_all_nodes(client, operator_id, parent_node_id, token):
|
||
headers = wiki_models.ListNodesHeaders()
|
||
headers.x_acs_dingtalk_access_token = token
|
||
|
||
request = wiki_models.ListNodesRequest(
|
||
operator_id=operator_id,
|
||
parent_node_id=parent_node_id
|
||
)
|
||
|
||
try:
|
||
resp = client.list_nodes_with_options(request, headers, util_models.RuntimeOptions())
|
||
except Exception as e:
|
||
print(f"请求 ListNodes 出错: {e}")
|
||
return []
|
||
|
||
if not hasattr(resp.body, "nodes") or not resp.body.nodes:
|
||
return []
|
||
|
||
nodes = []
|
||
for node in resp.body.nodes:
|
||
node_dict = {
|
||
"name": node.name,
|
||
"node_id": node.node_id,
|
||
"type": node.type,
|
||
"url": getattr(node, "url", None),
|
||
"hasChildren": getattr(node, "has_children", False),
|
||
}
|
||
|
||
if node_dict["hasChildren"]:
|
||
node_dict["children"] = list_all_nodes(client, operator_id, node.node_id, token)
|
||
else:
|
||
node_dict["children"] = []
|
||
|
||
nodes.append(node_dict)
|
||
|
||
return nodes
|
||
|
||
def get_workspaces_details(client, token, operator_id, workspace_ids):
|
||
batch_size = 10
|
||
all_details = []
|
||
for i in range(0, len(workspace_ids), batch_size):
|
||
batch = workspace_ids[i:i+batch_size]
|
||
get_request = wiki_models.GetWorkspacesRequest(
|
||
operator_id=operator_id,
|
||
workspace_ids=batch,
|
||
option=wiki_models.GetWorkspacesRequestOption(with_permission_role=False)
|
||
)
|
||
get_headers = wiki_models.GetWorkspacesHeaders()
|
||
get_headers.x_acs_dingtalk_access_token = token
|
||
|
||
resp = client.get_workspaces_with_options(get_request, get_headers, util_models.RuntimeOptions())
|
||
details = getattr(resp.body, 'workspaces', [])
|
||
for d in details:
|
||
workspace_dict = {
|
||
"name": d.name,
|
||
"workspace_id": d.workspace_id,
|
||
"type": d.type,
|
||
"url": d.url,
|
||
"root_node_id": d.root_node_id,
|
||
"nodes": list_all_nodes(client, operator_id, d.root_node_id, token)
|
||
}
|
||
all_details.append(workspace_dict)
|
||
return all_details
|
||
|
||
def get_unionid(token, userid):
|
||
"""通过 userId 获取 unionid"""
|
||
url = "https://oapi.dingtalk.com/topapi/v2/user/get"
|
||
resp = requests.post(url, params={"access_token": token}, json={"userid": userid}).json()
|
||
if resp.get("errcode") != 0:
|
||
raise Exception(f"获取unionid失败: {resp}")
|
||
return resp["result"]["unionid"]
|
||
|
||
|
||
def main():
|
||
token = get_access_token(APP_KEY,APP_SECRET)
|
||
|
||
|
||
USERID=get_admin_userid(token)
|
||
print("userId：",USERID)
|
||
operator_id = get_unionid(token, USERID)
|
||
client = create_wiki_client()
|
||
|
||
print("开始获取知识库列表...")
|
||
workspace_ids = list_workspaces(client, token, operator_id)
|
||
print(f"共获取到 {len(workspace_ids)} 个知识库ID")
|
||
|
||
print("开始获取知识库详情...")
|
||
get_workspaces_details(client, token, operator_id, workspace_ids)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|