Files
FabledCurator/backend/app/api/admin.py
T
bvandeusen 6281cb1e66
CI / lint (push) Successful in 3s
CI / frontend-build (push) Successful in 20s
CI / backend-lint-and-test (push) Successful in 27s
CI / integration (push) Successful in 3m18s
refactor(admin): consolidate Tier-A dry-run/apply handlers onto one helper (#753)
DRY pass on the cleanup/admin destructive-ops surface (task #753, hardened
process #594). Five Tier-A endpoints repeated the same get_json -> dry_run ->
run_sync(service_fn) -> jsonify block verbatim. Extract _run_dry_run_op(service_fn,
**kwargs); the five route handlers now delegate. reconcile keeps its source_id
validation and passes it through **kwargs.

The cleanup_service predicates were already shared between preview and apply
(find_*_conditions / find_duplicate_post_groups) — the post-data-loss fix — so no
backend-logic change; this is purely the HTTP-handler boilerplate.

Consumers (all routed through the helper, verified no copy left behind):
  prune_unused_tags, prune_bare_posts, reconcile_duplicate_posts (+source_id),
  purge_legacy_tags, reset_content_tagging.

Added route-level tests for prune-bare (apply) and reconcile (apply + source_id
passthrough + invalid-source_id 400) — the two helper consumers that previously
had only service-level coverage, so every consumer is exercised at the route.

Findings B (queued-response helper) and C (store dry-run POST helper) identified
but not applied this pass (operator scoped to A). The card preview->commit state
machine is deferred to a frontend pattern-consistency sweep.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-22 14:53:04 -04:00

416 lines
17 KiB
Python

"""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)
POST /api/admin/posts/prune-bare (Tier A)
POST /api/admin/tags/purge-legacy (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, text
from ..extensions import get_session
from ..models import Artist
from ..services.cleanup_service import project_artist_cascade, project_bulk_image_delete
from ._responses import error_response as _bad
admin_bp = Blueprint("admin", __name__, url_prefix="/api/admin")
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]
async def _run_dry_run_op(service_fn, **service_kwargs):
"""Shared body for the Tier-A dry-run/apply endpoints: read the `dry_run`
flag, run the cleanup_service predicate under `run_sync`, and return its
result dict. The SAME `service_fn` drives both preview and apply (the flag
just toggles), so a handler physically can't let its preview diverge from
its delete (rule 93). Default False preserves the existing contract — the UI
always passes `dry_run` explicitly (true to preview, false to apply). Extra
service kwargs (e.g. `source_id`) pass straight through."""
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: service_fn(sync_sess, dry_run=dry_run, **service_kwargs)
)
return jsonify(result)
@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
return await _run_dry_run_op(prune_unused_tags)
@admin_bp.route("/posts/prune-bare", methods=["POST"])
async def posts_prune_bare():
"""Tier-A: delete bare posts — Post rows with no linked images (primary OR
provenance) and no attachments. Dry-run preview list IS the prompt: UI calls
with dry_run=true first, shows the count + sample, operator confirms by
re-calling with dry_run=false. Same preview/apply-parity predicate as the
prune itself, so the preview can't diverge from the delete."""
from ..services.cleanup_service import prune_bare_posts
return await _run_dry_run_op(prune_bare_posts)
@admin_bp.route("/posts/reconcile-duplicates", methods=["POST"])
async def posts_reconcile_duplicates():
"""Tier-A: unify duplicate post rows for the same real post — the gallery-dl
(attachment-id) + native (post-id) duplicates — onto ONE post-id-keyed keeper,
moving image/provenance/attachment/link rows over. Images are untouched.
dry_run=true returns {groups, posts_to_merge, sample}; dry_run=false applies
and returns {groups, merged, sample}. Optional source_id scopes to one source.
Same find_duplicate_post_groups predicate drives preview + apply (rule 93)."""
from ..services.cleanup_service import reconcile_duplicate_posts
body = await request.get_json(silent=True) or {}
raw_source = body.get("source_id")
try:
source_id = int(raw_source) if raw_source is not None else None
except (TypeError, ValueError):
return _bad("invalid_source_id", detail="source_id must be an integer")
return await _run_dry_run_op(reconcile_duplicate_posts, source_id=source_id)
@admin_bp.route("/tags/purge-legacy", methods=["POST"])
async def tags_purge_legacy():
"""Tier-A: delete legacy IR-migration tags — archive/post/artist
kinds (e.g. `BlenderKnight:Hannah_BJ_Loops`) PLUS general tags with
a legacy name prefix (`source:*`, from IR's source kind that fell
back to general). dry-run preview returns per-kind + per-prefix
counts + a sample so the UI shows exactly what'll go before the
operator confirms with dry_run=false."""
from ..services.cleanup_service import purge_legacy_tags
return await _run_dry_run_op(purge_legacy_tags)
@admin_bp.route("/tags/reset-content", methods=["POST"])
async def tags_reset_content():
"""Tier-A: delete ALL general + character tags (the Camie-suggestable
content vocabulary) so the operator can re-tag from scratch via
auto-suggest. fandom + series tags + series_page ordering are preserved,
and image_prediction rows are untouched so suggestions repopulate.
dry-run preview returns per-kind counts + applications + a sample so the
UI shows exactly what'll go before the operator confirms (dry_run=false).
Irreversible except via DB backup restore."""
from ..services.cleanup_service import reset_content_tagging
return await _run_dry_run_op(reset_content_tagging)
@admin_bp.route("/tags/normalize", methods=["POST"])
async def tags_normalize():
"""#714: retro-normalize existing tags to the #701 canonical form (Title
Case + collapsed whitespace) and merge case/whitespace-variant duplicates.
dry_run=true (default) returns a projection inline — group/collision/rename
counts + a sample of the changes — so the UI shows exactly what'll happen.
dry_run=false dispatches the long-running maintenance task (the merge FK
repoints can touch many tags); the UI tails the activity dashboard for the
summary. Idempotent; back up first (the merges are irreversible)."""
from ..services.tag_service import normalize_existing_tags
body = await request.get_json(silent=True) or {}
dry_run = bool(body.get("dry_run", True))
if dry_run:
async with get_session() as session:
result = await normalize_existing_tags(session, dry_run=True)
return jsonify(result)
from ..tasks.admin import normalize_tags_task
async_result = normalize_tags_task.delay()
return jsonify({"task_id": async_result.id, "status": "queued"}), 202
@admin_bp.route("/maintenance/db-stats", methods=["GET"])
async def db_stats():
"""Per-table bloat readout (pg_stat_user_tables) for the high-churn tables
so the operator can see when a VACUUM is worth running."""
from ..tasks.maintenance import VACUUM_TABLES
wanted = set(VACUUM_TABLES)
async with get_session() as session:
rows = (await session.execute(text(
"SELECT relname, n_live_tup, n_dead_tup, last_vacuum, "
"last_autovacuum, last_analyze FROM pg_stat_user_tables"
))).all()
def _iso(v):
return v.isoformat() if v is not None else None
out = []
for r in rows:
if r.relname not in wanted:
continue
live = r.n_live_tup or 0
dead = r.n_dead_tup or 0
total = live + dead
out.append({
"table": r.relname,
"live": live,
"dead": dead,
"dead_pct": round(100 * dead / total, 1) if total else 0.0,
"last_vacuum": _iso(r.last_vacuum),
"last_autovacuum": _iso(r.last_autovacuum),
"last_analyze": _iso(r.last_analyze),
})
out.sort(key=lambda t: t["dead"], reverse=True)
return jsonify({"tables": out})
@admin_bp.route("/maintenance/vacuum", methods=["POST"])
async def trigger_vacuum():
"""Operator-triggered VACUUM (ANALYZE) over the high-churn tables — the
same maintenance-queue task the weekly Beat schedule runs."""
from ..tasks.maintenance import vacuum_analyze
vacuum_analyze.delay()
return jsonify({"status": "queued"}), 202
@admin_bp.route("/maintenance/reextract-archives", methods=["POST"])
async def trigger_reextract_archives():
"""Operator-triggered re-extract (#713): PostAttachments that are actually
archives but were filed opaquely (pre magic-byte gate) get extracted and
their members linked to the post. Idempotent; runs on the maintenance queue."""
from ..tasks.admin import reextract_archive_attachments_task
async_result = reextract_archive_attachments_task.delay()
return jsonify({"task_id": async_result.id, "status": "queued"}), 202
@admin_bp.route("/maintenance/prune-missing-files", methods=["POST"])
async def trigger_prune_missing_files():
"""Operator-triggered orphan repair (#859): delete ImageRecords whose backing
file is gone from disk (e.g. left by the external-attach unlink bug), so they
stop 404-ing on playback. The task aborts WITHOUT deleting if a large fraction
of files look missing (a filesystem/NFS stall). Maintenance queue;
operator-triggered only — never an unattended sweep."""
from ..tasks.admin import prune_missing_file_records_task
async_result = prune_missing_file_records_task.delay()
return jsonify({"task_id": async_result.id, "status": "queued"}), 202
@admin_bp.route("/maintenance/dedup-videos", methods=["POST"])
async def trigger_dedup_videos():
"""Tier-1 video dedup (#871). Body {"dry_run": bool}: dry_run=true previews
what would be removed (groups / redundant count / reclaimable bytes) WITHOUT
deleting; dry_run=false applies it (re-link posts to the keeper, then delete
the redundant copies). Either way it first re-probes NULL-duration videos so
the existing library participates. Returns the Celery task id — poll
/maintenance/task-result/<id> for the summary."""
from ..tasks.admin import dedup_videos_task
body = await request.get_json(silent=True) or {}
dry_run = bool(body.get("dry_run", True)) # default to the SAFE preview
async_result = dedup_videos_task.delay(dry_run=dry_run)
return jsonify({"task_id": async_result.id, "status": "queued"}), 202
@admin_bp.route("/maintenance/purge-gated-previews", methods=["POST"])
async def trigger_purge_gated_previews():
"""Cleanup (#874 follow-up). Body {"dry_run": bool}: dry_run=true previews how
many blurred locked-preview images (grabbed from tier-gated Patreon posts
before the fix) would be removed WITHOUT deleting; dry_run=false applies it.
Re-walks every enabled Patreon source read-only and matches by content hash, so
real content downloaded when access existed is provably spared. Returns the
Celery task id — poll /maintenance/task-result/<id> for the summary."""
from ..tasks.admin import purge_gated_previews_task
body = await request.get_json(silent=True) or {}
dry_run = bool(body.get("dry_run", True)) # default to the SAFE preview
async_result = purge_gated_previews_task.delay(dry_run=dry_run)
return jsonify({"task_id": async_result.id, "status": "queued"}), 202
@admin_bp.route("/maintenance/task-result/<task_id>", methods=["GET"])
async def maintenance_task_result(task_id: str):
"""Poll a maintenance Celery task's result (the summary dict it returns).
Used by the video-dedup card to show the dry-run projection before apply."""
from ..celery_app import celery
res = celery.AsyncResult(task_id)
ready = res.ready()
return jsonify({
"ready": ready,
"successful": res.successful() if ready else None,
"result": res.result if (ready and res.successful()) else None,
})