544 lines
22 KiB
Python
544 lines
22 KiB
Python
"""Raw-key settings UI for photo metadata cleanup."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
from tools.console import dark_field, dim, draw_screen, inverse, light_red, raw_key_input
|
|
from tools.exiftool import run_exiftool_json
|
|
from tools.filenames import naive_wall_time
|
|
from tools.media_metadata import DATETIME_RE
|
|
from tools.photo_planning import UserChoices
|
|
from tools.photo_records import IMAGE_EXTS, MediaRecord, build_records
|
|
from tools.timezones import local_timezone, parse_timezone_offset, timezone_to_string
|
|
|
|
TIME_ONLY_RE = re.compile(r"^(?P<h>\d{1,2})(?::?(?P<m>\d{2}))(?::?(?P<s>\d{2}))?$")
|
|
TOKEN_SHIFT_RE = re.compile(r"(\d+|[a-zA-Z]+)")
|
|
|
|
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_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 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 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}"
|
|
|
|
|
|
@dataclass
|
|
class TimezoneAnalysis:
|
|
ordered_zones: list[timezone]
|
|
gap_count: int
|
|
conflicting_gap_count: int
|
|
|
|
@property
|
|
def shifts(self) -> int:
|
|
return max(0, len(self.ordered_zones) - 1)
|
|
|
|
|
|
@dataclass
|
|
class SettingsState:
|
|
time_offset: timedelta | None
|
|
infer_missing_timestamps: bool
|
|
fixed_timezone: timezone
|
|
fixed_timezone_enabled: bool
|
|
fill_timezone_gaps: bool
|
|
rename_mode: str = "adjust_replace"
|
|
artist_enabled: bool = False
|
|
artist_value: str = ""
|
|
organize_files: bool = False
|
|
group_min_size: int = 2
|
|
process_pto_files: bool = True
|
|
|
|
|
|
def timeline_key(record: MediaRecord) -> tuple[datetime, str]:
|
|
value = record.original_time
|
|
if value is None:
|
|
return datetime.max, record.path.name.lower()
|
|
if value.tzinfo is not None:
|
|
value = value.astimezone(timezone.utc)
|
|
return naive_wall_time(value), record.path.name.lower()
|
|
|
|
|
|
def analyze_timezones(records: list[MediaRecord]) -> TimezoneAnalysis:
|
|
ordered = sorted(records, key=timeline_key)
|
|
runs: list[timezone] = []
|
|
for record in ordered:
|
|
if record.timezone_value is not None and (not runs or record.timezone_value != runs[-1]):
|
|
runs.append(record.timezone_value)
|
|
gaps, conflicts = timezone_gaps(ordered)
|
|
return TimezoneAnalysis(runs, len(gaps), len(conflicts))
|
|
|
|
|
|
def timezone_gaps(records: list[MediaRecord]) -> tuple[list[list[MediaRecord]], list[list[MediaRecord]]]:
|
|
gaps: list[list[MediaRecord]] = []
|
|
conflicts: list[list[MediaRecord]] = []
|
|
index = 0
|
|
while index < len(records):
|
|
if records[index].timezone_value is not None:
|
|
index += 1
|
|
continue
|
|
start = index
|
|
while index < len(records) and records[index].timezone_value is None:
|
|
index += 1
|
|
gap = records[start:index]
|
|
before = records[start - 1].timezone_value if start else None
|
|
after = records[index].timezone_value if index < len(records) else None
|
|
if before is not None and after is not None and before != after:
|
|
conflicts.append(gap)
|
|
else:
|
|
gaps.append(gap)
|
|
return gaps, conflicts
|
|
|
|
|
|
def apply_timezone_settings(records: list[MediaRecord], state: SettingsState) -> None:
|
|
for record in records:
|
|
record.resolved_timezone = state.fixed_timezone if state.fixed_timezone_enabled else record.timezone_value
|
|
if not state.fill_timezone_gaps or state.fixed_timezone_enabled:
|
|
return
|
|
ordered = sorted(records, key=timeline_key)
|
|
for gap in timezone_gaps(ordered)[0]:
|
|
start = ordered.index(gap[0])
|
|
end = ordered.index(gap[-1]) + 1
|
|
before = ordered[start - 1].resolved_timezone if start else None
|
|
after = ordered[end].resolved_timezone if end < len(ordered) else None
|
|
zone = before or after
|
|
if zone is not None:
|
|
for record in gap:
|
|
record.resolved_timezone = zone
|
|
|
|
|
|
def parse_clock_time(value: str, reference: datetime) -> datetime:
|
|
match = re.fullmatch(r"(\d{1,2}):(\d{1,2}):(\d{1,2})", value.strip())
|
|
if match is None:
|
|
raise ValueError("Time must be h:mm:ss.")
|
|
hour, minute, second = (int(part) for part in match.groups())
|
|
if hour > 23 or minute > 59 or second > 59:
|
|
raise ValueError("Time is outside the 24-hour range.")
|
|
return infer_reference_target(reference, f"{hour:02d}:{minute:02d}:{second:02d}")
|
|
|
|
|
|
def reference_record(executable: str, records: list[MediaRecord], value: str) -> MediaRecord:
|
|
text = value.strip().strip('"')
|
|
absolute = os.path.isabs(text) or bool(re.match(r"^[A-Za-z]:[\\/]", text) or text.startswith("\\\\"))
|
|
candidates = records if absolute else [record for record in records if record.path.name.lower() == text.lower()]
|
|
if not candidates and not absolute:
|
|
candidates = [record for record in records if record.path.stem.lower() == text.lower()]
|
|
if absolute:
|
|
path = Path(text)
|
|
if not path.is_file():
|
|
raise ValueError("File not found")
|
|
if path.suffix.lower() not in IMAGE_EXTS:
|
|
raise ValueError("Only photos are supported")
|
|
metadata = run_exiftool_json(executable, [path.resolve()])
|
|
candidates = build_records([path.resolve()], metadata)
|
|
photos = [record for record in candidates if record.is_image]
|
|
if not photos:
|
|
if candidates:
|
|
raise ValueError("Only photos are supported")
|
|
raise ValueError("Timeshift syntax error, or no such file was given")
|
|
if len(photos) != 1:
|
|
raise ValueError("Multiple photos matched. Please use an absolute path")
|
|
if photos[0].original_time is None:
|
|
raise ValueError("Reference photo has no capture timestamp")
|
|
return photos[0]
|
|
|
|
|
|
def option_text(text: str, selected: bool, active: bool = True) -> str:
|
|
value = f"{text}"
|
|
if not active:
|
|
return dim(value)
|
|
return inverse(value) if selected else value
|
|
|
|
|
|
def timezone_label(analysis: TimezoneAnalysis) -> str:
|
|
if not analysis.ordered_zones:
|
|
return "none"
|
|
if analysis.shifts > 2:
|
|
return "many timezones"
|
|
return ", ".join(timezone_to_string(zone) for zone in analysis.ordered_zones)
|
|
|
|
|
|
def default_artist(records: list[MediaRecord]) -> str:
|
|
values = {
|
|
str(value).strip()
|
|
for record in records
|
|
for tag in ("EXIF:Artist", "XMP:Artist", "EXIF:Author", "XMP:Author")
|
|
if (value := record.metadata.get(tag)) and str(value).strip()
|
|
}
|
|
return values.pop() if len(values) == 1 else ""
|
|
|
|
|
|
def settings_lines(
|
|
state: SettingsState,
|
|
analysis: TimezoneAnalysis,
|
|
working_dir: Path,
|
|
rows: list[str],
|
|
row: int,
|
|
option: int,
|
|
*,
|
|
editor: str | None = None,
|
|
text: str = "",
|
|
message: str = "",
|
|
reference: MediaRecord | None = None,
|
|
) -> list[str]:
|
|
selected = rows[row]
|
|
time_value = "none" if state.time_offset is None else format_timedelta(state.time_offset)
|
|
time_options = [
|
|
option_text("none", option == 0),
|
|
option_text(time_value if state.time_offset else "shift", option == 1),
|
|
]
|
|
rename_names = ["Adjust/replace", "Adjust/add", "Replace", "Add"]
|
|
rename_options = [option_text(name, option == index) for index, name in enumerate(rename_names)]
|
|
fixed = timezone_to_string(state.fixed_timezone)
|
|
fixed_option = f"Set {fixed} ({'on' if state.fixed_timezone_enabled else 'off'})"
|
|
timezone_options = [
|
|
option_text(fixed_option, option == 0),
|
|
option_text("Fill gaps", option == 1, analysis.gap_count > 0),
|
|
]
|
|
group_options = [option_text("Yes" if state.organize_files else "No", option == 0)]
|
|
if selected == "organize" or state.organize_files:
|
|
group_options.extend(
|
|
(
|
|
option_text(f"{state.group_min_size}+ files", option == 1, state.organize_files),
|
|
option_text(".pto", option == 2, state.organize_files and state.process_pto_files),
|
|
)
|
|
)
|
|
if selected != "time":
|
|
time_options = [time_value]
|
|
if selected != "rename":
|
|
rename_options = [rename_names[("adjust_replace", "adjust_add", "replace", "add").index(state.rename_mode)]]
|
|
if selected != "timezone":
|
|
enabled = []
|
|
if state.fixed_timezone_enabled:
|
|
enabled.append(f"Set {fixed}")
|
|
if state.fill_timezone_gaps:
|
|
enabled.append("Fill gaps")
|
|
timezone_options = enabled or ["none"]
|
|
artist = state.artist_value if state.artist_enabled else dim("no change")
|
|
if state.artist_enabled and not artist:
|
|
artist = light_red("(remove)")
|
|
if selected == "artist":
|
|
artist = option_text("no change", True) if not state.artist_enabled else option_text(artist, True)
|
|
if selected != "organize" and not state.organize_files:
|
|
group_options = ["No"]
|
|
lines = [
|
|
f"Working directory: {working_dir}",
|
|
"",
|
|
f"Time correction: {' '.join(time_options)}",
|
|
]
|
|
if "infer" in rows:
|
|
lines.append(f"Infer missing timestamps: {option_text('Yes' if state.infer_missing_timestamps else 'No', selected == 'infer')}")
|
|
lines.extend(
|
|
[
|
|
f"Timezone offsets: Current offsets: {timezone_label(analysis)}",
|
|
f"{'':28}{' '.join(timezone_options)}",
|
|
f"Rename files to timestamps: {' '.join(rename_options)}",
|
|
f"Artist/author: {artist}",
|
|
f"Organize/group: {' '.join(group_options)}",
|
|
]
|
|
)
|
|
if "confirm" in rows:
|
|
confirm_options = [option_text("Confirm", option == 0), option_text("Exit", option == 1)]
|
|
lines.extend(["", f"{'':28}{' '.join(confirm_options) if selected == 'confirm' else 'Confirm Exit'}"])
|
|
if editor:
|
|
lines.extend(["", light_red(message)] if message else [""])
|
|
if editor == "time-source":
|
|
lines.extend(["Enter timeshift (+/-h:mm:ss or +/-m:ss),", "or filename for photo used as a reference:"])
|
|
elif editor == "time-clock":
|
|
assert reference is not None
|
|
lines.extend([f'File "{reference.path.name}" selected as reference.', "Enter the time that this file should have (h:mm:ss):"])
|
|
lines.append("> " + (dark_field(text.ljust(6)) if editor == "timezone" else text))
|
|
elif message:
|
|
lines.extend(["", message])
|
|
controls = ["Enter: confirm"]
|
|
if not editor:
|
|
controls.insert(0, "Arrow keys: navigate")
|
|
if not editor and selected in {"infer", "timezone", "artist", "organize"}:
|
|
controls.append("Space: toggle")
|
|
if editor:
|
|
controls.append("Esc: undo")
|
|
lines.extend(["", " ".join(controls)])
|
|
return lines
|
|
|
|
|
|
def prompt_choices(
|
|
executable: str,
|
|
records: list[MediaRecord],
|
|
working_dir: Path,
|
|
has_pto_files: bool,
|
|
initial_state: SettingsState | None = None,
|
|
) -> tuple[UserChoices | None, SettingsState]:
|
|
|
|
analysis = analyze_timezones(records)
|
|
first_zone = analysis.ordered_zones[0] if analysis.ordered_zones else local_timezone()
|
|
state = copy.deepcopy(initial_state) if initial_state else SettingsState(
|
|
time_offset=None,
|
|
infer_missing_timestamps=True,
|
|
fixed_timezone=first_zone,
|
|
fixed_timezone_enabled=False,
|
|
fill_timezone_gaps=analysis.gap_count > 0 and analysis.shifts <= 2 and not analysis.conflicting_gap_count,
|
|
process_pto_files=has_pto_files,
|
|
artist_value=default_artist(records),
|
|
)
|
|
rows = ["time"]
|
|
if any(record.original_time is None for record in records):
|
|
rows.append("infer")
|
|
rows.extend(["timezone", "rename", "artist", "organize", "confirm"])
|
|
row = 0
|
|
option = 1 if state.time_offset is not None else 0
|
|
editor = None
|
|
text = message = ""
|
|
reference = None
|
|
|
|
def enter_row(index: int) -> None:
|
|
nonlocal row, option
|
|
row = index % len(rows)
|
|
if rows[row] == "rename":
|
|
option = ("adjust_replace", "adjust_add", "replace", "add").index(state.rename_mode)
|
|
elif rows[row] == "time":
|
|
option = 1 if state.time_offset is not None else 0
|
|
else:
|
|
option = 0
|
|
|
|
def finish_editor() -> None:
|
|
nonlocal editor, text, message, reference
|
|
editor = None
|
|
text = message = ""
|
|
reference = None
|
|
|
|
with raw_key_input() as next_key:
|
|
while True:
|
|
draw_screen(settings_lines(state, analysis, working_dir, rows, row, option, editor=editor, text=text, message=message, reference=reference))
|
|
key = next_key()
|
|
if editor:
|
|
if key == "esc":
|
|
finish_editor()
|
|
elif key == "backspace":
|
|
text = text[:-1]
|
|
elif key == "enter":
|
|
if editor == "timezone":
|
|
try:
|
|
state.fixed_timezone = parse_timezone_offset(text)
|
|
except ValueError:
|
|
message = "Invalid timezone offset"
|
|
else:
|
|
state.fixed_timezone_enabled = True
|
|
finish_editor()
|
|
elif editor == "artist":
|
|
state.artist_value = text
|
|
state.artist_enabled = True
|
|
finish_editor()
|
|
elif editor == "group":
|
|
if text.isdigit() and int(text) >= 2:
|
|
state.group_min_size = int(text)
|
|
finish_editor()
|
|
else:
|
|
message = "Group size must be 2 or more"
|
|
elif editor == "time-source":
|
|
try:
|
|
state.time_offset = parse_timeshift(text)
|
|
except ValueError:
|
|
try:
|
|
reference = reference_record(executable, records, text)
|
|
except ValueError as exc:
|
|
message = str(exc)
|
|
else:
|
|
editor, text, message = "time-clock", "", ""
|
|
else:
|
|
finish_editor()
|
|
enter_row(row + 1)
|
|
else:
|
|
try:
|
|
assert reference is not None and reference.original_time is not None
|
|
state.time_offset = parse_clock_time(text, naive_wall_time(reference.original_time)) - naive_wall_time(reference.original_time)
|
|
except ValueError as exc:
|
|
message = str(exc)
|
|
else:
|
|
finish_editor()
|
|
enter_row(row + 1)
|
|
elif key == "space" and editor != "timezone":
|
|
text += " "
|
|
elif len(key) == 1 and (editor != "timezone" or key in "+-:0123456789"):
|
|
if editor != "timezone" or len(text) < 6:
|
|
text += key
|
|
continue
|
|
|
|
if key == "up":
|
|
enter_row(row - 1)
|
|
continue
|
|
current = rows[row]
|
|
if current == "confirm" and key == "enter":
|
|
if option == 1:
|
|
return None, state
|
|
apply_timezone_settings(records, state)
|
|
artist_action = "leave" if not state.artist_enabled else ("set" if state.artist_value else "clear")
|
|
return UserChoices(
|
|
working_dir=working_dir,
|
|
time_offset=state.time_offset,
|
|
fixed_timezone=state.fixed_timezone if state.fixed_timezone_enabled else None,
|
|
fill_timezone_gaps=state.fill_timezone_gaps,
|
|
artist_action=artist_action,
|
|
artist_value=state.artist_value or None,
|
|
rename_mode=state.rename_mode,
|
|
organize_files=state.organize_files,
|
|
group_min_size=state.group_min_size,
|
|
process_pto_files=state.organize_files and state.process_pto_files,
|
|
infer_missing_timestamps=state.infer_missing_timestamps,
|
|
), state
|
|
if key == "down" or (key == "enter" and current != "time"):
|
|
enter_row(row + 1)
|
|
continue
|
|
|
|
if current == "time":
|
|
if key == "left":
|
|
option = 0
|
|
state.time_offset = None
|
|
elif key == "right":
|
|
option = 1
|
|
if state.time_offset is None:
|
|
editor, text = "time-source", ""
|
|
elif key == "enter" and option == 1:
|
|
editor, text = "time-source", ""
|
|
elif key == "enter":
|
|
enter_row(row + 1)
|
|
elif current == "infer" and key in {"left", "right", "space"}:
|
|
state.infer_missing_timestamps = not state.infer_missing_timestamps
|
|
elif current == "timezone":
|
|
if key == "left":
|
|
option = max(0, option - 1)
|
|
elif key == "right":
|
|
option = min(1 if analysis.gap_count and not analysis.conflicting_gap_count else 0, option + 1)
|
|
elif key == "space":
|
|
if option == 0:
|
|
state.fixed_timezone_enabled = not state.fixed_timezone_enabled
|
|
elif analysis.gap_count and not analysis.conflicting_gap_count:
|
|
state.fill_timezone_gaps = not state.fill_timezone_gaps
|
|
elif option == 0 and key in "+-0123456789":
|
|
editor, text = "timezone", ("+" if key.isdigit() else "") + key
|
|
elif current == "rename":
|
|
if key == "left":
|
|
option = max(0, option - 1)
|
|
elif key == "right":
|
|
option = min(3, option + 1)
|
|
state.rename_mode = ("adjust_replace", "adjust_add", "replace", "add")[option]
|
|
elif current == "artist":
|
|
if key == "space":
|
|
state.artist_enabled = not state.artist_enabled
|
|
elif len(key) == 1:
|
|
editor, text = "artist", key
|
|
elif current == "organize":
|
|
if key == "left":
|
|
option = max(0, option - 1)
|
|
elif key == "right" and state.organize_files:
|
|
option = min(2, option + 1)
|
|
elif key == "space":
|
|
if option == 0:
|
|
state.organize_files = not state.organize_files
|
|
elif option == 2:
|
|
state.process_pto_files = not state.process_pto_files
|
|
elif option == 1 and key.isdigit():
|
|
editor, text = "group", key
|
|
elif current == "confirm" and key in {"left", "right"}:
|
|
option = 1 - option
|
|
|