media-batch-tools/photo_metadata.py

#!/usr/bin/env python3
"""
Photo collection metadata cleanup.

Drag files or directories onto this script, or run:
    python3 photo_metadata.py path/to/file/or/directory [...]

The script reads metadata with exiftool, builds a full preview, then edits files in
place only after confirmation.
"""

from __future__ import annotations

import json
import math
import os
import re
import shutil
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 shutil import which
from typing import Iterable


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<Y>\d{4})[:\-]?(?P<M>\d{2})[:\-]?(?P<D>\d{2})"
    r"(?:[ T_])?"
    r"(?P<h>\d{2}):?(?P<m>\d{2}):?(?P<s>\d{2})"
    r"(?:[.,](?P<sub>\d+))?"
    r"(?:\s*(?P<tz>Z|[+-]\d{2}:?\d{2}))?"
)
TIME_ONLY_RE = re.compile(
    r"^(?P<h>\d{1,2})(?::?(?P<m>\d{2}))(?::?(?P<s>\d{2}))?$"
)
TZ_RE = re.compile(r"^(?P<sign>[+-])(?P<h>\d{1,2})(?::?(?P<m>\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<timestamp>\d{8}_\d{6})-[1-9]\d*$")


@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 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
    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 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
    move_to_working_dir: bool
    apply_filename_timestamps: bool


def ask_yes_no(prompt: str, default: bool) -> bool:
    suffix = "Y/n" if default else "y/N"
    while True:
        answer = prompt_input(f"{prompt} ({suffix}) ").strip().lower()
        if not answer:
            return default
        if answer in {"y", "yes"}:
            return True
        if answer in {"n", "no"}:
            return False
        print("Please answer y or n.")


def prompt_input(prompt: str) -> str:
    try:
        return input(prompt)
    except EOFError as exc:
        raise RuntimeError("Interactive input ended before choices were complete.") from exc


def parse_timezone_offset(value: str) -> timezone:
    match = TZ_RE.match(value.strip())
    if not match:
        raise ValueError("timezone must look like +09:00 or -05:30")
    sign = 1 if match.group("sign") == "+" else -1
    hours = int(match.group("h"))
    minutes = int(match.group("m") or "0")
    if minutes not in {0, 15, 30, 45}:
        raise ValueError("timezone minutes must be 00, 15, 30, or 45")
    return timezone(sign * timedelta(hours=hours, minutes=minutes))


