Improve metadata preview and skip no-op writes
This commit is contained in:
+197
-36
@@ -96,6 +96,15 @@ class XmlSidecar:
|
|||||||
device: str | None = None
|
device: str | None = None
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class TagWrite:
|
||||||
|
label: str
|
||||||
|
arg: str
|
||||||
|
current_value: str | None
|
||||||
|
new_value: str
|
||||||
|
will_write: bool
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class MediaRecord:
|
class MediaRecord:
|
||||||
path: Path
|
path: Path
|
||||||
@@ -119,6 +128,7 @@ class MediaRecord:
|
|||||||
target_path: Path | None = None
|
target_path: Path | None = None
|
||||||
target_sidecar: Path | None = None
|
target_sidecar: Path | None = None
|
||||||
rename_skip_reason: str | None = None
|
rename_skip_reason: str | None = None
|
||||||
|
write_plan: list[TagWrite] = field(default_factory=list)
|
||||||
warnings: list[str] = field(default_factory=list)
|
warnings: list[str] = field(default_factory=list)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
@@ -441,6 +451,55 @@ def first_tag(metadata: dict, tags: Iterable[str]) -> object | None:
|
|||||||
return None
|
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:
|
def find_sidecar(video_path: Path) -> XmlSidecar | None:
|
||||||
expected = f"{video_path.stem}M01.XML"
|
expected = f"{video_path.stem}M01.XML"
|
||||||
for child in video_path.parent.iterdir():
|
for child in video_path.parent.iterdir():
|
||||||
@@ -1044,18 +1103,57 @@ def plan_targets(records: list[MediaRecord], choices: UserChoices) -> None:
|
|||||||
record.target_sidecar = target_sidecar
|
record.target_sidecar = target_sidecar
|
||||||
|
|
||||||
|
|
||||||
def build_write_args(record: MediaRecord, choices: UserChoices) -> list[str]:
|
def build_write_plan(record: MediaRecord, choices: UserChoices) -> list[TagWrite]:
|
||||||
args: list[str] = []
|
writes: list[TagWrite] = []
|
||||||
if choices.artist_action == "set":
|
if choices.artist_action == "set":
|
||||||
assert choices.artist_value is not None
|
assert choices.artist_value is not None
|
||||||
args.extend([f"-Artist={choices.artist_value}", f"-Author={choices.artist_value}"])
|
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":
|
elif choices.artist_action == "clear":
|
||||||
args.extend(["-Artist=", "-Author="])
|
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 record.is_image:
|
||||||
if choices.time_offset is not None or record.parsed_from_filename:
|
if choices.time_offset is not None or record.parsed_from_filename:
|
||||||
if record.adjusted_time is not None:
|
if record.adjusted_time is not None:
|
||||||
args.append(f"-AllDates={record.adjusted_time.strftime('%Y:%m:%d %H:%M:%S')}")
|
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
|
should_write_tz = False
|
||||||
tz_to_write = choices.explicit_timezone
|
tz_to_write = choices.explicit_timezone
|
||||||
@@ -1070,39 +1168,84 @@ def build_write_args(record: MediaRecord, choices: UserChoices) -> list[str]:
|
|||||||
should_write_tz = True
|
should_write_tz = True
|
||||||
if should_write_tz and tz_to_write is not None:
|
if should_write_tz and tz_to_write is not None:
|
||||||
tz_text = timezone_to_string(tz_to_write)
|
tz_text = timezone_to_string(tz_to_write)
|
||||||
args.extend(
|
for label, arg_name, current_tag in (
|
||||||
[
|
("OffsetTimeOriginal", "EXIF:OffsetTimeOriginal", "EXIF:OffsetTimeOriginal"),
|
||||||
f"-EXIF:OffsetTimeOriginal={tz_text}",
|
("OffsetTimeDigitized", "EXIF:OffsetTimeDigitized", "EXIF:OffsetTimeDigitized"),
|
||||||
f"-EXIF:OffsetTimeDigitized={tz_text}",
|
("OffsetTime", "EXIF:OffsetTime", "EXIF:OffsetTime"),
|
||||||
f"-EXIF:OffsetTime={tz_text}",
|
):
|
||||||
]
|
add_tag_write(
|
||||||
)
|
writes,
|
||||||
|
label,
|
||||||
|
arg_name,
|
||||||
|
tag_value(record.metadata, current_tag),
|
||||||
|
tz_text,
|
||||||
|
)
|
||||||
|
|
||||||
if record.inferred_subsec is not None:
|
if record.inferred_subsec is not None:
|
||||||
args.extend(
|
subsec_text = f"{record.inferred_subsec:03d}"
|
||||||
[
|
for label, arg_name, current_tag in (
|
||||||
f"-SubSecTimeOriginal={record.inferred_subsec:03d}",
|
("SubSecTimeOriginal", "EXIF:SubSecTimeOriginal", "EXIF:SubSecTimeOriginal"),
|
||||||
f"-SubSecTimeDigitized={record.inferred_subsec:03d}",
|
("SubSecTimeDigitized", "EXIF:SubSecTimeDigitized", "EXIF:SubSecTimeDigitized"),
|
||||||
f"-SubSecTime={record.inferred_subsec:03d}",
|
("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:
|
elif record.is_video and choices.time_offset is not None and record.adjusted_time is not None:
|
||||||
utc_time = record.adjusted_time
|
utc_time = record.adjusted_time
|
||||||
if utc_time.tzinfo is None:
|
if utc_time.tzinfo is None:
|
||||||
utc_time = utc_time.replace(tzinfo=timezone.utc)
|
utc_time = utc_time.replace(tzinfo=timezone.utc)
|
||||||
utc_text = utc_time.astimezone(timezone.utc).strftime("%Y:%m:%d %H:%M:%S")
|
utc_text = utc_time.astimezone(timezone.utc).strftime("%Y:%m:%d %H:%M:%S")
|
||||||
args.extend(
|
for label, arg_name, current_tag in (
|
||||||
[
|
("QuickTime CreateDate", "QuickTime:CreateDate", "QuickTime:CreateDate"),
|
||||||
f"-QuickTime:CreateDate={utc_text}",
|
("QuickTime ModifyDate", "QuickTime:ModifyDate", "QuickTime:ModifyDate"),
|
||||||
f"-QuickTime:ModifyDate={utc_text}",
|
("QuickTime TrackCreateDate", "QuickTime:TrackCreateDate", "QuickTime:TrackCreateDate"),
|
||||||
f"-QuickTime:TrackCreateDate={utc_text}",
|
("QuickTime TrackModifyDate", "QuickTime:TrackModifyDate", "QuickTime:TrackModifyDate"),
|
||||||
f"-QuickTime:TrackModifyDate={utc_text}",
|
("QuickTime MediaCreateDate", "QuickTime:MediaCreateDate", "QuickTime:MediaCreateDate"),
|
||||||
f"-QuickTime:MediaCreateDate={utc_text}",
|
("QuickTime MediaModifyDate", "QuickTime:MediaModifyDate", "QuickTime:MediaModifyDate"),
|
||||||
f"-QuickTime:MediaModifyDate={utc_text}",
|
):
|
||||||
]
|
add_tag_write(
|
||||||
)
|
writes,
|
||||||
|
label,
|
||||||
|
arg_name,
|
||||||
|
tag_value(record.metadata, current_tag),
|
||||||
|
utc_text,
|
||||||
|
same_datetime_value,
|
||||||
|
)
|
||||||
|
|
||||||
return args
|
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 display_path_from_working_dir(path: Path, working_dir: Path) -> str:
|
||||||
|
try:
|
||||||
|
relative = path.resolve().relative_to(working_dir.resolve())
|
||||||
|
except ValueError:
|
||||||
|
return str(path)
|
||||||
|
return "." if str(relative) == "." else str(relative)
|
||||||
|
|
||||||
|
|
||||||
|
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 metadata_preview_text(write: TagWrite) -> str:
|
||||||
|
current = write.current_value if write.current_value is not None else "<missing>"
|
||||||
|
if write.will_write:
|
||||||
|
return f"metadata {write.label}: {current} -> {write.new_value}"
|
||||||
|
return f"metadata {write.label}: {current} -> {write.new_value} (unchanged, skipped)"
|
||||||
|
|
||||||
|
|
||||||
def print_preview(records: list[MediaRecord], choices: UserChoices) -> None:
|
def print_preview(records: list[MediaRecord], choices: UserChoices) -> None:
|
||||||
@@ -1137,17 +1280,33 @@ def print_preview(records: list[MediaRecord], choices: UserChoices) -> None:
|
|||||||
for record in sorted(records, key=lambda item: str(item.path).lower()):
|
for record in sorted(records, key=lambda item: str(item.path).lower()):
|
||||||
target = record.target_path or record.path
|
target = record.target_path or record.path
|
||||||
operation = []
|
operation = []
|
||||||
if target != record.path:
|
source_parent = record.path.parent.resolve()
|
||||||
operation.append(f"rename/move -> {target}")
|
target_parent = target.parent.resolve()
|
||||||
if record.inferred_subsec is not None:
|
source_name = record.path.name
|
||||||
operation.append(f"subsec {record.inferred_subsec:03d}")
|
target_name = target.name
|
||||||
|
is_moving = source_parent != target_parent
|
||||||
|
is_renaming = source_name != target_name
|
||||||
|
if is_moving:
|
||||||
|
source_display = path_display(record.path, choices, use_relative=True)
|
||||||
|
if is_renaming:
|
||||||
|
target_display = path_display(target, choices, use_relative=True)
|
||||||
|
else:
|
||||||
|
target_display = path_display(target.parent, choices, use_relative=True)
|
||||||
|
operation.append(f"move {source_display} -> {target_display}")
|
||||||
|
if is_renaming:
|
||||||
|
operation.append(f"rename {source_name} -> {target_name}")
|
||||||
|
for write in record.write_plan:
|
||||||
|
operation.append(metadata_preview_text(write))
|
||||||
if record.rename_skip_reason:
|
if record.rename_skip_reason:
|
||||||
operation.append(f"rename skipped: {record.rename_skip_reason}")
|
operation.append(f"rename skipped: {record.rename_skip_reason}")
|
||||||
for warning in record.warnings:
|
for warning in record.warnings:
|
||||||
operation.append(f"warning: {warning}")
|
operation.append(f"warning: {warning}")
|
||||||
if record.target_sidecar and record.sidecar:
|
if record.target_sidecar and record.sidecar:
|
||||||
operation.append(f"sidecar -> {record.target_sidecar}")
|
sidecar_source = path_display(record.sidecar.path, choices, use_relative=True)
|
||||||
rows.append((str(record.path), "; ".join(operation) or "metadata only/no change"))
|
sidecar_target = path_display(record.target_sidecar, choices, use_relative=True)
|
||||||
|
operation.append(f"sidecar move {sidecar_source} -> {sidecar_target}")
|
||||||
|
source_text = path_display(record.path, choices, use_relative=is_moving)
|
||||||
|
rows.append((source_text, "; ".join(operation) or "no change"))
|
||||||
|
|
||||||
path_width = min(max((len(row[0]) for row in rows), default=10), 70)
|
path_width = min(max((len(row[0]) for row in rows), default=10), 70)
|
||||||
for source, operation in rows:
|
for source, operation in rows:
|
||||||
@@ -1240,6 +1399,8 @@ def prepare_plan(records: list[MediaRecord], choices: UserChoices) -> list[list[
|
|||||||
if groups:
|
if groups:
|
||||||
assign_group_names(groups, choices.working_dir)
|
assign_group_names(groups, choices.working_dir)
|
||||||
plan_targets(records, choices)
|
plan_targets(records, choices)
|
||||||
|
for record in records:
|
||||||
|
record.write_plan = build_write_plan(record, choices)
|
||||||
return groups
|
return groups
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user