如何使用Python向OneDrive上传大文件

嘿，如果你需要用Python往OneDrive上传大文件，推荐用Microsoft Graph API来实现，搭配MSAL库处理身份认证，requests处理HTTP请求。下面是完整的实现步骤和代码示例：

第一步：安装依赖 #

先把需要的库装上，打开终端运行：

pip install requests msal python-dotenv

第二步：配置Azure AD应用 #

你需要先在Azure AD里注册一个公共客户端应用：

• 登录Azure门户，找到“Azure Active Directory” → “应用注册” → “新注册”
• 名称随便填，支持账户类型选“仅限个人Microsoft账户”（对应代码里的consumers权限）
• 重定向URI选“Web”，填http://localhost:8000/callback，然后完成注册
• 注册好后，复制应用的“客户端ID”（就是代码里的CLIENT_ID），存到.env文件里

第三步：编写代码 #

先创建一个.env文件，内容如下：

CLIENT_ID=你的Azure应用客户端ID
USER_ID=你的OneDrive账户邮箱（可选，部分场景需要）

然后是Python代码，我把完整流程写出来，包括授权、获取token、分片上传大文件：

import requests
from msal import PublicClientApplication
from dotenv import load_dotenv
import os

# 加载环境变量
load_dotenv()

# 配置参数
CLIENT_ID = os.getenv('CLIENT_ID')
AUTHORITY_URL = 'https://login.microsoftonline.com/consumers'
SCOPES = ['Files.ReadWrite.All']
REDIRECT_URI = 'http://localhost:8000/callback'
# 要上传的本地大文件路径
LOCAL_FILE_PATH = '/path/to/your/large/file'
# OneDrive里的目标文件路径（比如根目录下的test_large_file.zip）
ONEDRIVE_TARGET_PATH = '/test_large_file.zip'

def get_access_token():
    app = PublicClientApplication(CLIENT_ID, authority=AUTHORITY_URL)
    # 先检查有没有缓存的token
    accounts = app.get_accounts()
    if accounts:
        result = app.acquire_token_silent(SCOPES, account=accounts[0])
        if result:
            return result['access_token']
    
    # 没有缓存的话，生成授权URL让用户登录
    auth_url = app.get_authorization_request_url(SCOPES, redirect_uri=REDIRECT_URI)
    print(f"请打开下面的URL登录授权：\n{auth_url}")
    print("登录后，复制浏览器地址栏里的完整URL粘贴到下面：")
    redirect_response = input("> ")
    
    # 用授权码获取token
    result = app.acquire_token_by_authorization_code(
        redirect_response,
        scopes=SCOPES,
        redirect_uri=REDIRECT_URI
    )
    
    if 'access_token' in result:
        return result['access_token']
    else:
        raise Exception(f"获取token失败：{result.get('error_description')}")

def upload_large_file_to_onedrive(token):
    # 第一步：创建上传会话
    create_session_url = f'https://graph.microsoft.com/v1.0/me/drive/root:{ONEDRIVE_TARGET_PATH}:/createUploadSession'
    headers = {'Authorization': f'Bearer {token}'}
    session_response = requests.post(create_session_url, headers=headers)
    session_response.raise_for_status()
    upload_url = session_response.json()['uploadUrl']
    
    # 第二步：分块上传文件
    chunk_size = 10 * 1024 * 1024  # 10MB每块，你可以根据需求调整
    file_size = os.path.getsize(LOCAL_FILE_PATH)
    
    with open(LOCAL_FILE_PATH, 'rb') as f:
        chunk_num = 0
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            start = chunk_num * chunk_size
            end = start + len(chunk) - 1
            if end >= file_size:
                end = file_size - 1
            
            # 设置分块上传的请求头
            chunk_headers = {
                'Content-Length': str(len(chunk)),
                'Content-Range': f'bytes {start}-{end}/{file_size}'
            }
            
            # 发送分块请求
            chunk_response = requests.put(upload_url, headers=chunk_headers, data=chunk)
            chunk_response.raise_for_status()
            
            print(f"已上传块 {chunk_num+1}，进度：{end+1}/{file_size} bytes")
            chunk_num += 1
    
    print("大文件上传完成！")

if __name__ == '__main__':
    try:
        token = get_access_token()
        upload_large_file_to_onedrive(token)
    except Exception as e:
        print(f"出错了：{e}")

一些注意事项 #

• 分块大小：代码里用的是10MB每块，Graph API允许的分块大小是5MB到60MB之间，你可以根据自己的网络情况调整
• 权限配置：确保你的Azure应用已经添加了Files.ReadWrite.All权限，并且用户登录时同意了这个权限
• token缓存：MSAL库会自动缓存token，下次运行时不需要重复登录，除非token过期或者缓存被清除
• 错误处理：代码里加了基本的异常处理，实际使用时可以根据需求扩展，比如处理网络中断后的续传

备注：内容来源于stack exchange，提问作者tealy