def timezone_to_string(tz: timezone) -> str:
    offset = tz.utcoffset(None)
    if offset is None:
        return "+00:00"
    total_minutes = int(offset.total_seconds() // 60)
    sign = "+" if total_minutes >= 0 else "-"
    total_minutes = abs(total_minutes)
    return f"{sign}{total_minutes // 60:02d}:{total_minutes % 60:02d}"


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<h>\d+):)?(?P<m>\d{1,2}):(?P<s>\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 naive_wall_time(dt: datetime) -> datetime:
    return dt.replace(tzinfo=None)


def collect_media_files(args: list[str]) -> list[Path]:
    files: list[Path] = []
    seen: set[Path] = set()

    for arg in args:
        path = Path(arg).expanduser()
        if path.is_file():
            candidates = [path] if path.suffix.lower() in SUPPORTED_EXTS else []
        elif path.is_dir():
            candidates = [
                child
                for child in path.rglob("*")
                if child.is_file() and child.suffix.lower() in SUPPORTED_EXTS
            ]
        else:
            print(f"Skipping missing path: {arg}")
            continue

        for candidate in candidates:
            resolved = candidate.resolve()
            if resolved not in seen:
                seen.add(resolved)
                files.append(resolved)

    return sorted(files, key=lambda p: str(p).lower())


def existing_input_paths(args: list[str]) -> list[Path]:
    paths: list[Path] = []
    for arg in args:
        path = Path(arg).expanduser()
        if path.exists():
            paths.append(path.resolve())
    return paths


def determine_working_dir(args: list[str]) -> Path:
    paths = existing_input_paths(args)
    if not paths:
        raise RuntimeError("No existing input paths were provided.")
    if len(paths) == 1:
        path = paths[0]
        return path if path.is_dir() else path.parent

    try:
        common = Path(os.path.commonpath([str(path) for path in paths]))
    except ValueError as exc:
        raise RuntimeError("Input paths do not have a common ancestor.") from exc

    if str(common) == common.anchor:
        raise RuntimeError("Input paths only share the filesystem root; choose items closer together.")
    return common


def require_exiftool() -> str:
    executable = which("exiftool")
    if executable:
        return executable
    raise RuntimeError(
        "exiftool was not found on PATH. Install ExifTool and make sure the "
        "'exiftool' command works before running this script."
    )


def run_exiftool_json(executable: str, paths: list[Path]) -> list[dict]:
    results: list[dict] = []
    for start in range(0, len(paths), 80):
        chunk = paths[start : start + 80]
        command = [
            executable,
            "-j",
            "-G",
            "-api",
            "QuickTimeUTC=1",
            "-charset",
            "filename=UTF8",
            *[str(path) for path in chunk],
        ]
        completed = subprocess.run(command, check=True, capture_output=True, text=True)
        results.extend(json.loads(completed.stdout or "[]"))
    return results


def run_exiftool_write(executable: str, path: Path, args: list[str]) -> None:
    if not args:
        return
    command = [
        executable,
        "-overwrite_original",
        "-charset",
        "filename=UTF8",
        *args,
        str(path),
    ]
    subprocess.run(command, check=True)


def first_tag(metadata: dict, tags: Iterable[str]) -> object | None:
    for tag in tags:
        if tag in metadata:
            return metadata[tag]
    return None


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) -> 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_photos = False
    if sum(1 for record in records if record.is_image and record.adjusted_time) >= 2:
        group_photos = ask_yes_no("Create automatic burst/HDR groups?", default=True)

    move_to_working_dir = False
    if any(record.path.parent.resolve() != working_dir for record in records):
        move_to_working_dir = ask_yes_no(
            f"Move ungrouped 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,
        move_to_working_dir=move_to_working_dir,
        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 detect_photo_groups(records: list[MediaRecord]) -> 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)))

    return groups


def detect_groups_for_camera(records: list[MediaRecord]) -> 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))
        else:
            groups.extend(timestamp_groups(span, by_second))
    return [group for group in groups if len(group) >= 2]


def sequence_groups(records: list[MediaRecord]) -> 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) >= 2:
                groups.append(current)
            current = [record]
    if len(current) >= 2:
        groups.append(current)
    return groups


def timestamp_groups(
    span: list[datetime], by_second: dict[datetime, list[MediaRecord]]
) -> 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) >= 2:
            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:
            assert record.adjusted_time is not None
            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 format_filename_stem(dt: datetime) -> str:
    return naive_wall_time(dt).strftime("%Y%m%d_%H%M%S")


def normalized_extension(path: Path) -> str:
    return path.suffix.lower()


