FabledCurator/backend/app/services/cleanup_service.py

"""FC-3k: first-class admin destructive operations.

Projections are pure SELECTs used by both dry-run preview endpoints
and Tier-B count prompts. Mutations (Task 2) are called from sync
HTTP handlers (small ops) and from Celery tasks in
backend.app.tasks.admin (long ops).

This module is the PERMANENT home of artist-cascade + image-unlink
logic. (The legacy migrators/cleanup.py copy was removed with the rest of
the one-and-done GS/IR migration tooling.)
"""
from __future__ import annotations

import logging
import time
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any

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

from ..models import (
    Artist,
    ImageProvenance,
    ImageRecord,
    LibraryAuditRun,
    Post,
    PostAttachment,
    Tag,
)
from ..models.series_chapter import SeriesChapter
from ..models.series_page import SeriesPage
from ..models.tag import image_tag
from ..utils import safe_probe
from .importer import _VIDEO_DUP_ASPECT_TOL, _VIDEO_DUP_DURATION_TOL_SECONDS

log = logging.getLogger(__name__)

# Sentinel written to duration_seconds when a video was probed but ffprobe
# reported no usable duration (missing/corrupt file) — distinct from NULL (never
# probed) so the backfill doesn't re-probe it forever, and < 0 so it can never
# match a real duration in the dedup grouping (#871).
_VIDEO_DURATION_UNKNOWN = -1.0


def project_artist_cascade(session: Session, *, slug: str) -> dict:
    """Read-only projection of what delete_artist_cascade would touch.

    Returns:
        {
          "artist": {"id": int, "name": str, "slug": str},
          "projected": {
            "images": int,
            "sources": int,
            "thumbs": int,        # images with a thumbnail_path set
            "import_tasks": int,  # ImportTask rows referencing the artist's images
            "bytes_on_disk": int, # SUM(image_record.size_bytes) — column is NOT NULL
          },
        }
    Raises LookupError if slug not found. No mutations.
    """
    from ..models.import_task import ImportTask
    from ..models.source import Source

    artist = session.execute(
        select(Artist).where(Artist.slug == slug)
    ).scalar_one_or_none()
    if artist is None:
        raise LookupError(f"artist slug not found: {slug!r}")

    images_count = session.execute(
        select(func.count(ImageRecord.id))
        .where(ImageRecord.artist_id == artist.id)
    ).scalar_one()
    sources_count = session.execute(
        select(func.count(Source.id))
        .where(Source.artist_id == artist.id)
    ).scalar_one()
    thumbs_count = session.execute(
        select(func.count(ImageRecord.id))
        .where(ImageRecord.artist_id == artist.id)
        .where(ImageRecord.thumbnail_path.is_not(None))
    ).scalar_one()
    import_tasks_count = session.execute(
        select(func.count(ImportTask.id))
        .where(
            ImportTask.result_image_id.in_(
                select(ImageRecord.id).where(ImageRecord.artist_id == artist.id)
            )
        )
    ).scalar_one()
    bytes_on_disk = session.execute(
        select(func.coalesce(func.sum(ImageRecord.size_bytes), 0))
        .where(ImageRecord.artist_id == artist.id)
    ).scalar_one()

    return {
        "artist": {"id": artist.id, "name": artist.name, "slug": artist.slug},
        "projected": {
            "images": images_count,
            "sources": sources_count,
            "thumbs": thumbs_count,
            "import_tasks": import_tasks_count,
            "bytes_on_disk": int(bytes_on_disk),
        },
    }


def project_bulk_image_delete(
    session: Session, *, image_ids: list[int],
) -> dict:
    """Read-only projection of what delete_images would touch.

    Returns:
        {
          "images_found": int,
          "thumbs_to_unlink": int,
          "bytes_on_disk": int,
          "missing_ids": list[int],   # ids passed in that don't exist
        }
    No mutations.
    """
    if not image_ids:
        return {
            "images_found": 0,
            "thumbs_to_unlink": 0,
            "bytes_on_disk": 0,
            "missing_ids": [],
        }

    rows = session.execute(
        select(
            ImageRecord.id,
            ImageRecord.thumbnail_path,
            ImageRecord.size_bytes,
        ).where(ImageRecord.id.in_(image_ids))
    ).all()
    found_ids = {r.id for r in rows}
    missing = sorted(set(image_ids) - found_ids)
    return {
        "images_found": len(rows),
        "thumbs_to_unlink": sum(1 for r in rows if r.thumbnail_path),
        "bytes_on_disk": sum(r.size_bytes for r in rows),
        "missing_ids": missing,
    }


def count_tag_associations(session: Session, *, tag_id: int) -> int:
    """COUNT(*) FROM image_tag WHERE tag_id=?. For Tier-B prompt."""
    return session.execute(
        select(func.count())
        .select_from(image_tag)
        .where(image_tag.c.tag_id == tag_id)
    ).scalar_one()


