364 lines
9.8 KiB
Python
Executable File
364 lines
9.8 KiB
Python
Executable File
#!/usr/bin/env -S uv run --script
|
|
# /// script
|
|
# dependencies = ["click"]
|
|
# ///
|
|
|
|
import logging
|
|
import random
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import click
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s %(levelname)s: %(message)s",
|
|
)
|
|
logger = logging.getLogger(name=__name__)
|
|
|
|
VALID_EXTENSIONS = {".jpeg", ".jpg", ".png", ".webm", ".webp"}
|
|
|
|
|
|
def is_valid_image(file_path: Path) -> bool:
|
|
"""Check if file is a valid image type."""
|
|
return file_path.suffix.lower() in VALID_EXTENSIONS
|
|
|
|
|
|
def move_to_trash(file_paths: list[Path]) -> None:
|
|
"""Move files to Trash using trash CLI (single batched call)."""
|
|
resolved: list[str] = []
|
|
for file_path in file_paths:
|
|
if not file_path.is_file():
|
|
raise FileNotFoundError(f"File not found: {file_path}")
|
|
resolved.append(str(file_path.resolve()))
|
|
|
|
if resolved:
|
|
subprocess.run(["trash", *resolved], check=True)
|
|
|
|
|
|
def build_target_file_index(target_dirs: list[Path]) -> dict[str, Path]:
|
|
"""
|
|
Build a mapping of file stems to paths from target directories.
|
|
|
|
Scans target directories once and returns a dict for efficient lookups.
|
|
"""
|
|
index: dict[str, Path] = {}
|
|
|
|
for target_dir in target_dirs:
|
|
if not target_dir.is_dir():
|
|
continue
|
|
|
|
for candidate in target_dir.iterdir():
|
|
if not candidate.is_file():
|
|
continue
|
|
|
|
if not is_valid_image(file_path=candidate):
|
|
continue
|
|
|
|
index[candidate.stem] = candidate
|
|
|
|
return index
|
|
|
|
|
|
def find_edited_version(
|
|
image_path: Path,
|
|
target_index: dict[str, Path],
|
|
) -> Path | None:
|
|
"""
|
|
Search target file index for edited version of image.
|
|
|
|
An image is considered edited if its stem starts with the original stem
|
|
followed by '-edit'. Matches by stem only, ignoring extension.
|
|
|
|
Returns the path to the edited version or None if not found.
|
|
"""
|
|
original_stem = image_path.stem
|
|
search_pattern = f"{original_stem}-edit"
|
|
|
|
for indexed_stem, indexed_path in target_index.items():
|
|
if indexed_stem.startswith(search_pattern):
|
|
return indexed_path
|
|
|
|
return None
|
|
|
|
|
|
def build_directory_index(directory: Path) -> set[str]:
|
|
"""
|
|
Build a set of all valid image stems in a directory.
|
|
|
|
Args:
|
|
directory: Path to the directory to scan
|
|
|
|
Returns:
|
|
Set of file stems that are valid images
|
|
"""
|
|
stems: set[str] = set()
|
|
|
|
if not directory.is_dir():
|
|
return stems
|
|
|
|
for candidate in directory.iterdir():
|
|
if not candidate.is_file():
|
|
continue
|
|
|
|
if not is_valid_image(file_path=candidate):
|
|
continue
|
|
|
|
stems.add(candidate.stem)
|
|
|
|
return stems
|
|
|
|
|
|
def collect_images(paths: list[Path]) -> list[Path]:
|
|
images: list[Path] = []
|
|
for path in paths:
|
|
if path.is_file():
|
|
if is_valid_image(path):
|
|
images.append(path)
|
|
elif path.is_dir():
|
|
for candidate in sorted(path.iterdir()):
|
|
if candidate.is_file() and is_valid_image(candidate):
|
|
images.append(candidate)
|
|
return images
|
|
|
|
|
|
@click.group()
|
|
def cli() -> None:
|
|
"""Manage and process image files."""
|
|
pass
|
|
|
|
|
|
@cli.command()
|
|
@click.argument(
|
|
"image_paths",
|
|
nargs=-1,
|
|
required=True,
|
|
type=click.Path(exists=True, path_type=Path),
|
|
)
|
|
@click.option(
|
|
"--target-dir",
|
|
multiple=True,
|
|
required=True,
|
|
type=click.Path(exists=True, path_type=Path),
|
|
help="Directory to search for edited versions",
|
|
)
|
|
@click.option(
|
|
"--dry-run",
|
|
is_flag=True,
|
|
default=False,
|
|
help="Preview matches without moving to trash",
|
|
)
|
|
def dedupe(
|
|
image_paths: tuple[Path, ...],
|
|
target_dir: tuple[Path, ...],
|
|
dry_run: bool,
|
|
) -> None:
|
|
"""
|
|
Remove original images when edited versions exist in target directories.
|
|
|
|
IMAGE_PATHS: One or more image files to check (jpeg, jpg, png, webm, webp)
|
|
"""
|
|
target_dirs = list(target_dir)
|
|
valid_images = [path for path in image_paths if is_valid_image(file_path=path)]
|
|
|
|
if not valid_images:
|
|
logger.warning("No valid image files found")
|
|
return
|
|
|
|
target_index = build_target_file_index(target_dirs=target_dirs)
|
|
matches: list[tuple[Path, Path]] = []
|
|
|
|
for image_path in valid_images:
|
|
edited_version = find_edited_version(
|
|
image_path=image_path,
|
|
target_index=target_index,
|
|
)
|
|
|
|
if edited_version:
|
|
matches.append((image_path, edited_version))
|
|
logger.info(f"Found edited version: {image_path} -> {edited_version}")
|
|
|
|
if not matches:
|
|
logger.info("No edited versions found")
|
|
return
|
|
|
|
if dry_run:
|
|
click.echo(
|
|
click.style(
|
|
text="DRY RUN MODE - No files will be moved",
|
|
fg="yellow",
|
|
)
|
|
)
|
|
click.echo(f"Would move {len(matches)} image(s) to trash:")
|
|
for original, edited in matches:
|
|
click.echo(f" {original}")
|
|
return
|
|
|
|
originals = [original for original, _ in matches]
|
|
try:
|
|
move_to_trash(file_paths=originals)
|
|
for original in originals:
|
|
logger.info(f"Moved to trash: {original}")
|
|
except (FileNotFoundError, subprocess.CalledProcessError) as e:
|
|
logger.error(f"Failed to move files to trash: {e}")
|
|
|
|
click.echo(
|
|
click.style(
|
|
text=f"Moved {len(matches)} image(s) to trash",
|
|
fg="green",
|
|
)
|
|
)
|
|
|
|
|
|
@cli.command()
|
|
@click.argument(
|
|
"paths",
|
|
nargs=-1,
|
|
required=True,
|
|
type=click.Path(exists=True, path_type=Path),
|
|
)
|
|
@click.option(
|
|
"--dry-run",
|
|
is_flag=True,
|
|
default=False,
|
|
help="Preview matches without moving to trash",
|
|
)
|
|
def clean(paths: tuple[Path, ...], dry_run: bool) -> None:
|
|
"""
|
|
Remove originals when upscaled versions exist in the same basket.
|
|
|
|
PATHS: Image files and/or directories to scan (jpeg, jpg, png, webm, webp)
|
|
|
|
Scans all provided paths into a collection. If a file and its -upscaled
|
|
counterpart (stem starts with '{original}-upscaled', any extension)
|
|
are both in the collection, the original is moved to trash.
|
|
"""
|
|
all_images = collect_images(list(paths))
|
|
|
|
if not all_images:
|
|
logger.warning("No valid image files found")
|
|
return
|
|
|
|
image_index: dict[str, Path] = {img.stem: img for img in all_images}
|
|
trashables: set[Path] = set()
|
|
|
|
markers = ["-upscaled", "-edit"]
|
|
for image in all_images:
|
|
if any(marker in image.stem for marker in markers):
|
|
continue
|
|
|
|
upscaleds = [
|
|
p
|
|
for s, p in image_index.items()
|
|
if s.startswith(image.stem) and "-upscaled" in s
|
|
]
|
|
editeds = [
|
|
p
|
|
for s, p in image_index.items()
|
|
if s.startswith(image.stem) and "-edit" in s
|
|
]
|
|
if editeds or upscaleds:
|
|
logger.info(f"Will delete {image}")
|
|
trashables.add(image)
|
|
for upscaled in upscaleds:
|
|
low_res_edit = upscaled.with_stem(
|
|
upscaled.stem.removesuffix("-upscaled")
|
|
)
|
|
if low_res_edit in editeds:
|
|
logger.info(f"Will also delete low-res edit {low_res_edit}")
|
|
trashables.add(low_res_edit)
|
|
|
|
if not trashables:
|
|
logger.info("No upscaled or edited versions found")
|
|
return
|
|
|
|
if dry_run:
|
|
click.echo(click.style("DRY RUN MODE - No files will be moved", fg="yellow"))
|
|
click.echo(f"Would move {len(trashables)} image(s) to trash:")
|
|
for original in trashables:
|
|
click.echo(f" {original}")
|
|
return
|
|
|
|
try:
|
|
move_to_trash(file_paths=list(trashables))
|
|
for original in trashables:
|
|
logger.info(f"Moved to trash: {original}")
|
|
except (FileNotFoundError, subprocess.CalledProcessError) as e:
|
|
logger.error(f"Failed to move files to trash: {e}")
|
|
|
|
click.echo(click.style(f"Moved {len(trashables)} image(s) to trash", fg="green"))
|
|
|
|
|
|
@cli.command(name="list-editable")
|
|
@click.argument(
|
|
"image_paths",
|
|
nargs=-1,
|
|
required=True,
|
|
type=click.Path(exists=True, path_type=Path),
|
|
)
|
|
@click.option(
|
|
"--shuffle",
|
|
is_flag=True,
|
|
default=False,
|
|
help="Shuffle the output filenames",
|
|
)
|
|
@click.option(
|
|
"--null",
|
|
is_flag=True,
|
|
default=False,
|
|
help="Output null-terminated filenames for xargs -0",
|
|
)
|
|
def list_editable(image_paths: tuple[Path, ...], null: bool, shuffle: bool) -> None:
|
|
"""
|
|
List image files that don't have edited versions.
|
|
|
|
IMAGE_PATHS: One or more image files to check (jpeg, jpg, png, webm, webp)
|
|
|
|
Skips files with '-edit' in their stem and lists those without
|
|
matching edited versions (stem + '-edit' + optional suffix) in
|
|
the same directory.
|
|
"""
|
|
valid_images = [path for path in image_paths if is_valid_image(file_path=path)]
|
|
|
|
if not valid_images:
|
|
logger.warning("No valid image files found")
|
|
return
|
|
|
|
dir_indices: dict[Path, set[str]] = {}
|
|
editable_files: list[Path] = []
|
|
|
|
for image_path in valid_images:
|
|
if "-edit" in image_path.stem:
|
|
continue
|
|
|
|
directory = image_path.parent
|
|
if directory not in dir_indices:
|
|
dir_indices[directory] = build_directory_index(directory=directory)
|
|
|
|
dir_stems = dir_indices[directory]
|
|
original_stem = image_path.stem
|
|
search_pattern = f"{original_stem}-edit"
|
|
|
|
has_edited = any(stem.startswith(search_pattern) for stem in dir_stems)
|
|
|
|
if not has_edited:
|
|
editable_files.append(image_path)
|
|
|
|
if not editable_files:
|
|
logger.info("No editable files found")
|
|
return
|
|
|
|
if shuffle:
|
|
random.shuffle(editable_files)
|
|
|
|
for file_path in editable_files:
|
|
if null:
|
|
click.echo(str(file_path), nl=False)
|
|
click.echo("\0", nl=False)
|
|
else:
|
|
click.echo(str(file_path))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
cli()
|