diff --git a/upload_images_to_immich.py b/upload_images_to_immich.py new file mode 100755 index 0000000..b675208 --- /dev/null +++ b/upload_images_to_immich.py @@ -0,0 +1,169 @@ +#!/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)