US03-01: Aggregate Album Evidence #61

Merged
domverse merged 1 commits from us/US03-01-aggregate-album-evidence into main 2026-08-15 13:10:32 +02:00
6 changed files with 635 additions and 3 deletions

View File

@@ -336,11 +336,16 @@ rows:
Album = leaf folder relative to library, '(root)' at root, bare parent
outside the library. The single album-identity rule for stats, proposals
and folder-as-album upload. webapp.query.album_of is a copy — consolidate
to one implementation (see wa-album-of).
to one implementation (see wa-album-of). Extracted in US03-01 as
photo_pipeline.services.albums.album_label (parity test below); the album
evidence aggregator uses it. LibraryService._album_of still keeps its own
leaf-name grouping — fold it into this rule when the Library view is next
touched.
target: photo_pipeline/services/albums.py
tests:
- test_pa_album::test_album_label_leaf_folder_relative
- test_pa_album::test_webapp_album_of_mirrors_album_label
- test_pa_album::test_albums_service_album_label_matches_donor
status: characterized
- id: pa-config
@@ -546,9 +551,12 @@ rows:
classification: replace
rationale: >
Verbatim copy of photo_analyzer.album_label (equivalence characterized);
consolidated into the single albums-service implementation (pa-album-label).
consolidated into the single albums-service implementation (pa-album-label),
now photo_pipeline.services.albums.album_label (US03-01, parity test below).
target: photo_pipeline/services/albums.py
tests: [test_pa_album::test_webapp_album_of_mirrors_album_label]
tests:
- test_pa_album::test_webapp_album_of_mirrors_album_label
- test_pa_album::test_albums_service_album_label_matches_donor
status: characterized
- id: wa-allowlist

View File

