使用Python脚本通过Drive API向Google Drive上传超大文件
用Python实现Google Drive API v3超大文件可恢复上传
准备工作
- 启用Google Drive API,创建服务账号并下载
credentials.json文件;若用OAuth2授权,需创建OAuth客户端ID并下载client_secret.json。 - 若用服务账号,需将其邮箱添加到目标Google Drive文件夹的编辑权限中,确保能写入文件。
- 安装依赖包:
pip install google-api-python-client google-auth-httplib2 google-auth google-auth-oauthlib
方案一:服务账号(适合无人值守的服务器环境)
服务账号的凭据无需手动维护token有效期,API客户端会自动处理刷新,更适合长时间上传超大文件。
核心代码
import os from google.oauth2 import service_account from googleapiclient.discovery import build from googleapiclient.http import MediaFileUpload def upload_large_file(service_account_path, file_path, folder_id, chunk_size=100*1024*1024): # 加载服务账号凭据 credentials = service_account.Credentials.from_service_account_file( service_account_path, scopes=['https://www.googleapis.com/auth/drive'] ) # 构建Drive API客户端 drive_service = build('drive', 'v3', credentials=credentials) # 定义文件元数据(指定文件名和目标文件夹) file_metadata = { 'name': os.path.basename(file_path), 'parents': [folder_id] } # 创建可恢复上传媒体对象,指定分块大小 media = MediaFileUpload( file_path, chunksize=chunk_size, resumable=True, mimetype='application/zip' ) # 初始化上传会话 request = drive_service.files().create( body=file_metadata, media_body=media, fields='id' ) print(f"开始上传文件: {os.path.basename(file_path)}") response = None while response is None: status, response = request.next_chunk() if status: progress = int(status.progress() * 100) print(f"上传进度: {progress}%") print(f"上传完成!文件ID: {response.get('id')}") if __name__ == '__main__': # 替换为你的实际参数 SERVICE_ACCOUNT_JSON = '/path/to/credentials.json' LOCAL_ZIP_FILE = '/path/to/your/large/file.zip' TARGET_FOLDER_ID = 'your-google-drive-folder-id' # 从文件夹URL提取 upload_large_file(SERVICE_ACCOUNT_JSON, LOCAL_ZIP_FILE, TARGET_FOLDER_ID)
关键说明
- 分块大小:建议设为50MB-200MB,根据服务器带宽调整,太小会增加请求次数,太大易超时。
- 断点续传:上传中断后,再次运行脚本会自动从断点继续,无需重新上传整个文件。
- 权限验证:必须确保服务账号拥有目标文件夹的写入权限,否则会返回权限错误。
方案二:OAuth2授权(适合需要用户交互的场景)
若不想用服务账号,可通过OAuth2生成刷新令牌,实现token过期自动刷新。
核心代码
import os from google.oauth2.credentials import Credentials from google_auth_oauthlib.flow import InstalledAppFlow from googleapiclient.discovery import build from googleapiclient.http import MediaFileUpload SCOPES = ['https://www.googleapis.com/auth/drive'] def get_refreshed_credentials(credentials_path, token_path): creds = None # 读取已保存的token if os.path.exists(token_path): creds = Credentials.from_authorized_user_file(token_path, SCOPES) # 刷新或重新获取token if not creds or not creds.valid: if creds and creds.expired and creds.refresh_token: creds.refresh(Request()) else: flow = InstalledAppFlow.from_client_secrets_file(credentials_path, SCOPES) creds = flow.run_local_server(port=0) # 保存token供下次使用 with open(token_path, 'w') as token: token.write(creds.to_json()) return creds def upload_large_file_oauth(credentials_path, token_path, file_path, folder_id, chunk_size=100*1024*1024): creds = get_refreshed_credentials(credentials_path, token_path) drive_service = build('drive', 'v3', credentials=creds) file_metadata = { 'name': os.path.basename(file_path), 'parents': [folder_id] } media = MediaFileUpload( file_path, chunksize=chunk_size, resumable=True, mimetype='application/zip' ) request = drive_service.files().create( body=file_metadata, media_body=media, fields='id' ) print(f"开始上传文件: {os.path.basename(file_path)}") response = None while response is None: status, response = request.next_chunk() if status: progress = int(status.progress() * 100) print(f"上传进度: {progress}%") print(f"上传完成!文件ID: {response.get('id')}") if __name__ == '__main__': # 替换为你的实际参数 CLIENT_SECRET_JSON = '/path/to/client_secret.json' TOKEN_FILE = '/path/to/token.json' LOCAL_ZIP_FILE = '/path/to/your/large/file.zip' TARGET_FOLDER_ID = 'your-google-drive-folder-id' upload_large_file_oauth(CLIENT_SECRET_JSON, TOKEN_FILE, LOCAL_ZIP_FILE, TARGET_FOLDER_ID)
通用注意事项
- 服务器需保持网络稳定,可添加异常捕获和重试逻辑(比如捕获请求超时后等待10秒再继续)。
- 确保服务器磁盘空间充足,避免上传中途因磁盘满中断。
- 超大文件上传耗时久,建议用
nohup或screen将脚本放在后台运行,避免终端关闭中断进程。
内容的提问来源于stack exchange,提问作者starc52
相关产品推荐
相关产品推荐