def _unused_tag_conditions() -> list:
    """The WHERE conditions that define an 'unused' tag — the SINGLE source of
    truth shared by find_unused_tags (preview sample), the dry-run count, AND the
    live delete, so the preview can NEVER diverge from what the delete removes.

    A tag is "unused" iff it has zero references across ALL the ways a tag can be
    in use:
      - image_tag       (applied to an image)
      - series_page     (a series tag with ordered pages)
      - series_chapter  (a series tag with chapters but no pages yet)
      - tag.fandom_id   (a fandom referenced by a character)

    The fandom check is essential: fandom tags are NEVER applied to images — a
    character carries its fandom via fandom_id — so without it every assigned
    fandom looks "unused", and the FK is ondelete=SET NULL, so deleting one
    silently strips the fandom off all its characters. The delete had only the
    first two checks while the preview sample had all four, so the preview showed
    a safe list but the delete removed every fandom anyway (operator-flagged
    2026-06-08). Defining the predicate once makes that impossible.
    """
    used_via_image_tag = select(image_tag.c.tag_id).distinct()
    used_via_series = select(SeriesPage.series_tag_id).where(
        SeriesPage.series_tag_id.is_not(None)
    ).distinct()
    used_via_chapter = select(SeriesChapter.series_tag_id).distinct()
    used_via_fandom = select(Tag.fandom_id).where(
        Tag.fandom_id.is_not(None)
    ).distinct()
    return [
        Tag.id.not_in(used_via_image_tag),
        Tag.id.not_in(used_via_series),
        Tag.id.not_in(used_via_chapter),
        Tag.id.not_in(used_via_fandom),
    ]


def find_unused_tags(
    session: Session, *, limit: int | None = None,
) -> list[Tag]:
    """Tags genuinely referenced by nothing — safe to sweep. Sorted by name.
    Shares its predicate with the live prune via _unused_tag_conditions()."""
    stmt = select(Tag).where(*_unused_tag_conditions()).order_by(Tag.name)
    if limit is not None:
        stmt = stmt.limit(limit)
    return list(session.execute(stmt).scalars().all())


def unlink_image_files(
    image: ImageRecord, images_root: Path,
) -> dict:
    """Best-effort unlink of all on-disk files for an ImageRecord.

    Targets: image.path (original), image.thumbnail_path (cached
    thumbnail), and the computed thumbs path at
    /images/thumbs/<sha256[:3]>/<sha256>.(jpg|png|webp) (tries all
    three extensions; missing extension is silently OK).

    Returns {"original": bool, "thumbnail": bool}. Missing files
    count as success (missing_ok semantics). OSErrors are swallowed
    and reported as False so the calling DB delete still proceeds.
    """
    out = {"original": False, "thumbnail": False}
    if image.path:
        try:
            Path(image.path).unlink(missing_ok=True)
            out["original"] = True
        except OSError:
            out["original"] = False
    # Custom thumbnail_path (when set) — try it first.
    if image.thumbnail_path:
        try:
            Path(image.thumbnail_path).unlink(missing_ok=True)
            out["thumbnail"] = True
        except OSError:
            out["thumbnail"] = False
    # Convention thumbs dir — try both extensions thumbnailer writes
    # (.jpg for opaque, .png for alpha). `.webp` used to be in this
    # tuple but the thumbnailer never writes it (operator-flagged in
    # the 2026-06-02 audit) — keep the tuple aligned with what
    # actually lands on disk.
    if image.sha256:
        bucket = image.sha256[:3]
        for ext in ("jpg", "png"):
            try:
                (images_root / "thumbs" / bucket / f"{image.sha256}.{ext}").unlink(
                    missing_ok=True,
                )
            except OSError:
                pass
    return out


def delete_artist_cascade(
    session: Session, *, artist_id: int, images_root: Path,
) -> dict:
    """Batched delete of an artist's images + the artist row.

    Mirrors the cleanup_artist_async pattern: 500-row batches,
    commit between batches so partial progress survives a worker
    kill. Idempotent on missing artist (returns zeroed counts).
    Postgres cascades handle image_tag / image_provenance /
    series_page / tag_suggestion_rejection from ImageRecord delete,
    and source / post / download_event / etc. from Artist delete
    (via Artist.sources cascade="all, delete-orphan").
    """
    artist = session.get(Artist, artist_id)
    if artist is None:
        return {
            "artist": None,
            "summary": {
                "images_deleted": 0,
                "files_deleted": 0,
                "thumbs_deleted": 0,
                "import_tasks_nulled": 0,
                "files_failed": 0,
            },
        }
    artist_info = {"id": artist.id, "name": artist.name, "slug": artist.slug}

    images_deleted = 0
    files_deleted = 0
    thumbs_deleted = 0
    files_failed = 0

    while True:
        rows = session.execute(
            select(ImageRecord)
            .where(ImageRecord.artist_id == artist.id)
            .limit(500)
        ).scalars().all()
        if not rows:
            break
        for img in rows:
            unlinked = unlink_image_files(img, images_root)
            if unlinked["original"]:
                files_deleted += 1
            else:
                files_failed += 1
            if unlinked["thumbnail"]:
                thumbs_deleted += 1
            session.delete(img)
            images_deleted += 1
        session.commit()

    # ImportTask.result_image_id FK is SET NULL on image delete (Postgres
    # handles this in the cascade above). We don't separately count those
    # in FC-3k — the legacy cleanup_artist_async did it via
    # source_path_prefix matching that's out of scope here.
    import_tasks_nulled = 0

    session.delete(artist)
    session.commit()

    return {
        "artist": artist_info,
        "summary": {
            "images_deleted": images_deleted,
            "files_deleted": files_deleted,
            "thumbs_deleted": thumbs_deleted,
            "import_tasks_nulled": import_tasks_nulled,
            "files_failed": files_failed,
        },
    }


