Compare commits

...

10 Commits

Author SHA1 Message Date
bvandeusen 2879ac6f2b Merge pull request 'v26.05.25.1: maintenance sweep + Camie v2 + corrupt-file handling + post-date gallery + clear-stuck escape hatch' (#11) from dev into main 2026-05-25 12:57:46 -04:00
bvandeusen 3a359f6c5e fix(ui): Quick scan button always visible (disabled when active batch present) + inline Clear stuck action — operator was clicking a spinner area thinking it was the button because activeBatch hid the Quick scan button entirely
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-25 12:43:37 -04:00
bvandeusen b6a917ac81 feat(import): /api/import/clear-stuck endpoint + Clear stuck UI button — escape hatch for the autoretry-loop case the automatic sweep can't break
Operator hit 3 large PNGs stuck in 'processing' for 2 days 2026-05-25:
the existing recover_interrupted_tasks flips processing > 5min back to
queued + .delay(), but if the underlying file is unfixably broken (e.g.,
PIL OSError, also patched in 68cffce), the loop never terminates and the
'Scanning...' banner sticks at 0/0 forever blocking new scans.

/api/import/clear-stuck:
- Flips every task in pending/queued/processing to 'failed' with a clear
  marker error message
- Finalizes any 'running' ImportBatch that has no remaining active children
- Idempotent + non-destructive: rows survive, can be retried once the
  underlying cause is resolved

UI button 'Clear stuck...' sits next to 'Retry failed' / 'Clear completed'
with a warning-tonal alert in the confirm dialog explaining what it does
and recommending Retry failed once the cause is fixed.

Tests: clears mixed non-terminal states, untouches complete rows,
finalizes orphan batch, no-op when nothing stuck.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-25 12:37:07 -04:00
bvandeusen c361032554 feat(gallery): sort/group/jump by COALESCE(post.post_date, image_record.created_at) — surface migrated content at its original publish date, not FC scan date
Operator hit this 2026-05-25 after the IR tag_apply landed: ~57k images
all scanned into FC in the same week share image_record.created_at, so
the gallery timeline collapses them into a single month bucket and
scroll orders them all together at the top. Their actual publish dates
(spread over years) were already available in Post.post_date but the
gallery never read it.

Backend wire-up:
- tag_apply phase 4 now sets ImageRecord.primary_post_id when creating
  ImageProvenance (only if currently NULL — preserves the canonical
  download-time linkage set by the importer for new FC ingests).
- gallery_service.py introduces _effective_date_col() =
  COALESCE(post.post_date, image_record.created_at), used in:
    * scroll() ORDER BY + cursor WHERE clauses
    * timeline() year/month group-by
    * jump_cursor() year/month filter
    * _neighbors() prev/next ordering
- Each method LEFT OUTER JOIN Post on primary_post_id so the COALESCE
  works for images without a post (NULL on the Post side, fall back
  to created_at).
- GalleryImage gains posted_at + effective_date fields; API /gallery
  /scroll exposes both alongside the existing created_at so the UI
  can render 'Posted on X (imported Y)' if desired.
- get_image_with_tags() returns posted_at for the modal.

Cursor format unchanged — the encoded datetime is now the effective_
date (whichever column won the COALESCE) and pagination remains
consistent.

To pick up new behavior for an already-migrated IR set: re-run
/api/migrate/tag_apply on the existing manifest (phase 4 is
idempotent; the new primary_post_id assignment backfills).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-25 12:30:18 -04:00
bvandeusen 9f54efdedf fix(migrators): tag_apply phase 4 now covers deviantart + pixiv (was silently dropping IR PostMetadata from those platforms)
_PLATFORM_PROFILE_URL had only patreon/subscribestar/hentaifoundry but
FC's extension_service.py recognizes 5 platforms. Any IR PostMetadata
with platform=deviantart or pixiv fell through _profile_url returning
None and the entry was silently skipped — explaining operator's
2026-05-25 finding that IR-migrated images had tags but no provenance
for the deviantart + pixiv subscriptions.

Pixiv caveat noted in comment: real profile URL takes numeric user_id
(https://www.pixiv.net/users/12345) but IR's PostMetadata.artist
stores display name. We slug the name and use it as if it were the id
so the artist->post->image linkage survives migration; the resulting
Source.url won't resolve in a browser and operator can fix via
Settings -> Subscriptions later if they want.

To recover existing IR-migrated state: re-run /api/migrate/tag_apply
on the existing manifest. Phase 4 is idempotent; new posts get
inserted only for the previously-skipped platforms.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-25 11:25:45 -04:00
bvandeusen 6b1bb87647 fix(tests): update test_ensure_camie_skips_when_present to v2 filenames (camie-tagger-v2.onnx + camie-tagger-v2-metadata.json) — pinned-test bounce from 3b3e756
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-25 11:12:55 -04:00
bvandeusen 68cffce322 fix(importer): catch PIL OSError during transparency + phash blocks, skip as invalid_image instead of letting Celery autoretry loop forever
Operator hit a corrupt JPEG in the IR set 2026-05-25: PIL.verify() only
validates header structure but doesn't catch truncated/broken pixel
data. The error surfaces later in _transparency_pct (via getchannel
'A' -> load) or compute_phash (load) — both blow up with OSError
'broken data stream when reading image file'. Celery's autoretry_for
then bounces the same file forever instead of marking it skipped.

Wrap both PIL.load-triggering call sites with try/except OSError ->
ImportResult(status=skipped, skip_reason=invalid_image).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-25 02:34:36 -04:00
bvandeusen 52445eb501 fix(ui): double directory card width (220px -> 440px) + bound preview slot height (min 150 / max 220 / overflow hidden / explicit display+object-position) so tall source images can't escape on browsers that don't compute aspect-ratio
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-25 02:32:05 -04:00
bvandeusen 3b3e7565fb fix(ml): align tagger + downloader with Camie v2 actual layout (model.onnx -> camie-tagger-v2.onnx + JSON metadata + ImageNet preprocessing + sigmoid on refined output)
The HF repo Camais03/camie-tagger-v2 has camie-tagger-v2.onnx (789 MB)
+ camie-tagger-v2-metadata.json (7.77 MB) at root, NOT model.onnx +
selected_tags.csv. Tags ship as nested JSON (dataset_info.tag_mapping)
not CSV. Per the published onnx_inference.py reference: input is NCHW
not NHWC, normalize with ImageNet mean/std, pad-square color (124,116,
104), sigmoid the second output (refined predictions) not the first.

Operator hit this during the IR migration ML backfill — download_models
silently fetched only 3 json files (allow_patterns matched nothing
useful), tagger.load() then raised RuntimeError. Fetched the actual
v2 layout via WebFetch, rewrote tagger to match published reference.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-25 02:25:30 -04:00
bvandeusen 9d5abb09f6 fix(maintenance): recover_interrupted_tasks also sweeps pending/queued orphans (>30 min) to failed
scan_directory creates ImportTask rows with status='pending' (commit) then
in a second pass transitions to 'queued' + .delay() (commit). Crashes in
that window leave rows orphaned with no recovery path. Operator hit 5490
such rows 2026-05-25; the existing sweep only handled 'processing'.
Flipping to 'failed' (not re-enqueue) lets the operator drain via the
existing /api/import/retry-failed endpoint at their own pace.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-25 02:00:29 -04:00
22 changed files with 973 additions and 149 deletions
+2
View File
@@ -42,6 +42,8 @@ async def scroll():
"width": i.width, "width": i.width,
"height": i.height, "height": i.height,
"created_at": i.created_at.isoformat(), "created_at": i.created_at.isoformat(),
"posted_at": i.posted_at.isoformat() if i.posted_at else None,
"effective_date": i.effective_date.isoformat(),
"thumbnail_url": i.thumbnail_url, "thumbnail_url": i.thumbnail_url,
"artist": i.artist, "artist": i.artist,
} }
+77
View File
@@ -120,6 +120,83 @@ async def retry_failed():
return jsonify({"retried": len(failed_ids)}) return jsonify({"retried": len(failed_ids)})
@import_admin_bp.route("/clear-stuck", methods=["POST"])
async def clear_stuck():
"""Force any non-terminal ImportTask (status in pending/queued/
processing) to 'failed' AND finalize any ImportBatch that ends up
with no active children. Escape hatch for the operator when the
automatic recover_interrupted_tasks sweep keeps re-queueing the
same stuck row forever (e.g., underlying file is genuinely broken
and the import keeps OSError-looping at PIL load).
Idempotent + non-destructive: rows survive as 'failed' so the
Retry-Failed button can re-attempt them once whatever was broken
is fixed. Banked 2026-05-25 — operator hit 3 large PNGs that
autoretry-looped for 2 days after a corrupt-data PIL OSError.
"""
async with get_session() as session:
stuck_ids = (
await session.execute(
select(ImportTask.id).where(
ImportTask.status.in_(["pending", "queued", "processing"])
)
)
).scalars().all()
if stuck_ids:
await session.execute(
update(ImportTask)
.where(ImportTask.id.in_(stuck_ids))
.values(
status="failed",
finished_at=datetime.now(UTC),
error=(
"manually cleared via /api/import/clear-stuck "
"— stuck in non-terminal state; retry once "
"underlying cause (corrupt file, missing model, "
"etc.) is resolved"
),
)
)
# Finalize any 'running' ImportBatch that no longer has any
# active children. The "Scanning..." banner is driven by
# /api/import/status finding a running batch; left untouched,
# it would persist forever after the stuck-task clear.
running_batches = (
await session.execute(
select(ImportBatch.id).where(ImportBatch.status == "running")
)
).scalars().all()
finalized_batches = 0
for batch_id in running_batches:
still_active = (
await session.execute(
select(ImportTask.id)
.where(ImportTask.batch_id == batch_id)
.where(ImportTask.status.in_(
["pending", "queued", "processing"]
))
.limit(1)
)
).scalar_one_or_none()
if still_active is None:
await session.execute(
update(ImportBatch)
.where(ImportBatch.id == batch_id)
.values(
status="complete",
finished_at=datetime.now(UTC),
)
)
finalized_batches += 1
await session.commit()
return jsonify({
"tasks_failed": len(stuck_ids),
"batches_finalized": finalized_batches,
})
@import_admin_bp.route("/clear-completed", methods=["POST"]) @import_admin_bp.route("/clear-completed", methods=["POST"])
async def clear_completed(): async def clear_completed():
body = await request.get_json(silent=True) or {} body = await request.get_json(silent=True) or {}
+21 -2
View File
@@ -25,12 +25,31 @@ def _snapshot(repo_id: str, dest: Path, allow_patterns: list[str] | None) -> Non
def ensure_camie() -> None: def ensure_camie() -> None:
"""Fetch Camie v2 weights + metadata.
v2 layout (HuggingFace Camais03/camie-tagger-v2): the ONNX file is
named camie-tagger-v2.onnx (not model.onnx) and tags ship inside
camie-tagger-v2-metadata.json (not selected_tags.csv). Both at root.
The repo also contains app/, game/, training/, images/ subdirs full
of setup/demo files we don't need — allow_patterns scopes the fetch
to just the inference essentials (~790 MB instead of ~2 GB).
"""
dest = MODEL_ROOT / "camie" dest = MODEL_ROOT / "camie"
if (dest / "model.onnx").is_file() and (dest / "selected_tags.csv").is_file(): model_file = dest / "camie-tagger-v2.onnx"
meta_file = dest / "camie-tagger-v2-metadata.json"
if model_file.is_file() and meta_file.is_file():
print(f"[download_models] Camie present at {dest}") print(f"[download_models] Camie present at {dest}")
return return
print(f"[download_models] Fetching {CAMIE_REPO} -> {dest}") print(f"[download_models] Fetching {CAMIE_REPO} -> {dest}")
_snapshot(CAMIE_REPO, dest, ["model.onnx", "selected_tags.csv", "*.json"]) _snapshot(
CAMIE_REPO, dest,
[
"camie-tagger-v2.onnx",
"camie-tagger-v2-metadata.json",
"config.json",
"config.yaml",
],
)
def ensure_siglip() -> None: def ensure_siglip() -> None:
+113 -54
View File
@@ -1,26 +1,34 @@
"""Cursor-paginated gallery queries. """Cursor-paginated gallery queries.
Cursor format: opaque base64-encoded "<iso8601_created_at>:<image_id>". Cursor format: opaque base64-encoded "<iso8601_effective_date>:<image_id>".
Pagination key is (created_at DESC, id DESC) so we don't drift when new
imports arrive between page loads. Decoding rejects malformed cursors with Pagination key is (effective_date DESC, id DESC) where effective_date is
a ValueError; the API layer translates that to HTTP 400. COALESCE(post.post_date, image_record.created_at) so the gallery surfaces
images by ORIGINAL publish date when known, falling back to FC's scan
date. Important for migrated content: ~57k IR images scanned in a single
week would otherwise all share the same created_at and pile up in one
month bucket. The effective_date spreads them across the years they
were originally published.
Decoding rejects malformed cursors with a ValueError; the API layer
translates that to HTTP 400.
""" """
import base64 import base64
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime from datetime import datetime
from sqlalchemy import and_, exists, func, or_, select from sqlalchemy import Select, and_, exists, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from ..models import Artist, ImageProvenance, ImageRecord, Source, Tag from ..models import Artist, ImageProvenance, ImageRecord, Post, Source, Tag
from ..models.tag import image_tag from ..models.tag import image_tag
CURSOR_SEPARATOR = "|" CURSOR_SEPARATOR = "|"
def encode_cursor(created_at: datetime, image_id: int) -> str: def encode_cursor(effective_date: datetime, image_id: int) -> str:
raw = f"{created_at.isoformat()}{CURSOR_SEPARATOR}{image_id}" raw = f"{effective_date.isoformat()}{CURSOR_SEPARATOR}{image_id}"
return base64.urlsafe_b64encode(raw.encode()).decode() return base64.urlsafe_b64encode(raw.encode()).decode()
@@ -33,6 +41,26 @@ def decode_cursor(cursor: str) -> tuple[datetime, int]:
raise ValueError(f"invalid cursor: {cursor!r}") from exc raise ValueError(f"invalid cursor: {cursor!r}") from exc
def _effective_date_col():
"""SQL expression: COALESCE(post.post_date, image_record.created_at).
Used as the canonical sort/group/filter key across the gallery so
images backfilled with primary_post_id (e.g. via tag_apply phase 4)
surface at their original publish date, not their FC import date.
Images without a Post (or with Post.post_date NULL) fall back to
image_record.created_at and still order coherently against
post-attached ones.
"""
return func.coalesce(Post.post_date, ImageRecord.created_at)
def _outer_join_primary_post(stmt: Select) -> Select:
"""LEFT JOIN Post on ImageRecord.primary_post_id so the COALESCE
above sees Post.post_date when available. Images without a post
survive the join as NULL on the Post side; COALESCE handles it."""
return stmt.outerjoin(Post, Post.id == ImageRecord.primary_post_id)
@dataclass(frozen=True) @dataclass(frozen=True)
class GalleryImage: class GalleryImage:
id: int id: int
@@ -41,7 +69,9 @@ class GalleryImage:
mime: str mime: str
width: int | None width: int | None
height: int | None height: int | None
created_at: datetime created_at: datetime # FC's row-insert time
effective_date: datetime # COALESCE(post.post_date, created_at)
posted_at: datetime | None # post.post_date if known, else None
thumbnail_url: str thumbnail_url: str
artist: dict | None = None artist: dict | None = None
@@ -78,7 +108,7 @@ def _require_single_filter(tag_id, post_id, artist_id) -> None:
def _provenance_clause(post_id, artist_id): def _provenance_clause(post_id, artist_id):
"""Correlated EXISTS clause (NOT a join) so an image with multiple """Correlated EXISTS clause (NOT a join) so an image with multiple
matching provenance rows is returned exactly once and the matching provenance rows is returned exactly once and the
(created_at DESC, id DESC) cursor ordering is unaffected.""" (effective_date DESC, id DESC) cursor ordering is unaffected."""
if post_id is not None: if post_id is not None:
return exists().where( return exists().where(
ImageProvenance.image_record_id == ImageRecord.id, ImageProvenance.image_record_id == ImageRecord.id,
@@ -125,7 +155,9 @@ class GalleryService:
raise ValueError("limit must be between 1 and 200") raise ValueError("limit must be between 1 and 200")
_require_single_filter(tag_id, post_id, artist_id) _require_single_filter(tag_id, post_id, artist_id)
stmt = select(ImageRecord) eff = _effective_date_col()
stmt = select(ImageRecord, Post.post_date, eff.label("eff"))
stmt = _outer_join_primary_post(stmt)
if tag_id is not None: if tag_id is not None:
stmt = stmt.join(image_tag, image_tag.c.image_record_id == ImageRecord.id).where( stmt = stmt.join(image_tag, image_tag.c.image_record_id == ImageRecord.id).where(
image_tag.c.tag_id == tag_id image_tag.c.tag_id == tag_id
@@ -138,34 +170,38 @@ class GalleryService:
cur_ts, cur_id = decode_cursor(cursor) cur_ts, cur_id = decode_cursor(cursor)
stmt = stmt.where( stmt = stmt.where(
or_( or_(
ImageRecord.created_at < cur_ts, eff < cur_ts,
and_(ImageRecord.created_at == cur_ts, ImageRecord.id < cur_id), and_(eff == cur_ts, ImageRecord.id < cur_id),
) )
) )
stmt = stmt.order_by(ImageRecord.created_at.desc(), ImageRecord.id.desc()).limit(limit + 1) stmt = stmt.order_by(eff.desc(), ImageRecord.id.desc()).limit(limit + 1)
rows = (await self.session.execute(stmt)).scalars().all() rows = (await self.session.execute(stmt)).all()
next_cursor = None next_cursor = None
if len(rows) > limit: if len(rows) > limit:
last = rows[limit - 1] last_record, _last_posted_at, last_eff = rows[limit - 1]
next_cursor = encode_cursor(last.created_at, last.id) next_cursor = encode_cursor(last_eff, last_record.id)
rows = rows[:limit] rows = rows[:limit]
artists = await _artists_for(self.session, [r.id for r in rows]) artists = await _artists_for(
self.session, [r[0].id for r in rows]
)
images = [ images = [
GalleryImage( GalleryImage(
id=r.id, id=record.id,
path=r.path, path=record.path,
sha256=r.sha256, sha256=record.sha256,
mime=r.mime, mime=record.mime,
width=r.width, width=record.width,
height=r.height, height=record.height,
created_at=r.created_at, created_at=record.created_at,
thumbnail_url=thumbnail_url(r.sha256, r.mime), effective_date=eff_date,
artist=artists.get(r.id), posted_at=posted_at,
thumbnail_url=thumbnail_url(record.sha256, record.mime),
artist=artists.get(record.id),
) )
for r in rows for record, posted_at, eff_date in rows
] ]
return GalleryPage( return GalleryPage(
images=images, images=images,
@@ -179,11 +215,13 @@ class GalleryService:
post_id: int | None = None, post_id: int | None = None,
artist_id: int | None = None, artist_id: int | None = None,
) -> list[TimelineBucket]: ) -> list[TimelineBucket]:
year_col = func.date_part("year", ImageRecord.created_at).label("yr") eff = _effective_date_col()
month_col = func.date_part("month", ImageRecord.created_at).label("mo") year_col = func.date_part("year", eff).label("yr")
month_col = func.date_part("month", eff).label("mo")
stmt = select( stmt = select(
year_col, month_col, func.count(ImageRecord.id).label("cnt") year_col, month_col, func.count(ImageRecord.id).label("cnt")
) )
stmt = _outer_join_primary_post(stmt)
_require_single_filter(tag_id, post_id, artist_id) _require_single_filter(tag_id, post_id, artist_id)
if tag_id is not None: if tag_id is not None:
stmt = stmt.join(image_tag, image_tag.c.image_record_id == ImageRecord.id).where( stmt = stmt.join(image_tag, image_tag.c.image_record_id == ImageRecord.id).where(
@@ -201,14 +239,17 @@ class GalleryService:
post_id: int | None = None, artist_id: int | None = None, post_id: int | None = None, artist_id: int | None = None,
) -> str | None: ) -> str | None:
"""Returns a cursor that, when passed to scroll(), positions at the """Returns a cursor that, when passed to scroll(), positions at the
first image of the given year-month. None if the bucket is empty. first image of the given year-month (by effective_date, not
created_at). None if the bucket is empty.
""" """
from sqlalchemy import extract from sqlalchemy import extract
stmt = select(ImageRecord).where( eff = _effective_date_col()
extract("year", ImageRecord.created_at) == year, stmt = select(ImageRecord, eff.label("eff")).where(
extract("month", ImageRecord.created_at) == month, extract("year", eff) == year,
extract("month", eff) == month,
) )
stmt = _outer_join_primary_post(stmt)
_require_single_filter(tag_id, post_id, artist_id) _require_single_filter(tag_id, post_id, artist_id)
if tag_id is not None: if tag_id is not None:
stmt = stmt.join(image_tag, image_tag.c.image_record_id == ImageRecord.id).where( stmt = stmt.join(image_tag, image_tag.c.image_record_id == ImageRecord.id).where(
@@ -217,13 +258,14 @@ class GalleryService:
prov = _provenance_clause(post_id, artist_id) prov = _provenance_clause(post_id, artist_id)
if prov is not None: if prov is not None:
stmt = stmt.where(prov) stmt = stmt.where(prov)
stmt = stmt.order_by(ImageRecord.created_at.desc(), ImageRecord.id.desc()).limit(1) stmt = stmt.order_by(eff.desc(), ImageRecord.id.desc()).limit(1)
first = (await self.session.execute(stmt)).scalar_one_or_none() first = (await self.session.execute(stmt)).first()
if first is None: if first is None:
return None return None
record, eff_date = first
# Cursor is exclusive; we encode a cursor with id+1 so the row itself # Cursor is exclusive; we encode a cursor with id+1 so the row itself
# is the first result in the next scroll(). # is the first result in the next scroll().
return encode_cursor(first.created_at, first.id + 1) return encode_cursor(eff_date, record.id + 1)
async def get_image_with_tags(self, image_id: int) -> dict | None: async def get_image_with_tags(self, image_id: int) -> dict | None:
record = await self.session.get(ImageRecord, image_id) record = await self.session.get(ImageRecord, image_id)
@@ -236,6 +278,13 @@ class GalleryService:
.order_by(Tag.kind.asc(), Tag.name.asc()) .order_by(Tag.kind.asc(), Tag.name.asc())
) )
tags = (await self.session.execute(tag_stmt)).scalars().all() tags = (await self.session.execute(tag_stmt)).scalars().all()
# Fetch the canonical post.post_date for this image (if any) so
# the modal can show "Posted on <date>" alongside import date.
posted_at = None
if record.primary_post_id is not None:
posted_at = (await self.session.execute(
select(Post.post_date).where(Post.id == record.primary_post_id)
)).scalar_one_or_none()
neighbors = await self._neighbors(record) neighbors = await self._neighbors(record)
# Direct artist FK — used by the modal's ProvenancePanel as a # Direct artist FK — used by the modal's ProvenancePanel as a
# fallback when ImageProvenance is empty (i.e., filesystem- # fallback when ImageProvenance is empty (i.e., filesystem-
@@ -256,6 +305,7 @@ class GalleryService:
"size_bytes": record.size_bytes, "size_bytes": record.size_bytes,
"integrity_status": record.integrity_status, "integrity_status": record.integrity_status,
"created_at": record.created_at.isoformat(), "created_at": record.created_at.isoformat(),
"posted_at": posted_at.isoformat() if posted_at else None,
"thumbnail_url": thumbnail_url(record.sha256, record.mime), "thumbnail_url": thumbnail_url(record.sha256, record.mime),
"image_url": f"/images/{record.path.split('/images/', 1)[-1]}", "image_url": f"/images/{record.path.split('/images/', 1)[-1]}",
"artist": ( "artist": (
@@ -275,34 +325,41 @@ class GalleryService:
} }
async def _neighbors(self, record: ImageRecord) -> dict: async def _neighbors(self, record: ImageRecord) -> dict:
prev_stmt = ( # Compute the boundary image's effective_date in Python (one query
select(ImageRecord.id) # below + the SELECT we already have on `record`) and use it for
.where( # the neighbor comparison. Cheaper than re-deriving in SQL via
# correlated subquery.
boundary_eff = record.created_at
if record.primary_post_id is not None:
post_date = (await self.session.execute(
select(Post.post_date).where(Post.id == record.primary_post_id)
)).scalar_one_or_none()
if post_date is not None:
boundary_eff = post_date
eff = _effective_date_col()
prev_stmt = _outer_join_primary_post(
select(ImageRecord.id).where(
or_( or_(
ImageRecord.created_at > record.created_at, eff > boundary_eff,
and_( and_(
ImageRecord.created_at == record.created_at, eff == boundary_eff,
ImageRecord.id > record.id, ImageRecord.id > record.id,
), ),
) )
) )
.order_by(ImageRecord.created_at.asc(), ImageRecord.id.asc()) ).order_by(eff.asc(), ImageRecord.id.asc()).limit(1)
.limit(1) next_stmt = _outer_join_primary_post(
) select(ImageRecord.id).where(
next_stmt = (
select(ImageRecord.id)
.where(
or_( or_(
ImageRecord.created_at < record.created_at, eff < boundary_eff,
and_( and_(
ImageRecord.created_at == record.created_at, eff == boundary_eff,
ImageRecord.id < record.id, ImageRecord.id < record.id,
), ),
) )
) )
.order_by(ImageRecord.created_at.desc(), ImageRecord.id.desc()) ).order_by(eff.desc(), ImageRecord.id.desc()).limit(1)
.limit(1)
)
prev_id = (await self.session.execute(prev_stmt)).scalar_one_or_none() prev_id = (await self.session.execute(prev_stmt)).scalar_one_or_none()
next_id = (await self.session.execute(next_stmt)).scalar_one_or_none() next_id = (await self.session.execute(next_stmt)).scalar_one_or_none()
return {"prev_id": prev_id, "next_id": next_id} return {"prev_id": prev_id, "next_id": next_id}
@@ -311,9 +368,11 @@ class GalleryService:
def _group_by_year_month( def _group_by_year_month(
images: list[GalleryImage], images: list[GalleryImage],
) -> list[tuple[int, int, list[int]]]: ) -> list[tuple[int, int, list[int]]]:
"""Group by effective_date's year/month so migrated content surfaces
in the publish-date buckets, not the FC-scan-date bucket."""
groups: list[tuple[int, int, list[int]]] = [] groups: list[tuple[int, int, list[int]]] = []
for img in images: for img in images:
y, m = img.created_at.year, img.created_at.month y, m = img.effective_date.year, img.effective_date.month
if groups and groups[-1][0] == y and groups[-1][1] == m: if groups and groups[-1][0] == y and groups[-1][1] == m:
groups[-1][2].append(img.id) groups[-1][2].append(img.id)
else: else:
+22 -3
View File
@@ -280,7 +280,18 @@ class Importer:
) )
if self.settings.skip_transparent and has_alpha: if self.settings.skip_transparent and has_alpha:
pct = self._transparency_pct(source) try:
pct = self._transparency_pct(source)
except OSError as exc:
# PIL.verify() at line 263 only validates header structure;
# truncated/corrupt pixel data only surfaces when load()
# actually decodes (here via getchannel('A')). Convert to
# invalid_image skip so the Celery autoretry loop doesn't
# bounce the same broken file forever.
return ImportResult(
status="skipped", skip_reason=SkipReason.invalid_image,
error=f"PIL load failed during transparency check: {exc}",
)
if pct >= self.settings.transparency_threshold: if pct >= self.settings.transparency_threshold:
return ImportResult( return ImportResult(
status="skipped", skip_reason=SkipReason.too_transparent, status="skipped", skip_reason=SkipReason.too_transparent,
@@ -302,8 +313,16 @@ class Importer:
# Perceptual near-dup (images only; videos keep phash NULL). # Perceptual near-dup (images only; videos keep phash NULL).
phash = None phash = None
if not is_video(source): if not is_video(source):
with Image.open(source) as im: try:
phash = compute_phash(im) with Image.open(source) as im:
phash = compute_phash(im)
except OSError as exc:
# Same rationale as the transparency-check guard above:
# broken-pixel-data files pass verify() but blow up here.
return ImportResult(
status="skipped", skip_reason=SkipReason.invalid_image,
error=f"PIL load failed during phash compute: {exc}",
)
if phash is not None: if phash is not None:
cand_rows = self.session.execute( cand_rows = self.session.execute(
select( select(
+35 -4
View File
@@ -37,13 +37,25 @@ from ...utils.slug import slugify
from .ir_ingest import manifest_path from .ir_ingest import manifest_path
# Per-platform artist-profile URL — used as Source.url when restoring # Per-platform artist-profile URL — used as Source.url when restoring
# IR PostMetadata into FC. Keep this table in sync with # IR PostMetadata into FC. Must cover every platform that
# backend/app/services/extension_service.py:_PLATFORM_PATTERNS and # backend/app/services/extension_service.py:_PLATFORM_PATTERNS
# extension/lib/platforms.js. # recognizes; an entry missing here silently drops ALL PostMetadata for
# that platform during phase 4 (operator hit this 2026-05-25:
# DeviantArt + Pixiv posts in the IR migration produced empty
# ImageProvenance because they fell through this table).
#
# Pixiv caveat: the real profile URL takes a numeric user_id
# (https://www.pixiv.net/users/12345), but IR's PostMetadata.artist
# stores the display name not the id. We use the slugified name here
# so we preserve the artist→post→image linkage; the resulting Source.url
# won't resolve in a browser and the operator may want to manually fix
# it via Settings → Subscriptions once the migration lands.
_PLATFORM_PROFILE_URL = { _PLATFORM_PROFILE_URL = {
"patreon": "https://www.patreon.com/{slug}", "patreon": "https://www.patreon.com/{slug}",
"subscribestar": "https://www.subscribestar.com/{slug}", "subscribestar": "https://www.subscribestar.com/{slug}",
"hentaifoundry": "https://www.hentai-foundry.com/user/{slug}", "hentaifoundry": "https://www.hentai-foundry.com/user/{slug}",
"deviantart": "https://www.deviantart.com/{slug}",
"pixiv": "https://www.pixiv.net/users/{slug}",
} }
@@ -110,7 +122,14 @@ async def _ensure_provenance(
db: AsyncSession, *, db: AsyncSession, *,
image_id: int, post_id: int, source_id: int, dry_run: bool, image_id: int, post_id: int, source_id: int, dry_run: bool,
) -> bool: ) -> bool:
"""Returns True if a new ImageProvenance row was inserted.""" """Returns True if a new ImageProvenance row was inserted.
Also sets ImageRecord.primary_post_id to this post if the image
doesn't already have one — preserves any primary_post_id already
assigned at download time by the importer (don't clobber). This is
the linkage gallery_service.py uses to surface Post.post_date as
the image's effective date for sort/group/jump/neighbor nav.
"""
existing = (await db.execute( existing = (await db.execute(
select(ImageProvenance.id).where( select(ImageProvenance.id).where(
ImageProvenance.image_record_id == image_id, ImageProvenance.image_record_id == image_id,
@@ -118,6 +137,18 @@ async def _ensure_provenance(
ImageProvenance.source_id == source_id, ImageProvenance.source_id == source_id,
) )
)).scalar_one_or_none() )).scalar_one_or_none()
# Whether-or-not the provenance row already exists, ensure the
# image's primary_post_id is set so the gallery date-coalesce works.
# Idempotent: only writes when currently NULL.
if not dry_run:
await db.execute(
ImageRecord.__table__.update()
.where(ImageRecord.id == image_id)
.where(ImageRecord.primary_post_id.is_(None))
.values(primary_post_id=post_id)
)
if existing is not None: if existing is not None:
return False return False
if dry_run: if dry_run:
+82 -44
View File
@@ -4,12 +4,14 @@ CPU-only, single-image at a time. Loaded lazily inside the ml-worker
process; NOT thread-safe — the ml queue worker must run --concurrency=1 process; NOT thread-safe — the ml queue worker must run --concurrency=1
(set by the FC-1 entrypoint). (set by the FC-1 entrypoint).
Camie's selected_tags.csv columns: tag_id,name,category,count v2 layout reference: HuggingFace Camais03/camie-tagger-v2 root has
where category is a string: general|character|copyright|artist|meta|rating|year camie-tagger-v2.onnx (789 MB) + camie-tagger-v2-metadata.json (7.77 MB)
(unlike WD14's integer Danbooru category ids). + config.json. Tags ship as nested JSON, not CSV. Preprocessing and
output handling follow the published onnx_inference.py reference:
ImageNet normalize, NCHW layout, sigmoid on refined logits (output[1]).
""" """
import csv import json
import os import os
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
@@ -28,6 +30,8 @@ ImageFile.LOAD_TRUNCATED_IMAGES = True
MODEL_NAME = os.environ.get("CAMIE_MODEL_NAME", "camie-tagger-v2") MODEL_NAME = os.environ.get("CAMIE_MODEL_NAME", "camie-tagger-v2")
_MODEL_DIR = Path(os.environ.get("ML_MODEL_DIR", "/models")) / "camie" _MODEL_DIR = Path(os.environ.get("ML_MODEL_DIR", "/models")) / "camie"
_MODEL_FILE = f"{MODEL_NAME}.onnx"
_METADATA_FILE = f"{MODEL_NAME}-metadata.json"
# Below this confidence, predictions aren't stored (keeps the JSON compact). # Below this confidence, predictions aren't stored (keeps the JSON compact).
STORE_FLOOR = float(os.environ.get("TAGGER_STORE_FLOOR", "0.05")) STORE_FLOOR = float(os.environ.get("TAGGER_STORE_FLOOR", "0.05"))
@@ -39,6 +43,12 @@ STORE_FLOOR = float(os.environ.get("TAGGER_STORE_FLOOR", "0.05"))
# stored at STORE_FLOOR but artist never surfaces. # stored at STORE_FLOOR but artist never surfaces.
SURFACED_CATEGORIES = {"character", "copyright", "general"} SURFACED_CATEGORIES = {"character", "copyright", "general"}
# ImageNet preprocessing constants (per Camie v2 onnx_inference.py).
_IMAGENET_MEAN = np.array([0.485, 0.456, 0.406], dtype=np.float32)
_IMAGENET_STD = np.array([0.229, 0.224, 0.225], dtype=np.float32)
# Square-pad color ≈ ImageNet mean × 255 (matches reference inference).
_PAD_COLOR = (124, 116, 104)
@dataclass(frozen=True) @dataclass(frozen=True)
class TagPrediction: class TagPrediction:
@@ -51,34 +61,48 @@ class Tagger:
def __init__(self, model_dir: Path | None = None): def __init__(self, model_dir: Path | None = None):
self._model_dir = model_dir or _MODEL_DIR self._model_dir = model_dir or _MODEL_DIR
self._session = None # onnxruntime.InferenceSession once load()ed self._session = None # onnxruntime.InferenceSession once load()ed
self._tag_meta: list[dict] | None = None self._tag_names: list[str] | None = None
self._tag_categories: list[str] | None = None
self._input_name: str | None = None self._input_name: str | None = None
self._output_name: str | None = None self._input_size: int = 512
self._input_size: int = 448
def load(self) -> None: def load(self) -> None:
if self._session is not None: if self._session is not None:
return return
model_path = self._model_dir / "model.onnx" model_path = self._model_dir / _MODEL_FILE
tags_path = self._model_dir / "selected_tags.csv" meta_path = self._model_dir / _METADATA_FILE
if not model_path.is_file(): if not model_path.is_file():
raise RuntimeError( raise RuntimeError(
f"Camie model.onnx missing at {model_path}. " f"Camie {_MODEL_FILE} missing at {model_path}. "
f"Populate /models via the ml-worker downloader." f"Populate /models via the ml-worker downloader."
) )
if not tags_path.is_file(): if not meta_path.is_file():
raise RuntimeError( raise RuntimeError(
f"Camie selected_tags.csv missing at {tags_path}. " f"Camie {_METADATA_FILE} missing at {meta_path}. "
f"Populate /models via the ml-worker downloader." f"Populate /models via the ml-worker downloader."
) )
tag_meta: list[dict] = [] with open(meta_path) as f:
with open(tags_path, newline="") as f: metadata = json.load(f)
reader = csv.DictReader(f)
for row in reader: # Per Camie v2 onnx_inference.py: idx_to_tag is keyed by str(idx);
tag_meta.append( # tag_to_category maps tag_name -> category. Project to two parallel
{"name": row["name"], "category": row["category"]} # lists indexed by output position for O(1) lookup in the hot path.
) ds = metadata["dataset_info"]
idx_to_tag = ds["tag_mapping"]["idx_to_tag"]
tag_to_category = ds["tag_mapping"]["tag_to_category"]
total = ds["total_tags"]
names: list[str] = []
cats: list[str] = []
for i in range(total):
name = idx_to_tag.get(str(i), f"unknown-{i}")
names.append(name)
cats.append(tag_to_category.get(name, "general"))
# Input size from metadata; fall back to 512 (the v2 default).
self._input_size = int(
metadata.get("model_info", {}).get("img_size", 512)
)
# Lazy import — kept after the file-existence checks so the # Lazy import — kept after the file-existence checks so the
# missing-model RuntimeError still fires first in environments # missing-model RuntimeError still fires first in environments
@@ -89,51 +113,65 @@ class Tagger:
str(model_path), providers=["CPUExecutionProvider"] str(model_path), providers=["CPUExecutionProvider"]
) )
self._input_name = session.get_inputs()[0].name self._input_name = session.get_inputs()[0].name
self._output_name = session.get_outputs()[0].name
input_shape = session.get_inputs()[0].shape
for dim in input_shape:
if isinstance(dim, int) and dim > 1:
self._input_size = dim
break
# Assign sentinels last so a partial load isn't observable. # Assign sentinels last so a partial load isn't observable.
self._tag_meta = tag_meta self._tag_names = names
self._tag_categories = cats
self._session = session self._session = session
def _preprocess(self, image_path: Path) -> np.ndarray: def _preprocess(self, image_path: Path) -> np.ndarray:
img = Image.open(image_path) img = Image.open(image_path)
# Camie handles RGBA natively but we still composite onto white so # Composite RGBA onto neutral so transparency doesn't bias the model.
# transparency doesn't bias the model (same as IR's WD14 path). if img.mode == "RGBA":
if img.mode != "RGBA": bg = Image.new("RGBA", img.size, (255, 255, 255, 255))
img = img.convert("RGBA") bg.paste(img, mask=img.split()[3])
bg = Image.new("RGBA", img.size, (255, 255, 255, 255)) img = bg.convert("RGB")
bg.paste(img, mask=img.split()[3]) elif img.mode != "RGB":
img = bg.convert("RGB") img = img.convert("RGB")
# Pad to square with ImageNet-mean color, then bicubic resize.
w, h = img.size w, h = img.size
side = max(w, h) side = max(w, h)
square = Image.new("RGB", (side, side), (255, 255, 255)) square = Image.new("RGB", (side, side), _PAD_COLOR)
square.paste(img, ((side - w) // 2, (side - h) // 2)) square.paste(img, ((side - w) // 2, (side - h) // 2))
square = square.resize( square = square.resize(
(self._input_size, self._input_size), Image.BICUBIC (self._input_size, self._input_size), Image.BICUBIC
) )
arr = np.array(square, dtype=np.float32)
return arr[np.newaxis, :, :, :] # NHWC arr = np.array(square, dtype=np.float32) / 255.0 # HWC, [0,1]
arr = (arr - _IMAGENET_MEAN) / _IMAGENET_STD # ImageNet normalize
arr = arr.transpose(2, 0, 1) # HWC -> CHW
return arr[np.newaxis, :, :, :] # NCHW
def infer(self, image_path: Path) -> dict[str, TagPrediction]: def infer(self, image_path: Path) -> dict[str, TagPrediction]:
"""Run Camie on one image. Returns {name: TagPrediction}, only """Run Camie v2 on one image. Returns {name: TagPrediction} with
entries with confidence >= STORE_FLOOR (across all categories — confidence >= STORE_FLOOR (across all categories — the suggestion
the suggestion service does category filtering later).""" service does category filtering later).
v2 emits multiple outputs; we use the refined predictions
(output[1] per onnx_inference.py). Sigmoid is applied to raw
logits to produce [0,1] confidence scores.
"""
self.load() self.load()
x = self._preprocess(image_path) x = self._preprocess(image_path)
out = self._session.run([self._output_name], {self._input_name: x})[0][0] outputs = self._session.run(None, {self._input_name: x})
# Refined predictions if present (v2 emits initial + refined),
# fall back to initial for single-output forks.
logits = outputs[1] if len(outputs) > 1 else outputs[0]
# Squeeze batch dim, apply sigmoid.
probs = 1.0 / (1.0 + np.exp(-logits[0]))
results: dict[str, TagPrediction] = {} results: dict[str, TagPrediction] = {}
for idx, score in enumerate(out): names = self._tag_names
cats = self._tag_categories
for idx, score in enumerate(probs):
conf = float(score) conf = float(score)
if conf < STORE_FLOOR: if conf < STORE_FLOOR:
continue continue
meta = self._tag_meta[idx] if idx >= len(names):
results[meta["name"]] = TagPrediction( # Output longer than metadata declared — shouldn't happen but
name=meta["name"], category=meta["category"], confidence=conf # don't crash the import pipeline if v2 metadata desynchronizes.
continue
results[names[idx]] = TagPrediction(
name=names[idx], category=cats[idx], confidence=conf
) )
return results return results
+58 -20
View File
@@ -16,6 +16,7 @@ from ._sync_engine import sync_session_factory as _sync_session_factory
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
STUCK_THRESHOLD_MINUTES = 5 STUCK_THRESHOLD_MINUTES = 5
ORPHAN_PENDING_THRESHOLD_MINUTES = 30
OLD_TASK_DAYS = 7 OLD_TASK_DAYS = 7
PHASH_PAGE = 500 PHASH_PAGE = 500
VERIFY_PAGE = 200 VERIFY_PAGE = 200
@@ -26,40 +27,77 @@ TASK_RUN_KEEP_FAILURE_SECONDS = 7 * 24 * 3600 # 7 days
@celery.task(name="backend.app.tasks.maintenance.recover_interrupted_tasks") @celery.task(name="backend.app.tasks.maintenance.recover_interrupted_tasks")
def recover_interrupted_tasks() -> int: def recover_interrupted_tasks() -> int:
"""Find ImportTask rows stuck in 'processing' for >5 min and re-queue them. """Recover stuck ImportTask rows. Two distinct stuck states:
Why 5 min: import_media_file is sub-second for the vast majority of 1. 'processing' > 5 min — worker crash mid-import. Re-queue via
files; even a large-video transcode caps at the per-task soft_time_limit .delay() and let the import retry. Was 30 min historically;
(5 min) defined on the task itself. Anything still 'processing' after tightened 2026-05-24 after operator hit a 2224-row zombie pile.
that window is a confirmed crash (worker died, DB disconnect mid-flush, import_media_file is sub-second for the vast majority of files and
OOM) and must be recycled. Was 30 min historically; tightened capped at the per-task soft_time_limit (5 min), so anything still
2026-05-24 after operator hit a 2224-row zombie pile during the IR 'processing' after that window is a confirmed crash.
migration scan.
2. 'pending' or 'queued' > 30 min — enqueue-phase crash. scan_directory
creates rows with status='pending' (commit), then in a second pass
transitions to 'queued' and calls .delay() (commit). If the scanner
crashes between those two commits, rows are orphaned in 'pending'
(never enqueued) with no recovery path — invisible to the
'processing' sweep above. Flagged 2026-05-25 by operator hitting a
5490-row orphan pile. Flip these to 'failed' (not re-enqueue) so
the operator drains them via /api/import/retry-failed at their own
pace; bulk-re-enqueueing 5000+ rows would thundering-herd the
import worker.
Returns total rows touched (recovered + marked failed).
""" """
SessionLocal = _sync_session_factory() SessionLocal = _sync_session_factory()
cutoff = datetime.now(UTC) - timedelta(minutes=STUCK_THRESHOLD_MINUTES) now = datetime.now(UTC)
processing_cutoff = now - timedelta(minutes=STUCK_THRESHOLD_MINUTES)
orphan_cutoff = now - timedelta(minutes=ORPHAN_PENDING_THRESHOLD_MINUTES)
with SessionLocal() as session: with SessionLocal() as session:
stuck_ids = session.execute( stuck_ids = session.execute(
select(ImportTask.id) select(ImportTask.id)
.where(ImportTask.status == "processing") .where(ImportTask.status == "processing")
.where(ImportTask.started_at < cutoff) .where(ImportTask.started_at < processing_cutoff)
).scalars().all() ).scalars().all()
if not stuck_ids: orphan_ids = session.execute(
select(ImportTask.id)
.where(ImportTask.status.in_(["pending", "queued"]))
.where(ImportTask.created_at < orphan_cutoff)
).scalars().all()
if not stuck_ids and not orphan_ids:
return 0 return 0
session.execute( if stuck_ids:
update(ImportTask) session.execute(
.where(ImportTask.id.in_(stuck_ids)) update(ImportTask)
.values(status="queued", started_at=None, error="recovered from stuck state") .where(ImportTask.id.in_(stuck_ids))
) .values(status="queued", started_at=None, error="recovered from stuck state")
)
if orphan_ids:
session.execute(
update(ImportTask)
.where(ImportTask.id.in_(orphan_ids))
.values(
status="failed",
error=(
"orphan pending/queued swept by recover_interrupted_tasks "
"(scanner likely crashed mid-enqueue); retry via "
"/api/import/retry-failed"
),
)
)
session.commit() session.commit()
from .import_file import import_media_file if stuck_ids:
for tid in stuck_ids: from .import_file import import_media_file
import_media_file.delay(tid) for tid in stuck_ids:
import_media_file.delay(tid)
return len(stuck_ids) return len(stuck_ids) + len(orphan_ids)
@celery.task(name="backend.app.tasks.maintenance.cleanup_old_tasks") @celery.task(name="backend.app.tasks.maintenance.cleanup_old_tasks")
@@ -38,9 +38,18 @@ function onCardClick() {
.fc-artistcard { cursor: pointer; } .fc-artistcard { cursor: pointer; }
.fc-artistcard__previews { .fc-artistcard__previews {
display: grid; grid-template-columns: repeat(3, 1fr); display: grid; grid-template-columns: repeat(3, 1fr);
gap: 2px; aspect-ratio: 3 / 1; background: rgb(var(--v-theme-surface-light)); gap: 2px; aspect-ratio: 3 / 1;
/* Explicit floor + ceiling so tall source images can't escape the
preview slot even on browsers where aspect-ratio doesn't compute. */
min-height: 150px; max-height: 220px;
overflow: hidden;
background: rgb(var(--v-theme-surface-light));
}
.fc-artistcard__previews img {
display: block;
width: 100%; height: 100%;
object-fit: cover; object-position: center;
} }
.fc-artistcard__previews img { width: 100%; height: 100%; object-fit: cover; }
.fc-artistcard__noimg { .fc-artistcard__noimg {
grid-column: 1 / -1; display: flex; align-items: center; grid-column: 1 / -1; display: flex; align-items: center;
justify-content: center; justify-content: center;
+12 -2
View File
@@ -106,9 +106,19 @@ function submit() {
.fc-tagcard { cursor: pointer; } .fc-tagcard { cursor: pointer; }
.fc-tagcard__previews { .fc-tagcard__previews {
display: grid; grid-template-columns: repeat(3, 1fr); display: grid; grid-template-columns: repeat(3, 1fr);
gap: 2px; aspect-ratio: 3 / 1; background: rgb(var(--v-theme-surface-light)); gap: 2px; aspect-ratio: 3 / 1;
/* Explicit floor + ceiling so tall source images can't escape the
preview slot even on browsers where aspect-ratio doesn't compute
(older Safari, embedded webviews). */
min-height: 150px; max-height: 220px;
overflow: hidden;
background: rgb(var(--v-theme-surface-light));
}
.fc-tagcard__previews img {
display: block;
width: 100%; height: 100%;
object-fit: cover; object-position: center;
} }
.fc-tagcard__previews img { width: 100%; height: 100%; object-fit: cover; }
.fc-tagcard__noimg { .fc-tagcard__noimg {
grid-column: 1 / -1; display: flex; align-items: center; grid-column: 1 / -1; display: flex; align-items: center;
justify-content: center; justify-content: center;
@@ -17,6 +17,12 @@
> >
Retry failed Retry failed
</v-btn> </v-btn>
<v-btn
variant="text" rounded="pill" size="small" color="warning"
:disabled="!hasStuck" @click="onClearStuckOpen"
>
Clear stuck
</v-btn>
<v-btn <v-btn
variant="text" rounded="pill" size="small" color="error" variant="text" rounded="pill" size="small" color="error"
@click="onClearOpen" @click="onClearOpen"
@@ -69,6 +75,31 @@
</v-card-actions> </v-card-actions>
</v-card> </v-card>
</v-dialog> </v-dialog>
<v-dialog v-model="clearStuckDialog" max-width="480">
<v-card>
<v-card-title>Clear stuck tasks</v-card-title>
<v-card-text>
<v-alert type="warning" variant="tonal" density="compact" class="mb-3">
Force every <strong>pending / queued / processing</strong> task to
<strong>failed</strong> and finalize any active batch that
has no remaining work. Use this when the automatic recovery
sweep keeps re-queueing the same row (e.g., corrupt file in
an autoretry loop, or worker model missing).
</v-alert>
<p class="text-body-2">
Tasks remain in the database with status=<code>failed</code>;
click <em>Retry failed</em> once the underlying cause is
resolved to re-queue them.
</p>
</v-card-text>
<v-card-actions>
<v-spacer />
<v-btn @click="clearStuckDialog = false">Cancel</v-btn>
<v-btn color="warning" rounded="pill" @click="onClearStuckConfirm">Clear stuck</v-btn>
</v-card-actions>
</v-card>
</v-dialog>
</v-card> </v-card>
</template> </template>
@@ -80,6 +111,7 @@ const store = useImportStore()
const statusFilter = ref(null) const statusFilter = ref(null)
const clearDialog = ref(false) const clearDialog = ref(false)
const clearAgeDays = ref(7) const clearAgeDays = ref(7)
const clearStuckDialog = ref(false)
const statusOptions = [ const statusOptions = [
{ title: 'All', value: null }, { title: 'All', value: null },
@@ -100,6 +132,9 @@ const headers = [
] ]
const hasFailed = computed(() => store.tasks.some(t => t.status === 'failed')) const hasFailed = computed(() => store.tasks.some(t => t.status === 'failed'))
const hasStuck = computed(() => store.tasks.some(
t => t.status === 'pending' || t.status === 'queued' || t.status === 'processing'
))
function statusColor(s) { function statusColor(s) {
return { return {
@@ -138,4 +173,9 @@ async function onClearConfirm() {
await store.clearCompleted(clearAgeDays.value) await store.clearCompleted(clearAgeDays.value)
clearDialog.value = false clearDialog.value = false
} }
function onClearStuckOpen() { clearStuckDialog.value = true }
async function onClearStuckConfirm() {
await store.clearStuck()
clearStuckDialog.value = false
}
</script> </script>
@@ -2,7 +2,7 @@
<v-card> <v-card>
<v-card-title>Trigger scan</v-card-title> <v-card-title>Trigger scan</v-card-title>
<v-card-text> <v-card-text>
<div v-if="store.activeBatch" class="d-flex align-center" style="gap: 12px;"> <div v-if="store.activeBatch" class="d-flex align-center mb-3" style="gap: 12px;">
<v-progress-circular <v-progress-circular
indeterminate color="accent" size="20" indeterminate color="accent" size="20"
/> />
@@ -13,20 +13,40 @@
failed {{ store.activeBatch.failed }} / failed {{ store.activeBatch.failed }} /
{{ store.activeBatch.total_files }} files {{ store.activeBatch.total_files }} files
</span> </span>
<v-spacer />
<v-btn
variant="text" rounded="pill" size="small" color="warning"
:loading="clearing" @click="onClearStuck"
>
Clear stuck
</v-btn>
</div> </div>
<div v-else>
<p class="text-body-2 mb-3"> <p class="text-body-2 mb-3">
<span v-if="!store.activeBatch">
Run a quick scan of the import directory. Deep scan (pHash dedup, Run a quick scan of the import directory. Deep scan (pHash dedup,
archives) lands in FC-2d. archives) lands in FC-2d.
</p> </span>
<v-btn color="primary" rounded="pill" @click="trigger" :loading="busy"> <span v-else>
<v-icon start>mdi-magnify-scan</v-icon> An active batch is in progress. Wait for it to finish, or click
Quick scan <em>Clear stuck</em> above if it has been wedged with no
</v-btn> measurable progress.
<v-alert v-if="store.triggerError" type="error" variant="tonal" class="mt-3" closable> </span>
{{ store.triggerError }} </p>
</v-alert>
</div> <v-btn
color="primary" rounded="pill"
:disabled="!!store.activeBatch"
:loading="busy"
@click="trigger"
>
<v-icon start>mdi-magnify-scan</v-icon>
Quick scan
</v-btn>
<v-alert v-if="store.triggerError" type="error" variant="tonal" class="mt-3" closable>
{{ store.triggerError }}
</v-alert>
</v-card-text> </v-card-text>
</v-card> </v-card>
</template> </template>
@@ -37,9 +57,21 @@ import { useImportStore } from '../../stores/import.js'
const store = useImportStore() const store = useImportStore()
const busy = ref(false) const busy = ref(false)
const clearing = ref(false)
async function trigger() { async function trigger() {
busy.value = true busy.value = true
try { await store.triggerScan() } catch {} finally { busy.value = false } try { await store.triggerScan() } catch {} finally { busy.value = false }
} }
async function onClearStuck() {
clearing.value = true
try {
await store.clearStuck()
} catch {
// store surfaces error via triggerError if needed
} finally {
clearing.value = false
}
}
</script> </script>
+8 -1
View File
@@ -92,6 +92,13 @@ export const useImportStore = defineStore('import', () => {
await loadTasks(true) await loadTasks(true)
} }
async function clearStuck() {
const body = await api.post('/api/import/clear-stuck')
await loadTasks(true)
await refreshStatus()
return body
}
const hasMore = computed(() => tasksNextCursor.value !== null) const hasMore = computed(() => tasksNextCursor.value !== null)
return { return {
@@ -101,6 +108,6 @@ export const useImportStore = defineStore('import', () => {
triggerError, triggerError,
loadSettings, patchSettings, loadSettings, patchSettings,
refreshStatus, triggerScan, refreshStatus, triggerScan,
loadTasks, setStatusFilter, retryFailed, clearCompleted loadTasks, setStatusFilter, retryFailed, clearCompleted, clearStuck
} }
}) })
+1 -1
View File
@@ -84,7 +84,7 @@ onUnmounted(() => observer && observer.disconnect())
} }
.fc-artists__grid { .fc-artists__grid {
display: grid; display: grid;
grid-template-columns: repeat(auto-fill, minmax(220px, 1fr)); grid-template-columns: repeat(auto-fill, minmax(440px, 1fr));
gap: 12px; gap: 12px;
} }
.fc-artists__sentinel { .fc-artists__sentinel {
+1 -1
View File
@@ -236,7 +236,7 @@ async function onDeleteTagConfirm() {
} }
.fc-tags__grid { .fc-tags__grid {
display: grid; display: grid;
grid-template-columns: repeat(auto-fill, minmax(220px, 1fr)); grid-template-columns: repeat(auto-fill, minmax(440px, 1fr));
gap: 12px; gap: 12px;
} }
.fc-tags__sentinel { .fc-tags__sentinel {
+55
View File
@@ -86,6 +86,61 @@ async def test_clear_completed(client, db):
assert body["deleted"] == 1 assert body["deleted"] == 1
@pytest.mark.asyncio
async def test_clear_stuck_fails_non_terminal_and_finalizes_orphan_batch(client, db):
"""Operator-flagged 2026-05-25: 3 large PNGs got stuck in 'processing'
for 2 days, the active ImportBatch never finalized, and the UI's
'Scanning...' banner persisted with 0/0 files. /api/import/clear-stuck
is the escape hatch to break the autoretry loop manually."""
from sqlalchemy import select as _select
batch = ImportBatch(triggered_by="manual", source_path="/import", scan_mode="quick")
db.add(batch)
await db.flush()
# Three stuck rows in mixed non-terminal states.
db.add(ImportTask(
batch_id=batch.id, source_path="/p1", task_type="media", status="processing",
))
db.add(ImportTask(
batch_id=batch.id, source_path="/p2", task_type="media", status="queued",
))
db.add(ImportTask(
batch_id=batch.id, source_path="/p3", task_type="media", status="pending",
))
# One already-complete row should be untouched.
db.add(ImportTask(
batch_id=batch.id, source_path="/done", task_type="media",
status="complete", finished_at=datetime.now(UTC),
))
await db.commit()
resp = await client.post("/api/import/clear-stuck")
body = await resp.get_json()
assert resp.status_code == 200
assert body["tasks_failed"] == 3
assert body["batches_finalized"] == 1
statuses = {
row.status for row in
(await db.execute(_select(ImportTask).where(ImportTask.batch_id == batch.id)))
.scalars().all()
}
assert statuses == {"failed", "complete"}
batch_status = (await db.execute(
_select(ImportBatch.status).where(ImportBatch.id == batch.id)
)).scalar_one()
assert batch_status == "complete"
@pytest.mark.asyncio
async def test_clear_stuck_no_op_when_nothing_stuck(client, db):
resp = await client.post("/api/import/clear-stuck")
body = await resp.get_json()
assert resp.status_code == 200
assert body == {"tasks_failed": 0, "batches_finalized": 0}
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_trigger_accepts_deep(client, monkeypatch): async def test_trigger_accepts_deep(client, monkeypatch):
# Stub the task dispatch — assert the API accepts 'deep' and forwards # Stub the task dispatch — assert the API accepts 'deep' and forwards
+7 -2
View File
@@ -9,11 +9,16 @@ from backend.app.scripts import download_models as dm
def test_ensure_camie_skips_when_present(tmp_path, monkeypatch): def test_ensure_camie_skips_when_present(tmp_path, monkeypatch):
"""v2 layout (HF Camais03/camie-tagger-v2): the ONNX file is named
camie-tagger-v2.onnx (not model.onnx) and tags ship inside
camie-tagger-v2-metadata.json (not selected_tags.csv). Both at root.
Updated 2026-05-25 after the actual repo layout was confirmed via
WebFetch — the old assertion pinned the v1 filenames."""
monkeypatch.setattr(dm, "MODEL_ROOT", tmp_path) monkeypatch.setattr(dm, "MODEL_ROOT", tmp_path)
camie = tmp_path / "camie" camie = tmp_path / "camie"
camie.mkdir(parents=True) camie.mkdir(parents=True)
(camie / "model.onnx").write_bytes(b"x") (camie / "camie-tagger-v2.onnx").write_bytes(b"x")
(camie / "selected_tags.csv").write_text("tag_id,name,category,count\n") (camie / "camie-tagger-v2-metadata.json").write_text("{}")
with patch.object(dm, "_snapshot") as snap: with patch.object(dm, "_snapshot") as snap:
dm.ensure_camie() dm.ensure_camie()
snap.assert_not_called() snap.assert_not_called()
+132
View File
@@ -133,3 +133,135 @@ async def test_get_image_with_tags_includes_integrity_status(db):
svc = GalleryService(db) svc = GalleryService(db)
payload = await svc.get_image_with_tags(img.id) payload = await svc.get_image_with_tags(img.id)
assert payload["integrity_status"] == "ok" assert payload["integrity_status"] == "ok"
async def _seed_image_with_post(
db, *, sha: str, image_created_at, post_date, artist_name="test-artist",
platform="patreon", external_post_id="42",
):
"""Helper: seed an Artist + Source + Post and one ImageRecord whose
primary_post_id points at that Post. Used for date-coalesce tests."""
from backend.app.models import Artist, Post, Source
artist = Artist(name=artist_name, slug=artist_name.lower().replace(" ", "-"))
db.add(artist)
await db.flush()
source = Source(
artist_id=artist.id, platform=platform,
url=f"https://www.{platform}.com/{artist.slug}",
)
db.add(source)
await db.flush()
post = Post(
source_id=source.id, external_post_id=external_post_id,
post_title="A Post", post_date=post_date,
)
db.add(post)
await db.flush()
img = ImageRecord(
path=f"/images/test/{sha[:8]}.jpg",
sha256=sha, size_bytes=1000, mime="image/jpeg",
width=100, height=100,
origin="imported_filesystem", integrity_status="unknown",
primary_post_id=post.id,
)
img.created_at = image_created_at
db.add(img)
await db.flush()
return img, post
@pytest.mark.asyncio
async def test_scroll_sorts_by_post_date_when_available(db):
"""Operator-flagged 2026-05-25: ~57k IR images all imported in the
same week sort by image.created_at and pile up in one month bucket.
Once primary_post_id is wired (via tag_apply phase 4), the gallery
should sort by Post.post_date instead, spreading them across the
actual publish years."""
base_import = _now()
# Image A: imported NOW, but post was made 2 years ago.
img_a, _ = await _seed_image_with_post(
db, sha="a" * 64,
image_created_at=base_import,
post_date=base_import - timedelta(days=730),
artist_name="Aria", external_post_id="A-1",
)
# Image B: imported NOW (1 min later), post made YESTERDAY.
img_b, _ = await _seed_image_with_post(
db, sha="b" * 64,
image_created_at=base_import - timedelta(minutes=1),
post_date=base_import - timedelta(days=1),
artist_name="Bea", external_post_id="B-1",
)
# Image C: filesystem-imported, no primary_post_id, created 5 days ago.
img_c = ImageRecord(
path="/images/test/c.jpg", sha256="c" * 64,
size_bytes=1000, mime="image/jpeg",
width=100, height=100,
origin="imported_filesystem", integrity_status="unknown",
)
img_c.created_at = base_import - timedelta(days=5)
db.add(img_c)
await db.flush()
svc = GalleryService(db)
page = await svc.scroll(cursor=None, limit=10)
# Effective-date order: B (yesterday) > C (5 days ago) > A (2 years ago)
assert [i.id for i in page.images] == [img_b.id, img_c.id, img_a.id]
# API exposes both fields explicitly so the UI can show "Posted X / Imported Y".
a_payload = next(i for i in page.images if i.id == img_a.id)
assert a_payload.posted_at is not None
assert a_payload.posted_at < a_payload.created_at
c_payload = next(i for i in page.images if i.id == img_c.id)
assert c_payload.posted_at is None
assert c_payload.effective_date == c_payload.created_at
@pytest.mark.asyncio
async def test_timeline_buckets_use_post_date_when_available(db):
"""Timeline group-by must follow the same effective_date rule so the
UI's year/month navigation surfaces publish-date buckets, not the
single FC-scan bucket all migrated images share."""
base = datetime(2026, 6, 15, 12, 0, tzinfo=UTC)
await _seed_image_with_post(
db, sha="1" * 64,
image_created_at=base,
post_date=datetime(2024, 3, 10, tzinfo=UTC),
artist_name="Carl", external_post_id="C-1",
)
await _seed_image_with_post(
db, sha="2" * 64,
image_created_at=base,
post_date=datetime(2024, 3, 11, tzinfo=UTC),
artist_name="Dee", external_post_id="D-1",
)
await _seed_image_with_post(
db, sha="3" * 64,
image_created_at=base,
post_date=datetime(2025, 9, 1, tzinfo=UTC),
artist_name="Eli", external_post_id="E-1",
)
svc = GalleryService(db)
buckets = await svc.timeline()
bucket_keys = {(b.year, b.month, b.count) for b in buckets}
# Two posts in 2024-03, one in 2025-09 — even though all imported in 2026-06.
assert (2024, 3, 2) in bucket_keys
assert (2025, 9, 1) in bucket_keys
# The FC-import bucket should NOT appear since all 3 images have post_date.
assert not any(b.year == 2026 and b.month == 6 for b in buckets)
@pytest.mark.asyncio
async def test_get_image_with_tags_includes_posted_at_when_present(db):
base = _now()
img, _ = await _seed_image_with_post(
db, sha="f" * 64,
image_created_at=base,
post_date=base - timedelta(days=365),
artist_name="Fred", external_post_id="F-1",
)
svc = GalleryService(db)
payload = await svc.get_image_with_tags(img.id)
assert payload["posted_at"] is not None
# Image's own created_at is still surfaced separately.
assert payload["created_at"] != payload["posted_at"]
+55
View File
@@ -134,3 +134,58 @@ def test_root_level_file_has_no_artist(importer, import_layout):
importer.import_one(src) importer.import_one(src)
artists = importer.session.execute(select(Artist)).scalars().all() artists = importer.session.execute(select(Artist)).scalars().all()
assert artists == [] assert artists == []
def test_pil_load_oserror_in_transparency_check_skips_not_raises(
importer, import_layout, monkeypatch,
):
"""PIL.verify() only validates header structure — broken pixel data
only surfaces when load() actually decodes. The importer must catch
the OSError and return a skipped: invalid_image result so the Celery
autoretry loop doesn't bounce the same broken file forever.
Operator hit this 2026-05-25 with a corrupt JPEG in the IR set."""
import_root, _ = import_layout
src = import_root / "Bob" / "corrupt.png"
# Make a real RGBA PNG so the has_alpha path engages.
_make_png_rgba(src, (100, 100), alpha=128)
importer.settings.skip_transparent = True
importer.settings.transparency_threshold = 0.5
# Force the next _transparency_pct call to raise as if PIL's load()
# blew up on truncated pixel data.
def _boom(_self, _src):
raise OSError("broken data stream when reading image file")
monkeypatch.setattr(
type(importer), "_transparency_pct", _boom,
)
result = importer.import_one(src)
assert result.status == "skipped"
assert result.skip_reason == SkipReason.invalid_image
assert "transparency check" in (result.error or "")
def test_pil_load_oserror_in_phash_compute_skips_not_raises(
importer, import_layout, monkeypatch,
):
"""Same shape as the transparency-check guard, but for the phash
compute block — the OTHER place PIL.load() runs implicitly during
the dedup pipeline."""
import_root, _ = import_layout
src = import_root / "Carol" / "corrupt.jpg"
_make_jpeg(src)
# Disable transparency check so we reach the phash compute block.
importer.settings.skip_transparent = False
from backend.app.services import importer as importer_module
def _boom(_im):
raise OSError("broken data stream when reading image file")
monkeypatch.setattr(importer_module, "compute_phash", _boom)
result = importer.import_one(src)
assert result.status == "skipped"
assert result.skip_reason == SkipReason.invalid_image
assert "phash compute" in (result.error or "")
+96
View File
@@ -68,6 +68,102 @@ def test_recover_interrupted_only_old(db_sync, monkeypatch):
assert dispatched == [stale.id] assert dispatched == [stale.id]
def test_recover_interrupted_sweeps_pending_orphans_to_failed(db_sync, monkeypatch):
"""A scan that creates ImportTask rows but crashes before the second
pass (transition to 'queued' + .delay()) leaves rows orphaned at
status='pending'. The sweep flips them to 'failed' so the operator
can drain via /api/import/retry-failed without thundering-herding.
Banked 2026-05-25 after operator hit 5490 stuck pending rows.
"""
from backend.app.tasks import import_file
monkeypatch.setattr(import_file.import_media_file, "delay", lambda *_: None)
batch_id = _make_batch(db_sync)
now = datetime.now(UTC)
fresh_pending = ImportTask(
batch_id=batch_id, source_path="/import/fresh.jpg", task_type="media",
status="pending",
)
db_sync.add(fresh_pending)
db_sync.flush()
# created_at defaults to now() server-side; fresh row stays untouched.
# Two stale rows simulating the orphan pile: one 'pending', one
# 'queued' (scanner crashed AFTER transitioning some rows but
# before all). Both should sweep.
stale_pending = ImportTask(
batch_id=batch_id, source_path="/import/stale1.jpg", task_type="media",
status="pending",
)
stale_queued = ImportTask(
batch_id=batch_id, source_path="/import/stale2.jpg", task_type="media",
status="queued",
)
db_sync.add_all([stale_pending, stale_queued])
db_sync.flush()
# Backdate created_at past the orphan cutoff (30 min).
from sqlalchemy import update as _upd
db_sync.execute(
_upd(ImportTask)
.where(ImportTask.id.in_([stale_pending.id, stale_queued.id]))
.values(created_at=now - timedelta(hours=2))
)
db_sync.commit()
from backend.app.tasks.maintenance import recover_interrupted_tasks
touched = recover_interrupted_tasks.apply().get()
assert touched == 2
db_sync.refresh(fresh_pending)
db_sync.refresh(stale_pending)
db_sync.refresh(stale_queued)
assert fresh_pending.status == "pending" # fresh row untouched
assert stale_pending.status == "failed"
assert stale_queued.status == "failed"
assert "orphan" in (stale_pending.error or "")
def test_recover_interrupted_handles_both_stuck_and_orphans(db_sync, monkeypatch):
"""One sweep tick handles both 'processing' crashes AND
'pending'/'queued' orphans in a single pass."""
from backend.app.tasks import import_file
dispatched: list[int] = []
monkeypatch.setattr(
import_file.import_media_file, "delay", dispatched.append
)
batch_id = _make_batch(db_sync)
now = datetime.now(UTC)
stuck = ImportTask(
batch_id=batch_id, source_path="/import/stuck.jpg", task_type="media",
status="processing", started_at=now - timedelta(hours=2),
)
orphan = ImportTask(
batch_id=batch_id, source_path="/import/orphan.jpg", task_type="media",
status="pending",
)
db_sync.add_all([stuck, orphan])
db_sync.flush()
from sqlalchemy import update as _upd
db_sync.execute(
_upd(ImportTask).where(ImportTask.id == orphan.id)
.values(created_at=now - timedelta(hours=2))
)
db_sync.commit()
from backend.app.tasks.maintenance import recover_interrupted_tasks
touched = recover_interrupted_tasks.apply().get()
assert touched == 2
db_sync.refresh(stuck)
db_sync.refresh(orphan)
assert stuck.status == "queued"
assert orphan.status == "failed"
assert dispatched == [stuck.id] # stuck rows re-enqueue; orphans don't
def test_cleanup_old_deletes_finished_old(db_sync): def test_cleanup_old_deletes_finished_old(db_sync):
batch_id = _make_batch(db_sync) batch_id = _make_batch(db_sync)
now = datetime.now(UTC) now = datetime.now(UTC)
+4 -1
View File
@@ -40,5 +40,8 @@ def test_get_tagger_singleton():
def test_load_raises_when_model_missing(tmp_path): def test_load_raises_when_model_missing(tmp_path):
t = Tagger(model_dir=tmp_path / "nonexistent") t = Tagger(model_dir=tmp_path / "nonexistent")
with pytest.raises(RuntimeError, match="model.onnx missing"): # Match the trailing "missing at <path>" rather than the specific
# filename, so a future model-version bump (camie-tagger-v3.onnx, etc.)
# doesn't bounce this test.
with pytest.raises(RuntimeError, match=r"\.onnx missing at "):
t.load() t.load()
+97
View File
@@ -227,6 +227,67 @@ async def test_image_posts_creates_source_post_provenance(db, tmp_path):
)).scalar_one() )).scalar_one()
assert prov_count == 1 assert prov_count == 1
# Phase 4 must also set ImageRecord.primary_post_id so the gallery's
# effective_date COALESCE can surface Post.post_date. Operator-flagged
# 2026-05-25: without this, IR-migrated images keep sorting by FC's
# scan date instead of the original publish date.
primary_post_id = (await db.execute(
select(ImageRecord.primary_post_id).where(ImageRecord.id == img_id)
)).scalar_one()
canonical_post_id = (await db.execute(
select(Post.id).where(Post.external_post_id == "10001")
)).scalar_one()
assert primary_post_id == canonical_post_id
@pytest.mark.asyncio
async def test_image_posts_primary_post_id_not_clobbered(db, tmp_path):
"""If the importer already set primary_post_id (e.g. a downloaded
image with a known provenance), phase 4 must NOT overwrite it when
re-running tag_apply against the IR migration. The existing
download-time linkage is the source of truth."""
sha = "9" * 64
await _seed_image(db, sha, suffix="9")
# Pre-set primary_post_id to a sentinel Post so we can detect a clobber.
img_id = (await db.execute(
select(ImageRecord.id).where(ImageRecord.sha256 == sha)
)).scalar_one()
# Build an existing Source + Post for the sentinel.
art = Artist(name="Pre-existing", slug="pre-existing")
db.add(art)
await db.flush()
src = Source(
artist_id=art.id, platform="patreon",
url="https://www.patreon.com/pre-existing",
)
db.add(src)
await db.flush()
sentinel_post = Post(
source_id=src.id, external_post_id="sentinel-99",
post_title="Pre-existing",
)
db.add(sentinel_post)
await db.flush()
await db.execute(
ImageRecord.__table__.update()
.where(ImageRecord.id == img_id)
.values(primary_post_id=sentinel_post.id)
)
await db.commit()
_write_manifest(tmp_path, image_posts=[
{**_POST_ENTRY, "image_sha256s": [sha]},
])
await tag_apply.apply_async(db, images_root=tmp_path, dry_run=False)
# The migration created a NEW Post (external_post_id="10001") and a
# new ImageProvenance, but primary_post_id must still point at the
# original sentinel.
primary_post_id = (await db.execute(
select(ImageRecord.primary_post_id).where(ImageRecord.id == img_id)
)).scalar_one()
assert primary_post_id == sentinel_post.id
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_image_posts_idempotent_on_rerun(db, tmp_path): async def test_image_posts_idempotent_on_rerun(db, tmp_path):
@@ -281,6 +342,42 @@ async def test_image_posts_unknown_platform_skipped(db, tmp_path):
assert src_count == 0 assert src_count == 0
@pytest.mark.parametrize("platform,expected_url", [
("deviantart", "https://www.deviantart.com/maewix"),
("pixiv", "https://www.pixiv.net/users/maewix"),
])
@pytest.mark.asyncio
async def test_image_posts_extended_platforms_create_source(
db, tmp_path, platform, expected_url,
):
"""Regression for 2026-05-25 operator-reported bug: phase 4's
_PLATFORM_PROFILE_URL had only patreon/subscribestar/hentaifoundry,
silently dropping deviantart + pixiv PostMetadata from the IR migration."""
sha = f"{platform[0]}" * 64
await _seed_image(db, sha, suffix=f"_{platform}")
await db.commit()
_write_manifest(tmp_path, image_posts=[
{**_POST_ENTRY, "platform": platform, "image_sha256s": [sha]},
])
result = await tag_apply.apply_async(db, images_root=tmp_path, dry_run=False)
assert result["counts"]["rows_inserted"] >= 1
src_url = (await db.execute(
select(Source.url).where(Source.platform == platform)
)).scalar_one()
assert src_url == expected_url
# ImageProvenance row was created.
prov_count = (await db.execute(
select(func.count(ImageProvenance.id))
.join(Source, Source.id == ImageProvenance.source_id)
.where(Source.platform == platform)
)).scalar_one()
assert prov_count == 1
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_image_posts_dry_run_makes_no_writes(db, tmp_path): async def test_image_posts_dry_run_makes_no_writes(db, tmp_path):
sha = "2" * 64 sha = "2" * 64