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

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

from sqlalchemy import func, or_, 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}


# 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


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