def delete_images(
    session: Session, *, image_ids: list[int], images_root: Path,
) -> dict:
    """Delete a list of images in 500-row batches with commit between.

    Postgres CASCADE on image_tag / image_provenance / series_page /
    tag_suggestion_rejection / post_attachment(FK SET NULL) handles
    the DB side; this function handles file unlinks first then row
    deletes. Idempotent on missing IDs (returned as missing_ids;
    no error). On partial OSError, the row is still deleted and
    files_failed is incremented.
    """
    if not image_ids:
        return {
            "images_deleted": 0,
            "files_deleted": 0,
            "thumbs_deleted": 0,
            "files_failed": 0,
            "missing_ids": [],
        }

    seen_ids: set[int] = set()
    images_deleted = 0
    files_deleted = 0
    thumbs_deleted = 0
    files_failed = 0

    pending = list(image_ids)
    while pending:
        batch_ids = pending[:500]
        pending = pending[500:]
        rows = session.execute(
            select(ImageRecord).where(ImageRecord.id.in_(batch_ids))
        ).scalars().all()
        for img in rows:
            seen_ids.add(img.id)
            unlinked = unlink_image_files(img, images_root)
            if unlinked["original"]:
                files_deleted += 1
            else:
                files_failed += 1
            if unlinked["thumbnail"]:
                thumbs_deleted += 1
            session.delete(img)
            images_deleted += 1
        session.commit()

    missing = sorted(set(image_ids) - seen_ids)
    return {
        "images_deleted": images_deleted,
        "files_deleted": files_deleted,
        "thumbs_deleted": thumbs_deleted,
        "files_failed": files_failed,
        "missing_ids": missing,
    }


def delete_tag(session: Session, *, tag_id: int) -> dict:
    """Simple DELETE FROM tag WHERE id=?.

    Postgres cascades the rest (image_tag, tag_alias, tag_allowlist,
    tag_reference_embedding, tag_suggestion_rejection, series_page).
    Returns counts BEFORE delete so the caller can surface them.
    Raises LookupError if tag_id not found.
    """
    tag = session.get(Tag, tag_id)
    if tag is None:
        raise LookupError(f"tag id not found: {tag_id}")
    associations_count = count_tag_associations(session, tag_id=tag_id)
    info = {"id": tag.id, "name": tag.name, "kind": tag.kind.value}
    session.delete(tag)
    session.commit()
    return {"deleted": info, "associations_removed": associations_count}


def prune_unused_tags(session: Session, *, dry_run: bool = False) -> dict:
    """Find tags with zero references and (unless dry_run) delete them.

    Returns:
      dry_run=True:  {"count": N, "sample_names": [first 50]}
      dry_run=False: {"deleted": N, "sample_names": [first 50]}

    Implementation note: the previous SELECT-ids → DELETE-WHERE-IN
    pattern was vulnerable to the psycopg 65535-parameter ceiling on
    libraries with tag explosions. The live delete runs a single DELETE
    with the SAME predicate (_unused_tag_conditions) the preview uses, so
    the row count scales without binding every id as a parameter AND the
    delete can never remove a tag the preview deemed safe. Audit 2026-06-02;
    predicate unified 2026-06-08.
    """
    conditions = _unused_tag_conditions()
    sample_rows = find_unused_tags(session, limit=50)
    sample = [t.name for t in sample_rows]
    if dry_run:
        count = session.execute(
            select(func.count()).select_from(Tag).where(*conditions)
        ).scalar_one()
        return {"count": count, "sample_names": sample}
    result = session.execute(Tag.__table__.delete().where(*conditions))
    session.commit()
    return {"deleted": result.rowcount or 0, "sample_names": sample}


def _bare_post_conditions() -> list:
    """The WHERE conditions that define a 'bare' post — the SINGLE source of truth
    shared by find_bare_posts (preview sample), the dry-run count, AND the live
    delete, so the preview can NEVER diverge from what the delete removes
    ([[feedback_preview_apply_parity]]).

    A post is "bare" iff NOTHING is attached to it across every way content links
    to a post:
      - image_record.primary_post_id  (a post's own canonical images)
      - image_provenance.post_id      (cross-posted/duplicate images linked here)
      - post_attachment.post_id        (preserved non-art files)

    These are exactly the shells the empty-post flood produced: the native Patreon
    ingester synthesized a Post per walked post, but when its only content was a
    duplicate image/attachment that linked to an EARLIER post, the new post was
    left with none of the three (operator-flagged 2026-06-08). Every FK to post.id
    is SET NULL or CASCADE, so deleting a bare post is non-destructive by
    construction — there is nothing pointing at it to orphan. Must run AFTER the
    provenance-render fix so a post that DOES have a hidden provenance link is
    spared, not deleted.
    """
    has_primary = select(ImageRecord.id).where(
        ImageRecord.primary_post_id == Post.id
    )
    has_provenance = select(ImageProvenance.id).where(
        ImageProvenance.post_id == Post.id
    )
    has_attachment = select(PostAttachment.id).where(
        PostAttachment.post_id == Post.id
    )
    return [
        ~has_primary.exists(),
        ~has_provenance.exists(),
        ~has_attachment.exists(),
    ]


