如何通过API访问Google Workspace Shared Drive文件？权限异常排查与解决

问题描述 #

持有标准Google Shared Drive的editor权限时，通过API访问文件完全正常；但在Google Workspace环境下，即使拥有Shared Drive的manager权限，仍无法通过API访问其中文件。已排查API凭证、权限配置并重新完成认证，问题依旧，预期manager权限应具备与标准Shared Drive editor权限一致的API访问能力。

重现代码 #

import os
import openai
from pydrive.auth import GoogleAuth
from pydrive.drive import GoogleDrive
import shutil

# OpenAI API Configuration
openai.api_key = 'API-key-here'

# Supported file formats for transcription
SUPPORTED_FORMATS = ['flac', 'm4a', 'mp3', 'mp4', 'mpeg', 'mpga', 'oga', 'ogg', 'wav', 'webm']

# Function to authenticate and get an instance of Google Drive
def authenticate_google_drive():
    try:
        gauth = GoogleAuth()
        # Try to load saved credentials
        gauth.LoadCredentialsFile("mycreds.txt")
        if gauth.credentials is None:
            # Local authentication
            gauth.LocalWebserverAuth()
        elif gauth.access_token_expired:
            # Refresh expired token
            gauth.Refresh()
        else:
            # Authorize the existing token
            gauth.Authorize()
        # Save the credentials
        gauth.SaveCredentialsFile("mycreds.txt")
        return GoogleDrive(gauth)
    except Exception as e:
        raise RuntimeError(f"Failed to authenticate Google Drive: {e}")

# Authenticate and get a Google Drive instance
try:
    drive = authenticate_google_drive()
except RuntimeError as e:
    print(e)
    exit(1)

# Input and result folder IDs
input_folder_id = 'ID-here'
result_folder_id = 'ID-here'

# Function to verify the existence of a folder
def verify_folder_exists(folder_id):
    try:
        folder = drive.CreateFile({'id': folder_id})
        folder.FetchMetadata(fields='title')
        return True
    except Exception as e:
        raise RuntimeError(f"Error verifying folder with ID {folder_id}: {e}")

try:
    if not verify_folder_exists(input_folder_id):
        raise RuntimeError(f"Input folder with ID {input_folder_id} not found or access denied.")
except RuntimeError as e:
    print(e)
    exit(1)

try:
    if not verify_folder_exists(result_folder_id):
        raise RuntimeError(f"Result folder with ID {result_folder_id} not found or access denied.")
except RuntimeError as e:
    print(e)
    exit(1)

# Function to download files from the input folder and its subfolders
def download_files_from_drive(folder_id, destination_folder):
    try:
        query = f"'{folder_id}' in parents and trashed=false"
        file_list = drive.ListFile({'q': query}).GetList()
        if not os.path.exists(destination_folder):
            os.makedirs(destination_folder)
        for file in file_list:
            if file['mimeType'] == 'application/vnd.google-apps.folder':
                # Create subfolder locally and download files within it
                subfolder_path = os.path.join(destination_folder, file['title'])
                os.makedirs(subfolder_path, exist_ok=True)
                download_files_from_drive(file['id'], subfolder_path)
            else:
                file_ext = file['title'].split('.')[-1].lower()
                if file_ext in SUPPORTED_FORMATS:
                    print(f"Downloading file {file['title']}...")
                    file.GetContentFile(os.path.join(destination_folder, file['title']))
                else:
                    print(f"File {file['title']} is not in a supported format and will be skipped.")
    except Exception as e:
        raise RuntimeError(f"Error downloading files from folder with ID {folder_id}: {e}")

# Function to transcribe files using Whisper
def transcribe_files(input_folder, output_folder):
    if not os.path.exists(output_folder):
        os.makedirs(output_folder)
    for root, _, files in os.walk(input_folder):
        for filename in files:
            input_path = os.path.join(root, filename)
            output_subfolder = os.path.join(output_folder, os.path.relpath(root, input_folder))
            os.makedirs(output_subfolder, exist_ok=True)
            output_path = os.path.join(output_subfolder, os.path.splitext(filename)[0] + '.txt')
            try:
                with open(input_path, 'rb') as audio_file:
                    response = openai.Audio.transcribe("whisper-1", audio_file)
                    with open(output_path, 'w') as output_file:
                        output_file.write(response['text'])
                print(f"Transcription of {filename} saved in {output_path}")
            except openai.error.OpenAIError as oe:
                print(f"OpenAI API error transcribing {filename}: {oe}")
            except Exception as e:
                print(f"Unknown error transcribing {filename}: {e}")