@@ -0,0 +1,265 @@
"""AlbumService — aggregate per-folder evidence for album proposals (US03-01).
Read-only. For each leaf folder it summarises the *already stored* analysis and
safety evidence of the canonical, active assets living there, so a later stage can
name the folder without resending images or exposing raw model responses. It writes
nothing and generates no names (that is US03-02/03).
Folder identity is the full parent path of an asset's ``current_path`` — not just
the leaf name — so two folders that share a leaf never merge. Aggregation is
deterministic (everything sorted, no clock/random) and each folder's evidence
carries a content ``version`` derived only from the inputs that could change it, so
any relevant asset change (new analysis, decision change, move, or becoming a
duplicate) invalidates exactly that folder's version and nothing else.
Documented evidence rules (acceptance criterion 2):
- **Duplicate (non-canonical) assets** — excluded entirely; the canonical carries
the evidence.
- **Missing / archived assets** — excluded: only ``availability_state == 'active'``
assets with a ``current_path`` belong to an active folder.
- **NSFW assets** — counted in the folder total and reported as ``nsfw_count``, but
contribute NO analysis-derived evidence (tags, description, setting, location,
year, people). They are never analysed, and their content must not shape a name
beyond mere presence.
- **Undecided / deferred / not-yet-analysed assets** — counted in the folder total
and reported as ``pending_count``; they contribute no evidence until confirmed
SFW and analysed.
- **Only confirmed-SFW analysed assets** shape the descriptive evidence.
- **Missing fields** (null/empty) simply do not contribute.
- **Conflicting fields** (e.g. several ``approx_year`` values) are never silently
collapsed: the full sorted distribution is returned, plus a ``dominant_year`` that
is set only when one value strictly outnumbers the rest; ties set it to ``None``
and raise ``year_conflict``.
Reuses ``_tags`` from :mod:`photo_pipeline.services.library` (the shared JSON-tag
decode). ponytail: neighbouring-folder names/date ranges and user naming
conventions (concept §7 inputs) are proposal-generation concerns — add them in
US03-03 when a name is actually produced.
"""
from __future__ import annotations
import hashlib
import json
from collections import Counter
from pathlib import Path
from sqlalchemy import select
from sqlalchemy.orm import sessionmaker
from photo_pipeline.models import AnalysisResult, Asset, SafetyReview
from photo_pipeline.services.library import _tags
SFW = "sfw"
NSFW = "nsfw"
ANALYZED = "analyzed"
# Cap the descriptions list so a huge folder can't produce an unbounded payload;
# tags/facets are already bounded by their distinct-value count.
MAX_DESCRIPTIONS = 50
def album_label(path: str, library_roots: tuple[Path, ...] = ()) -> str:
"""The canonical album identity: the leaf folder directly containing ``path``,
shown as a path relative to its library root so nested albums stay distinct
(``Urlaub/Rom`` vs ``Urlaub/Venedig``); ``(root)`` for a photo directly in a
root, and the bare parent name for a photo outside every configured root.
Ported from ``photo_analyzer.album_label`` (donor_ledger.yaml: pa-album-label);
``webapp/query.album_of`` was a verbatim copy (wa-album-of). This is the single
implementation for stats, proposals, and folder-as-album upload.
"""
parent = Path(path).parent
for root in library_roots:
try:
rel = parent.relative_to(root)
return "(root)" if str(rel) == "." else str(rel)
except ValueError:
continue
return parent.name or str(parent)
def _counts(counter: Counter) -> list[dict]:
"""Deterministic ``[{value, count}]``, most frequent first, ties broken by value."""
return [
{"value": value, "count": count}
for value, count in sorted(counter.items(), key=lambda kv: (-kv[1], str(kv[0])))
]
def _dominant(counter: Counter):
"""The single strictly-most-frequent value, or ``(None, conflict)``. ``conflict``
is true when more than one distinct value is present and no strict winner exists."""
if not counter:
return None, False
ranked = counter.most_common()
if len(ranked) == 1:
return ranked[0][0], False
top_value, top_count = ranked[0]
if top_count > ranked[1][1]:
return top_value, False
return None, True
class AlbumService:
def __init__(
self, session_factory: sessionmaker, *, library_roots: tuple[Path, ...] = ()
) -> None:
self._session_factory = session_factory
self._library_roots = tuple(Path(root) for root in library_roots)
def _album_of(self, path: str) -> str:
return album_label(path, self._library_roots)
def _latest_decisions(self, session) -> dict[str, str | None]:
"""asset_id -> latest safety decision (``sfw``/``nsfw``/``deferred``/None)."""
latest: dict[str, str | None] = {}
for review in session.scalars(select(SafetyReview).order_by(SafetyReview.created_at)):
latest[review.asset_id] = review.decision
return latest
def folder_count(self) -> int:
"""Number of distinct active-canonical folders (for paging callers)."""
with self._session_factory() as session:
paths = session.scalars(
select(Asset.current_path).where(
Asset.canonical_asset_id.is_(None),
Asset.availability_state == "active",
Asset.current_path.is_not(None),
)
)
return len({self._album_of(p) for p in paths})
def aggregate_evidence(self, *, offset: int = 0, limit: int | None = None) -> dict:
"""Return a page of per-folder evidence, folders sorted by path.
``{folders: [...], total: int, offset, limit}``. ``limit=None`` returns all.
"""
with self._session_factory() as session:
decisions = self._latest_decisions(session)
rows = list(
session.execute(
select(Asset.id, Asset.current_path, AnalysisResult)
.outerjoin(AnalysisResult, AnalysisResult.asset_id == Asset.id)
.where(
Asset.canonical_asset_id.is_(None),
Asset.availability_state == "active",
Asset.current_path.is_not(None),
)
.order_by(Asset.current_path, Asset.id)
)
)
folders: dict[str, list] = {}
for asset_id, path, result in rows:
folders.setdefault(self._album_of(path), []).append((asset_id, path, result))
ordered = sorted(folders)
total = len(ordered)
page = ordered[offset:] if limit is None else ordered[offset : offset + limit]
evidence = [self._folder_evidence(folder, folders[folder], decisions) for folder in page]
return {"folders": evidence, "total": total, "offset": offset, "limit": limit}
def _folder_evidence(self, folder: str, members: list, decisions: dict) -> dict:
descriptions: list[str] = []
tags: Counter = Counter()
settings: Counter = Counter()
time_of_day: Counter = Counter()
seasons: Counter = Counter()
locations: Counter = Counter()
people: Counter = Counter()
years: Counter = Counter()
analyzed_count = nsfw_count = pending_count = 0
fingerprint: list[tuple] = []
for asset_id, path, result in members:
decision = decisions.get(asset_id)
status = result.status if result is not None else None
# Fingerprint: every input that could change this folder's evidence.
fingerprint.append(
(
asset_id,
path,
decision,
status,
result.description if result is not None else None,
result.tags if result is not None else None,
result.people_count if result is not None else None,
result.setting if result is not None else None,
result.time_of_day if result is not None else None,
result.season if result is not None else None,
_clean(result.location_hint) if result is not None else None,
result.approx_year if result is not None else None,
)
)
if decision == NSFW:
nsfw_count += 1
continue
eligible = decision == SFW and status == ANALYZED and result is not None
if not eligible:
pending_count += 1
continue
analyzed_count += 1
if result.description and result.description.strip():
descriptions.append(result.description.strip())
for tag in _tags(result.tags):
tags[tag] += 1
if result.setting:
settings[result.setting] += 1
if result.time_of_day:
time_of_day[result.time_of_day] += 1
if result.season:
seasons[result.season] += 1
location = _clean(result.location_hint)
if location:
locations[location] += 1
if result.people_count is not None:
people["3+" if result.people_count >= 3 else str(result.people_count)] += 1
if result.approx_year is not None:
years[result.approx_year] += 1
dominant_year, year_conflict = _dominant(years)
# ``folder`` is the album label (identity): a library-relative path like
# ``Urlaub/Rom``, or ``(root)``. Name is its leaf; parent is the prefix.
parent = folder.rsplit("/", 1)[0] if "/" in folder else ""
name = folder.rsplit("/", 1)[-1]
return {
"album": folder,
"folder_name": name,
"parent_path": parent,
"asset_count": len(members),
"analyzed_count": analyzed_count,
"nsfw_count": nsfw_count,
"pending_count": pending_count,
"descriptions": sorted(descriptions)[:MAX_DESCRIPTIONS],
"tags": _counts(tags),
"settings": _counts(settings),
"time_of_day": _counts(time_of_day),
"seasons": _counts(seasons),
"locations": _counts(locations),
"people_counts": _counts(people),
"years": _counts(years),
"dominant_year": dominant_year,
"year_conflict": year_conflict,
"version": _version(fingerprint),
}
def _clean(location: str | None) -> str | None:
"""Drop empty/``null`` location hints (the model emits the string ``"null"``)."""
if not location:
return None
trimmed = location.strip()
if not trimmed or trimmed.lower() == "null":
return None
return trimmed
def _version(fingerprint: list[tuple]) -> str:
"""Deterministic content hash of a folder's evidence inputs. Independent of row
order (the inputs are sorted) so only a real change moves it."""
payload = json.dumps(sorted(fingerprint, key=lambda t: t[0]), ensure_ascii=False)
return hashlib.sha256(payload.encode("utf-8")).hexdigest()

