Python调用Google Drive API无法导出所有子文件夹文件至CSV

需要将Google Drive主文件夹（含1.5万个一级子文件夹，总计约6.5万个文件）的所有文件名称、日期等信息导出到CSV，但运行下方Python代码后，仅获取到约1.8万个文件（均为各子文件夹中最新上传的文件），子文件夹信息能完整获取。目标为共享文件夹，暂不认为此因素有影响，需排查是否触发API限制或代码存在问题。

原运行代码 #

import httplib2
import os
from apiclient import discovery
from oauth2client import client
from oauth2client import tools
from oauth2client.file import Storage

try:
    import argparse
    flags = argparse.ArgumentParser(parents=[tools.argparser]).parse_args()
except ImportError:
    flags = None
import pandas as pd

# If modifying these scopes, delete your previously saved credentials
# at ~/.credentials/drive-python-quickstart.json

SCOPES = 'https://www.googleapis.com/auth/drive.metadata.readonly'
CLIENT_SECRET_FILE = 'credentials.json'
APPLICATION_NAME = 'Drive API Python Quickstart'

folder_id = '########' # Set to id of the parent folder you want to list (should be the content folder)
folder_list = []
all_folders = []
file_list = []

def get_credentials():
    """Gets valid user credentials from storage.
    If nothing has been stored, or if the stored credentials are invalid,
    the OAuth2 flow is completed to obtain the new credentials.
    Returns:
        Credentials, the obtained credential.
    """

    home_dir = os.path.expanduser('~')
    credential_dir = os.path.join(home_dir, '.credentials')
    if not os.path.exists(credential_dir):
        os.makedirs(credential_dir)
    credential_path = os.path.join(credential_dir,'drive-python-quickstart.json')
    store = Storage(credential_path)
    credentials = store.get()
    if not credentials or credentials.invalid:
        flow = client.flow_from_clientsecrets(CLIENT_SECRET_FILE, SCOPES)
        flow.user_agent = APPLICATION_NAME
        if flags:
            credentials = tools.run_flow(flow, store, flags)
        else:  # Needed only for compatibility with Python 2.6
            credentials = tools.run(flow, store)
        print('Storing credentials to ' + credential_path)
    return credentials


def get_root_folder(): # Gets folder list from original root folder
    credentials = get_credentials()
    http = credentials.authorize(httplib2.Http())
    service = discovery.build('drive', 'v3', http=http)
    results = service.files().list(q="mimeType = 'application/vnd.google-apps.folder' and '"+folder_id+"' in parents",
        pageSize=1000, fields="nextPageToken, files(id, mimeType)", supportsAllDrives=True, includeItemsFromAllDrives=True).execute()
    folders = results.get('files', [])
    if not folders:
        print('No folders found.')
    else:
        for folder in folders:
            id = folder.get('id')
            folder_list.append(id)

def get_all_folders(folder_list): # Creates list of all sub folder under root, keeps going until no folders underneath
    for folder in folder_list:
        additional_folders = []
        credentials = get_credentials()
        http = credentials.authorize(httplib2.Http())
        service = discovery.build('drive', 'v3', http=http)
        results = service.files().list(
            q="mimeType = 'application/vnd.google-apps.folder' and '" +folder+ "' in parents",
            pageSize=1000, fields="nextPageToken, files(id, mimeType)", supportsAllDrives=True, includeItemsFromAllDrives=True).execute()
        items = results.get('files', [])
        for item in items:
            id = item.get('id')
            additional_folders.append(id)
        if not additional_folders:
            pass
        else:
            all_folders.extend(additional_folders)
            folder_list = additional_folders
            get_all_folders(folder_list)

def merge(): # Merges sub folder list with full list
    global full_list
    full_list = all_folders + folder_list
    full_list.append(folder_id)