def find_bare_posts(
    session: Session, *, limit: int | None = None,
) -> list[Post]:
    """Posts with zero linked images (primary OR provenance) AND zero
    attachments — safe to sweep. Sorted by id. Shares its predicate with the
    live prune via _bare_post_conditions()."""
    stmt = select(Post).where(*_bare_post_conditions()).order_by(Post.id)
    if limit is not None:
        stmt = stmt.limit(limit)
    return list(session.execute(stmt).scalars().all())


def _bare_post_label(post: Post) -> str:
    """Human label for the preview sample — mirrors the feed's fallback title."""
    if post.post_title:
        return post.post_title
    return f"Post {post.external_post_id or post.id}"


def prune_bare_posts(session: Session, *, dry_run: bool = False) -> dict:
    """Find posts with no images and no attachments and (unless dry_run) delete
    them.

    Returns:
      dry_run=True:  {"count": N, "sample_names": [first 50]}
      dry_run=False: {"deleted": N, "sample_names": [first 50]}

    The live delete runs a single DELETE with the SAME predicate
    (_bare_post_conditions) the preview uses, so the row count scales without
    binding every id as a parameter ([[reference_psycopg_65535_param_ceiling]])
    AND the delete can never remove a post the preview deemed kept.
    """
    conditions = _bare_post_conditions()
    sample_rows = find_bare_posts(session, limit=50)
    sample = [_bare_post_label(p) for p in sample_rows]
    if dry_run:
        count = session.execute(
            select(func.count()).select_from(Post).where(*conditions)
        ).scalar_one()
        return {"count": count, "sample_names": sample}
    result = session.execute(Post.__table__.delete().where(*conditions))
    session.commit()
    return {"deleted": result.rowcount or 0, "sample_names": sample}


# Legacy tags FC no longer uses, in two shapes:
#   (1) kinds the tag input never produces — archive/post/artist.
#       provenance (post grouping) + archive membership are their own
#       systems now, and artists are first-class Artist/Source rows.
#       meta/rating were already hard-deleted by alembic 0023.
#   (2) name prefixes from IR kinds FC never adopted — `source:*`.
#       ImageRepo had a `source` kind; FC's enum doesn't, so ir_ingest
#       fell those back to `general` (kind=general, name="source:patreon"
#       etc.). They can't be caught by kind, so we match the name prefix.
PURGEABLE_TAG_KINDS = ("archive", "post", "artist")
LEGACY_NAME_PREFIXES = ("source:",)


def _legacy_tag_predicate():
    name_clauses = [Tag.name.like(f"{p}%") for p in LEGACY_NAME_PREFIXES]
    return or_(Tag.kind.in_(PURGEABLE_TAG_KINDS), *name_clauses)


def purge_legacy_tags(session: Session, *, dry_run: bool = False) -> dict:
    """Count (dry_run) or delete legacy IR-migration tags: archive/post/
    artist-kind tags PLUS general tags whose name matches a legacy
    prefix (source:*).

    CASCADE on image_tag / tag_alias / tag_allowlist /
    tag_reference_embedding / tag_suggestion_rejection / series_page
    clears the related rows on the parent DELETE.

    Returns:
      {"by_kind": {kind: count, ...},        # kind-matched rows
       "by_prefix": {"source:*": count},     # name-prefix-matched rows
       "count": total, "sample_names": [first 50],
       and on live runs "deleted": total}
    """
    predicate = _legacy_tag_predicate()
    rows = session.execute(
        select(Tag.id, Tag.name, Tag.kind).where(predicate)
    ).all()
    by_kind: dict[str, int] = {}
    by_prefix: dict[str, int] = {}
    for _id, name, kind in rows:
        # Classify by name-prefix first so a source:* row counts once,
        # under the prefix bucket, regardless of its (general) kind.
        matched_prefix = next(
            (p for p in LEGACY_NAME_PREFIXES if name.startswith(p)), None,
        )
        if matched_prefix is not None:
            label = f"{matched_prefix}*"
            by_prefix[label] = by_prefix.get(label, 0) + 1
        else:
            key = kind.value if hasattr(kind, "value") else str(kind)
            by_kind[key] = by_kind.get(key, 0) + 1
    sample = [name for _id, name, _kind in rows[:50]]
    total = len(rows)
    result = {
        "by_kind": by_kind, "by_prefix": by_prefix,
        "count": total, "sample_names": sample,
    }
    if dry_run:
        return result
    if total:
        session.execute(Tag.__table__.delete().where(predicate))
        session.commit()
    result["deleted"] = total
    return result


# The Camie-suggestable CONTENT vocabulary. "Reset content tagging" wipes
# these so the operator can re-tag from scratch via auto-suggest. fandom +
# series (and series_page ordering) are deliberately NOT here — they're kept.
RESETTABLE_TAG_KINDS = ("general", "character")


