From f50f4654657783f999e8104607d195747149f6cc Mon Sep 17 00:00:00 2001 From: ajp_anton Date: Tue, 9 Jun 2026 15:07:46 +0000 Subject: [PATCH] Improve grouping controls and cleanup --- grouping.py | 19 ++++++++++++++++- photo_metadata.py | 52 ++++++++++++++++++++++++++++++++++------------- 2 files changed, 56 insertions(+), 15 deletions(-) diff --git a/grouping.py b/grouping.py index 57fa3a4..a1b14be 100644 --- a/grouping.py +++ b/grouping.py @@ -17,7 +17,13 @@ from pathlib import Path from tools.console import pause_if_interactive from tools.filenames import group_stem -from tools.filesystem import collect_files, determine_working_dir, unique_path +from tools.filesystem import ( + cleanup_roots_from_args, + collect_files, + determine_working_dir, + remove_empty_dirs, + unique_path, +) def collect_group_files(args: list[str]) -> list[Path]: @@ -41,6 +47,14 @@ def group_files(files: list[Path], working_dir: Path) -> Path | None: return target_dir +def print_removed_dirs(removed_dirs: list[Path]) -> None: + if not removed_dirs: + return + print("Removed empty director" + ("y:" if len(removed_dirs) == 1 else "ies:")) + for directory in removed_dirs: + print(f" - {directory}") + + def run(argv: list[str]) -> int: if len(argv) < 2: print("Usage: python3 grouping.py path/to/file/or/directory [...]") @@ -52,6 +66,7 @@ def run(argv: list[str]) -> int: print(exc) return 1 + cleanup_roots = cleanup_roots_from_args(argv[1:]) files = collect_group_files(argv[1:]) if not files: print("No files to group.") @@ -60,6 +75,8 @@ def run(argv: list[str]) -> int: target_dir = group_files(files, working_dir) if target_dir is not None: print(f"Moved {len(files)} file(s) into {target_dir}") + removed_dirs = remove_empty_dirs(cleanup_roots, keep_dirs={working_dir.resolve()}) + print_removed_dirs(removed_dirs) return 0 diff --git a/photo_metadata.py b/photo_metadata.py index 7b70f9b..66a8ea7 100644 --- a/photo_metadata.py +++ b/photo_metadata.py @@ -189,6 +189,7 @@ class UserChoices: artist_value: str | None rename_files: bool group_photos: bool + group_min_size: int move_to_working_dir: bool process_pto_files: bool apply_filename_timestamps: bool @@ -207,6 +208,23 @@ def ask_yes_no(prompt: str, default: bool) -> bool: print("Please answer y or n.") +def ask_group_min_size() -> int: + while True: + answer = prompt_input( + "Create automatic burst/HDR groups? Enter=yes, n=no, or minimum group size [2]: " + ).strip().lower() + if not answer or answer in {"y", "yes"}: + return 2 + if answer in {"n", "no"}: + return 0 + if answer.isdigit(): + value = int(answer) + if value <= 0: + return 0 + return max(value, 2) + print("Please answer y, n, or an integer minimum group size.") + + def parse_timezone_offset(value: str) -> timezone: match = TZ_RE.match(value.strip()) if not match: @@ -739,9 +757,10 @@ def prompt_choices(records: list[MediaRecord], working_dir: Path, has_pto_files: rename_files = ask_yes_no("Rename files to timestamps?", default=True) explicit_tz, authoritative_tz, write_missing_tz, video_tz = prompt_timezone(records, rename_files) artist_action, artist_value = prompt_artist_action() - group_photos = False + group_min_size = 0 if sum(1 for record in records if record.is_image and record.adjusted_time) >= 2: - group_photos = ask_yes_no("Create automatic burst/HDR groups?", default=True) + group_min_size = ask_group_min_size() + group_photos = group_min_size >= 2 process_pto_files = False if has_pto_files: @@ -772,6 +791,7 @@ def prompt_choices(records: list[MediaRecord], working_dir: Path, has_pto_files: artist_value=artist_value, rename_files=rename_files, group_photos=group_photos, + group_min_size=group_min_size, move_to_working_dir=move_to_working_dir, process_pto_files=process_pto_files, apply_filename_timestamps=apply_filename_timestamps, @@ -806,7 +826,7 @@ def group_sort_key(record: MediaRecord) -> tuple: ) -def detect_photo_groups(records: list[MediaRecord]) -> list[list[MediaRecord]]: +def detect_photo_groups(records: list[MediaRecord], min_size: int = 2) -> list[list[MediaRecord]]: groups: list[list[MediaRecord]] = [] by_camera: dict[str, list[MediaRecord]] = defaultdict(list) for record in records: @@ -814,12 +834,12 @@ def detect_photo_groups(records: list[MediaRecord]) -> list[list[MediaRecord]]: by_camera[record.camera_key].append(record) for camera_records in by_camera.values(): - groups.extend(detect_groups_for_camera(sorted(camera_records, key=photo_sort_key))) + groups.extend(detect_groups_for_camera(sorted(camera_records, key=photo_sort_key), min_size)) return groups -def detect_groups_for_camera(records: list[MediaRecord]) -> list[list[MediaRecord]]: +def detect_groups_for_camera(records: list[MediaRecord], min_size: int) -> list[list[MediaRecord]]: by_second: dict[datetime, list[MediaRecord]] = defaultdict(list) for record in records: assert record.adjusted_time is not None @@ -843,13 +863,13 @@ def detect_groups_for_camera(records: list[MediaRecord]) -> list[list[MediaRecor continue span_records = [record for second in span for record in by_second[second]] if all(record.sequence is not None for record in span_records) and len(span_records) >= 2: - groups.extend(sequence_groups(span_records)) + groups.extend(sequence_groups(span_records, min_size)) else: - groups.extend(timestamp_groups(span, by_second)) - return [group for group in groups if len(group) >= 2] + groups.extend(timestamp_groups(span, by_second, min_size)) + return [group for group in groups if len(group) >= min_size] -def sequence_groups(records: list[MediaRecord]) -> list[list[MediaRecord]]: +def sequence_groups(records: list[MediaRecord], min_size: int) -> list[list[MediaRecord]]: ordered = sorted(records, key=photo_sort_key) groups: list[list[MediaRecord]] = [] current: list[MediaRecord] = [ordered[0]] @@ -858,16 +878,16 @@ def sequence_groups(records: list[MediaRecord]) -> list[list[MediaRecord]]: if record.sequence is not None and previous.sequence is not None and record.sequence > previous.sequence: current.append(record) else: - if len(current) >= 2: + if len(current) >= min_size: groups.append(current) current = [record] - if len(current) >= 2: + if len(current) >= min_size: groups.append(current) return groups def timestamp_groups( - span: list[datetime], by_second: dict[datetime, list[MediaRecord]] + span: list[datetime], by_second: dict[datetime, list[MediaRecord]], min_size: int ) -> list[list[MediaRecord]]: multi_indices = [index for index, second in enumerate(span) if len(by_second[second]) >= 2] if not multi_indices: @@ -893,7 +913,7 @@ def timestamp_groups( for second in seconds: used_seconds.add(second) records = [record for second in seconds for record in sorted(by_second[second], key=photo_sort_key)] - if len(records) >= 2: + if len(records) >= min_size: groups.append(records) return groups @@ -1448,6 +1468,10 @@ def print_preview(records: list[MediaRecord], choices: UserChoices, pto_plans: l print("\nPreview") print(f"Working directory: {choices.working_dir}") print(f"Move ungrouped files into working directory: {'yes' if choices.move_to_working_dir else 'no'}") + print( + "Automatic burst/HDR grouping: " + + (f"minimum {choices.group_min_size}" if choices.group_photos else "no") + ) print(f"Panorama project files (.pto): {'group/update' if choices.process_pto_files else 'ignore'}") print(f"Time shift: {format_timedelta(choices.time_offset) if choices.time_offset else 'none'}") print( @@ -1571,7 +1595,7 @@ def prepare_plan( group_candidates: list[list[MediaRecord]] = [] if choices.group_photos: - group_candidates.extend(detect_photo_groups(records)) + group_candidates.extend(detect_photo_groups(records, choices.group_min_size)) if choices.process_pto_files: group_candidates.extend(reference.records for reference in pto_references)