US01-03: Discover and Reconcile Stable Assets (#48)
This commit was merged in pull request #48.
This commit is contained in:
1
photo_pipeline/services/__init__.py
Normal file
1
photo_pipeline/services/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""Application services — importable domain logic shared by API, worker, and CLI."""
|
||||
47
photo_pipeline/services/hashing.py
Normal file
47
photo_pipeline/services/hashing.py
Normal file
@@ -0,0 +1,47 @@
|
||||
"""Content hashing for identity and reconciliation.
|
||||
|
||||
- ``sha256_file`` is exact byte identity: it drives move/copy/replace matching.
|
||||
- ``pixel_sha256`` hashes the normalized decoded pixels (EXIF orientation applied,
|
||||
converted to RGB), so an EXIF-only edit keeps the same value while a genuine
|
||||
content change does not. That distinction separates "metadata changed" from
|
||||
"replaced".
|
||||
|
||||
Refactored from photo_analyzer._sha1_file/_phash_image: SHA-256 replaces SHA-1 as
|
||||
byte identity and a normalized pixel hash is added (see donor_ledger.yaml:
|
||||
pa-hashing). Perceptual hashing stays with the duplicate engine (US01-04).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
from pathlib import Path
|
||||
|
||||
from PIL import Image, ImageOps
|
||||
|
||||
PIXEL_HASH_VERSION = 1
|
||||
_CHUNK = 1 << 20
|
||||
|
||||
|
||||
def sha256_file(path: Path | str) -> str:
|
||||
digest = hashlib.sha256()
|
||||
with open(path, "rb") as handle:
|
||||
for chunk in iter(lambda: handle.read(_CHUNK), b""):
|
||||
digest.update(chunk)
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def pixel_sha256(path: Path | str) -> str:
|
||||
with Image.open(path) as image:
|
||||
oriented = ImageOps.exif_transpose(image)
|
||||
rgb = oriented.convert("RGB")
|
||||
header = f"{PIXEL_HASH_VERSION}:{rgb.width}x{rgb.height}:".encode()
|
||||
payload = header + rgb.tobytes()
|
||||
return hashlib.sha256(payload).hexdigest()
|
||||
|
||||
|
||||
def safe_pixel_sha256(path: Path | str) -> str | None:
|
||||
"""``pixel_sha256`` but returns None for undecodable images instead of raising."""
|
||||
try:
|
||||
return pixel_sha256(path)
|
||||
except Exception:
|
||||
return None
|
||||
197
photo_pipeline/services/inventory.py
Normal file
197
photo_pipeline/services/inventory.py
Normal file
@@ -0,0 +1,197 @@
|
||||
"""InventoryService — discovery and reconciliation against stable asset identity.
|
||||
|
||||
A scan discovers eligible files through the shared path policy, hashes each one,
|
||||
and reconciles them against the ``assets`` table so identity survives moves and
|
||||
renames. Every discovered or absent path is classified as one occurrence:
|
||||
|
||||
- ``new`` — a file with no matching identity;
|
||||
- ``moved`` — bytes match an asset whose old path is gone (id retained);
|
||||
- ``copied`` — bytes match an asset still present (distinct new id);
|
||||
- ``replaced`` — same path, different pixels;
|
||||
- ``metadata_changed`` — same path and pixels, different bytes (e.g. EXIF edit);
|
||||
- ``unchanged`` — same path and bytes;
|
||||
- ``missing`` — a known active asset whose file is gone (kept, flagged).
|
||||
|
||||
Missing files are never pruned (that would break identity); the asset is retained
|
||||
with ``missing_at`` set. Archived assets are left untouched. Rescanning unchanged
|
||||
input makes no durable change.
|
||||
|
||||
Extracted from photo_analyzer.discover_photos/reconcile_moved/prune_missing
|
||||
(see donor_ledger.yaml: pa-discovery, pa-prune-missing).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from collections import Counter
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime, timezone
|
||||
from enum import Enum
|
||||
from pathlib import Path
|
||||
from typing import Iterable
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session, sessionmaker
|
||||
|
||||
from photo_pipeline import path_policy
|
||||
from photo_pipeline.models import Asset, AssetPath
|
||||
from photo_pipeline.services import hashing
|
||||
|
||||
|
||||
class Occurrence(str, Enum):
|
||||
NEW = "new"
|
||||
MOVED = "moved"
|
||||
COPIED = "copied"
|
||||
REPLACED = "replaced"
|
||||
METADATA_CHANGED = "metadata_changed"
|
||||
UNCHANGED = "unchanged"
|
||||
MISSING = "missing"
|
||||
|
||||
|
||||
@dataclass
|
||||
class ScanResult:
|
||||
counts: dict[str, int] = field(default_factory=dict)
|
||||
occurrences: dict[str, str] = field(default_factory=dict) # path -> occurrence
|
||||
asset_ids: dict[str, str] = field(default_factory=dict) # path -> asset id
|
||||
|
||||
|
||||
class InventoryService:
|
||||
def __init__(self, session_factory: sessionmaker) -> None:
|
||||
self._session_factory = session_factory
|
||||
|
||||
def scan(self, roots: Iterable[Path | str] | Path | str) -> ScanResult:
|
||||
if isinstance(roots, (str, Path)):
|
||||
roots = [roots]
|
||||
normalized_roots = [path_policy.normalize_root(r) for r in roots]
|
||||
discovered = path_policy.discover(normalized_roots)
|
||||
discovered_paths = {str(p) for p in discovered}
|
||||
now = datetime.now(timezone.utc)
|
||||
|
||||
result = ScanResult()
|
||||
with self._session_factory() as session:
|
||||
assets = list(session.execute(select(Asset)).scalars())
|
||||
by_path = {a.current_path: a for a in assets if a.current_path}
|
||||
by_sha: dict[str, list[Asset]] = {}
|
||||
for asset in assets:
|
||||
if asset.current_sha256:
|
||||
by_sha.setdefault(asset.current_sha256, []).append(asset)
|
||||
|
||||
seen_ids: set[str] = set()
|
||||
for path in discovered:
|
||||
asset, occ = self._reconcile_one(
|
||||
session, path, discovered_paths, by_path, by_sha, now
|
||||
)
|
||||
seen_ids.add(asset.id)
|
||||
result.occurrences[str(path)] = occ.value
|
||||
result.asset_ids[str(path)] = asset.id
|
||||
|
||||
for asset in assets:
|
||||
if asset.availability_state != "active" or asset.id in seen_ids:
|
||||
continue
|
||||
if asset.current_path and asset.current_path not in discovered_paths:
|
||||
if not Path(asset.current_path).exists():
|
||||
if asset.missing_at is None:
|
||||
asset.missing_at = now
|
||||
asset.state_version += 1
|
||||
asset.updated_at = now
|
||||
result.occurrences[asset.current_path] = Occurrence.MISSING.value
|
||||
|
||||
session.commit()
|
||||
|
||||
result.counts = dict(Counter(result.occurrences.values()))
|
||||
return result
|
||||
|
||||
def _reconcile_one(
|
||||
self,
|
||||
session: Session,
|
||||
path: Path,
|
||||
discovered_paths: set[str],
|
||||
by_path: dict[str, Asset],
|
||||
by_sha: dict[str, list[Asset]],
|
||||
now: datetime,
|
||||
) -> tuple[Asset, Occurrence]:
|
||||
path_str = str(path)
|
||||
sha = hashing.sha256_file(path)
|
||||
pixel = hashing.safe_pixel_sha256(path)
|
||||
size = path.stat().st_size
|
||||
|
||||
existing = by_path.get(path_str)
|
||||
if existing is not None:
|
||||
if existing.current_sha256 == sha:
|
||||
occ = Occurrence.UNCHANGED
|
||||
else:
|
||||
same_pixels = pixel is not None and existing.pixel_sha256 == pixel
|
||||
occ = Occurrence.METADATA_CHANGED if same_pixels else Occurrence.REPLACED
|
||||
existing.current_sha256 = sha
|
||||
existing.pixel_sha256 = pixel
|
||||
existing.byte_size = size
|
||||
existing.state_version += 1
|
||||
existing.updated_at = now
|
||||
self._open_path(session, existing.id, path_str, now, occ.value)
|
||||
if existing.missing_at is not None:
|
||||
existing.missing_at = None
|
||||
existing.state_version += 1
|
||||
existing.updated_at = now
|
||||
return existing, occ
|
||||
|
||||
candidates = by_sha.get(sha, [])
|
||||
moved_from = next(
|
||||
(
|
||||
a
|
||||
for a in candidates
|
||||
if a.current_path
|
||||
and a.current_path not in discovered_paths
|
||||
and not Path(a.current_path).exists()
|
||||
),
|
||||
None,
|
||||
)
|
||||
if moved_from is not None:
|
||||
self._close_path(session, moved_from.id, moved_from.current_path, now)
|
||||
moved_from.current_path = path_str
|
||||
moved_from.byte_size = size
|
||||
moved_from.missing_at = None
|
||||
moved_from.state_version += 1
|
||||
moved_from.updated_at = now
|
||||
self._open_path(session, moved_from.id, path_str, now, Occurrence.MOVED.value)
|
||||
by_path[path_str] = moved_from
|
||||
return moved_from, Occurrence.MOVED
|
||||
|
||||
occ = Occurrence.COPIED if candidates else Occurrence.NEW
|
||||
asset = Asset(
|
||||
id=str(uuid.uuid4()),
|
||||
original_path=path_str,
|
||||
current_path=path_str,
|
||||
current_sha256=sha,
|
||||
pixel_sha256=pixel,
|
||||
hash_version=hashing.PIXEL_HASH_VERSION,
|
||||
byte_size=size,
|
||||
discovered_at=now,
|
||||
availability_state="active",
|
||||
state_version=1,
|
||||
)
|
||||
session.add(asset)
|
||||
session.flush()
|
||||
self._open_path(session, asset.id, path_str, now, occ.value)
|
||||
by_sha.setdefault(sha, []).append(asset)
|
||||
by_path[path_str] = asset
|
||||
return asset, occ
|
||||
|
||||
@staticmethod
|
||||
def _open_path(session: Session, asset_id: str, path: str, now: datetime, reason: str) -> None:
|
||||
session.add(
|
||||
AssetPath(asset_id=asset_id, path=path, valid_from=now, reason=reason)
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _close_path(session: Session, asset_id: str, path: str | None, now: datetime) -> None:
|
||||
if not path:
|
||||
return
|
||||
rows = session.execute(
|
||||
select(AssetPath).where(
|
||||
AssetPath.asset_id == asset_id,
|
||||
AssetPath.path == path,
|
||||
AssetPath.valid_until.is_(None),
|
||||
)
|
||||
).scalars()
|
||||
for row in rows:
|
||||
row.valid_until = now
|
||||
Reference in New Issue
Block a user