def group_stem_from_name(name: str) -> str:
    stem = Path(name).stem
    match = ORDERED_TIMESTAMP_STEM_RE.match(stem)
    if match:
        return match.group("timestamp")
    return stem


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

    for record in records:
        if record.adjusted_time is None:
            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 = format_filename_stem(local_begin)
        else:
            record.new_stem = format_filename_stem(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 assign_groups(records: list[MediaRecord], groups: list[list[MediaRecord]]) -> None:
    for index, group in enumerate(groups, start=1):
        ordered = sorted(group, key=photo_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 base_name not in used_names and not (working_dir / base_name).exists():
        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=photo_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 build_write_args(record: MediaRecord, choices: UserChoices) -> list[str]:
    args: list[str] = []
    if choices.artist_action == "set":
        assert choices.artist_value is not None
        args.extend([f"-Artist={choices.artist_value}", f"-Author={choices.artist_value}"])
    elif choices.artist_action == "clear":
        args.extend(["-Artist=", "-Author="])

    if record.is_image:
        if choices.time_offset is not None or record.parsed_from_filename:
            if record.adjusted_time is not None:
                args.append(f"-AllDates={record.adjusted_time.strftime('%Y:%m:%d %H:%M:%S')}")

        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)
            args.extend(
                [
                    f"-EXIF:OffsetTimeOriginal={tz_text}",
                    f"-EXIF:OffsetTimeDigitized={tz_text}",
                    f"-EXIF:OffsetTime={tz_text}",
                ]
            )

        if record.inferred_subsec is not None:
            args.extend(
                [
                    f"-SubSecTimeOriginal={record.inferred_subsec:03d}",
                    f"-SubSecTimeDigitized={record.inferred_subsec:03d}",
                    f"-SubSecTime={record.inferred_subsec:03d}",
                ]
            )
    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")
        args.extend(
            [
                f"-QuickTime:CreateDate={utc_text}",
                f"-QuickTime:ModifyDate={utc_text}",
                f"-QuickTime:TrackCreateDate={utc_text}",
                f"-QuickTime:TrackModifyDate={utc_text}",
                f"-QuickTime:MediaCreateDate={utc_text}",
                f"-QuickTime:MediaModifyDate={utc_text}",
            ]
        )

    return args


def print_preview(records: list[MediaRecord], choices: UserChoices) -> 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(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")

    rows = []
    for record in sorted(records, key=lambda item: str(item.path).lower()):
        target = record.target_path or record.path
        operation = []
        if target != record.path:
            operation.append(f"rename/move -> {target}")
        if record.inferred_subsec is not None:
            operation.append(f"subsec {record.inferred_subsec:03d}")
        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:
            operation.append(f"sidecar -> {record.target_sidecar}")
        rows.append((str(record.path), "; ".join(operation) or "metadata only/no change"))

    path_width = min(max((len(row[0]) for row in rows), default=10), 70)
    for source, operation in rows:
        print(f"{source:<{path_width}}  {operation}")


def unique_existing_target(path: Path, planned_sources: set[Path]) -> Path:
    if not path.exists() or path.resolve() in planned_sources:
        return path
    for index in range(1, 10000):
        candidate = path.with_name(f"{path.stem}-{index}{path.suffix}")
        if not candidate.exists() or candidate.resolve() in planned_sources:
            return candidate
    raise RuntimeError(f"Could not find available target name for {path}")


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):
        try:
            write_args = build_write_args(record, choices)
            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}")
        print(f"Processed {index} of {len(records)}")

    return failures


def prepare_plan(records: list[MediaRecord], choices: UserChoices) -> list[list[MediaRecord]]:
    apply_time_offset(records, choices.time_offset)

    groups: list[list[MediaRecord]] = []
    if choices.group_photos:
        groups = detect_photo_groups(records)
        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)
    return groups


def pause_if_interactive() -> None:
    if sys.stdin.isatty():
        try:
            input("\nPress Enter to close...")
        except EOFError:
            pass


def run(argv: list[str]) -> int:
    if len(argv) < 2:
        print("Usage: python3 photo_metadata.py path/to/file/or/directory [...]")
        return 2

    try:
        working_dir = determine_working_dir(argv[1:])
    except RuntimeError as exc:
        print(exc)
        return 1

    files = collect_media_files(argv[1:])
    if not files:
        print("No supported media files found.")
        return 1

    try:
        exiftool = require_exiftool()
    except RuntimeError as exc:
        print(exc)
        return 1

    print(f"Found {len(files)} supported media file(s). Reading metadata...")
    try:
        metadata = run_exiftool_json(exiftool, files)
    except (subprocess.CalledProcessError, json.JSONDecodeError) as exc:
        print(f"Failed to read metadata with exiftool: {exc}")
        return 1

    records = build_records(files, metadata)
    try:
        choices = prompt_choices(records, working_dir)
    except RuntimeError as exc:
        print(exc)
        return 1

    prepare_plan(records, choices)
    print_preview(records, choices)

    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

    failures = apply_changes(exiftool, records, choices)
    if failures:
        print("\nSome files failed:")
        for failure in failures:
            print(f" - {failure}")
        return 1

    print("\nDone.")
    return 0


def main(argv: list[str]) -> int:
    try:
        return run(argv)
    finally:
        pause_if_interactive()


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))