From c9431a1df8a89d9023a99c2947c8045d3f7452a3 Mon Sep 17 00:00:00 2001 From: Abdussamet Kocak Date: Thu, 10 Jul 2025 04:56:53 +0300 Subject: [PATCH] refactor(immich): Use native python in favor of 3rd party CLI for image uploads --- upload_images_to_immich.py | 372 +++++++++++++++++++++++++++---------- 1 file changed, 273 insertions(+), 99 deletions(-) diff --git a/upload_images_to_immich.py b/upload_images_to_immich.py index b675208..b515446 100755 --- a/upload_images_to_immich.py +++ b/upload_images_to_immich.py @@ -1,23 +1,241 @@ -#!/usr/bin/env python3 +#!/usr/bin/env -S uv run --script +# /// script +# dependencies = ["httpx"] +# /// +import dataclasses import os -import shutil -import subprocess -import tempfile +import re import logging import argparse +import uuid +import hashlib from pathlib import Path from collections import defaultdict -from typing import List +from typing import List, Dict +from datetime import datetime +import httpx # Configure logging -logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s") +logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s: %(message)s") logger = logging.getLogger(__name__) +logging.getLogger("httpx").setLevel(logging.WARNING) 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]: +class ImmichClient: + """Client for interacting with Immich API.""" + + def __init__(self, base_url: str, api_key: str): + self.base_url = base_url.rstrip("/") + "/api" + self.api_key = api_key + self.device_uuid = str(uuid.uuid4()) + self.client = httpx.Client( + timeout=60, + headers={ + "x-api-key": self.api_key, + }, + base_url=self.base_url, + ) + self._album_id_by_name = {} + self._tag_id_by_name = {} + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.client.close() + + def get_or_create_album(self, name: str) -> str: + """Get existing album or create new one.""" + # Check cache first + try: + return self._album_id_by_name[name] + except KeyError: + pass + + # Try to find existing album + try: + res = self.client.get("/albums") + res.raise_for_status() + albums = res.json() + + for album in albums: + existing_album = album["albumName"] + self._album_id_by_name[existing_album] = album["id"] + if existing_album == name: + logger.info(f"Found existing album: {name} (ID: {album['id']})") + return album["id"] + except Exception as e: + logger.warning(f"Error searching for album: {e}") + + # Create new album + logger.info(f"Creating new album: {name}") + res = self.client.post("/albums", json={"albumName": name}) + res.raise_for_status() + album = res.json() + album_id = album["id"] + self._album_id_by_name[name] = album_id + return album_id + + def create_tags(self, tags: list[str]) -> dict[str, str]: + """Create or get tags, return mapping of tag_name -> tag_id.""" + if not tags: + return {} + + # Check cache for existing tags + cached_tags = {} + missing_tags = [] + + for tag in tags: + if tag in self._tag_id_by_name: + cached_tags[tag] = self._tag_id_by_name[tag] + else: + missing_tags.append(tag) + + if not missing_tags: + logger.debug(f"All tags found in cache: {tags}") + return cached_tags + + logger.info(f"Creating/retrieving tags: {missing_tags}") + res = self.client.put("/tags", json={"tags": missing_tags}) + res.raise_for_status() + + # Update cache with new tags + new_tags = {} + for tag in res.json(): + tag_name = tag["name"] + tag_id = tag["id"] + self._tag_id_by_name[tag_name] = tag_id + new_tags[tag_name] = tag_id + + # Combine cached and new tags + all_tags = {**cached_tags, **new_tags} + return all_tags + + def add_asset_to_album(self, album_id: str, asset_id: str) -> None: + """Add asset to album.""" + logger.debug(f"Adding asset {asset_id} to album {album_id}") + res = self.client.put(f"/albums/{album_id}/assets", json={"ids": [asset_id]}) + res.raise_for_status() + + def link_tag_to_asset(self, tag_id: str, asset_id: str) -> None: + """Link tag to asset.""" + logger.debug(f"Linking tag {tag_id} to asset {asset_id}") + res = self.client.put(f"/tags/{tag_id}/assets", json={"ids": [asset_id]}) + res.raise_for_status() + + def upload_image(self, image_path: Path) -> Dict: + """Upload single image to Immich.""" + logger.info(f"Uploading image: {image_path}") + + stat = image_path.stat() + filename = image_path.name + ext = image_path.suffix.lower() + + device_asset_id = f"{filename}-{stat.st_size}" + + file_time = datetime.fromtimestamp(stat.st_mtime).isoformat() + "Z" + + # Calculate SHA1 checksum for duplicate detection + sha1_hash = hashlib.sha1() + with image_path.open("rb") as f: + for chunk in iter(lambda: f.read(4096), b""): + sha1_hash.update(chunk) + checksum = sha1_hash.hexdigest() + + with image_path.open("rb") as f: + response = self.client.post( + "/assets", + headers={ + "x-immich-checksum": checksum, + }, + files={ + "assetData": (filename, f, "image/jpeg"), + }, + data={ + "deviceAssetId": device_asset_id, + "deviceId": self.device_uuid, + "assetType": "image", + "fileCreatedAt": file_time, + "fileModifiedAt": file_time, + "isFavorite": "false", + "fileExtension": ext, + "duration": "0", + "isReadOnly": "false", + "isArchived": "false", + }, + ) + response.raise_for_status() + return response.json() + + +def upload_image( + client: ImmichClient, + image_path: Path, + albums: list[str] | None = None, + tags: list[str] | None = None, +) -> None: + """ + Upload a single image to Immich with optional albums and tags. + + Args: + image_path: Path to the image file + albums: List of album names to add the image to + tags: List of tags to apply to the image + immich_client: ImmichClient instance to use (defaults to global instance) + """ + + albums = albums or [] + tags = tags or [] + + try: + # Upload the image + asset_response = client.upload_image(image_path) + asset_id = asset_response["id"] + status = asset_response.get("status", "uploaded") + + logger.info(f"Uploaded: {image_path.name} (status={status})") + + # Skip album/tag processing for duplicates + if status == "duplicate": + logger.info(f"Skipping album/tag processing for duplicate: {image_path.name}") + return + + # Handle albums + album_ids = [] + for album_name in albums: + try: + album_id = client.get_or_create_album(album_name) + album_ids.append(album_id) + client.add_asset_to_album(album_id, asset_id) + logger.debug(f"Added to album: {album_name}") + except Exception as e: + logger.error(f"Failed to add to album {album_name}: {e}") + + # Handle tags + if tags: + try: + tag_id_by_name = client.create_tags(tags) + for tag_name, tag_id in tag_id_by_name.items(): + client.link_tag_to_asset(tag_id, asset_id) + logger.debug(f"Applied tag: {tag_name}") + except Exception as e: + logger.error(f"Failed to apply tags: {e}") + + except Exception as e: + logger.error(f"Failed to upload {image_path}: {e}") + raise + + +@dataclasses.dataclass +class ParsedFilename: + actors: list[str] + studio: str | None = None + + +def parse_image_filename(image_path: Path) -> ParsedFilename | None: """Parse image filename to extract actor names.""" # Remove extension and split by ' -- ' filename = image_path.stem @@ -25,10 +243,13 @@ def parse_image_filename(image_path: Path) -> List[str]: actors_part = filename.split(" -- ")[0] # Split by comma and clean up whitespace actors = [actor.strip() for actor in actors_part.split(",")] - return actors + studio: str | None = None + if m := re.search(r" -- @(\S+)", filename): + studio = m.group(1) + return ParsedFilename(actors=actors, studio=studio) logger.warning(f"Filename doesn't match expected format: {filename}") - return [] + return None def list_images(folder_path: Path) -> List[Path]: @@ -47,9 +268,9 @@ def group_images_by_actor(images: List[Path]) -> dict: actor_groups = defaultdict(list) for image in images: - actors = parse_image_filename(image) - if actors: - first_actor = actors[0].lower() + parsed = parse_image_filename(image) + if parsed: + first_actor = parsed.actors[0].lower() actor_groups[first_actor].append(image) logger.debug(f"Assigned {image.name} to actor: {first_actor}") else: @@ -59,111 +280,64 @@ def group_images_by_actor(images: List[Path]) -> dict: 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) + parser.add_argument("image_paths", type=Path, nargs="+", help="Path to the images to upload") return parser.parse_args() -def main(source_folder: Path): +def main(): + args = parse_args() + image_paths: list[Path] = args.image_paths + """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}") + if not image_paths: + logger.error("No images found to process") 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 + logger.info(f"Found {len(image_paths)} images to process") # Create _uploaded directory - uploaded_dir = source_path / "_uploaded" + # if they're all in the same directory + if not all(image_paths[0].parent == path.parent for path in image_paths): + logger.error("All images must be in the same directory") + return + uploaded_dir = image_paths[0].parent / "_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}") + immich = ImmichClient(IMMICH_SERVER_URL, IMMICH_API_KEY) - for actor_name, actor_images in actor_groups.items(): - logger.info(f"Processing {len(actor_images)} images for actor: {actor_name}") + for image_path in image_paths: + try: + # Parse filename to extract actor names + parsed = parse_image_filename(image_path) + if not parsed: + logger.warning(f"Skipping image with unparseable filename: {image_path.name}") + continue - # Step 4: Create actor folder and copy images - actor_folder = temp_path / actor_name - actor_folder.mkdir(exist_ok=True) + # Convert actor names to lowercase for albums and tags - 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}") + logger.info(f"Processing {image_path.name} with actors: {', '.join(parsed.actors)}") - # 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}") + actor_names = [actor.lower() for actor in parsed.actors] + tags = [*actor_names] + if parsed.studio: + tags.append(parsed.studio.lower()) - except Exception as e: - logger.error(f"Error processing actor {actor_name}: {e}") + upload_image( + client=immich, + image_path=image_path, + albums=actor_names, + tags=tags, + ) + + dest_path = uploaded_dir / image_path.name + logger.info(f"Moving {image_path.name} to {dest_path}") + image_path.rename(dest_path) # Use rename to move the file + except Exception as e: + logger.error(f"Error processing image {image_path.name}: {e}") if __name__ == "__main__": - args = parse_args() - main(args.source_folder) + main()