def reset_content_tagging(session: Session, *, dry_run: bool = False) -> dict:
    """Count (dry_run) or DELETE every general + character tag so the operator
    can re-tag from scratch via the Camie auto-suggest.

    PRESERVED: fandom + series tags and their series_page ordering, plus every
    image's image_prediction rows (untouched) so suggestions
    repopulate immediately. CASCADE on image_tag / tag_alias / tag_allowlist /
    tag_reference_embedding / tag_suggestion_rejection clears each deleted
    tag's applications + metadata. Tag.fandom_id is SET NULL, so deleting
    character tags never touches the fandom rows. Irreversible except via DB
    backup restore.

    Returns:
      {"by_kind": {"general": N, "character": M},
       "count": total tags,
       "applications": image_tag rows that will be / were removed,
       "sample_names": [first 50],
       and on live runs "deleted": total}
    """
    predicate = Tag.kind.in_(RESETTABLE_TAG_KINDS)
    rows = session.execute(
        select(Tag.id, Tag.name, Tag.kind).where(predicate)
    ).all()
    by_kind: dict[str, int] = {}
    for _id, _name, kind in rows:
        key = kind.value if hasattr(kind, "value") else str(kind)
        by_kind[key] = by_kind.get(key, 0) + 1
    # Headline impact: applications (image_tag rows) that vanish via cascade.
    applications = session.execute(
        select(func.count())
        .select_from(image_tag)
        .where(image_tag.c.tag_id.in_(select(Tag.id).where(predicate)))
    ).scalar_one()
    sample = [name for _id, name, _kind in rows[:50]]
    total = len(rows)
    result = {
        "by_kind": by_kind,
        "count": total,
        "applications": applications,
        "sample_names": sample,
    }
    if dry_run:
        return result
    if total:
        session.execute(Tag.__table__.delete().where(predicate))
        session.commit()
    result["deleted"] = total
    return result


# ---------------------------------------------------------------------------
# FC-Cleanup additions (2026-05-26): retroactive audit of import-filter rules.
# ---------------------------------------------------------------------------

_MIN_DIM_SAMPLE_CAP = 50


def project_min_dimension_violations(
    session: Session, *, min_width: int, min_height: int,
) -> dict:
    """Return {count, sample_ids} for image_record rows with width or
    height below the thresholds. Synchronous SQL — no PIL inspection
    needed since width/height are stored columns."""
    base = select(ImageRecord.id).where(
        (ImageRecord.width < min_width) | (ImageRecord.height < min_height)
    )
    count = session.execute(
        select(func.count()).select_from(base.subquery())
    ).scalar_one()
    sample_ids = session.execute(
        base.order_by(ImageRecord.id).limit(_MIN_DIM_SAMPLE_CAP)
    ).scalars().all()
    return {"count": count, "sample_ids": list(sample_ids)}


def delete_min_dimension_violations(
    session: Session, *, min_width: int, min_height: int, images_root: Path,
) -> int:
    """Delete every image_record where width<min_w OR height<min_h.
    Routes through delete_images so file-unlink + cascading FKs
    (image_tag / image_provenance / etc.) are handled uniformly."""
    ids = session.execute(
        select(ImageRecord.id).where(
            (ImageRecord.width < min_width) | (ImageRecord.height < min_height)
        )
    ).scalars().all()
    if not ids:
        return 0
    result = delete_images(
        session, image_ids=list(ids), images_root=images_root,
    )
    return result["images_deleted"]


# ---------------------------------------------------------------------------
# Audit lifecycle (transparency + single_color async scans).
# ---------------------------------------------------------------------------


class AuditAlreadyRunning(Exception):
    """Another audit_run is currently in status='running' — wait or
    cancel it before starting a new one. Surfaces as HTTP 409 in the
    /api/cleanup/audit POST endpoint."""


class AuditNotReady(Exception):
    """apply_audit_run called on an audit whose status is not 'ready'."""


class ConfirmTokenMismatch(Exception):
    """Operator-supplied confirm token did not match server-recomputed token."""


_VALID_RULES = ("transparency", "single_color")


_AUDIT_GUARD_THRESHOLD_MINUTES = 135  # matches LIBRARY_AUDIT_STALL_THRESHOLD_MINUTES


def start_audit_run(
    session: Session, *, rule: str, params: dict[str, Any],
) -> int:
    """Create a LibraryAuditRun row in status='running' and dispatch the
    scan_library_for_rule Celery task. Returns the new audit_id.

    Concurrent-runs guard: raises AuditAlreadyRunning if any audit_run
    has status='running' AND started recently. Audit 2026-06-02 made
    the guard age-aware: a SIGKILL'd run leaves a row in 'running'
    that the recovery sweep flips on its next pass (~5 min), but a
    fresh start_audit_run between the SIGKILL and the sweep would
    previously block forever. Past the threshold, treat the running
    row as stale and let the sweep clean it up — the new run still
    gets to start.
    """
    if rule not in _VALID_RULES:
        raise ValueError(f"unknown rule {rule!r}; expected one of {_VALID_RULES}")
    cutoff = datetime.now(UTC) - timedelta(minutes=_AUDIT_GUARD_THRESHOLD_MINUTES)
    existing = session.execute(
        select(LibraryAuditRun.id)
        .where(LibraryAuditRun.status == "running")
        .where(LibraryAuditRun.started_at >= cutoff)
    ).scalar_one_or_none()
    if existing is not None:
        raise AuditAlreadyRunning(existing)
    audit = LibraryAuditRun(
        rule=rule,
        params=params,
        status="running",
        scanned_count=0,
        matched_count=0,
        matched_ids=[],
    )
    session.add(audit)
    session.flush()
    audit_id = audit.id
    # Dispatch after flush so audit_id is populated; commit happens in
    # the API handler so the audit row + dispatch are visible together.
    from ..tasks.library_audit import scan_library_for_rule
    scan_library_for_rule.delay(audit_id)
    return audit_id