def get_file_list(): # Runs over each folder generating file list, for files over 1000 uses nextpagetoken to run additional requests, picks up metadata included in the request
    for folder in full_list:
        credentials = get_credentials()
        http = credentials.authorize(httplib2.Http())
        service = discovery.build('drive', 'v3', http=http)
        page_token = None
        while True:
            results = service.files().list(
                q="'" + folder + "' in parents",
                pageSize=1000, fields="nextPageToken, files(name, md5Checksum, mimeType, size, createdTime, modifiedTime, id, parents, trashed)", pageToken=page_token, supportsAllDrives=True, includeItemsFromAllDrives=True).execute()
            items = results.get('files', [])
            for item in items:
                name = item['name']
                checksum = item.get('md5Checksum')
                size = item.get('size', '-')
                id = item.get('id')
                mimeType = item.get('mimeType', '-')
                createdTime = item.get('createdTime', 'No date')
                modifiedTime = item.get('modifiedTime', 'No date')
                parents = item.get('parents')
                trashed = item.get('trashed')
                file_list.append([name, checksum, mimeType, size, createdTime, modifiedTime, id, parents, trashed])
            page_token = results.get('nextPageToken', None)
            if page_token is None:
                break
    files = pd.DataFrame(file_list,columns=['file_name','checksum_md5','mimeType','size', 'date_created', 'date_last_modified','google_id', 'google_parent_id', 'trashed'])
    files.drop(files[files['trashed'] == True].index, inplace=True) #removes files which have True listed in trashed, these are files which had been moved to the recycle bin
    foldernumbers = files['mimeType'].str.contains('application/vnd.google-apps.folder').sum()
    filenumbers = (~files['mimeType'].str.contains('application/vnd.google-apps.folder')).sum()
    print('Number of folders is: ', foldernumbers)
    print('Number of files is: ', filenumbers)
    files.to_csv('D:/GoogleAPIMetadata.csv', index=False)


if __name__ == '__main__':
    print('Collecting folder id list')
    get_root_folder()
    get_all_folders(folder_list)
    merge()
    print('Generating file metadata list')
    get_file_list()

问题排查与修复方案 #

1. 递归遍历逻辑缺陷 #

原代码get_all_folders函数存在逻辑错误：遍历一级子文件夹时，若某个子文件夹包含子目录，会将folder_list替换为该子目录列表并递归，导致后续的一级子文件夹被跳过，大量深层文件夹未被纳入扫描范围，对应文件自然无法导出。

修复后的递归函数：

def get_all_folders(service, folder_list):
    for folder in folder_list:
        page_token = None
        while True:
            results = service.files().list(
                q="mimeType = 'application/vnd.google-apps.folder' and '" +folder+ "' in parents",
                pageSize=1000, fields="nextPageToken, files(id, mimeType)", supportsAllDrives=True, includeItemsFromAllDrives=True, pageToken=page_token).execute()
            items = results.get('files', [])
            additional_folders = [item['id'] for item in items]
            if additional_folders:
                all_folders.extend(additional_folders)
                # 直接递归当前子文件夹的子目录，不修改原folder_list
                get_all_folders(service, additional_folders)
            page_token = results.get('nextPageToken')
            if not page_token:
                break

2. 文件夹列表分页遗漏 #

原代码在get_root_folder和get_all_folders中未处理nextPageToken，当文件夹数量超过1000时，仅能获取第一页数据，导致部分文件夹未被收录。修复时需添加分页循环，确保获取所有文件夹。

3. API资源浪费与速率限制 #

• 原代码每次请求都重复创建service实例，增加请求次数，易触发Google Drive API的速率限制（普通用户1000请求/100秒），导致部分请求失败。
• 查询文件时未排除文件夹类型，导致文件夹元数据被加入列表后又被过滤，浪费API配额。

优化方案：

• 在主函数中一次性创建service实例，传递给所有函数复用。
• 查询文件时直接排除文件夹类型：

  q="'" + folder + "' in parents and mimeType != 'application/vnd.google-apps.folder'"
  

完整优化代码 #

import httplib2
import os
from apiclient import discovery
from oauth2client import client
from oauth2client import tools
from oauth2client.file import Storage

try:
    import argparse
    flags = argparse.ArgumentParser(parents=[tools.argparser]).parse_args()
except ImportError:
    flags = None
import pandas as pd

SCOPES = 'https://www.googleapis.com/auth/drive.metadata.readonly'
CLIENT_SECRET_FILE = 'credentials.json'
APPLICATION_NAME = 'Drive API Python Quickstart'

folder_id = '########'
folder_list = []
all_folders = []
file_list = []

