FabledCurator/backend/app/services/library_layout.py

"""Milestone #421: where an image file BELONGS, and which rows are not there.

The library is keyed on the Artist row's `slug` — one directory per artist.
It grew a second (and third, and fourth) home for many of them because
`Importer._copy_to_library` used to name the destination after the IMPORT
folder while the downloader wrote under the slug. That writer is fixed
(`utils.paths.canonical_subdir`, task #4244); this module is the other half —
finding the rows whose files are still in the old places, and saying where
each one goes.

## Preview and apply share these predicates, they do not re-derive them

`_misplaced_conditions` and `destination_for` are the whole decision. The
report (task #4245) and the move (task #4246) both spread them rather than
writing their own — the house shape for rule 93, snippet #3087. A preview
that computes its set differently from the apply is a preview that can lie,
and here the apply RENAMES the operator's art.

Nothing in this module writes. It reads rows, and it stats files when asked.
"""
from __future__ import annotations

from dataclasses import dataclass, field
from pathlib import Path

from sqlalchemy import func, select
from sqlalchemy.orm import Session

from ..models import Artist, ImageRecord
from ..utils.paths import canonical_subdir

# Top-level directories under the images root that are STORES, not artists.
# A sweep that treats these as misplaced artwork would relocate the
# thumbnail cache, the attachment blobs, or the credential key.
#   thumbs/      sha-addressed thumbnail cache (NOT path-keyed — see below)
#   attachments/ sha-addressed non-media blobs
#   cookies/, secrets/   credential material
#   _backups/, _quarantine/  backup artifacts; files pulled out of the library
RESERVED_TOP_LEVEL = frozenset({
    "thumbs", "attachments", "cookies", "secrets", "_backups", "_quarantine",
})


def canonical_dir(images_root: Path, slug: str) -> Path:
    """The one directory an artist's files belong under."""
    return images_root / slug


def _misplaced_conditions(images_root: Path, artist_id: int, slug: str) -> list:
    """Rows of `artist_id` whose file is NOT under that artist's canonical
    directory. Spread into both halves — never restated.

    The prefix carries a trailing separator on purpose: without it, artist
    `ara` would match every path under `arbuzbudesh/`, and the sweep would
    report one artist's whole library as correctly placed while quietly
    skipping another's.

    `startswith` compiles to LIKE, where `_` and `%` are wildcards, and this
    does not escape them. That is safe ONLY because `utils.slug.slugify`
    reduces a slug to `[a-z0-9-]` — neither character can reach the pattern.
    Widen that charset and this needs `autoescape=True`, or `poch4n_art`
    starts matching `poch4nXart` too.
    """
    prefix = f"{canonical_dir(images_root, slug)}/"
    return [
        ImageRecord.artist_id == artist_id,
        ImageRecord.path.is_not(None),
        ~ImageRecord.path.startswith(prefix),
    ]


def destination_for(path: str, images_root: Path, slug: str) -> Path | None:
    """Where `path`'s file belongs, or None when this row must not be moved.

    None means: the path is outside the images root, or its top-level segment
    is a reserved store. Both are refusals rather than errors — a row pointing
    somewhere unexpected is exactly what should NOT be relocated automatically.

    ## The one place this diverges from `canonical_subdir`

    A file sitting at the images ROOT with a known artist moves under that
    artist's directory here, where `canonical_subdir` would leave it alone.
    The two answer different questions. At import time an empty subdir means
    no artist was resolved, so there is nothing to canonicalise against. Here
    the row already CARRIES an artist_id, so a file at the root is an anomaly
    with a known correct home — which is the whole point of the sweep.

    (The 660 unattributed files at the root have no artist_id at all and are
    not reachable from these predicates; task #4247 decides those.)
    """
    p = Path(path)
    try:
        rel_dir = p.parent.relative_to(images_root)
    except ValueError:
        return None
    parts = rel_dir.parts
    if parts and parts[0] in RESERVED_TOP_LEVEL:
        return None
    sub = canonical_subdir(str(rel_dir) if str(rel_dir) != "." else "", slug)
    if not sub:
        # At the root, with an artist — see the docstring above.
        return canonical_dir(images_root, slug) / p.name
    return images_root / sub / p.name