def apply_audit_run(
    session: Session, *, audit_id: int, confirm_token: str, images_root: Path,
) -> int:
    """Delete all images in audit_run.matched_ids after confirming token.
    Marks audit status='applied'. Routes through delete_images so files
    + cascading FK rows are handled uniformly."""
    audit = session.execute(
        select(LibraryAuditRun).where(LibraryAuditRun.id == audit_id)
    ).scalar_one_or_none()
    if audit is None:
        raise ValueError(f"audit_run {audit_id} not found")
    if audit.status != "ready":
        raise AuditNotReady(audit.status)
    # Token format matches modal/DestructiveConfirmModal.vue convention:
    # ${action}-${kind}-${runId}. The modal hardcodes action ∈ {'restore',
    # 'delete'}; "apply audit" is semantically a delete of the matched
    # images, so we use 'delete-audit-<id>' (not 'apply-audit-<id>').
    expected = f"delete-audit-{audit_id}"
    if confirm_token != expected:
        raise ConfirmTokenMismatch(expected)
    ids = list(audit.matched_ids or [])
    deleted = 0
    if ids:
        result = delete_images(session, image_ids=ids, images_root=images_root)
        deleted = result["images_deleted"]
    session.execute(
        update(LibraryAuditRun)
        .where(LibraryAuditRun.id == audit_id)
        .values(status="applied", finished_at=datetime.now(UTC))
    )
    return deleted


def cancel_audit_run(session: Session, *, audit_id: int) -> None:
    """Flip a running audit_run to 'cancelled'. The scan task checks
    for status=='cancelled' between batches and exits cleanly."""
    session.execute(
        update(LibraryAuditRun)
        .where(LibraryAuditRun.id == audit_id)
        .where(LibraryAuditRun.status == "running")
        .values(status="cancelled", finished_at=datetime.now(UTC))
    )


# -- archive-attachment re-extraction (#713 part 2) ------------------------

_ARCHIVE_EXT_FOR_FORMAT = {"zip": ".zip", "rar": ".rar", "7z": ".7z"}


def _reextract_archive_to_post(
    importer, archive_path: Path, post, source_row, artist, images_root: Path,
) -> list[int]:
    """Extract one stored archive and link its members to `post`.

    The stored attachment has no adjacent sidecar (it lives in the sha-addressed
    attachment store). Stage a copy + a reconstructed sidecar UNDER the artist's
    library dir (`images_root/<slug>/<platform>/<post>/`) — the importer
    re-derives the artist from the path AND copies members relative to it, so the
    members land in the real library and resolve to the right artist — then re-run
    `attach_in_place`: the archive extracts and `find_or_create_post` re-attaches
    the members to the SAME Post (source_id + external_post_id). Removes only the
    staged archive + sidecar afterward; the imported member files stay. Returns
    the new member image ids.
    """
    import json
    import shutil

    from .archive_extractor import detect_archive_format

    fmt = detect_archive_format(archive_path)
    ext = _ARCHIVE_EXT_FOR_FORMAT.get(fmt or "", ".zip")
    platform = source_row.platform if source_row is not None else "imported"
    sidecar = {
        "category": source_row.platform if source_row is not None
        else (post.raw_metadata or {}).get("category"),
        "id": post.external_post_id,
        "title": post.post_title or "",
        "content": post.description or "",
        "published_at": post.post_date.isoformat() if post.post_date else None,
        "url": post.post_url,
    }
    work = images_root / artist.slug / platform / str(post.external_post_id)
    work.mkdir(parents=True, exist_ok=True)
    staged = work / f"archive{ext}"  # clean ext → is_archive + find_sidecar
    sidecar_path = staged.with_suffix(".json")
    try:
        shutil.copy2(archive_path, staged)
        sidecar_path.write_text(json.dumps(sidecar))
        res = importer.attach_in_place(staged, artist=artist, source=source_row)
        return list(res.member_image_ids or [])
    finally:
        # Drop only the staged archive + sidecar; the extracted member files
        # were copied into the library alongside them and must stay.
        staged.unlink(missing_ok=True)
        sidecar_path.unlink(missing_ok=True)


