#!/usr/bin/env -S uv run # /// script # requires-python = ">=3.14" # dependencies = [ # "bottle>=0.13.0", # ] # /// import argparse import functools import os import re import socket import sys import threading import time import webbrowser from pathlib import Path from datetime import datetime from typing import TypedDict import bottle IMAGE_EXTENSIONS: set[str] = {".jpg", ".jpeg", ".png", ".webp", ".webm"} class ImageGroup(TypedDict): """Type definition for grouped images.""" base: str original: Path edits: list[Path] def find_available_port() -> int: """Find an available random port.""" with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.bind(("127.0.0.1", 0)) s.listen(1) port = s.getsockname()[1] return port def is_image(path: str | Path) -> bool: """Check if file is a supported image.""" return Path(path).suffix.lower() in IMAGE_EXTENSIONS def discover_images(paths: list[str]) -> list[Path]: """Discover all images from mixed list of files and folders.""" images = [] for path_str in paths: path = Path(path_str) if path.is_file() and is_image(path): images.append(path.resolve()) elif path.is_dir(): for item in path.iterdir(): if item.is_file() and is_image(item): images.append(item.resolve()) return sorted(set(images)) def group_images(images: list[Path]) -> list[ImageGroup]: """ Group images by original + edits. Original: image.jpg Edits: image-edit1.jpg, image-edit2.jpg, etc. Returns list of dicts: {original: Path, edits: [Path, ...]} """ groups = {} for img in images: stem = img.stem # Check if this is an edit match = re.match(r"^(.+?)-edit", stem) if match: base = match.group(1) if base not in groups: groups[base] = {"original": None, "edits": []} groups[base]["edits"].append(img) else: # This is an original if stem not in groups: groups[stem] = {"original": None, "edits": []} groups[stem]["original"] = img # Filter out groups without originals, convert to list result = [] for base, group in groups.items(): if group["original"]: result.append(ImageGroup(base=base, original=group["original"], edits=sorted(group["edits"]))) return sorted(result, key=lambda g: g["original"].name) def safe_image_path(filepath: str) -> Path: """Verify image path is safe (prevent directory traversal).""" # Search for the image in all_images by filename for img in all_images: if img.name == filepath: return img raise ValueError(f"Unauthorized path: {filepath}") def hardlink_images(src_paths: list[Path | str], dest_dir: str | Path) -> None: """Hard link multiple images to destination directory.""" dest_path = Path(dest_dir) make_dirs(dest_path) for src in src_paths: src = Path(src) dest = dest_path / src.name try: # Remove existing file if present if dest.exists(): dest.unlink() os.link(src, dest) except Exception as e: print(f"Error hard linking {src} to {dest}: {e}", file=sys.stderr) raise # Global state app = bottle.Bottle() all_images = set() image_groups = [] current_group_idx = 0 last_heartbeat = datetime.now() should_exit = False picks_dir = Path() flags = [] flag_dirs = {} HTML_TEMPLATE = """ Image Culler

Keyboard Shortcuts

