diff --git a/photo_pipeline/services/naming.py b/photo_pipeline/services/naming.py new file mode 100644 index 0000000..6759986 --- /dev/null +++ b/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} diff --git a/tests/story_traceability.json b/tests/story_traceability.json index 0811ea9..f837702 100644 --- a/tests/story_traceability.json +++ b/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" ] } } diff --git a/tests/unit/test_naming_policy.py b/tests/unit/test_naming_policy.py new file mode 100644 index 0000000..db9fb99 --- /dev/null +++ b/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