View File

@@ -22,6 +22,19 @@ def test_webapp_album_of_mirrors_album_label():
assert query.album_of(p, lib) == pa.album_label(Path(p), lib), p
def test_albums_service_album_label_matches_donor():
# Parity: the extracted single implementation (US03-01) matches the donor rule
# (donor_ledger.yaml: pa-album-label, wa-album-of). The service takes a tuple of
# library roots; the donor takes one Path.
from photo_pipeline.services.albums import album_label
lib = Path("/lib")
for p in ("/lib/Urlaub/Rom/a.jpg", "/lib/a.jpg", "/other/x/a.jpg", "/lib/x/y/a.jpg"):
assert album_label(p, (lib,)) == pa.album_label(Path(p), lib), p
# No configured roots → bare parent name, like the donor with library=None.
assert album_label("/lib/x/a.jpg", ()) == pa.album_label(Path("/lib/x/a.jpg"), None)
def test_fit_label_left_truncates():
assert pa.fit_label("short", 10) == "short"
assert pa.fit_label("Urlaub/Somewhere/Rom", 8) == "…ere/Rom"

View File

@@ -0,0 +1,225 @@
"""Repository tests for AlbumService.aggregate_evidence (US03-01): folder grouping,
the documented exclusion/conflict rules, ordering, paging, and — the heart of the
story — that each folder's version is stable across reads and invalidated by exactly
the relevant asset changes, before and after those changes.
"""
import uuid
from datetime import datetime, timedelta, timezone
from pathlib import Path
from photo_pipeline.config import Config
from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
from photo_pipeline.models import AnalysisResult, Asset, SafetyReview
from photo_pipeline.services.albums import AlbumService
NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
LIB = (Path("/lib"),) # library root, so album labels are library-relative
def _service(sf):
return AlbumService(sf, library_roots=LIB)
def _factory(tmp_path):
(tmp_path / "data").mkdir()
config = Config.from_env({"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data")})
run_migrations(config.database_url)
return create_session_factory(create_db_engine(config.database_url))
def _asset(sf, path, *, canonical_of=None, availability="active", asset_id=None):
asset_id = asset_id or str(uuid.uuid4())
with sf() as session:
session.add(
Asset(
id=asset_id,
original_path=path,
current_path=path,
discovered_at=NOW,
hash_version=1,
availability_state=availability,
canonical_asset_id=canonical_of,
)
)
session.commit()
return asset_id
def _decide(sf, asset_id, decision, *, at=NOW):
with sf() as session:
session.add(
SafetyReview(id=str(uuid.uuid4()), asset_id=asset_id, decision=decision, created_at=at)
)
session.commit()
def _analyze(sf, asset_id, **fields):
with sf() as session:
row = session.get(AnalysisResult, asset_id) or AnalysisResult(asset_id=asset_id)
row.status = fields.pop("status", "analyzed")
for key, value in fields.items():
setattr(row, key, value)
session.add(row)
session.commit()
def _sfw_analyzed(sf, path, **fields):
asset_id = _asset(sf, path)
_decide(sf, asset_id, "sfw")
_analyze(sf, asset_id, **fields)
return asset_id
def _folders(sf):
return {f["album"]: f for f in _service(sf).aggregate_evidence()["folders"]}
def test_groups_by_full_folder_path_and_summarises_sfw_evidence(tmp_path):
sf = _factory(tmp_path)
_sfw_analyzed(
sf,
"/lib/rome/a.jpg",
description="Colosseum",
tags='["ruins", "city"]',
setting="outdoor",
approx_year=2019,
people_count=2,
location_hint="Rome",
)
_sfw_analyzed(
sf,
"/lib/rome/b.jpg",
description="Forum",
tags='["ruins", "history"]',
setting="outdoor",
approx_year=2019,
people_count=4,
location_hint="Rome",
)
rome = _folders(sf)["rome"]
assert rome["folder_name"] == "rome" and rome["parent_path"] == ""
assert rome["asset_count"] == 2 and rome["analyzed_count"] == 2
assert rome["descriptions"] == ["Colosseum", "Forum"] # sorted
assert rome["tags"][0] == {"value": "ruins", "count": 2} # most frequent first
assert rome["locations"] == [{"value": "Rome", "count": 2}]
assert rome["people_counts"] == [{"value": "2", "count": 1}, {"value": "3+", "count": 1}]
assert rome["dominant_year"] == 2019 and rome["year_conflict"] is False
def test_exclusion_rules_for_dup_nsfw_deferred_and_missing(tmp_path):
sf = _factory(tmp_path)
canonical = _sfw_analyzed(sf, "/lib/mix/keep.jpg", description="kept", tags='["a"]')
# Duplicate variant of the canonical: excluded entirely.
_asset(sf, "/lib/mix/dup.jpg", canonical_of=canonical)
# NSFW: counted, but no descriptive evidence.
nsfw = _asset(sf, "/lib/mix/private.jpg")
_decide(sf, nsfw, "nsfw")
# Deferred and undecided: counted as pending, no evidence.
deferred = _asset(sf, "/lib/mix/later.jpg")
_decide(sf, deferred, "deferred")
_asset(sf, "/lib/mix/unknown.jpg") # no decision at all
# Archived: not in an active folder, excluded entirely.
_asset(sf, "/lib/mix/gone.jpg", availability="archived_offline")
mix = _folders(sf)["mix"]
# Only canonical, active assets are counted: keep + nsfw + deferred + unknown.
# The duplicate variant and the archived asset are excluded entirely.
assert mix["asset_count"] == 4
assert mix["nsfw_count"] == 1
assert mix["pending_count"] == 2 # deferred + unknown
assert mix["analyzed_count"] == 1
assert mix["descriptions"] == ["kept"] and mix["tags"] == [{"value": "a", "count": 1}]
def test_conflicting_year_reports_distribution_and_no_dominant(tmp_path):
sf = _factory(tmp_path)
_sfw_analyzed(sf, "/lib/trip/a.jpg", approx_year=2018)
_sfw_analyzed(sf, "/lib/trip/b.jpg", approx_year=2019)
trip = _folders(sf)["trip"]
assert trip["years"] == [{"value": 2018, "count": 1}, {"value": 2019, "count": 1}]
assert trip["dominant_year"] is None and trip["year_conflict"] is True
def test_folders_are_ordered_and_pageable(tmp_path):
sf = _factory(tmp_path)
for folder in ("/lib/c", "/lib/a", "/lib/b"):
_sfw_analyzed(sf, f"{folder}/x.jpg", description="x")
service = _service(sf)
assert service.folder_count() == 3
full = service.aggregate_evidence()
assert [f["album"] for f in full["folders"]] == ["a", "b", "c"]
page = service.aggregate_evidence(offset=1, limit=1)
assert page["total"] == 3 and page["offset"] == 1
assert [f["album"] for f in page["folders"]] == ["b"]
def test_version_is_stable_across_repeated_reads(tmp_path):
sf = _factory(tmp_path)
_sfw_analyzed(sf, "/lib/rome/a.jpg", description="d", approx_year=2019)
first = _folders(sf)["rome"]["version"]
second = _folders(sf)["rome"]["version"]
assert first == second
def test_version_invalidates_on_new_analysis(tmp_path):
sf = _factory(tmp_path)
_sfw_analyzed(sf, "/lib/rome/a.jpg", description="d")
before = _folders(sf)["rome"]["version"]
added = _asset(sf, "/lib/rome/b.jpg")
_decide(sf, added, "sfw")
_analyze(sf, added, description="new")
after = _folders(sf)["rome"]
assert after["version"] != before and after["analyzed_count"] == 2
def test_version_invalidates_when_a_decision_changes_to_nsfw(tmp_path):
sf = _factory(tmp_path)
asset = _sfw_analyzed(sf, "/lib/rome/a.jpg", description="d")
before = _folders(sf)["rome"]["version"]
_decide(sf, asset, "nsfw", at=NOW + timedelta(hours=1)) # newer decision wins
after = _folders(sf)["rome"]
assert after["version"] != before
assert after["nsfw_count"] == 1 and after["analyzed_count"] == 0
def test_moving_an_asset_invalidates_both_folders_and_not_a_third(tmp_path):
sf = _factory(tmp_path)
mover = _sfw_analyzed(sf, "/lib/src/a.jpg", description="d")
_sfw_analyzed(sf, "/lib/dst/keep.jpg", description="k")
_sfw_analyzed(sf, "/lib/other/z.jpg", description="z")
before = _folders(sf)
other_before = before["other"]["version"]
with sf() as session:
asset = session.get(Asset, mover)
asset.current_path = "/lib/dst/a.jpg"
session.commit()
after = _folders(sf)
assert "src" not in after # source folder is now empty
assert after["dst"]["asset_count"] == 2
assert after["dst"]["version"] != before["dst"]["version"]
assert after["other"]["version"] == other_before # untouched folder is stable
def test_version_invalidates_when_an_asset_becomes_a_duplicate(tmp_path):
sf = _factory(tmp_path)
canonical = _sfw_analyzed(sf, "/lib/rome/a.jpg", description="a")
variant = _sfw_analyzed(sf, "/lib/rome/b.jpg", description="b")
before = _folders(sf)["rome"]["version"]
with sf() as session:
session.get(Asset, variant).canonical_asset_id = canonical
session.commit()
after = _folders(sf)["rome"]
assert after["version"] != before
assert after["asset_count"] == 1 # the variant no longer contributes

