FabledCurator/backend/app/api/admin.py

"""FC-3k: /api/admin — destructive admin actions.

Five action surfaces:
  POST   /api/admin/artists/<slug>/cascade-delete  (Tier C)
  POST   /api/admin/images/bulk-delete             (Tier C)
  DELETE /api/admin/tags/<int:tag_id>              (Tier B)
  POST   /api/admin/tags/<int:dest_id>/merge       (Tier B)
  POST   /api/admin/tags/prune-unused              (Tier A)
  GET    /api/admin/tags/<int:tag_id>/usage-count  (helper)

Tier-C ops take a dry_run body flag (returns projection inline,
no dispatch) and a confirm body field (server-recomputed token).
Long-running ops dispatch a maintenance-queue Celery task; the UI
tails FC-3i's /api/system/activity/runs to surface progress.
"""
from __future__ import annotations

import hashlib

from quart import Blueprint, jsonify, request
from sqlalchemy import select

from ..extensions import get_session
from ..models import Artist
from ..services.cleanup_service import project_artist_cascade, project_bulk_image_delete

admin_bp = Blueprint("admin", __name__, url_prefix="/api/admin")


def _bad(error: str, *, status: int = 400, **extra):
    body = {"error": error}
    body.update(extra)
    return jsonify(body), status


def _bulk_image_confirm_token(image_ids: list[int]) -> str:
    """Stable 8-hex token derived from the sorted id list. Mutates
    when the selection changes; stays the same across modal opens of
    the same selection so the operator can paste without confusion."""
    canon = ",".join(str(i) for i in sorted(image_ids))
    digest = hashlib.sha256(canon.encode("utf-8")).hexdigest()
    return digest[:8]


@admin_bp.route("/artists/<slug>/cascade-delete", methods=["POST"])
async def artist_cascade_delete(slug: str):
    body = await request.get_json(silent=True) or {}
    dry_run = bool(body.get("dry_run", False))
    supplied_confirm = body.get("confirm", "")

    async with get_session() as session:
        artist = (await session.execute(
            select(Artist).where(Artist.slug == slug)
        )).scalar_one_or_none()
        if artist is None:
            return _bad("not_found", status=404)
        artist_id = artist.id

        projected = await session.run_sync(
            lambda sync_sess: project_artist_cascade(sync_sess, slug=slug)
        )

    if dry_run:
        return jsonify(projected)

    expected = f"delete-artist-{artist_id}"
    if supplied_confirm != expected:
        return _bad(
            "confirm_mismatch",
            detail=f"confirm must equal {expected!r}",
            expected=expected,
        )

    from ..tasks.admin import delete_artist_cascade_task
    async_result = delete_artist_cascade_task.delay(artist_id=artist_id)
    return jsonify({"task_id": async_result.id}), 202


@admin_bp.route("/images/bulk-delete", methods=["POST"])
async def images_bulk_delete():
    body = await request.get_json(silent=True) or {}
    image_ids = body.get("image_ids")
    if not isinstance(image_ids, list) or not image_ids:
        return _bad("invalid_image_ids", detail="image_ids must be non-empty list of int")
    try:
        image_ids = [int(i) for i in image_ids]
    except (TypeError, ValueError):
        return _bad("invalid_image_ids", detail="image_ids must contain only ints")

    dry_run = bool(body.get("dry_run", False))
    supplied_confirm = body.get("confirm", "")

    async with get_session() as session:
        projected = await session.run_sync(
            lambda sync_sess: project_bulk_image_delete(
                sync_sess, image_ids=image_ids,
            )
        )

    sha8 = _bulk_image_confirm_token(image_ids)
    expected = f"delete-images-{sha8}"

    if dry_run:
        # Hand the canonical Tier-C confirm token back with the
        # projection so the frontend doesn't have to recompute SHA-256
        # client-side via crypto.subtle (Secure-Context-gated,
        # undefined on plain-HTTP origins per the homelab posture).
        # Operator-flagged 2026-05-27.
        projected["confirm_token"] = expected
        return jsonify(projected)


    if supplied_confirm != expected:
        return _bad(
            "confirm_mismatch",
            detail=f"confirm must equal {expected!r}",
            expected=expected,
        )

    from ..tasks.admin import bulk_delete_images_task
    async_result = bulk_delete_images_task.delay(image_ids=image_ids)
    return jsonify({"task_id": async_result.id}), 202


@admin_bp.route("/tags/<int:tag_id>", methods=["DELETE"])
async def tag_delete(tag_id: int):
    """Tier-B sync delete. UI yes/no modal is the only confirmation."""
    from ..services.cleanup_service import delete_tag

    async with get_session() as session:
        try:
            result = await session.run_sync(
                lambda sync_sess: delete_tag(sync_sess, tag_id=tag_id)
            )
        except LookupError:
            return _bad("not_found", status=404)
    return jsonify(result)


@admin_bp.route("/tags/<int:dest_id>/merge", methods=["POST"])
async def tag_merge(dest_id: int):
    """Wraps TagService.merge. Source repoints to dest, dest survives,
    source row deleted, protective alias auto-created if source was
    ML-applied or allowlisted."""
    from ..services.tag_service import TagMergeConflict, TagService, TagValidationError

    body = await request.get_json(silent=True) or {}
    source_id = body.get("source_id")
    if not isinstance(source_id, int) or source_id == dest_id:
        return _bad("invalid_source_id", detail="source_id must be int and differ from dest")

    async with get_session() as session:
        try:
            result = await TagService(session).merge(
                source_id=source_id, target_id=dest_id,
            )
        except TagMergeConflict as exc:
            return _bad("merge_conflict", status=409, detail=str(exc))
        except TagValidationError as exc:
            return _bad("tag_kind_mismatch", detail=str(exc))
        except LookupError:
            return _bad("not_found", status=404)

    # MergeResult is a frozen dataclass — flatten to dict.
    return jsonify({
        "result": {
            "target_id": result.target_id,
            "target_name": result.target_name,
            "target_kind": result.target_kind,
            "merged_count": result.merged_count,
            "alias_created": result.alias_created,
            "source_deleted": result.source_deleted,
        },
    })


@admin_bp.route("/tags/<int:tag_id>/usage-count", methods=["GET"])
async def tag_usage_count(tag_id: int):
    """Helper for the Tier-B yes/no prompt; surfaces "N associations"
    in the dialog so the operator knows what they're nuking."""
    from ..services.cleanup_service import count_tag_associations

    async with get_session() as session:
        count = await session.run_sync(
            lambda sync_sess: count_tag_associations(
                sync_sess, tag_id=tag_id,
            )
        )
    return jsonify({"count": count})


@admin_bp.route("/tags/prune-unused", methods=["POST"])
async def tags_prune_unused():
    """Tier-A: dry-run preview list IS the prompt. UI calls with
    dry_run=true first, shows the list, operator clicks button to
    re-call with dry_run=false."""
    from ..services.cleanup_service import prune_unused_tags

    body = await request.get_json(silent=True) or {}
    dry_run = bool(body.get("dry_run", False))

    async with get_session() as session:
        result = await session.run_sync(
            lambda sync_sess: prune_unused_tags(
                sync_sess, dry_run=dry_run,
            )
        )
    return jsonify(result)