ajp_anton/media-batch-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improve grouping controls and cleanup

Browse Source
This commit is contained in:
ajp_anton
2026-06-09 15:07:46 +00:00
parent b6024d8b65
commit f50f465465
2 changed files with 56 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
grouping.py
+18 -1
View File
@@ -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
photo_metadata.py
+38 -14
View File
@@ -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)