608 lines
19 KiB
Python
608 lines
19 KiB
Python
|
"""
|
|||
|
|
file: dingpan.py
|
|||
|
|
date: 2024-1112
|
|||
|
|
|
|||
|
|
访问钉钉的钉盘一个接口
|
|||
|
|
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import requests
|
|||
|
|
import json
|
|||
|
|
from datetime import datetime
|
|||
|
|
|
|||
|
|
#获取access_token
|
|||
|
|
def get_dingtalk_access_token(app_key, app_secret):
|
|||
|
|
# 定义请求的URL
|
|||
|
|
url = 'https://api.dingtalk.com/v1.0/oauth2/accessToken'
|
|||
|
|
|
|||
|
|
# 定义请求体的数据
|
|||
|
|
payload = {
|
|||
|
|
"appKey": app_key,
|
|||
|
|
"appSecret": app_secret
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 设置请求头，指定内容类型为JSON
|
|||
|
|
headers = {
|
|||
|
|
'Content-Type': 'application/json'
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 发送POST请求
|
|||
|
|
response = requests.post(url, data=json.dumps(payload), headers=headers)
|
|||
|
|
|
|||
|
|
# 检查响应状态码是否为200（HTTP OK）
|
|||
|
|
if response.status_code == 200:
|
|||
|
|
# 解析返回的JSON数据
|
|||
|
|
response_data = response.json()
|
|||
|
|
# 检查是否有access_token字段
|
|||
|
|
if 'accessToken' in response_data:
|
|||
|
|
return response_data['accessToken']
|
|||
|
|
else:
|
|||
|
|
print('Error: Access token not found in the response.')
|
|||
|
|
return None
|
|||
|
|
else:
|
|||
|
|
print(f'Error: Received status code {response.status_code}')
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
|
|||
|
|
#获取用户信息
|
|||
|
|
def get_user_details(access_token, userid, language='zh_CN'):
|
|||
|
|
# 定义请求的URL
|
|||
|
|
url = f'https://oapi.dingtalk.com/topapi/v2/user/get?access_token={access_token}'
|
|||
|
|
|
|||
|
|
# 构建请求体的数据
|
|||
|
|
payload = {
|
|||
|
|
"language": language,
|
|||
|
|
"userid": userid
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 设置请求头，指定内容类型为JSON
|
|||
|
|
headers = {
|
|||
|
|
'Content-Type': 'application/json'
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 发送POST请求
|
|||
|
|
response = requests.post(url, data=json.dumps(payload), headers=headers)
|
|||
|
|
|
|||
|
|
# 检查响应状态码是否为200（HTTP OK）
|
|||
|
|
if response.status_code == 200:
|
|||
|
|
# 解析返回的JSON数据
|
|||
|
|
response_data = response.json()
|
|||
|
|
# 检查是否有错误码
|
|||
|
|
if response_data.get('errcode') == 0:
|
|||
|
|
return response_data['result']
|
|||
|
|
else:
|
|||
|
|
print(f'Error: {response_data.get("errmsg")}')
|
|||
|
|
return None
|
|||
|
|
else:
|
|||
|
|
print(f'Error: Received status code {response.status_code}')
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
#以获取指定用户在组织空间下的文件夹列表
|
|||
|
|
def get_drive_spaces(access_token,union_id, space_type="org", next_token=None, max_results=50):
|
|||
|
|
# 定义请求的URL
|
|||
|
|
url = 'https://api.dingtalk.com/v1.0/drive/spaces'
|
|||
|
|
|
|||
|
|
# 构建查询参数
|
|||
|
|
params = {
|
|||
|
|
'unionId': union_id,
|
|||
|
|
'spaceType': space_type,
|
|||
|
|
'maxResults': max_results
|
|||
|
|
}
|
|||
|
|
if next_token:
|
|||
|
|
params['nextToken'] = next_token
|
|||
|
|
|
|||
|
|
# 设置请求头
|
|||
|
|
headers = {
|
|||
|
|
'x-acs-dingtalk-access-token': access_token,
|
|||
|
|
'Content-Type': 'application/json'
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 发送GET请求
|
|||
|
|
response = requests.get(url, params=params, headers=headers)
|
|||
|
|
|
|||
|
|
# 检查响应状态码是否为200（HTTP OK）
|
|||
|
|
if response.status_code == 200:
|
|||
|
|
# 解析返回的JSON数据
|
|||
|
|
response_data = response.json()
|
|||
|
|
return response_data["spaces"]
|
|||
|
|
else:
|
|||
|
|
print(f'Error: Received status code {response.status_code} {response.text}')
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
#用户的人数
|
|||
|
|
def get_user_count(access_token, only_active=False):
|
|||
|
|
# 定义请求的URL
|
|||
|
|
url = f'https://oapi.dingtalk.com/topapi/user/count?access_token={access_token}'
|
|||
|
|
|
|||
|
|
# 构建请求体的数据
|
|||
|
|
payload = {
|
|||
|
|
"only_active": str(only_active).lower()
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 设置请求头，指定内容类型为JSON
|
|||
|
|
headers = {
|
|||
|
|
'Content-Type': 'application/json'
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 发送POST请求
|
|||
|
|
response = requests.post(url, data=json.dumps(payload), headers=headers)
|
|||
|
|
|
|||
|
|
# 检查响应状态码是否为200（HTTP OK）
|
|||
|
|
if response.status_code == 200:
|
|||
|
|
# 解析返回的JSON数据
|
|||
|
|
response_data = response.json()
|
|||
|
|
# 检查是否有错误码
|
|||
|
|
if response_data.get('errcode') == 0:
|
|||
|
|
return response_data['result']['count']
|
|||
|
|
else:
|
|||
|
|
print(f'Error: {response_data.get("errmsg")}')
|
|||
|
|
return None
|
|||
|
|
else:
|
|||
|
|
print(f'Error: Received status code {response.status_code}')
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
#获取用户的id列表，
|
|||
|
|
def get_user_list_id(access_token, dept_id):
|
|||
|
|
# 定义请求的URL
|
|||
|
|
url = f'https://oapi.dingtalk.com/topapi/user/listid?access_token={access_token}'
|
|||
|
|
|
|||
|
|
# 构建请求体的数据
|
|||
|
|
payload = {
|
|||
|
|
"dept_id": dept_id
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 设置请求头，指定内容类型为JSON
|
|||
|
|
headers = {
|
|||
|
|
'Content-Type': 'application/json'
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 发送POST请求
|
|||
|
|
response = requests.post(url, data=json.dumps(payload), headers=headers)
|
|||
|
|
|
|||
|
|
# 检查响应状态码是否为200（HTTP OK）
|
|||
|
|
if response.status_code == 200:
|
|||
|
|
# 解析返回的JSON数据
|
|||
|
|
response_data = response.json()
|
|||
|
|
# 检查是否有错误码
|
|||
|
|
if response_data.get('errcode') == 0:
|
|||
|
|
return response_data['result']["userid_list"]
|
|||
|
|
else:
|
|||
|
|
print(f'Error: {response_data.get("errmsg")}')
|
|||
|
|
return None
|
|||
|
|
else:
|
|||
|
|
print(f'Error: Received status code {response.status_code}')
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
#得到子部门的列表
|
|||
|
|
def get_sub_departments(access_token, dept_id, language='zh_CN'):
|
|||
|
|
# 定义请求的URL
|
|||
|
|
url = f'https://oapi.dingtalk.com/topapi/v2/department/listsub?access_token={access_token}'
|
|||
|
|
|
|||
|
|
# 构建请求体的数据
|
|||
|
|
payload = {
|
|||
|
|
"language": language,
|
|||
|
|
"dept_id": dept_id
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 设置请求头，指定内容类型为JSON
|
|||
|
|
headers = {
|
|||
|
|
'Content-Type': 'application/json'
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 发送POST请求
|
|||
|
|
response = requests.post(url, data=json.dumps(payload), headers=headers)
|
|||
|
|
|
|||
|
|
# 检查响应状态码是否为200（HTTP OK）
|
|||
|
|
if response.status_code == 200:
|
|||
|
|
# 解析返回的JSON数据
|
|||
|
|
response_data = response.json()
|
|||
|
|
# 检查是否有错误码
|
|||
|
|
if response_data.get('errcode') == 0:
|
|||
|
|
return response_data['result']
|
|||
|
|
else:
|
|||
|
|
print(f'Error: {response_data.get("errmsg")}')
|
|||
|
|
return None
|
|||
|
|
else:
|
|||
|
|
print(f'Error: Received status code {response.status_code}')
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
|
|||
|
|
#得到组织内的管理员
|
|||
|
|
def get_admin_users(access_token):
|
|||
|
|
# 定义请求的URL
|
|||
|
|
url = f'https://oapi.dingtalk.com/topapi/user/listadmin?access_token={access_token}'
|
|||
|
|
|
|||
|
|
# 构建请求体的数据
|
|||
|
|
payload = {}
|
|||
|
|
|
|||
|
|
# 设置请求头，指定内容类型为JSON
|
|||
|
|
headers = {
|
|||
|
|
'Content-Type': 'application/json'
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 发送POST请求
|
|||
|
|
response = requests.post(url, data=json.dumps(payload), headers=headers)
|
|||
|
|
|
|||
|
|
# 检查响应状态码是否为200（HTTP OK）
|
|||
|
|
if response.status_code == 200:
|
|||
|
|
# 解析返回的JSON数据
|
|||
|
|
response_data = response.json()
|
|||
|
|
# 检查是否有错误码
|
|||
|
|
if response_data.get('errcode') == 0:
|
|||
|
|
return response_data['result']
|
|||
|
|
else:
|
|||
|
|
print(f'Error: {response_data.get("errmsg")}')
|
|||
|
|
return None
|
|||
|
|
else:
|
|||
|
|
print(f'Error: Received status code {response.status_code}')
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
|
|||
|
|
#所有文件和文件夹
|
|||
|
|
"""
|
|||
|
|
返回的文件信息
|
|||
|
|
[
|
|||
|
|
{
|
|||
|
|
"modifiedTime": "Fri Jun 21 15:43:42 CST 2024",
|
|||
|
|
"creatorId": "QksETWLqFiPBiSZ8oE1IcsqwiEiE",
|
|||
|
|
"modifierId": "QksETWLqFiPBiSZ8oE1IcsqwiEiE",
|
|||
|
|
"type": "FOLDER",
|
|||
|
|
"version": 1,
|
|||
|
|
"uuid": "7QG4Yx2JpL7eXAbrCz6m3Y4xJ9dEq3XD",
|
|||
|
|
"partitionType": "PUBLIC_OSS_PARTITION",
|
|||
|
|
"parentId": "143956017570",
|
|||
|
|
"spaceId": "24447524446",
|
|||
|
|
"path": "/分类分级/05用户手册",
|
|||
|
|
"createTime": "Fri Jun 21 15:43:42 CST 2024",
|
|||
|
|
"storageDriver": "DINGTALK",
|
|||
|
|
"name": "05用户手册",
|
|||
|
|
"id": "143956920642",
|
|||
|
|
"properties": {
|
|||
|
|
"readOnly": false
|
|||
|
|
},
|
|||
|
|
"status": "NORMAL",
|
|||
|
|
"appProperties": {}
|
|||
|
|
}
|
|||
|
|
]
|
|||
|
|
"""
|
|||
|
|
def list_all_dentries(space_id, union_id, access_token, next_token=None, max_results=1000, order="DESC", with_thumbnail=False):
|
|||
|
|
# 定义请求的URL
|
|||
|
|
url = f'https://api.dingtalk.com/v1.0/storage/spaces/{space_id}/dentries/listAll?unionId={union_id}'
|
|||
|
|
|
|||
|
|
# 构建请求体的数据
|
|||
|
|
payload = {
|
|||
|
|
"option": {
|
|||
|
|
"nextToken": next_token,
|
|||
|
|
"maxResults": max_results,
|
|||
|
|
"order": order,
|
|||
|
|
"withThumbnail": with_thumbnail
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 设置请求头，指定内容类型为JSON
|
|||
|
|
headers = {
|
|||
|
|
'x-acs-dingtalk-access-token': access_token,
|
|||
|
|
'Content-Type': 'application/json'
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 发送POST请求
|
|||
|
|
response = requests.post(url, data=json.dumps(payload), headers=headers)
|
|||
|
|
|
|||
|
|
# 检查响应状态码是否为200（HTTP OK）
|
|||
|
|
if response.status_code == 200:
|
|||
|
|
# 解析返回的JSON数据
|
|||
|
|
response_data = response.json()
|
|||
|
|
# 检查是否有错误码
|
|||
|
|
if 'error' not in response_data and "nextToken" in response_data:
|
|||
|
|
return response_data["dentries"],response_data["nextToken"]
|
|||
|
|
elif 'error' not in response_data:
|
|||
|
|
#遍历完没有文件了
|
|||
|
|
return response_data["dentries"],None
|
|||
|
|
else:
|
|||
|
|
print(f'Error: {response_data.get("error").get("message")}')
|
|||
|
|
return [],None
|
|||
|
|
else:
|
|||
|
|
print(f'Error: Received status code {response.status_code}')
|
|||
|
|
return [],None
|
|||
|
|
|
|||
|
|
|
|||
|
|
#等到下载文件的信息，可能是多个url,然后再调用url进行下载
|
|||
|
|
"""
|
|||
|
|
{
|
|||
|
|
"protocol": "HEADER_SIGNATURE",
|
|||
|
|
"headerSignatureInfo": {
|
|||
|
|
"headers": {
|
|||
|
|
"Authorization": "OSS LTAIjmWpzHta71rc:/2tIMDsXGyehOirAzCHx1cI0C8o=",
|
|||
|
|
"x-oss-date": "Tue, 03 Jun 2025 06:47:27 GMT"
|
|||
|
|
},
|
|||
|
|
"resourceUrls": [
|
|||
|
|
"https://sh-dualstack.trans.dingtalk.com/yundisk0/iAEIAqRmaWxlA6h5dW5kaXNrMATOIVA04gXNE48GzQ3bB85mqzrtCM0CmQ.file"
|
|||
|
|
],
|
|||
|
|
"expirationSeconds": 900,
|
|||
|
|
"internalResourceUrls": [
|
|||
|
|
"lippi-space-sh.oss-cn-shanghai-internal.aliyuncs.com/yundisk0/iAEIAqRmaWxlA6h5dW5kaXNrMATOIVA04gXNE48GzQ3bB85mqzrtCM0CmQ.file"
|
|||
|
|
],
|
|||
|
|
"region": "SHANGHAI"
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
"""
|
|||
|
|
def get_download_info(space_id, dentry_id,union_id, access_token, version=None, prefer_intranet=False):
|
|||
|
|
# 定义请求的URL
|
|||
|
|
url = f'https://api.dingtalk.com/v1.0/storage/spaces/{space_id}/dentries/{dentry_id}/downloadInfos/query?unionId={union_id}'
|
|||
|
|
|
|||
|
|
# 构建请求体的数据
|
|||
|
|
payload = {
|
|||
|
|
"option": {
|
|||
|
|
"version": version,
|
|||
|
|
"preferIntranet": prefer_intranet
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 设置请求头，指定内容类型为JSON
|
|||
|
|
headers = {
|
|||
|
|
'x-acs-dingtalk-access-token': access_token,
|
|||
|
|
'Content-Type': 'application/json'
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 发送POST请求
|
|||
|
|
response = requests.post(url, data=json.dumps(payload), headers=headers)
|
|||
|
|
|
|||
|
|
# 检查响应状态码是否为200（HTTP OK）
|
|||
|
|
if response.status_code == 200:
|
|||
|
|
# 解析返回的JSON数据
|
|||
|
|
response_data = response.json()
|
|||
|
|
# 检查是否有错误码
|
|||
|
|
if 'error' not in response_data:
|
|||
|
|
return response_data
|
|||
|
|
else:
|
|||
|
|
print(f'Error: {response_data.get("error").get("message")}')
|
|||
|
|
return None
|
|||
|
|
else:
|
|||
|
|
print(f'Error: Received status code {response.status_code},{response.text}')
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
|
|||
|
|
#等到下载文件的信息，可能是多个url,然后再调用url进行下载
|
|||
|
|
def download_file(urls,headers,file=""):
|
|||
|
|
|
|||
|
|
# 设置请求头，指定内容类型为JSON
|
|||
|
|
# headers = {
|
|||
|
|
# 'Authorization': header_auth,
|
|||
|
|
# "x-oss-date": oss_date
|
|||
|
|
# }
|
|||
|
|
|
|||
|
|
# 发送POST请求
|
|||
|
|
response_data=bytearray(b'')
|
|||
|
|
for url in urls:
|
|||
|
|
response = requests.get(url, headers=headers)
|
|||
|
|
|
|||
|
|
# 检查响应状态码是否为200（HTTP OK）
|
|||
|
|
if response.status_code == 200:
|
|||
|
|
# 解析返回的JSON数据
|
|||
|
|
response_data += response.content
|
|||
|
|
return response_data
|
|||
|
|
else:
|
|||
|
|
print(f'Error: Received status code {response.status_code},{response.text}')
|
|||
|
|
return None
|
|||
|
|
#end download_file
|
|||
|
|
|
|||
|
|
|
|||
|
|
#显示前50文件
|
|||
|
|
def ding_list_top_50_files(app_key,app_secret,main_dir=None):
|
|||
|
|
access_token = get_dingtalk_access_token(app_key, app_secret)
|
|||
|
|
|
|||
|
|
if access_token:
|
|||
|
|
print(f'Access Token: {access_token}')
|
|||
|
|
else:
|
|||
|
|
print('Failed to obtain access token.')
|
|||
|
|
admin_users = get_admin_users(access_token)
|
|||
|
|
|
|||
|
|
|
|||
|
|
user_details = get_user_details(access_token, admin_users[0]["userid"])
|
|||
|
|
|
|||
|
|
if user_details:
|
|||
|
|
print(json.dumps(user_details, indent=2, ensure_ascii=False))
|
|||
|
|
else:
|
|||
|
|
print('Failed to retrieve user details.')
|
|||
|
|
|
|||
|
|
|
|||
|
|
spaces=[]
|
|||
|
|
spaces_info = get_drive_spaces(access_token,user_details["unionid"])
|
|||
|
|
|
|||
|
|
if spaces_info:
|
|||
|
|
print(json.dumps(spaces_info))
|
|||
|
|
else:
|
|||
|
|
print('Failed to retrieve drive spaces information.')
|
|||
|
|
|
|||
|
|
if main_dir:
|
|||
|
|
for space in spaces_info:
|
|||
|
|
if space["spaceName"] in main_dir.split(","):
|
|||
|
|
print("钉盘",space["spaceName"])
|
|||
|
|
spaces.append(space)
|
|||
|
|
else:
|
|||
|
|
spaces = spaces_info
|
|||
|
|
|
|||
|
|
print(spaces)
|
|||
|
|
|
|||
|
|
files=[]
|
|||
|
|
#遍历目录
|
|||
|
|
for space in spaces:
|
|||
|
|
nextToken = None
|
|||
|
|
|
|||
|
|
dentries,nextToken = list_all_dentries(space["spaceId"], user_details["unionid"],access_token,nextToken)
|
|||
|
|
|
|||
|
|
if dentries:
|
|||
|
|
#print(json.dumps(dentries, indent=2, ensure_ascii=False))
|
|||
|
|
for dentrie in dentries:
|
|||
|
|
files.append({"path":dentrie["path"],
|
|||
|
|
"fname": dentrie["path"].split("/")[-1],
|
|||
|
|
"mtime": datetime.strptime(dentrie["modifiedTime"], "%a %b %d %H:%M:%S %Z %Y"),
|
|||
|
|
})
|
|||
|
|
else:
|
|||
|
|
print('Failed to retrieve dentries.')
|
|||
|
|
print("总文件",len(files))
|
|||
|
|
return files
|
|||
|
|
|
|||
|
|
#入口函数
|
|||
|
|
def ding_scan_directory(app_key,app_secret,main_dir=None):
|
|||
|
|
access_token = get_dingtalk_access_token(app_key, app_secret)
|
|||
|
|
|
|||
|
|
if access_token:
|
|||
|
|
print(f'Access Token: {access_token}')
|
|||
|
|
else:
|
|||
|
|
print('Failed to 获得access_token.')
|
|||
|
|
admin_users = get_admin_users(access_token)
|
|||
|
|
user_details = get_user_details(access_token, admin_users[0]["userid"])
|
|||
|
|
|
|||
|
|
if user_details:
|
|||
|
|
print(json.dumps(user_details, indent=2, ensure_ascii=False))
|
|||
|
|
else:
|
|||
|
|
print('Failed to 用户详细信息')
|
|||
|
|
|
|||
|
|
spaces=[]
|
|||
|
|
spaces_info = get_drive_spaces(access_token,user_details["unionid"])
|
|||
|
|
|
|||
|
|
if spaces_info:
|
|||
|
|
print(json.dumps(spaces_info))
|
|||
|
|
else:
|
|||
|
|
print('Failed to 获得钉盘信息')
|
|||
|
|
|
|||
|
|
if main_dir:
|
|||
|
|
for space in spaces_info:
|
|||
|
|
if space["spaceName"] in main_dir.split(","):
|
|||
|
|
print("钉盘",space["spaceName"])
|
|||
|
|
spaces.append(space)
|
|||
|
|
else:
|
|||
|
|
spaces = spaces_info
|
|||
|
|
|
|||
|
|
print(spaces)
|
|||
|
|
|
|||
|
|
files=[]
|
|||
|
|
#遍历目录
|
|||
|
|
for space in spaces:
|
|||
|
|
nextToken = None
|
|||
|
|
while 1:
|
|||
|
|
dentries,nextToken = list_all_dentries(space["spaceId"], user_details["unionid"],access_token,nextToken)
|
|||
|
|
|
|||
|
|
if dentries:
|
|||
|
|
#print(json.dumps(dentries, indent=2, ensure_ascii=False))
|
|||
|
|
for dentrie in dentries:
|
|||
|
|
files.append({"path":dentrie["path"],
|
|||
|
|
"fname": dentrie["path"].split("/")[-1],
|
|||
|
|
"id":dentrie["id"],
|
|||
|
|
"mtime": datetime.strptime(dentrie["modifiedTime"], "%a %b %d %H:%M:%S %Z %Y"),
|
|||
|
|
"unionid": user_details["unionid"],
|
|||
|
|
"spaceId": space["spaceId"],
|
|||
|
|
"access_token": access_token
|
|||
|
|
})
|
|||
|
|
else:
|
|||
|
|
print('Failed to retrieve dentries.')
|
|||
|
|
if nextToken==None:
|
|||
|
|
break
|
|||
|
|
print("总文件",len(files))
|
|||
|
|
return files
|
|||
|
|
|
|||
|
|
#单次下载文件
|
|||
|
|
def ding_get_file_content_init(app_key,app_secret,spaceId,file_id):
|
|||
|
|
access_token = get_dingtalk_access_token(app_key, app_secret)
|
|||
|
|
|
|||
|
|
if access_token:
|
|||
|
|
print(f'Access Token: {access_token}')
|
|||
|
|
else:
|
|||
|
|
print('Failed to obtain access token.')
|
|||
|
|
admin_users = get_admin_users(access_token)
|
|||
|
|
|
|||
|
|
|
|||
|
|
user_details = get_user_details(access_token, admin_users[0]["userid"])
|
|||
|
|
|
|||
|
|
if user_details:
|
|||
|
|
print(json.dumps(user_details, indent=2, ensure_ascii=False))
|
|||
|
|
else:
|
|||
|
|
print('Failed to retrieve user details.')
|
|||
|
|
unionid = user_details["unionid"]
|
|||
|
|
|
|||
|
|
return ding_get_file_content(spaceId,file_id,unionid,access_token)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def ding_get_file_content(spaceId,file_id,unionid,access_token):
|
|||
|
|
download_info = get_download_info(spaceId,file_id,unionid, access_token)
|
|||
|
|
if download_info:
|
|||
|
|
|
|||
|
|
return download_file(download_info["headerSignatureInfo"]["resourceUrls"],download_info["headerSignatureInfo"]["headers"])
|
|||
|
|
else:
|
|||
|
|
print('Failed to retrieve download info.')
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__=="__main__":
|
|||
|
|
# 示例调用
|
|||
|
|
app_key = 'app_key'
|
|||
|
|
app_secret = 'app_secret'
|
|||
|
|
access_token = get_dingtalk_access_token(app_key, app_secret)
|
|||
|
|
|
|||
|
|
if access_token:
|
|||
|
|
print(f'Access Token: {access_token}')
|
|||
|
|
else:
|
|||
|
|
print('Failed to obtain access token.')
|
|||
|
|
|
|||
|
|
user_count = get_user_count(access_token, only_active=False)
|
|||
|
|
print("用户数",user_count)
|
|||
|
|
|
|||
|
|
admin_users = get_admin_users(access_token)
|
|||
|
|
print("管理员",admin_users)
|
|||
|
|
|
|||
|
|
# sub_departments = get_sub_departments(access_token, 1)
|
|||
|
|
# print("部门",sub_departments)
|
|||
|
|
|
|||
|
|
# user_list_id = get_user_list_id(access_token, 1)
|
|||
|
|
# print("用户id",user_list_id)
|
|||
|
|
|
|||
|
|
|
|||
|
|
user_details = get_user_details(access_token, admin_users[0]["userid"])
|
|||
|
|
|
|||
|
|
if user_details:
|
|||
|
|
print(json.dumps(user_details, indent=2, ensure_ascii=False))
|
|||
|
|
else:
|
|||
|
|
print('Failed to retrieve user details.')
|
|||
|
|
|
|||
|
|
spaces_info = get_drive_spaces(access_token,user_details["unionid"])
|
|||
|
|
print("钉盘")
|
|||
|
|
if spaces_info:
|
|||
|
|
print(json.dumps(spaces_info))
|
|||
|
|
else:
|
|||
|
|
print('Failed to retrieve drive spaces information.')
|
|||
|
|
|
|||
|
|
print("存储",spaces_info[0]["spaceName"])
|
|||
|
|
|
|||
|
|
|
|||
|
|
#遍历目录
|
|||
|
|
count=0
|
|||
|
|
nextToken = None
|
|||
|
|
files=[]
|
|||
|
|
while 1:
|
|||
|
|
dentries,nextToken = list_all_dentries(spaces_info[0]["spaceId"], user_details["unionid"],access_token,nextToken)
|
|||
|
|
|
|||
|
|
if dentries:
|
|||
|
|
print(json.dumps(dentries, indent=2, ensure_ascii=False))
|
|||
|
|
count += len(dentries)
|
|||
|
|
files.extend(dentries)
|
|||
|
|
print("文件数量",count)
|
|||
|
|
else:
|
|||
|
|
print('Failed to retrieve dentries.')
|
|||
|
|
if nextToken==None:
|
|||
|
|
break
|
|||
|
|
|
|||
|
|
for file in files:
|
|||
|
|
print(file["path"])
|
|||
|
|
#更具id下载某一个文件
|
|||
|
|
download_info = get_download_info(spaces_info[1]["spaceId"],file["id"], user_details["unionid"], access_token)
|
|||
|
|
if download_info:
|
|||
|
|
if len(download_info["headerSignatureInfo"]["resourceUrls"]) >1:
|
|||
|
|
print(json.dumps(download_info, indent=2, ensure_ascii=False))
|
|||
|
|
#download_file(download_info["headerSignatureInfo"]["resourceUrls"],download_info["headerSignatureInfo"]["headers"])
|
|||
|
|
else:
|
|||
|
|
print('Failed to retrieve download info.')
|
|||
|
|
|
|||
|
|
print(json.dumps(user_details, indent=2, ensure_ascii=False))
|