View File

@@ -68,6 +68,10 @@
"US02-07": [
"tests/e2e/test_phase_b_pipeline.py",
"tests/e2e/test_analysis_browser.py"
],
"US03-01": [
"tests/unit/test_album_evidence.py",
"tests/integration/test_album_evidence_repo.py"
]
}
}

View File

@@ -0,0 +1,117 @@
"""Unit + property tests for the album-evidence aggregation helpers (US03-01):
conflict/dominant rules, deterministic ordering, location cleaning, and the content
version's stability and sensitivity.
"""
import random
from collections import Counter
from pathlib import Path
from photo_pipeline.services.albums import (
_clean,
_counts,
_dominant,
_version,
album_label,
)
def test_album_label_is_library_relative_and_keeps_nested_albums_distinct():
lib = (Path("/lib"),)
assert album_label("/lib/Urlaub/Rom/a.jpg", lib) == "Urlaub/Rom"
assert album_label("/lib/Urlaub/Venedig/a.jpg", lib) == "Urlaub/Venedig"
assert album_label("/lib/a.jpg", lib) == "(root)" # directly in the root
assert album_label("/elsewhere/x/a.jpg", lib) == "x" # outside → bare parent
assert album_label("/lib/x/a.jpg", ()) == "x" # no configured root → bare parent
# Donor parity lives in tests/characterization/test_pa_album.py.
def test_clean_drops_empty_and_literal_null_locations():
assert _clean(None) is None
assert _clean("") is None
assert _clean(" ") is None
assert _clean("null") is None
assert _clean("NULL") is None
assert _clean(" Rome ") == "Rome"
def test_counts_order_by_frequency_then_value():
counts = _counts(Counter({"beach": 1, "sunset": 3, "sea": 1}))
assert counts == [
{"value": "sunset", "count": 3},
{"value": "beach", "count": 1}, # tie broken alphabetically
{"value": "sea", "count": 1},
]
def test_dominant_requires_a_strict_winner():
assert _dominant(Counter()) == (None, False)
assert _dominant(Counter({2019: 3})) == (2019, False)
assert _dominant(Counter({2019: 3, 2020: 1})) == (2019, False)
# A tie is a conflict: no dominant value, flagged.
assert _dominant(Counter({2019: 2, 2020: 2})) == (None, True)
def test_version_is_stable_and_row_order_independent():
a = [("id1", "/f/a.jpg", "sfw", "analyzed", "d", None, 1, None, None, None, None, 2019)]
b = list(a)
two = a + [("id2", "/f/b.jpg", "sfw", "analyzed", "e", None, 2, None, None, None, None, 2019)]
assert _version(a) == _version(b)
# Reversed input order must not change the version (inputs are sorted).
assert _version(two) == _version(list(reversed(two)))
def test_version_changes_when_any_input_field_changes():
base = (
"id1",
"/f/a.jpg",
"sfw",
"analyzed",
"desc",
None,
1,
"outdoor",
None,
None,
None,
2019,
)
baseline = _version([base])
# Flip each field in turn; every change must move the hash.
for index in range(len(base)):
changed = list(base)
current = changed[index]
changed[index] = 99 if isinstance(current, int) else "CHANGED"
assert _version([tuple(changed)]) != baseline, f"field {index} did not affect the version"
def test_property_dominant_matches_a_brute_force_definition():
rng = random.Random(1234)
for _ in range(500):
counter = Counter()
for _ in range(rng.randint(0, 6)):
counter[rng.randint(2000, 2005)] += 1
value, conflict = _dominant(counter)
if not counter:
assert value is None and conflict is False
continue
top = max(counter.values())
winners = [k for k, v in counter.items() if v == top]
if len(winners) == 1:
assert value == winners[0] and conflict is False
else:
assert value is None and conflict is True
def test_property_counts_is_a_permutation_sorted_by_frequency():
rng = random.Random(99)
for _ in range(300):
counter = Counter()
for _ in range(rng.randint(0, 20)):
counter[rng.choice("abcde")] += 1
counts = _counts(counter)
# Same multiset of (value, count) pairs as the counter.
assert {(c["value"], c["count"]) for c in counts} == set(counter.items())
# Non-increasing frequency.
freqs = [c["count"] for c in counts]
assert freqs == sorted(freqs, reverse=True)