immich-extern-to-album/src/main.py

import logging
import os
import posixpath
import sys
import time
from pathlib import PurePosixPath
from typing import Any

import requests


IMMICH_URL = os.environ.get("IMMICH_URL", "").rstrip("/")
IMMICH_API_KEY = os.environ.get("IMMICH_API_KEY", "")
LIBRARY_ID = os.environ.get("IMMICH_LIBRARY_ID", "")

# Path as it appears in Immich's originalPath field.
EXTERNAL_ROOT = os.environ.get(
    "EXTERNAL_ROOT",
    "/mnt/nextcloud-photos",
).rstrip("/")

# root:
#   /mnt/nextcloud-photos/Korea_2016_07_08/photo.jpg -> Korea_2016_07_08
#
# leaf:
#   /mnt/nextcloud-photos/2026/Vacation/photo.jpg -> Vacation
#
# relative:
#   /mnt/nextcloud-photos/2026/Vacation/photo.jpg -> 2026 - Vacation
ALBUM_MODE = os.environ.get("ALBUM_MODE", "root").lower()

ALBUM_SEPARATOR = os.environ.get("ALBUM_SEPARATOR", " - ")
PAGE_SIZE = int(os.environ.get("PAGE_SIZE", "1000"))
DRY_RUN = os.environ.get("DRY_RUN", "false").lower() == "true"
SYNC_INTERVAL_SECONDS = int(os.environ.get("SYNC_INTERVAL_SECONDS", "1800"))

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)

session = requests.Session()
session.headers.update(
    {
        "x-api-key": IMMICH_API_KEY,
        "Accept": "application/json",
        "Content-Type": "application/json",
    }
)


def check_configuration() -> None:
    missing = []

    if not IMMICH_URL:
        missing.append("IMMICH_URL")

    if not IMMICH_API_KEY:
        missing.append("IMMICH_API_KEY")

    if not LIBRARY_ID:
        missing.append("IMMICH_LIBRARY_ID")

    if missing:
        raise RuntimeError(
            "The following environment variables are missing: " + ", ".join(missing)
        )

    if ALBUM_MODE not in {"root", "leaf", "relative"}:
        raise RuntimeError("ALBUM_MODE must be root, leaf, or relative")

    if not EXTERNAL_ROOT.startswith("/"):
        raise RuntimeError("EXTERNAL_ROOT must be an absolute path")

    if SYNC_INTERVAL_SECONDS < 0:
        raise RuntimeError("SYNC_INTERVAL_SECONDS must be zero or greater")


def api_request(
    method: str,
    path: str,
    *,
    json: dict[str, Any] | None = None,
) -> Any:
    url = f"{IMMICH_URL}/api{path}"

    response = session.request(
        method,
        url,
        json=json,
        timeout=120,
    )

    if not response.ok:
        raise RuntimeError(
            f"{method} {url} failed: {response.status_code} {response.text}"
        )

    if response.status_code == 204 or not response.content:
        return None

    return response.json()


def get_albums() -> dict[str, dict[str, Any]]:
    albums = api_request("GET", "/albums?isOwned=true")

    return {album["albumName"]: album for album in albums}


def create_album(album_name: str) -> dict[str, Any]:
    logging.info("Creating album: %s", album_name)

    if DRY_RUN:
        return {
            "id": f"dry-run:{album_name}",
            "albumName": album_name,
            "assets": [],
        }

    return api_request(
        "POST",
        "/albums",
        json={
            "albumName": album_name,
            "assetIds": [],
        },
    )


def search_assets(**filters: Any) -> list[dict[str, Any]]:
    assets: list[dict[str, Any]] = []
    page = 1

    while True:
        logging.info("Reading asset page %d", page)

        result = api_request(
            "POST",
            "/search/metadata",
            json={
                **filters,
                "page": page,
                "size": PAGE_SIZE,
            },
        )

        asset_page = result.get("assets", {})
        items = asset_page.get("items", [])
        assets.extend(items)

        next_page = asset_page.get("nextPage")
        if not items or next_page is None:
            break

        page = int(next_page)

    return assets


def get_album_asset_ids(album_id: str) -> set[str]:
    assets = search_assets(albumIds=[album_id])
    return {asset["id"] for asset in assets if asset.get("id")}


def add_assets_to_album(
    album_id: str,
    asset_ids: list[str],
) -> None:
    if not asset_ids:
        return

    logging.info(
        "Adding %d asset(s) to the album",
        len(asset_ids),
    )

    if DRY_RUN:
        return

    # Send smaller batches to keep the request body at a reasonable size.
    batch_size = 500

    for start in range(0, len(asset_ids), batch_size):
        batch = asset_ids[start : start + batch_size]

        api_request(
            "PUT",
            f"/albums/{album_id}/assets",
            json={
                "ids": batch,
            },
        )


def get_album_name(original_path: str) -> str | None:
    normalized_root = posixpath.normpath(EXTERNAL_ROOT)
    normalized_path = posixpath.normpath(original_path.replace("\\", "/"))

    prefix = normalized_root + "/"

    if not normalized_path.startswith(prefix):
        return None

    relative_path = normalized_path[len(prefix) :]
    relative = PurePosixPath(relative_path)

    # A file directly in EXTERNAL_ROOT has no album directory.
    if len(relative.parts) < 2:
        return None

    directory_parts = relative.parts[:-1]

    if ALBUM_MODE == "root":
        return directory_parts[0]

    if ALBUM_MODE == "leaf":
        return directory_parts[-1]

    return ALBUM_SEPARATOR.join(directory_parts)


def get_external_assets() -> list[dict[str, Any]]:
    assets = search_assets(libraryId=LIBRARY_ID)
    logging.info("%d external assets found", len(assets))
    return assets


def synchronize() -> None:
    existing_albums = get_albums()
    assets = get_external_assets()

    desired_albums: dict[str, list[str]] = {}

    for asset in assets:
        asset_id = asset.get("id")
        original_path = asset.get("originalPath")

        if not asset_id or not original_path:
            continue

        album_name = get_album_name(original_path)

        if not album_name:
            continue

        desired_albums.setdefault(album_name, []).append(asset_id)

    logging.info(
        "%d album directories derived from asset paths",
        len(desired_albums),
    )

    for album_name, desired_asset_ids in sorted(desired_albums.items()):
        album = existing_albums.get(album_name)

        if album is None:
            album = create_album(album_name)
            existing_albums[album_name] = album

        album_id = album["id"]

        if DRY_RUN:
            existing_asset_ids: set[str] = set()
        else:
            # Immich v3 album responses no longer contain the asset list.
            existing_asset_ids = get_album_asset_ids(album_id)

        missing_asset_ids = sorted(set(desired_asset_ids) - existing_asset_ids)

        if missing_asset_ids:
            logging.info(
                "Album '%s': %d new asset(s)",
                album_name,
                len(missing_asset_ids),
            )
            add_assets_to_album(album_id, missing_asset_ids)
        else:
            logging.info(
                "Album '%s' is up to date",
                album_name,
            )

    logging.info("Synchronization completed")


def main() -> int:
    check_configuration()

    while True:
        try:
            synchronize()
        except Exception:
            if SYNC_INTERVAL_SECONDS == 0:
                raise
            logging.exception("Synchronization failed; the next run will retry")

        if SYNC_INTERVAL_SECONDS == 0:
            break

        logging.info(
            "Next synchronization in %d seconds",
            SYNC_INTERVAL_SECONDS,
        )
        time.sleep(SYNC_INTERVAL_SECONDS)

    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except Exception:
        logging.exception("Synchronization failed")
        sys.exit(1)