def reextract_archive_attachments(
    session: Session,
    *,
    images_root: Path,
    time_budget_seconds: float | None = None,
    after_id: int = 0,
) -> dict:
    """Re-process existing PostAttachments that are ACTUALLY archives but were
    filed opaquely before #713 part 1 (extension-only is_archive missed mangled /
    extension-less Patreon attachment names). For each: extract the members,
    import them, and link them to the attachment's post.

    Idempotent — members dedupe by sha256, the archive dedupes by sha — so it's
    safe to run repeatedly. Returns a summary dict for task_run.metadata.

    Time-boxed + resumable: scans PostAttachments in ascending id order starting
    after ``after_id``. When ``time_budget_seconds`` elapses, stops and reports
    ``partial=True`` + ``resume_after_id`` (the last scanned id) so the task can
    re-enqueue itself and continue — a large archive back-catalog can't run the
    task into the Celery time limit or hog the maintenance lane. A bare re-run
    (after_id=0) would never advance because an already-extracted archive is
    still an archive on disk, so the cursor is what guarantees forward progress.
    """
    from ..models import ImportSettings, Post, PostAttachment, Source
    from ..tasks.ml import tag_and_embed
    from ..tasks.thumbnail import generate_thumbnail
    from .archive_extractor import is_archive
    from .importer import Importer
    from .thumbnailer import Thumbnailer

    summary = {
        "scanned": 0, "archives": 0, "members_imported": 0,
        "posts_touched": 0, "skipped_no_post": 0, "skipped_no_artist": 0,
        "errors": 0, "partial": False, "resume_after_id": after_id,
    }
    settings = ImportSettings.load_sync(session)
    importer = Importer(
        session=session, images_root=images_root, import_root=images_root,
        thumbnailer=Thumbnailer(images_root=images_root), settings=settings,
    )

    attachments = session.execute(
        select(PostAttachment)
        .where(PostAttachment.id > after_id)
        .order_by(PostAttachment.id)
    ).scalars().all()
    enqueue_ids: list[int] = []
    start = time.monotonic()
    for att in attachments:
        summary["scanned"] += 1
        summary["resume_after_id"] = att.id
        stored = Path(att.path)
        try:
            if not stored.is_file() or not is_archive(stored):
                continue
        except OSError:
            continue
        summary["archives"] += 1
        if att.post_id is None:
            summary["skipped_no_post"] += 1
            continue
        post = session.get(Post, att.post_id)
        if post is None:
            summary["skipped_no_post"] += 1
            continue
        artist = session.get(Artist, att.artist_id) if att.artist_id else None
        if artist is None and post.artist_id:
            artist = session.get(Artist, post.artist_id)
        if artist is None or not artist.slug:
            # The importer re-derives the artist from the staged path, so we need
            # a real artist+slug to anchor under. (Shouldn't happen for
            # subscription posts; skip rather than orphan the members.)
            summary["skipped_no_artist"] += 1
            continue
        source_row = session.get(Source, post.source_id) if post.source_id else None
        try:
            ids = _reextract_archive_to_post(
                importer, stored, post, source_row, artist, images_root,
            )
            session.commit()
        except Exception as exc:  # one bad archive must not strand the rest
            session.rollback()
            summary["errors"] += 1
            log.warning("re-extract failed for attachment %s: %s", att.id, exc)
            continue
        if ids:
            summary["members_imported"] += len(ids)
            summary["posts_touched"] += 1
            enqueue_ids.extend(ids)

        # Time-box the chunk. resume_after_id already points at this attachment,
        # so the next run starts strictly after it. Checked after the commit so a
        # half-extracted archive never straddles the boundary.
        if (
            time_budget_seconds is not None
            and time.monotonic() - start >= time_budget_seconds
        ):
            summary["partial"] = True
            break
    else:
        # Loop ran to exhaustion — nothing left to resume.
        summary["partial"] = False

    # Thumbnails + ML for the newly-imported members (best-effort; off the
    # critical path — a Redis hiccup must not fail the whole re-extract).
    for img_id in enqueue_ids:
        try:
            generate_thumbnail.delay(img_id)
            tag_and_embed.delay(img_id)
        except Exception as exc:
            log.warning("re-extract enqueue failed for image %s: %s", img_id, exc)
    return summary


# ---- Tier-1 video dedup (#871) ------------------------------------------


def _aspect_matches(w, h, cw, ch) -> bool:
    """Same aspect ratio within tolerance; missing dims don't block (duration is
    the primary signal). Mirrors Importer._video_aspect_matches."""
    if not (w and h and cw and ch):
        return True
    return abs((w / h) - (cw / ch)) <= _VIDEO_DUP_ASPECT_TOL


def backfill_video_durations(session: Session) -> int:
    """Populate image_record.duration_seconds for video rows imported before #871
    (NULL). Idempotent — only NULL rows are touched, so a re-run after a timeout
    naturally resumes. A probe that yields no duration writes the
    _VIDEO_DURATION_UNKNOWN sentinel so the file isn't re-probed forever (and can
    never match a real duration). Returns the count of rows given a real duration.
    """
    populated = 0
    while True:
        rows = session.execute(
            select(ImageRecord.id, ImageRecord.path)
            .where(
                ImageRecord.mime.like("video/%"),
                ImageRecord.duration_seconds.is_(None),
            )
            .order_by(ImageRecord.id)
            .limit(500)
        ).all()
        if not rows:
            break
        for rid, path in rows:
            probe = safe_probe.probe_video(Path(path))
            dur = probe.duration if probe.ok and probe.duration else None
            session.execute(
                update(ImageRecord)
                .where(ImageRecord.id == rid)
                .values(
                    duration_seconds=dur if dur is not None else _VIDEO_DURATION_UNKNOWN
                )
            )
            if dur is not None:
                populated += 1
        session.commit()
    return populated


