获取onedrive token替换alist闭源api获取的token

1
import http.server
2
import webbrowser
3
import urllib.parse
4
import requests
5
import json
6
import os
7

8
# ========== 输入区 ==========
9
CLIENT_ID = input("🔑 请输入 Azure client_id: ").strip() or " client_id"
10
CLIENT_SECRET = input("🔒 请输入 Azure client_secret: ").strip() or "client_secret"
11
ALIST_REMARK = input("📝 Alist 挂载备注（如 OneDrive）: ").strip() or "My OneDrive"
12
ALIST_PATH = input("📂 Alist 显示路径（如 /onedrive）: ").strip() or "/onedrive"
13
ONEDRIVE_ROOT = input("📁 OneDrive 根路径（如 / 或 /某个子目录）: ").strip() or "/"
14
ALIST_URL = input("📁  Alist 服务地址: ").strip() or "http://localhost:5244"
15
ALIST_USER = input("👤 Alist 管理员用户名: ").strip() or "admin"
16
ALIST_PASS = input("🔑 Alist 管理员密码: ").strip() or "123456"
17

18
# ========== 常量区 ==========
19
REDIRECT_URI = "http://localhost:53682/"
20
SCOPES = "Files.ReadWrite.All offline_access User.Read Sites.Read.All"
21
TOKEN_URL = "https://login.microsoftonline.com/common/oauth2/v2.0/token"
22
CONFIG_OUTPUT_PATH = "alist_onedrive_config.json"
23

24
auth_code = None
25

26
class OAuthHandler(http.server.BaseHTTPRequestHandler):
27
    def do_GET(self):
28
        global auth_code
29
        params = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)
30
        if "code" in params:
31
            auth_code = params["code"][0]
32
            self.send_response(200)
33
            self.send_header("Content-type", "text/html; charset=utf-8")
34
            self.end_headers()
35
            self.wfile.write("<h1>✅ 授权成功！请返回终端。</h1>".encode("utf-8"))
36
        else:
37
            self.send_response(400)
38
            self.send_header("Content-type", "text/html; charset=utf-8")
39
            self.end_headers()
40
            self.wfile.write("<h1>❌ 授权失败，未获取到 code。</h1>".encode("utf-8"))
41

42

43
def start_server():
44
    server = http.server.HTTPServer(('localhost', 53682), OAuthHandler)
45
    server.handle_request()
46

47
def get_alist_token(alist_url: str, username: str, password: str, otp: str = None) -> str:
48
    login_url = f"{alist_url}/api/auth/login"
49
    payload = {
50
        "username": username,
51
        "password": password
52
    }
53
    if otp:
54
        payload["otp_code"] = otp
55

56
    res = requests.post(login_url, json=payload)
57
    if res.status_code != 200:
58
        raise Exception("❌ 无法连接 Alist 登录接口")
59
    data = res.json()
60
    if data.get("code") != 200:
61
        raise Exception(f"❌ 登录失败：{data.get('message')}")
62
    return data["data"]["token"]
63

64
def main():
65
    print("🌐 打开浏览器进行 OneDrive 授权...")
66
    auth_url = (
67
        f"https://login.microsoftonline.com/common/oauth2/v2.0/authorize"
68
        f"?client_id={CLIENT_ID}"
69
        f"&response_type=code"
70
        f"&redirect_uri={urllib.parse.quote(REDIRECT_URI)}"
71
        f"&scope={urllib.parse.quote(SCOPES)}"
72
    )
73
    webbrowser.open(auth_url)
74
    start_server()
75

76
    if not auth_code:
77
        print("❌ 未获取到授权码。")
78
        return
79

80
    print("🔄 获取 refresh_token 中...")
81
    data = {
82
        "client_id": CLIENT_ID,
83
        "client_secret": CLIENT_SECRET,
84
        "code": auth_code,
85
        "redirect_uri": REDIRECT_URI,
86
        "grant_type": "authorization_code"
87
    }
88
    response = requests.post(TOKEN_URL, data=data)
89
    if response.status_code != 200:
90
        print("❌ 获取 token 失败：")
91
        print(response.text)
92
        return
93

94
    tokens = response.json()
95
    refresh_token = tokens.get("refresh_token")
96
    if not refresh_token:
97
        print("❌ token 响应中未找到 refresh_token。")
98
        return
99

100
    # 获取 Alist 管理 token
101
    TWO_FA_CODE = input("📲 请输入两步验证码（如启用）[可留空]: ").strip()
102
    print("🔐 正在登录 Alist...")
103
    alist_token = get_alist_token(ALIST_URL, ALIST_USER, ALIST_PASS, TWO_FA_CODE)
104

105
    # 构造符合 Alist API 要求格式的配置
106
    payload = {
107
        "mount_path": ALIST_PATH,
108
        "order": 0,
109
        "remark": ALIST_REMARK,
110
        "cache_expiration": 30,
111
        "web_proxy": False,
112
        "webdav_policy": "302_redirect",
113
        "down_proxy_url": "",
114
        "order_by": "",
115
        "order_direction": "",
116
        "extract_folder": "",
117
        "enable_sign": False,
118
        "driver": "Onedrive",
119
        "addition": json.dumps({
120
            "root_folder_path": ONEDRIVE_ROOT,
121
            "region": "global",
122
            "is_sharepoint": False,
123
            "client_id": CLIENT_ID,
124
            "client_secret": CLIENT_SECRET,
125
            "redirect_uri": REDIRECT_URI,
126
            "refresh_token": refresh_token,
127
            "site_id": "",
128
            "chunk_size": 5,
129
            "custom_host": ""
130
        }, ensure_ascii=False)
131
    }
132

133
    print("🚀 向 Alist 推送挂载配置...")
134
    api_url = f"{ALIST_URL}/api/admin/storage/create"
135
    headers = {
136
        "Content-Type": "application/json",
137
        "Authorization": alist_token
138
    }
139
    res = requests.post(api_url, json=payload, headers=headers)
140

141
    if res.status_code == 200 and res.json().get("code") == 200:
142
        print("✅ 成功添加 OneDrive 挂载！请在 Alist 后台查看。")
143
    else:
144
        print("❌ 推送失败：")
145
        print(res.text)
146

147
    # 同时写入配置文件备份
148
    print(f"💾 正在写入配置到 {CONFIG_OUTPUT_PATH}...")
149
    with open(CONFIG_OUTPUT_PATH, "w", encoding="utf-8") as f:
150
        json.dump(payload, f, ensure_ascii=False, indent=2)
151

152

153
if __name__ == "__main__":
154
    main()