Import your Google Photos albums to your existing Immich library

main.py

import requests
import json
import pathlib

"""
This script is designed to import your Google Photos albums into Immich.

It expects:
- A directory containing all your Google Photos albums, exported from Google Takeout. Each album should be a directory, and each photo should be a file inside that directory.
- Your photos should ALREADY be uploaded to Immich. This script will not upload them for you, it will only try to find them in your library to add them to existing or new albums.
- An Immich instance running at BASE_URL
- An API key for your Immich instance

Images are matched by filename, and albums are matched by name. If an image is not found, it will be skipped. If an album is not found, it will be created.
"""

BASE_URL = "https://your.immich.instance.example.com"
INPUT_DIR = pathlib.Path("/home/exampleuser/Downloads/Takeout/Google Photos")
API_KEY = "YOUR-API-KEY-HERE"

HEADERS = {
    "Content-Type": "application/json",
    "Accept": "application/json",
    "x-api-key": API_KEY,
}


def main():
    # List all directories in the input dir
    albums = [album for album in INPUT_DIR.iterdir() if album.is_dir()]

    all_assets = get_all_assets()
    all_albums = get_all_albums()

    # Remember directories which yielded an error
    errored_albums = []
    successful_albums = []

    # For each directory, list all files inside. For each file, find its asset ID
    for album in albums:
        album_id = get_album_id_for_name(all_albums, album.name)
        asset_ids = [
            get_asset_id_for_filename(all_assets, file)
            for file in album.iterdir()
            if file.is_file() and file.suffix != ".json"
        ]

        print("")

        try:
            insert_in_album(album_id, asset_ids=[id for id in asset_ids if id])
            successful_albums.append(album.name)
        except Exception as e:
            print(e)
            errored_albums.append(album.name)

    print(f"Successful albums: {successful_albums}")
    print(f"Errored albums: {errored_albums}")

    print("Done!")


def create_album(album_name: str, asset_ids: list[str]):
    payload = json.dumps(
        {
            "albumName": album_name,
            "assetIds": asset_ids,
        }
    )

    print(f"[Album] Creating album {album_name}")
    print(f"[Album] Payload: {payload}")

    response = requests.request(
        "POST", f"{BASE_URL}/api/album", headers=HEADERS, data=payload
    )
    response.raise_for_status()

    return response.json()


def insert_in_album(album_id: str, asset_ids: list[str]):
    payload = json.dumps(
        {
            "ids": asset_ids,
        }
    )

    print(f"Inserting {len(asset_ids)} assets into album {album_id}")

    response = requests.request(
        "PUT", f"{BASE_URL}/api/album/{album_id}/assets", headers=HEADERS, data=payload
    )
    response.raise_for_status()

    return response.json()


def get_asset_id_for_filename(all_assets: list[dict], file: pathlib.Path) -> str | None:
    print(".", end="")

    file_without_extension = file.stem

    # Find the asset ID for the file by finding it in all_assets
    for asset in all_assets:
        if asset["originalFileName"] == file_without_extension:
            # print(f"Found {file_without_extension}")
            return asset["id"]

    # print(f"Could not find {file_without_extension}")
    return None


def get_album_id_for_name(all_albums: list[dict], album_name: str) -> str | None:
    print(f"[Album] Searching for {album_name}")

    # Find the album ID for the album by finding it in all_albums
    for album in all_albums:
        if album["albumName"] == album_name:
            print(f"[Album] Found {album_name}")
            return album["id"]

    print(f"[Album] Could not find {album_name} in all albums, creating new one")

    response = create_album(album_name, [])

    print(f"[Album] Created album {album_name}, id {response['id']}")

    return response["id"]


def get_all_assets():
    response = requests.request(
        "GET", f"{BASE_URL}/api/asset", headers=HEADERS, data={}
    )

    print(f"[Load] Found {len(response.json())} assets")

    return response.json()


def get_all_albums():
    response = requests.request(
        "GET", f"{BASE_URL}/api/album", headers=HEADERS, data={}
    )

    print(f"[Load] Found {len(response.json())} albums")

    return response.json()


if __name__ == "__main__":
    main()