US04-01: Build and Export Rename Plans (#66)
This commit was merged in pull request #66.
This commit is contained in:
426
photo_pipeline/services/renames.py
Normal file
426
photo_pipeline/services/renames.py
Normal file
@@ -0,0 +1,426 @@
|
||||
"""RenameService — build, validate, and export guarded rename plans (US04-01).
|
||||
|
||||
A plan turns approved album proposals into an itemized, reviewable preview: for every
|
||||
album, the source folder, the destination folder, the assets that move with it, their
|
||||
expected hashes, and every validation issue found. **Nothing here touches the
|
||||
filesystem** — it only reads to validate. Applying a plan is US04-03.
|
||||
|
||||
Validation is the point of the story, so every refusal is a structured issue rather
|
||||
than an exception (concept §7 "guarded rename plan"):
|
||||
|
||||
- ``source_missing`` — the registered folder is gone
|
||||
- ``source_changed`` — an asset's bytes changed since inventory
|
||||
- ``duplicate_target`` — two operations claim the same destination
|
||||
- ``destination_exists`` — an unexpected path already occupies the destination
|
||||
- ``case_only`` — informational: needs the staged procedure on
|
||||
case-insensitive filesystems (not a blocker)
|
||||
- ``unicode_collision`` — two destinations differ only by Unicode normalization
|
||||
- ``root_escape`` — a path resolves outside the configured library
|
||||
- ``excluded_path`` — a path lies under ``_IGNORE/``
|
||||
- ``symlink`` — source or destination is a symlink
|
||||
- ``cross_filesystem`` — destination is on another device (copy-verify-delete)
|
||||
|
||||
A plan with any blocking issue is ``invalid`` and can never be applied. Issues that
|
||||
are purely informational (``case_only``, ``cross_filesystem``) are recorded on the
|
||||
operation so the apply stage picks the right procedure.
|
||||
|
||||
The JSON export is portable evidence: schema version, checksum, and the full operation
|
||||
list, with no credentials and no absolute host paths beyond the library itself.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import unicodedata
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from photo_pipeline import path_policy
|
||||
from photo_pipeline.models import AlbumProposal, Asset, RenameOperation, RenamePlan
|
||||
from photo_pipeline.services.albums import album_label
|
||||
|
||||
EXPORT_SCHEMA_VERSION = 1
|
||||
MOVE_FOLDER = "move_folder"
|
||||
|
||||
# Issues that make a plan unapplyable. Everything else is advisory.
|
||||
BLOCKING_CODES = frozenset(
|
||||
{
|
||||
"source_missing",
|
||||
"source_changed",
|
||||
"duplicate_target",
|
||||
"destination_exists",
|
||||
"unicode_collision",
|
||||
"root_escape",
|
||||
"excluded_path",
|
||||
"symlink",
|
||||
"no_library_root",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
class RenameError(RuntimeError):
|
||||
"""Invalid request against a plan (unknown id, nothing to plan)."""
|
||||
|
||||
|
||||
def _now() -> datetime:
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
def _issue(code: str, message: str) -> dict:
|
||||
return {"code": code, "message": message}
|
||||
|
||||
|
||||
def _nfc(value: str) -> str:
|
||||
return unicodedata.normalize("NFC", value)
|
||||
|
||||
|
||||
def plan_checksum(operations: list[dict]) -> str:
|
||||
"""Content hash over the operations that define a plan. Two plans with the same
|
||||
moves have the same checksum, so a stale confirmation is detectable."""
|
||||
payload = json.dumps(
|
||||
[
|
||||
{
|
||||
"sequence": op["sequence"],
|
||||
"source_path": op["source_path"],
|
||||
"destination_path": op["destination_path"],
|
||||
"asset_ids": sorted(op["asset_ids"]),
|
||||
}
|
||||
for op in sorted(operations, key=lambda op: op["sequence"])
|
||||
],
|
||||
ensure_ascii=False,
|
||||
sort_keys=True,
|
||||
)
|
||||
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
class RenameService:
|
||||
def __init__(self, session_factory: sessionmaker, *, library_roots: tuple = ()) -> None:
|
||||
self._session_factory = session_factory
|
||||
self._roots = tuple(Path(root) for root in library_roots)
|
||||
|
||||
# ── planning ──────────────────────────────────────────────────────────────
|
||||
|
||||
def build_plan(self) -> dict:
|
||||
"""Build a plan from every approved album proposal. Read-only."""
|
||||
with self._session_factory() as session:
|
||||
approved = list(
|
||||
session.scalars(
|
||||
select(AlbumProposal)
|
||||
.where(AlbumProposal.status == "approved")
|
||||
.order_by(AlbumProposal.album)
|
||||
)
|
||||
)
|
||||
if not approved:
|
||||
raise RenameError("no approved album proposals to plan")
|
||||
assets = list(
|
||||
session.scalars(
|
||||
select(Asset).where(
|
||||
Asset.canonical_asset_id.is_(None),
|
||||
Asset.availability_state == "active",
|
||||
Asset.current_path.is_not(None),
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
by_album: dict[str, list[Asset]] = {}
|
||||
for asset in assets:
|
||||
by_album.setdefault(album_label(asset.current_path, self._roots), []).append(asset)
|
||||
|
||||
operations: list[dict] = []
|
||||
for sequence, proposal in enumerate(approved):
|
||||
members = sorted(by_album.get(proposal.album, []), key=lambda a: a.current_path)
|
||||
operations.append(self._operation(sequence, proposal, members))
|
||||
|
||||
self._detect_cross_operation_issues(operations)
|
||||
return self._persist(operations)
|
||||
|
||||
def _operation(self, sequence: int, proposal: AlbumProposal, members: list[Asset]) -> dict:
|
||||
final_name = proposal.final_name or proposal.proposed_name or ""
|
||||
issues: list[dict] = []
|
||||
|
||||
source = self._source_folder(members)
|
||||
destination = source.parent / final_name if source else None
|
||||
|
||||
operation = {
|
||||
"sequence": sequence,
|
||||
"operation": MOVE_FOLDER,
|
||||
"album": proposal.album,
|
||||
"source_path": str(source) if source else "",
|
||||
"destination_path": str(destination) if destination else "",
|
||||
"asset_ids": [asset.id for asset in members],
|
||||
"expected_sha256": {
|
||||
asset.id: asset.current_sha256 for asset in members if asset.current_sha256
|
||||
},
|
||||
"asset_count": len(members),
|
||||
"same_filesystem": None,
|
||||
"case_only": False,
|
||||
"issues": issues,
|
||||
}
|
||||
|
||||
if source is None:
|
||||
issues.append(
|
||||
_issue("source_missing", f"no active assets remain in {proposal.album!r}")
|
||||
)
|
||||
return operation
|
||||
if not final_name:
|
||||
issues.append(_issue("empty_name", f"proposal for {proposal.album!r} has no name"))
|
||||
return operation
|
||||
|
||||
self._validate_paths(operation, source, destination, issues)
|
||||
self._validate_sources(operation, members, issues)
|
||||
return operation
|
||||
|
||||
def _source_folder(self, members: list[Asset]) -> Path | None:
|
||||
if not members:
|
||||
return None
|
||||
return Path(members[0].current_path).parent
|
||||
|
||||
def _validate_paths(
|
||||
self, operation: dict, source: Path, destination: Path, issues: list[dict]
|
||||
) -> None:
|
||||
if not self._roots:
|
||||
issues.append(_issue("no_library_root", "no library root is configured"))
|
||||
return
|
||||
|
||||
for label, path in (("source", source), ("destination", destination)):
|
||||
if path_policy.is_excluded(path):
|
||||
issues.append(_issue("excluded_path", f"{label} is under an excluded directory"))
|
||||
root = self._root_for(path)
|
||||
if root is None:
|
||||
issues.append(
|
||||
_issue("root_escape", f"{label} {path} is outside every library root")
|
||||
)
|
||||
|
||||
if source.is_symlink() or destination.is_symlink():
|
||||
issues.append(_issue("symlink", "refusing to rename through a symlink"))
|
||||
|
||||
if not source.exists():
|
||||
issues.append(_issue("source_missing", f"source folder {source} does not exist"))
|
||||
return
|
||||
|
||||
# A symlinked source could still resolve outside the root even when its
|
||||
# literal path looks fine; check the resolved location too.
|
||||
root = self._root_for(source)
|
||||
if root is not None:
|
||||
try:
|
||||
path_policy.resolve_within(root, source)
|
||||
except path_policy.PathPolicyError as error:
|
||||
issues.append(_issue("root_escape", str(error)))
|
||||
|
||||
operation["case_only"] = (
|
||||
source != destination and source.as_posix().lower() == destination.as_posix().lower()
|
||||
)
|
||||
operation["same_filesystem"] = self._same_filesystem(source, destination)
|
||||
if operation["same_filesystem"] is False:
|
||||
issues.append(
|
||||
_issue(
|
||||
"cross_filesystem", "destination is on another filesystem; copy-verify-delete"
|
||||
)
|
||||
)
|
||||
|
||||
# A destination that already exists is only acceptable for a case-only
|
||||
# rename, where source and destination are the same directory entry.
|
||||
if destination.exists() and not operation["case_only"]:
|
||||
issues.append(_issue("destination_exists", f"destination {destination} already exists"))
|
||||
|
||||
def _validate_sources(self, operation: dict, members: list[Asset], issues: list[dict]) -> None:
|
||||
"""Every asset must still be present with the bytes inventory recorded."""
|
||||
for asset in members:
|
||||
path = Path(asset.current_path)
|
||||
if not path.exists():
|
||||
issues.append(_issue("source_missing", f"asset file {path} is missing"))
|
||||
continue
|
||||
expected = asset.byte_size
|
||||
if expected is not None and path.stat().st_size != expected:
|
||||
issues.append(
|
||||
_issue("source_changed", f"asset file {path} changed since inventory")
|
||||
)
|
||||
|
||||
def _root_for(self, path: Path) -> Path | None:
|
||||
for root in self._roots:
|
||||
resolved_root = Path(root)
|
||||
if path == resolved_root or resolved_root in path.parents:
|
||||
return resolved_root
|
||||
return None
|
||||
|
||||
def _same_filesystem(self, source: Path, destination: Path) -> bool | None:
|
||||
anchor = destination.parent
|
||||
if not source.exists() or not anchor.exists():
|
||||
return None
|
||||
try:
|
||||
return os.stat(source).st_dev == os.stat(anchor).st_dev
|
||||
except OSError:
|
||||
return None
|
||||
|
||||
def _detect_cross_operation_issues(self, operations: list[dict]) -> None:
|
||||
"""Two operations must never target the same destination — including when the
|
||||
names differ only by case or by Unicode normalization."""
|
||||
exact: dict[str, int] = {}
|
||||
folded: dict[str, int] = {}
|
||||
for operation in operations:
|
||||
destination = operation["destination_path"]
|
||||
if not destination:
|
||||
continue
|
||||
if destination in exact:
|
||||
operation["issues"].append(
|
||||
_issue("duplicate_target", f"destination {destination} is claimed twice")
|
||||
)
|
||||
else:
|
||||
exact[destination] = operation["sequence"]
|
||||
|
||||
key = _nfc(destination).casefold()
|
||||
if key in folded and folded[key] != operation["sequence"]:
|
||||
operation["issues"].append(
|
||||
_issue(
|
||||
"unicode_collision",
|
||||
f"destination {destination} collides with another target after "
|
||||
"case/Unicode normalization",
|
||||
)
|
||||
)
|
||||
else:
|
||||
folded.setdefault(key, operation["sequence"])
|
||||
|
||||
def _persist(self, operations: list[dict]) -> dict:
|
||||
blockers = sorted(
|
||||
{
|
||||
issue["code"]
|
||||
for operation in operations
|
||||
for issue in operation["issues"]
|
||||
if issue["code"] in BLOCKING_CODES
|
||||
}
|
||||
)
|
||||
now = _now()
|
||||
plan_id = str(uuid.uuid4())
|
||||
with self._session_factory() as session:
|
||||
session.add(
|
||||
RenamePlan(
|
||||
id=plan_id,
|
||||
state="invalid" if blockers else "validated",
|
||||
schema_version=EXPORT_SCHEMA_VERSION,
|
||||
checksum=plan_checksum(operations),
|
||||
blockers=json.dumps(blockers),
|
||||
operation_count=len(operations),
|
||||
version=1,
|
||||
validated_at=now,
|
||||
)
|
||||
)
|
||||
# Flush the parent before its operations so the foreign key resolves.
|
||||
session.flush()
|
||||
for operation in operations:
|
||||
session.add(
|
||||
RenameOperation(
|
||||
id=str(uuid.uuid4()),
|
||||
plan_id=plan_id,
|
||||
sequence=operation["sequence"],
|
||||
operation=operation["operation"],
|
||||
album=operation["album"],
|
||||
source_path=operation["source_path"],
|
||||
destination_path=operation["destination_path"],
|
||||
asset_ids=json.dumps(operation["asset_ids"]),
|
||||
expected_sha256=json.dumps(operation["expected_sha256"]),
|
||||
asset_count=operation["asset_count"],
|
||||
same_filesystem=operation["same_filesystem"],
|
||||
case_only=operation["case_only"],
|
||||
issues=json.dumps(operation["issues"], ensure_ascii=False),
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
return self.get(plan_id)
|
||||
|
||||
# ── reads ─────────────────────────────────────────────────────────────────
|
||||
|
||||
def get(self, plan_id: str) -> dict | None:
|
||||
with self._session_factory() as session:
|
||||
plan = session.get(RenamePlan, plan_id)
|
||||
if plan is None:
|
||||
return None
|
||||
operations = list(
|
||||
session.scalars(
|
||||
select(RenameOperation)
|
||||
.where(RenameOperation.plan_id == plan_id)
|
||||
.order_by(RenameOperation.sequence)
|
||||
)
|
||||
)
|
||||
return _plan_dict(plan, operations)
|
||||
|
||||
def list_plans(self) -> dict:
|
||||
with self._session_factory() as session:
|
||||
plans = list(session.scalars(select(RenamePlan).order_by(RenamePlan.created_at.desc())))
|
||||
return {
|
||||
"items": [
|
||||
{
|
||||
"id": plan.id,
|
||||
"state": plan.state,
|
||||
"operation_count": plan.operation_count,
|
||||
"blockers": json.loads(plan.blockers or "[]"),
|
||||
"checksum": plan.checksum,
|
||||
"version": plan.version,
|
||||
}
|
||||
for plan in plans
|
||||
],
|
||||
"total": len(plans),
|
||||
}
|
||||
|
||||
def export(self, plan_id: str) -> dict:
|
||||
"""Portable JSON evidence for a plan. Contains no credentials."""
|
||||
plan = self.get(plan_id)
|
||||
if plan is None:
|
||||
raise RenameError(f"unknown rename plan {plan_id!r}")
|
||||
return {
|
||||
"schema_version": plan["schema_version"],
|
||||
"plan_id": plan["id"],
|
||||
"state": plan["state"],
|
||||
"checksum": plan["checksum"],
|
||||
"blockers": plan["blockers"],
|
||||
"operations": [
|
||||
{
|
||||
"sequence": operation["sequence"],
|
||||
"operation": operation["operation"],
|
||||
"album": operation["album"],
|
||||
"source_path": operation["source_path"],
|
||||
"destination_path": operation["destination_path"],
|
||||
"asset_ids": operation["asset_ids"],
|
||||
"asset_count": operation["asset_count"],
|
||||
"case_only": operation["case_only"],
|
||||
"same_filesystem": operation["same_filesystem"],
|
||||
"issues": operation["issues"],
|
||||
}
|
||||
for operation in plan["operations"]
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _plan_dict(plan: RenamePlan, operations: list[RenameOperation]) -> dict:
|
||||
return {
|
||||
"id": plan.id,
|
||||
"state": plan.state,
|
||||
"schema_version": plan.schema_version,
|
||||
"checksum": plan.checksum,
|
||||
"blockers": json.loads(plan.blockers or "[]"),
|
||||
"operation_count": plan.operation_count,
|
||||
"version": plan.version,
|
||||
"applicable": plan.state == "validated",
|
||||
"operations": [
|
||||
{
|
||||
"sequence": operation.sequence,
|
||||
"operation": operation.operation,
|
||||
"album": operation.album,
|
||||
"source_path": operation.source_path,
|
||||
"destination_path": operation.destination_path,
|
||||
"asset_ids": json.loads(operation.asset_ids or "[]"),
|
||||
"expected_sha256": json.loads(operation.expected_sha256 or "{}"),
|
||||
"asset_count": operation.asset_count,
|
||||
"same_filesystem": operation.same_filesystem,
|
||||
"case_only": operation.case_only,
|
||||
"issues": json.loads(operation.issues or "[]"),
|
||||
"journal_state": operation.journal_state,
|
||||
}
|
||||
for operation in operations
|
||||
],
|
||||
}
|
||||
Reference in New Issue
Block a user