bvandeusen
fb05c5eef7
find_unused_tags only excluded tags with image_tag or series_page references, so it flagged every fandom as 'unused' — fandoms are NEVER applied to images (a character carries its fandom via tag.fandom_id), and the FK is ondelete=SET NULL, so deleting one silently strips the fandom off all its characters (operator-flagged 2026-06-08: artist-OC fandoms showing as unused). Exclude tags referenced as a character's fandom_id, and (same class of gap) tags referenced by a series_chapter (an all-placeholder series has chapters but no pages yet). A genuinely orphaned fandom with no characters is still swept. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
855 lines
32 KiB
Python
855 lines
32 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 migrators/cleanup.py copy was removed with the rest of
|
|
the one-and-done GS/IR migration tooling.)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import time
|
|
from datetime import UTC, datetime, timedelta
|
|
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_chapter import SeriesChapter
|
|
from ..models.series_page import SeriesPage
|
|
from ..models.tag import image_tag
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
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 genuinely referenced by nothing — safe to sweep.
|
|
|
|
Sorted by name. Used by both dry-run preview and the live prune. A tag is
|
|
"unused" iff it has zero references across ALL the ways a tag can be in use:
|
|
- image_tag (applied to an image)
|
|
- series_page (a series tag with ordered pages)
|
|
- series_chapter (a series tag with chapters but no pages yet, e.g.
|
|
all-placeholder)
|
|
- tag.fandom_id (a fandom referenced by a character)
|
|
|
|
The fandom check is essential: fandom tags are NEVER applied to images — a
|
|
character carries its fandom via fandom_id — so without it every assigned
|
|
fandom looked "unused", and the FK is ondelete=SET NULL, so deleting one
|
|
would silently strip the fandom off all its characters (operator-flagged
|
|
2026-06-08).
|
|
"""
|
|
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()
|
|
used_via_chapter = select(SeriesChapter.series_tag_id).distinct()
|
|
used_via_fandom = select(Tag.fandom_id).where(
|
|
Tag.fandom_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))
|
|
.where(Tag.id.not_in(used_via_chapter))
|
|
.where(Tag.id.not_in(used_via_fandom))
|
|
.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 both extensions thumbnailer writes
|
|
# (.jpg for opaque, .png for alpha). `.webp` used to be in this
|
|
# tuple but the thumbnailer never writes it (operator-flagged in
|
|
# the 2026-06-02 audit) — keep the tuple aligned with what
|
|
# actually lands on disk.
|
|
if image.sha256:
|
|
bucket = image.sha256[:3]
|
|
for ext in ("jpg", "png"):
|
|
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]}
|
|
|
|
Implementation note: the previous SELECT-ids → DELETE-WHERE-IN
|
|
pattern was vulnerable to the psycopg 65535-parameter ceiling on
|
|
libraries with tag explosions. The live delete now runs a single
|
|
DELETE with the same NOT-IN predicate find_unused_tags uses, so
|
|
the row count scales without binding every id as a parameter.
|
|
Audit 2026-06-02.
|
|
"""
|
|
sample_rows = find_unused_tags(session, limit=50)
|
|
sample = [t.name for t in sample_rows]
|
|
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()
|
|
if dry_run:
|
|
count = session.execute(
|
|
select(func.count())
|
|
.select_from(Tag)
|
|
.where(Tag.id.not_in(used_via_image_tag))
|
|
.where(Tag.id.not_in(used_via_series))
|
|
).scalar_one()
|
|
return {"count": count, "sample_names": sample}
|
|
result = session.execute(
|
|
Tag.__table__.delete()
|
|
.where(Tag.id.not_in(used_via_image_tag))
|
|
.where(Tag.id.not_in(used_via_series))
|
|
)
|
|
session.commit()
|
|
return {"deleted": result.rowcount or 0, "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
|
|
|
|
|
|
# The Camie-suggestable CONTENT vocabulary. "Reset content tagging" wipes
|
|
# these so the operator can re-tag from scratch via auto-suggest. fandom +
|
|
# series (and series_page ordering) are deliberately NOT here — they're kept.
|
|
RESETTABLE_TAG_KINDS = ("general", "character")
|
|
|
|
|
|
def reset_content_tagging(session: Session, *, dry_run: bool = False) -> dict:
|
|
"""Count (dry_run) or DELETE every general + character tag so the operator
|
|
can re-tag from scratch via the Camie auto-suggest.
|
|
|
|
PRESERVED: fandom + series tags and their series_page ordering, plus every
|
|
image's image_record.tagger_predictions (untouched) so suggestions
|
|
repopulate immediately. CASCADE on image_tag / tag_alias / tag_allowlist /
|
|
tag_reference_embedding / tag_suggestion_rejection clears each deleted
|
|
tag's applications + metadata. Tag.fandom_id is SET NULL, so deleting
|
|
character tags never touches the fandom rows. Irreversible except via DB
|
|
backup restore.
|
|
|
|
Returns:
|
|
{"by_kind": {"general": N, "character": M},
|
|
"count": total tags,
|
|
"applications": image_tag rows that will be / were removed,
|
|
"sample_names": [first 50],
|
|
and on live runs "deleted": total}
|
|
"""
|
|
predicate = Tag.kind.in_(RESETTABLE_TAG_KINDS)
|
|
rows = session.execute(
|
|
select(Tag.id, Tag.name, Tag.kind).where(predicate)
|
|
).all()
|
|
by_kind: dict[str, int] = {}
|
|
for _id, _name, kind in rows:
|
|
key = kind.value if hasattr(kind, "value") else str(kind)
|
|
by_kind[key] = by_kind.get(key, 0) + 1
|
|
# Headline impact: applications (image_tag rows) that vanish via cascade.
|
|
applications = session.execute(
|
|
select(func.count())
|
|
.select_from(image_tag)
|
|
.where(image_tag.c.tag_id.in_(select(Tag.id).where(predicate)))
|
|
).scalar_one()
|
|
sample = [name for _id, name, _kind in rows[:50]]
|
|
total = len(rows)
|
|
result = {
|
|
"by_kind": by_kind,
|
|
"count": total,
|
|
"applications": applications,
|
|
"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")
|
|
|
|
|
|
_AUDIT_GUARD_THRESHOLD_MINUTES = 135 # matches LIBRARY_AUDIT_STALL_THRESHOLD_MINUTES
|
|
|
|
|
|
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' AND started recently. Audit 2026-06-02 made
|
|
the guard age-aware: a SIGKILL'd run leaves a row in 'running'
|
|
that the recovery sweep flips on its next pass (~5 min), but a
|
|
fresh start_audit_run between the SIGKILL and the sweep would
|
|
previously block forever. Past the threshold, treat the running
|
|
row as stale and let the sweep clean it up — the new run still
|
|
gets to start.
|
|
"""
|
|
if rule not in _VALID_RULES:
|
|
raise ValueError(f"unknown rule {rule!r}; expected one of {_VALID_RULES}")
|
|
cutoff = datetime.now(UTC) - timedelta(minutes=_AUDIT_GUARD_THRESHOLD_MINUTES)
|
|
existing = session.execute(
|
|
select(LibraryAuditRun.id)
|
|
.where(LibraryAuditRun.status == "running")
|
|
.where(LibraryAuditRun.started_at >= cutoff)
|
|
).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))
|
|
)
|
|
|
|
|
|
# -- archive-attachment re-extraction (#713 part 2) ------------------------
|
|
|
|
_ARCHIVE_EXT_FOR_FORMAT = {"zip": ".zip", "rar": ".rar", "7z": ".7z"}
|
|
|
|
|
|
def _reextract_archive_to_post(
|
|
importer, archive_path: Path, post, source_row, artist, images_root: Path,
|
|
) -> list[int]:
|
|
"""Extract one stored archive and link its members to `post`.
|
|
|
|
The stored attachment has no adjacent sidecar (it lives in the sha-addressed
|
|
attachment store). Stage a copy + a reconstructed sidecar UNDER the artist's
|
|
library dir (`images_root/<slug>/<platform>/<post>/`) — the importer
|
|
re-derives the artist from the path AND copies members relative to it, so the
|
|
members land in the real library and resolve to the right artist — then re-run
|
|
`attach_in_place`: the archive extracts and `find_or_create_post` re-attaches
|
|
the members to the SAME Post (source_id + external_post_id). Removes only the
|
|
staged archive + sidecar afterward; the imported member files stay. Returns
|
|
the new member image ids.
|
|
"""
|
|
import json
|
|
import shutil
|
|
|
|
from .archive_extractor import detect_archive_format
|
|
|
|
fmt = detect_archive_format(archive_path)
|
|
ext = _ARCHIVE_EXT_FOR_FORMAT.get(fmt or "", ".zip")
|
|
platform = source_row.platform if source_row is not None else "imported"
|
|
sidecar = {
|
|
"category": source_row.platform if source_row is not None
|
|
else (post.raw_metadata or {}).get("category"),
|
|
"id": post.external_post_id,
|
|
"title": post.post_title or "",
|
|
"content": post.description or "",
|
|
"published_at": post.post_date.isoformat() if post.post_date else None,
|
|
"url": post.post_url,
|
|
}
|
|
work = images_root / artist.slug / platform / str(post.external_post_id)
|
|
work.mkdir(parents=True, exist_ok=True)
|
|
staged = work / f"archive{ext}" # clean ext → is_archive + find_sidecar
|
|
sidecar_path = staged.with_suffix(".json")
|
|
try:
|
|
shutil.copy2(archive_path, staged)
|
|
sidecar_path.write_text(json.dumps(sidecar))
|
|
res = importer.attach_in_place(staged, artist=artist, source=source_row)
|
|
return list(res.member_image_ids or [])
|
|
finally:
|
|
# Drop only the staged archive + sidecar; the extracted member files
|
|
# were copied into the library alongside them and must stay.
|
|
staged.unlink(missing_ok=True)
|
|
sidecar_path.unlink(missing_ok=True)
|
|
|
|
|
|
def reextract_archive_attachments(
|
|
session: Session,
|
|
*,
|
|
images_root: Path,
|
|
time_budget_seconds: float | None = None,
|
|
after_id: int = 0,
|
|
) -> dict:
|
|
"""Re-process existing PostAttachments that are ACTUALLY archives but were
|
|
filed opaquely before #713 part 1 (extension-only is_archive missed mangled /
|
|
extension-less Patreon attachment names). For each: extract the members,
|
|
import them, and link them to the attachment's post.
|
|
|
|
Idempotent — members dedupe by sha256, the archive dedupes by sha — so it's
|
|
safe to run repeatedly. Returns a summary dict for task_run.metadata.
|
|
|
|
Time-boxed + resumable: scans PostAttachments in ascending id order starting
|
|
after ``after_id``. When ``time_budget_seconds`` elapses, stops and reports
|
|
``partial=True`` + ``resume_after_id`` (the last scanned id) so the task can
|
|
re-enqueue itself and continue — a large archive back-catalog can't run the
|
|
task into the Celery time limit or hog the maintenance lane. A bare re-run
|
|
(after_id=0) would never advance because an already-extracted archive is
|
|
still an archive on disk, so the cursor is what guarantees forward progress.
|
|
"""
|
|
from ..models import ImportSettings, Post, PostAttachment, Source
|
|
from ..tasks.ml import tag_and_embed
|
|
from ..tasks.thumbnail import generate_thumbnail
|
|
from .archive_extractor import is_archive
|
|
from .importer import Importer
|
|
from .thumbnailer import Thumbnailer
|
|
|
|
summary = {
|
|
"scanned": 0, "archives": 0, "members_imported": 0,
|
|
"posts_touched": 0, "skipped_no_post": 0, "skipped_no_artist": 0,
|
|
"errors": 0, "partial": False, "resume_after_id": after_id,
|
|
}
|
|
settings = ImportSettings.load_sync(session)
|
|
importer = Importer(
|
|
session=session, images_root=images_root, import_root=images_root,
|
|
thumbnailer=Thumbnailer(images_root=images_root), settings=settings,
|
|
)
|
|
|
|
attachments = session.execute(
|
|
select(PostAttachment)
|
|
.where(PostAttachment.id > after_id)
|
|
.order_by(PostAttachment.id)
|
|
).scalars().all()
|
|
enqueue_ids: list[int] = []
|
|
start = time.monotonic()
|
|
for att in attachments:
|
|
summary["scanned"] += 1
|
|
summary["resume_after_id"] = att.id
|
|
stored = Path(att.path)
|
|
try:
|
|
if not stored.is_file() or not is_archive(stored):
|
|
continue
|
|
except OSError:
|
|
continue
|
|
summary["archives"] += 1
|
|
if att.post_id is None:
|
|
summary["skipped_no_post"] += 1
|
|
continue
|
|
post = session.get(Post, att.post_id)
|
|
if post is None:
|
|
summary["skipped_no_post"] += 1
|
|
continue
|
|
artist = session.get(Artist, att.artist_id) if att.artist_id else None
|
|
if artist is None and post.artist_id:
|
|
artist = session.get(Artist, post.artist_id)
|
|
if artist is None or not artist.slug:
|
|
# The importer re-derives the artist from the staged path, so we need
|
|
# a real artist+slug to anchor under. (Shouldn't happen for
|
|
# subscription posts; skip rather than orphan the members.)
|
|
summary["skipped_no_artist"] += 1
|
|
continue
|
|
source_row = session.get(Source, post.source_id) if post.source_id else None
|
|
try:
|
|
ids = _reextract_archive_to_post(
|
|
importer, stored, post, source_row, artist, images_root,
|
|
)
|
|
session.commit()
|
|
except Exception as exc: # one bad archive must not strand the rest
|
|
session.rollback()
|
|
summary["errors"] += 1
|
|
log.warning("re-extract failed for attachment %s: %s", att.id, exc)
|
|
continue
|
|
if ids:
|
|
summary["members_imported"] += len(ids)
|
|
summary["posts_touched"] += 1
|
|
enqueue_ids.extend(ids)
|
|
|
|
# Time-box the chunk. resume_after_id already points at this attachment,
|
|
# so the next run starts strictly after it. Checked after the commit so a
|
|
# half-extracted archive never straddles the boundary.
|
|
if (
|
|
time_budget_seconds is not None
|
|
and time.monotonic() - start >= time_budget_seconds
|
|
):
|
|
summary["partial"] = True
|
|
break
|
|
else:
|
|
# Loop ran to exhaustion — nothing left to resume.
|
|
summary["partial"] = False
|
|
|
|
# Thumbnails + ML for the newly-imported members (best-effort; off the
|
|
# critical path — a Redis hiccup must not fail the whole re-extract).
|
|
for img_id in enqueue_ids:
|
|
try:
|
|
generate_thumbnail.delay(img_id)
|
|
tag_and_embed.delay(img_id)
|
|
except Exception as exc:
|
|
log.warning("re-extract enqueue failed for image %s: %s", img_id, exc)
|
|
return summary
|