def get_credentials():
    home_dir = os.path.expanduser('~')
    credential_dir = os.path.join(home_dir, '.credentials')
    if not os.path.exists(credential_dir):
        os.makedirs(credential_dir)
    credential_path = os.path.join(credential_dir,'drive-python-quickstart.json')
    store = Storage(credential_path)
    credentials = store.get()
    if not credentials or credentials.invalid:
        flow = client.flow_from_clientsecrets(CLIENT_SECRET_FILE, SCOPES)
        flow.user_agent = APPLICATION_NAME
        if flags:
            credentials = tools.run_flow(flow, store, flags)
        else:
            credentials = tools.run(flow, store)
        print('Storing credentials to ' + credential_path)
    return credentials

def get_root_folder(service):
    page_token = None
    while True:
        results = service.files().list(q="mimeType = 'application/vnd.google-apps.folder' and '"+folder_id+"' in parents",
            pageSize=1000, fields="nextPageToken, files(id, mimeType)", supportsAllDrives=True, includeItemsFromAllDrives=True, pageToken=page_token).execute()
        folders = results.get('files', [])
        for folder in folders:
            folder_list.append(folder['id'])
        page_token = results.get('nextPageToken')
        if not page_token:
            break

def get_all_folders(service, folder_list):
    for folder in folder_list:
        page_token = None
        while True:
            results = service.files().list(
                q="mimeType = 'application/vnd.google-apps.folder' and '" +folder+ "' in parents",
                pageSize=1000, fields="nextPageToken, files(id, mimeType)", supportsAllDrives=True, includeItemsFromAllDrives=True, pageToken=page_token).execute()
            items = results.get('files', [])
            additional_folders = [item['id'] for item in items]
            if additional_folders:
                all_folders.extend(additional_folders)
                get_all_folders(service, additional_folders)
            page_token = results.get('nextPageToken')
            if not page_token:
                break

def merge():
    global full_list
    full_list = all_folders + folder_list
    full_list.append(folder_id)

def get_file_list(service):
    for folder in full_list:
        page_token = None
        while True:
            results = service.files().list(
                q="'" + folder + "' in parents and mimeType != 'application/vnd.google-apps.folder'",
                pageSize=1000, fields="nextPageToken, files(name, md5Checksum, mimeType, size, createdTime, modifiedTime, id, parents, trashed)", pageToken=page_token, supportsAllDrives=True, includeItemsFromAllDrives=True).execute()
            items = results.get('files', [])
            for item in items:
                file_list.append([
                    item['name'],
                    item.get('md5Checksum'),
                    item.get('mimeType', '-'),
                    item.get('size', '-'),
                    item.get('createdTime', 'No date'),
                    item.get('modifiedTime', 'No date'),
                    item.get('id'),
                    item.get('parents'),
                    item.get('trashed')
                ])
            page_token = results.get('nextPageToken')
            if not page_token:
                break
    files = pd.DataFrame(file_list,columns=['file_name','checksum_md5','mimeType','size', 'date_created', 'date_last_modified','google_id', 'google_parent_id', 'trashed'])
    files = files[files['trashed'] != True]
    foldernumbers = files['mimeType'].str.contains('application/vnd.google-apps.folder').sum()
    filenumbers = (~files['mimeType'].str.contains('application/vnd.google-apps.folder')).sum()
    print('Number of folders is: ', foldernumbers)
    print('Number of files is: ', filenumbers)
    files.to_csv('D:/GoogleAPIMetadata.csv', index=False)

if __name__ == '__main__':
    print('Collecting folder id list')
    credentials = get_credentials()
    http = credentials.authorize(httplib2.Http())
    service = discovery.build('drive', 'v3', http=http)
    get_root_folder(service)
    get_all_folders(service, folder_list)
    merge()
    print('Generating file metadata list')
    get_file_list(service)

额外注意事项 #

• 若仍存在文件遗漏，需检查是否有文件被标记为trashed（代码已过滤），或共享文件夹中部分文件权限不足导致无法访问。
• 若触发API速率限制，可在请求循环中添加time.sleep(0.1)延迟，避免短时间内发送过多请求。

内容的提问来源于stack exchange，提问作者Funkeh-Monkeh