P Toggle pick (all edits or hover one)
←→ Navigate any image
K Skip to next pending
Flags:
1-9 Assign/unassign to flag
/ Toggle reject flag
""" @app.get("/") def index() -> str: return HTML_TEMPLATE @app.get("/api/images") def api_images(): """Return grouped images and current flag assignments.""" global image_groups, flags, flag_dirs result = [] for group in image_groups: result.append( { "base": group["base"], "original": {"name": group["original"].name, "path": str(group["original"])}, "edits": [{"name": e.name, "path": str(e)} for e in group["edits"]], } ) # Build flag assignments by checking which flag directories contain which original images flag_assignments = {} for flag, flag_dir in flag_dirs.items(): if flag_dir.exists(): for item in flag_dir.iterdir(): if item.is_file(): # Map the file name to its flag flag_assignments[item.name] = flag return {"groups": result, "flags_list": flags, "flag_assignments": flag_assignments} @app.get("/image/") def serve_image(filename: str): """Serve an image file safely.""" try: target = safe_image_path(filename) return bottle.static_file(target.name, root=str(target.parent), mimetype="image/jpeg") except ValueError: bottle.response.status = 403 return {"error": "Unauthorized path"} @app.post("/api/pick") def api_pick(): """Hard link picked images to _picks directory.""" global image_groups, current_group_idx, picks_dir # Create directory before writing make_dirs(picks_dir) data = bottle.request.json original_name = data.get("original") edits_names = data.get("edits", []) include_original = data.get("includeOriginal", False) files_to_pick = [] # Add original if requested if include_original: for img in all_images: if img.name == original_name: files_to_pick.append(img) break # Add edits by finding them in all_images for edit_name in edits_names: for img in all_images: if img.name == edit_name: files_to_pick.append(img) break try: hardlink_images(files_to_pick, picks_dir) return {"status": "ok"} except Exception as e: bottle.response.status = 500 return {"error": str(e)} @app.post("/api/unpick") def api_unpick(): """Remove picked images from _picks directory.""" global picks_dir data = bottle.request.json original_name = data.get("original") try: # Find and remove the original and any edits from picks directory if picks_dir.exists(): for item in picks_dir.iterdir(): if item.is_file() and (item.name == original_name or item.stem.startswith(original_name.rsplit(".", 1)[0] + "-edit")): item.unlink() return {"status": "ok"} except Exception as e: bottle.response.status = 500 return {"error": str(e)} @app.post("/api/assign-flag") def api_assign_flag(): """Assign an image to a flag.""" global flag_dirs, all_images data = bottle.request.json original_name = data.get("original") flag_name = data.get("flag") # Find original file original_path = None for img in all_images: if img.name == original_name: original_path = img break if not original_path or flag_name not in flag_dirs: bottle.response.status = 400 return {"error": "Invalid image or flag"} # Create directory before writing make_dirs(flag_dirs[flag_name]) try: hardlink_images([original_path], flag_dirs[flag_name]) return {"status": "ok"} except Exception as e: bottle.response.status = 500 return {"error": str(e)} @app.post("/api/unassign-flag") def api_unassign_flag(): """Remove an image from a flag.""" global flag_dirs data = bottle.request.json original_name = data.get("original") flag_name = data.get("flag") if flag_name not in flag_dirs: bottle.response.status = 400 return {"error": "Invalid flag"} try: target_file = flag_dirs[flag_name] / original_name if target_file.exists(): target_file.unlink() return {"status": "ok"} except Exception as e: bottle.response.status = 500 return {"error": str(e)} @app.post("/api/heartbeat") def api_heartbeat(): """Update last heartbeat timestamp.""" global last_heartbeat last_heartbeat = datetime.now() return {"status": "ok"} @app.post("/api/shutdown") def api_shutdown(): """Shutdown the server.""" global should_exit should_exit = True return {"status": "ok"} @functools.cache def make_dirs(path: Path) -> None: """Create directory if it doesn't exist.""" path.mkdir(parents=True, exist_ok=True) def heartbeat_monitor() -> None: """Monitor for shutdown signal.""" global should_exit while not should_exit: time.sleep(0.5) print("\nShutting down server.", file=sys.stderr) os._exit(0) def main() -> None: global all_images, image_groups, last_heartbeat, picks_dir, flags, flag_dirs parser = argparse.ArgumentParser(description="Image culling app for AI image2image transforms") parser.add_argument("paths", nargs="+", help="Image files or folders to cull") parser.add_argument("--picks-dir", default=None, help="Directory for picked images (default: $cwd/_picks)") parser.add_argument("--rejects-dir", default=None, help="Directory for rejected images (default: $cwd/_rejects)") parser.add_argument("--flags", default=None, help="Comma-separated flag names (e.g., 'a,b,c,d')") parser.add_argument("--flags-dir", default=None, help="Directory for flag subdirectories (default: $cwd)") args = parser.parse_args() # Set directories cwd = Path.cwd() picks_dir = Path(args.picks_dir or (cwd / "_picks")) rejects_dir = Path(args.rejects_dir or (cwd / "_rejects")) flags_dir = Path(args.flags_dir or cwd) # Set up flags if args.flags: flags = [f.strip() for f in args.flags.split(",")] for flag in flags: flag_dir = flags_dir / f"_picks_flag_{flag}" flag_dirs[flag] = flag_dir # Always add reject as a flag at the end flags.append("reject") flag_dirs["reject"] = rejects_dir # Discover images images = discover_images(args.paths) all_images = set(images) if not images: print("No images found in provided paths.", file=sys.stderr) sys.exit(1) # Group images image_groups = group_images(images) if not image_groups: print("No original images found (looking for files matching naming convention).", file=sys.stderr) sys.exit(1) print(f"Found {len(image_groups)} original image(s) with variations", file=sys.stderr) # Find available port port = find_available_port() url = f"http://127.0.0.1:{port}" # Start heartbeat monitor monitor_thread = threading.Thread(target=heartbeat_monitor, daemon=True) monitor_thread.start() # Open browser print(f"Opening browser at {url}", file=sys.stderr) webbrowser.open(url) # Start Flask server last_heartbeat = datetime.now() bottle.run(app, host="127.0.0.1", port=port, quiet=True) if __name__ == "__main__": main()