| … | … | @@ -0,0 +1,608 @@ |
| 1 | #!/usr/bin/env python3 | |
| 2 | """Media preprocessing for portfolio site. | |
| 3 | ||
| 4 | Handles image preprocessing, video preprocessing, and manifest generation. | |
| 5 | ||
| 6 | Usage: | |
| 7 | python3 process.py images <source_dir> --collection name | |
| 8 | python3 process.py videos <source_dir> --collection name | |
| 9 | python3 process.py manifest | |
| 10 | python3 process.py all <source_dir> --collection name | |
| 11 | """ | |
| 12 | ||
| 13 | import argparse | |
| 14 | import json | |
| 15 | import os | |
| 16 | import re | |
| 17 | import subprocess | |
| 18 | import sys | |
| 19 | ||
| 20 | from PIL import Image | |
| 21 | ||
| 22 | SUPPORTED_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".pdf", ".nef", ".psd", ".tiff", ".tif"} | |
| 23 | VIDEO_EXTENSIONS = {".mp4", ".mov", ".webm", ".avi", ".mkv"} | |
| 24 | IMAGE_EXTENSIONS_WEB = {".jpg", ".jpeg", ".png", ".gif", ".webp"} | |
| 25 | ||
| 26 | DEFAULTS = { | |
| 27 | "fullsize": 1500, | |
| 28 | "thumbnail": 400, | |
| 29 | "quality": 85, | |
| 30 | "video_thumbnail_time": 2, | |
| 31 | } | |
| 32 | ||
| 33 | ||
| 34 | def load_config(path="site.json"): | |
| 35 | """Read site.json for processing settings, fall back to defaults. | |
| 36 | ||
| 37 | Expects a JSON structure like: | |
| 38 | {"processing": {"thumbnail": 400, "fullsize": 1500, "quality": 85, "video_thumbnail_time": 2}} | |
| 39 | """ | |
| 40 | config = dict(DEFAULTS) | |
| 41 | if path and os.path.exists(path): | |
| 42 | with open(path, "r") as f: | |
| 43 | data = json.load(f) | |
| 44 | if data and "processing" in data: | |
| 45 | proc = data["processing"] | |
| 46 | for key in ("fullsize", "thumbnail", "quality", "video_thumbnail_time"): | |
| 47 | if key in proc: | |
| 48 | config[key] = int(proc[key]) | |
| 49 | return config | |
| 50 | ||
| 51 | ||
| 52 | def load_site_config(path="site.json"): | |
| 53 | """Load full site.json for manifest generation.""" | |
| 54 | if path and os.path.exists(path): | |
| 55 | with open(path, "r") as f: | |
| 56 | return json.load(f) | |
| 57 | return {} | |
| 58 | ||
| 59 | ||
| 60 | def slugify(name): | |
| 61 | """Convert a name to a URL-friendly slug. | |
| 62 | ||
| 63 | "Paint and Pixels" -> "paint-and-pixels" | |
| 64 | """ | |
| 65 | name = name.lower().strip() | |
| 66 | name = re.sub(r"[^\w\s-]", "", name) | |
| 67 | name = re.sub(r"[\s_]+", "-", name) | |
| 68 | name = re.sub(r"-+", "-", name) | |
| 69 | return name.strip("-") | |
| 70 | ||
| 71 | ||
| 72 | def resize_image(img, max_edge): | |
| 73 | """Resize image so longest edge equals max_edge. Never upscale. | |
| 74 | ||
| 75 | Returns a new Image (or the same one if no resize needed). | |
| 76 | """ | |
| 77 | w, h = img.size | |
| 78 | longest = max(w, h) | |
| 79 | ||
| 80 | if longest <= max_edge: | |
| 81 | return img.copy() | |
| 82 | ||
| 83 | ratio = max_edge / longest | |
| 84 | new_w = int(w * ratio) | |
| 85 | new_h = int(h * ratio) | |
| 86 | return img.resize((new_w, new_h), Image.LANCZOS) | |
| 87 | ||
| 88 | ||
| 89 | def _convert_to_rgb(img): | |
| 90 | """Convert RGBA, P, LA, or other modes to RGB on a black background.""" | |
| 91 | if img.mode == "RGB": | |
| 92 | return img | |
| 93 | ||
| 94 | if img.mode in ("RGBA", "LA", "PA"): | |
| 95 | background = Image.new("RGB", img.size, (0, 0, 0)) | |
| 96 | # Ensure we have RGBA for alpha composite | |
| 97 | if img.mode != "RGBA": | |
| 98 | img = img.convert("RGBA") | |
| 99 | background.paste(img, mask=img.split()[3]) | |
| 100 | return background | |
| 101 | ||
| 102 | if img.mode == "P": | |
| 103 | # Palette mode -- convert through RGBA to handle transparency | |
| 104 | rgba = img.convert("RGBA") | |
| 105 | background = Image.new("RGB", rgba.size, (0, 0, 0)) | |
| 106 | background.paste(rgba, mask=rgba.split()[3]) | |
| 107 | return background | |
| 108 | ||
| 109 | # All other modes (L, CMYK, etc.) | |
| 110 | return img.convert("RGB") | |
| 111 | ||
| 112 | ||
| 113 | def process_file(src, dst_dir, thumbs_dir, fullsize=1500, thumbnail=400, quality=85): | |
| 114 | """Process a single image file. | |
| 115 | ||
| 116 | - Converts to JPEG | |
| 117 | - Resizes to fullsize (longest edge) | |
| 118 | - Creates thumbnail | |
| 119 | - Skips if output exists and is newer than source | |
| 120 | - Never upscales | |
| 121 | """ | |
| 122 | basename = os.path.splitext(os.path.basename(src))[0] | |
| 123 | out_name = basename + ".jpg" | |
| 124 | fullres_path = os.path.join(dst_dir, out_name) | |
| 125 | thumb_path = os.path.join(thumbs_dir, out_name) | |
| 126 | ||
| 127 | src_mtime = os.path.getmtime(src) | |
| 128 | ||
| 129 | # Skip if both outputs exist and are newer than source | |
| 130 | if (os.path.exists(fullres_path) and os.path.exists(thumb_path) | |
| 131 | and os.path.getmtime(fullres_path) >= src_mtime | |
| 132 | and os.path.getmtime(thumb_path) >= src_mtime): | |
| 133 | return False # skipped | |
| 134 | ||
| 135 | try: | |
| 136 | img = Image.open(src) | |
| 137 | img.load() # force load (needed for some formats) | |
| 138 | except Exception as e: | |
| 139 | print(f" WARNING: Could not open {src}: {e}", file=sys.stderr) | |
| 140 | return False | |
| 141 | ||
| 142 | img = _convert_to_rgb(img) | |
| 143 | ||
| 144 | # Full-res | |
| 145 | full_img = resize_image(img, fullsize) | |
| 146 | full_img.save(fullres_path, "JPEG", quality=quality, optimize=True) | |
| 147 | ||
| 148 | # Thumbnail | |
| 149 | thumb_img = resize_image(img, thumbnail) | |
| 150 | thumb_img.save(thumb_path, "JPEG", quality=quality, optimize=True) | |
| 151 | ||
| 152 | return True # processed | |
| 153 | ||
| 154 | ||
| 155 | def process_collection(src_dir, dst_dir, fullsize=1500, thumbnail=400, quality=85, | |
| 156 | root_only=False): | |
| 157 | """Process all images in src_dir, writing outputs to dst_dir. | |
| 158 | ||
| 159 | Args: | |
| 160 | src_dir: Source directory containing images. | |
| 161 | dst_dir: Destination directory for processed images. | |
| 162 | fullsize: Maximum edge size for full-res images. | |
| 163 | thumbnail: Maximum edge size for thumbnails. | |
| 164 | quality: JPEG quality (1-100). | |
| 165 | root_only: If True, only process files in the root of src_dir (no subdirs). | |
| 166 | """ | |
| 167 | os.makedirs(dst_dir, exist_ok=True) | |
| 168 | thumbs_dir = os.path.join(dst_dir, "thumbs") | |
| 169 | os.makedirs(thumbs_dir, exist_ok=True) | |
| 170 | ||
| 171 | processed = 0 | |
| 172 | skipped = 0 | |
| 173 | ||
| 174 | if root_only: | |
| 175 | entries = [os.path.join(src_dir, f) for f in sorted(os.listdir(src_dir)) | |
| 176 | if os.path.isfile(os.path.join(src_dir, f))] | |
| 177 | else: | |
| 178 | entries = [] | |
| 179 | for dirpath, _dirnames, filenames in os.walk(src_dir): | |
| 180 | for fname in sorted(filenames): | |
| 181 | entries.append(os.path.join(dirpath, fname)) | |
| 182 | ||
| 183 | for filepath in entries: | |
| 184 | ext = os.path.splitext(filepath)[1].lower() | |
| 185 | if ext and ext not in SUPPORTED_EXTENSIONS: | |
| 186 | continue # has an extension but not a supported one | |
| 187 | ||
| 188 | if process_file(filepath, dst_dir, thumbs_dir, fullsize, thumbnail, quality): | |
| 189 | processed += 1 | |
| 190 | print(f" Processed: {os.path.basename(filepath)}") | |
| 191 | else: | |
| 192 | skipped += 1 | |
| 193 | ||
| 194 | return processed, skipped | |
| 195 | ||
| 196 | ||
| 197 | # --------------------------------------------------------------------------- | |
| 198 | # Video processing | |
| 199 | # --------------------------------------------------------------------------- | |
| 200 | ||
| 201 | def process_video_file(src, dst_dir, thumbs_dir, thumbnail_time=2): | |
| 202 | """Process a single video: faststart MP4 + poster frame thumbnail. | |
| 203 | ||
| 204 | - Transcodes to h264+aac MP4 with -movflags +faststart | |
| 205 | - Extracts poster frame as JPEG thumbnail | |
| 206 | - Skips if output exists and is newer than source | |
| 207 | ||
| 208 | Returns True if processed, False if skipped. | |
| 209 | """ | |
| 210 | basename = os.path.splitext(os.path.basename(src))[0] | |
| 211 | out_name = basename + ".mp4" | |
| 212 | thumb_name = basename + ".jpg" | |
| 213 | video_path = os.path.join(dst_dir, out_name) | |
| 214 | thumb_path = os.path.join(thumbs_dir, thumb_name) | |
| 215 | ||
| 216 | src_mtime = os.path.getmtime(src) | |
| 217 | ||
| 218 | # Skip if both outputs exist and are newer than source | |
| 219 | if (os.path.exists(video_path) and os.path.exists(thumb_path) | |
| 220 | and os.path.getmtime(video_path) >= src_mtime | |
| 221 | and os.path.getmtime(thumb_path) >= src_mtime): | |
| 222 | return False # skipped | |
| 223 | ||
| 224 | os.makedirs(dst_dir, exist_ok=True) | |
| 225 | os.makedirs(thumbs_dir, exist_ok=True) | |
| 226 | ||
| 227 | # Transcode to h264+aac with faststart | |
| 228 | try: | |
| 229 | subprocess.run([ | |
| 230 | "ffmpeg", "-y", "-i", src, | |
| 231 | "-c:v", "libx264", "-preset", "medium", "-crf", "23", | |
| 232 | "-c:a", "aac", "-b:a", "128k", | |
| 233 | "-movflags", "+faststart", | |
| 234 | video_path | |
| 235 | ], capture_output=True, check=True) | |
| 236 | except subprocess.CalledProcessError as e: | |
| 237 | print(f" WARNING: Could not transcode {src}: {e.stderr.decode()}", file=sys.stderr) | |
| 238 | return False | |
| 239 | ||
| 240 | # Extract poster frame as thumbnail | |
| 241 | # Try at the requested time first, fall back to first frame | |
| 242 | for seek_args in [["-ss", str(thumbnail_time)], []]: | |
| 243 | try: | |
| 244 | subprocess.run( | |
| 245 | ["ffmpeg", "-y", "-i", src] + seek_args + | |
| 246 | ["-vframes", "1", "-q:v", "2", thumb_path], | |
| 247 | capture_output=True, check=True | |
| 248 | ) | |
| 249 | except subprocess.CalledProcessError: | |
| 250 | pass | |
| 251 | if os.path.exists(thumb_path) and os.path.getsize(thumb_path) > 0: | |
| 252 | break | |
| 253 | else: | |
| 254 | if not (os.path.exists(thumb_path) and os.path.getsize(thumb_path) > 0): | |
| 255 | print(f" WARNING: Could not extract thumbnail from {src}", file=sys.stderr) | |
| 256 | ||
| 257 | return True # processed | |
| 258 | ||
| 259 | ||
| 260 | def get_video_duration(path): | |
| 261 | """Get video duration in seconds using ffprobe. | |
| 262 | ||
| 263 | Returns duration as a float, or 0.0 if it cannot be determined. | |
| 264 | """ | |
| 265 | try: | |
| 266 | result = subprocess.run([ | |
| 267 | "ffprobe", "-v", "error", | |
| 268 | "-show_entries", "format=duration", | |
| 269 | "-of", "default=noprint_wrappers=1:nokey=1", | |
| 270 | path | |
| 271 | ], capture_output=True, text=True, check=True) | |
| 272 | return float(result.stdout.strip()) | |
| 273 | except (subprocess.CalledProcessError, ValueError): | |
| 274 | return 0.0 | |
| 275 | ||
| 276 | ||
| 277 | def process_video_collection(src_dir, dst_dir, thumbnail_time=2): | |
| 278 | """Process all videos in src_dir. Returns (processed, skipped).""" | |
| 279 | os.makedirs(dst_dir, exist_ok=True) | |
| 280 | thumbs_dir = os.path.join(dst_dir, "thumbs") | |
| 281 | os.makedirs(thumbs_dir, exist_ok=True) | |
| 282 | ||
| 283 | processed = 0 | |
| 284 | skipped = 0 | |
| 285 | ||
| 286 | for filepath in sorted( | |
| 287 | os.path.join(src_dir, f) for f in os.listdir(src_dir) | |
| 288 | if os.path.isfile(os.path.join(src_dir, f)) | |
| 289 | ): | |
| 290 | ext = os.path.splitext(filepath)[1].lower() | |
| 291 | if ext not in VIDEO_EXTENSIONS: | |
| 292 | continue | |
| 293 | ||
| 294 | if process_video_file(filepath, dst_dir, thumbs_dir, thumbnail_time): | |
| 295 | processed += 1 | |
| 296 | print(f" Processed: {os.path.basename(filepath)}") | |
| 297 | else: | |
| 298 | skipped += 1 | |
| 299 | ||
| 300 | return processed, skipped | |
| 301 | ||
| 302 | ||
| 303 | # --------------------------------------------------------------------------- | |
| 304 | # Manifest generation | |
| 305 | # --------------------------------------------------------------------------- | |
| 306 | ||
| 307 | def generate_manifest(base_dir, site_config): | |
| 308 | """Scan images/ and videos/ directories, generate manifest dict. | |
| 309 | ||
| 310 | - Scans images/<collection>/ for image files, checks for .md sidecars | |
| 311 | - Scans videos/<collection>/ for .mp4 files, gets duration | |
| 312 | - Orders collections per site_config['collections']['order'] | |
| 313 | - Ignores collections in site_config['collections']['ignore'] | |
| 314 | ||
| 315 | Returns: {'collections': [...], 'items': [...]} | |
| 316 | """ | |
| 317 | images_dir = os.path.join(base_dir, "images") | |
| 318 | videos_dir = os.path.join(base_dir, "videos") | |
| 319 | ||
| 320 | collections_config = site_config.get("collections", {}) | |
| 321 | order = collections_config.get("order", []) | |
| 322 | ignore = set(collections_config.get("ignore", [])) | |
| 323 | # Always ignore thumbs directories | |
| 324 | ignore.add("thumbs") | |
| 325 | ||
| 326 | # Discover all collection names from images/ and videos/ | |
| 327 | collection_names = set() | |
| 328 | ||
| 329 | if os.path.isdir(images_dir): | |
| 330 | for name in os.listdir(images_dir): | |
| 331 | if os.path.isdir(os.path.join(images_dir, name)) and name not in ignore: | |
| 332 | collection_names.add(name) | |
| 333 | ||
| 334 | if os.path.isdir(videos_dir): | |
| 335 | for name in os.listdir(videos_dir): | |
| 336 | if os.path.isdir(os.path.join(videos_dir, name)) and name not in ignore: | |
| 337 | collection_names.add(name) | |
| 338 | ||
| 339 | # Sort: ordered collections first, then remaining alphabetically | |
| 340 | ordered = [c for c in order if c in collection_names] | |
| 341 | remaining = sorted(collection_names - set(ordered)) | |
| 342 | sorted_collections = ordered + remaining | |
| 343 | ||
| 344 | collections = [] | |
| 345 | items = [] | |
| 346 | ||
| 347 | for coll_name in sorted_collections: | |
| 348 | coll_images_dir = os.path.join(images_dir, coll_name) | |
| 349 | coll_videos_dir = os.path.join(videos_dir, coll_name) | |
| 350 | coll_items = [] | |
| 351 | ||
| 352 | # Scan images | |
| 353 | if os.path.isdir(coll_images_dir): | |
| 354 | for fname in sorted(os.listdir(coll_images_dir)): | |
| 355 | if fname == "thumbs" or fname.endswith(".md"): | |
| 356 | continue | |
| 357 | ext = os.path.splitext(fname)[1].lower() | |
| 358 | if ext not in IMAGE_EXTENSIONS_WEB: | |
| 359 | continue | |
| 360 | ||
| 361 | basename = os.path.splitext(fname)[0] | |
| 362 | item = { | |
| 363 | "type": "image", | |
| 364 | "collection": coll_name, | |
| 365 | "filename": fname, | |
| 366 | "path": f"images/{coll_name}/{fname}", | |
| 367 | "thumb": f"images/{coll_name}/thumbs/{fname}", | |
| 368 | } | |
| 369 | ||
| 370 | # Check for .md sidecar | |
| 371 | md_path = os.path.join(coll_images_dir, basename + ".md") | |
| 372 | if os.path.exists(md_path): | |
| 373 | with open(md_path, "r") as f: | |
| 374 | item["sidecar"] = f.read().strip() | |
| 375 | ||
| 376 | coll_items.append(item) | |
| 377 | ||
| 378 | # Scan videos | |
| 379 | if os.path.isdir(coll_videos_dir): | |
| 380 | for fname in sorted(os.listdir(coll_videos_dir)): | |
| 381 | if fname == "thumbs": | |
| 382 | continue | |
| 383 | ext = os.path.splitext(fname)[1].lower() | |
| 384 | if ext != ".mp4": | |
| 385 | continue | |
| 386 | ||
| 387 | basename = os.path.splitext(fname)[0] | |
| 388 | video_path = os.path.join(coll_videos_dir, fname) | |
| 389 | duration = get_video_duration(video_path) | |
| 390 | ||
| 391 | item = { | |
| 392 | "type": "video", | |
| 393 | "collection": coll_name, | |
| 394 | "filename": fname, | |
| 395 | "path": f"videos/{coll_name}/{fname}", | |
| 396 | "thumb": f"videos/{coll_name}/thumbs/{basename}.jpg", | |
| 397 | "duration": round(duration, 2), | |
| 398 | } | |
| 399 | coll_items.append(item) | |
| 400 | ||
| 401 | image_count = sum(1 for i in coll_items if i["type"] == "image") | |
| 402 | video_count = sum(1 for i in coll_items if i["type"] == "video") | |
| 403 | ||
| 404 | collections.append({ | |
| 405 | "name": coll_name, | |
| 406 | "images": image_count, | |
| 407 | "videos": video_count, | |
| 408 | "total": len(coll_items), | |
| 409 | }) | |
| 410 | ||
| 411 | items.extend(coll_items) | |
| 412 | ||
| 413 | return {"collections": collections, "items": items} | |
| 414 | ||
| 415 | ||
| 416 | def write_manifest(base_dir, site_config): | |
| 417 | """Generate and write manifest.json to base_dir.""" | |
| 418 | manifest = generate_manifest(base_dir, site_config) | |
| 419 | manifest_path = os.path.join(base_dir, "manifest.json") | |
| 420 | with open(manifest_path, "w") as f: | |
| 421 | json.dump(manifest, f, indent=2) | |
| 422 | f.write("\n") | |
| 423 | total_items = len(manifest["items"]) | |
| 424 | total_colls = len(manifest["collections"]) | |
| 425 | print(f"Manifest written: {total_colls} collections, {total_items} items -> {manifest_path}") | |
| 426 | return manifest | |
| 427 | ||
| 428 | ||
| 429 | # --------------------------------------------------------------------------- | |
| 430 | # CLI | |
| 431 | # --------------------------------------------------------------------------- | |
| 432 | ||
| 433 | def main(): | |
| 434 | parser = argparse.ArgumentParser( | |
| 435 | description="Media preprocessing for portfolio site." | |
| 436 | ) | |
| 437 | subparsers = parser.add_subparsers(dest="command", help="Processing command") | |
| 438 | ||
| 439 | # images subcommand | |
| 440 | images_parser = subparsers.add_parser("images", help="Process images") | |
| 441 | images_parser.add_argument("source_dir", help="Source directory containing raw images") | |
| 442 | images_parser.add_argument("--collection", default=None, | |
| 443 | help="Process source_dir as a single named collection") | |
| 444 | images_parser.add_argument("--root-only", action="store_true", | |
| 445 | help="Only process files in the root of source_dir (no subdirs)") | |
| 446 | images_parser.add_argument("--output", default="images", | |
| 447 | help="Base output directory (default: images)") | |
| 448 | ||
| 449 | # videos subcommand | |
| 450 | videos_parser = subparsers.add_parser("videos", help="Process videos") | |
| 451 | videos_parser.add_argument("source_dir", help="Source directory containing raw videos") | |
| 452 | videos_parser.add_argument("--collection", default=None, | |
| 453 | help="Process source_dir as a single named collection") | |
| 454 | videos_parser.add_argument("--output", default="videos", | |
| 455 | help="Base output directory (default: videos)") | |
| 456 | ||
| 457 | # manifest subcommand | |
| 458 | subparsers.add_parser("manifest", help="Generate manifest.json from existing media") | |
| 459 | ||
| 460 | # all subcommand | |
| 461 | all_parser = subparsers.add_parser("all", help="Process images + videos, then generate manifest") | |
| 462 | all_parser.add_argument("source_dir", help="Source directory containing raw media") | |
| 463 | all_parser.add_argument("--collection", default=None, | |
| 464 | help="Process source_dir as a single named collection") | |
| 465 | all_parser.add_argument("--root-only", action="store_true", | |
| 466 | help="Only process image files in the root of source_dir") | |
| 467 | ||
| 468 | args = parser.parse_args() | |
| 469 | ||
| 470 | if not args.command: | |
| 471 | parser.print_help() | |
| 472 | sys.exit(1) | |
| 473 | ||
| 474 | config = load_config("site.json") | |
| 475 | site_config = load_site_config("site.json") | |
| 476 | fullsize = config["fullsize"] | |
| 477 | thumbnail = config["thumbnail"] | |
| 478 | quality = config["quality"] | |
| 479 | video_thumbnail_time = config["video_thumbnail_time"] | |
| 480 | ||
| 481 | if args.command == "images": | |
| 482 | if not os.path.isdir(args.source_dir): | |
| 483 | print(f"Error: {args.source_dir} is not a directory", file=sys.stderr) | |
| 484 | sys.exit(1) | |
| 485 | ||
| 486 | if args.collection: | |
| 487 | slug = slugify(args.collection) | |
| 488 | dst_dir = os.path.join(args.output, slug) | |
| 489 | print(f"Processing image collection '{args.collection}' -> {dst_dir}") | |
| 490 | processed, skipped = process_collection( | |
| 491 | args.source_dir, dst_dir, fullsize, thumbnail, quality, | |
| 492 | root_only=args.root_only | |
| 493 | ) | |
| 494 | print(f"Done: {processed} processed, {skipped} skipped") | |
| 495 | else: | |
| 496 | total_processed = 0 | |
| 497 | total_skipped = 0 | |
| 498 | for entry in sorted(os.listdir(args.source_dir)): | |
| 499 | subdir = os.path.join(args.source_dir, entry) | |
| 500 | if not os.path.isdir(subdir): | |
| 501 | continue | |
| 502 | slug = slugify(entry) | |
| 503 | dst_dir = os.path.join(args.output, slug) | |
| 504 | print(f"Processing image collection '{entry}' -> {dst_dir}") | |
| 505 | processed, skipped = process_collection( | |
| 506 | subdir, dst_dir, fullsize, thumbnail, quality | |
| 507 | ) | |
| 508 | total_processed += processed | |
| 509 | total_skipped += skipped | |
| 510 | print(f" {processed} processed, {skipped} skipped") | |
| 511 | print(f"\nTotal: {total_processed} processed, {total_skipped} skipped") | |
| 512 | ||
| 513 | # Auto-regenerate manifest | |
| 514 | write_manifest(".", site_config) | |
| 515 | ||
| 516 | elif args.command == "videos": | |
| 517 | if not os.path.isdir(args.source_dir): | |
| 518 | print(f"Error: {args.source_dir} is not a directory", file=sys.stderr) | |
| 519 | sys.exit(1) | |
| 520 | ||
| 521 | if args.collection: | |
| 522 | slug = slugify(args.collection) | |
| 523 | dst_dir = os.path.join(args.output, slug) | |
| 524 | print(f"Processing video collection '{args.collection}' -> {dst_dir}") | |
| 525 | processed, skipped = process_video_collection( | |
| 526 | args.source_dir, dst_dir, video_thumbnail_time | |
| 527 | ) | |
| 528 | print(f"Done: {processed} processed, {skipped} skipped") | |
| 529 | else: | |
| 530 | total_processed = 0 | |
| 531 | total_skipped = 0 | |
| 532 | for entry in sorted(os.listdir(args.source_dir)): | |
| 533 | subdir = os.path.join(args.source_dir, entry) | |
| 534 | if not os.path.isdir(subdir): | |
| 535 | continue | |
| 536 | slug = slugify(entry) | |
| 537 | dst_dir = os.path.join(args.output, slug) | |
| 538 | print(f"Processing video collection '{entry}' -> {dst_dir}") | |
| 539 | processed, skipped = process_video_collection( | |
| 540 | subdir, dst_dir, video_thumbnail_time | |
| 541 | ) | |
| 542 | total_processed += processed | |
| 543 | total_skipped += skipped | |
| 544 | print(f" {processed} processed, {skipped} skipped") | |
| 545 | print(f"\nTotal: {total_processed} processed, {total_skipped} skipped") | |
| 546 | ||
| 547 | # Auto-regenerate manifest | |
| 548 | write_manifest(".", site_config) | |
| 549 | ||
| 550 | elif args.command == "manifest": | |
| 551 | write_manifest(".", site_config) | |
| 552 | ||
| 553 | elif args.command == "all": | |
| 554 | if not os.path.isdir(args.source_dir): | |
| 555 | print(f"Error: {args.source_dir} is not a directory", file=sys.stderr) | |
| 556 | sys.exit(1) | |
| 557 | ||
| 558 | if args.collection: | |
| 559 | slug = slugify(args.collection) | |
| 560 | ||
| 561 | # Process images | |
| 562 | img_dst = os.path.join("images", slug) | |
| 563 | print(f"Processing image collection '{args.collection}' -> {img_dst}") | |
| 564 | img_processed, img_skipped = process_collection( | |
| 565 | args.source_dir, img_dst, fullsize, thumbnail, quality, | |
| 566 | root_only=args.root_only | |
| 567 | ) | |
| 568 | print(f" Images: {img_processed} processed, {img_skipped} skipped") | |
| 569 | ||
| 570 | # Process videos | |
| 571 | vid_dst = os.path.join("videos", slug) | |
| 572 | print(f"Processing video collection '{args.collection}' -> {vid_dst}") | |
| 573 | vid_processed, vid_skipped = process_video_collection( | |
| 574 | args.source_dir, vid_dst, video_thumbnail_time | |
| 575 | ) | |
| 576 | print(f" Videos: {vid_processed} processed, {vid_skipped} skipped") | |
| 577 | else: | |
| 578 | total_img_processed = 0 | |
| 579 | total_img_skipped = 0 | |
| 580 | total_vid_processed = 0 | |
| 581 | total_vid_skipped = 0 | |
| 582 | for entry in sorted(os.listdir(args.source_dir)): | |
| 583 | subdir = os.path.join(args.source_dir, entry) | |
| 584 | if not os.path.isdir(subdir): | |
| 585 | continue | |
| 586 | slug = slugify(entry) | |
| 587 | ||
| 588 | img_dst = os.path.join("images", slug) | |
| 589 | print(f"Processing '{entry}' images -> {img_dst}") | |
| 590 | ip, isk = process_collection(subdir, img_dst, fullsize, thumbnail, quality) | |
| 591 | total_img_processed += ip | |
| 592 | total_img_skipped += isk | |
| 593 | ||
| 594 | vid_dst = os.path.join("videos", slug) | |
| 595 | print(f"Processing '{entry}' videos -> {vid_dst}") | |
| 596 | vp, vsk = process_video_collection(subdir, vid_dst, video_thumbnail_time) | |
| 597 | total_vid_processed += vp | |
| 598 | total_vid_skipped += vsk | |
| 599 | ||
| 600 | print(f"\nTotal images: {total_img_processed} processed, {total_img_skipped} skipped") | |
| 601 | print(f"Total videos: {total_vid_processed} processed, {total_vid_skipped} skipped") | |
| 602 | ||
| 603 | # Auto-regenerate manifest | |
| 604 | write_manifest(".", site_config) | |
| 605 | ||
| 606 | ||
| 607 | if __name__ == "__main__": | |
| 608 | main() |
| … | … | @@ -0,0 +1,192 @@ |
| 1 | #!/usr/bin/env python3 | |
| 2 | """Rename images or collections and keep everything in sync. | |
| 3 | ||
| 4 | Usage: | |
| 5 | python3 rename.py image <collection> <old-name> <new-name> | |
| 6 | python3 rename.py collection <old-name> <new-name> | |
| 7 | python3 rename.py list [collection] | |
| 8 | ||
| 9 | Examples: | |
| 10 | python3 rename.py list | |
| 11 | python3 rename.py list in-dreams | |
| 12 | python3 rename.py image in-dreams "1009494620_364f90e07e_o" "tangled-roots" | |
| 13 | python3 rename.py collection in-dreams "dreams" | |
| 14 | ||
| 15 | Names are without extensions -- the script handles .jpg and .md automatically. | |
| 16 | """ | |
| 17 | ||
| 18 | import json | |
| 19 | import os | |
| 20 | import sys | |
| 21 | ||
| 22 | BASE = os.path.dirname(os.path.abspath(__file__)) | |
| 23 | IMAGES_DIR = os.path.join(BASE, "images") | |
| 24 | CONFIG_PATH = os.path.join(BASE, "site.json") | |
| 25 | ||
| 26 | ||
| 27 | def load_config(): | |
| 28 | if os.path.exists(CONFIG_PATH): | |
| 29 | with open(CONFIG_PATH, "r") as f: | |
| 30 | return json.load(f) | |
| 31 | return {} | |
| 32 | ||
| 33 | ||
| 34 | def save_config(config): | |
| 35 | with open(CONFIG_PATH, "w") as f: | |
| 36 | json.dump(config, f, indent=2) | |
| 37 | f.write("\n") | |
| 38 | ||
| 39 | ||
| 40 | def list_collections(): | |
| 41 | ignore = {"old", "_archive", "thumbs"} | |
| 42 | # Also read ignore list from site.json | |
| 43 | config = load_config() | |
| 44 | collections_config = config.get("collections", {}) | |
| 45 | ignore.update(collections_config.get("ignore", [])) | |
| 46 | ||
| 47 | collections = [] | |
| 48 | if not os.path.isdir(IMAGES_DIR): | |
| 49 | return collections | |
| 50 | for name in sorted(os.listdir(IMAGES_DIR)): | |
| 51 | path = os.path.join(IMAGES_DIR, name) | |
| 52 | if os.path.isdir(path) and name not in ignore: | |
| 53 | collections.append(name) | |
| 54 | return collections | |
| 55 | ||
| 56 | ||
| 57 | def list_images(collection): | |
| 58 | coll_dir = os.path.join(IMAGES_DIR, collection) | |
| 59 | if not os.path.isdir(coll_dir): | |
| 60 | print(f"Error: collection '{collection}' not found", file=sys.stderr) | |
| 61 | sys.exit(1) | |
| 62 | ||
| 63 | images = [] | |
| 64 | for name in sorted(os.listdir(coll_dir)): | |
| 65 | if name == "thumbs" or name.endswith(".md"): | |
| 66 | continue | |
| 67 | filepath = os.path.join(coll_dir, name) | |
| 68 | if os.path.isfile(filepath): | |
| 69 | images.append(name) | |
| 70 | return images | |
| 71 | ||
| 72 | ||
| 73 | def rename_image(collection, old_name, new_name): | |
| 74 | coll_dir = os.path.join(IMAGES_DIR, collection) | |
| 75 | thumbs_dir = os.path.join(coll_dir, "thumbs") | |
| 76 | ||
| 77 | if not os.path.isdir(coll_dir): | |
| 78 | print(f"Error: collection '{collection}' not found", file=sys.stderr) | |
| 79 | sys.exit(1) | |
| 80 | ||
| 81 | old_jpg = old_name + ".jpg" | |
| 82 | new_jpg = new_name + ".jpg" | |
| 83 | old_md = old_name + ".md" | |
| 84 | new_md = new_name + ".md" | |
| 85 | ||
| 86 | moved = [] | |
| 87 | ||
| 88 | # Full-res | |
| 89 | src = os.path.join(coll_dir, old_jpg) | |
| 90 | dst = os.path.join(coll_dir, new_jpg) | |
| 91 | if os.path.exists(src): | |
| 92 | os.rename(src, dst) | |
| 93 | moved.append(f" {old_jpg} -> {new_jpg}") | |
| 94 | else: | |
| 95 | print(f"Warning: {src} not found", file=sys.stderr) | |
| 96 | ||
| 97 | # Thumbnail | |
| 98 | src = os.path.join(thumbs_dir, old_jpg) | |
| 99 | dst = os.path.join(thumbs_dir, new_jpg) | |
| 100 | if os.path.exists(src): | |
| 101 | os.rename(src, dst) | |
| 102 | moved.append(f" thumbs/{old_jpg} -> thumbs/{new_jpg}") | |
| 103 | ||
| 104 | # Markdown sidecar | |
| 105 | src = os.path.join(coll_dir, old_md) | |
| 106 | dst = os.path.join(coll_dir, new_md) | |
| 107 | if os.path.exists(src): | |
| 108 | os.rename(src, dst) | |
| 109 | moved.append(f" {old_md} -> {new_md}") | |
| 110 | ||
| 111 | # Update site.json collections order if present | |
| 112 | config = load_config() | |
| 113 | collections_config = config.get("collections", {}) | |
| 114 | order = collections_config.get("order", []) | |
| 115 | # Collection-level order does not track individual images, | |
| 116 | # so no update needed here unless a per-image order is added later. | |
| 117 | ||
| 118 | if moved: | |
| 119 | print(f"Renamed '{old_name}' -> '{new_name}' in {collection}/:") | |
| 120 | for m in moved: | |
| 121 | print(m) | |
| 122 | else: | |
| 123 | print("Nothing was renamed.") | |
| 124 | ||
| 125 | ||
| 126 | def rename_collection(old_name, new_name): | |
| 127 | old_dir = os.path.join(IMAGES_DIR, old_name) | |
| 128 | new_dir = os.path.join(IMAGES_DIR, new_name) | |
| 129 | ||
| 130 | if not os.path.isdir(old_dir): | |
| 131 | print(f"Error: collection '{old_name}' not found", file=sys.stderr) | |
| 132 | sys.exit(1) | |
| 133 | ||
| 134 | if os.path.exists(new_dir): | |
| 135 | print(f"Error: '{new_name}' already exists", file=sys.stderr) | |
| 136 | sys.exit(1) | |
| 137 | ||
| 138 | os.rename(old_dir, new_dir) | |
| 139 | print(f"Renamed directory: {old_name}/ -> {new_name}/") | |
| 140 | ||
| 141 | # Update site.json collections order | |
| 142 | config = load_config() | |
| 143 | collections_config = config.get("collections", {}) | |
| 144 | order = collections_config.get("order", []) | |
| 145 | if old_name in order: | |
| 146 | order = [new_name if n == old_name else n for n in order] | |
| 147 | collections_config["order"] = order | |
| 148 | config["collections"] = collections_config | |
| 149 | save_config(config) | |
| 150 | print(f"Updated site.json: '{old_name}' -> '{new_name}'") | |
| 151 | ||
| 152 | ||
| 153 | def main(): | |
| 154 | if len(sys.argv) < 2: | |
| 155 | print(__doc__) | |
| 156 | sys.exit(1) | |
| 157 | ||
| 158 | cmd = sys.argv[1] | |
| 159 | ||
| 160 | if cmd == "list": | |
| 161 | if len(sys.argv) >= 3: | |
| 162 | collection = sys.argv[2] | |
| 163 | images = list_images(collection) | |
| 164 | print(f"{collection}/ ({len(images)} images):") | |
| 165 | for img in images: | |
| 166 | print(f" {img}") | |
| 167 | else: | |
| 168 | collections = list_collections() | |
| 169 | for coll in collections: | |
| 170 | images = list_images(coll) | |
| 171 | print(f"{coll}/ ({len(images)} images)") | |
| 172 | ||
| 173 | elif cmd == "image": | |
| 174 | if len(sys.argv) != 5: | |
| 175 | print("Usage: rename.py image <collection> <old-name> <new-name>") | |
| 176 | sys.exit(1) | |
| 177 | rename_image(sys.argv[2], sys.argv[3], sys.argv[4]) | |
| 178 | ||
| 179 | elif cmd == "collection": | |
| 180 | if len(sys.argv) != 4: | |
| 181 | print("Usage: rename.py collection <old-name> <new-name>") | |
| 182 | sys.exit(1) | |
| 183 | rename_collection(sys.argv[2], sys.argv[3]) | |
| 184 | ||
| 185 | else: | |
| 186 | print(f"Unknown command: {cmd}") | |
| 187 | print(__doc__) | |
| 188 | sys.exit(1) | |
| 189 | ||
| 190 | ||
| 191 | if __name__ == "__main__": | |
| 192 | main() |
| … | … | @@ -0,0 +1,427 @@ |
| 1 | """Unit tests for process.py media preprocessing.""" | |
| 2 | ||
| 3 | import json | |
| 4 | import os | |
| 5 | import subprocess | |
| 6 | import tempfile | |
| 7 | import time | |
| 8 | import unittest | |
| 9 | ||
| 10 | from PIL import Image | |
| 11 | ||
| 12 | from process import ( | |
| 13 | process_collection, | |
| 14 | process_video_file, | |
| 15 | process_video_collection, | |
| 16 | get_video_duration, | |
| 17 | generate_manifest, | |
| 18 | slugify, | |
| 19 | load_config, | |
| 20 | ) | |
| 21 | ||
| 22 | ||
| 23 | class TestSlugify(unittest.TestCase): | |
| 24 | """Tests for the slugify helper.""" | |
| 25 | ||
| 26 | def test_basic_slugify(self): | |
| 27 | self.assertEqual(slugify("Paint and Pixels"), "paint-and-pixels") | |
| 28 | ||
| 29 | def test_special_characters(self): | |
| 30 | self.assertEqual(slugify("Hello, World!"), "hello-world") | |
| 31 | ||
| 32 | def test_multiple_spaces(self): | |
| 33 | self.assertEqual(slugify("too many spaces"), "too-many-spaces") | |
| 34 | ||
| 35 | def test_already_slug(self): | |
| 36 | self.assertEqual(slugify("already-a-slug"), "already-a-slug") | |
| 37 | ||
| 38 | ||
| 39 | class TestLoadConfig(unittest.TestCase): | |
| 40 | """Tests for config loading.""" | |
| 41 | ||
| 42 | def test_load_config_defaults(self): | |
| 43 | """Missing config file should return defaults.""" | |
| 44 | config = load_config("/nonexistent/path.json") | |
| 45 | self.assertEqual(config["fullsize"], 1500) | |
| 46 | self.assertEqual(config["thumbnail"], 400) | |
| 47 | self.assertEqual(config["quality"], 85) | |
| 48 | self.assertEqual(config["video_thumbnail_time"], 2) | |
| 49 | ||
| 50 | def test_load_config_from_file(self): | |
| 51 | """Config file should override defaults.""" | |
| 52 | with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: | |
| 53 | json.dump({"processing": {"fullsize": 2000, "quality": 90}}, f) | |
| 54 | f.flush() | |
| 55 | config = load_config(f.name) | |
| 56 | os.unlink(f.name) | |
| 57 | self.assertEqual(config["fullsize"], 2000) | |
| 58 | self.assertEqual(config["quality"], 90) | |
| 59 | # Non-overridden values keep defaults | |
| 60 | self.assertEqual(config["thumbnail"], 400) | |
| 61 | ||
| 62 | ||
| 63 | class TestProcessCollection(unittest.TestCase): | |
| 64 | """Tests for the image preprocessing pipeline.""" | |
| 65 | ||
| 66 | def test_process_single_image(self): | |
| 67 | """A 4000x3000 image should be resized to 1500x1125 full-res and 400x300 thumbnail.""" | |
| 68 | with tempfile.TemporaryDirectory() as tmpdir: | |
| 69 | src_dir = os.path.join(tmpdir, "src") | |
| 70 | dst_dir = os.path.join(tmpdir, "dst") | |
| 71 | os.makedirs(src_dir) | |
| 72 | ||
| 73 | # Create a 4000x3000 test image | |
| 74 | img = Image.new("RGB", (4000, 3000), color=(255, 0, 0)) | |
| 75 | img.save(os.path.join(src_dir, "big.jpg")) | |
| 76 | ||
| 77 | process_collection(src_dir, dst_dir, fullsize=1500, thumbnail=400, quality=85) | |
| 78 | ||
| 79 | # Check full-res output exists and has correct dimensions | |
| 80 | fullres_path = os.path.join(dst_dir, "big.jpg") | |
| 81 | self.assertTrue(os.path.exists(fullres_path), "Full-res output should exist") | |
| 82 | with Image.open(fullres_path) as out: | |
| 83 | self.assertEqual(out.size, (1500, 1125)) | |
| 84 | ||
| 85 | # Check thumbnail exists and has correct dimensions | |
| 86 | thumb_path = os.path.join(dst_dir, "thumbs", "big.jpg") | |
| 87 | self.assertTrue(os.path.exists(thumb_path), "Thumbnail should exist") | |
| 88 | with Image.open(thumb_path) as out: | |
| 89 | self.assertEqual(out.size, (400, 300)) | |
| 90 | ||
| 91 | def test_skip_already_processed(self): | |
| 92 | """Processing again should not overwrite existing outputs (same mtime).""" | |
| 93 | with tempfile.TemporaryDirectory() as tmpdir: | |
| 94 | src_dir = os.path.join(tmpdir, "src") | |
| 95 | dst_dir = os.path.join(tmpdir, "dst") | |
| 96 | os.makedirs(src_dir) | |
| 97 | ||
| 98 | # Create and process a test image | |
| 99 | img = Image.new("RGB", (2000, 1000), color=(0, 255, 0)) | |
| 100 | img.save(os.path.join(src_dir, "test.jpg")) | |
| 101 | ||
| 102 | process_collection(src_dir, dst_dir) | |
| 103 | ||
| 104 | fullres_path = os.path.join(dst_dir, "test.jpg") | |
| 105 | thumb_path = os.path.join(dst_dir, "thumbs", "test.jpg") | |
| 106 | ||
| 107 | # Record mtimes after first processing | |
| 108 | mtime_full_1 = os.path.getmtime(fullres_path) | |
| 109 | mtime_thumb_1 = os.path.getmtime(thumb_path) | |
| 110 | ||
| 111 | # Small delay to ensure any rewrite would have a different mtime | |
| 112 | time.sleep(0.1) | |
| 113 | ||
| 114 | # Process again | |
| 115 | process_collection(src_dir, dst_dir) | |
| 116 | ||
| 117 | # mtimes should be unchanged (files were skipped) | |
| 118 | mtime_full_2 = os.path.getmtime(fullres_path) | |
| 119 | mtime_thumb_2 = os.path.getmtime(thumb_path) | |
| 120 | ||
| 121 | self.assertEqual(mtime_full_1, mtime_full_2, | |
| 122 | "Full-res mtime should not change on re-process") | |
| 123 | self.assertEqual(mtime_thumb_1, mtime_thumb_2, | |
| 124 | "Thumbnail mtime should not change on re-process") | |
| 125 | ||
| 126 | def test_png_to_jpeg(self): | |
| 127 | """PNG input should produce .jpg output.""" | |
| 128 | with tempfile.TemporaryDirectory() as tmpdir: | |
| 129 | src_dir = os.path.join(tmpdir, "src") | |
| 130 | dst_dir = os.path.join(tmpdir, "dst") | |
| 131 | os.makedirs(src_dir) | |
| 132 | ||
| 133 | # Create a PNG with alpha channel (RGBA) | |
| 134 | img = Image.new("RGBA", (2000, 1500), color=(0, 0, 255, 128)) | |
| 135 | img.save(os.path.join(src_dir, "alpha.png")) | |
| 136 | ||
| 137 | process_collection(src_dir, dst_dir) | |
| 138 | ||
| 139 | # Output should be .jpg, not .png | |
| 140 | jpg_path = os.path.join(dst_dir, "alpha.jpg") | |
| 141 | png_path = os.path.join(dst_dir, "alpha.png") | |
| 142 | self.assertTrue(os.path.exists(jpg_path), | |
| 143 | "PNG should be converted to .jpg output") | |
| 144 | self.assertFalse(os.path.exists(png_path), | |
| 145 | "Original .png extension should not appear in output") | |
| 146 | ||
| 147 | # Verify it is actually a JPEG | |
| 148 | with Image.open(jpg_path) as out: | |
| 149 | self.assertEqual(out.format, "JPEG") | |
| 150 | ||
| 151 | def test_small_image_not_upscaled(self): | |
| 152 | """A 300x200 image should not be upscaled.""" | |
| 153 | with tempfile.TemporaryDirectory() as tmpdir: | |
| 154 | src_dir = os.path.join(tmpdir, "src") | |
| 155 | dst_dir = os.path.join(tmpdir, "dst") | |
| 156 | os.makedirs(src_dir) | |
| 157 | ||
| 158 | # Create a small image (smaller than both fullsize and thumbnail targets) | |
| 159 | img = Image.new("RGB", (300, 200), color=(128, 128, 0)) | |
| 160 | img.save(os.path.join(src_dir, "small.jpg")) | |
| 161 | ||
| 162 | process_collection(src_dir, dst_dir, fullsize=1500, thumbnail=400, quality=85) | |
| 163 | ||
| 164 | # Full-res should keep original size (no upscaling) | |
| 165 | fullres_path = os.path.join(dst_dir, "small.jpg") | |
| 166 | self.assertTrue(os.path.exists(fullres_path)) | |
| 167 | with Image.open(fullres_path) as out: | |
| 168 | self.assertEqual(out.size, (300, 200), | |
| 169 | "Small image should not be upscaled for full-res") | |
| 170 | ||
| 171 | # Thumbnail should also keep original size (no upscaling) | |
| 172 | thumb_path = os.path.join(dst_dir, "thumbs", "small.jpg") | |
| 173 | self.assertTrue(os.path.exists(thumb_path)) | |
| 174 | with Image.open(thumb_path) as out: | |
| 175 | self.assertEqual(out.size, (300, 200), | |
| 176 | "Small image should not be upscaled for thumbnail") | |
| 177 | ||
| 178 | ||
| 179 | class TestVideoProcessing(unittest.TestCase): | |
| 180 | """Tests for the video preprocessing pipeline.""" | |
| 181 | ||
| 182 | def _create_test_video(self, path, duration=3): | |
| 183 | """Create minimal test video with ffmpeg.""" | |
| 184 | os.makedirs(os.path.dirname(path), exist_ok=True) | |
| 185 | subprocess.run([ | |
| 186 | "ffmpeg", "-y", "-f", "lavfi", "-i", | |
| 187 | f"color=c=red:s=320x240:d={duration}", | |
| 188 | "-f", "lavfi", "-i", f"anullsrc=r=44100:cl=mono", | |
| 189 | "-t", str(duration), "-c:v", "libx264", "-c:a", "aac", | |
| 190 | "-shortest", path | |
| 191 | ], capture_output=True, check=True) | |
| 192 | ||
| 193 | def test_process_video(self): | |
| 194 | """Video should produce faststart MP4 + poster thumbnail.""" | |
| 195 | with tempfile.TemporaryDirectory() as tmpdir: | |
| 196 | src_dir = os.path.join(tmpdir, "src") | |
| 197 | dst_dir = os.path.join(tmpdir, "dst") | |
| 198 | thumbs_dir = os.path.join(dst_dir, "thumbs") | |
| 199 | ||
| 200 | src_path = os.path.join(src_dir, "clip.mp4") | |
| 201 | self._create_test_video(src_path, duration=3) | |
| 202 | ||
| 203 | result = process_video_file(src_path, dst_dir, thumbs_dir, thumbnail_time=1) | |
| 204 | self.assertTrue(result, "Video should be processed (not skipped)") | |
| 205 | ||
| 206 | # Output video should exist | |
| 207 | out_path = os.path.join(dst_dir, "clip.mp4") | |
| 208 | self.assertTrue(os.path.exists(out_path), "Output MP4 should exist") | |
| 209 | self.assertGreater(os.path.getsize(out_path), 0, "Output MP4 should not be empty") | |
| 210 | ||
| 211 | # Poster thumbnail should exist | |
| 212 | thumb_path = os.path.join(thumbs_dir, "clip.jpg") | |
| 213 | self.assertTrue(os.path.exists(thumb_path), "Poster thumbnail should exist") | |
| 214 | with Image.open(thumb_path) as img: | |
| 215 | self.assertEqual(img.format, "JPEG") | |
| 216 | ||
| 217 | def test_video_skip_already_processed(self): | |
| 218 | """Already-processed videos should be skipped.""" | |
| 219 | with tempfile.TemporaryDirectory() as tmpdir: | |
| 220 | src_dir = os.path.join(tmpdir, "src") | |
| 221 | dst_dir = os.path.join(tmpdir, "dst") | |
| 222 | thumbs_dir = os.path.join(dst_dir, "thumbs") | |
| 223 | ||
| 224 | src_path = os.path.join(src_dir, "clip.mp4") | |
| 225 | self._create_test_video(src_path, duration=2) | |
| 226 | ||
| 227 | # First run should process | |
| 228 | result1 = process_video_file(src_path, dst_dir, thumbs_dir) | |
| 229 | self.assertTrue(result1, "First run should process") | |
| 230 | ||
| 231 | out_path = os.path.join(dst_dir, "clip.mp4") | |
| 232 | thumb_path = os.path.join(thumbs_dir, "clip.jpg") | |
| 233 | mtime_vid = os.path.getmtime(out_path) | |
| 234 | mtime_thumb = os.path.getmtime(thumb_path) | |
| 235 | ||
| 236 | time.sleep(0.1) | |
| 237 | ||
| 238 | # Second run should skip | |
| 239 | result2 = process_video_file(src_path, dst_dir, thumbs_dir) | |
| 240 | self.assertFalse(result2, "Second run should skip") | |
| 241 | ||
| 242 | self.assertEqual(os.path.getmtime(out_path), mtime_vid, | |
| 243 | "Video mtime should not change on re-process") | |
| 244 | self.assertEqual(os.path.getmtime(thumb_path), mtime_thumb, | |
| 245 | "Thumbnail mtime should not change on re-process") | |
| 246 | ||
| 247 | def test_get_video_duration(self): | |
| 248 | """Should return correct duration for a test video.""" | |
| 249 | with tempfile.TemporaryDirectory() as tmpdir: | |
| 250 | path = os.path.join(tmpdir, "test.mp4") | |
| 251 | self._create_test_video(path, duration=4) | |
| 252 | ||
| 253 | duration = get_video_duration(path) | |
| 254 | # Allow some tolerance for encoding | |
| 255 | self.assertGreater(duration, 3.0) | |
| 256 | self.assertLess(duration, 5.0) | |
| 257 | ||
| 258 | def test_get_video_duration_invalid(self): | |
| 259 | """Should return 0.0 for non-video files.""" | |
| 260 | with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f: | |
| 261 | f.write(b"not a video") | |
| 262 | f.flush() | |
| 263 | duration = get_video_duration(f.name) | |
| 264 | os.unlink(f.name) | |
| 265 | self.assertEqual(duration, 0.0) | |
| 266 | ||
| 267 | def test_process_video_collection(self): | |
| 268 | """Should process all videos in a directory.""" | |
| 269 | with tempfile.TemporaryDirectory() as tmpdir: | |
| 270 | src_dir = os.path.join(tmpdir, "src") | |
| 271 | dst_dir = os.path.join(tmpdir, "dst") | |
| 272 | os.makedirs(src_dir) | |
| 273 | ||
| 274 | # Create two test videos | |
| 275 | self._create_test_video(os.path.join(src_dir, "a.mp4"), duration=2) | |
| 276 | self._create_test_video(os.path.join(src_dir, "b.mp4"), duration=2) | |
| 277 | ||
| 278 | # Also create a non-video file (should be ignored) | |
| 279 | with open(os.path.join(src_dir, "readme.txt"), "w") as f: | |
| 280 | f.write("not a video") | |
| 281 | ||
| 282 | processed, skipped = process_video_collection(src_dir, dst_dir) | |
| 283 | self.assertEqual(processed, 2) | |
| 284 | self.assertEqual(skipped, 0) | |
| 285 | ||
| 286 | self.assertTrue(os.path.exists(os.path.join(dst_dir, "a.mp4"))) | |
| 287 | self.assertTrue(os.path.exists(os.path.join(dst_dir, "b.mp4"))) | |
| 288 | self.assertTrue(os.path.exists(os.path.join(dst_dir, "thumbs", "a.jpg"))) | |
| 289 | self.assertTrue(os.path.exists(os.path.join(dst_dir, "thumbs", "b.jpg"))) | |
| 290 | ||
| 291 | ||
| 292 | class TestManifestGeneration(unittest.TestCase): | |
| 293 | """Tests for manifest generation.""" | |
| 294 | ||
| 295 | def _setup_images(self, base_dir, collection, filenames): | |
| 296 | """Create dummy image files in images/<collection>/.""" | |
| 297 | coll_dir = os.path.join(base_dir, "images", collection) | |
| 298 | thumbs_dir = os.path.join(coll_dir, "thumbs") | |
| 299 | os.makedirs(coll_dir, exist_ok=True) | |
| 300 | os.makedirs(thumbs_dir, exist_ok=True) | |
| 301 | for fname in filenames: | |
| 302 | img = Image.new("RGB", (100, 100), color=(255, 0, 0)) | |
| 303 | img.save(os.path.join(coll_dir, fname)) | |
| 304 | img.save(os.path.join(thumbs_dir, fname)) | |
| 305 | ||
| 306 | def _setup_videos(self, base_dir, collection, filenames): | |
| 307 | """Create dummy video files in videos/<collection>/.""" | |
| 308 | coll_dir = os.path.join(base_dir, "videos", collection) | |
| 309 | thumbs_dir = os.path.join(coll_dir, "thumbs") | |
| 310 | os.makedirs(coll_dir, exist_ok=True) | |
| 311 | os.makedirs(thumbs_dir, exist_ok=True) | |
| 312 | for fname in filenames: | |
| 313 | # Create a minimal valid MP4 for duration detection | |
| 314 | # (just an empty file -- duration will be 0.0) | |
| 315 | with open(os.path.join(coll_dir, fname), "wb") as f: | |
| 316 | f.write(b"\x00" * 100) | |
| 317 | # Create thumbnail | |
| 318 | basename = os.path.splitext(fname)[0] | |
| 319 | img = Image.new("RGB", (100, 100), color=(0, 0, 255)) | |
| 320 | img.save(os.path.join(thumbs_dir, basename + ".jpg")) | |
| 321 | ||
| 322 | def test_generate_manifest_images(self): | |
| 323 | """Manifest should include all images with correct structure.""" | |
| 324 | with tempfile.TemporaryDirectory() as tmpdir: | |
| 325 | self._setup_images(tmpdir, "nature", ["sunset.jpg", "forest.jpg"]) | |
| 326 | ||
| 327 | config = {"collections": {"order": ["nature"], "ignore": []}} | |
| 328 | manifest = generate_manifest(tmpdir, config) | |
| 329 | ||
| 330 | self.assertEqual(len(manifest["collections"]), 1) | |
| 331 | self.assertEqual(manifest["collections"][0]["name"], "nature") | |
| 332 | self.assertEqual(manifest["collections"][0]["images"], 2) | |
| 333 | self.assertEqual(manifest["collections"][0]["total"], 2) | |
| 334 | ||
| 335 | self.assertEqual(len(manifest["items"]), 2) | |
| 336 | item = manifest["items"][0] | |
| 337 | self.assertEqual(item["type"], "image") | |
| 338 | self.assertEqual(item["collection"], "nature") | |
| 339 | self.assertIn("path", item) | |
| 340 | self.assertIn("thumb", item) | |
| 341 | ||
| 342 | def test_generate_manifest_with_sidecar(self): | |
| 343 | """Images with .md sidecars should include sidecar content.""" | |
| 344 | with tempfile.TemporaryDirectory() as tmpdir: | |
| 345 | self._setup_images(tmpdir, "art", ["piece.jpg"]) | |
| 346 | ||
| 347 | # Write a sidecar | |
| 348 | sidecar_path = os.path.join(tmpdir, "images", "art", "piece.md") | |
| 349 | with open(sidecar_path, "w") as f: | |
| 350 | f.write("Oil on canvas, 2024") | |
| 351 | ||
| 352 | config = {"collections": {"order": ["art"], "ignore": []}} | |
| 353 | manifest = generate_manifest(tmpdir, config) | |
| 354 | ||
| 355 | self.assertEqual(len(manifest["items"]), 1) | |
| 356 | self.assertEqual(manifest["items"][0]["sidecar"], "Oil on canvas, 2024") | |
| 357 | ||
| 358 | def test_generate_manifest_respects_order(self): | |
| 359 | """Collections should follow site.json order.""" | |
| 360 | with tempfile.TemporaryDirectory() as tmpdir: | |
| 361 | self._setup_images(tmpdir, "alpha", ["a.jpg"]) | |
| 362 | self._setup_images(tmpdir, "beta", ["b.jpg"]) | |
| 363 | self._setup_images(tmpdir, "gamma", ["g.jpg"]) | |
| 364 | ||
| 365 | config = {"collections": {"order": ["gamma", "alpha", "beta"], "ignore": []}} | |
| 366 | manifest = generate_manifest(tmpdir, config) | |
| 367 | ||
| 368 | names = [c["name"] for c in manifest["collections"]] | |
| 369 | self.assertEqual(names, ["gamma", "alpha", "beta"]) | |
| 370 | ||
| 371 | def test_generate_manifest_unordered_collections(self): | |
| 372 | """Collections not in order list should appear alphabetically after ordered ones.""" | |
| 373 | with tempfile.TemporaryDirectory() as tmpdir: | |
| 374 | self._setup_images(tmpdir, "alpha", ["a.jpg"]) | |
| 375 | self._setup_images(tmpdir, "beta", ["b.jpg"]) | |
| 376 | self._setup_images(tmpdir, "gamma", ["g.jpg"]) | |
| 377 | ||
| 378 | config = {"collections": {"order": ["gamma"], "ignore": []}} | |
| 379 | manifest = generate_manifest(tmpdir, config) | |
| 380 | ||
| 381 | names = [c["name"] for c in manifest["collections"]] | |
| 382 | self.assertEqual(names, ["gamma", "alpha", "beta"]) | |
| 383 | ||
| 384 | def test_generate_manifest_ignores(self): | |
| 385 | """Ignored collections should not appear.""" | |
| 386 | with tempfile.TemporaryDirectory() as tmpdir: | |
| 387 | self._setup_images(tmpdir, "keep", ["a.jpg"]) | |
| 388 | self._setup_images(tmpdir, "old", ["b.jpg"]) | |
| 389 | self._setup_images(tmpdir, "_archive", ["c.jpg"]) | |
| 390 | ||
| 391 | config = {"collections": {"order": [], "ignore": ["old", "_archive"]}} | |
| 392 | manifest = generate_manifest(tmpdir, config) | |
| 393 | ||
| 394 | names = [c["name"] for c in manifest["collections"]] | |
| 395 | self.assertEqual(names, ["keep"]) | |
| 396 | self.assertEqual(len(manifest["items"]), 1) | |
| 397 | ||
| 398 | def test_generate_manifest_mixed_media(self): | |
| 399 | """Manifest should include both images and videos from same collection.""" | |
| 400 | with tempfile.TemporaryDirectory() as tmpdir: | |
| 401 | self._setup_images(tmpdir, "portfolio", ["photo.jpg"]) | |
| 402 | self._setup_videos(tmpdir, "portfolio", ["clip.mp4"]) | |
| 403 | ||
| 404 | config = {"collections": {"order": ["portfolio"], "ignore": []}} | |
| 405 | manifest = generate_manifest(tmpdir, config) | |
| 406 | ||
| 407 | self.assertEqual(len(manifest["collections"]), 1) | |
| 408 | coll = manifest["collections"][0] | |
| 409 | self.assertEqual(coll["images"], 1) | |
| 410 | self.assertEqual(coll["videos"], 1) | |
| 411 | self.assertEqual(coll["total"], 2) | |
| 412 | ||
| 413 | types = {item["type"] for item in manifest["items"]} | |
| 414 | self.assertEqual(types, {"image", "video"}) | |
| 415 | ||
| 416 | def test_generate_manifest_empty(self): | |
| 417 | """Empty directories should produce empty manifest.""" | |
| 418 | with tempfile.TemporaryDirectory() as tmpdir: | |
| 419 | config = {"collections": {"order": [], "ignore": []}} | |
| 420 | manifest = generate_manifest(tmpdir, config) | |
| 421 | ||
| 422 | self.assertEqual(manifest["collections"], []) | |
| 423 | self.assertEqual(manifest["items"], []) | |
| 424 | ||
| 425 | ||
| 426 | if __name__ == "__main__": | |
| 427 | unittest.main() |