upload-photos.py

#!/Users/jvkersch/miniconda3/envs/google-photos/bin/python

# Script to upload folders of photos to Google Photos
# Written with ChatGPT.
#
# Requires a Python environment with Google API packages:
# pip install google-auth google-auth-oauthlib google-auth-httplib2 requests

import os
import requests
import json
import google.auth
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
import sys

# If modifying these SCOPES, delete the file token.json.
SCOPES = ["https://www.googleapis.com/auth/photoslibrary"]


def load_credentials():
    creds = None
    if os.path.exists("token.json"):
        creds = Credentials.from_authorized_user_file("token.json", SCOPES)
    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.json", "w") as token:
            token.write(creds.to_json())
    return creds


def get_albums(creds):
    url = "https://photoslibrary.googleapis.com/v1/albums"
    headers = {
        "Authorization": f"Bearer {creds.token}",
        "Content-Type": "application/json",
    }
    response = requests.get(url, headers=headers)
    response.raise_for_status()  # Raise an exception for HTTP errors
    return response.json().get("albums", [])


def create_album(creds, album_title):
    print(f"Creating album '{album_title}'")
    url = "https://photoslibrary.googleapis.com/v1/albums"
    headers = {
        "Authorization": f"Bearer {creds.token}",
        "Content-Type": "application/json",
    }
    data = json.dumps({"album": {"title": album_title}})
    response = requests.post(url, headers=headers, data=data)
    response.raise_for_status()  # Raise an exception for HTTP errors
    return response.json()


def upload_photo(creds, file_path):
    url = "https://photoslibrary.googleapis.com/v1/uploads"
    headers = {
        "Authorization": f"Bearer {creds.token}",
        "Content-Type": "application/octet-stream",
        "X-Goog-Upload-File-Name": os.path.basename(file_path),
        "X-Goog-Upload-Protocol": "raw",
    }
    with open(file_path, "rb") as file:
        response = requests.post(url, headers=headers, data=file)
    response.raise_for_status()  # Raise an exception for HTTP errors
    return response.content.decode("utf-8")


def add_photo_to_album(creds, upload_token, album_id):
    url = "https://photoslibrary.googleapis.com/v1/mediaItems:batchCreate"
    headers = {
        "Authorization": f"Bearer {creds.token}",
        "Content-Type": "application/json",
    }
    data = json.dumps(
        {
            "albumId": album_id,
            "newMediaItems": [
                {
                    "description": "Uploaded with Google Photos API",
                    "simpleMediaItem": {"uploadToken": upload_token},
                }
            ],
        }
    )
    response = requests.post(url, headers=headers, data=data)
    response.raise_for_status()  # Raise an exception for HTTP errors
    return response.json()


def main():
    if len(sys.argv) != 2:
        print("Usage: python upload_photos.py <photos_folder>")
        sys.exit(1)

    photos_folder = sys.argv[1]
    if not os.path.isdir(photos_folder):
        print("Invalid directory path")
        sys.exit(1)

    creds = load_credentials()
    folder_name = os.path.split(os.path.abspath(photos_folder))[-1]

    # Check if album already exists
    albums = get_albums(creds)
    for album in albums:
        if album["title"] == folder_name:
            print(f"Album '{folder_name}' already exists.")
            sys.exit(1)

    # Create a new album
    album = create_album(creds, folder_name)
    album_id = album["id"]

    for file_name in os.listdir(photos_folder):
        file_path = os.path.join(photos_folder, file_name)
        if os.path.isfile(file_path):
            try:
                upload_token = upload_photo(creds, file_path)
                if upload_token:
                    add_photo_to_album(creds, upload_token, album_id)
                    print(f"Uploaded {file_name} to album {folder_name}")
            except requests.exceptions.HTTPError as err:
                print(f"Error uploading {file_name}: {err}")


if __name__ == "__main__":
    main()