@dataclass
class ArtistLayout:
    """One artist's verdict."""

    artist_id: int
    name: str
    slug: str
    canonical_rows: int = 0
    misplaced_rows: int = 0
    # The non-canonical top-level directories this artist's files sit in —
    # "Conto", "StickySpoodge", … This is what makes the report readable as
    # the family list the disk survey found.
    stray_dirs: list[str] = field(default_factory=list)
    collisions: list[str] = field(default_factory=list)
    missing_files: int = 0
    unmovable: int = 0


@dataclass
class LayoutReport:
    artists: list[ArtistLayout] = field(default_factory=list)
    total_rows: int = 0
    canonical_rows: int = 0
    misplaced_rows: int = 0
    collision_count: int = 0
    missing_files: int = 0
    unmovable: int = 0
    unattributed_rows: int = 0

    def as_dict(self) -> dict:
        return {
            "total_rows": self.total_rows,
            "canonical_rows": self.canonical_rows,
            "misplaced_rows": self.misplaced_rows,
            "collision_count": self.collision_count,
            "missing_files": self.missing_files,
            "unmovable": self.unmovable,
            "unattributed_rows": self.unattributed_rows,
            "artists": [
                {
                    "artist_id": a.artist_id,
                    "name": a.name,
                    "slug": a.slug,
                    "canonical_rows": a.canonical_rows,
                    "misplaced_rows": a.misplaced_rows,
                    "stray_dirs": a.stray_dirs,
                    "collisions": a.collisions,
                    "missing_files": a.missing_files,
                    "unmovable": a.unmovable,
                }
                for a in self.artists
                if a.misplaced_rows or a.collisions
            ],
        }


def survey_layout(
    session: Session, images_root: Path, *, check_disk: bool = True,
) -> LayoutReport:
    """Read-only blast radius for the consolidation.

    `check_disk` stats every destination to find rows that would collide with
    a file already there, and sources that have already gone missing. It is
    the honest number and it is what the apply will refuse on, but it costs
    one stat per misplaced row over NFS — turn it off when you only want
    counts.
    """
    report = LayoutReport()
    report.total_rows = session.execute(
        select(func.count(ImageRecord.id))
    ).scalar_one()
    report.unattributed_rows = session.execute(
        select(func.count(ImageRecord.id)).where(ImageRecord.artist_id.is_(None))
    ).scalar_one()

    artists = session.execute(
        select(Artist).order_by(Artist.slug)
    ).scalars().all()

    for artist in artists:
        layout = ArtistLayout(
            artist_id=artist.id, name=artist.name, slug=artist.slug,
        )
        conds = _misplaced_conditions(images_root, artist.id, artist.slug)
        owned = session.execute(
            select(func.count(ImageRecord.id))
            .where(ImageRecord.artist_id == artist.id)
        ).scalar_one()
        rows = session.execute(
            select(ImageRecord.id, ImageRecord.path).where(*conds)
        ).all()
        layout.misplaced_rows = len(rows)
        layout.canonical_rows = owned - len(rows)

        strays: set[str] = set()
        destinations: dict[str, int] = {}
        for row_id, path in rows:
            dest = destination_for(path, images_root, artist.slug)
            if dest is None:
                layout.unmovable += 1
                continue
            try:
                top = Path(path).parent.relative_to(images_root).parts
                strays.add(top[0] if top else "<root>")
            except ValueError:
                strays.add("<outside>")
            key = str(dest)
            if key in destinations:
                layout.collisions.append(key)
            else:
                destinations[key] = row_id
            if check_disk:
                if not Path(path).exists():
                    layout.missing_files += 1
                elif dest.exists():
                    layout.collisions.append(key)

        layout.stray_dirs = sorted(strays)
        report.artists.append(layout)
        report.canonical_rows += layout.canonical_rows
        report.misplaced_rows += layout.misplaced_rows
        report.collision_count += len(layout.collisions)
        report.missing_files += layout.missing_files
        report.unmovable += layout.unmovable

    return report