

钉钉下载表格数据的时候,表格数据并不一定会通过接口返回数据,所以导致有时候下载失败,是因为钉钉在中途加入了一个链接来存储数据,所以这一次我加入了请求数据的一步,现在基本上可以肯定,能够完美的下载钉钉表格啦
class DingXlsxDownloads:
def __init__(self, download, token, key, timeout=10):
self.TOKEN = token
self.KEY = key
self.DOWNLOADURL = download
self.TIMEOUT = timeout
def get_content(self) -> dict:
headers = {
"a-doc-key": self.KEY,
"content-type": "application/json",
}
cookies = {
"doc_atoken": self.TOKEN,
}
url = "https://alidocs.dingtalk.com/api/document/data"
response = requests.post(url, headers=headers, cookies=cookies,
data=json.dumps({}, separators=(',', ':'))).json()
try:
return {"name": response["data"]["fileMetaInfo"]["name"], "size":
response["data"]["documentContent"]["checkpoint"]["cpOssSize"], "content": json.loads(
response["data"]["documentContent"]["checkpoint"]["content"])}
except Exception as e:
content = requests.get(response["data"]["documentContent"]["checkpoint"]["ossUrl"])
return {"name": response["data"]["fileMetaInfo"]["name"], "size":
response["data"]["documentContent"]["checkpoint"]["cpOssSize"], "content": json.loads(content.text)}
def get_urls(self, name, size) -> dict:
headers = {
"a-doc-key": self.KEY,
"content-type": "application/json",
}
cookies = {
"doc_atoken": self.TOKEN,
}
url = "https://alidocs.dingtalk.com/core/api/resources/9/upload_info"
data = {
"size": size,
"resourceName": name,
"contentType": ""
}
data = json.dumps(data, separators=(',', ':'))
urls = requests.post(url, headers=headers, cookies=cookies, data=data).json()
return urls
def put_data(self, url, content) -> bool:
data = json.dumps(content, separators=(',', ':'))
response = requests.put(url, data=data)
if response.status_code == 200:
return True
else:
return False
def get_jobid(self, url) -> str:
headers = {
"a-doc-key": self.KEY,
"content-type": "application/json",
}
cookies = {
"doc_atoken": self.TOKEN
}
data = {
"exportType": "dingTalksheetToxlsx",
"storagePath": url
}
data = json.dumps(data, separators=(',', ':'))
response = requests.post("https://alidocs.dingtalk.com/core/api/document/submitExportJob", headers=headers,
cookies=cookies, data=data).json()
return response["data"]["jobId"]
def get_download_url(self, job_id) -> str:
headers = {
"a-doc-key": self.KEY,
}
cookies = {
"doc_atoken": self.TOKEN,
}
url = "https://alidocs.dingtalk.com/core/api/document/queryExportJobInfo"
params = {
"jobId": job_id
}
response = requests.get(url, headers=headers, cookies=cookies, params=params).json()
return response["data"]["ossUrl"]
def run(self):
dic = self.get_content()
url_dic = self.get_urls(dic["name"], dic["size"])
result = self.put_data(url_dic["data"]["uploadUrl"], dic["content"])
if result:
job_id = self.get_jobid(url_dic['data']["storagePath"])
for i in range(self.TIMEOUT):
try:
download_url = self.get_download_url(job_id)
print("请求成功")
data = requests.get(download_url)
if data.status_code == 200:
with open(self.DOWNLOADURL, 'wb') as file:
file.write(data.content)
return self.DOWNLOADURL
else:
return f'请求失败,状态码:{data.status_code}'
except Exception as e:
print(f"下载请求失败{i + 1}次")
time.sleep(1)
continue
else:
return "下载失败"