You need to enable JavaScript to run this app.
最新活动
大模型
产品
解决方案
定价
生态与合作
支持与服务
开发者
了解我们

如何基于Python API通过URL直接将网络文件上传至Google Drive?

直接从网络链接上传文件到Google Drive(Python + Drive REST v3)

嘿,我之前刚好碰到过一模一样的需求!完全不用先把文件存本地,直接把网络文件的数据流传给Google Drive API就行,给你整理了一套可行的方案:

核心思路

你平时上传本地文件时,是把本地文件的字节流传给Drive API;那从网络链接上传的话,只需要先通过网络请求获取文件的流式数据,然后直接把这个数据流喂给API就行,全程不用写入本地磁盘。

准备工作

先确保你的Python环境装了必要的依赖包,打开终端跑这条命令:

pip install google-api-python-client google-auth-httplib2 google-auth-oauthlib requests

完整代码实现

下面是可以直接用的代码,我加了详细注释:

1. 导入依赖模块

import io
import os
import requests
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from googleapiclient.http import MediaIoBaseUpload
from google.auth.transport.requests import Request

2. 获取Drive服务实例(OAuth2认证,适配Installed Applications)

这部分和你上传本地文件的认证逻辑一致,会自动处理token的保存和刷新:

# 定义需要的API权限,drive.file足够满足上传文件的需求(更安全),如果需要全权限可以用'drive'
SCOPES = ['https://www.googleapis.com/auth/drive.file']

def get_drive_service():
    creds = None
    # token.json会自动保存你的认证信息,避免每次都要扫码
    if os.path.exists('token.json'):
        creds = Credentials.from_authorized_user_file('token.json', SCOPES)
    # 处理token过期或不存在的情况
    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
        else:
            # 这里替换成你的OAuth2客户端密钥文件(从Google Cloud控制台下载的credentials.json)
            flow = InstalledAppFlow.from_client_secrets_file(
                'credentials.json', SCOPES)
            creds = flow.run_local_server(port=0)
        # 保存token到本地
        with open('token.json', 'w') as token:
            token.write(creds.to_json())
    # 返回Drive v3服务实例
    return build('drive', 'v3', credentials=creds)

3. 核心上传函数(从网络链接直接上传)

这个函数会流式获取网络文件,然后直接传给Drive API:

def upload_from_url(drive_service, file_url, drive_filename, parent_folder_id=None):
    try:
        # 流式获取网络文件,stream=True避免一次性把大文件加载到内存
        with requests.get(file_url, stream=True) as r:
            r.raise_for_status()  # 检查网络请求是否成功
            
            # 把网络数据流写入BytesIO(内存中的流),也可以直接用r.raw更省内存
            file_stream = io.BytesIO()
            for chunk in r.iter_content(chunk_size=8192):
                file_stream.write(chunk)
            file_stream.seek(0)  # 把流指针移到开头,方便API读取
            
            # 构造Drive文件的元数据(文件名、父文件夹等)
            file_metadata = {'name': drive_filename}
            if parent_folder_id:
                file_metadata['parents'] = [parent_folder_id]
            
            # 构造媒体上传对象,自动从响应头获取MIME类型(让Drive正确识别文件类型)
            mime_type = r.headers.get('Content-Type', 'application/octet-stream')
            media = MediaIoBaseUpload(file_stream, mimetype=mime_type)
            
            # 调用Drive API上传文件
            uploaded_file = drive_service.files().create(
                body=file_metadata,
                media_body=media,
                fields='id,name'  # 指定返回的字段,减少数据量
            ).execute()
            
            print(f"✅ 文件上传成功!文件名: {uploaded_file['name']},文件ID: {uploaded_file['id']}")
            return uploaded_file['id']
    except Exception as e:
        print(f"❌ 上传失败: {str(e)}")
        return None

4. 调用示例

把你的网络文件链接、想要保存的文件名填进去就行:

if __name__ == '__main__':
    # 获取Drive服务
    drive_service = get_drive_service()
    
    # 替换成你的网络文件链接
    target_url = "https://example.com/your-target-file.jpg"
    # 替换成你想在Drive中保存的文件名
    save_name = "我的网络文件.jpg"
    # 可选:如果要上传到指定文件夹,替换成文件夹ID,否则传None
    target_folder_id = "your-folder-id-here"
    
    # 执行上传
    upload_from_url(drive_service, target_url, save_name, target_folder_id)

一些实用提示

  • 大文件优化:如果上传超大文件,建议直接用r.raw代替BytesIO,比如media = MediaIoBaseUpload(r.raw, mimetype=mime_type),记得在requests.get里加上stream=True, timeout=300,避免超时。
  • 权限选择:尽量用drive.file权限,比全drive权限更安全,它只允许访问你通过这个应用上传的文件。
  • 错误处理:可以根据需求扩展异常捕获,比如处理网络超时、API配额耗尽等情况。
  • MIME类型:如果响应头里没有返回正确的MIME类型,你可以手动指定,比如上传PDF就写mimetype='application/pdf'

内容的提问来源于stack exchange,提问作者Notagenericmember

火山引擎 最新活动