python爬虫系列之下载在线文档Excel -- 石墨
一、简介
本文讲述使用python下载石墨在线文档中的Excel数据，下方有完整代码，方便食用。
- 思路
文档导出流程如下
使用抓包工具获取下载过程触发的接口，使用requests进行调用，即可完成下载。主要接口有三个，如下：获取文件名字接口、创建导出任务接口、查询任务进度接口。
这三个接口之间的关系：
- 提供在线文档地址
- 通过在线文档地址获取excel文件名
- 通过excel名字创建导出任务
- 通过创建任务的id，查询任务进度
- 任务进度100%时，会返回文件down_url
二、实现步骤
1. 数据准备
-
提供石墨文档中excel文档地址唯一标识字段，以便进行下载，获取excel文档id
document_id = Wr3DVRmBzVUdQkJQ
-
cookie值获取，通过接口工具抓到的接口，获取cookie信息，首先需要进行登陆。这里我用的抓包工具为 fiddler
cookie_value = ********
2. 获取excel文件名接口
获取文件名字接口，其中入参_为时间戳,调用该接口代码如下:
def get_excel_name(self):
"""
# 获取excel文件名
:return:
"""
timestamp = int(time() * 1000)
excel_name_url = self.document_url + f'?_={timestamp}&collaboratorCount=true'
res = requests.get(url=excel_name_url, headers=self.headers)
if res.json()['name']:
print("文件名获取成功")
return res.json()['name']
3.创建导出任务
创建导出任务，一个必填字段name为excel文件名。通过get_excel_name()获取，file字段为document_id= Wr3DVRmBzVUdQkJQ后缀中获取。
创建导出任务接口如下图
def export_excel_task(self):
document_name = self.get_excel_name()
bfb = urllib.parse.quote(str(document_name)) # 转为 url 编码
document_id = str(self.document_url).split('/')[-1]
export_url = self.document_url + f'/export?type=xlsx&file={document_id}&returnJson=1&name={bfb}&isAsync=1&timezoneOffset=-8'
res = requests.get(url=export_url, headers=self.headers)
data_dict = res.json()
if data_dict['status'] == 0:
print("导出任务id获取成功")
return data_dict['data']['taskId']
return "导出任务创建失败"
4. 查询进度进度，拿到文件下载地址
查询任务进度接口，在点击导出excel时会出现该接口信息，频繁调用该接口，当progress进度为100时，该接口返回值会出现file_url信息。
def query_progess(self):
taskId = self.export_excel_task()
progress_url = self.document_url + f'/export/progress?taskId={taskId}'
start_time = time()
file_url = ''
# 循环调用查询进度接口,直到进度为100,出现下载url
while True:
sleep(3)
res = requests.get(url=progress_url, headers = self.headers)
data = res.json()['data']
progress = data['progress']
if progress == 100:
file_url = data['downloadUrl']
break
elif time() - start_time > 30:
print("数据准备超时,请排查")
break
if file_url:
print("excel文件下载地址获取成功")
return file_url
5. 下载文件
石墨文档中该file_url会进行重定向,调用后通过返回值可以拿到真实的url
然后requests调用重定向后的url，就可以拿到文件流，写到excel文件中即可完成下载。
def download_file(self, file_name):
file_url = self.query_progess()
if file_url:
# self.headers['content-type'] = 'application/octet-stream'
res = requests.get(url=file_url)
sleep(2)
real_url = res.url
real_res = requests.get(url=real_url)
with open(file_name, 'wb') as f:
f.write(real_res.content)
print('下载成功,文件名: ' + file_name)
else:
print("下载文件地址获取失败, 下载excel文件不成功")
三、完整代码
因文档中不能包含shimo地址，这里用截图放出来，代码中用到{shimo}的地方注意替换。
# -*- coding: UTF-8 -*-
"""
@Project :small-tools
@File :shimo.py
@Author :silen
@Time :2022/5/27 22:06
@Description :
"""
import urllib.parse
from datetime import datetime
from time import time, sleep
import requests
class ShiMo():
def __init__(self, document_id, cookie_value):
# excel文档地址
self.document_url = {shimo_url} + 'lizard-api/files/' + document_id
self.headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/101.0.4951.64 Safari/537.36 Edg/101.0.1210.53',
'Cookie': cookie_value,
'Accept': 'application/nd.shimo.v2+json',
'Accept-Encoding': 'gzip, deflate, br',
'Referer': {shimo_url} + f'sheets/{document_id}',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
'Host': {shimo_url},
}
def get_excel_name(self):
"""
# 获取excel文件名
:return:
"""
timestamp = int(time() * 1000)
excel_name_url = self.document_url + f'?_={timestamp}&collaboratorCount=true'
res = requests.get(url=excel_name_url, headers=self.headers)
if res.json()['name']:
print("文件名获取成功")
return res.json()['name']
def export_excel_task(self):
document_name = self.get_excel_name()
bfb = urllib.parse.quote(str(document_name)) # 转为 url 编码
document_id = str(self.document_url).split('/')[-1]
export_url = self.document_url + f'/export?type=xlsx&file={document_id}&returnJson=1&name={bfb}&isAsync=1&timezoneOffset=-8'
res = requests.get(url=export_url, headers=self.headers)
data_dict = res.json()
if data_dict['status'] == 0:
print("导出任务id获取成功")
return data_dict['data']['taskId']
return "导出任务创建失败"
def query_progess(self):
taskId = self.export_excel_task()
progress_url = self.document_url + f'/export/progress?taskId={taskId}'
start_time = time()
file_url = ''
# 循环调用查询进度接口,直到进度为100,出现下载url
while True:
sleep(3)
res = requests.get(url=progress_url, headers = self.headers)
data = res.json()['data']
progress = data['progress']
if progress == 100:
file_url = data['downloadUrl']
break
elif time() - start_time > 30:
print("数据准备超时,请排查")
break
if file_url:
print("excel文件下载地址获取成功")
return file_url
def download_file(self, file_name):
file_url = self.query_progess()
if file_url:
# self.headers['content-type'] = 'application/octet-stream'
res = requests.get(url=file_url)
sleep(2)
real_url = res.url
real_res = requests.get(url=real_url)
with open(file_name, 'wb') as f:
f.write(real_res.content)
print('下载成功,文件名: ' + file_name)
else:
print("下载文件地址获取失败, 下载excel文件不成功")
if __name__ == '__main__':
cookie_value = '***'
document_id = 'Wr3DVRmBzVUdQkJQ'
shimo = ShiMo(document_id, cookie_value)
# 文档名称
name = shimo.get_excel_name()
current_datetime = datetime.strftime(datetime.now(), '%Y_%m_%d_%H_%M_%S')
file_name = f'{current_datetime + name}.xlsx'
shimo.download_file(file_name)
四、效果演示
运行脚本截图：
成功下载的文档如下所示
如果我的文章对你有帮助，点个关注呗
如有疑问欢迎留言！
