bvandeusenandClaude Opus 5
2e0f8f8c61
PostAttachment's two FKs are both ON DELETE SET NULL, so a deleted post or artist left the row behind rather than taking it. Nothing ever pruned those rows, and nothing in the repo had ever unlinked a file under the attachment store — so both rows and bytes accumulated permanently, invisible to every existing diagnostic. Why a disk->DB reconciliation rather than a row sweep: the store is sha-addressed and idempotent, so ONE blob backs MANY rows. Deleting a row does not free its blob, and since the artist cascade (#3066) now deletes its attachment rows outright, a freed blob has no DB pointer left to find it by. Walking the store and asking "does any row still reference this sha?" catches orphans from every cause, including ones no future delete path will think to report. Preview and apply share `_orphan_attachment_conditions` (rule 93). The dry-run derives its surviving-sha set by NEGATING that same predicate, so it is honest about blobs the delete would free rather than counting them as still-referenced — the one place this was easy to get backwards, so it has its own parity test. Guards, each with a reason: - A blob is written before its row commits, so a just-stored file legitimately has no referencing row. Files under 6h are never judged — same guard and reasoning as ORPHAN_TEMP_MIN_AGE_HOURS. - `.partial` staging files belong to cleanup_orphaned_temp_files; skipped rather than raced. - The sha is parsed as the first 64 chars, not via Path.stem: store() takes the extension from the source filename, and a URL-encoded basename yields a multi-dot suffix that would make stem eat part of the sha. - A 900s walk budget reports partial=True instead of running to the task's hard limit (rule 89). - TASK_STUCK_THRESHOLD_MINUTES override at 30 (= time_limit 25 + 5). Without it a healthy 20-minute walk is phantom-flagged 'RecoverySweep' at the bare 5-min default — the #883 failure class; its invariant test is mirrored here. Defaults to the safe preview at both the task and the route, unlike the other maintenance triggers: this apply unlinks files. Operator-triggered only, never on a beat. Ships with its UI (rule 27): AttachmentReclaimCard in Cleanup → Duplicates & leftovers, built on the existing useMaintenanceTask/MaintenanceTile shapes, so a run survives navigating away. Surfaces files_failed and partial explicitly, since both change what the numbers mean. Also promotes humanBytes to utils/bytes.js — it was byte-identical in VideoDedupCard and GatedPurgeCard and this card would have been the third copy. The three divergent `formatBytes` helpers are deliberately left alone. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1747 lines
69 KiB
Python
1747 lines
69 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 and_, delete, func, or_, select, update
|
|
from sqlalchemy.orm import Session, aliased
|
|
|
|
from ..models import (
|
|
Artist,
|
|
ExternalLink,
|
|
ImageProvenance,
|
|
ImageRecord,
|
|
LibraryAuditRun,
|
|
PatreonFailedMedia,
|
|
PatreonSeenMedia,
|
|
Post,
|
|
PostAttachment,
|
|
Tag,
|
|
)
|
|
from ..models.series_chapter import SeriesChapter
|
|
from ..models.series_page import SeriesPage
|
|
from ..models.tag import image_tag
|
|
from ..utils import safe_probe
|
|
from .importer import _VIDEO_DUP_ASPECT_TOL, _VIDEO_DUP_DURATION_TOL_SECONDS
|
|
from .platforms import PLATFORMS
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
# Sentinel written to duration_seconds when a video was probed but ffprobe
|
|
# reported no usable duration (missing/corrupt file) — distinct from NULL (never
|
|
# probed) so the backfill doesn't re-probe it forever, and < 0 so it can never
|
|
# match a real duration in the dedup grouping (#871).
|
|
_VIDEO_DURATION_UNKNOWN = -1.0
|
|
|
|
|
|
# -- artist-cascade predicates (rule 93: ONE definition, preview + apply) ---
|
|
# project_artist_cascade (preview) and delete_artist_cascade (apply) both build
|
|
# their queries from these. The preview used to re-derive its own — which is how
|
|
# it came to count images and stay silent about posts and attachments while the
|
|
# apply destroyed both. Same failure shape as the 2026-06-08 fandom-tag
|
|
# deletion, where a re-implemented delete predicate diverged from the preview's.
|
|
# Returned as condition LISTS spread into `.where(*conds)`, matching
|
|
# _unused_tag_conditions / _bare_post_conditions below.
|
|
|
|
|
|
def _artist_images_conditions(artist_id: int) -> list:
|
|
"""Images the cascade deletes (rows AND their on-disk files)."""
|
|
return [ImageRecord.artist_id == artist_id]
|
|
|
|
|
|
def _artist_posts_conditions(artist_id: int) -> list:
|
|
"""Posts the cascade destroys. The apply never names these — post.artist_id
|
|
is ondelete=CASCADE, so Postgres takes them when the artist row goes — which
|
|
is exactly why the preview has to name them: an artist whose posts are
|
|
body-only (no images) otherwise previews as `images: 0` and reads as an
|
|
empty artist, while every captured body/description/external-link set is
|
|
destroyed."""
|
|
return [Post.artist_id == artist_id]
|
|
|
|
|
|
def _artist_attachments_conditions(artist_id: int) -> list:
|
|
"""Attachments the cascade deletes. Matched by artist_id OR by the owning
|
|
post's artist: artist_id is nullable (_capture_attachment leaves it NULL
|
|
when no artist resolved), so neither arm alone covers every row. The
|
|
sha-addressed blobs are NOT unlinked (one blob backs many rows) — these are
|
|
row counts, and the bytes are not part of this operation's footprint."""
|
|
return [
|
|
or_(
|
|
PostAttachment.artist_id == artist_id,
|
|
PostAttachment.post_id.in_(
|
|
select(Post.id).where(*_artist_posts_conditions(artist_id))
|
|
),
|
|
)
|
|
]
|
|
|
|
|
|
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,
|
|
"posts": int, # hard-deleted by the post.artist_id CASCADE
|
|
"attachments": int, # rows deleted; the sha-addressed blobs stay
|
|
"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
|
|
},
|
|
}
|
|
Every count is built from the shared `_artist_*_conditions` predicates the
|
|
apply uses, so the two halves cannot drift (rule 93).
|
|
|
|
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_conds = _artist_images_conditions(artist.id)
|
|
|
|
images_count = session.execute(
|
|
select(func.count(ImageRecord.id)).where(*images_conds)
|
|
).scalar_one()
|
|
posts_count = session.execute(
|
|
select(func.count(Post.id))
|
|
.where(*_artist_posts_conditions(artist.id))
|
|
).scalar_one()
|
|
attachments_count = session.execute(
|
|
select(func.count(PostAttachment.id))
|
|
.where(*_artist_attachments_conditions(artist.id))
|
|
).scalar_one()
|
|
# Sources have no shared predicate: the apply never queries them either, it
|
|
# gets them from the Artist.sources ORM cascade. Counted directly here.
|
|
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(*images_conds)
|
|
.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(*images_conds)
|
|
)
|
|
)
|
|
).scalar_one()
|
|
bytes_on_disk = session.execute(
|
|
select(func.coalesce(func.sum(ImageRecord.size_bytes), 0))
|
|
.where(*images_conds)
|
|
).scalar_one()
|
|
|
|
return {
|
|
"artist": {"id": artist.id, "name": artist.name, "slug": artist.slug},
|
|
"projected": {
|
|
"images": images_count,
|
|
"posts": posts_count,
|
|
"attachments": attachments_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:
|
|
"""Images affected by deleting this tag — the Tier-B blast-radius prompt.
|
|
Mirrors the gallery/directory membership predicate: images carrying the tag
|
|
DIRECTLY, plus — when it's a fandom — images carrying one of its characters
|
|
(member.fandom_id == tag_id). DISTINCT so each image counts once. Without
|
|
the character leg a fandom would report 0 here yet its delete still strips
|
|
the fandom off every character, badly understating the prompt."""
|
|
member = aliased(Tag)
|
|
return session.execute(
|
|
select(func.count(image_tag.c.image_record_id.distinct())).where(
|
|
or_(
|
|
image_tag.c.tag_id == tag_id,
|
|
image_tag.c.tag_id.in_(
|
|
select(member.id).where(member.fandom_id == tag_id)
|
|
),
|
|
)
|
|
)
|
|
).scalar_one()
|
|
|
|
|
|
def _unused_tag_conditions() -> list:
|
|
"""The WHERE conditions that define an 'unused' tag — the SINGLE source of
|
|
truth shared by find_unused_tags (preview sample), the dry-run count, AND the
|
|
live delete, so the preview can NEVER diverge from what the delete removes.
|
|
|
|
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)
|
|
- 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 looks "unused", and the FK is ondelete=SET NULL, so deleting one
|
|
silently strips the fandom off all its characters. The delete had only the
|
|
first two checks while the preview sample had all four, so the preview showed
|
|
a safe list but the delete removed every fandom anyway (operator-flagged
|
|
2026-06-08). Defining the predicate once makes that impossible.
|
|
"""
|
|
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()
|
|
return [
|
|
Tag.id.not_in(used_via_image_tag),
|
|
Tag.id.not_in(used_via_series),
|
|
Tag.id.not_in(used_via_chapter),
|
|
Tag.id.not_in(used_via_fandom),
|
|
# System tags (#128) ship with zero applications and must survive a
|
|
# prune — the training-hygiene machinery keys on the rows.
|
|
Tag.is_system.is_(False),
|
|
]
|
|
|
|
|
|
def find_unused_tags(
|
|
session: Session, *, limit: int | None = None,
|
|
) -> list[Tag]:
|
|
"""Tags genuinely referenced by nothing — safe to sweep. Sorted by name.
|
|
Shares its predicate with the live prune via _unused_tag_conditions()."""
|
|
stmt = select(Tag).where(*_unused_tag_conditions()).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").
|
|
|
|
The artist's post_attachment rows are cleared EXPLICITLY before the
|
|
artist row goes — see the comment at that step; leaving them to the
|
|
cascade aborts the whole delete on a unique violation.
|
|
"""
|
|
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,
|
|
"posts_deleted": 0,
|
|
"attachments_deleted": 0,
|
|
"files_failed": 0,
|
|
},
|
|
}
|
|
artist_info = {"id": artist.id, "name": artist.name, "slug": artist.slug}
|
|
|
|
# Counted BEFORE the delete: Postgres takes these via the post.artist_id
|
|
# CASCADE when the artist row goes, so afterwards there is nothing left to
|
|
# count. Reported so the summary can be checked against the preview's
|
|
# `posts` — the parity rule 93 asks for is only testable if both halves
|
|
# actually state the number.
|
|
posts_deleted = session.execute(
|
|
select(func.count(Post.id)).where(*_artist_posts_conditions(artist.id))
|
|
).scalar_one()
|
|
|
|
images_deleted = 0
|
|
files_deleted = 0
|
|
thumbs_deleted = 0
|
|
files_failed = 0
|
|
|
|
while True:
|
|
rows = session.execute(
|
|
select(ImageRecord)
|
|
.where(*_artist_images_conditions(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
|
|
|
|
# Clear the artist's attachments BEFORE the artist row, or the delete below
|
|
# aborts. Deleting an artist CASCADEs to Post (post.artist_id is
|
|
# ondelete=CASCADE), which SET NULLs post_attachment.post_id — and
|
|
# `uq_post_attachment_null_post_sha` is a partial UNIQUE on sha256 ALONE
|
|
# WHERE post_id IS NULL, so any two of this artist's attachments sharing a
|
|
# sha collapse onto one another and raise. That is an ORDINARY shape, not a
|
|
# corrupt one: _capture_attachment deliberately writes one row per post over
|
|
# a single sha-addressed blob (a creator who attaches the same pdf to two
|
|
# posts has two rows), and a pre-existing filesystem-import row with the same
|
|
# sha and a NULL post_id collides on its own. Migration 0043 reasoned only
|
|
# about upgrade-time safety and never about this later SET NULL.
|
|
# _repoint_post_links guards the identical collision class in the reconcile
|
|
# path; this is its artist-cascade counterpart.
|
|
#
|
|
# Which rows count as the artist's — and why the blobs are left on disk —
|
|
# is _artist_attachments_conditions, shared with the preview.
|
|
attachments_deleted = session.execute(
|
|
delete(PostAttachment)
|
|
.where(*_artist_attachments_conditions(artist.id))
|
|
).rowcount or 0
|
|
session.commit()
|
|
|
|
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,
|
|
"posts_deleted": posts_deleted,
|
|
"attachments_deleted": attachments_deleted,
|
|
"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_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}")
|
|
if tag.is_system:
|
|
raise ValueError(f"'{tag.name}' is a system tag and cannot be deleted")
|
|
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 runs a single DELETE
|
|
with the SAME predicate (_unused_tag_conditions) the preview uses, so
|
|
the row count scales without binding every id as a parameter AND the
|
|
delete can never remove a tag the preview deemed safe. Audit 2026-06-02;
|
|
predicate unified 2026-06-08.
|
|
"""
|
|
conditions = _unused_tag_conditions()
|
|
sample_rows = find_unused_tags(session, limit=50)
|
|
sample = [t.name for t in sample_rows]
|
|
if dry_run:
|
|
count = session.execute(
|
|
select(func.count()).select_from(Tag).where(*conditions)
|
|
).scalar_one()
|
|
return {"count": count, "sample_names": sample}
|
|
result = session.execute(Tag.__table__.delete().where(*conditions))
|
|
session.commit()
|
|
return {"deleted": result.rowcount or 0, "sample_names": sample}
|
|
|
|
|
|
def _bare_post_conditions() -> list:
|
|
"""The WHERE conditions that define a 'bare' post — the SINGLE source of truth
|
|
shared by find_bare_posts (preview sample), the dry-run count, AND the live
|
|
delete, so the preview can NEVER diverge from what the delete removes
|
|
([[feedback_preview_apply_parity]]).
|
|
|
|
A post is "bare" iff NOTHING is attached to it across every way content links
|
|
to a post:
|
|
- image_record.primary_post_id (a post's own canonical images)
|
|
- image_provenance.post_id (cross-posted/duplicate images linked here)
|
|
- post_attachment.post_id (preserved non-art files)
|
|
|
|
These are exactly the shells the empty-post flood produced: the native Patreon
|
|
ingester synthesized a Post per walked post, but when its only content was a
|
|
duplicate image/attachment that linked to an EARLIER post, the new post was
|
|
left with none of the three (operator-flagged 2026-06-08). Every FK to post.id
|
|
is SET NULL or CASCADE, so deleting a bare post is non-destructive by
|
|
construction — there is nothing pointing at it to orphan. Must run AFTER the
|
|
provenance-render fix so a post that DOES have a hidden provenance link is
|
|
spared, not deleted.
|
|
"""
|
|
has_primary = select(ImageRecord.id).where(
|
|
ImageRecord.primary_post_id == Post.id
|
|
)
|
|
has_provenance = select(ImageProvenance.id).where(
|
|
ImageProvenance.post_id == Post.id
|
|
)
|
|
has_attachment = select(PostAttachment.id).where(
|
|
PostAttachment.post_id == Post.id
|
|
)
|
|
return [
|
|
~has_primary.exists(),
|
|
~has_provenance.exists(),
|
|
~has_attachment.exists(),
|
|
]
|
|
|
|
|
|
def find_bare_posts(
|
|
session: Session, *, limit: int | None = None,
|
|
) -> list[Post]:
|
|
"""Posts with zero linked images (primary OR provenance) AND zero
|
|
attachments — safe to sweep. Sorted by id. Shares its predicate with the
|
|
live prune via _bare_post_conditions()."""
|
|
stmt = select(Post).where(*_bare_post_conditions()).order_by(Post.id)
|
|
if limit is not None:
|
|
stmt = stmt.limit(limit)
|
|
return list(session.execute(stmt).scalars().all())
|
|
|
|
|
|
def _bare_post_label(post: Post) -> str:
|
|
"""Human label for the preview sample — mirrors the feed's fallback title."""
|
|
if post.post_title:
|
|
return post.post_title
|
|
return f"Post {post.external_post_id or post.id}"
|
|
|
|
|
|
def prune_bare_posts(session: Session, *, dry_run: bool = False) -> dict:
|
|
"""Find posts with no images and no attachments 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]}
|
|
|
|
The live delete runs a single DELETE with the SAME predicate
|
|
(_bare_post_conditions) the preview uses, so the row count scales without
|
|
binding every id as a parameter ([[reference_psycopg_65535_param_ceiling]])
|
|
AND the delete can never remove a post the preview deemed kept.
|
|
"""
|
|
conditions = _bare_post_conditions()
|
|
sample_rows = find_bare_posts(session, limit=50)
|
|
sample = [_bare_post_label(p) for p in sample_rows]
|
|
if dry_run:
|
|
count = session.execute(
|
|
select(func.count()).select_from(Post).where(*conditions)
|
|
).scalar_one()
|
|
return {"count": count, "sample_names": sample}
|
|
result = session.execute(Post.__table__.delete().where(*conditions))
|
|
session.commit()
|
|
return {"deleted": result.rowcount or 0, "sample_names": sample}
|
|
|
|
|
|
# -- duplicate-post reconciliation (gallery-dl → native migration) ----------
|
|
# An artist first downloaded by gallery-dl gets Post rows keyed by the per-
|
|
# ATTACHMENT id (gallery-dl's `id`); a later native walk keys the SAME real post
|
|
# by the post id. The two never dedup (uq_post_source_external_id is on
|
|
# external_post_id) → duplicate post rows. The real post id is recoverable in-DB
|
|
# from raw_metadata["post_id"] (both eras store the sidecar there). We unify each
|
|
# group onto ONE post row keyed the way the CURRENT native downloader keys it
|
|
# (post id), so future native walks match and the dup can't recur. Images are
|
|
# untouched (content-addressed/deduped already); only post rows + their link
|
|
# rows move. Milestone #73 / note #917.
|
|
|
|
|
|
def _canonical_post_id(post: Post) -> str | None:
|
|
"""The real platform post id used to group duplicate rows: raw_metadata
|
|
['post_id'] when present (the true id, stored by both gallery-dl and native
|
|
imports), else external_post_id. None when neither is usable."""
|
|
rm = post.raw_metadata or {}
|
|
pid = rm.get("post_id")
|
|
if pid is not None and str(pid).strip():
|
|
return str(pid)
|
|
return post.external_post_id or None
|
|
|
|
|
|
def find_duplicate_post_groups(
|
|
session: Session, *, source_id: int | None = None,
|
|
) -> list[list[Post]]:
|
|
"""Groups of >1 Post that are the SAME real post (same source_id + canonical
|
|
post id) — the gallery-dl(attachment-id) / native(post-id) duplicates. Shared
|
|
by the dry-run preview and the live reconcile (preview/apply parity)."""
|
|
stmt = select(Post)
|
|
if source_id is not None:
|
|
stmt = stmt.where(Post.source_id == source_id)
|
|
groups: dict[tuple, list[Post]] = {}
|
|
for post in session.execute(stmt.order_by(Post.id)).scalars().all():
|
|
cpid = _canonical_post_id(post)
|
|
if not cpid:
|
|
continue
|
|
groups.setdefault((post.source_id, cpid), []).append(post)
|
|
return [posts for posts in groups.values() if len(posts) > 1]
|
|
|
|
|
|
def _choose_keeper(posts: list[Post], cpid: str) -> Post:
|
|
"""The surviving row for a dup group: prefer one already keyed by the
|
|
canonical post id (the native format we keep), then the most complete
|
|
(has description, has date), then the lowest id for stability."""
|
|
native = [p for p in posts if p.external_post_id == cpid]
|
|
pool = native or posts
|
|
return min(pool, key=lambda p: (not p.description, not p.post_date, p.id))
|
|
|
|
|
|
def _repoint_post_links(session: Session, loser_id: int, keeper_id: int) -> None:
|
|
"""Move every link row from a loser post to the keeper, conflict-safe against
|
|
each table's uniqueness (drop the loser's row when the keeper already has the
|
|
equivalent, else re-point). Images themselves are never touched."""
|
|
# ImageRecord.primary_post_id — no uniqueness; straight re-point.
|
|
session.execute(
|
|
update(ImageRecord)
|
|
.where(ImageRecord.primary_post_id == loser_id)
|
|
.values(primary_post_id=keeper_id)
|
|
)
|
|
# ImageProvenance — unique (image_record_id, post_id).
|
|
dup_imgs = select(ImageProvenance.image_record_id).where(
|
|
ImageProvenance.post_id == keeper_id
|
|
)
|
|
# Before dropping the colliding loser rows, carry their from_attachment_id
|
|
# (which archive the file came out of, milestone #87) onto the keeper's
|
|
# surviving row when the keeper didn't record one. For the gallery-dl→native
|
|
# case this very milestone targets, the keeper is the native stub (no
|
|
# archive) and the loser is the gallery-dl row that extracted the member, so
|
|
# a blind delete would silently lose the containing-archive linkage.
|
|
for img_id, att_id in session.execute(
|
|
select(ImageProvenance.image_record_id, ImageProvenance.from_attachment_id)
|
|
.where(
|
|
ImageProvenance.post_id == loser_id,
|
|
ImageProvenance.image_record_id.in_(dup_imgs),
|
|
ImageProvenance.from_attachment_id.is_not(None),
|
|
)
|
|
).all():
|
|
session.execute(
|
|
update(ImageProvenance)
|
|
.where(
|
|
ImageProvenance.post_id == keeper_id,
|
|
ImageProvenance.image_record_id == img_id,
|
|
ImageProvenance.from_attachment_id.is_(None),
|
|
)
|
|
.values(from_attachment_id=att_id)
|
|
)
|
|
session.execute(
|
|
delete(ImageProvenance).where(
|
|
ImageProvenance.post_id == loser_id,
|
|
ImageProvenance.image_record_id.in_(dup_imgs),
|
|
)
|
|
)
|
|
session.execute(
|
|
update(ImageProvenance)
|
|
.where(ImageProvenance.post_id == loser_id)
|
|
.values(post_id=keeper_id)
|
|
)
|
|
# PostAttachment — partial unique (post_id, sha256) where post_id NOT NULL.
|
|
dup_shas = select(PostAttachment.sha256).where(PostAttachment.post_id == keeper_id)
|
|
session.execute(
|
|
delete(PostAttachment).where(
|
|
PostAttachment.post_id == loser_id,
|
|
PostAttachment.sha256.in_(dup_shas),
|
|
)
|
|
)
|
|
session.execute(
|
|
update(PostAttachment)
|
|
.where(PostAttachment.post_id == loser_id)
|
|
.values(post_id=keeper_id)
|
|
)
|
|
# ExternalLink — unique (post_id, url).
|
|
dup_urls = select(ExternalLink.url).where(ExternalLink.post_id == keeper_id)
|
|
session.execute(
|
|
delete(ExternalLink).where(
|
|
ExternalLink.post_id == loser_id,
|
|
ExternalLink.url.in_(dup_urls),
|
|
)
|
|
)
|
|
session.execute(
|
|
update(ExternalLink)
|
|
.where(ExternalLink.post_id == loser_id)
|
|
.values(post_id=keeper_id)
|
|
)
|
|
|
|
|
|
def _fill_missing_post_fields(keeper: Post, loser: Post) -> None:
|
|
"""Backfill the keeper's empty metadata from a loser that has it (the native
|
|
stub is often bare; the gallery-dl row carries date/title/body/raw_metadata)."""
|
|
if not keeper.post_date and loser.post_date:
|
|
keeper.post_date = loser.post_date
|
|
if not keeper.post_title and loser.post_title:
|
|
keeper.post_title = loser.post_title
|
|
if not keeper.description and loser.description:
|
|
keeper.description = loser.description
|
|
if not keeper.raw_metadata and loser.raw_metadata:
|
|
keeper.raw_metadata = loser.raw_metadata
|
|
if not keeper.attachment_count and loser.attachment_count:
|
|
keeper.attachment_count = loser.attachment_count
|
|
|
|
|
|
def _canonical_post_url(post: Post, cpid: str) -> str | None:
|
|
"""Permalink for the unified post, via the platform's derive_post_url hook
|
|
(subscribestar/patreon synthesize `…/posts/<id>`). Falls back to the keeper's
|
|
existing url when no hook applies."""
|
|
platform = (post.raw_metadata or {}).get("category")
|
|
info = PLATFORMS.get(platform) if platform else None
|
|
if info is not None and info.derive_post_url is not None:
|
|
derived = info.derive_post_url({"post_id": cpid})
|
|
if derived:
|
|
return derived
|
|
return post.post_url
|
|
|
|
|
|
def reconcile_duplicate_posts(
|
|
session: Session, *, source_id: int | None = None, dry_run: bool = False,
|
|
) -> dict:
|
|
"""Unify duplicate post rows (gallery-dl attachment-id + native post-id) onto
|
|
one keeper per real post, re-keyed to the post id. Images untouched.
|
|
|
|
Returns:
|
|
dry_run=True: {"groups": G, "posts_to_merge": L, "sample": [...]}
|
|
dry_run=False: {"groups": G, "merged": L, "sample": [...]}
|
|
where L = rows that would be (were) deleted after merging into keepers. The
|
|
SAME find_duplicate_post_groups predicate drives preview and apply (rule 93).
|
|
"""
|
|
groups = find_duplicate_post_groups(session, source_id=source_id)
|
|
sample: list[dict] = []
|
|
losers_total = 0
|
|
for posts in groups:
|
|
cpid = _canonical_post_id(posts[0])
|
|
keeper = _choose_keeper(posts, cpid)
|
|
losers = [p for p in posts if p.id != keeper.id]
|
|
losers_total += len(losers)
|
|
if len(sample) < 50:
|
|
sample.append({
|
|
"post_id": cpid,
|
|
"rows": len(posts),
|
|
"keeper_id": keeper.id,
|
|
"title": keeper.post_title or f"Post {cpid}",
|
|
})
|
|
if dry_run:
|
|
continue
|
|
for loser in losers:
|
|
_repoint_post_links(session, loser.id, keeper.id)
|
|
_fill_missing_post_fields(keeper, loser)
|
|
keeper.external_post_id = cpid
|
|
new_url = _canonical_post_url(keeper, cpid)
|
|
if new_url:
|
|
keeper.post_url = new_url
|
|
session.flush()
|
|
for loser in losers:
|
|
session.delete(loser)
|
|
if dry_run:
|
|
return {"groups": len(groups), "posts_to_merge": losers_total, "sample": sample}
|
|
session.commit()
|
|
return {"groups": len(groups), "merged": losers_total, "sample": sample}
|
|
|
|
|
|
# The CONTENT vocabulary. "Reset content tagging" wipes these so the operator
|
|
# can re-tag from scratch. 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. NB: the deleted applications include the tagging
|
|
heads' training positives — suggestions do NOT repopulate on their own; the
|
|
heads retrain from whatever the operator re-tags. (The API route gates the
|
|
live run behind a preview-derived confirm token for exactly this reason.)
|
|
|
|
PRESERVED: fandom + series tags and their series_page ordering, AND the
|
|
system hygiene tags (#128) WITH their applications — the reset re-tags
|
|
CONTENT concepts, while wip/banner flags describe the file itself and
|
|
re-flagging hundreds of banners by hand would be pure loss. CASCADE on
|
|
image_tag / tag_alias / 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 = and_(
|
|
Tag.kind.in_(RESETTABLE_TAG_KINDS), Tag.is_system.is_(False)
|
|
)
|
|
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 cpu_embed_enabled, embed_image
|
|
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).
|
|
do_embed = cpu_embed_enabled()
|
|
for img_id in enqueue_ids:
|
|
try:
|
|
generate_thumbnail.delay(img_id)
|
|
if do_embed:
|
|
embed_image.delay(img_id)
|
|
except Exception as exc:
|
|
log.warning("re-extract enqueue failed for image %s: %s", img_id, exc)
|
|
return summary
|
|
|
|
|
|
# ---- Tier-1 video dedup (#871) ------------------------------------------
|
|
|
|
|
|
def _aspect_matches(w, h, cw, ch) -> bool:
|
|
"""Same aspect ratio within tolerance; missing dims don't block (duration is
|
|
the primary signal). Mirrors Importer._video_aspect_matches."""
|
|
if not (w and h and cw and ch):
|
|
return True
|
|
return abs((w / h) - (cw / ch)) <= _VIDEO_DUP_ASPECT_TOL
|
|
|
|
|
|
def backfill_video_durations(session: Session) -> int:
|
|
"""Populate image_record.duration_seconds for video rows imported before #871
|
|
(NULL). Idempotent — only NULL rows are touched, so a re-run after a timeout
|
|
naturally resumes. A probe that yields no duration writes the
|
|
_VIDEO_DURATION_UNKNOWN sentinel so the file isn't re-probed forever (and can
|
|
never match a real duration). Returns the count of rows given a real duration.
|
|
"""
|
|
populated = 0
|
|
while True:
|
|
rows = session.execute(
|
|
select(ImageRecord.id, ImageRecord.path)
|
|
.where(
|
|
ImageRecord.mime.like("video/%"),
|
|
ImageRecord.duration_seconds.is_(None),
|
|
)
|
|
.order_by(ImageRecord.id)
|
|
.limit(500)
|
|
).all()
|
|
if not rows:
|
|
break
|
|
for rid, path in rows:
|
|
probe = safe_probe.probe_video(Path(path))
|
|
dur = probe.duration if probe.ok and probe.duration else None
|
|
session.execute(
|
|
update(ImageRecord)
|
|
.where(ImageRecord.id == rid)
|
|
.values(
|
|
duration_seconds=dur if dur is not None else _VIDEO_DURATION_UNKNOWN
|
|
)
|
|
)
|
|
if dur is not None:
|
|
populated += 1
|
|
session.commit()
|
|
return populated
|
|
|
|
|
|
def _video_dup_group(members: list) -> dict:
|
|
"""Pick the keeper (highest pixel area, then largest bytes, then lowest id for
|
|
stability) and describe the group."""
|
|
keeper = max(
|
|
members,
|
|
key=lambda m: ((m.width or 0) * (m.height or 0), m.size_bytes or 0, -m.id),
|
|
)
|
|
losers = [m for m in members if m.id != keeper.id]
|
|
return {
|
|
"artist_id": keeper.artist_id,
|
|
"keeper_id": keeper.id,
|
|
"loser_ids": [m.id for m in losers],
|
|
"duration": keeper.duration_seconds,
|
|
"count": len(members),
|
|
"reclaim_bytes": sum((m.size_bytes or 0) for m in losers),
|
|
}
|
|
|
|
|
|
def find_video_dup_groups(session: Session) -> list[dict]:
|
|
"""Cluster videos that are the same content (#871): same artist, duration
|
|
within tolerance, matching aspect ratio. Returns groups of >1 member. Greedy
|
|
sweep over duration-sorted rows, anchored to each cluster's first member so the
|
|
cluster's duration span never exceeds the tolerance (no chain drift)."""
|
|
rows = session.execute(
|
|
select(
|
|
ImageRecord.id, ImageRecord.artist_id, ImageRecord.duration_seconds,
|
|
ImageRecord.width, ImageRecord.height, ImageRecord.size_bytes,
|
|
)
|
|
.where(
|
|
ImageRecord.mime.like("video/%"),
|
|
ImageRecord.duration_seconds.is_not(None),
|
|
ImageRecord.duration_seconds > 0,
|
|
ImageRecord.artist_id.is_not(None),
|
|
)
|
|
.order_by(
|
|
ImageRecord.artist_id, ImageRecord.duration_seconds, ImageRecord.id
|
|
)
|
|
).all()
|
|
groups: list[dict] = []
|
|
cluster: list = []
|
|
anchor = None
|
|
for r in rows:
|
|
if (
|
|
anchor is not None
|
|
and r.artist_id == anchor.artist_id
|
|
and (r.duration_seconds - anchor.duration_seconds)
|
|
<= _VIDEO_DUP_DURATION_TOL_SECONDS
|
|
and _aspect_matches(r.width, r.height, anchor.width, anchor.height)
|
|
):
|
|
cluster.append(r)
|
|
else:
|
|
if len(cluster) > 1:
|
|
groups.append(_video_dup_group(cluster))
|
|
cluster = [r]
|
|
anchor = r
|
|
if len(cluster) > 1:
|
|
groups.append(_video_dup_group(cluster))
|
|
return groups
|
|
|
|
|
|
def _relink_provenance_to_keeper(
|
|
session: Session, *, loser_id: int, keeper_id: int
|
|
) -> int:
|
|
"""Ensure the keeper has an ImageProvenance row for every post the loser was
|
|
linked to, so deleting the loser never drops the video off a post. Returns the
|
|
number of new keeper↔post links added."""
|
|
rows = session.execute(
|
|
select(ImageProvenance.post_id, ImageProvenance.source_id)
|
|
.where(ImageProvenance.image_record_id == loser_id)
|
|
).all()
|
|
added = 0
|
|
for post_id, source_id in rows:
|
|
exists = session.execute(
|
|
select(ImageProvenance.id).where(
|
|
ImageProvenance.image_record_id == keeper_id,
|
|
ImageProvenance.post_id == post_id,
|
|
)
|
|
).scalar_one_or_none()
|
|
if exists is None:
|
|
session.add(ImageProvenance(
|
|
image_record_id=keeper_id, post_id=post_id, source_id=source_id,
|
|
))
|
|
session.flush()
|
|
added += 1
|
|
return added
|
|
|
|
|
|
def dedup_videos(
|
|
session: Session, *, images_root: Path, dry_run: bool = False
|
|
) -> dict:
|
|
"""Find and (unless dry_run) collapse Tier-1 video duplicates (#871).
|
|
|
|
Re-probes NULL-duration videos first so the existing library participates,
|
|
then clusters by artist + duration + aspect and keeps the highest-res copy per
|
|
cluster. On apply, each loser's post links are re-pointed to the keeper BEFORE
|
|
the loser record + file are deleted, so no post loses the video. dry_run shares
|
|
the same discovery predicate and returns the projection without deleting
|
|
(rule 93).
|
|
|
|
NOTE: tags/curation on a loser are NOT merged onto the keeper — videos rarely
|
|
carry hand-curation and merging would add FK-juggling risk. Flagged as a
|
|
follow-up if it ever matters.
|
|
"""
|
|
backfill_video_durations(session)
|
|
groups = find_video_dup_groups(session)
|
|
redundant = sum(len(g["loser_ids"]) for g in groups)
|
|
reclaim = sum(g["reclaim_bytes"] for g in groups)
|
|
sample = [
|
|
{
|
|
"keeper_id": g["keeper_id"],
|
|
"redundant": len(g["loser_ids"]),
|
|
"duration": round(g["duration"], 1),
|
|
}
|
|
for g in groups[:50]
|
|
]
|
|
if dry_run:
|
|
return {
|
|
"groups": len(groups), "redundant": redundant,
|
|
"reclaim_bytes": reclaim, "sample": sample,
|
|
}
|
|
|
|
relinked = 0
|
|
for g in groups:
|
|
for loser_id in g["loser_ids"]:
|
|
relinked += _relink_provenance_to_keeper(
|
|
session, loser_id=loser_id, keeper_id=g["keeper_id"]
|
|
)
|
|
session.commit()
|
|
|
|
loser_ids = [lid for g in groups for lid in g["loser_ids"]]
|
|
deleted = delete_images(session, image_ids=loser_ids, images_root=images_root)
|
|
log.info(
|
|
"video dedup: %d group(s), %d redundant removed, %d post link(s) re-pointed",
|
|
len(groups), deleted["images_deleted"], relinked,
|
|
)
|
|
return {
|
|
"groups": len(groups), "redundant": redundant,
|
|
"reclaim_bytes": reclaim, "deleted": deleted["images_deleted"],
|
|
"files_deleted": deleted["files_deleted"],
|
|
"relinked_posts": relinked, "sample": sample,
|
|
}
|
|
|
|
|
|
# ---- Gated-post blurred-preview cleanup (#874 follow-up) -----------------
|
|
#
|
|
# Before the #874 ingester fix, tier-gated Patreon posts (current_user_can_view
|
|
# == False) had their BLURRED locked-preview media downloaded as if real. That
|
|
# flag was never persisted (the sidecar/Post.raw_metadata store only
|
|
# category/id/title/content/url), so we can't tell from the DB which stored
|
|
# images are blurred previews. This cleanup RE-WALKS each feed to re-derive what
|
|
# Patreon serves as gated NOW, then matches by CONTENT HASH: an ImageRecord whose
|
|
# stored `source_filehash` equals a currently-served blurred file's hash IS that
|
|
# blurred preview. Because the hash is content-addressed, a real file the operator
|
|
# downloaded when they HAD access has a different hash and can never match — so
|
|
# regained-then-lost-access content is provably spared (operator's hard
|
|
# requirement, 2026-06-16). NULL source_filehash → can't verify → kept + reported.
|
|
|
|
|
|
def collect_gated_previews(client, campaign_id: str) -> dict[str, list[str]]:
|
|
"""Read-only walk of one Patreon feed → `{external_post_id: [filehashes]}` for
|
|
every GATED post, where the list is the CDN filehashes of the blurred preview
|
|
media Patreon is serving for it right now. No downloads.
|
|
|
|
`client` is a PatreonClient (or test stub) exposing the same seams the
|
|
ingester uses: `post_is_gated`, `iter_posts`, `extract_media`. extract_media
|
|
on a gated post yields exactly the blurred preview items (the original bug), so
|
|
their filehashes are the match keys the purge needs.
|
|
"""
|
|
gated: dict[str, list[str]] = {}
|
|
is_gated = getattr(client, "post_is_gated", None)
|
|
if is_gated is None:
|
|
return gated
|
|
for post, included, _cursor in client.iter_posts(campaign_id):
|
|
if not is_gated(post):
|
|
continue
|
|
pid = str(post.get("id") or "")
|
|
if not pid:
|
|
continue
|
|
hashes = sorted(
|
|
{m.filehash for m in client.extract_media(post, included) if m.filehash}
|
|
)
|
|
gated[pid] = hashes
|
|
return gated
|
|
|
|
|
|
def _gated_source_hashes(gated_map: dict) -> dict[int, set[str]]:
|
|
"""{source_id: set(blurred filehashes)} from the discovery map
|
|
{source_id: {external_post_id: [filehashes]}}."""
|
|
out: dict[int, set[str]] = {}
|
|
for sid, posts in gated_map.items():
|
|
bucket: set[str] = set()
|
|
for hashes in posts.values():
|
|
bucket.update(hashes)
|
|
out[int(sid)] = bucket
|
|
return out
|
|
|
|
|
|
def _match_gated_preview_images(
|
|
session: Session, all_hashes: set[str],
|
|
) -> dict[int, int]:
|
|
"""{image_id: size_bytes} for ImageRecords whose source_filehash is one of the
|
|
currently-served blurred filehashes — the confirmed blurred previews. Chunked
|
|
under the psycopg parameter ceiling."""
|
|
matched: dict[int, int] = {}
|
|
hash_list = list(all_hashes)
|
|
for i in range(0, len(hash_list), 500):
|
|
chunk = hash_list[i:i + 500]
|
|
for rid, size in session.execute(
|
|
select(ImageRecord.id, ImageRecord.size_bytes)
|
|
.where(ImageRecord.source_filehash.in_(chunk))
|
|
).all():
|
|
matched[rid] = size or 0
|
|
return matched
|
|
|
|
|
|
def _count_unverifiable_gated_images(session: Session, gated_map: dict) -> int:
|
|
"""Images linked (via provenance) to a gated post but with a NULL
|
|
source_filehash — we can't prove they're blurred previews, so they're KEPT and
|
|
only reported. Counts distinct images so one shared across posts counts once."""
|
|
image_ids: set[int] = set()
|
|
for sid, posts in gated_map.items():
|
|
ext_ids = list(posts.keys())
|
|
if not ext_ids:
|
|
continue
|
|
post_ids = session.execute(
|
|
select(Post.id).where(
|
|
Post.source_id == int(sid), Post.external_post_id.in_(ext_ids)
|
|
)
|
|
).scalars().all()
|
|
if not post_ids:
|
|
continue
|
|
rows = session.execute(
|
|
select(ImageProvenance.image_record_id)
|
|
.join(ImageRecord, ImageRecord.id == ImageProvenance.image_record_id)
|
|
.where(
|
|
ImageProvenance.post_id.in_(list(post_ids)),
|
|
ImageRecord.source_filehash.is_(None),
|
|
)
|
|
).scalars().all()
|
|
image_ids.update(rows)
|
|
return len(image_ids)
|
|
|
|
|
|
def purge_gated_previews(
|
|
session: Session, *, gated_map: dict, images_root: Path, dry_run: bool = False,
|
|
) -> dict:
|
|
"""Find and (unless dry_run) delete the blurred locked-preview images grabbed
|
|
from gated posts before the #874 fix (see module section above).
|
|
|
|
`gated_map` is the discovery output `{source_id: {external_post_id:
|
|
[filehashes]}}` (assembled by collect_gated_previews per source). Matching is
|
|
by content hash, so real content is provably spared. dry_run shares the exact
|
|
same match predicate and returns the projection without deleting (rule 93).
|
|
|
|
On apply: delete the matched ImageRecords + files (provenance cascades), clear
|
|
the seen/dead-letter ledger rows for those blurred hashes so the REAL media
|
|
re-ingests if access is later regained, and delete the gated post records that
|
|
are left bare.
|
|
"""
|
|
per_source = _gated_source_hashes(gated_map)
|
|
all_hashes: set[str] = set()
|
|
for bucket in per_source.values():
|
|
all_hashes |= bucket
|
|
gated_posts = sum(len(posts) for posts in gated_map.values())
|
|
|
|
matched = _match_gated_preview_images(session, all_hashes)
|
|
reclaim = sum(matched.values())
|
|
unverifiable = _count_unverifiable_gated_images(session, gated_map)
|
|
sample = [{"image_id": rid} for rid in sorted(matched)[:50]]
|
|
|
|
projection = {
|
|
"gated_posts": gated_posts,
|
|
"matched": len(matched),
|
|
"reclaim_bytes": reclaim,
|
|
"unverifiable": unverifiable,
|
|
"sample": sample,
|
|
}
|
|
if dry_run:
|
|
return projection
|
|
|
|
deleted = delete_images(
|
|
session, image_ids=list(matched), images_root=images_root,
|
|
)
|
|
|
|
# Clear the seen + dead-letter ledger for the blurred hashes so a later
|
|
# recovery/backfill re-ingests the REAL media if access is regained.
|
|
ledger_cleared = 0
|
|
for sid, hashes in per_source.items():
|
|
hl = list(hashes)
|
|
for i in range(0, len(hl), 500):
|
|
chunk = hl[i:i + 500]
|
|
res = session.execute(
|
|
delete(PatreonSeenMedia).where(
|
|
PatreonSeenMedia.source_id == sid,
|
|
PatreonSeenMedia.filehash.in_(chunk),
|
|
)
|
|
)
|
|
ledger_cleared += res.rowcount or 0
|
|
session.execute(
|
|
delete(PatreonFailedMedia).where(
|
|
PatreonFailedMedia.source_id == sid,
|
|
PatreonFailedMedia.filehash.in_(chunk),
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
# Delete gated post records left bare by the deletion (shares the bare-post
|
|
# predicate so a gated post that still has real content is never removed).
|
|
bare = _bare_post_conditions()
|
|
posts_deleted = 0
|
|
for sid, posts in gated_map.items():
|
|
ext_ids = list(posts.keys())
|
|
for i in range(0, len(ext_ids), 500):
|
|
chunk = ext_ids[i:i + 500]
|
|
res = session.execute(
|
|
Post.__table__.delete().where(
|
|
Post.source_id == int(sid),
|
|
Post.external_post_id.in_(chunk),
|
|
*bare,
|
|
)
|
|
)
|
|
posts_deleted += res.rowcount or 0
|
|
session.commit()
|
|
|
|
log.info(
|
|
"gated-preview purge: %d gated post(s), %d blurred image(s) deleted, "
|
|
"%d ledger row(s) cleared, %d bare post(s) removed, %d unverifiable kept",
|
|
gated_posts, deleted["images_deleted"], ledger_cleared, posts_deleted,
|
|
unverifiable,
|
|
)
|
|
return {
|
|
**projection,
|
|
"deleted": deleted["images_deleted"],
|
|
"files_deleted": deleted["files_deleted"],
|
|
"ledger_cleared": ledger_cleared,
|
|
"posts_deleted": posts_deleted,
|
|
}
|
|
|
|
|
|
# -- orphaned attachment reclamation ---------------------------------------
|
|
# PostAttachment's two FKs are both ON DELETE SET NULL, so a deleted post or
|
|
# artist leaves the row behind rather than taking it. Nothing ever pruned those
|
|
# rows, and nothing has ever unlinked a file under the attachment store — so
|
|
# both rows and bytes accumulated permanently and were invisible to every
|
|
# existing diagnostic.
|
|
#
|
|
# Why this is a DISK->DB reconciliation rather than a row sweep: the store is
|
|
# sha-addressed and idempotent (attachment_store.store), so ONE blob backs MANY
|
|
# rows. Deleting a row therefore does not free its blob, and — since the artist
|
|
# cascade now deletes its attachment rows outright — a freed blob has no DB
|
|
# pointer left to find it by. Walking the store and asking "does any row still
|
|
# reference this sha?" catches orphans from every cause, including ones no
|
|
# future delete path will think to report.
|
|
|
|
# A blob is written by attachment_store.store BEFORE its row is inserted and
|
|
# committed, so a just-stored file legitimately has no referencing row for a
|
|
# moment. Same guard, same reasoning as ORPHAN_TEMP_MIN_AGE_HOURS in
|
|
# tasks/maintenance.py: never judge a file younger than this.
|
|
_ATTACHMENT_ORPHAN_MIN_AGE_HOURS = 6
|
|
|
|
# Wall-clock budget for the store walk (rule 89). A library with a large
|
|
# attachment store shouldn't be able to run this past its soft time limit; on
|
|
# exhaustion it reports partial=True and the operator re-runs to finish.
|
|
_ATTACHMENT_RECLAIM_BUDGET_SECONDS = 900
|
|
|
|
# The store names files `<sha256><ext>`. Parse the sha as the first 64 chars
|
|
# rather than via Path.stem: store() takes the extension straight from the
|
|
# source filename, and a URL-encoded basename yields a multi-dot "suffix"
|
|
# (see [[reference_url_encoded_basename_suffix]]) that would make stem eat part
|
|
# of the sha. Validating the 64 chars as hex also skips anything else in the
|
|
# tree that isn't a stored blob.
|
|
_SHA256_HEX_LEN = 64
|
|
|
|
|
|
def _orphan_attachment_conditions() -> list:
|
|
"""PostAttachment rows belonging to nothing: both FKs nulled by a deleted
|
|
post AND a deleted artist. A row with post_id NULL but an artist_id is the
|
|
deliberate filesystem-import case (importer._capture_attachment writes it
|
|
that way) and is NOT an orphan — it is still attributed."""
|
|
return [
|
|
PostAttachment.post_id.is_(None),
|
|
PostAttachment.artist_id.is_(None),
|
|
]
|
|
|
|
|
|
def _is_sha_named(name: str) -> bool:
|
|
"""True when `name` starts with a 64-char lowercase-hex sha256."""
|
|
if len(name) < _SHA256_HEX_LEN:
|
|
return False
|
|
head = name[:_SHA256_HEX_LEN]
|
|
return all(c in "0123456789abcdef" for c in head)
|
|
|
|
|
|
def reclaim_orphaned_attachments(
|
|
session: Session, *, images_root: Path, dry_run: bool = False,
|
|
) -> dict:
|
|
"""Prune unattributed PostAttachment rows, then unlink store blobs that no
|
|
surviving row references.
|
|
|
|
Returns (same discovery keys either way, so the UI renders one shape):
|
|
{"rows": int, # orphan rows found / deleted
|
|
"files": int, # unreferenced blobs found / unlinked
|
|
"bytes": int, # their total size
|
|
"scanned": int, # blobs examined
|
|
"skipped_recent": int, # blobs under the min-age guard
|
|
"files_failed": int, # unlink raised (apply only)
|
|
"partial": bool} # walk hit the time budget
|
|
|
|
dry_run computes exactly what the apply would do and mutates nothing — the
|
|
surviving-sha set is derived by NEGATING the same orphan predicate the
|
|
delete uses, so the preview cannot disagree with the apply (rule 93).
|
|
"""
|
|
started = time.monotonic()
|
|
orphan_conds = _orphan_attachment_conditions()
|
|
|
|
if dry_run:
|
|
rows = session.execute(
|
|
select(func.count(PostAttachment.id)).where(*orphan_conds)
|
|
).scalar_one()
|
|
else:
|
|
rows = session.execute(
|
|
delete(PostAttachment).where(*orphan_conds)
|
|
).rowcount or 0
|
|
session.commit()
|
|
|
|
# Shas that still have a home. In the apply path the orphan rows are already
|
|
# gone, so `NOT orphan` is redundant but harmless; in the dry-run path it is
|
|
# what makes the projection honest about blobs the delete would free. One
|
|
# predicate, one query, both modes.
|
|
surviving_shas = set(session.execute(
|
|
select(PostAttachment.sha256).where(~and_(*orphan_conds)).distinct()
|
|
).scalars())
|
|
|
|
root = Path(images_root) / "attachments"
|
|
cutoff = (
|
|
datetime.now(UTC).timestamp()
|
|
- _ATTACHMENT_ORPHAN_MIN_AGE_HOURS * 3600
|
|
)
|
|
files = 0
|
|
freed_bytes = 0
|
|
scanned = 0
|
|
skipped_recent = 0
|
|
files_failed = 0
|
|
partial = False
|
|
|
|
if root.is_dir():
|
|
for path in root.rglob("*"):
|
|
if time.monotonic() - started >= _ATTACHMENT_RECLAIM_BUDGET_SECONDS:
|
|
partial = True
|
|
break
|
|
# .partial staging files belong to cleanup_orphaned_temp_files —
|
|
# leave them alone rather than racing an in-flight store().
|
|
if path.suffix in (".part", ".partial") or not path.is_file():
|
|
continue
|
|
if not _is_sha_named(path.name):
|
|
continue
|
|
scanned += 1
|
|
sha = path.name[:_SHA256_HEX_LEN]
|
|
if sha in surviving_shas:
|
|
continue
|
|
try:
|
|
st = path.stat()
|
|
if st.st_mtime >= cutoff:
|
|
skipped_recent += 1
|
|
continue
|
|
size = st.st_size
|
|
if not dry_run:
|
|
path.unlink()
|
|
files += 1
|
|
freed_bytes += size
|
|
except OSError as exc:
|
|
files_failed += 1
|
|
log.warning("reclaim_orphaned_attachments: %s: %s", path, exc)
|
|
|
|
if not dry_run and (rows or files):
|
|
log.info(
|
|
"attachment reclaim: %d orphan row(s) deleted, %d blob(s) unlinked "
|
|
"(%d bytes), %d failed, partial=%s",
|
|
rows, files, freed_bytes, files_failed, partial,
|
|
)
|
|
return {
|
|
"rows": rows,
|
|
"files": files,
|
|
"bytes": freed_bytes,
|
|
"scanned": scanned,
|
|
"skipped_recent": skipped_recent,
|
|
"files_failed": files_failed,
|
|
"partial": partial,
|
|
}
|