playground/upload_images_to_immich.py

#!/usr/bin/env python3
import os
import shutil
import subprocess
import tempfile
import logging
import argparse
from pathlib import Path
from collections import defaultdict
from typing import List

# Configure logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

IMMICH_SERVER_URL = "https://ph.abdus.dev"  # Replace with your Immich server URL
IMMICH_API_KEY = os.getenv("IMMICH_API_KEY")


def parse_image_filename(image_path: Path) -> List[str]:
    """Parse image filename to extract actor names."""
    # Remove extension and split by ' -- '
    filename = image_path.stem
    if " -- " in filename:
        actors_part = filename.split(" -- ")[0]
        # Split by comma and clean up whitespace
        actors = [actor.strip() for actor in actors_part.split(",")]
        return actors

    logger.warning(f"Filename doesn't match expected format: {filename}")
    return []


def list_images(folder_path: Path) -> List[Path]:
    """List all image files in the folder."""
    image_extensions = {".jpg", ".jpeg", ".png", ".gif", ".bmp", ".tiff", ".webp"}
    images = []

    for ext in image_extensions:
        images.extend(folder_path.glob(f"*{ext}"))

    return images


def group_images_by_actor(images: List[Path]) -> dict:
    """Group images by first actor name."""
    actor_groups = defaultdict(list)

    for image in images:
        actors = parse_image_filename(image)
        if actors:
            first_actor = actors[0].lower()
            actor_groups[first_actor].append(image)
            logger.debug(f"Assigned {image.name} to actor: {first_actor}")
        else:
            logger.warning(f"Skipping image with no actors: {image.name}")

    logger.info(f"Grouped images into {len(actor_groups)} actor categories")
    return actor_groups


def upload_to_immich(image_dir_path: Path, actor_name: str) -> bool:
    """Upload images to Immich using immich-go CLI."""
    api_key = IMMICH_API_KEY
    if not api_key:
        logger.error("API_KEY environment variable not found")
        return False

    try:
        cmd = [
            "immich-go",
            "upload",
            "--api-key",
            api_key,
            "--server",
            IMMICH_SERVER_URL,
            "--pause-immich-jobs=FALSE",
            "--no-ui",
            "--log-level",
            "WARN",
            "from-folder",
            str(image_dir_path),
            "--into-album",
            actor_name,
        ]

        logger.info(f"Uploading {actor_name} images to Immich...")
        proc = subprocess.run(cmd, check=True)
        logger.info(f"Successfully uploaded {actor_name} images")
        return True

    except subprocess.CalledProcessError as e:
        logger.error(f"Error uploading {actor_name} images: {e}")
        logger.error(f"Command output: {e.stdout}")
        logger.error(f"Command error: {e.stderr}")
        return False
    except Exception as e:
        logger.error(f"Unexpected error uploading {actor_name} images: {e}")
        return False


def parse_args():
    """Parse command-line arguments."""
    parser = argparse.ArgumentParser(description="Upload images to Immich, grouping them by actor name from filename")
    parser.add_argument("source_folder", help="Path to the folder containing images to upload", type=Path)
    return parser.parse_args()


def main(source_folder: Path):
    """Main function to process and upload images."""
    source_path = Path(source_folder)

    if not source_path.exists() or not source_path.is_dir():
        logger.error(f"Source folder does not exist: {source_folder}")
        return

    # Step 1: List all images
    images = list_images(source_path)
    if not images:
        logger.error("No images found in the source folder")
        return

    # Step 2: Group images by first actor
    actor_groups = group_images_by_actor(images)
    if not actor_groups:
        logger.error("No valid actor groups found")
        return

    # Create _uploaded directory
    uploaded_dir = source_path / "_uploaded"
    uploaded_dir.mkdir(exist_ok=True)

    # Step 3: Create temp folder and process each actor
    with tempfile.TemporaryDirectory() as temp_dir:
        temp_path = Path(temp_dir)
        logger.info(f"Using temporary directory: {temp_path}")

        for actor_name, actor_images in actor_groups.items():
            logger.info(f"Processing {len(actor_images)} images for actor: {actor_name}")

            # Step 4: Create actor folder and copy images
            actor_folder = temp_path / actor_name
            actor_folder.mkdir(exist_ok=True)

            try:
                for image in actor_images:
                    dest_path = actor_folder / image.name
                    shutil.copy2(image, dest_path)
                    logger.debug(f"Copied {image.name} to {actor_folder}")

                # Step 5: Upload to Immich
                success = upload_to_immich(actor_folder, actor_name)
                if success:
                    logger.info(f"Completed processing for actor: {actor_name}")
                    # Move uploaded files to _uploaded directory
                    for image in actor_images:
                        dest_path = uploaded_dir / image.name
                        shutil.move(str(image), str(dest_path))
                        logger.debug(f"Moved {image.name} to {uploaded_dir}")
                else:
                    logger.error(f"Failed to upload images for actor: {actor_name}")

            except Exception as e:
                logger.error(f"Error processing actor {actor_name}: {e}")


if __name__ == "__main__":
    args = parse_args()
    main(args.source_folder)