US03-02: Define Album Naming Policy (#62)

photo_pipeline/services/naming.py

@@ -0,0 +1,186 @@
+"""Album naming policy: render candidate album names from evidence fields and
+validate them for supported filesystems (US03-02). Pure logic — it never touches
+the disk; it only proposes and inspects strings.
+
+The policy is template-driven. A template is an ordered tuple of evidence field
+names; rendering drops fields that are missing/empty and joins the rest with the
+separator, and the first template in the chain that yields a non-empty result wins.
+This gives the concept's graceful fallbacks (§7)::
+
+    date place event  →  "2019-07 — Rome — Summer Holiday"
+    date event        →  "2019 — Summer Holiday"
+    place event       →  "Rome — Summer Holiday"
+    event             →  "Summer Holiday"
+
+``date`` combines year and (optional) month into ``YYYY`` / ``YYYY-MM`` (album_naming.md:
+year leads, ISO format). Naming conventions follow the concept and album_naming.md:
+UTF-8 kept (umlauts survive), no underscores/double spaces, a single ``—`` qualifier
+separator.
+
+Validation is non-destructive: it sanitizes a proposed string, reports every change
+and blocking problem as a structured issue, and offers filesystem-safe suggestions
+(including collision-disambiguated variants) — but only the caller ever renames.
+"""
+
+from __future__ import annotations
+
+import re
+import unicodedata
+from dataclasses import dataclass, field
+
+# Characters no supported filesystem tolerates in a path component, plus the path
+# separators themselves (a name must never introduce a new path level) and control
+# characters. Replaced with a space during sanitisation.
+FORBIDDEN = set('/\\:*?"<>|') | {chr(c) for c in range(0x20)}
+_FORBIDDEN_RE = re.compile("[" + re.escape("".join(sorted(FORBIDDEN))) + "]")
+_WHITESPACE_RE = re.compile(r"\s+")
+
+# Case-insensitive reserved basenames on Windows; rejected so a name stays portable.
+RESERVED = (
+    {"CON", "PRN", "AUX", "NUL"}
+    | {f"COM{n}" for n in range(1, 10)}
+    | {f"LPT{n}" for n in range(1, 10)}
+)
+
+DEFAULT_SEPARATOR = " — "
+DEFAULT_MAX_LENGTH = 120
+DEFAULT_TEMPLATES = (
+    ("date", "place", "event"),
+    ("date", "event"),
+    ("place", "event"),
+    ("date", "place"),
+    ("event",),
+    ("place",),
+    ("date",),
+)
+ALLOWED_FIELDS = ("date", "year", "month", "place", "event")
+
+
+@dataclass(frozen=True)
+class NamingPolicy:
+    separator: str = DEFAULT_SEPARATOR
+    max_length: int = DEFAULT_MAX_LENGTH
+    templates: tuple[tuple[str, ...], ...] = DEFAULT_TEMPLATES
+    empty_fallback: str = "(untitled)"
+    allowed_fields: tuple[str, ...] = ALLOWED_FIELDS
+
+
+@dataclass
+class ValidationResult:
+    name: str  # the sanitised name ("" when nothing survives)
+    valid: bool
+    issues: list[dict] = field(default_factory=list)
+    suggestions: list[str] = field(default_factory=list)
+
+
+def date_field(year: int | None, month: int | None) -> str | None:
+    """``YYYY-MM`` when a month is known, ``YYYY`` with only a year, else ``None``."""
+    if year is None:
+        return None
+    return f"{year}-{month:02d}" if month else str(year)
+
+
+def build_fields(evidence: dict) -> dict:
+    """Project an US03-01 folder-evidence dict onto the naming fields. Deterministic:
+    ``place`` is the single most common location, ``event`` is the folder's own name.
+    The richer AI ``event`` naming is US03-03; this keeps a usable default."""
+    locations = evidence.get("locations") or []
+    place = locations[0]["value"] if locations else None
+    return {
+        "date": date_field(evidence.get("dominant_year"), evidence.get("month")),
+        "year": evidence.get("dominant_year"),
+        "month": evidence.get("month"),
+        "place": place,
+        "event": evidence.get("folder_name") or None,
+    }
+
+
+def render_name(fields: dict, policy: NamingPolicy = NamingPolicy()) -> str:
+    """Render the first template whose available fields produce a non-empty name."""
+    for template in policy.templates:
+        pieces = [str(fields[name]).strip() for name in template if _present(fields.get(name))]
+        if pieces:
+            return policy.separator.join(pieces)
+    return ""
+
+
+def _present(value) -> bool:
+    return value is not None and str(value).strip() != ""
+
+
+def sanitize(name: str, policy: NamingPolicy = NamingPolicy()) -> tuple[str, list[dict]]:
+    """Return a filesystem-safe name plus the list of changes made, each a structured
+    issue. Never raises and never returns a name containing a path separator, control
+    character, ``..`` traversal, or leading/trailing space/dot."""
+    issues: list[dict] = []
+    normalized = unicodedata.normalize("NFC", name)
+    if normalized != name:
+        issues.append(_issue("normalized_unicode", "applied Unicode NFC normalization"))
+
+    replaced = _FORBIDDEN_RE.sub(" ", normalized)
+    if replaced != normalized:
+        issues.append(_issue("removed_forbidden_characters", "replaced forbidden characters"))
+
+    # Strip spaces and dots together from the ends in one pass: doing them
+    # separately isn't idempotent ("x. ." would leave a trailing "x."). Interior
+    # dots (e.g. "St. Louis") are preserved; leading dots (hidden files) are not.
+    collapsed = _WHITESPACE_RE.sub(" ", replaced).strip(" .")
+    if collapsed != replaced:
+        issues.append(_issue("normalized_whitespace", "collapsed whitespace and trimmed"))
+
+    if len(collapsed) > policy.max_length:
+        collapsed = collapsed[: policy.max_length].strip(" .")
+        issues.append(_issue("truncated", f"truncated to {policy.max_length} characters"))
+
+    return collapsed, issues
+
+
+def validate(
+    name: str,
+    *,
+    existing: set[str] | None = None,
+    policy: NamingPolicy = NamingPolicy(),
+) -> ValidationResult:
+    """Sanitise and inspect a proposed name. Reports every change and blocking problem
+    and offers safe suggestions; mutates nothing. ``existing`` is the set of names
+    already taken on the target (compared case-insensitively)."""
+    taken = {e.casefold() for e in (existing or set())}
+    sanitized, issues = sanitize(name, policy)
+    suggestions: list[str] = []
+    valid = True
+
+    if not sanitized:
+        valid = False
+        issues.append(_issue("empty", "name is empty after sanitization"))
+        suggestions.append(policy.empty_fallback)
+        return ValidationResult(name="", valid=False, issues=issues, suggestions=suggestions)
+
+    if sanitized.upper() in RESERVED:
+        valid = False
+        issues.append(_issue("reserved_name", f"{sanitized!r} is a reserved filesystem name"))
+        suggestions.append(_disambiguate(f"{sanitized} (album)", taken, policy))
+
+    if sanitized.casefold() in taken:
+        valid = False
+        issues.append(_issue("collision", f"{sanitized!r} already exists on the target"))
+        suggestions.append(_disambiguate(sanitized, taken, policy))
+
+    return ValidationResult(name=sanitized, valid=valid, issues=issues, suggestions=suggestions)
+
+
+def _disambiguate(name: str, taken: set[str], policy: NamingPolicy) -> str:
+    """Append ``(2)``, ``(3)`` … until the name is free (case-insensitively), bounded
+    by the length limit."""
+    if name.casefold() not in taken and name.upper() not in RESERVED:
+        return name
+    for suffix in range(2, 1000):
+        candidate = f"{name} ({suffix})"
+        if len(candidate) > policy.max_length:
+            candidate = f"{name[: policy.max_length - len(f' ({suffix})')].strip()} ({suffix})"
+        if candidate.casefold() not in taken and candidate.upper() not in RESERVED:
+            return candidate
+    return name  # pathological: 998 collisions — caller still sees the collision issue
+
+
+def _issue(code: str, message: str) -> dict:
+    return {"code": code, "message": message}

tests/story_traceability.json

@@ -72,6 +72,9 @@
     "US03-01": [
       "tests/unit/test_album_evidence.py",
       "tests/integration/test_album_evidence_repo.py"
+    ],
+    "US03-02": [
+      "tests/unit/test_naming_policy.py"
     ]
   }
 }

