From eb5937a6297beb956e365456441656a9d3306469 Mon Sep 17 00:00:00 2001 From: ajp_anton Date: Mon, 24 Aug 2026 16:34:20 +0000 Subject: [PATCH] Refactor photo metadata workflow --- README.md | 11 +- photo_metadata.py | 1635 +---------------------- tools/console.py | 209 ++- tools/exiftool.py | 92 +- tools/media_metadata.py | 63 + tools/metadata_copy.py | 52 +- tools/photo_execution.py | 166 +++ tools/photo_grouping.py | 136 ++ tools/photo_planning.py | 383 ++++++ tools/photo_records.py | 251 ++++ tools/photo_settings.py | 543 ++++++++ tools/video_probe.py | 47 +- unit_tests/test_exiftool.py | 34 +- unit_tests/test_photo_metadata_logic.py | 148 ++ unit_tests/test_video_logic.py | 68 +- 15 files changed, 2111 insertions(+), 1727 deletions(-) create mode 100644 tools/media_metadata.py create mode 100644 tools/photo_execution.py create mode 100644 tools/photo_grouping.py create mode 100644 tools/photo_planning.py create mode 100644 tools/photo_records.py create mode 100644 tools/photo_settings.py create mode 100644 unit_tests/test_photo_metadata_logic.py diff --git a/README.md b/README.md index df32af4..ff6925c 100644 --- a/README.md +++ b/README.md @@ -20,23 +20,26 @@ Current features: - reads and writes metadata through `exiftool` from `PATH` - optional direct time shifts, such as `+1:30`, `-02:00:00`, or `+1h 2m` - optional reference-clock correction using a source photo timestamp or filename plus the correct time -- timezone handling for photos and local-time video filenames +- keyboard-driven settings table with timezone gap filling and optional fixed timezone offsets - artist/author set, clear, or leave unchanged -- timestamp-based renaming to `YYYYMMDD_HHMMSS.ext` +- timestamp-based rename modes: adjust timestamps already in filenames, add one, or replace the stem - video filenames use the beginning timestamp by subtracting rounded duration from the video metadata timestamp - Sony-style video sidecars like `C0011.MP4` plus `C0011M01.XML` are moved and renamed with the video - automatic photo burst/HDR grouping per camera - optional panorama project (`.pto`) grouping and reference updates -- inferred subsecond metadata for grouped same-second photo bursts +- inferred subsecond metadata for grouped same-second photo bursts when the group has no existing subseconds +- optional capture-time inference from a safe timestamp filename, then `ModifyDate` as a fallback The working directory is determined from the dragged items. A single dragged folder is the working directory. For multiple dragged files/folders, their last common ancestor is the working directory. Automatic burst/HDR and `.pto` group folders are created inside that working directory. -If dragged folders contain files in subdirectories and grouping is enabled, ungrouped files are moved into the working directory too. This makes the run organize everything into a deterministic flat reset state, with grouped files in their group folders and the remaining files directly in the working directory. If grouping is disabled, the script asks whether to flatten subdirectory files into the working directory. +The `Organize/group` setting controls all file movement. When enabled, ungrouped files are moved into the working directory and groups are created there; when disabled, files remain in their original directories. Panorama-project grouping is also enabled from that row. Panorama project files use the `.pto` extension. When enabled, every media file referenced by a `.pto` file belongs to the same group. If those files already belong to burst/HDR groups, the groups are merged. `.pto` files are moved into the group folder with their referenced files, renamed from the first and last referenced media stems, and their file references are rewritten to the new relative paths. Reference-clock correction is useful when one photo shows a reliable clock. For example, if `DSC01234.JPG` has camera metadata `2026:06:02 22:10:00`, but the clock in the photo shows `22:13:25`, choose reference mode, enter `DSC01234.JPG` as the source, and enter `22:13:25` as the correct time. The script computes a `+00:03:25` shift and applies that correction to the selected files. If you omit the date in the correct time, the script chooses the date closest to the source timestamp, within 12 hours. +The settings table requires an ANSI-capable terminal because it uses arrow keys and inline editing. Use Up/Down to select a row, Left/Right to select an option, Space to toggle an option, and Enter to continue. The preview separates common changes, inferred timestamps, warnings, and file operations before asking for confirmation. + ### `grouping.py` Fast manual grouping for selected files: diff --git a/photo_metadata.py b/photo_metadata.py index 6828335..f8f6cab 100644 --- a/photo_metadata.py +++ b/photo_metadata.py @@ -11,1571 +11,31 @@ place only after confirmation. from __future__ import annotations -import math -import os -import re -import shutil +import json import subprocess import sys -import xml.etree.ElementTree as ET -from collections import defaultdict -from dataclasses import dataclass, field -from datetime import datetime, timedelta, timezone -from pathlib import Path -from typing import Iterable -from tools.console import pause_if_interactive, prompt_input, prompt_yes_no as ask_yes_no -from tools.exiftool import require_exiftool, run_exiftool_json, run_exiftool_write -from tools.filenames import ( - format_filename_stem, - group_stem_from_name, - naive_wall_time, - normalized_extension, +from tools.console import ( + ProgressView, + clear_screen, + pause_if_interactive, + prompt_yes_no as ask_yes_no, ) +from tools.exiftool import require_exiftool, run_exiftool_json +from tools.photo_planning import ( + plan_pto_updates, + prepare_plan, + read_pto_references, +) +from tools.photo_execution import apply_changes, apply_pto_changes, print_preview, print_removed_dirs +from tools.photo_records import SUPPORTED_EXTS, build_records +from tools.photo_settings import prompt_choices from tools.filesystem import ( cleanup_roots_from_args, collect_files, determine_working_dir, - display_path_from_working_dir, remove_empty_dirs, - unique_existing_target, ) -from tools.timezones import parse_timezone_offset, timezone_to_string - - -IMAGE_EXTS = {".jpg", ".jpeg", ".heic", ".arw"} -VIDEO_EXTS = {".mp4", ".mov", ".mts"} -SUPPORTED_EXTS = IMAGE_EXTS | VIDEO_EXTS - -IMAGE_TIME_TAGS = ( - "EXIF:DateTimeOriginal", - "EXIF:CreateDate", - "EXIF:DateTime", - "EXIF:ModifyDate", - "XMP:DateTimeOriginal", - "XMP:CreateDate", -) -IMAGE_OFFSET_TAGS = ( - "EXIF:OffsetTimeOriginal", - "EXIF:OffsetTimeDigitized", - "EXIF:OffsetTime", -) -VIDEO_TIME_TAGS = ( - "QuickTime:MediaCreateDate", - "QuickTime:TrackCreateDate", - "QuickTime:CreateDate", - "QuickTime:MediaModifyDate", - "QuickTime:TrackModifyDate", - "QuickTime:ModifyDate", - "XMP:DateTimeOriginal", - "XMP:CreateDate", -) -VIDEO_BEGIN_TIME_TAGS = ( - "H264:DateTimeOriginal", -) -VIDEO_DURATION_TAGS = ( - "QuickTime:Duration", - "Composite:Duration", - "File:Duration", - "M2TS:Duration", -) -SEQUENCE_TAGS = ( - "MakerNotes:SequenceNumber", - "MakerNotes:SequenceNumberOriginal", - "MakerNotes:ImageNumber", - "EXIF:ImageNumber", - "SequenceNumber", -) - -DATETIME_RE = re.compile( - r"(?P\d{4})[:\-]?(?P\d{2})[:\-]?(?P\d{2})" - r"(?:[ T_])?" - r"(?P\d{2}):?(?P\d{2}):?(?P\d{2})" - r"(?:[.,](?P\d+))?" - r"(?:\s*(?PZ|[+-]\d{2}:?\d{2}))?" -) -TIME_ONLY_RE = re.compile( - r"^(?P\d{1,2})(?::?(?P\d{2}))(?::?(?P\d{2}))?$" -) -OFFSET_RE = re.compile(r"^([+-])(?:(\d+):)?(\d{1,2})(?::(\d{2}))?$") -TOKEN_SHIFT_RE = re.compile(r"(\d+|[a-zA-Z]+)") -ORDERED_TIMESTAMP_STEM_RE = re.compile(r"^(?P\d{8}_\d{6})-[1-9]\d*$") -TIMESTAMP_RANGE_STEM_RE = re.compile( - r"^(?P\d{8}_\d{6})-(?P\d{8}_\d{6})(?P_.*)?$" -) -CAMERA_RANGE_STEM_RE = re.compile( - r"^(?P[A-Za-z]*)(?P\d+)-(?P=prefix)(?P\d+)(?P_.*)$" -) - - -@dataclass -class XmlSidecar: - path: Path - creation: datetime | None = None - duration_seconds: float | None = None - timezone_value: timezone | None = None - device: str | None = None - - -@dataclass -class TagWrite: - label: str - arg: str - current_value: str | None - new_value: str - will_write: bool - - -@dataclass -class MediaRecord: - path: Path - metadata: dict - kind: str - original_time: datetime | None = None - adjusted_time: datetime | None = None - timezone_value: timezone | None = None - subsec: int | None = None - inferred_subsec: int | None = None - sequence: int | None = None - camera_key: str = "" - duration_seconds: float | None = None - video_time_is_beginning: bool = False - parsed_from_filename: bool = False - sidecar: XmlSidecar | None = None - new_stem: str | None = None - final_name: str | None = None - group_id: int | None = None - group_name: str | None = None - target_path: Path | None = None - target_sidecar: Path | None = None - rename_skip_reason: str | None = None - write_plan: list[TagWrite] = field(default_factory=list) - warnings: list[str] = field(default_factory=list) - - @property - def is_image(self) -> bool: - return self.kind == "image" - - @property - def is_video(self) -> bool: - return self.kind == "video" - - -@dataclass -class PtoPlan: - path: Path - target_path: Path - referenced_records: list[MediaRecord] - updated_text: str - replacements: list[tuple[str, str]] - - -@dataclass -class PtoReference: - path: Path - original_text: str - records: list[MediaRecord] - - -@dataclass -class UserChoices: - working_dir: Path - time_offset: timedelta | None - explicit_timezone: timezone | None - authoritative_timezone: timezone | None - write_missing_photo_timezones: bool - video_filename_timezone: timezone | None - artist_action: str - 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 - - -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_datetime(value: str, default_date: datetime | None = None) -> datetime: - text = value.strip() - match = DATETIME_RE.search(text) - if match: - tzinfo = None - tz_text = match.group("tz") - if tz_text == "Z": - tzinfo = timezone.utc - elif tz_text: - tzinfo = parse_timezone_offset(tz_text) - return datetime( - int(match.group("Y")), - int(match.group("M")), - int(match.group("D")), - int(match.group("h")), - int(match.group("m")), - int(match.group("s")), - tzinfo=tzinfo, - ) - - match = TIME_ONLY_RE.match(text) - if match and default_date is not None: - return default_date.replace( - hour=int(match.group("h")), - minute=int(match.group("m")), - second=int(match.group("s") or "0"), - microsecond=0, - ) - - raise ValueError("timestamp must include YYYYMMDD_HHMMSS or a time with reference date") - - -def parse_exif_datetime(value: object, assume_utc: bool = False) -> datetime | None: - if value is None: - return None - try: - dt = parse_datetime(str(value)) - except ValueError: - return None - if assume_utc and dt.tzinfo is None: - dt = dt.replace(tzinfo=timezone.utc) - return dt - - -def parse_subsec(value: object) -> int | None: - if value is None: - return None - match = re.search(r"\d+", str(value)) - if not match: - return None - digits = match.group(0)[:3].ljust(3, "0") - return int(digits) - - -def parse_sequence(value: object) -> int | None: - if value is None: - return None - match = re.search(r"\d+", str(value)) - if not match: - return None - return int(match.group(0)) - - -def parse_duration_seconds(value: object) -> float | None: - if value is None: - return None - text = str(value).strip() - - number_match = re.match(r"^([0-9]+(?:\.[0-9]+)?)\s*(?:s|sec|seconds)?$", text, re.I) - if number_match: - return float(number_match.group(1)) - - hms_match = re.match( - r"^(?:(?P\d+):)?(?P\d{1,2}):(?P\d{1,2}(?:\.\d+)?)$", text - ) - if hms_match: - hours = int(hms_match.group("h") or "0") - minutes = int(hms_match.group("m")) - seconds = float(hms_match.group("s")) - return hours * 3600 + minutes * 60 + seconds - - embedded = re.search(r"([0-9]+(?:\.[0-9]+)?)\s*s", text, re.I) - if embedded: - return float(embedded.group(1)) - - return None - - -def round_half_up(value: float) -> int: - return int(math.floor(value + 0.5)) - - -def parse_timeshift(value: str) -> timedelta: - text = value.strip().replace(" ", "") - if not text or text[0] not in "+-": - raise ValueError("time shift must start with + or -") - sign = 1 if text[0] == "+" else -1 - body = text[1:] - - if re.fullmatch(r"\d+(?::\d{2}){0,2}", body): - parts = [int(part) for part in body.split(":")] - parts = [0] * (3 - len(parts)) + parts - return sign * timedelta(hours=parts[0], minutes=parts[1], seconds=parts[2]) - - units = { - "h": 3600, - "hr": 3600, - "hrs": 3600, - "hour": 3600, - "hours": 3600, - "m": 60, - "min": 60, - "mins": 60, - "minute": 60, - "minutes": 60, - "s": 1, - "sec": 1, - "secs": 1, - "second": 1, - "seconds": 1, - } - tokens = TOKEN_SHIFT_RE.findall(body) - if not tokens or len(tokens) % 2 != 0: - raise ValueError("time shift units must look like +1h30m or -2 minutes") - seconds = 0 - for amount, unit in zip(tokens[::2], tokens[1::2]): - if not amount.isdigit() or unit.lower() not in units: - raise ValueError("invalid time shift unit") - seconds += int(amount) * units[unit.lower()] - return sign * timedelta(seconds=seconds) - - -def collect_media_files(args: list[str]) -> list[Path]: - return collect_files(args, allowed_exts=SUPPORTED_EXTS, print_missing=True) - - -def collect_pto_files(args: list[str]) -> list[Path]: - return collect_files(args, allowed_exts={".pto"}) - - -def first_tag(metadata: dict, tags: Iterable[str]) -> object | None: - for tag in tags: - if tag in metadata: - return metadata[tag] - return None - - -def tag_value(metadata: dict, tag: str) -> str | None: - value = metadata.get(tag) - if value is None: - return None - return str(value) - - -def same_text_value(current: str | None, new_value: str) -> bool: - return (current or "") == new_value - - -def same_datetime_value(current: str | None, new_value: str) -> bool: - if current is None: - return False - current_dt = parse_exif_datetime(current) - new_dt = parse_exif_datetime(new_value) - if current_dt is None or new_dt is None: - return current == new_value - return naive_wall_time(current_dt) == naive_wall_time(new_dt) - - -def same_subsec_value(current: str | None, new_value: str) -> bool: - if current is None: - return False - current_subsec = parse_subsec(current) - if current_subsec is None: - return current == new_value - return f"{current_subsec:03d}" == new_value - - -def add_tag_write( - writes: list[TagWrite], - label: str, - arg_name: str, - current_value: str | None, - new_value: str, - same_value=same_text_value, -) -> None: - writes.append( - TagWrite( - label=label, - arg=f"-{arg_name}={new_value}", - current_value=current_value, - new_value=new_value, - will_write=not same_value(current_value, new_value), - ) - ) - - -def find_sidecar(video_path: Path) -> XmlSidecar | None: - expected = f"{video_path.stem}M01.XML" - for child in video_path.parent.iterdir(): - if child.is_file() and child.name.lower() == expected.lower(): - return parse_xml_sidecar(child) - return None - - -def parse_xml_sidecar(path: Path) -> XmlSidecar: - sidecar = XmlSidecar(path=path) - try: - root = ET.parse(path).getroot() - except ET.ParseError as exc: - sidecar.device = f"unreadable XML: {exc}" - return sidecar - - duration_value: float | None = None - fps_value: float | None = None - - def strip_ns(name: str) -> str: - return name.rsplit("}", 1)[-1] - - for element in root.iter(): - name = strip_ns(element.tag) - if name == "CreationDate": - value = element.attrib.get("value") - if value: - sidecar.creation = parse_exif_datetime(value) - if sidecar.creation and sidecar.creation.tzinfo: - sidecar.timezone_value = sidecar.creation.tzinfo - elif name == "Duration": - value = element.attrib.get("value") - if value: - duration_value = parse_duration_seconds(value) - elif name == "VideoFrame": - fps_text = element.attrib.get("captureFps") or element.attrib.get("formatFps") - if fps_text: - fps_match = re.search(r"\d+(?:\.\d+)?", fps_text) - if fps_match: - fps_value = float(fps_match.group(0)) - elif name == "Device": - manufacturer = element.attrib.get("manufacturer") - model = element.attrib.get("modelName") - sidecar.device = " ".join(part for part in (manufacturer, model) if part) - - if duration_value is not None and fps_value: - sidecar.duration_seconds = duration_value / fps_value - else: - sidecar.duration_seconds = duration_value - - return sidecar - - -def build_records(paths: list[Path], metadata_items: list[dict]) -> list[MediaRecord]: - by_source = { - Path(item.get("SourceFile", "")).resolve(): item - for item in metadata_items - if item.get("SourceFile") - } - records: list[MediaRecord] = [] - - for path in paths: - metadata = by_source.get(path.resolve(), {}) - ext = path.suffix.lower() - kind = "video" if ext in VIDEO_EXTS else "image" - record = MediaRecord(path=path, metadata=metadata, kind=kind) - - if record.is_image: - record.original_time = parse_exif_datetime(first_tag(metadata, IMAGE_TIME_TAGS)) - record.adjusted_time = record.original_time - offset_value = first_tag(metadata, IMAGE_OFFSET_TAGS) - if offset_value: - try: - record.timezone_value = parse_timezone_offset(str(offset_value)) - except ValueError: - record.warnings.append(f"Invalid timezone offset metadata: {offset_value}") - record.subsec = parse_subsec( - first_tag( - metadata, - ( - "EXIF:SubSecTimeOriginal", - "EXIF:SubSecTimeDigitized", - "EXIF:SubSecTime", - ), - ) - ) - else: - begin_time_value = first_tag(metadata, VIDEO_BEGIN_TIME_TAGS) - if ext == ".mts" and begin_time_value is not None: - record.original_time = parse_exif_datetime(begin_time_value, assume_utc=True) - record.video_time_is_beginning = True - else: - record.original_time = parse_exif_datetime( - first_tag(metadata, VIDEO_TIME_TAGS), assume_utc=True - ) - record.adjusted_time = record.original_time - record.timezone_value = timezone.utc - record.duration_seconds = parse_duration_seconds(first_tag(metadata, VIDEO_DURATION_TAGS)) - record.sidecar = find_sidecar(path) - if record.sidecar and record.duration_seconds is not None: - xml_duration = record.sidecar.duration_seconds - if xml_duration is not None and abs(xml_duration - record.duration_seconds) > 1: - record.warnings.append( - f"XML duration {xml_duration:g}s differs from metadata duration " - f"{record.duration_seconds:g}s" - ) - - record.sequence = parse_sequence(first_tag(metadata, SEQUENCE_TAGS)) - record.camera_key = camera_key(record) - records.append(record) - - return records - - -def camera_key(record: MediaRecord) -> str: - metadata = record.metadata - make = first_tag(metadata, ("EXIF:Make", "QuickTime:Make", "MakerNotes:Make")) or "" - if not make: - make = first_tag(metadata, ("H264:Make",)) or "" - model = first_tag(metadata, ("EXIF:Model", "QuickTime:Model", "MakerNotes:Model")) or "" - if not model: - model = first_tag(metadata, ("H264:Model",)) or "" - serial = first_tag( - metadata, - ( - "EXIF:SerialNumber", - "MakerNotes:SerialNumber", - "MakerNotes:InternalSerialNumber", - "Composite:SerialNumber", - ), - ) - if make or model or serial: - return "|".join(str(part).strip() for part in (make, model, serial) if part) - return f"dir:{record.path.parent.resolve()}" - - -def parse_filename_timestamp(record: MediaRecord) -> datetime | None: - try: - return parse_datetime(record.path.stem) - except ValueError: - return None - - -def resolve_reference_time(records: list[MediaRecord], value: str) -> datetime: - text = value.strip() - lower = text.lower() - for record in records: - candidates = { - str(record.path).lower(), - record.path.name.lower(), - record.path.stem.lower(), - } - if lower in candidates: - if record.adjusted_time is None: - raise ValueError(f"{value} has no readable timestamp") - return naive_wall_time(record.adjusted_time) - return naive_wall_time(parse_datetime(text)) - - -def infer_reference_target(source_time: datetime, target_text: str) -> datetime: - try: - return naive_wall_time(parse_datetime(target_text)) - except ValueError: - pass - - base = naive_wall_time(source_time) - target_same_date = parse_datetime(target_text, default_date=base) - candidates = [ - target_same_date - timedelta(days=1), - target_same_date, - target_same_date + timedelta(days=1), - ] - - def score(candidate: datetime) -> tuple[int, float]: - delta_hours = abs((candidate - base).total_seconds()) / 3600 - same_date_penalty = 0 if candidate.date() == base.date() else 1 - return (0 if delta_hours <= 12 else 1, delta_hours + same_date_penalty) - - return min(candidates, key=score) - - -def prompt_time_offset(records: list[MediaRecord]) -> timedelta | None: - while True: - mode = prompt_input("Time correction: Enter=none, o=offset, r=reference photo/clock: ").strip().lower() - if mode == "": - return None - if mode in {"o", "offset"}: - while True: - text = prompt_input("Enter time shift (+1:30, -02:00:00, +1h 2m): ").strip() - try: - return parse_timeshift(text) - except ValueError as exc: - print(exc) - if mode in {"r", "ref", "reference"}: - while True: - source_text = prompt_input("Reference source timestamp, filename, or path: ").strip() - target_text = prompt_input("Correct reference-clock time: ").strip() - try: - source_time = resolve_reference_time(records, source_text) - target_time = infer_reference_target(source_time, target_text) - offset = target_time - source_time - print(f"Computed time shift: {format_timedelta(offset)}") - return offset - except ValueError as exc: - print(exc) - print("Please choose Enter, o, or r.") - - -def format_timedelta(value: timedelta) -> str: - total_seconds = int(value.total_seconds()) - sign = "+" if total_seconds >= 0 else "-" - total_seconds = abs(total_seconds) - hours = total_seconds // 3600 - minutes = (total_seconds % 3600) // 60 - seconds = total_seconds % 60 - return f"{sign}{hours:02d}:{minutes:02d}:{seconds:02d}" - - -def prompt_timezone(records: list[MediaRecord], rename_files: bool) -> tuple[timezone | None, timezone | None, bool, timezone | None]: - while True: - text = prompt_input("Enter timezone offset (+09:00), or blank to keep/infer: ").strip() - if not text: - explicit = None - break - try: - explicit = parse_timezone_offset(text) - break - except ValueError as exc: - print(exc) - - photo_zones = { - timezone_to_string(record.timezone_value): record.timezone_value - for record in records - if record.is_image and record.timezone_value is not None - } - missing_photo_zones = [ - record for record in records if record.is_image and record.timezone_value is None - ] - - authoritative = explicit - if authoritative is None and len(photo_zones) == 1: - authoritative = next(iter(photo_zones.values())) - elif authoritative is None and len(photo_zones) > 1: - print("Photo timezone offsets conflict; no authoritative timezone inferred.") - - write_missing = False - if authoritative is not None and explicit is not None: - write_missing = True - elif authoritative is not None and missing_photo_zones and photo_zones: - write_missing = ask_yes_no( - f"Write inferred timezone {timezone_to_string(authoritative)} to " - f"{len(missing_photo_zones)} photo(s) missing it?", - default=True, - ) - - video_tz = None - has_videos = any(record.is_video for record in records) - if rename_files and has_videos: - if authoritative is not None and explicit is None: - if ask_yes_no( - f"Use photo timezone {timezone_to_string(authoritative)} for video filenames?", - default=True, - ): - video_tz = authoritative - elif authoritative is not None: - video_tz = authoritative - - while video_tz is None: - text = prompt_input("Video renaming needs local timezone (+09:00): ").strip() - try: - video_tz = parse_timezone_offset(text) - except ValueError as exc: - print(exc) - - return explicit, authoritative, write_missing, video_tz - - -def prompt_artist_action() -> tuple[str, str | None]: - while True: - text = prompt_input("Artist/author: Enter=leave unchanged, s=set, c=clear: ").strip().lower() - if text == "": - return "leave", None - if text in {"c", "clear"}: - return "clear", None - if text in {"s", "set"}: - value = prompt_input("Artist/author value: ").strip() - if value: - return "set", value - print("Use clear if you want an empty artist/author.") - else: - print("Please choose Enter, s, or c.") - - -def prompt_choices(records: list[MediaRecord], working_dir: Path, has_pto_files: bool) -> UserChoices: - missing_time = [record for record in records if record.original_time is None] - apply_filename_timestamps = False - if missing_time: - apply_filename_timestamps = ask_yes_no( - f"{len(missing_time)} file(s) have no readable metadata timestamp. " - "Try timestamp from filename?", - default=True, - ) - if apply_filename_timestamps: - for record in missing_time: - parsed = parse_filename_timestamp(record) - if parsed is not None: - record.original_time = parsed - record.adjusted_time = parsed - record.parsed_from_filename = True - - time_offset = prompt_time_offset(records) - 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_min_size = 0 - if sum(1 for record in records if record.is_image and record.adjusted_time) >= 2: - group_min_size = ask_group_min_size() - group_photos = group_min_size >= 2 - - process_pto_files = False - if has_pto_files: - process_pto_files = ask_yes_no( - "Use panorama project files (.pto) for grouping and update their references?", - default=True, - ) - - move_to_working_dir = False - has_subdir_media = any(record.path.parent.resolve() != working_dir for record in records) - if has_subdir_media: - if group_photos or process_pto_files: - move_to_working_dir = True - else: - move_to_working_dir = ask_yes_no( - f"Move files from subdirectories into the working directory ({working_dir})?", - default=False, - ) - - return UserChoices( - working_dir=working_dir, - time_offset=time_offset, - explicit_timezone=explicit_tz, - authoritative_timezone=authoritative_tz, - write_missing_photo_timezones=write_missing_tz, - video_filename_timezone=video_tz, - artist_action=artist_action, - 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, - ) - - -def apply_time_offset(records: list[MediaRecord], offset: timedelta | None) -> None: - if offset is None: - return - for record in records: - if record.adjusted_time is not None: - record.adjusted_time = record.adjusted_time + offset - - -def photo_sort_key(record: MediaRecord) -> tuple: - assert record.adjusted_time is not None - return ( - naive_wall_time(record.adjusted_time), - record.subsec if record.subsec is not None else -1, - record.sequence if record.sequence is not None else -1, - record.path.name.lower(), - ) - - -def group_sort_key(record: MediaRecord) -> tuple: - timestamp = naive_wall_time(record.adjusted_time) if record.adjusted_time is not None else datetime.max - return ( - timestamp, - record.subsec if record.subsec is not None else -1, - record.sequence if record.sequence is not None else -1, - str(record.path).lower(), - ) - - -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: - if record.is_image and record.adjusted_time is not None: - 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), min_size)) - - return groups - - -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 - by_second[naive_wall_time(record.adjusted_time).replace(microsecond=0)].append(record) - - seconds = sorted(by_second) - spans: list[list[datetime]] = [] - current: list[datetime] = [] - for second in seconds: - if not current or second == current[-1] + timedelta(seconds=1): - current.append(second) - else: - spans.append(current) - current = [second] - if current: - spans.append(current) - - groups: list[list[MediaRecord]] = [] - for span in spans: - if max(len(by_second[second]) for second in span) < 2: - 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, min_size)) - else: - 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], min_size: int) -> list[list[MediaRecord]]: - ordered = sorted(records, key=photo_sort_key) - groups: list[list[MediaRecord]] = [] - current: list[MediaRecord] = [ordered[0]] - for record in ordered[1:]: - previous = current[-1] - if record.sequence is not None and previous.sequence is not None and record.sequence > previous.sequence: - current.append(record) - else: - if len(current) >= min_size: - groups.append(current) - current = [record] - if len(current) >= min_size: - groups.append(current) - return groups - - -def timestamp_groups( - 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: - return [] - - clusters: list[list[int]] = [[multi_indices[0]]] - for index in multi_indices[1:]: - if index - clusters[-1][-1] <= 2: - clusters[-1].append(index) - else: - clusters.append([index]) - - groups: list[list[MediaRecord]] = [] - used_seconds: set[datetime] = set() - for cluster in clusters: - start = cluster[0] - end = cluster[-1] - if start > 0 and len(by_second[span[start - 1]]) == 1: - start -= 1 - if end + 1 < len(span) and len(by_second[span[end + 1]]) == 1: - end += 1 - seconds = [second for second in span[start : end + 1] if second not in used_seconds] - 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) >= min_size: - groups.append(records) - return groups - - -def infer_group_subseconds(groups: list[list[MediaRecord]]) -> None: - for group in groups: - by_second: dict[datetime, list[MediaRecord]] = defaultdict(list) - for record in group: - if not record.is_image or record.adjusted_time is None: - continue - second = naive_wall_time(record.adjusted_time).replace(microsecond=0) - by_second[second].append(record) - - seconds = sorted(by_second) - counts = {second: len(records) for second, records in by_second.items()} - for index, second in enumerate(seconds): - records = sorted(by_second[second], key=photo_sort_key) - count = len(records) - if count <= 1: - continue - fps = count - if len(seconds) == 2: - fps = max(counts[seconds[0]], counts[seconds[1]]) - elif index == 0 and len(seconds) > 1: - fps = max(count, counts[seconds[index + 1]]) - elif index == len(seconds) - 1 and len(seconds) > 1: - fps = max(count, counts[seconds[index - 1]]) - - start_slot = 0 - if index == 0 and fps > count: - start_slot = fps - count - - for item_index, record in enumerate(records): - record.inferred_subsec = int((start_slot + item_index) * 1000 / fps) - - -def filename_range_stem(original_stem: str, base_dt: datetime) -> str | None: - match = TIMESTAMP_RANGE_STEM_RE.match(original_stem) - if not match: - return None - try: - old_start = parse_datetime(match.group("start")) - old_end = parse_datetime(match.group("end")) - except ValueError: - return None - interval = old_end - old_start - new_start = naive_wall_time(base_dt) - new_end = new_start + interval - suffix = match.group("suffix") or "" - return f"{format_filename_stem(new_start)}-{format_filename_stem(new_end)}{suffix}" - - -def unique_records_by_stem(records: list[MediaRecord]) -> dict[str, MediaRecord]: - buckets: dict[str, list[MediaRecord]] = defaultdict(list) - for record in records: - buckets[record.path.stem.lower()].append(record) - return {stem: bucket[0] for stem, bucket in buckets.items() if len(bucket) == 1} - - -def camera_range_stem(record: MediaRecord, records_by_stem: dict[str, MediaRecord]) -> str | None: - match = CAMERA_RANGE_STEM_RE.match(record.path.stem) - if not match: - return None - - first_number = int(match.group("first")) - last_number = int(match.group("last")) - if last_number <= first_number: - return None - - prefix = match.group("prefix") - first_record = records_by_stem.get(f"{prefix}{match.group('first')}".lower()) - last_record = records_by_stem.get(f"{prefix}{match.group('last')}".lower()) - first_time = first_record.adjusted_time if first_record else None - last_time = last_record.adjusted_time if last_record else None - own_time = record.adjusted_time - - if own_time is not None: - start_time = own_time - end_time = last_time or own_time - else: - start_time = first_time or last_time - end_time = last_time or first_time - - if start_time is None or end_time is None: - return None - - suffix = match.group("suffix") - return f"{format_filename_stem(start_time)}-{format_filename_stem(end_time)}{suffix}" - - -def filename_stem_for_dt(record: MediaRecord, dt: datetime) -> str: - range_stem = filename_range_stem(record.path.stem, dt) - if range_stem is not None: - return range_stem - return format_filename_stem(dt) - - -def plan_names(records: list[MediaRecord], choices: UserChoices) -> None: - if not choices.rename_files: - for record in records: - record.final_name = record.path.name - return - - records_by_stem = unique_records_by_stem(records) - for record in records: - if record.adjusted_time is None: - range_stem = camera_range_stem(record, records_by_stem) - if range_stem is not None: - record.new_stem = range_stem - else: - record.rename_skip_reason = "missing timestamp" - continue - if record.is_video: - if record.duration_seconds is None and not record.parsed_from_filename: - record.rename_skip_reason = "missing duration" - continue - if choices.video_filename_timezone is None: - record.rename_skip_reason = "missing video timezone" - continue - if record.parsed_from_filename and record.original_time is not None: - begin = record.adjusted_time - elif record.video_time_is_beginning: - begin = record.adjusted_time - else: - duration = round_half_up(record.duration_seconds or 0) - begin = record.adjusted_time - timedelta(seconds=duration) - if begin.tzinfo is None: - begin = begin.replace(tzinfo=timezone.utc) - local_begin = begin.astimezone(choices.video_filename_timezone) - record.new_stem = filename_stem_for_dt(record, local_begin) - else: - record.new_stem = camera_range_stem(record, records_by_stem) or filename_stem_for_dt( - record, record.adjusted_time - ) - - assign_unique_names(records) - - -def assign_unique_names(records: list[MediaRecord]) -> None: - buckets: dict[tuple[int | str, str], list[MediaRecord]] = defaultdict(list) - for record in records: - if record.new_stem is None: - record.final_name = record.path.name - continue - container: int | str = record.group_id if record.group_id is not None else "all" - buckets[(container, normalized_extension(record.path))].append(record) - - for (_container, ext), bucket in buckets.items(): - by_stem: dict[str, list[MediaRecord]] = defaultdict(list) - for record in sorted(bucket, key=lambda item: (item.new_stem or "", item.path.name.lower())): - assert record.new_stem is not None - by_stem[record.new_stem].append(record) - - for stem, records_with_stem in by_stem.items(): - if len(records_with_stem) == 1: - records_with_stem[0].final_name = f"{stem}{ext}" - continue - for index, record in enumerate(records_with_stem, start=1): - record.final_name = f"{stem}-{index}{ext}" - - -def merge_record_groups( - records: list[MediaRecord], group_candidates: list[list[MediaRecord]] -) -> list[list[MediaRecord]]: - record_indexes = {record.path.resolve(): index for index, record in enumerate(records)} - parents = list(range(len(records))) - grouped_indexes: set[int] = set() - - def find(index: int) -> int: - while parents[index] != index: - parents[index] = parents[parents[index]] - index = parents[index] - return index - - def union(left: int, right: int) -> None: - left_root = find(left) - right_root = find(right) - if left_root != right_root: - parents[right_root] = left_root - - for candidate in group_candidates: - indexes = [ - record_indexes[record.path.resolve()] - for record in candidate - if record.path.resolve() in record_indexes - ] - if not indexes: - continue - grouped_indexes.update(indexes) - first = indexes[0] - for index in indexes[1:]: - union(first, index) - - merged: dict[int, list[MediaRecord]] = defaultdict(list) - for index in grouped_indexes: - merged[find(index)].append(records[index]) - - return sorted( - [sorted(group, key=group_sort_key) for group in merged.values()], - key=lambda group: group_sort_key(group[0]), - ) - - -def assign_groups(records: list[MediaRecord], groups: list[list[MediaRecord]]) -> None: - for record in records: - record.group_id = None - record.group_name = None - for index, group in enumerate(groups, start=1): - ordered = sorted(group, key=group_sort_key) - for record in ordered: - record.group_id = index - - -def unique_group_name(base_name: str, working_dir: Path, used_names: set[str]) -> str: - if (working_dir / base_name).exists(): - used_names.add(base_name) - return base_name - - if base_name not in used_names: - used_names.add(base_name) - return base_name - - for index in range(1, 10000): - candidate = f"{base_name}-{index}" - if candidate not in used_names and not (working_dir / candidate).exists(): - used_names.add(candidate) - return candidate - - raise RuntimeError(f"Could not find an available group directory name for {base_name}") - - -def assign_group_names(groups: list[list[MediaRecord]], working_dir: Path) -> None: - used_names: set[str] = set() - for group in groups: - ordered = sorted(group, key=group_sort_key) - first = group_stem_from_name(ordered[0].final_name or ordered[0].path.name) - last = group_stem_from_name(ordered[-1].final_name or ordered[-1].path.name) - group_name = unique_group_name(f"{first}-{last}", working_dir, used_names) - for record in ordered: - record.group_name = group_name - - -def plan_targets(records: list[MediaRecord], choices: UserChoices) -> None: - for record in records: - if record.final_name is None: - record.final_name = record.path.name - out_dir = record.path.parent - if record.group_name: - out_dir = choices.working_dir / record.group_name - elif choices.move_to_working_dir: - out_dir = choices.working_dir - record.target_path = out_dir / record.final_name - if record.is_video and record.sidecar is not None: - target_sidecar = out_dir / f"{Path(record.final_name).stem}M01.XML" - if target_sidecar.resolve() != record.sidecar.path.resolve(): - record.target_sidecar = target_sidecar - - -def normalize_reference(value: str) -> str: - return value.replace("\\", "/") - - -def pto_target_for_base( - path: Path, target_dir: Path, base_name: str, used_targets: set[Path] -) -> Path: - desired = target_dir / f"{base_name}.pto" - if desired.resolve() == path.resolve() or (not desired.exists() and desired not in used_targets): - used_targets.add(desired) - return desired - - for index in range(1, 10000): - candidate = target_dir / f"{base_name}-{index}.pto" - if candidate.resolve() == path.resolve() or ( - not candidate.exists() and candidate not in used_targets - ): - used_targets.add(candidate) - return candidate - - raise RuntimeError(f"Could not find an available .pto name for {base_name}") - - -def pto_reference_map(pto_path: Path, records: list[MediaRecord]) -> dict[str, MediaRecord]: - pto_dir = pto_path.parent - basename_counts: dict[str, int] = defaultdict(int) - for record in records: - basename_counts[record.path.name] += 1 - - mapping: dict[str, MediaRecord] = {} - for record in records: - old_abs = record.path.resolve() - candidates = { - normalize_reference(str(old_abs)), - normalize_reference(os.path.relpath(old_abs, pto_dir)), - } - if basename_counts[record.path.name] == 1: - candidates.add(record.path.name) - for candidate in candidates: - mapping[candidate] = record - return mapping - - -PTO_QUOTED_VALUE_RE = re.compile(r"([\"'])(.*?)(\1)") - - -def read_pto_text(path: Path) -> str: - try: - return path.read_text(encoding="utf-8") - except UnicodeDecodeError: - return path.read_text(encoding="utf-8", errors="replace") - - -def read_pto_references(pto_files: list[Path], records: list[MediaRecord]) -> list[PtoReference]: - references: list[PtoReference] = [] - for pto_path in pto_files: - original_text = read_pto_text(pto_path) - mapping = pto_reference_map(pto_path, records) - if not mapping: - continue - - referenced: list[MediaRecord] = [] - for match in PTO_QUOTED_VALUE_RE.finditer(original_text): - record = mapping.get(normalize_reference(match.group(2))) - if record is not None and record not in referenced: - referenced.append(record) - - if referenced: - references.append(PtoReference(path=pto_path, original_text=original_text, records=referenced)) - return references - - -def replacement_for_pto_value( - value: str, mapping: dict[str, MediaRecord], pto_target_dir: Path -) -> tuple[str, MediaRecord] | None: - normalized = normalize_reference(value) - record = mapping.get(normalized) - if record is None: - return None - assert record.target_path is not None - replacement = normalize_reference(os.path.relpath(record.target_path.resolve(), pto_target_dir)) - return replacement, record - - -def plan_pto_updates(pto_references: list[PtoReference], records: list[MediaRecord]) -> list[PtoPlan]: - plans: list[PtoPlan] = [] - used_targets: set[Path] = set() - for pto_reference in pto_references: - pto_path = pto_reference.path - original_text = pto_reference.original_text - referenced = sorted(pto_reference.records, key=group_sort_key) - if not referenced: - continue - - target_dir = pto_path.parent - if referenced[0].group_name: - target_dir = referenced[0].target_path.parent if referenced[0].target_path else target_dir - - mapping = pto_reference_map(pto_path, records) - if not mapping: - continue - - replacements: list[tuple[str, str]] = [] - - def replace_match(match: re.Match) -> str: - quote = match.group(1) - value = match.group(2) - replacement = replacement_for_pto_value(value, mapping, target_dir) - if replacement is None: - return match.group(0) - new_value, record = replacement - if value != new_value: - replacements.append((value, new_value)) - return f"{quote}{new_value}{quote}" - - updated_text = PTO_QUOTED_VALUE_RE.sub(replace_match, original_text) - first = group_stem_from_name(referenced[0].target_path.name) - last = group_stem_from_name(referenced[-1].target_path.name) - target_path = pto_target_for_base(pto_path, target_dir, f"{first}-{last}", used_targets) - if updated_text == original_text and target_path.resolve() == pto_path.resolve(): - continue - plans.append( - PtoPlan( - path=pto_path, - target_path=target_path, - referenced_records=referenced, - updated_text=updated_text, - replacements=replacements, - ) - ) - - return plans - - -def build_write_plan(record: MediaRecord, choices: UserChoices) -> list[TagWrite]: - writes: list[TagWrite] = [] - if choices.artist_action == "set": - assert choices.artist_value is not None - add_tag_write( - writes, - "Artist", - "Artist", - tag_value(record.metadata, "EXIF:Artist") or tag_value(record.metadata, "XMP:Artist"), - choices.artist_value, - ) - add_tag_write( - writes, - "Author", - "Author", - tag_value(record.metadata, "EXIF:Author") or tag_value(record.metadata, "XMP:Author"), - choices.artist_value, - ) - elif choices.artist_action == "clear": - add_tag_write( - writes, - "Artist", - "Artist", - tag_value(record.metadata, "EXIF:Artist") or tag_value(record.metadata, "XMP:Artist"), - "", - ) - add_tag_write( - writes, - "Author", - "Author", - tag_value(record.metadata, "EXIF:Author") or tag_value(record.metadata, "XMP:Author"), - "", - ) - - if record.is_image: - if choices.time_offset is not None or record.parsed_from_filename: - if record.adjusted_time is not None: - dt_text = record.adjusted_time.strftime("%Y:%m:%d %H:%M:%S") - for label, arg_name, current_tag in ( - ("DateTimeOriginal", "EXIF:DateTimeOriginal", "EXIF:DateTimeOriginal"), - ("CreateDate", "EXIF:CreateDate", "EXIF:CreateDate"), - ("ModifyDate", "EXIF:ModifyDate", "EXIF:ModifyDate"), - ): - add_tag_write( - writes, - label, - arg_name, - tag_value(record.metadata, current_tag), - dt_text, - same_datetime_value, - ) - - should_write_tz = False - tz_to_write = choices.explicit_timezone - if tz_to_write is not None: - should_write_tz = True - elif ( - choices.write_missing_photo_timezones - and record.timezone_value is None - and choices.authoritative_timezone is not None - ): - tz_to_write = choices.authoritative_timezone - should_write_tz = True - if should_write_tz and tz_to_write is not None: - tz_text = timezone_to_string(tz_to_write) - for label, arg_name, current_tag in ( - ("OffsetTimeOriginal", "EXIF:OffsetTimeOriginal", "EXIF:OffsetTimeOriginal"), - ("OffsetTimeDigitized", "EXIF:OffsetTimeDigitized", "EXIF:OffsetTimeDigitized"), - ("OffsetTime", "EXIF:OffsetTime", "EXIF:OffsetTime"), - ): - add_tag_write( - writes, - label, - arg_name, - tag_value(record.metadata, current_tag), - tz_text, - ) - - if record.inferred_subsec is not None: - subsec_text = f"{record.inferred_subsec:03d}" - for label, arg_name, current_tag in ( - ("SubSecTimeOriginal", "EXIF:SubSecTimeOriginal", "EXIF:SubSecTimeOriginal"), - ("SubSecTimeDigitized", "EXIF:SubSecTimeDigitized", "EXIF:SubSecTimeDigitized"), - ("SubSecTime", "EXIF:SubSecTime", "EXIF:SubSecTime"), - ): - add_tag_write( - writes, - label, - arg_name, - tag_value(record.metadata, current_tag), - subsec_text, - same_subsec_value, - ) - elif record.is_video and choices.time_offset is not None and record.adjusted_time is not None: - utc_time = record.adjusted_time - if utc_time.tzinfo is None: - utc_time = utc_time.replace(tzinfo=timezone.utc) - utc_text = utc_time.astimezone(timezone.utc).strftime("%Y:%m:%d %H:%M:%S") - for label, arg_name, current_tag in ( - ("QuickTime CreateDate", "QuickTime:CreateDate", "QuickTime:CreateDate"), - ("QuickTime ModifyDate", "QuickTime:ModifyDate", "QuickTime:ModifyDate"), - ("QuickTime TrackCreateDate", "QuickTime:TrackCreateDate", "QuickTime:TrackCreateDate"), - ("QuickTime TrackModifyDate", "QuickTime:TrackModifyDate", "QuickTime:TrackModifyDate"), - ("QuickTime MediaCreateDate", "QuickTime:MediaCreateDate", "QuickTime:MediaCreateDate"), - ("QuickTime MediaModifyDate", "QuickTime:MediaModifyDate", "QuickTime:MediaModifyDate"), - ): - add_tag_write( - writes, - label, - arg_name, - tag_value(record.metadata, current_tag), - utc_text, - same_datetime_value, - ) - - return writes - - -def build_write_args(record: MediaRecord, choices: UserChoices) -> list[str]: - if not record.write_plan: - record.write_plan = build_write_plan(record, choices) - return [write.arg for write in record.write_plan if write.will_write] - - -def path_display(path: Path, choices: UserChoices, use_relative: bool) -> str: - if use_relative: - return display_path_from_working_dir(path, choices.working_dir) - return str(path) - - -def directory_display(path: Path, choices: UserChoices) -> str: - text = path_display(path, choices, use_relative=True) - if text != "." and not text.endswith(os.sep): - return f"{text}{os.sep}" - return text - - -def move_rename_preview( - source: Path, - target: Path, - choices: UserChoices, - source_label: str | None = None, -) -> str | None: - is_moving = source.parent.resolve() != target.parent.resolve() - is_renaming = source.name != target.name - source_text = source_label or path_display(source, choices, use_relative=is_moving) - - if is_moving and is_renaming: - target_text = path_display(target, choices, use_relative=True) - return f"move+rename {source_text} -> {target_text}" - if is_moving: - target_text = directory_display(target.parent, choices) - return f"move {source_text} -> {target_text}" - if is_renaming: - return f"rename {source_label or source.name} -> {target.name}" - return None - - -def print_preview(records: list[MediaRecord], choices: UserChoices, pto_plans: list[PtoPlan]) -> None: - 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( - "Photo timezone write: " - + ( - timezone_to_string(choices.explicit_timezone) - if choices.explicit_timezone - else ( - f"missing only -> {timezone_to_string(choices.authoritative_timezone)}" - if choices.write_missing_photo_timezones and choices.authoritative_timezone - else "unchanged" - ) - ) - ) - print( - "Video filename timezone: " - + (timezone_to_string(choices.video_filename_timezone) if choices.video_filename_timezone else "n/a") - ) - if choices.artist_action == "set": - print(f"Artist/author: set to {choices.artist_value}") - elif choices.artist_action == "clear": - print("Artist/author: clear") - else: - print("Artist/author: unchanged") - - for record in sorted(records, key=lambda item: str(item.path).lower()): - target = record.target_path or record.path - operation = [] - operation_text = move_rename_preview(record.path, target, choices) - if operation_text: - operation.append(operation_text) - if record.rename_skip_reason: - operation.append(f"rename skipped: {record.rename_skip_reason}") - for warning in record.warnings: - operation.append(f"warning: {warning}") - if record.target_sidecar and record.sidecar: - sidecar_source = path_display(record.sidecar.path, choices, use_relative=True) - sidecar_target = path_display(record.target_sidecar, choices, use_relative=True) - operation.append(f"sidecar move {sidecar_source} -> {sidecar_target}") - if operation: - print("; ".join(operation)) - - if pto_plans: - print("\nPTO updates:") - for plan in pto_plans: - source = path_display(plan.path, choices, use_relative=True) - operation = move_rename_preview(plan.path, plan.target_path, choices, source_label=source) - print(operation or source) - for old_ref, new_ref in plan.replacements: - print(f" {old_ref} -> {new_ref}") - - -def has_planned_media_change(record: MediaRecord, write_args: list[str]) -> bool: - if write_args: - return True - if record.target_path and record.target_path.resolve() != record.path.resolve(): - return True - if record.sidecar and record.target_sidecar and record.sidecar.path.exists(): - return True - return False - - -def apply_changes( - executable: str, records: list[MediaRecord], choices: UserChoices -) -> list[str]: - failures: list[str] = [] - planned_sources = {record.path.resolve() for record in records} - - for index, record in enumerate(records, start=1): - changed = False - try: - write_args = build_write_args(record, choices) - changed = has_planned_media_change(record, write_args) - run_exiftool_write(executable, record.path, write_args) - - 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: - target_sidecar = unique_existing_target(record.target_sidecar, planned_sources) - target_sidecar.parent.mkdir(parents=True, exist_ok=True) - if record.sidecar.path.exists(): - shutil.move(str(record.sidecar.path), str(target_sidecar)) - except (OSError, subprocess.CalledProcessError, RuntimeError) as exc: - failures.append(f"{record.path}: {exc}") - action = "Processed" if changed else "Skipped" - print(f"{action} {index} of {len(records)}") - - return failures - - -def apply_pto_changes(pto_plans: list[PtoPlan]) -> list[str]: - failures: list[str] = [] - for plan in pto_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(removed_dirs: list[Path]) -> None: - if not removed_dirs: - return - print("\nRemoved empty director" + ("y:" if len(removed_dirs) == 1 else "ies:")) - for directory in removed_dirs: - print(f" - {directory}") - - -def prepare_plan( - records: list[MediaRecord], choices: UserChoices, pto_references: list[PtoReference] -) -> list[list[MediaRecord]]: - apply_time_offset(records, choices.time_offset) - - group_candidates: list[list[MediaRecord]] = [] - if choices.group_photos: - 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) - - groups = merge_record_groups(records, group_candidates) - if groups: - assign_groups(records, groups) - infer_group_subseconds(groups) - - plan_names(records, choices) - if groups: - assign_group_names(groups, choices.working_dir) - plan_targets(records, choices) - for record in records: - record.write_plan = build_write_plan(record, choices) - return groups def run(argv: list[str]) -> int: @@ -1590,8 +50,8 @@ def run(argv: list[str]) -> int: return 1 cleanup_roots = cleanup_roots_from_args(argv[1:]) - files = collect_media_files(argv[1:]) - pto_files = collect_pto_files(argv[1:]) + files = collect_files(argv[1:], allowed_exts=SUPPORTED_EXTS, print_missing=True) + pto_files = collect_files(argv[1:], allowed_exts={".pto"}) if not files: print("No supported media files found.") return 1 @@ -1602,34 +62,53 @@ def run(argv: list[str]) -> int: print(exc) return 1 - print(f"Found {len(files)} supported media file(s). Reading metadata...") + print(f"Found {len(files)} supported media file(s).") + progress = ProgressView(len(files), "Reading metadata") + read_count = 0 + + def update_read_progress() -> None: + nonlocal read_count + read_count += 1 + progress.update(read_count) + try: - metadata = run_exiftool_json(exiftool, files) + metadata = run_exiftool_json(exiftool, files, progress=update_read_progress) except (subprocess.CalledProcessError, json.JSONDecodeError) as exc: + progress.finish() print(f"Failed to read metadata with exiftool: {exc}") return 1 + progress.finish(keep=True) - records = build_records(files, metadata) - try: - choices = prompt_choices(records, working_dir, has_pto_files=bool(pto_files)) - except RuntimeError as exc: - print(exc) - return 1 + settings_state = None + while True: + records = build_records(files, metadata) + try: + choices, settings_state = prompt_choices( + exiftool, + records, + working_dir, + has_pto_files=bool(pto_files), + initial_state=settings_state, + ) + except RuntimeError as exc: + print(exc) + return 1 + if choices is None: + print("Nothing changed.") + return 0 - pto_references = read_pto_references(pto_files, records) if choices.process_pto_files else [] - prepare_plan(records, choices, pto_references) - pto_plans = plan_pto_updates(pto_references, records) if choices.process_pto_files else [] - print_preview(records, choices, pto_plans) + pto_references = read_pto_references(pto_files, records) if choices.process_pto_files else [] + prepare_plan(records, choices, pto_references) + pto_plans = plan_pto_updates(pto_references, records) if choices.process_pto_files else [] + clear_screen(scrollback=True) + print_preview(records, choices, pto_plans) - try: - proceed = ask_yes_no("Proceed with these changes?", default=True) - except RuntimeError as exc: - print(exc) - return 1 - - if not proceed: - print("Nothing changed.") - return 0 + try: + if ask_yes_no("Proceed with these changes?", default=True): + break + except RuntimeError as exc: + print(exc) + return 1 failures = apply_changes(exiftool, records, choices) failures.extend(apply_pto_changes(pto_plans)) diff --git a/tools/console.py b/tools/console.py index d4b82e5..6580868 100644 --- a/tools/console.py +++ b/tools/console.py @@ -1,14 +1,19 @@ from __future__ import annotations import os +import select +import shutil import sys +import textwrap import time from collections import deque +from contextlib import contextmanager from math import floor, log10 from threading import Lock _WINDOWS_ANSI_ENABLED: bool | None = None +_WINDOWS_CONSOLE_OUTPUT: bool | None = None def prompt_input(prompt: str) -> str: @@ -44,7 +49,7 @@ def supports_color() -> bool: def supports_ansi() -> bool: - if not sys.stdout.isatty(): + if not stream_is_interactive(sys.stdout): return False if os.name != "nt": return os.environ.get("TERM", "dumb") != "dumb" @@ -56,6 +61,10 @@ def supports_ansi() -> bool: ) +def stream_is_interactive(stream) -> bool: + return stream.isatty() or (os.name == "nt" and stream is sys.stdout and _windows_console_output()) + + def color(text: object, code: str) -> str: value = str(text) if not supports_color(): @@ -79,12 +88,74 @@ def red_strikethrough(text: object) -> str: return color(text, "91;9") -def clear_screen() -> None: - if not sys.stdout.isatty(): +def clear_screen(*, scrollback: bool = False) -> None: + if not stream_is_interactive(sys.stdout): + return + if supports_ansi(): + sys.stdout.write("\x1b[3J" if scrollback else "") + sys.stdout.write("\x1b[H\x1b[2J") + sys.stdout.flush() return os.system("cls" if os.name == "nt" else "clear") +def require_terminal_ui() -> None: + if not (stream_is_interactive(sys.stdin) and stream_is_interactive(sys.stdout) and supports_ansi()): + raise RuntimeError("This settings dialog requires an ANSI-capable interactive terminal.") + + +@contextmanager +def raw_key_input(): + require_terminal_ui() + if os.name == "nt": + yield read_key + return + + import termios + import tty + + fd = sys.stdin.fileno() + settings = termios.tcgetattr(fd) + try: + tty.setraw(fd) + yield read_key + finally: + termios.tcsetattr(fd, termios.TCSADRAIN, settings) + + +def read_key() -> str: + if os.name == "nt": + import msvcrt + + key = msvcrt.getwch() + if key in {"\x00", "\xe0"}: + return {"H": "up", "P": "down", "K": "left", "M": "right"}.get(msvcrt.getwch(), "") + else: + key = sys.stdin.read(1) + if key == "\x1b" and select.select([sys.stdin], [], [], 0.03)[0]: + key += sys.stdin.read(2) + return {"\x1b[A": "up", "\x1b[B": "down", "\x1b[D": "left", "\x1b[C": "right"}.get(key, "esc") + return {"\r": "enter", "\n": "enter", "\x1b": "esc", "\x08": "backspace", "\x7f": "backspace", " ": "space"}.get(key, key) + + +def draw_screen(lines: list[str]) -> None: + require_terminal_ui() + sys.stdout.write("\x1b[H\x1b[2J" + "\n".join(lines)) + sys.stdout.flush() + + +def inverse(text: str) -> str: + return f"\x1b[30;47m{text}\x1b[0m" + + +def dim(text: str) -> str: + return f"\x1b[90m{text}\x1b[0m" + + +def dark_field(text: str) -> str: + return f"\x1b[30;100m{text}\x1b[0m" + + def _enable_windows_ansi() -> bool: global _WINDOWS_ANSI_ENABLED if _WINDOWS_ANSI_ENABLED is not None: @@ -105,6 +176,23 @@ def _enable_windows_ansi() -> bool: return False +def _windows_console_output() -> bool: + global _WINDOWS_CONSOLE_OUTPUT + if _WINDOWS_CONSOLE_OUTPUT is not None: + return _WINDOWS_CONSOLE_OUTPUT + try: + import ctypes + + mode = ctypes.c_uint32() + handle = ctypes.windll.kernel32.GetStdHandle(-11) + _WINDOWS_CONSOLE_OUTPUT = bool( + ctypes.windll.kernel32.GetConsoleMode(handle, ctypes.byref(mode)) + ) + except Exception: + _WINDOWS_CONSOLE_OUTPUT = False + return _WINDOWS_CONSOLE_OUTPUT + + def format_duration(seconds: float) -> str: seconds = max(0, int(round(seconds))) minutes, second = divmod(seconds, 60) @@ -120,30 +208,42 @@ class ProgressView: total: int, label: str, *, + total_work: float | None = None, stream=None, embedded_percent: bool = False, show_rate: bool = False, + show_elapsed: bool = True, + bar_width: int = 24, ) -> None: self.total = max(0, total) + self.total_work = total if total_work is None else max(0.0, total_work) self.label = label self.stream = stream or sys.stdout self.started_at = time.monotonic() - self.samples: deque[tuple[int, float]] = deque() + self.samples: deque[tuple[float, float]] = deque() self.last_width = 0 + self.last_line_count = 0 self.embedded_percent = embedded_percent self.show_rate = show_rate + self.show_elapsed = show_elapsed + self.bar_width = bar_width - def update(self, processed: int, detail: str = "") -> None: - self._write(self.update_line(processed, detail)) + def update( + self, processed: int, detail: str = "", *, completed_work: float | None = None + ) -> None: + self._write(self.update_line(processed, detail, completed_work=completed_work)) - def update_line(self, processed: int, detail: str = "") -> str: + def update_line( + self, processed: int, detail: str = "", *, completed_work: float | None = None + ) -> str: processed = max(0, min(processed, self.total)) + work_done = min(self.total_work, completed_work if completed_work is not None else processed) now = time.monotonic() - if not self.samples or self.samples[-1][0] != processed: - self.samples.append((processed, now)) + if not self.samples or self.samples[-1][0] != work_done: + self.samples.append((work_done, now)) while len(self.samples) > max(2, processed // 2 + 2): self.samples.popleft() - return self._line(processed, detail, now) + return self._line(processed, detail, now, completed_work=work_done) def update_external( self, @@ -167,25 +267,60 @@ class ProgressView: return self._line(processed, detail, None, elapsed=elapsed, eta=eta) def finish(self, *, keep: bool = False) -> None: - if self.stream.isatty(): + if stream_is_interactive(self.stream): if keep: self.stream.write("\n") else: - self.stream.write("\r" + " " * self.last_width + "\r") + self._clear_live_lines() self.stream.flush() elif self.last_width and not keep: self.stream.write("\n") self.last_width = 0 + self.last_line_count = 0 - def _write(self, line: str) -> None: - if self.stream.isatty(): + def write_lines(self, lines: list[str]) -> None: + self._write(lines) + + def _write(self, content: str | list[str]) -> None: + lines = [content] if isinstance(content, str) else content + if stream_is_interactive(self.stream) and supports_ansi(): + lines = [wrapped for line in lines for wrapped in _wrap_to_terminal(line, self.stream)] + if self.last_line_count > 1: + self.stream.write(f"\x1b[{self.last_line_count - 1}F") + rows = max(self.last_line_count, len(lines)) + for index in range(rows): + self.stream.write("\r\x1b[2K") + if index < len(lines): + self.stream.write(lines[index]) + if index < rows - 1: + self.stream.write("\n") + if rows > len(lines): + self.stream.write(f"\x1b[{rows - len(lines)}F") + self.stream.flush() + self.last_width = max(map(len, lines), default=0) + self.last_line_count = len(lines) + elif stream_is_interactive(self.stream): + line = " ".join(lines) padding = max(0, self.last_width - len(line)) self.stream.write("\r" + line + (" " * padding)) self.stream.flush() self.last_width = len(line) else: - self.stream.write(line + "\n") - self.last_width = len(line) + self.stream.write("\n".join(lines) + "\n") + self.last_width = max(map(len, lines), default=0) + + def _clear_live_lines(self) -> None: + if not self.last_line_count or not supports_ansi(): + self.stream.write("\r" + " " * self.last_width + "\r") + return + if self.last_line_count > 1: + self.stream.write(f"\x1b[{self.last_line_count - 1}F") + for index in range(self.last_line_count): + self.stream.write("\r\x1b[2K") + if index < self.last_line_count - 1: + self.stream.write("\n") + if self.last_line_count > 1: + self.stream.write(f"\x1b[{self.last_line_count - 1}F") def _line( self, @@ -195,13 +330,15 @@ class ProgressView: *, elapsed: float | None = None, eta: float | None = None, + completed_work: float | None = None, ) -> str: - prefix = self.label if not detail else f"{self.label}: {detail}" + prefix = self.label if self.total <= 2: - return f"{prefix} {processed}/{self.total}" + return " ".join(part for part in (prefix, f"{processed}/{self.total}", detail) if part) - percent = 0.0 if self.total == 0 else processed / self.total - bar = _progress_bar(percent, embedded_percent=self.embedded_percent) + work_done = processed if completed_work is None else completed_work + percent = 0.0 if self.total_work == 0 else work_done / self.total_work + bar = _progress_bar(percent, embedded_percent=self.embedded_percent, width=self.bar_width) elapsed_value = elapsed if elapsed is not None else (now or time.monotonic()) - self.started_at parts = [ prefix, @@ -212,22 +349,25 @@ class ProgressView: parts.append(f"{percent * 100:5.1f}%") if self.show_rate and processed > 0 and elapsed_value > 0: parts.append(f"{_format_significant(processed / elapsed_value, 2)} fps") - parts.append(f"elapsed {format_duration(elapsed_value)}") - eta_value = eta if eta is not None else self._eta(processed) + if self.show_elapsed: + parts.append(f"elapsed {format_duration(elapsed_value)}") + eta_value = eta if eta is not None else self._eta(processed, work_done) if processed >= 2 and processed < self.total and eta_value is not None: parts.append(f"ETA {format_duration(eta_value)}") + if detail: + parts.append(detail) return " ".join(parts) - def _eta(self, processed: int) -> float | None: + def _eta(self, processed: int, work_done: float) -> float | None: if processed < 2 or self.total <= processed or len(self.samples) < 2: return None - oldest_processed, oldest_time = self.samples[0] - newest_processed, newest_time = self.samples[-1] - delta_items = newest_processed - oldest_processed + oldest_work, oldest_time = self.samples[0] + newest_work, newest_time = self.samples[-1] + delta_work = newest_work - oldest_work delta_time = newest_time - oldest_time - if delta_items <= 0 or delta_time <= 0: + if delta_work <= 0 or delta_time <= 0: return None - return (self.total - processed) * (delta_time / delta_items) + return (self.total_work - work_done) * (delta_time / delta_work) class PipelineProgressView: @@ -311,8 +451,7 @@ class PipelineProgressView: self._shown = True -def _progress_bar(percent: float, *, embedded_percent: bool) -> str: - width = 24 +def _progress_bar(percent: float, *, embedded_percent: bool, width: int = 24) -> str: done = round(max(0.0, min(percent, 1.0)) * width) if not embedded_percent: return "[" + ("#" * done).ljust(width, "-") + "]" @@ -335,6 +474,16 @@ def _center_progress_text(text: str, width: int, fill: str) -> str: return content.center(width, fill) +def _wrap_to_terminal(line: str, stream) -> list[str]: + try: + width = shutil.get_terminal_size().columns + except OSError: + return [line] + if width <= 0 or len(line) <= width: + return [line] + return textwrap.wrap(line, width=width, break_long_words=False, break_on_hyphens=False) or [line] + + def _format_significant(value: float, digits: int) -> str: if value == 0: return "0" diff --git a/tools/exiftool.py b/tools/exiftool.py index 410c99f..8d7c469 100644 --- a/tools/exiftool.py +++ b/tools/exiftool.py @@ -2,8 +2,11 @@ from __future__ import annotations import json import os +import re import subprocess import tempfile +from collections.abc import Callable, Iterator +from contextlib import contextmanager from pathlib import Path from typing import Any @@ -18,15 +21,12 @@ def _use_windows_argument_file() -> bool: return os.name == "nt" -def run_exiftool_command( - executable: str, - args: list[str], - **run_kwargs: Any, -) -> subprocess.CompletedProcess: - """Run ExifTool with Unicode-safe filename arguments on Windows.""" +@contextmanager +def _prepared_command(executable: str, args: list[str]) -> Iterator[tuple[list[str], str | None]]: command = [executable, "-charset", "filename=UTF8", *args] if not _use_windows_argument_file(): - return subprocess.run(command, **run_kwargs) + yield command, None + return # ExifTool can't reliably receive arbitrary Unicode paths on the Windows # command line. The outer path is ASCII and the UTF-8 argument file holds @@ -34,11 +34,20 @@ def run_exiftool_command( with tempfile.TemporaryDirectory(prefix="mbt-exiftool-") as temp_dir: argument_file = Path(temp_dir) / "arguments.txt" argument_file.write_text("\n".join(args) + "\n", encoding="utf-8") - return subprocess.run( - [executable, "-charset", "filename=UTF8", "-@", argument_file.name], - cwd=temp_dir, - **run_kwargs, - ) + yield [executable, "-charset", "filename=UTF8", "-@", argument_file.name], temp_dir + + +def run_exiftool_command( + executable: str, + args: list[str], + **run_kwargs: Any, +) -> subprocess.CompletedProcess: + """Run ExifTool with Unicode-safe filename arguments on Windows.""" + with _prepared_command(executable, args) as (command, cwd): + return subprocess.run(command, cwd=cwd, **run_kwargs) + + +_PROGRESS_RE = re.compile(r"^======== .* \[\d+/(\d+)\]$") def run_exiftool_json( @@ -46,6 +55,7 @@ def run_exiftool_json( paths: list[Path], *, quicktime_utc: bool = True, + progress: Callable[[], None] | None = None, ) -> list[dict]: results: list[dict] = [] for start in range(0, len(paths), 80): @@ -57,14 +67,45 @@ def run_exiftool_json( if quicktime_utc: args.extend(["-api", "QuickTimeUTC=1"]) args.extend(str(path) for path in chunk) - completed = run_exiftool_command( - executable, - args, - check=True, - capture_output=True, - text=True, - ) - results.extend(json.loads(completed.stdout or "[]")) + if progress is None: + completed = run_exiftool_command( + executable, + args, + check=True, + capture_output=True, + text=True, + ) + results.extend(json.loads(completed.stdout or "[]")) + continue + + with _prepared_command(executable, ["-progress", *args]) as (command, cwd): + with tempfile.TemporaryFile(mode="w+t", encoding="utf-8") as output: + process = subprocess.Popen( + command, + cwd=cwd, + stdout=output, + stderr=subprocess.PIPE, + text=True, + ) + assert process.stderr is not None + errors = [] + started_file = False + for line in process.stderr: + if _PROGRESS_RE.match(line.rstrip("\r\n")): + # ExifTool emits this marker before reading the named + # file, so it confirms that the previous one finished. + if started_file: + progress() + started_file = True + else: + errors.append(line) + return_code = process.wait() + if return_code: + raise subprocess.CalledProcessError(return_code, command, stderr="".join(errors)) + if started_file: + progress() + output.seek(0) + results.extend(json.load(output)) return results @@ -76,4 +117,13 @@ def run_exiftool_write(executable: str, path: Path, args: list[str]) -> None: *args, str(path), ] - run_exiftool_command(executable, command, check=True) + completed = run_exiftool_command( + executable, + command, + check=False, + capture_output=True, + text=True, + ) + if completed.returncode: + detail = (completed.stderr or completed.stdout or "").strip() + raise RuntimeError(f"ExifTool failed: {detail or f'exit code {completed.returncode}'}") diff --git a/tools/media_metadata.py b/tools/media_metadata.py new file mode 100644 index 0000000..d2103c2 --- /dev/null +++ b/tools/media_metadata.py @@ -0,0 +1,63 @@ +"""Small, shared helpers for values returned by ExifTool.""" + +from __future__ import annotations + +import math +import re +from collections.abc import Iterable, Mapping +from datetime import datetime, timedelta, timezone + + +DATETIME_RE = re.compile( + r"(?P\d{4})[:\-]?(?P\d{2})[:\-]?(?P\d{2})" + r"(?:[ T_])?" + r"(?P\d{2}):?(?P\d{2}):?(?P\d{2})" + r"(?:[.,](?P\d+))?" + r"(?:\s*(?PZ|[+-]\d{2}:?\d{2}))?" +) + + +def first_tag(metadata: Mapping[str, object], tags: Iterable[str]) -> object | None: + for tag in tags: + if tag in metadata: + return metadata[tag] + return None + + +def tag_value(metadata: Mapping[str, object], tag: str) -> str | None: + value = metadata.get(tag) + return None if value is None else str(value) + + +def parse_exif_datetime(value: object, assume_utc: bool = False) -> datetime | None: + if value is None: + return None + match = DATETIME_RE.search(str(value).strip()) + if not match: + return None + try: + tz_text = match.group("tz") + if tz_text == "Z": + tzinfo = timezone.utc + elif tz_text: + sign = 1 if tz_text[0] == "+" else -1 + digits = tz_text[1:].replace(":", "") + tzinfo = timezone(sign * timedelta(hours=int(digits[:2]), minutes=int(digits[2:]))) + else: + tzinfo = None + parsed = datetime( + int(match.group("Y")), + int(match.group("M")), + int(match.group("D")), + int(match.group("h")), + int(match.group("m")), + int(match.group("s")), + tzinfo=tzinfo, + ) + return parsed.replace(tzinfo=timezone.utc) if assume_utc and tzinfo is None else parsed + except ValueError: + return None + + +def round_half_up(value: float) -> int: + return math.floor(value + 0.5) diff --git a/tools/metadata_copy.py b/tools/metadata_copy.py index 95eae30..134b3a4 100644 --- a/tools/metadata_copy.py +++ b/tools/metadata_copy.py @@ -1,11 +1,8 @@ from __future__ import annotations -import shutil -import subprocess from pathlib import Path from tools.exiftool import run_exiftool_command -from tools.filesystem import unique_path EXCLUDED_COPY_TAGS = [ @@ -70,15 +67,12 @@ def copy_meaningful_metadata( *, extra_excluded_tags: list[str] | None = None, ) -> None: - if destination.suffix.lower() in {".mp4", ".mov"}: - copy_meaningful_metadata_with_exiftool( - exiftool, - source, - destination, - extra_excluded_tags=extra_excluded_tags, - ) - return - copy_container_metadata_with_ffmpeg(source, destination) + copy_meaningful_metadata_with_exiftool( + exiftool, + source, + destination, + extra_excluded_tags=extra_excluded_tags, + ) def copy_meaningful_metadata_with_exiftool( @@ -104,37 +98,3 @@ def copy_meaningful_metadata_with_exiftool( capture_output=True, text=True, ) - - -def copy_container_metadata_with_ffmpeg(source: Path, destination: Path) -> None: - ffmpeg = shutil.which("ffmpeg") - if not ffmpeg: - print(" metadata: ffmpeg not found; skipped best-effort metadata remux") - return - - temp_output = destination.with_name(f"{destination.stem}.metadata-copy{destination.suffix}") - temp_output = unique_path(temp_output) - command = [ - ffmpeg, - "-y", - "-hide_banner", - "-loglevel", - "error", - "-i", - str(destination), - "-i", - str(source), - "-map", - "0", - "-map_metadata", - "1", - "-c", - "copy", - str(temp_output), - ] - try: - subprocess.run(command, check=True, capture_output=True, text=True) - temp_output.replace(destination) - finally: - if temp_output.exists(): - temp_output.unlink() diff --git a/tools/photo_execution.py b/tools/photo_execution.py new file mode 100644 index 0000000..61a454a --- /dev/null +++ b/tools/photo_execution.py @@ -0,0 +1,166 @@ +"""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}") diff --git a/tools/photo_grouping.py b/tools/photo_grouping.py new file mode 100644 index 0000000..4d27348 --- /dev/null +++ b/tools/photo_grouping.py @@ -0,0 +1,136 @@ +"""Burst/HDR group detection for photo metadata cleanup.""" + +from __future__ import annotations + +from collections import defaultdict +from datetime import datetime, timedelta +from typing import TYPE_CHECKING + +from tools.filenames import naive_wall_time + +if TYPE_CHECKING: + from photo_metadata import MediaRecord + + +def photo_sort_key(record: MediaRecord) -> tuple: + assert record.adjusted_time is not None + return ( + naive_wall_time(record.adjusted_time), + record.subsec if record.subsec is not None else -1, + record.sequence if record.sequence is not None else -1, + record.path.name.lower(), + ) + + +def group_sort_key(record: MediaRecord) -> tuple: + timestamp = naive_wall_time(record.adjusted_time) if record.adjusted_time else datetime.max + return ( + timestamp, + record.subsec if record.subsec is not None else -1, + record.sequence if record.sequence is not None else -1, + str(record.path).lower(), + ) + + +def detect_photo_groups(records: list[MediaRecord], min_size: int = 2) -> list[list[MediaRecord]]: + by_camera: dict[str, list[MediaRecord]] = defaultdict(list) + for record in records: + if record.is_image and record.adjusted_time is not None: + by_camera[record.camera_key].append(record) + return [ + group + for camera_records in by_camera.values() + for group in detect_groups_for_camera(sorted(camera_records, key=photo_sort_key), min_size) + ] + + +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 + by_second[naive_wall_time(record.adjusted_time).replace(microsecond=0)].append(record) + + seconds = sorted(by_second) + spans: list[list[datetime]] = [] + for second in seconds: + if not spans or second != spans[-1][-1] + timedelta(seconds=1): + spans.append([]) + spans[-1].append(second) + + groups: list[list[MediaRecord]] = [] + for span in spans: + if max(len(by_second[second]) for second in span) < 2: + continue + span_records = [record for second in span for record in by_second[second]] + if len(span_records) >= 2 and all(record.sequence is not None for record in span_records): + groups.extend(sequence_groups(span_records, min_size)) + else: + 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], min_size: int) -> list[list[MediaRecord]]: + groups: list[list[MediaRecord]] = [] + for record in sorted(records, key=photo_sort_key): + if not groups or record.sequence <= groups[-1][-1].sequence: + groups.append([]) + groups[-1].append(record) + return [group for group in groups if len(group) >= min_size] + + +def timestamp_groups( + 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: + return [] + + clusters: list[list[int]] = [[multi_indices[0]]] + for index in multi_indices[1:]: + if index - clusters[-1][-1] > 2: + clusters.append([]) + clusters[-1].append(index) + + groups: list[list[MediaRecord]] = [] + used_seconds: set[datetime] = set() + for cluster in clusters: + start, end = cluster[0], cluster[-1] + if start and len(by_second[span[start - 1]]) == 1: + start -= 1 + if end + 1 < len(span) and len(by_second[span[end + 1]]) == 1: + end += 1 + seconds = [second for second in span[start : end + 1] if second not in used_seconds] + used_seconds.update(seconds) + records = [record for second in seconds for record in sorted(by_second[second], key=photo_sort_key)] + if len(records) >= min_size: + groups.append(records) + return groups + + +def infer_group_subseconds(groups: list[list[MediaRecord]]) -> None: + for group in groups: + if any(record.subsec is not None for record in group): + continue + by_second: dict[datetime, list[MediaRecord]] = defaultdict(list) + for record in group: + if record.is_image and record.adjusted_time is not None: + by_second[naive_wall_time(record.adjusted_time).replace(microsecond=0)].append(record) + + seconds = sorted(by_second) + counts = {second: len(records) for second, records in by_second.items()} + for index, second in enumerate(seconds): + records = sorted(by_second[second], key=photo_sort_key) + count = len(records) + if count <= 1: + continue + if len(seconds) == 2: + fps = max(counts.values()) + elif index == 0: + fps = max(count, counts[seconds[1]]) + elif index == len(seconds) - 1: + fps = max(count, counts[seconds[-2]]) + else: + fps = count + start_slot = fps - count if index == 0 and fps > count else 0 + for item_index, record in enumerate(records): + record.inferred_subsec = int((start_slot + item_index) * 1000 / fps) diff --git a/tools/photo_planning.py b/tools/photo_planning.py new file mode 100644 index 0000000..319561d --- /dev/null +++ b/tools/photo_planning.py @@ -0,0 +1,383 @@ +"""Pure planning for photo metadata updates, grouping, and PTO projects.""" + +from __future__ import annotations + +import os +import re +from collections import defaultdict +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from pathlib import Path + +from tools.filenames import format_filename_stem, group_stem_from_name, naive_wall_time, normalized_extension +from tools.media_metadata import first_tag, parse_exif_datetime, round_half_up, tag_value +from tools.photo_grouping import detect_photo_groups, group_sort_key, infer_group_subseconds +from tools.photo_records import FILENAME_TIMESTAMP_RE, IMAGE_MODIFY_TIME_TAGS, MediaRecord, TagWrite, parse_filename_timestamp, parse_subsec +from tools.timezones import timezone_to_string + + +IMAGE_WRITE_TIME_TAGS = ( + "EXIF:DateTimeOriginal", "EXIF:CreateDate", "XMP:DateTimeOriginal", "XMP:CreationDate", + "XMP:CreateDate", "EXIF:ModifyDate", "XMP:ModifyDate", +) +OFFSET_TAG_FOR_TIME_TAG = { + "EXIF:DateTimeOriginal": "EXIF:OffsetTimeOriginal", + "EXIF:CreateDate": "EXIF:OffsetTimeDigitized", + "EXIF:ModifyDate": "EXIF:OffsetTime", +} +CAMERA_RANGE_STEM_RE = re.compile(r"^(?P[A-Za-z]*)(?P\d+)-(?P=prefix)(?P\d+)(?P_.*)$") +PTO_QUOTED_VALUE_RE = re.compile(r"([\"'])(.*?)(\1)") + + +@dataclass +class UserChoices: + working_dir: Path + time_offset: timedelta | None + fixed_timezone: timezone | None + fill_timezone_gaps: bool + artist_action: str + artist_value: str | None + rename_mode: str + organize_files: bool + group_min_size: int + process_pto_files: bool + infer_missing_timestamps: bool + + @property + def group_photos(self) -> bool: + return self.organize_files and self.group_min_size >= 2 + + @property + def move_to_working_dir(self) -> bool: + return self.organize_files + + +@dataclass +class PtoPlan: + path: Path + target_path: Path + referenced_records: list[MediaRecord] + updated_text: str + replacements: list[tuple[str, str]] + + +@dataclass +class PtoReference: + path: Path + original_text: str + records: list[MediaRecord] + + +def apply_time_offset(records: list[MediaRecord], offset: timedelta | None) -> None: + if offset is not None: + for record in records: + if record.adjusted_time is not None: + record.adjusted_time += offset + + +def infer_missing_timestamps(records: list[MediaRecord], enabled: bool) -> None: + if not enabled: + return + for record in records: + if record.original_time is not None: + continue + parsed, source = parse_filename_timestamp(record), "filename" + if parsed is None and record.is_image: + parsed, source = parse_exif_datetime(first_tag(record.metadata, IMAGE_MODIFY_TIME_TAGS)), "ModifyDate" + if parsed is None: + continue + if record.is_video: + if record.resolved_timezone is None or record.duration_seconds is None: + continue + parsed = parsed.replace(tzinfo=record.resolved_timezone).astimezone(timezone.utc) + parsed += timedelta(seconds=record.duration_seconds) + record.original_time = record.adjusted_time = parsed + record.inferred_timestamp_source = source + record.parsed_from_filename = source == "filename" + + +def filename_timestamp_stem(stem: str, timestamp: datetime, mode: str) -> str: + matches = [(match, parse_exif_datetime(match.group(0))) for match in FILENAME_TIMESTAMP_RE.finditer(stem)] + matches = [(match, value) for match, value in matches if value is not None] + if mode.startswith("adjust") and matches: + first_time = matches[0][1] + pieces, position, base = [], 0, naive_wall_time(timestamp) + for match, old_time in matches: + replacement = format_filename_stem(base + (old_time - first_time)).replace("_", match.group(0)[8]) + pieces.extend((stem[position:match.start()], replacement)) + position = match.end() + return "".join((*pieces, stem[position:])) + formatted = format_filename_stem(timestamp) + return f"{formatted} {stem}" if mode.endswith("add") else formatted + + +def plan_names(records: list[MediaRecord], choices: UserChoices) -> None: + by_stem = _unique_records_by_stem(records) + for record in records: + record.new_stem = record.rename_skip_reason = None + if record.adjusted_time is None: + record.new_stem = _camera_range_stem(record, by_stem) + record.rename_skip_reason = None if record.new_stem else "missing timestamp" + elif record.is_video: + if record.duration_seconds is None: + record.rename_skip_reason = "missing duration" + elif record.resolved_timezone is None: + record.rename_skip_reason = "missing video timezone" + else: + begin = record.adjusted_time if record.video_time_is_beginning else record.adjusted_time - timedelta(seconds=round_half_up(record.duration_seconds)) + if begin.tzinfo is None: + begin = begin.replace(tzinfo=timezone.utc) + record.new_stem = filename_timestamp_stem(record.path.stem, begin.astimezone(record.resolved_timezone), choices.rename_mode) + else: + record.new_stem = filename_timestamp_stem(record.path.stem, record.adjusted_time, choices.rename_mode) + _assign_unique_names(records) + + +def _unique_records_by_stem(records: list[MediaRecord]) -> dict[str, MediaRecord]: + buckets: dict[str, list[MediaRecord]] = defaultdict(list) + for record in records: + buckets[record.path.stem.lower()].append(record) + return {stem: bucket[0] for stem, bucket in buckets.items() if len(bucket) == 1} + + +def _camera_range_stem(record: MediaRecord, by_stem: dict[str, MediaRecord]) -> str | None: + match = CAMERA_RANGE_STEM_RE.match(record.path.stem) + if not match or int(match.group("last")) <= int(match.group("first")): + return None + prefix = match.group("prefix") + first = by_stem.get(f"{prefix}{match.group('first')}".lower()) + last = by_stem.get(f"{prefix}{match.group('last')}".lower()) + start = record.adjusted_time or (first.adjusted_time if first else None) or (last.adjusted_time if last else None) + end = (last.adjusted_time if last else None) or (first.adjusted_time if first else None) or start + return f"{format_filename_stem(start)}-{format_filename_stem(end)}{match.group('suffix')}" if start and end else None + + +def _assign_unique_names(records: list[MediaRecord]) -> None: + buckets: dict[tuple[int | str, str], list[MediaRecord]] = defaultdict(list) + for record in records: + if record.new_stem: + buckets[(record.group_id if record.group_id is not None else str(record.path.parent.resolve()), record.new_stem.lower())].append(record) + for bucket in buckets.values(): + if len(bucket) == 1: + bucket[0].final_name = f"{bucket[0].new_stem}{normalized_extension(bucket[0].path)}" + continue + for index, record in enumerate(sorted(bucket, key=group_sort_key), 1): + record.final_name = f"{record.new_stem}-{index}{normalized_extension(record.path)}" + + +def merge_record_groups(records: list[MediaRecord], candidates: list[list[MediaRecord]]) -> list[list[MediaRecord]]: + indexes = {record.path.resolve(): index for index, record in enumerate(records)} + parents = list(range(len(records))) + grouped: set[int] = set() + + def find(index: int) -> int: + while parents[index] != index: + parents[index] = parents[parents[index]] + index = parents[index] + return index + + for candidate in candidates: + members = [indexes[record.path.resolve()] for record in candidate if record.path.resolve() in indexes] + if members: + grouped.update(members) + for index in members[1:]: + parents[find(index)] = find(members[0]) + merged: dict[int, list[MediaRecord]] = defaultdict(list) + for index in grouped: + merged[find(index)].append(records[index]) + return sorted((sorted(group, key=group_sort_key) for group in merged.values()), key=lambda group: group_sort_key(group[0])) + + +def assign_groups(records: list[MediaRecord], groups: list[list[MediaRecord]]) -> None: + for record in records: + record.group_id = record.group_name = None + for index, group in enumerate(groups, 1): + for record in group: + record.group_id = index + + +def _group_name(base: str, working_dir: Path, used: set[str]) -> str: + if (working_dir / base).exists() or base not in used: + used.add(base) + return base + for index in range(1, 10000): + name = f"{base}-{index}" + if name not in used and not (working_dir / name).exists(): + used.add(name) + return name + raise RuntimeError(f"Could not find an available group directory name for {base}") + + +def assign_group_names(groups: list[list[MediaRecord]], working_dir: Path) -> None: + used: set[str] = set() + for group in groups: + ordered = sorted(group, key=group_sort_key) + first, last = (group_stem_from_name(record.final_name or record.path.name) for record in (ordered[0], ordered[-1])) + name = _group_name(f"{first}-{last}", working_dir, used) + for record in group: + record.group_name = name + + +def plan_targets(records: list[MediaRecord], choices: UserChoices) -> None: + for record in records: + directory = choices.working_dir if choices.organize_files else record.path.parent + if record.group_name: + directory /= record.group_name + name = record.final_name or record.path.name + record.target_path = directory / name + if record.sidecar: + record.target_sidecar = directory / f"{Path(name).stem}M01.XML" + + +def normalize_reference(value: str) -> str: + return value.replace("\\", "/") + + +def _pto_mapping(pto_path: Path, records: list[MediaRecord]) -> dict[str, MediaRecord]: + basename_counts: dict[str, int] = defaultdict(int) + for record in records: + basename_counts[record.path.name] += 1 + mapping: dict[str, MediaRecord] = {} + for record in records: + candidates = {normalize_reference(str(record.path.resolve())), normalize_reference(os.path.relpath(record.path.resolve(), pto_path.parent))} + if basename_counts[record.path.name] == 1: + candidates.add(record.path.name) + for candidate in candidates: + mapping[normalize_reference(candidate)] = record + return mapping + + +def _pto_target(path: Path, directory: Path, base: str, used: set[Path]) -> Path: + for index in range(10000): + target = directory / f"{base}{'' if index == 0 else f'-{index}'}.pto" + if target.resolve() == path.resolve() or (not target.exists() and target not in used): + used.add(target) + return target + raise RuntimeError(f"Could not find an available .pto name for {base}") + + +def read_pto_references(pto_files: list[Path], records: list[MediaRecord]) -> list[PtoReference]: + references = [] + for path in pto_files: + try: + text = path.read_text(encoding="utf-8") + except UnicodeDecodeError: + text = path.read_text(encoding="utf-8", errors="replace") + mapping = _pto_mapping(path, records) + found = [] + for match in PTO_QUOTED_VALUE_RE.finditer(text): + record = mapping.get(normalize_reference(match.group(2))) + if record is not None and record not in found: + found.append(record) + if found: + references.append(PtoReference(path, text, found)) + return references + + +def plan_pto_updates(references: list[PtoReference], records: list[MediaRecord]) -> list[PtoPlan]: + plans, used = [], set() + for reference in references: + referenced = sorted(reference.records, key=group_sort_key) + target_dir = referenced[0].target_path.parent if referenced[0].group_name and referenced[0].target_path else reference.path.parent + mapping, replacements = _pto_mapping(reference.path, records), [] + + def replace(match: re.Match) -> str: + value = match.group(2) + record = mapping.get(normalize_reference(value)) + if record is None or record.target_path is None: + return match.group(0) + replacement = normalize_reference(os.path.relpath(record.target_path.resolve(), target_dir)) + if value != replacement: + replacements.append((value, replacement)) + return f"{match.group(1)}{replacement}{match.group(1)}" + + text = PTO_QUOTED_VALUE_RE.sub(replace, reference.original_text) + base = f"{group_stem_from_name(referenced[0].target_path.name)}-{group_stem_from_name(referenced[-1].target_path.name)}" + target = _pto_target(reference.path, target_dir, base, used) + if text != reference.original_text or target.resolve() != reference.path.resolve(): + plans.append(PtoPlan(reference.path, target, referenced, text, replacements)) + return plans + + +def _same_text(current: str | None, value: str) -> bool: + return (current or "") == value + + +def _same_datetime(current: str | None, value: str) -> bool: + current_time, new_time = parse_exif_datetime(current), parse_exif_datetime(value) + return current == value if current_time is None or new_time is None else naive_wall_time(current_time) == naive_wall_time(new_time) + + +def _same_subsec(current: str | None, value: str) -> bool: + parsed = parse_subsec(current) + return current == value if parsed is None else f"{parsed:03d}" == value + + +def _add_write(writes: list[TagWrite], label: str, tag: str, current: str | None, value: str, same=_same_text) -> None: + writes.append(TagWrite(label, f"-{tag}={value}", current, value, not same(current, value))) + + +def build_write_plan(record: MediaRecord, choices: UserChoices) -> list[TagWrite]: + writes: list[TagWrite] = [] + if choices.artist_action != "leave": + value = choices.artist_value or "" + for label, tag in (("Artist", "Artist"), ("Author", "Author")): + _add_write(writes, label, tag, tag_value(record.metadata, f"EXIF:{tag}") or tag_value(record.metadata, f"XMP:{tag}"), value) + if record.is_image: + capture_tags: set[str] = set() + if record.inferred_timestamp_source and record.adjusted_time: + capture_tags.add("EXIF:DateTimeOriginal") + _add_write(writes, "DateTimeOriginal", "EXIF:DateTimeOriginal", tag_value(record.metadata, "EXIF:DateTimeOriginal"), record.adjusted_time.strftime("%Y:%m:%d %H:%M:%S"), _same_datetime) + if record.original_time and record.adjusted_time: + original, adjusted = naive_wall_time(record.original_time), naive_wall_time(record.adjusted_time) + for tag in IMAGE_WRITE_TIME_TAGS: + current = parse_exif_datetime(tag_value(record.metadata, tag)) + if current is None: + continue + current_time = naive_wall_time(current) + matches_capture = abs((current_time - original).total_seconds()) <= 1 + if matches_capture: + capture_tags.add(tag) + target = current_time + choices.time_offset if choices.time_offset and matches_capture else current_time + if tag.endswith("ModifyDate") and target < adjusted: + target, capture_tags = adjusted, capture_tags | {tag} + if "ModifyDate was earlier than capture time and will be clamped" not in record.warnings: + record.warnings.append("ModifyDate was earlier than capture time and will be clamped") + if target != current_time: + _add_write(writes, tag.split(":", 1)[1], tag, tag_value(record.metadata, tag), target.strftime("%Y:%m:%d %H:%M:%S"), _same_datetime) + if record.resolved_timezone and (choices.fixed_timezone or record.timezone_value is None): + for time_tag, offset_tag in OFFSET_TAG_FOR_TIME_TAG.items(): + if time_tag in capture_tags: + _add_write(writes, offset_tag.split(":", 1)[1], offset_tag, tag_value(record.metadata, offset_tag), timezone_to_string(record.resolved_timezone)) + if record.inferred_subsec is not None: + for label, tag in (("SubSecTimeOriginal", "EXIF:SubSecTimeOriginal"), ("SubSecTimeDigitized", "EXIF:SubSecTimeDigitized"), ("SubSecTime", "EXIF:SubSecTime")): + _add_write(writes, label, tag, tag_value(record.metadata, tag), f"{record.inferred_subsec:03d}", _same_subsec) + elif record.is_video and (choices.time_offset or record.inferred_timestamp_source) and record.adjusted_time: + value = record.adjusted_time.replace(tzinfo=record.adjusted_time.tzinfo or timezone.utc).astimezone(timezone.utc).strftime("%Y:%m:%d %H:%M:%S") + for tag in ("QuickTime:CreateDate", "QuickTime:ModifyDate", "QuickTime:TrackCreateDate", "QuickTime:TrackModifyDate", "QuickTime:MediaCreateDate", "QuickTime:MediaModifyDate"): + _add_write(writes, tag.split(":", 1)[1], tag, tag_value(record.metadata, tag), value, _same_datetime) + return writes + + +def build_write_args(record: MediaRecord, choices: UserChoices) -> list[str]: + if not record.write_plan: + record.write_plan = build_write_plan(record, choices) + return [write.arg for write in record.write_plan if write.will_write] + + +def prepare_plan(records: list[MediaRecord], choices: UserChoices, references: list[PtoReference]) -> list[list[MediaRecord]]: + infer_missing_timestamps(records, choices.infer_missing_timestamps) + apply_time_offset(records, choices.time_offset) + burst_groups = detect_photo_groups(records, choices.group_min_size) if choices.group_photos else [] + groups = merge_record_groups(records, [*burst_groups, *(reference.records for reference in references if choices.process_pto_files)]) + if groups: + assign_groups(records, groups) + if choices.organize_files: + infer_group_subseconds(burst_groups) + plan_names(records, choices) + if groups: + assign_group_names(groups, choices.working_dir) + plan_targets(records, choices) + for record in records: + record.write_plan = build_write_plan(record, choices) + return groups diff --git a/tools/photo_records.py b/tools/photo_records.py new file mode 100644 index 0000000..77510f3 --- /dev/null +++ b/tools/photo_records.py @@ -0,0 +1,251 @@ +"""Photo/video record types and ExifTool metadata decoding.""" + +from __future__ import annotations + +import re +import xml.etree.ElementTree as ET +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +from pathlib import Path + +from tools.filenames import naive_wall_time +from tools.media_metadata import first_tag, parse_exif_datetime +from tools.timezones import parse_timezone_offset + + +IMAGE_EXTS = {".jpg", ".jpeg", ".heic", ".arw"} +VIDEO_EXTS = {".mp4", ".mov", ".mts"} +SUPPORTED_EXTS = IMAGE_EXTS | VIDEO_EXTS + +IMAGE_CAPTURE_TIME_TAGS = ( + "Composite:SubSecDateTimeOriginal", + "Composite:SubSecCreateDate", + "EXIF:DateTimeOriginal", + "XMP:DateTimeOriginal", + "XMP:CreationDate", + "XMP:CreateDate", + "EXIF:CreateDate", +) +IMAGE_MODIFY_TIME_TAGS = ("Composite:SubSecModifyDate", "EXIF:ModifyDate", "XMP:ModifyDate") +IMAGE_OFFSET_TAGS = ( + "EXIF:OffsetTimeOriginal", + "EXIF:OffsetTimeDigitized", + "EXIF:OffsetTime", +) +VIDEO_TIME_TAGS = ( + "QuickTime:MediaCreateDate", + "QuickTime:TrackCreateDate", + "QuickTime:CreateDate", + "QuickTime:MediaModifyDate", + "QuickTime:TrackModifyDate", + "QuickTime:ModifyDate", + "XMP:DateTimeOriginal", + "XMP:CreateDate", +) +VIDEO_BEGIN_TIME_TAGS = ("H264:DateTimeOriginal",) +VIDEO_DURATION_TAGS = ( + "QuickTime:Duration", + "Composite:Duration", + "File:Duration", + "M2TS:Duration", +) +VIDEO_TIMEZONE_TAGS = ("QuickTime:AndroidTimeZone", "QuickTime:Keys:AndroidTimeZone") +SEQUENCE_TAGS = ( + "MakerNotes:SequenceNumber", + "MakerNotes:SequenceNumberOriginal", + "MakerNotes:ImageNumber", + "EXIF:ImageNumber", + "SequenceNumber", +) +FILENAME_TIMESTAMP_RE = re.compile(r"(?P\d{8})[-_](?P