补充一下的钉钉下载表格的代码
评论
收藏

补充一下的钉钉下载表格的代码

经验分享
曾诚
2024-06-25 14:18·浏览量:305
曾诚
发布于 2024-06-25 14:18305浏览

钉钉下载表格数据的时候,表格数据并不一定会通过接口返回数据,所以导致有时候下载失败,是因为钉钉在中途加入了一个链接来存储数据,所以这一次我加入了请求数据的一步,现在基本上可以肯定,能够完美的下载钉钉表格啦

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 "下载失败"

庆祝

收藏2
全部评论1
最新
发布评论
评论