python调用百度网盘开放平台接口上传本地文件

醉一场红尘 · 发布于 2023-6-7 08:58:15

本文章是为如何在没有GUI的环境下，使用指令行上传文件到百度网盘提供一个思路，其他操作请自行查询官方文档拓展。

前期工作

记录AppKey和SecreKey，后续开发需要使用

工作流程

https://openapi.baidu.com/oauth/2.0/authorize?response_type=code&client_id=YOUR_APP_KEY&redirect_uri=oob&scope=basic,netdisk&display=tv&qrcode=0&force_login=1

# 使用受权码
python main.py --code your_code
# 使用缓存，每次获取都会缓存一份缓存，access_token 30天过期，refresh_token 10年过期
python main.py

复制代码

例程

批量上传指定目录所有文件到百度网盘

import argparse
import requests
import json
import os
import hashlib
from urllib.parse import urlencode
app_name ='你的应用名称'# apps对应百度云盘我的应用数据文件夹
remote_path_prefix ='/apps/'+ app_name
json_path ='./token.json'# access_token获取地址
access_token_api ='https://openapi.baidu.com/oauth/2.0/token'# 预创建文件接口
precreate_api ='https://pan.baidu.com/rest/2.0/xpan/file?'# 分片上传api
upload_api ='https://d.pcs.baidu.com/rest/2.0/pcs/superfile2?'# 创建文件api
create_api ='https://pan.baidu.com/rest/2.0/xpan/file?'
app_key ='YOUR_APP_KEY'
secret_key ='YOUR_SECRET_KEY'classBaiduPanHelper:def__init__(self, code):
self.code = code
self.refresh_token =''
self.access_token =''# 创建文件defcreate(self, remote_path, size, uploadid, block_list):
params ={'method':'create','access_token': self.access_token,}
api = create_api + urlencode(params)
data ={'path': remote_path,'size': size,'isdir':0,'uploadid': uploadid,'block_list': block_list
}
response = requests.post(api, data=data)# 分片上传defupload(self, remote_path, uploadid, partseq, file_path):
data ={}
files =[('file',open(file_path,'rb'))]
params ={'method':'upload','access_token': self.access_token,'path': remote_path,'type':'tmpfile','uploadid': uploadid,'partseq': partseq
}
api = upload_api + urlencode(params)
res = requests.post(api, data=data, files=files)# 预上传defprecreate(self, file_path):
remote_path = remote_path_prefix
size = os.path.getsize(file_path)
arr = file_path.split('/')for item in arr[1::]:
remote_path = os.path.join(remote_path, item)
block_list =[]withopen(file_path,'rb')as f:
data = f.read()
file_md5 = hashlib.md5(data).hexdigest()
block_list.append(file_md5)
f.close()
block_list = json.dumps(block_list)
params ={'method':'precreate','access_token': self.access_token,}
data ={'path': remote_path,'size': size,'isdir':0,'autoinit':1,'block_list': block_list
}
api = precreate_api + urlencode(params)
res = requests.post(api, data=data)
json_resp = json.loads(res.content)if"path"in json_resp:return json_resp['uploadid'], remote_path, size, block_list
else:return'', remote_path, size, block_list
# 获得缓存的tokendefget_cache_token(self):withopen(json_path,'r')as f:
data = json.load(f)
self.refresh_token = data['refresh_token']
self.access_token = data['access_token']print(self.refresh_token)print(self.access_token)
f.close()# 根据受权码获取tokendefget_access_token(self):
data ={'grant_type':'authorization_code','code': self.code,'client_id': app_key,'client_secret': secret_key,'redirect_uri':'oob'}
res = requests.post(access_token_api, data=data)
json_resp = json.loads(res.content)
self.refresh_token = json_resp['refresh_token']
self.access_token = json_resp['access_token']withopen(json_path,'wb')as f:
f.write(res.content)
f.close()defupdate_file(self, file_path):# 1. 预上传
uploadid, remote_path, size, block_list = self.precreate(file_path)# 2. 分片上传（文件切片这里没有做，超级会员单文件最大20G）
self.upload(remote_path, uploadid,0, file_path)# 3. 创建文件
self.create(remote_path, size, uploadid, block_list)if __name__ =='__main__':
parser = argparse.ArgumentParser(description='百度云盘api测试')
parser.add_argument('--code',type=str, default='',help='受权码')
args = parser.parse_args()
bdpan = BaiduPanHelper(args.code)if bdpan.code is'':print('CODE is null')
bdpan.get_cache_token()else:print('CODE:'+ args.code)
bdpan.get_access_token()# 批量上传某个目录下的所有文件for root, dirs, files in os.walk('./test'):forfilein files:
file_path = os.path.join(root,file)
bdpan.update_file(file_path)print(file_path)

复制代码

上传效果