167 lines
8.0 KiB
Python
167 lines
8.0 KiB
Python
"""Preview and apply a prepared photo metadata plan."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
from concurrent.futures import Future, ThreadPoolExecutor
|
|
from pathlib import Path
|
|
|
|
from tools.console import ProgressView
|
|
from tools.exiftool import run_exiftool_write
|
|
from tools.filesystem import display_path_from_working_dir, unique_existing_target
|
|
from tools.photo_planning import PtoPlan, UserChoices, build_write_args
|
|
from tools.photo_records import MediaRecord
|
|
from tools.timezones import timezone_to_string
|
|
|
|
|
|
EXIFTOOL_WRITE_WORKERS = 4
|
|
METADATA_WORK_OVERHEAD = 512 * 1024
|
|
FILE_OPERATION_WORK = 16 * 1024
|
|
|
|
|
|
def _time_offset_text(value) -> str:
|
|
seconds = int(value.total_seconds())
|
|
sign, seconds = ("+" if seconds >= 0 else "-"), abs(seconds)
|
|
return f"{sign}{seconds // 3600:02d}:{seconds % 3600 // 60:02d}:{seconds % 60:02d}"
|
|
|
|
|
|
def _display(path: Path, choices: UserChoices, relative: bool) -> str:
|
|
return display_path_from_working_dir(path, choices.working_dir) if relative else str(path)
|
|
|
|
|
|
def move_rename_preview(source: Path, target: Path, choices: UserChoices, label: str | None = None) -> str | None:
|
|
moving, renaming = source.parent.resolve() != target.parent.resolve(), source.name != target.name
|
|
source_text = label or _display(source, choices, moving)
|
|
if moving and renaming:
|
|
return f"move+rename {source_text} -> {_display(target, choices, True)}"
|
|
if moving:
|
|
destination = _display(target.parent, choices, True)
|
|
return f"move {source_text} -> {destination if destination == '.' or destination.endswith(os.sep) else destination + os.sep}"
|
|
return f"rename {label or source.name} -> {target.name}" if renaming else None
|
|
|
|
|
|
def print_preview(records: list[MediaRecord], choices: UserChoices, pto_plans: list[PtoPlan]) -> None:
|
|
print("\nPreview\n\nCommon changes")
|
|
print(f" Working directory: {choices.working_dir}")
|
|
print(f" Time correction: {_time_offset_text(choices.time_offset) if choices.time_offset else 'none'}")
|
|
if choices.fixed_timezone:
|
|
print(f" Timezone offset: {timezone_to_string(choices.fixed_timezone)}")
|
|
elif choices.fill_timezone_gaps:
|
|
print(" Timezone offsets: fill missing gaps")
|
|
print(f" Filename mode: {choices.rename_mode.replace('_', '/')}")
|
|
print(f" Organize/group: {'yes' if choices.organize_files else 'no'}")
|
|
if choices.organize_files:
|
|
print(f" Burst/HDR minimum: {choices.group_min_size}+ files")
|
|
print(f" Panorama project files: {'yes' if choices.process_pto_files else 'no'}")
|
|
if choices.artist_action != "leave":
|
|
print(f" Artist/author: {'clear' if choices.artist_action == 'clear' else 'set'}")
|
|
inferred = [record for record in records if record.inferred_timestamp_source]
|
|
if inferred:
|
|
print("\nInferred timestamps")
|
|
for record in inferred:
|
|
print(f" {record.path.name}: {record.adjusted_time} from {record.inferred_timestamp_source}")
|
|
warnings = [(record.path.name, warning) for record in records for warning in record.warnings]
|
|
if warnings:
|
|
print("\nWarnings")
|
|
for name, warning in warnings:
|
|
print(f" {name}: {warning}")
|
|
print("\nFile operations")
|
|
for record in sorted(records, key=lambda item: str(item.path).lower()):
|
|
operation = move_rename_preview(record.path, record.target_path or record.path, choices)
|
|
if operation:
|
|
print(operation)
|
|
if record.rename_skip_reason:
|
|
print(f" {record.path.name}: rename skipped: {record.rename_skip_reason}")
|
|
if record.target_sidecar and record.sidecar:
|
|
print(f" sidecar move {_display(record.sidecar.path, choices, True)} -> {_display(record.target_sidecar, choices, True)}")
|
|
if pto_plans:
|
|
print("\nPTO updates:")
|
|
for plan in pto_plans:
|
|
operation = move_rename_preview(plan.path, plan.target_path, choices, _display(plan.path, choices, True))
|
|
print(operation or _display(plan.path, choices, True))
|
|
for old, new in plan.replacements:
|
|
print(f" {old} -> {new}")
|
|
|
|
|
|
def _operation(record: MediaRecord) -> str:
|
|
target = record.target_path or record.path
|
|
if target.parent.resolve() != record.path.parent.resolve():
|
|
return "move+rename" if target.name != record.path.name else "move"
|
|
return "rename" if target.name != record.path.name else "no-op"
|
|
|
|
|
|
def _work(record: MediaRecord, args: list[str]) -> float:
|
|
if not args:
|
|
return FILE_OPERATION_WORK
|
|
try:
|
|
return FILE_OPERATION_WORK + METADATA_WORK_OVERHEAD + record.path.stat().st_size
|
|
except OSError:
|
|
return FILE_OPERATION_WORK + METADATA_WORK_OVERHEAD
|
|
|
|
|
|
def _move_record(record: MediaRecord, planned_sources: set[Path]) -> None:
|
|
if record.target_path and record.target_path != record.path:
|
|
target = unique_existing_target(record.target_path, planned_sources)
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.move(str(record.path), str(target))
|
|
record.target_path = target
|
|
if record.sidecar and record.target_sidecar and record.sidecar.path.exists():
|
|
target = unique_existing_target(record.target_sidecar, planned_sources)
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.move(str(record.sidecar.path), str(target))
|
|
|
|
|
|
def apply_changes(executable: str, records: list[MediaRecord], choices: UserChoices) -> list[str]:
|
|
failures, sources = [], {record.path.resolve() for record in records}
|
|
args = {record.path: build_write_args(record, choices) for record in records}
|
|
work = {record.path: _work(record, args[record.path]) for record in records}
|
|
progress = ProgressView(len(records), "Progress", total_work=sum(work.values()), show_elapsed=False, bar_width=16)
|
|
counts = {key: 0 for key in ("succeeded", "failed", "metadata", "rename", "move", "move+rename", "no-op")}
|
|
completed = 0.0
|
|
with ThreadPoolExecutor(max_workers=EXIFTOOL_WRITE_WORKERS) as workers:
|
|
futures: dict[Path, Future[None]] = {record.path: workers.submit(run_exiftool_write, executable, record.path, write_args) for record in records if (write_args := args[record.path])}
|
|
for index, record in enumerate(records, 1):
|
|
operation = _operation(record)
|
|
try:
|
|
if args[record.path]:
|
|
futures[record.path].result()
|
|
counts["metadata"] += 1
|
|
_move_record(record, sources)
|
|
counts["succeeded"] += 1
|
|
counts[operation] += 1 if operation != "no-op" or not args[record.path] else 0
|
|
except (OSError, subprocess.CalledProcessError, RuntimeError) as exc:
|
|
counts["failed"] += 1
|
|
failures.append(f"{record.path}: {exc}")
|
|
completed += work[record.path]
|
|
progress.write_lines([
|
|
"Applying changes",
|
|
f" {progress.update_line(index, completed_work=completed)}",
|
|
f" Results: succeeded {counts['succeeded']} failed {counts['failed']} metadata {counts['metadata']} no-op {counts['no-op']}",
|
|
f" File ops: rename {counts['rename']} move {counts['move']} rename+move {counts['move+rename']}",
|
|
])
|
|
progress.finish(keep=True)
|
|
return failures
|
|
|
|
|
|
def apply_pto_changes(plans: list[PtoPlan]) -> list[str]:
|
|
failures = []
|
|
for plan in plans:
|
|
try:
|
|
plan.path.write_text(plan.updated_text, encoding="utf-8")
|
|
if plan.target_path.resolve() != plan.path.resolve():
|
|
target = unique_existing_target(plan.target_path, {plan.path.resolve()})
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.move(str(plan.path), str(target))
|
|
except OSError as exc:
|
|
failures.append(f"{plan.path}: {exc}")
|
|
return failures
|
|
|
|
|
|
def print_removed_dirs(directories: list[Path]) -> None:
|
|
if directories:
|
|
print("\nRemoved empty director" + ("y:" if len(directories) == 1 else "ies:"))
|
|
for directory in directories:
|
|
print(f" - {directory}")
|