diff --git a/upload_images_to_immich.py b/upload_images_to_immich.py index b515446..4b706a5 100755 --- a/upload_images_to_immich.py +++ b/upload_images_to_immich.py @@ -10,7 +10,6 @@ import argparse import uuid import hashlib from pathlib import Path -from collections import defaultdict from typing import List, Dict from datetime import datetime import httpx @@ -48,14 +47,11 @@ class ImmichClient: 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() @@ -70,7 +66,6 @@ class ImmichClient: 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() @@ -80,7 +75,6 @@ class ImmichClient: 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 {} @@ -98,7 +92,7 @@ class ImmichClient: logger.debug(f"All tags found in cache: {tags}") return cached_tags - logger.info(f"Creating/retrieving tags: {missing_tags}") + logger.debug(f"Creating tags: {missing_tags}") res = self.client.put("/tags", json={"tags": missing_tags}) res.raise_for_status() @@ -191,19 +185,17 @@ def upload_image( 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") + asset = client.upload_image(image_path) + asset_id = asset["id"] + status = asset.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 + # 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: @@ -214,7 +206,6 @@ def upload_image( 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) @@ -235,25 +226,20 @@ class ParsedFilename: 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 ' -- ' +def parse_filename(image_path: Path) -> ParsedFilename | None: 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(",")] 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 None 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 = [] @@ -263,25 +249,7 @@ def list_images(folder_path: Path) -> List[Path]: 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: - 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: - 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 parse_args(): - """Parse command-line arguments.""" parser = argparse.ArgumentParser(description="Upload images to Immich, grouping them by actor name from filename") parser.add_argument("image_paths", type=Path, nargs="+", help="Path to the images to upload") return parser.parse_args() @@ -291,15 +259,12 @@ def main(): args = parse_args() image_paths: list[Path] = args.image_paths - """Main function to process and upload images.""" if not image_paths: logger.error("No images found to process") return logger.info(f"Found {len(image_paths)} images to process") - # Create _uploaded directory - # 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 @@ -309,15 +274,16 @@ def main(): immich = ImmichClient(IMMICH_SERVER_URL, IMMICH_API_KEY) for image_path in image_paths: + if not image_path.is_file(): + logger.warning(f"Skipping non-file path: {image_path}") + continue + try: - # Parse filename to extract actor names - parsed = parse_image_filename(image_path) + parsed = parse_filename(image_path) if not parsed: logger.warning(f"Skipping image with unparseable filename: {image_path.name}") continue - # Convert actor names to lowercase for albums and tags - logger.info(f"Processing {image_path.name} with actors: {', '.join(parsed.actors)}") actor_names = [actor.lower() for actor in parsed.actors] @@ -334,7 +300,7 @@ def main(): 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 + image_path.rename(dest_path) except Exception as e: logger.error(f"Error processing image {image_path.name}: {e}")