| 1 | #!/usr/bin/env python3 |
| 2 | """ |
| 3 | PPT Master - Image Treatment |
| 4 | |
| 5 | Create a non-destructive PNG derivative from one project-local bitmap while |
| 6 | preserving its display dimensions, alpha mask, and any matching web provenance. |
| 7 | |
| 8 | Usage: |
| 9 | python3 scripts/image_treat.py <project_path> <source> --output <filename.png> [options] |
| 10 | |
| 11 | Examples: |
| 12 | python3 scripts/image_treat.py projects/demo hero.jpg --output hero_soft.png --blur 12 |
| 13 | python3 scripts/image_treat.py projects/demo hero.jpg --output hero_duotone.png \ |
| 14 | --contrast 1.1 --duotone "#14213D" "#FCA311" |
| 15 | |
| 16 | Dependencies: |
| 17 | Pillow; project image-search dependencies when copying web provenance |
| 18 | |
| 19 | Treatments run in this fixed order: brightness, contrast, tone treatment |
| 20 | (desaturate / grayscale / duotone), then Gaussian blur. |
| 21 | """ |
| 22 | |
| 23 | from __future__ import annotations |
| 24 | |
| 25 | import argparse |
| 26 | import copy |
| 27 | import math |
| 28 | import os |
| 29 | import re |
| 30 | import sys |
| 31 | import tempfile |
| 32 | from io import BytesIO |
| 33 | from pathlib import Path |
| 34 | from typing import Optional |
| 35 | |
| 36 | from PIL import ( |
| 37 | Image, |
| 38 | ImageCms, |
| 39 | ImageEnhance, |
| 40 | ImageFilter, |
| 41 | ImageOps, |
| 42 | UnidentifiedImageError, |
| 43 | ) |
| 44 | |
| 45 | |
| 46 | _SCRIPTS_DIR = Path(__file__).resolve().parent |
| 47 | if str(_SCRIPTS_DIR) not in sys.path: |
| 48 | sys.path.insert(0, str(_SCRIPTS_DIR)) |
| 49 | |
| 50 | from console_encoding import configure_utf8_stdio # noqa: E402 |
| 51 | |
| 52 | configure_utf8_stdio() |
| 53 | |
| 54 | |
| 55 | _HEX_COLOR_RE = re.compile(r"^#?([0-9A-Fa-f]{6})$") |
| 56 | |
| 57 | |
| 58 | def _read_sources_manifest(path: Path) -> dict: |
| 59 | from image_search import _read_existing_manifest |
| 60 | |
| 61 | return _read_existing_manifest(path) |
| 62 | |
| 63 | |
| 64 | def _write_sources_manifest(path: Path, item: dict) -> Path: |
| 65 | from image_search import write_sources_manifest |
| 66 | |
| 67 | return write_sources_manifest(path, item) |
| 68 | |
| 69 | |
| 70 | def _finite_float(value: str) -> float: |
| 71 | try: |
| 72 | number = float(value) |
| 73 | except ValueError as exc: |
| 74 | raise argparse.ArgumentTypeError(f"expected a number, got {value!r}") from exc |
| 75 | if not math.isfinite(number): |
| 76 | raise argparse.ArgumentTypeError("value must be finite") |
| 77 | return number |
| 78 | |
| 79 | |
| 80 | def _nonnegative_float(value: str) -> float: |
| 81 | number = _finite_float(value) |
| 82 | if number < 0: |
| 83 | raise argparse.ArgumentTypeError("value must be greater than or equal to 0") |
| 84 | return number |
| 85 | |
| 86 | |
| 87 | def _positive_float(value: str) -> float: |
| 88 | number = _finite_float(value) |
| 89 | if number <= 0: |
| 90 | raise argparse.ArgumentTypeError("value must be greater than 0") |
| 91 | return number |
| 92 | |
| 93 | |
| 94 | def _unit_float(value: str) -> float: |
| 95 | number = _finite_float(value) |
| 96 | if not 0 <= number <= 1: |
| 97 | raise argparse.ArgumentTypeError("value must be between 0 and 1") |
| 98 | return number |
| 99 | |
| 100 | |
| 101 | def _hex_color(value: str) -> str: |
| 102 | match = _HEX_COLOR_RE.fullmatch(value) |
| 103 | if match is None: |
| 104 | raise argparse.ArgumentTypeError("color must be #RRGGBB or RRGGBB") |
| 105 | return f"#{match.group(1).upper()}" |
| 106 | |
| 107 | |
| 108 | def build_parser() -> argparse.ArgumentParser: |
| 109 | parser = argparse.ArgumentParser( |
| 110 | description=( |
| 111 | "Create a project-local PNG derivative. Processing order: brightness -> " |
| 112 | "contrast -> desaturate/grayscale/duotone -> blur." |
| 113 | ), |
| 114 | formatter_class=argparse.RawDescriptionHelpFormatter, |
| 115 | ) |
| 116 | parser.add_argument("project_path", help="Project root containing an images/ directory.") |
| 117 | parser.add_argument( |
| 118 | "source", |
| 119 | help="Existing bare bitmap filename under <project_path>/images (no path components).", |
| 120 | ) |
| 121 | parser.add_argument( |
| 122 | "--output", |
| 123 | required=True, |
| 124 | help="New bare .png filename under <project_path>/images; existing files are never replaced.", |
| 125 | ) |
| 126 | parser.add_argument( |
| 127 | "--brightness", |
| 128 | type=_positive_float, |
| 129 | help="Brightness factor greater than 0; 1 is unchanged.", |
| 130 | ) |
| 131 | parser.add_argument( |
| 132 | "--contrast", |
| 133 | type=_positive_float, |
| 134 | help="Contrast factor greater than 0; 1 is unchanged.", |
| 135 | ) |
| 136 | tone_group = parser.add_mutually_exclusive_group() |
| 137 | tone_group.add_argument( |
| 138 | "--desaturate", |
| 139 | type=_unit_float, |
| 140 | help="Remove this fraction of color (0..1); 1 is grayscale.", |
| 141 | ) |
| 142 | tone_group.add_argument( |
| 143 | "--grayscale", |
| 144 | action="store_true", |
| 145 | help="Convert RGB content to grayscale while preserving any alpha mask.", |
| 146 | ) |
| 147 | tone_group.add_argument( |
| 148 | "--duotone", |
| 149 | nargs=2, |
| 150 | type=_hex_color, |
| 151 | metavar=("SHADOW", "HIGHLIGHT"), |
| 152 | help="Map sRGB luminance between two sRGB #RRGGBB colors.", |
| 153 | ) |
| 154 | parser.add_argument( |
| 155 | "--blur", |
| 156 | type=_nonnegative_float, |
| 157 | help="Gaussian blur radius greater than or equal to 0; 0 is unchanged.", |
| 158 | ) |
| 159 | return parser |
| 160 | |
| 161 | |
| 162 | def _validate_bare_filename(value: str, *, field_name: str) -> str: |
| 163 | if ( |
| 164 | not value.strip() |
| 165 | or value in {".", ".."} |
| 166 | or "/" in value |
| 167 | or "\\" in value |
| 168 | or ":" in value |
| 169 | or Path(value).is_absolute() |
| 170 | ): |
| 171 | raise ValueError( |
| 172 | f"{field_name} must be a bare filename without path components: {value!r}" |
| 173 | ) |
| 174 | return value |
| 175 | |
| 176 | |
| 177 | def _assert_output_absent(output_path: Path) -> None: |
| 178 | for child in output_path.parent.iterdir(): |
| 179 | if child.name.casefold() == output_path.name.casefold(): |
| 180 | raise ValueError(f"output already exists or conflicts by casing: {child}") |
| 181 | |
| 182 | |
| 183 | def _resolve_paths(project_value: str, source_name: str, output_name: str) -> tuple[Path, Path, Path]: |
| 184 | source_name = _validate_bare_filename(source_name, field_name="source") |
| 185 | output_name = _validate_bare_filename(output_name, field_name="output") |
| 186 | if Path(output_name).suffix.casefold() != ".png": |
| 187 | raise ValueError("output must use the .png extension") |
| 188 | if source_name.casefold() == output_name.casefold(): |
| 189 | raise ValueError("output must differ from source, including filename casing") |
| 190 | |
| 191 | project_input = Path(project_value).expanduser() |
| 192 | try: |
| 193 | project_path = project_input.resolve(strict=True) |
| 194 | except (OSError, RuntimeError) as exc: |
| 195 | raise ValueError(f"project path does not resolve safely: {project_input} ({exc})") from exc |
| 196 | if not project_path.is_dir(): |
| 197 | raise ValueError(f"project path is not a directory: {project_path}") |
| 198 | |
| 199 | images_link = project_path / "images" |
| 200 | if images_link.is_symlink(): |
| 201 | raise ValueError(f"project images directory must not be a symlink: {images_link}") |
| 202 | if not images_link.is_dir(): |
| 203 | raise ValueError(f"project images directory does not exist: {images_link}") |
| 204 | images_path = images_link.resolve(strict=True) |
| 205 | if images_path.parent != project_path: |
| 206 | raise ValueError(f"project images directory escapes the project root: {images_link}") |
| 207 | |
| 208 | source_link = images_path / source_name |
| 209 | if source_link.is_symlink(): |
| 210 | raise ValueError(f"source must not be a symlink: {source_link}") |
| 211 | try: |
| 212 | source_path = source_link.resolve(strict=True) |
| 213 | except (OSError, RuntimeError) as exc: |
| 214 | raise ValueError(f"source does not resolve safely: {source_link} ({exc})") from exc |
| 215 | if source_path.parent != images_path or not source_path.is_file(): |
| 216 | raise ValueError(f"source must be a file directly under project images/: {source_link}") |
| 217 | |
| 218 | output_path = images_path / output_name |
| 219 | _assert_output_absent(output_path) |
| 220 | if output_path.exists() or output_path.is_symlink(): |
| 221 | raise ValueError(f"output already exists: {output_path}") |
| 222 | if output_path.parent.resolve() != images_path: |
| 223 | raise ValueError(f"output escapes the project images directory: {output_path}") |
| 224 | |
| 225 | return images_path, source_path, output_path |
| 226 | |
| 227 | |
| 228 | def _treatment_plan(args: argparse.Namespace) -> list[dict]: |
| 229 | plan: list[dict] = [] |
| 230 | if args.brightness is not None and args.brightness != 1: |
| 231 | plan.append({"operation": "brightness", "factor": args.brightness}) |
| 232 | if args.contrast is not None and args.contrast != 1: |
| 233 | plan.append({"operation": "contrast", "factor": args.contrast}) |
| 234 | if args.desaturate is not None and args.desaturate > 0: |
| 235 | plan.append({"operation": "desaturate", "amount": args.desaturate}) |
| 236 | elif args.grayscale: |
| 237 | plan.append({"operation": "grayscale"}) |
| 238 | elif args.duotone is not None: |
| 239 | plan.append( |
| 240 | { |
| 241 | "operation": "duotone", |
| 242 | "shadow": args.duotone[0], |
| 243 | "highlight": args.duotone[1], |
| 244 | } |
| 245 | ) |
| 246 | if args.blur is not None and args.blur > 0: |
| 247 | plan.append({"operation": "blur", "radius": args.blur}) |
| 248 | if not plan: |
| 249 | raise ValueError( |
| 250 | "select at least one effective treatment; identity values such as " |
| 251 | "--brightness 1, --desaturate 0, or --blur 0 do not change the image" |
| 252 | ) |
| 253 | return plan |
| 254 | |
| 255 | |
| 256 | def _validate_blur_radius(radius: Optional[float], width: int, height: int) -> None: |
| 257 | if radius is None or radius <= 0: |
| 258 | return |
| 259 | effective_maximum = max(width, height) |
| 260 | if radius > effective_maximum: |
| 261 | raise ValueError( |
| 262 | f"blur radius {radius:g} exceeds the effective maximum " |
| 263 | f"{effective_maximum} for a {width}x{height} image; choose --blur " |
| 264 | f"at or below {effective_maximum}" |
| 265 | ) |
| 266 | |
| 267 | |
| 268 | def _convert_duotone_source_to_srgb(image: Image.Image) -> tuple[Image.Image, bytes]: |
| 269 | output_profile = ImageCms.ImageCmsProfile(ImageCms.createProfile("sRGB")) |
| 270 | output_icc = output_profile.tobytes() |
| 271 | source_icc = image.info.get("icc_profile") |
| 272 | if not source_icc: |
| 273 | return image.convert("RGB"), output_icc |
| 274 | if not isinstance(source_icc, bytes): |
| 275 | raise ValueError( |
| 276 | "source ICC profile has an unsupported representation; convert the " |
| 277 | "source to sRGB or remove the invalid profile, then retry" |
| 278 | ) |
| 279 | try: |
| 280 | input_profile = ImageCms.ImageCmsProfile(BytesIO(source_icc)) |
| 281 | converted = ImageCms.profileToProfile( |
| 282 | image, |
| 283 | input_profile, |
| 284 | output_profile, |
| 285 | outputMode="RGB", |
| 286 | ) |
| 287 | except (ImageCms.PyCMSError, OSError, TypeError, ValueError) as exc: |
| 288 | raise ValueError( |
| 289 | "source ICC profile is invalid or incompatible with the image mode; " |
| 290 | "convert the source to sRGB or remove the invalid profile, then retry" |
| 291 | ) from exc |
| 292 | if converted is None: |
| 293 | raise RuntimeError("ICC conversion did not produce an image") |
| 294 | return converted, output_icc |
| 295 | |
| 296 | |
| 297 | def _has_alpha(image: Image.Image) -> bool: |
| 298 | return "A" in image.getbands() or "transparency" in image.info |
| 299 | |
| 300 | |
| 301 | def _apply_treatments( |
| 302 | source_path: Path, |
| 303 | temporary_path: Path, |
| 304 | args: argparse.Namespace, |
| 305 | ) -> tuple[int, int]: |
| 306 | try: |
| 307 | with Image.open(source_path) as source: |
| 308 | if int(getattr(source, "n_frames", 1)) != 1: |
| 309 | raise RuntimeError( |
| 310 | f"animated or multi-frame images are unsupported: {source_path}" |
| 311 | ) |
| 312 | oriented = ImageOps.exif_transpose(source) |
| 313 | try: |
| 314 | oriented.load() |
| 315 | width, height = oriented.size |
| 316 | _validate_blur_radius(args.blur, width, height) |
| 317 | has_alpha = _has_alpha(oriented) |
| 318 | rgba = oriented.convert("RGBA") if has_alpha else None |
| 319 | alpha = rgba.getchannel("A") if rgba is not None else None |
| 320 | if args.duotone is not None: |
| 321 | rgb, icc_profile = _convert_duotone_source_to_srgb(oriented) |
| 322 | else: |
| 323 | keep_icc = oriented.mode in {"RGB", "RGBA"} |
| 324 | icc_profile = oriented.info.get("icc_profile") if keep_icc else None |
| 325 | rgb = rgba.convert("RGB") if rgba is not None else oriented.convert("RGB") |
| 326 | |
| 327 | if args.brightness is not None and args.brightness != 1: |
| 328 | rgb = ImageEnhance.Brightness(rgb).enhance(args.brightness) |
| 329 | if args.contrast is not None and args.contrast != 1: |
| 330 | rgb = ImageEnhance.Contrast(rgb).enhance(args.contrast) |
| 331 | if args.desaturate is not None and args.desaturate > 0: |
| 332 | rgb = ImageEnhance.Color(rgb).enhance(1 - args.desaturate) |
| 333 | elif args.grayscale: |
| 334 | rgb = ImageOps.grayscale(rgb).convert("RGB") |
| 335 | elif args.duotone is not None: |
| 336 | rgb = ImageOps.colorize( |
| 337 | ImageOps.grayscale(rgb), |
| 338 | black=args.duotone[0], |
| 339 | white=args.duotone[1], |
| 340 | ) |
| 341 | if args.blur is not None and args.blur > 0: |
| 342 | rgb = rgb.filter(ImageFilter.GaussianBlur(radius=args.blur)) |
| 343 | |
| 344 | result = rgb.convert("RGBA") if alpha is not None else rgb |
| 345 | if alpha is not None: |
| 346 | result.putalpha(alpha) |
| 347 | save_options = {"format": "PNG"} |
| 348 | if isinstance(icc_profile, bytes) and icc_profile: |
| 349 | save_options["icc_profile"] = icc_profile |
| 350 | result.save(temporary_path, **save_options) |
| 351 | result.close() |
| 352 | if rgba is not None: |
| 353 | rgba.close() |
| 354 | if alpha is not None: |
| 355 | alpha.close() |
| 356 | rgb.close() |
| 357 | return width, height |
| 358 | finally: |
| 359 | if oriented is not source: |
| 360 | oriented.close() |
| 361 | except (OSError, UnidentifiedImageError) as exc: |
| 362 | raise RuntimeError(f"unable to read or treat source image {source_path}: {exc}") from exc |
| 363 | |
| 364 | |
| 365 | def _rewrite_attribution_text(item: dict, source_name: str, output_name: str) -> str: |
| 366 | value = item.get("attribution_text") |
| 367 | if not isinstance(value, str) or not value: |
| 368 | return value if isinstance(value, str) else "" |
| 369 | if not value.startswith(source_name): |
| 370 | return value |
| 371 | remainder = value[len(source_name) :] |
| 372 | if remainder and not (remainder[0].isspace() or remainder[0] in "—–-:"): |
| 373 | return value |
| 374 | return output_name + remainder |
| 375 | |
| 376 | |
| 377 | def _prepare_provenance( |
| 378 | manifest_path: Path, |
| 379 | source_name: str, |
| 380 | output_name: str, |
| 381 | treatments: list[dict], |
| 382 | ) -> Optional[dict]: |
| 383 | if manifest_path.is_symlink(): |
| 384 | raise ValueError(f"image source manifest must not be a symlink: {manifest_path}") |
| 385 | if not manifest_path.exists(): |
| 386 | return None |
| 387 | |
| 388 | payload = _read_sources_manifest(manifest_path) |
| 389 | items = payload.get("items") or [] |
| 390 | source_item = None |
| 391 | for item in items: |
| 392 | filename = item.get("filename") |
| 393 | if not isinstance(filename, str): |
| 394 | continue |
| 395 | if filename.casefold() == output_name.casefold(): |
| 396 | raise ValueError( |
| 397 | f"output already has an image_sources.json record: {filename!r}" |
| 398 | ) |
| 399 | if filename.casefold() == source_name.casefold(): |
| 400 | if filename != source_name: |
| 401 | raise ValueError( |
| 402 | "source filename casing does not match image_sources.json: " |
| 403 | f"{source_name!r} vs {filename!r}" |
| 404 | ) |
| 405 | source_item = item |
| 406 | |
| 407 | if source_item is None: |
| 408 | return None |
| 409 | derived_item = copy.deepcopy(source_item) |
| 410 | derived_item["filename"] = output_name |
| 411 | derived_item["attribution_text"] = _rewrite_attribution_text( |
| 412 | source_item, |
| 413 | source_name, |
| 414 | output_name, |
| 415 | ) |
| 416 | derived_item["derived_from"] = source_name |
| 417 | derived_item["treatments"] = copy.deepcopy(treatments) |
| 418 | return derived_item |
| 419 | |
| 420 | |
| 421 | def _assert_manifest_output_absent(manifest_path: Path, output_name: str) -> None: |
| 422 | payload = _read_sources_manifest(manifest_path) |
| 423 | for item in payload.get("items") or []: |
| 424 | filename = item.get("filename") |
| 425 | if isinstance(filename, str) and filename.casefold() == output_name.casefold(): |
| 426 | raise RuntimeError( |
| 427 | f"output already has an image_sources.json record: {filename!r}" |
| 428 | ) |
| 429 | |
| 430 | |
| 431 | def _is_staged_output(temporary_path: Path, output_path: Path) -> bool: |
| 432 | try: |
| 433 | return not output_path.is_symlink() and os.path.samefile(temporary_path, output_path) |
| 434 | except OSError: |
| 435 | return False |
| 436 | |
| 437 | |
| 438 | def _write_derivative( |
| 439 | source_path: Path, |
| 440 | output_path: Path, |
| 441 | manifest_path: Path, |
| 442 | provenance_item: Optional[dict], |
| 443 | args: argparse.Namespace, |
| 444 | ) -> None: |
| 445 | fd, temporary_name = tempfile.mkstemp( |
| 446 | prefix=f".{output_path.stem}.", |
| 447 | suffix=".tmp.png", |
| 448 | dir=str(output_path.parent), |
| 449 | ) |
| 450 | os.close(fd) |
| 451 | temporary_path = Path(temporary_name) |
| 452 | try: |
| 453 | width, height = _apply_treatments(source_path, temporary_path, args) |
| 454 | if provenance_item is not None: |
| 455 | provenance_item["width"] = width |
| 456 | provenance_item["height"] = height |
| 457 | _assert_manifest_output_absent(manifest_path, output_path.name) |
| 458 | _assert_output_absent(output_path) |
| 459 | try: |
| 460 | os.chmod(temporary_path, 0o644) |
| 461 | except OSError: |
| 462 | pass |
| 463 | try: |
| 464 | os.link(temporary_path, output_path) |
| 465 | except FileExistsError as exc: |
| 466 | raise RuntimeError( |
| 467 | f"output appeared during processing and will not be replaced: {output_path}" |
| 468 | ) from exc |
| 469 | if provenance_item is not None: |
| 470 | try: |
| 471 | if not _is_staged_output(temporary_path, output_path): |
| 472 | raise RuntimeError( |
| 473 | f"installed output changed before provenance commit: {output_path}" |
| 474 | ) |
| 475 | _assert_manifest_output_absent(manifest_path, output_path.name) |
| 476 | _write_sources_manifest(manifest_path, provenance_item) |
| 477 | except Exception as exc: |
| 478 | try: |
| 479 | if _is_staged_output(temporary_path, output_path): |
| 480 | output_path.unlink() |
| 481 | except OSError as cleanup_exc: |
| 482 | raise RuntimeError( |
| 483 | f"manifest update failed and output rollback also failed: {cleanup_exc}" |
| 484 | ) from exc |
| 485 | raise RuntimeError(f"manifest update failed; output was rolled back: {exc}") from exc |
| 486 | finally: |
| 487 | try: |
| 488 | temporary_path.unlink() |
| 489 | except FileNotFoundError: |
| 490 | pass |
| 491 | |
| 492 | |
| 493 | def main(argv: Optional[list[str]] = None) -> int: |
| 494 | parser = build_parser() |
| 495 | args = parser.parse_args(argv) |
| 496 | try: |
| 497 | images_path, source_path, output_path = _resolve_paths( |
| 498 | args.project_path, |
| 499 | args.source, |
| 500 | args.output, |
| 501 | ) |
| 502 | treatments = _treatment_plan(args) |
| 503 | manifest_path = images_path / "image_sources.json" |
| 504 | provenance_item = _prepare_provenance( |
| 505 | manifest_path, |
| 506 | args.source, |
| 507 | args.output, |
| 508 | treatments, |
| 509 | ) |
| 510 | _write_derivative( |
| 511 | source_path, |
| 512 | output_path, |
| 513 | manifest_path, |
| 514 | provenance_item, |
| 515 | args, |
| 516 | ) |
| 517 | except (OSError, RuntimeError, ValueError) as exc: |
| 518 | print(f"Error: {exc}", file=sys.stderr) |
| 519 | return 1 |
| 520 | |
| 521 | print(output_path) |
| 522 | return 0 |
| 523 | |
| 524 | |
| 525 | if __name__ == "__main__": |
| 526 | raise SystemExit(main()) |
| 527 |