1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
| import json
import requests
def qr_print(code):
"""
打印授权二维码
:param code:
:return:
"""
import qrcode
import io
qr = qrcode.QRCode(
version=1,
error_correction=qrcode.constants.ERROR_CORRECT_L,
box_size=5,
border=4,
)
qr.add_data(code)
qr.make(fit=True)
f = io.StringIO()
qr.print_ascii(out=f)
f.seek(0)
print(f.read())
def sendEmail(link):
"""
发送邮箱百度网盘登录连接
:param link:
:return:
"""
import smtplib
import time
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
sender = 'x@qq.com'
receivers = 'x@qq.com'
message = MIMEMultipart('related')
subject = '发送百度网盘扫码验证' + time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
message['Subject'] = subject
message['From'] = sender
message['To'] = receivers
ss = '<p>发送百度网盘扫码验证</p><p><a href="' + link + '">百度网盘登录验证</a></p>'
content = MIMEText(ss, 'html', 'utf-8')
message.attach(content)
try:
server = smtplib.SMTP_SSL("smtp.qq.com", 465)
server.login(sender, "x")
server.sendmail(sender, receivers, message.as_string())
server.quit()
print("邮件发送成功")
except smtplib.SMTPException as e:
print(e)
class BaiduPanHelper:
def __init__(self):
self.appkey = 'x'
self.secretkey = "x"
self.device_code = "x"
self.user_code = "x"
self.access_token = ""
self.refresh_token = ""
self.expires_in = ""
self.verification_url = ""
self.app_name = 'ReportAnalysisBack'
self.remote_path_prefix = '/apps/' + self.app_name
self.json_path = './token.json'
print()
def login(self):
self._device_code_login()
input("请在120秒内完成授权并回车继续程序:-:")
self._device_code_get_token()
self.device_code_refresh_token()
def _device_code_login(self):
"""
采用设备码授权登录
https://pan.baidu.com/union/doc/fl1x114ti
:return:
"""
url = "https://openapi.baidu.com/oauth/2.0/device/code?response_type=device_code&client_id=" + self.appkey + "&scope=basic,netdisk"
print("设备码授权登录url:", url)
payload = {}
r = requests.request("GET", url, data=payload).text.encode('utf8')
print("设备授权码信息：", r)
self.device_code = json.loads(r).get('device_code')
self.user_code = json.loads(r).get('user_code')
self.verification_url = json.loads(r).get('verification_url')
print(self.device_code, self.user_code)
print("通过手机扫码下方二维码或者打开此连接输入授权码：",
self.verification_url)
print("授权码：", self.user_code)
print("或者直接打开此链接：", "https://openapi.baidu.com/device?display=mobile&code=" + self.user_code)
qr_print("https://openapi.baidu.com/device?display=mobile&code=" + self.user_code)
return self.device_code, self.user_code
def _device_code_get_token(self):
"""
用 Device Code 轮询换取 Access Token
:return:
通过 Device Code 轮询换取 Access Token。换取Access Token 的实现依赖于以下请求链接：
https://openapi.baidu.com/oauth/2.0/token?grant_type=device_token&
code=第一步生成的设备码device_code&
client_id=您应用的AppKey&
client_secret=您应用的SecretKey
关于应用的相关信息，您可在控制台，点进去您对应的应用，查看应用详情获得。
{
expires_in: 2592000,
refresh_token: "",
access_token: "",
session_secret: "",
session_key: "",
scope: "basic netdisk"
}
"""
url = "https://openapi.baidu.com/oauth/2.0/token?grant_type=device_token&code=" + self.device_code + "&client_id=" + self.appkey + "&client_secret=" + self.secretkey
print("设备码授权轮询换取 Access Token url:", url)
payload = {}
r = requests.request("GET", url, data=payload).text.encode('utf8')
print(r)
if json.loads(r).get('error') != 'invalid_request':
self.access_token = json.loads(r).get('access_token')
self.refresh_token = json.loads(r).get('refresh_token')
print(self.access_token, self.refresh_token)
return self.access_token, self.refresh_token
else:
return 'error'
def device_code_refresh_token(self):
"""
刷新 Access Token
设备码模式下支持刷新 Access Token。
通过 Refresh Token 刷新，具体依赖于以下链接：
GET https://openapi.baidu.com/oauth/2.0/token?
grant_type=refresh_token&
refresh_token=Refresh Token的值&
client_id=您应用的AppKey&
client_secret=您应用的SecretKey
以上链接示例中参数仅给出了必选参数。
关于应用的相关信息，您可在控制台，点进去您对应的应用，查看应用详情获得。
关于 Refresh Token的值，在换取 Access Token 凭证时，您可在响应信息中拿到。
:return:
"""
url = "https://openapi.baidu.com/oauth/2.0/token?grant_type=refresh_token&refresh_token=" + self.refresh_token + "&client_id=" + self.appkey + "&client_secret=" + self.secretkey
print("设备码授刷新Access Token url:", url)
payload = {}
r = requests.request("GET", url, data=payload).text.encode('utf8')
print(r)
if json.loads(r).get('error') != 'invalid_request':
self.writ_access_token(r)
self.access_token = json.loads(r).get('access_token')
self.refresh_token = json.loads(r).get('refresh_token')
print(self.access_token, self.refresh_token)
return self.access_token, self.refresh_token
else:
self.login()
def get_uinfo(self):
"""
{"avatar_url":"https://dss0.bdstatic.com/7Ls0a8Sm1A5BphGlnYG/sys/portrait/item/netdisk.1.413dda19.x.jpg","baidu_name":"hqlandv","errmsg":"succ","errno":0,"netdisk_name":"","request_id":"x","uk":x,"vip_type":2}
baidu_name string 百度帐号
netdisk_name string 网盘帐号
avatar_url string 头像地址
vip_type int 会员类型，0普通用户、1普通会员、2超级会员
uk int 用户ID
:param self:
:return:
"""
url = "https://pan.baidu.com/rest/2.0/xpan/nas?access_token=" + self.access_token + "&method=uinfo"
payload = {}
r = requests.request("GET", url, data=payload).text.encode('utf8')
print("百度网盘用户信息：", r)
def get_uinfo_t(self):
"""
用获取用户信息来判断是否有效
:return:
"""
url = "https://pan.baidu.com/rest/2.0/xpan/nas?access_token=" + self.access_token + "&method=uinfo"
payload = {}
r = requests.request("GET", url, data=payload).text.encode('utf8')
try:
print("测试token是否能用")
if json.loads(r).get('errmsg') == "Invalid Bduss":
print("token错误")
return False
else:
print("token能用")
return True
except:
return False
def get_file_list(self):
"""
TODO 完善列表获取
https://pan.baidu.com/union/doc/nksg0sat9
/rest/2.0/xpan/file?method=list&access_token=xx
method string 是 list URL参数 本接口固定为list
access_token string 是 x URL参数 接口鉴权参数
dir string 否 x URL参数 需要list的目录，以/开头的绝对路径, 默认为/
路径包含中文时需要UrlEncode编码
给出的示例的路径是/测试目录的UrlEncode编码
order string 否 name URL参数 排序字段：默认为name；
time表示先按文件类型排序，后按修改时间排序；
name表示先按文件类型排序，后按文件名称排序；
size表示先按文件类型排序，后按文件大小排序。
desc int 否 1 URL参数 默认为升序，设置为1实现降序 （注：排序的对象是当前目录下所有文件，不是当前分页下的文件）
start int 否 0 URL参数 起始位置，从0开始
limit int 否 100 URL参数 查询数目，默认为1000，建议最大不超过1000
web int 否 1 URL参数 值为1时，返回dir_empty属性和缩略图数据
folder int 否 0 URL参数 是否只返回文件夹，0 返回所有，1 只返回文件夹，且属性只返回path字段
showempty int 否 0 URL参数 是否返回dir_empty属性，0 不返回，1 返回
:param self:
:param dir:
:param order:
:return:
"""
url = "https://pan.baidu.com/rest/2.0/xpan/file?access_token=" + self.access_token + "&method=list"
print(url)
r = requests.get(url)
print("百度网盘目录列表：", r.json())
def _create(self):
"""
创建文件
:return:
"""
print()
def _upload(self):
"""
分片上传
:return:
"""
print()
def _precreate(self):
"""
预上传
:return:
"""
def update_file(self):
"""
文件上传
:return:
上传流程简介
上传流程是指，用户将本地文件上传到百度网盘云端服务器的过程。
文件上传分为三个阶段：预上传、分片上传、创建文件。
第二个阶段分片上传依赖第一个阶段预上传的结果，第三个阶段创建文件依赖第一个阶段预上传和第二阶段分片上传的结果，
串行完成这三个阶段任务后，本地文件成功上传到网盘服务器。
"""
self._precreate()
self._upload()
self._create()
print("上传文件成功")
def get_cache_token(self):
"""
取得缓存的token
:return:
"""
with open(self.json_path, 'r') as f:
data = json.load(f)
self.refresh_token = data['refresh_token']
self.access_token = data['access_token']
print("取得缓存的token")
print(self.refresh_token)
print(self.access_token)
f.close()
def writ_access_token(self, res):
"""
缓存token写入文件
:return:
"""
print("缓存token")
with open(self.json_path, 'wb') as f:
f.write(res)
f.close()
if __name__ == '__main__':
pan = BaiduPanHelper()
try:
pan.get_cache_token()
if pan.refresh_token != '' and pan.access_token != '':
print("有缓存token")
if not pan.get_uinfo_t():
pan.device_code_refresh_token()
else:
pan.login()
except:
pan.login()
pan.get_uinfo()
print("获取网盘目录信息")
pan.get_file_list()
|
微信
