使用Google APIs的Python脚本抛出状态码3和6的原因排查

错误原因分析 #

代码存在两个核心问题导致上传失败：

1. 上传数据格式错误：upload_image函数将图片二进制数据转成Base64编码后上传，但Google Photos的uploads接口要求直接提交原始二进制数据，编码后的内容无法被服务器正确解析为图片，触发状态码3（通用处理错误）和6（媒体内容无法处理）。
2. 批次数据未清空：batch_upload_images函数中new_media_items在批次循环外初始化，每次提交后未清空，导致后续批次重复提交之前的上传令牌，干扰接口正常处理。
3. （次要）create_album函数的日志打印参数错误，将album_title误写为photos_service，虽不影响功能但会导致日志混乱。

修复方案 #

1. 修正图片上传逻辑 #

移除Base64编码步骤，直接将原始二进制数据传入请求体。

2. 重置批次媒体项列表 #

在每个批次循环内初始化new_media_items，确保每次提交的都是当前批次的有效上传令牌。

3. 修正日志打印参数 #

修改后的完整代码 #

import time
from googleapiclient.discovery import build
from googleapiclient.http import MediaIoBaseDownload
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
from PIL import Image, ImageFile
import pickle
import os
import io
from bs4 import BeautifulSoup
import requests
import PySimpleGUI as sg


def create_album(photos_service, album_title):
    print(f"Start create_album with photo_service: {photos_service} album_title: {album_title}")
    # List all albums
    albums_list = photos_service.albums().list().execute()
    print(f"albums_list: {albums_list}")

    # Check if album already exists
    for album in albums_list.get('albums', []):
        if album.get('title') == album_title:
            return album  # Return the existing album

    # If album does not exist, create a new one
    create_album_body = {
        'album': {'title': album_title}
    }
    return photos_service.albums().create(body=create_album_body).execute()

def upload_image(image_path, creds):
    # Read the image file in binary mode
    with open(image_path, 'rb') as photo_file:
        binary_photo_data = photo_file.read()

    # Create the headers for the upload request
    headers = {
        'Authorization': 'Bearer ' + creds.token,
        'Content-type': 'application/octet-stream',
        'X-Goog-Upload-Protocol': 'raw',
        'X-Goog-Upload-File-Name': os.path.basename(image_path),
    }

    # Make the POST request to upload the image and get the upload token
    upload_response = requests.post(
        'https://photoslibrary.googleapis.com/v1/uploads',
        headers=headers,
        data=binary_photo_data
    )

    # Check if the upload was successful
    if upload_response.status_code == 200:
        upload_token = upload_response.text
        if upload_token:
            return upload_token
        else:
            print(f"Upload succeeded but no upload token received for image: {image_path}")
            return None
    else:
        print(f"Upload failed with status code: {upload_response.status_code} for image: {image_path}")
        return None
    

def batch_upload_images(photos_service, album_id, image_paths, creds):
    for i in range(0, len(image_paths), 30):
        batch = image_paths[i:i+30]
        new_media_items = []  # 每个批次初始化新列表
        for image_path in batch:
            # Upload the image and get the upload token
            upload_token = upload_image(image_path, creds)
            if upload_token is None:
                print(f"Failed to upload image: {image_path}")
                continue
            
            print(f"Uploaded image successfully: {image_path}")

            # Add the new media item to the list
            new_media_items.append({
                'description': 'My new photo',
                'simpleMediaItem': {'uploadToken': upload_token}
            })
            print(f"New media item added: {image_path}")

        # Create the media items in a single batch request
        if new_media_items:  # 只有当有有效媒体项时才提交
            file_metadata = {
                'albumId': album_id,
                'newMediaItems': new_media_items
            }
            response = photos_service.mediaItems().batchCreate(body=file_metadata).execute()
            print(f"Batch create response: {response}")

            for result in response.get('newMediaItemResults', []):
                status = result.get('status', {})
                print(f"Media item status: {status.get('message')} (code: {status.get('code')})")

        if i + 30 < len(image_paths):
            time.sleep(60)  # Wait for 60 seconds if there are more images to upload



if __name__ == '__main__':
    sg.theme('SandyBeach')
    ImageFile.LOAD_TRUNCATED_IMAGES = True

    layout = [
        [sg.Text('Please enter the save path, the file_id, and the album name')],
        [sg.Text('Save Path', size=(15, 1)), sg.InputText()],
        [sg.Text('File ID', size=(15, 1)), sg.InputText()],
        [sg.Text('Album Name', size=(15, 1)), sg.InputText()],
        [sg.Submit(), sg.Cancel()]
    ]

    window = sg.Window('Simple data entry window', layout)
    event, values = window.read()
    window.close()

    save_path = values[0] if values[0] is not None else 'default_save_path'
    album_name = values[2]

    if not os.path.exists('Keeps/' + save_path):
        os.makedirs('Keeps/' + save_path)

    SCOPES = ['https://www.googleapis.com/auth/photoslibrary', 'https://www.googleapis.com/auth/drive']

    creds = None
    if os.path.exists('token.pickle'):
        with open('token.pickle', 'rb') as token:
            if os.path.getsize('token.pickle') > 0:
                creds = pickle.load(token)
            else:
                creds = None

    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.json', SCOPES)
            creds = flow.run_local_server(port=0)
        with open('token.pickle', 'wb') as token:
            pickle.dump(creds, token)

    drive_service = build('drive', 'v3', credentials=creds)
    photos_service = build('photoslibrary', 'v1', credentials=creds, static_discovery=False)

    file_id = values[1]
    request = drive_service.files().export_media(fileId=file_id, mimeType='text/html')
    fh = io.BytesIO()
    downloader = MediaIoBaseDownload(fh, request)
    done = False
    while done is False:
        status, done = downloader.next_chunk()

    html_content = fh.getvalue().decode()
    soup = BeautifulSoup(html_content, 'html.parser')

    new_album = create_album(photos_service, album_name)
    print(f"New album: {new_album}")
    album_id = new_album['id']

    image_counter = 1
    image_paths = []
    for img in soup.find_all('img'):
        img_url = img['src']
        img_data = requests.get(img_url).content
        img_name = "Image" + str(image_counter)
        image_counter += 1

        img_object = Image.open(io.BytesIO(img_data))
        img_path = os.path.join('Keeps/' + save_path + '/', img_name + '.' + img_object.format)
        print(f"Saving image: {img_path}")
        img_object.save(img_path)

        image_paths.append(img_path)

        # Batch upload the images every 50 images
        if len(image_paths) >= 50:
            batch_upload_images(photos_service, album_id, image_paths, creds)
            image_paths = []
    
    if image_paths:
        batch_upload_images(photos_service, album_id, image_paths, creds)

关键修改点说明 #

• upload_image函数：移除Base64编码步骤，直接传入原始二进制数据；将creds作为参数传入，避免全局变量依赖。
• batch_upload_images函数：将new_media_items移至批次循环内初始化；添加creds参数；增加空列表判断，避免提交空请求。
• create_album函数：修正打印参数，将photos_service改为album_title。

验证步骤 #

1. 运行修改后的代码，输入保存路径、Google文档ID和相册名称。
2. 检查本地图片是否正常下载，Google相册是否成功创建。
3. 查看日志，确认上传令牌获取正常，批量创建媒体项返回成功状态。

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