75 lines
3.1 KiB
Python
75 lines
3.1 KiB
Python
"""Persistent per-step copy flags shared by repair commands and the GUI."""
|
|
from pathlib import Path
|
|
|
|
from copienator import CliError, EvaluationWorkspace, atomic_update_json, read_json
|
|
|
|
|
|
def _path(workspace: EvaluationWorkspace) -> Path:
|
|
return workspace.metadata_dir / "copy_errors.json"
|
|
|
|
|
|
def _validate(value) -> dict[str, dict[str, str]]:
|
|
if not isinstance(value, dict):
|
|
raise CliError("Invalid copy_errors.json: expected PDF filenames and step flags")
|
|
normalized = {}
|
|
for name, flags in value.items():
|
|
if (not isinstance(name, str) or "/" in name or "\\" in name
|
|
or Path(name).suffix.casefold() != ".pdf"):
|
|
raise CliError("Invalid copy_errors.json: invalid PDF filename")
|
|
# Older versions only reported errors from margin cropping.
|
|
if isinstance(flags, str):
|
|
flags = {"cutleft": flags}
|
|
if (not isinstance(flags, dict) or not flags
|
|
or any(step not in {"cutleft", "plotting"} or not isinstance(reason, str)
|
|
for step, reason in flags.items())):
|
|
raise CliError("Invalid copy_errors.json: invalid step flags")
|
|
normalized[name] = dict(flags)
|
|
return normalized
|
|
|
|
|
|
def copy_errors(workspace: EvaluationWorkspace) -> dict[str, str]:
|
|
"""Return a summary per copy, preserving the existing GUI-facing API."""
|
|
flags = _validate(read_json(_path(workspace), default={}))
|
|
return {name: "; ".join(reasons.values()) for name, reasons in flags.items()}
|
|
|
|
|
|
def mark_copy_error(workspace: EvaluationWorkspace, pdf: Path, reason: str,
|
|
*, source: str = "cutleft") -> None:
|
|
if source not in {"cutleft", "plotting"}:
|
|
raise ValueError(f"Unknown copy flag source: {source}")
|
|
def update(errors):
|
|
normalized = _validate(errors)
|
|
normalized.setdefault(pdf.name, {})[source] = reason
|
|
errors.clear()
|
|
errors.update(normalized)
|
|
atomic_update_json(_path(workspace), update, default_factory=dict)
|
|
|
|
|
|
def clear_copy_error(workspace: EvaluationWorkspace, pdf: Path,
|
|
*, source: str = "cutleft") -> None:
|
|
if source not in {"cutleft", "plotting"}:
|
|
raise ValueError(f"Unknown copy flag source: {source}")
|
|
if not _path(workspace).exists():
|
|
return
|
|
def update(errors):
|
|
normalized = _validate(errors)
|
|
flags = normalized.get(pdf.name, {})
|
|
flags.pop(source, None)
|
|
if not flags:
|
|
normalized.pop(pdf.name, None)
|
|
errors.clear()
|
|
errors.update(normalized)
|
|
atomic_update_json(_path(workspace), update, default_factory=dict)
|
|
|
|
|
|
def marked_copy_paths(workspace: EvaluationWorkspace, *, originals: bool = False) -> list[Path]:
|
|
directories = ([workspace.original_copies_dir, workspace.copies_dir, workspace.root]
|
|
if originals else [workspace.copies_dir])
|
|
result = []
|
|
for name in sorted(copy_errors(workspace), key=str.casefold):
|
|
path = next((directory / name for directory in directories if (directory / name).is_file()), None)
|
|
if path is None:
|
|
raise CliError(f"Marked copy not found: {name}")
|
|
result.append(path)
|
|
return result
|