157 lines
5.1 KiB
Python
157 lines
5.1 KiB
Python
"""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 sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..models import Artist, ImageRecord, 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())
|