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)) |