# Function to upload transcribed files to the result folder
def upload_files_to_drive(folder_id, source_folder):
    try:
        for root, _, files in os.walk(source_folder):
            for filename in files:
                file_path = os.path.join(root, filename)
                rel_path = os.path.relpath(file_path, source_folder)
                parents = [{'id': folder_id}]
                # Create subfolders on Google Drive if necessary
                for subfolder in rel_path.split(os.sep)[:-1]:
                    folder_query = f"'{folder_id}' in parents and trashed=false and title='{subfolder}' and mimeType='application/vnd.google-apps.folder'"
                    folder_list = drive.ListFile({'q': folder_query}).GetList()
                    if not folder_list:
                        new_folder = drive.CreateFile({'title': subfolder, 'parents': parents, 'mimeType': 'application/vnd.google-apps.folder'})
                        new_folder.Upload()
                        folder_id = new_folder['id']
                    else:
                        folder_id = folder_list[0]['id']
                    parents = [{'id': folder_id}]
                file_drive = drive.CreateFile({'title': filename, 'parents': parents})
                file_drive.SetContentFile(file_path)
                file_drive.Upload()
                print(f"File {filename} uploaded to the result folder.")
    except Exception as e:
        raise RuntimeError(f"Error uploading file {filename} to folder with ID {folder_id}: {e}")

# Temporary paths for download and upload
temp_download_folder = './downloaded_files'
temp_transcribed_folder = './transcribed_files'

# Executing the process
try:
    print(f"Downloading files from folder {input_folder_id} to {temp_download_folder}...")
    download_files_from_drive(input_folder_id, temp_download_folder)

    print(f"Transcribing files from folder {temp_download_folder} to {temp_transcribed_folder}...")
    transcribe_files(temp_download_folder, temp_transcribed_folder)

    print(f"Uploading transcribed files to folder {result_folder_id}...")
    upload_files_to_drive(result_folder_id, temp_transcribed_folder)

    # Cleaning up temporary files
    shutil.rmtree(temp_download_folder)
    shutil.rmtree(temp_transcribed_folder)

    print("Process completed successfully!")
except RuntimeError as e:
    print(e)
    exit(1)
except Exception as e:
    print(f"Unexpected error: {e}")
    exit(1)

可能的原因及解决方案 #

1. PyDrive未启用Shared Drive支持 #

PyDrive默认不开启Workspace Shared Drive（原Team Drive）的访问支持，需显式配置：

• 修改authenticate_google_drive函数，添加Shared Drive相关设置：

def authenticate_google_drive():
    try:
        gauth = GoogleAuth()
        # 启用Shared Drive支持
        gauth.settings['enable_team_drives'] = True
        # 确保权限范围包含Drive访问权限
        gauth.settings['oauth_scope'] = [
            'https://www.googleapis.com/auth/drive',
            'https://www.googleapis.com/auth/drive.file'
        ]
        # Try to load saved credentials
        gauth.LoadCredentialsFile("mycreds.txt")
        if gauth.credentials is None:
            # Local authentication
            gauth.LocalWebserverAuth()
        elif gauth.access_token_expired:
            # Refresh expired token
            gauth.Refresh()
        else:
            # Authorize the existing token
            gauth.Authorize()
        # Save the credentials
        gauth.SaveCredentialsFile("mycreds.txt")
        return GoogleDrive(gauth)
    except Exception as e:
        raise RuntimeError(f"Failed to authenticate Google Drive: {e}")

• 或者在项目根目录的settings.yaml文件中添加配置：

enable_team_drives: true
oauth_scope:
  - https://www.googleapis.com/auth/drive
  - https://www.googleapis.com/auth/drive.file

2. API请求未添加supportsAllDrives参数 #

Workspace Shared Drive的文件查询需要显式指定supportsAllDrives=True，否则API会忽略Shared Drive内容：

• 修改verify_folder_exists函数的FetchMetadata调用：

def verify_folder_exists(folder_id):
    try:
        folder = drive.CreateFile({'id': folder_id})
        folder.FetchMetadata(fields='title', supportsAllDrives=True)
        return True
    except Exception as e:
        raise RuntimeError(f"Error verifying folder with ID {folder_id}: {e}")

• 修改download_files_from_drive中的文件查询：

file_list = drive.ListFile({'q': query, 'supportsAllDrives': True}).GetList()

• 修改upload_files_to_drive中的子文件夹查询：

folder_list = drive.ListFile({'q': folder_query, 'supportsAllDrives': True}).GetList()

3. Workspace管理员限制API访问 #

检查Workspace管理员是否设置了Drive API访问限制：

• 确认管理员未禁用第三方应用访问Shared Drive
• 确认Drive API在Workspace控制台中处于启用状态

4. 凭证权限范围不足 #

确保认证时请求的权限范围包含https://www.googleapis.com/auth/drive，避免仅使用drive.file权限（该权限仅允许访问由应用创建的文件，无法访问Shared Drive中已存在的文件）。

总结 #

优先检查PyDrive的Shared Drive配置和API请求中的supportsAllDrives参数，这是导致Workspace Shared Drive访问失败的最常见原因。若问题仍存在，需联系Workspace管理员确认权限政策，或检查凭证的权限范围是否满足要求。

内容的提问来源于stack exchange，提问作者user1