125 lines
3.7 KiB
Python
125 lines
3.7 KiB
Python
from flask import Flask, request, jsonify, redirect, url_for
|
||
from config import AppID,AppKey,Secretkey,Signkey,CODE_SCOPE,REDIRECT_URI
|
||
from configToken import ACCESS_TOKEN,REFRESH_TOKEN
|
||
import requests
|
||
import urllib
|
||
app = Flask(__name__)
|
||
def update_config_token(access_token, refresh_token):
|
||
"""
|
||
更新 config.py 中的 ACCESS_TOKEN 和 REFRESH_TOKEN
|
||
"""
|
||
with open("configToken.py", "w") as f:
|
||
f.write(f'ACCESS_TOKEN = "{access_token}"\n')
|
||
f.write(f'REFRESH_TOKEN = "{refresh_token}"\n')
|
||
|
||
@app.route('/oauth_url', methods=['GET'])
|
||
def get_baidu_oauth_url():
|
||
"""
|
||
返回百度 OAuth 2.0 授权 URL
|
||
"""
|
||
base_url = "https://openapi.baidu.com/oauth/2.0/authorize"
|
||
response_type = "code"
|
||
|
||
# 构建完整 URL
|
||
oauth_url = (
|
||
f"{base_url}?"
|
||
f"response_type={response_type}&"
|
||
f"client_id={AppKey}&"
|
||
f"redirect_uri={REDIRECT_URI}&"
|
||
f"scope={CODE_SCOPE}&"
|
||
f"device_id={AppID}"
|
||
)
|
||
return jsonify({"oauth_url": oauth_url})
|
||
|
||
@app.route('/get_token', methods=['GET'])
|
||
def get_baidu_token():
|
||
"""
|
||
使用用户授权码换取 access_token
|
||
GET 参数: code
|
||
"""
|
||
code = request.args.get('code')
|
||
if not code:
|
||
return jsonify({"error": "缺少 code 参数"}), 400
|
||
|
||
headers = {
|
||
'User-Agent': 'pan.baidu.com'
|
||
}
|
||
token_url = "https://openapi.baidu.com/oauth/2.0/token"
|
||
params = {
|
||
"grant_type": "authorization_code",
|
||
"code": code,
|
||
"client_id": AppKey,
|
||
"client_secret": Secretkey,
|
||
"redirect_uri": REDIRECT_URI
|
||
}
|
||
|
||
response = requests.get(token_url, params=params,headers=headers)
|
||
data = response.json()
|
||
|
||
# 如果返回了 access_token 和 refresh_token，就更新 config.py
|
||
if "access_token" in data and "refresh_token" in data:
|
||
update_config_token(data["access_token"], data["refresh_token"])
|
||
|
||
return jsonify(data)
|
||
|
||
@app.route('/refresh_token', methods=['GET'])
|
||
def refresh_baidu_token():
|
||
"""
|
||
使用 refresh_token 刷新 access_token
|
||
GET 参数: refresh_token
|
||
"""
|
||
refresh_token = REFRESH_TOKEN
|
||
token_url = "https://openapi.baidu.com/oauth/2.0/token"
|
||
params = {
|
||
"grant_type": "refresh_token",
|
||
"refresh_token": refresh_token,
|
||
"client_id": AppKey,
|
||
"client_secret": Secretkey,
|
||
}
|
||
|
||
response = requests.get(token_url, params=params)
|
||
data = response.json()
|
||
|
||
# 如果返回了 access_token 和 refresh_token，就更新 config.py
|
||
if "access_token" in data and "refresh_token" in data:
|
||
update_config_token(data["access_token"], data["refresh_token"])
|
||
|
||
return jsonify(data)
|
||
|
||
|
||
@app.route('/list_files', methods=['GET'])
|
||
def list_files():
|
||
"""
|
||
获取百度网盘指定目录下的文件列表
|
||
GET 参数:
|
||
- access_token: 必须
|
||
- path: 必须，目录绝对路径，中文需URL编码
|
||
- order: 可选，排序字段，time/name/size
|
||
"""
|
||
path = request.args.get('path')
|
||
order = request.args.get('order', 'type') # 默认按文件类型排序
|
||
|
||
if not path:
|
||
return jsonify({"error": "缺少 access_token 或 path 参数"}), 400
|
||
BAIDU_PAN_API = "https://pan.baidu.com/rest/2.0/xpan/multimedia"
|
||
# 中文路径 URL 编码
|
||
path_encoded = urllib.parse.quote(path)
|
||
|
||
params = {
|
||
"method": "listall",
|
||
"access_token": ACCESS_TOKEN,
|
||
"path": path_encoded,
|
||
"order": order
|
||
}
|
||
|
||
response = requests.get(BAIDU_PAN_API, params=params)
|
||
try:
|
||
data = response.json()
|
||
except Exception as e:
|
||
return jsonify({"error": "返回结果不是 JSON", "detail": str(e)}), 500
|
||
|
||
return jsonify(data)
|
||
|
||
if __name__ == '__main__':
|
||
app.run(host='0.0.0.0', port=5000, debug=True)
|