photoanalyzer/photo_pipeline/services/naming.py

"""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}