"""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 copy at backend/app/services/migrators/cleanup.py
stays in place until FC-3j; FC-3j will replace its body with thin
re-exports from this module and then delete the wrapper.
"""
from __future__ import annotations

from datetime import UTC, datetime
from pathlib import Path
from typing import Any

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

from ..models import Artist, ImageRecord, LibraryAuditRun, Tag
from ..models.series_page import SeriesPage
from ..models.tag import image_tag


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 find_unused_tags(
    session: Session, *, limit: int | None = None,
) -> list[Tag]:
    """Tags with no image_tag rows AND no series_page rows.

    Sorted by name. Used by both dry-run preview and the live prune.
    A tag is "unused" iff it has zero rows in image_tag AND zero rows
    in series_page (so we don't accidentally prune a series tag that
    happens to have no images yet).
    """
    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()
    stmt = (
        select(Tag)
        .where(Tag.id.not_in(used_via_image_tag))
        .where(Tag.id.not_in(used_via_series))
        .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 all extensions; missing OK.
    if image.sha256:
        bucket = image.sha256[:3]
        for ext in ("jpg", "png", "webp"):
            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]}
    """
    unused = find_unused_tags(session)
    sample = [t.name for t in unused[:50]]
    if dry_run:
        return {"count": len(unused), "sample_names": sample}
    ids = [t.id for t in unused]
    if ids:
        session.execute(
            Tag.__table__.delete().where(Tag.id.in_(ids))
        )
        session.commit()
    return {"deleted": len(ids), "sample_names": sample}


# ---------------------------------------------------------------------------
# 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")


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'. Operator must cancel or wait."""
    if rule not in _VALID_RULES:
        raise ValueError(f"unknown rule {rule!r}; expected one of {_VALID_RULES}")
    existing = session.execute(
        select(LibraryAuditRun.id).where(LibraryAuditRun.status == "running")
    ).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))
    )