def _video_dup_group(members: list) -> dict:
    """Pick the keeper (highest pixel area, then largest bytes, then lowest id for
    stability) and describe the group."""
    keeper = max(
        members,
        key=lambda m: ((m.width or 0) * (m.height or 0), m.size_bytes or 0, -m.id),
    )
    losers = [m for m in members if m.id != keeper.id]
    return {
        "artist_id": keeper.artist_id,
        "keeper_id": keeper.id,
        "loser_ids": [m.id for m in losers],
        "duration": keeper.duration_seconds,
        "count": len(members),
        "reclaim_bytes": sum((m.size_bytes or 0) for m in losers),
    }


def find_video_dup_groups(session: Session) -> list[dict]:
    """Cluster videos that are the same content (#871): same artist, duration
    within tolerance, matching aspect ratio. Returns groups of >1 member. Greedy
    sweep over duration-sorted rows, anchored to each cluster's first member so the
    cluster's duration span never exceeds the tolerance (no chain drift)."""
    rows = session.execute(
        select(
            ImageRecord.id, ImageRecord.artist_id, ImageRecord.duration_seconds,
            ImageRecord.width, ImageRecord.height, ImageRecord.size_bytes,
        )
        .where(
            ImageRecord.mime.like("video/%"),
            ImageRecord.duration_seconds.is_not(None),
            ImageRecord.duration_seconds > 0,
            ImageRecord.artist_id.is_not(None),
        )
        .order_by(
            ImageRecord.artist_id, ImageRecord.duration_seconds, ImageRecord.id
        )
    ).all()
    groups: list[dict] = []
    cluster: list = []
    anchor = None
    for r in rows:
        if (
            anchor is not None
            and r.artist_id == anchor.artist_id
            and (r.duration_seconds - anchor.duration_seconds)
            <= _VIDEO_DUP_DURATION_TOL_SECONDS
            and _aspect_matches(r.width, r.height, anchor.width, anchor.height)
        ):
            cluster.append(r)
        else:
            if len(cluster) > 1:
                groups.append(_video_dup_group(cluster))
            cluster = [r]
            anchor = r
    if len(cluster) > 1:
        groups.append(_video_dup_group(cluster))
    return groups


def _relink_provenance_to_keeper(
    session: Session, *, loser_id: int, keeper_id: int
) -> int:
    """Ensure the keeper has an ImageProvenance row for every post the loser was
    linked to, so deleting the loser never drops the video off a post. Returns the
    number of new keeper↔post links added."""
    rows = session.execute(
        select(ImageProvenance.post_id, ImageProvenance.source_id)
        .where(ImageProvenance.image_record_id == loser_id)
    ).all()
    added = 0
    for post_id, source_id in rows:
        exists = session.execute(
            select(ImageProvenance.id).where(
                ImageProvenance.image_record_id == keeper_id,
                ImageProvenance.post_id == post_id,
            )
        ).scalar_one_or_none()
        if exists is None:
            session.add(ImageProvenance(
                image_record_id=keeper_id, post_id=post_id, source_id=source_id,
            ))
            session.flush()
            added += 1
    return added


def dedup_videos(
    session: Session, *, images_root: Path, dry_run: bool = False
) -> dict:
    """Find and (unless dry_run) collapse Tier-1 video duplicates (#871).

    Re-probes NULL-duration videos first so the existing library participates,
    then clusters by artist + duration + aspect and keeps the highest-res copy per
    cluster. On apply, each loser's post links are re-pointed to the keeper BEFORE
    the loser record + file are deleted, so no post loses the video. dry_run shares
    the same discovery predicate and returns the projection without deleting
    (rule 93).

    NOTE: tags/curation on a loser are NOT merged onto the keeper — videos rarely
    carry hand-curation and merging would add FK-juggling risk. Flagged as a
    follow-up if it ever matters.
    """
    backfill_video_durations(session)
    groups = find_video_dup_groups(session)
    redundant = sum(len(g["loser_ids"]) for g in groups)
    reclaim = sum(g["reclaim_bytes"] for g in groups)
    sample = [
        {
            "keeper_id": g["keeper_id"],
            "redundant": len(g["loser_ids"]),
            "duration": round(g["duration"], 1),
        }
        for g in groups[:50]
    ]
    if dry_run:
        return {
            "groups": len(groups), "redundant": redundant,
            "reclaim_bytes": reclaim, "sample": sample,
        }

    relinked = 0
    for g in groups:
        for loser_id in g["loser_ids"]:
            relinked += _relink_provenance_to_keeper(
                session, loser_id=loser_id, keeper_id=g["keeper_id"]
            )
        session.commit()

    loser_ids = [lid for g in groups for lid in g["loser_ids"]]
    deleted = delete_images(session, image_ids=loser_ids, images_root=images_root)
    log.info(
        "video dedup: %d group(s), %d redundant removed, %d post link(s) re-pointed",
        len(groups), deleted["images_deleted"], relinked,
    )
    return {
        "groups": len(groups), "redundant": redundant,
        "reclaim_bytes": reclaim, "deleted": deleted["images_deleted"],
        "files_deleted": deleted["files_deleted"],
        "relinked_posts": relinked, "sample": sample,
    }