tests/unit/test_naming_policy.py

@@ -0,0 +1,150 @@
+"""Golden + property tests for the album naming policy (US03-02): template rendering
+and fallbacks, evidence projection, sanitisation/normalisation, reserved names,
+collisions, and the invariant that a sanitised name can never escape a root.
+"""
+
+import random
+import unicodedata
+from pathlib import Path
+
+from photo_pipeline.services.naming import (
+    FORBIDDEN,
+    NamingPolicy,
+    build_fields,
+    date_field,
+    render_name,
+    sanitize,
+    validate,
+)
+
+SEP = NamingPolicy().separator
+
+
+def _codes(result):
+    return {issue["code"] for issue in result.issues}
+
+
+# ── rendering + fallbacks ────────────────────────────────────────────────────
+
+
+def test_full_template_renders_date_place_event():
+    fields = {"date": "2019-07", "place": "Rome", "event": "Summer Holiday"}
+    assert render_name(fields) == SEP.join(["2019-07", "Rome", "Summer Holiday"])
+
+
+def test_template_fallbacks_drop_missing_fields_in_priority_order():
+    assert render_name({"place": "Rome", "event": "Summer Holiday"}) == SEP.join(
+        ["Rome", "Summer Holiday"]
+    )
+    assert render_name({"date": "2016", "event": "Team Event"}) == SEP.join(["2016", "Team Event"])
+    assert render_name({"date": "2019", "place": "Rome"}) == SEP.join(["2019", "Rome"])
+    assert render_name({"event": "Family Portraits"}) == "Family Portraits"
+    assert render_name({"place": "Rome"}) == "Rome"
+    assert render_name({}) == ""
+
+
+def test_date_field_combines_year_and_month_iso():
+    assert date_field(2019, 7) == "2019-07"
+    assert date_field(2019, None) == "2019"
+    assert date_field(None, 7) is None
+
+
+def test_build_fields_projects_evidence_and_renders():
+    evidence = {
+        "dominant_year": 2019,
+        "locations": [{"value": "Rome", "count": 3}, {"value": "Napoli", "count": 1}],
+        "folder_name": "trip",
+    }
+    fields = build_fields(evidence)
+    assert fields["date"] == "2019" and fields["place"] == "Rome" and fields["event"] == "trip"
+    assert render_name(fields) == SEP.join(["2019", "Rome", "trip"])
+
+
+# ── golden validation cases ──────────────────────────────────────────────────
+
+
+def test_valid_name_passes_unchanged():
+    result = validate("2019-07 — Rome — Summer Holiday")
+    assert result.valid and result.name == "2019-07 — Rome — Summer Holiday"
+    assert result.issues == []
+
+
+def test_normalization_collapses_whitespace_and_strips_forbidden():
+    result = validate("  2019   Rome/Trip  ")
+    assert result.name == "2019 Rome Trip"  # slash → space, collapsed, trimmed
+    assert "removed_forbidden_characters" in _codes(result)
+    assert "normalized_whitespace" in _codes(result)
+    assert result.valid
+
+
+def test_unicode_is_nfc_normalized_keeping_umlauts():
+    decomposed = unicodedata.normalize("NFD", "Café")  # e + combining acute
+    result = validate(decomposed)
+    assert result.name == "Café" and "normalized_unicode" in _codes(result)
+
+
+def test_reserved_name_is_rejected_with_a_safe_suggestion():
+    result = validate("con")
+    assert not result.valid and "reserved_name" in _codes(result)
+    assert result.suggestions == ["con (album)"]
+
+
+def test_collision_is_flagged_and_disambiguated():
+    result = validate("2019 Rome", existing={"2019 rome"})  # case-insensitive
+    assert not result.valid and "collision" in _codes(result)
+    assert result.suggestions == ["2019 Rome (2)"]
+
+
+def test_disambiguation_skips_multiple_taken_variants():
+    result = validate("Rome", existing={"rome", "rome (2)", "rome (3)"})
+    assert result.suggestions == ["Rome (4)"]
+
+
+def test_empty_after_sanitization_falls_back():
+    result = validate("///:::")
+    assert not result.valid and "empty" in _codes(result)
+    assert result.name == "" and result.suggestions == ["(untitled)"]
+
+
+def test_overlong_name_is_truncated_to_the_limit():
+    policy = NamingPolicy(max_length=20)
+    result = validate("A" * 50, policy=policy)
+    assert len(result.name) <= 20 and "truncated" in _codes(result)
+
+
+# ── property: sanitisation keeps names safe and inside the root ──────────────
+
+_ALPHABET = 'abcÄ Rome/..\\:*?"<>|.\t\n\x00 é2019—()'
+
+
+def test_property_sanitized_names_are_filesystem_safe_and_cannot_escape_root(tmp_path):
+    rng = random.Random(20260815)
+    root = tmp_path.resolve()
+    policy = NamingPolicy()
+    for _ in range(2000):
+        raw = "".join(rng.choice(_ALPHABET) for _ in range(rng.randint(0, 40)))
+        name, _issues = sanitize(raw, policy)
+
+        assert not (set(name) & FORBIDDEN), f"forbidden char survived: {name!r}"
+        assert name == name.strip() and not name.endswith("."), name
+        assert len(name) <= policy.max_length
+        if not name:
+            continue
+        assert ".." not in Path(name).parts and "/" not in name and "\\" not in name
+        # Joining to a root can never escape it.
+        resolved = (root / name).resolve()
+        assert resolved == root or root in resolved.parents, resolved
+
+
+def test_property_valid_names_never_collide_with_existing(tmp_path):
+    rng = random.Random(7)
+    policy = NamingPolicy()
+    for _ in range(500):
+        existing = {f"album {i}" for i in range(rng.randint(0, 5))}
+        raw = rng.choice(["album 0", "Album 1", "Rome", "2019 Trip", "album 3"])
+        result = validate(raw, existing=existing, policy=policy)
+        if result.valid:
+            assert result.name.casefold() not in {e.casefold() for e in existing}
+        else:
+            # An invalid result always offers at least one suggestion to move forward.
+            assert result.suggestions