From 9d5abb09f67a2356bbad93fe5031c701f74c1065 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 25 May 2026 02:00:29 -0400 Subject: [PATCH 1/9] 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) --- backend/app/tasks/maintenance.py | 78 +++++++++++++++++++------- tests/test_maintenance.py | 96 ++++++++++++++++++++++++++++++++ 2 files changed, 154 insertions(+), 20 deletions(-) diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 1a2e028..4ef590a 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -16,6 +16,7 @@ from ._sync_engine import sync_session_factory as _sync_session_factory log = logging.getLogger(__name__) STUCK_THRESHOLD_MINUTES = 5 +ORPHAN_PENDING_THRESHOLD_MINUTES = 30 OLD_TASK_DAYS = 7 PHASH_PAGE = 500 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") 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 - files; even a large-video transcode caps at the per-task soft_time_limit - (5 min) defined on the task itself. Anything still 'processing' after - that window is a confirmed crash (worker died, DB disconnect mid-flush, - OOM) and must be recycled. Was 30 min historically; tightened - 2026-05-24 after operator hit a 2224-row zombie pile during the IR - migration scan. + 1. 'processing' > 5 min — worker crash mid-import. Re-queue via + .delay() and let the import retry. Was 30 min historically; + tightened 2026-05-24 after operator hit a 2224-row zombie pile. + import_media_file is sub-second for the vast majority of files and + capped at the per-task soft_time_limit (5 min), so anything still + 'processing' after that window is a confirmed crash. + + 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() - 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: stuck_ids = session.execute( select(ImportTask.id) .where(ImportTask.status == "processing") - .where(ImportTask.started_at < cutoff) + .where(ImportTask.started_at < processing_cutoff) ).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 - session.execute( - update(ImportTask) - .where(ImportTask.id.in_(stuck_ids)) - .values(status="queued", started_at=None, error="recovered from stuck state") - ) + if stuck_ids: + session.execute( + update(ImportTask) + .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() - from .import_file import import_media_file - for tid in stuck_ids: - import_media_file.delay(tid) + if stuck_ids: + from .import_file import import_media_file + 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") diff --git a/tests/test_maintenance.py b/tests/test_maintenance.py index 6e08a91..7fc04e9 100644 --- a/tests/test_maintenance.py +++ b/tests/test_maintenance.py @@ -68,6 +68,102 @@ def test_recover_interrupted_only_old(db_sync, monkeypatch): 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): batch_id = _make_batch(db_sync) now = datetime.now(UTC) From 3b3e7565fb65dcc351c2a53040ad91c432caa275 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 25 May 2026 02:25:30 -0400 Subject: [PATCH 2/9] fix(ml): align tagger + downloader with Camie v2 actual layout (model.onnx -> camie-tagger-v2.onnx + JSON metadata + ImageNet preprocessing + sigmoid on refined output) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- backend/app/scripts/download_models.py | 23 ++++- backend/app/services/ml/tagger.py | 126 ++++++++++++++++--------- tests/test_ml_tagger.py | 5 +- 3 files changed, 107 insertions(+), 47 deletions(-) diff --git a/backend/app/scripts/download_models.py b/backend/app/scripts/download_models.py index 39131a1..4b5e6d3 100644 --- a/backend/app/scripts/download_models.py +++ b/backend/app/scripts/download_models.py @@ -25,12 +25,31 @@ def _snapshot(repo_id: str, dest: Path, allow_patterns: list[str] | None) -> Non 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" - 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}") return 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: diff --git a/backend/app/services/ml/tagger.py b/backend/app/services/ml/tagger.py index 92de3a3..5277b44 100644 --- a/backend/app/services/ml/tagger.py +++ b/backend/app/services/ml/tagger.py @@ -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 (set by the FC-1 entrypoint). -Camie's selected_tags.csv columns: tag_id,name,category,count -where category is a string: general|character|copyright|artist|meta|rating|year -(unlike WD14's integer Danbooru category ids). +v2 layout reference: HuggingFace Camais03/camie-tagger-v2 root has +camie-tagger-v2.onnx (789 MB) + camie-tagger-v2-metadata.json (7.77 MB) ++ 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 from dataclasses import dataclass 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_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). 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. 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) class TagPrediction: @@ -51,34 +61,48 @@ class Tagger: def __init__(self, model_dir: Path | None = None): self._model_dir = model_dir or _MODEL_DIR 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._output_name: str | None = None - self._input_size: int = 448 + self._input_size: int = 512 def load(self) -> None: if self._session is not None: return - model_path = self._model_dir / "model.onnx" - tags_path = self._model_dir / "selected_tags.csv" + model_path = self._model_dir / _MODEL_FILE + meta_path = self._model_dir / _METADATA_FILE if not model_path.is_file(): 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." ) - if not tags_path.is_file(): + if not meta_path.is_file(): 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." ) - tag_meta: list[dict] = [] - with open(tags_path, newline="") as f: - reader = csv.DictReader(f) - for row in reader: - tag_meta.append( - {"name": row["name"], "category": row["category"]} - ) + with open(meta_path) as f: + metadata = json.load(f) + + # Per Camie v2 onnx_inference.py: idx_to_tag is keyed by str(idx); + # tag_to_category maps tag_name -> category. Project to two parallel + # 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 # missing-model RuntimeError still fires first in environments @@ -89,51 +113,65 @@ class Tagger: str(model_path), providers=["CPUExecutionProvider"] ) 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. - self._tag_meta = tag_meta + self._tag_names = names + self._tag_categories = cats self._session = session def _preprocess(self, image_path: Path) -> np.ndarray: img = Image.open(image_path) - # Camie handles RGBA natively but we still composite onto white so - # transparency doesn't bias the model (same as IR's WD14 path). - if img.mode != "RGBA": - img = img.convert("RGBA") - bg = Image.new("RGBA", img.size, (255, 255, 255, 255)) - bg.paste(img, mask=img.split()[3]) - img = bg.convert("RGB") + # Composite RGBA onto neutral so transparency doesn't bias the model. + if img.mode == "RGBA": + bg = Image.new("RGBA", img.size, (255, 255, 255, 255)) + bg.paste(img, mask=img.split()[3]) + img = bg.convert("RGB") + elif img.mode != "RGB": + img = img.convert("RGB") + # Pad to square with ImageNet-mean color, then bicubic resize. w, h = img.size 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 = square.resize( (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]: - """Run Camie on one image. Returns {name: TagPrediction}, only - entries with confidence >= STORE_FLOOR (across all categories — - the suggestion service does category filtering later).""" + """Run Camie v2 on one image. Returns {name: TagPrediction} with + confidence >= STORE_FLOOR (across all categories — the suggestion + 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() 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] = {} - for idx, score in enumerate(out): + names = self._tag_names + cats = self._tag_categories + for idx, score in enumerate(probs): conf = float(score) if conf < STORE_FLOOR: continue - meta = self._tag_meta[idx] - results[meta["name"]] = TagPrediction( - name=meta["name"], category=meta["category"], confidence=conf + if idx >= len(names): + # Output longer than metadata declared — shouldn't happen but + # 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 diff --git a/tests/test_ml_tagger.py b/tests/test_ml_tagger.py index 79cfe51..17c449c 100644 --- a/tests/test_ml_tagger.py +++ b/tests/test_ml_tagger.py @@ -40,5 +40,8 @@ def test_get_tagger_singleton(): def test_load_raises_when_model_missing(tmp_path): t = Tagger(model_dir=tmp_path / "nonexistent") - with pytest.raises(RuntimeError, match="model.onnx missing"): + # Match the trailing "missing at " 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() From 52445eb501596479d246018848c5eb42deb9ccd9 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 25 May 2026 02:32:05 -0400 Subject: [PATCH 3/9] 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) --- frontend/src/components/discovery/ArtistCard.vue | 13 +++++++++++-- frontend/src/components/discovery/TagCard.vue | 14 ++++++++++++-- frontend/src/views/ArtistsView.vue | 2 +- frontend/src/views/TagsView.vue | 2 +- 4 files changed, 25 insertions(+), 6 deletions(-) diff --git a/frontend/src/components/discovery/ArtistCard.vue b/frontend/src/components/discovery/ArtistCard.vue index 55564cb..6c4bace 100644 --- a/frontend/src/components/discovery/ArtistCard.vue +++ b/frontend/src/components/discovery/ArtistCard.vue @@ -38,9 +38,18 @@ function onCardClick() { .fc-artistcard { cursor: pointer; } .fc-artistcard__previews { 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 { grid-column: 1 / -1; display: flex; align-items: center; justify-content: center; diff --git a/frontend/src/components/discovery/TagCard.vue b/frontend/src/components/discovery/TagCard.vue index 4475aab..9eb27ad 100644 --- a/frontend/src/components/discovery/TagCard.vue +++ b/frontend/src/components/discovery/TagCard.vue @@ -106,9 +106,19 @@ function submit() { .fc-tagcard { cursor: pointer; } .fc-tagcard__previews { 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 { grid-column: 1 / -1; display: flex; align-items: center; justify-content: center; diff --git a/frontend/src/views/ArtistsView.vue b/frontend/src/views/ArtistsView.vue index dc32f82..94913a1 100644 --- a/frontend/src/views/ArtistsView.vue +++ b/frontend/src/views/ArtistsView.vue @@ -84,7 +84,7 @@ onUnmounted(() => observer && observer.disconnect()) } .fc-artists__grid { display: grid; - grid-template-columns: repeat(auto-fill, minmax(220px, 1fr)); + grid-template-columns: repeat(auto-fill, minmax(440px, 1fr)); gap: 12px; } .fc-artists__sentinel { diff --git a/frontend/src/views/TagsView.vue b/frontend/src/views/TagsView.vue index 552c7ea..0d748ef 100644 --- a/frontend/src/views/TagsView.vue +++ b/frontend/src/views/TagsView.vue @@ -236,7 +236,7 @@ async function onDeleteTagConfirm() { } .fc-tags__grid { display: grid; - grid-template-columns: repeat(auto-fill, minmax(220px, 1fr)); + grid-template-columns: repeat(auto-fill, minmax(440px, 1fr)); gap: 12px; } .fc-tags__sentinel { From 68cffce322f5bd9422fbf15538f50daa50610855 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 25 May 2026 02:34:36 -0400 Subject: [PATCH 4/9] fix(importer): catch PIL OSError during transparency + phash blocks, skip as invalid_image instead of letting Celery autoretry loop forever MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- backend/app/services/importer.py | 25 +++++++++++++-- tests/test_importer.py | 55 ++++++++++++++++++++++++++++++++ 2 files changed, 77 insertions(+), 3 deletions(-) diff --git a/backend/app/services/importer.py b/backend/app/services/importer.py index 51550c2..2548d5b 100644 --- a/backend/app/services/importer.py +++ b/backend/app/services/importer.py @@ -280,7 +280,18 @@ class Importer: ) 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: return ImportResult( status="skipped", skip_reason=SkipReason.too_transparent, @@ -302,8 +313,16 @@ class Importer: # Perceptual near-dup (images only; videos keep phash NULL). phash = None if not is_video(source): - with Image.open(source) as im: - phash = compute_phash(im) + try: + 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: cand_rows = self.session.execute( select( diff --git a/tests/test_importer.py b/tests/test_importer.py index b809de6..ab69ea0 100644 --- a/tests/test_importer.py +++ b/tests/test_importer.py @@ -134,3 +134,58 @@ def test_root_level_file_has_no_artist(importer, import_layout): importer.import_one(src) artists = importer.session.execute(select(Artist)).scalars().all() 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 "") From 6b1bb8764704f4f1420b824b1d065bb259a76fae Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 25 May 2026 11:12:55 -0400 Subject: [PATCH 5/9] =?UTF-8?q?fix(tests):=20update=20test=5Fensure=5Fcami?= =?UTF-8?q?e=5Fskips=5Fwhen=5Fpresent=20to=20v2=20filenames=20(camie-tagge?= =?UTF-8?q?r-v2.onnx=20+=20camie-tagger-v2-metadata.json)=20=E2=80=94=20pi?= =?UTF-8?q?nned-test=20bounce=20from=203b3e756?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/test_download_models.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/tests/test_download_models.py b/tests/test_download_models.py index edd13be..366bd81 100644 --- a/tests/test_download_models.py +++ b/tests/test_download_models.py @@ -9,11 +9,16 @@ from backend.app.scripts import download_models as dm 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) camie = tmp_path / "camie" camie.mkdir(parents=True) - (camie / "model.onnx").write_bytes(b"x") - (camie / "selected_tags.csv").write_text("tag_id,name,category,count\n") + (camie / "camie-tagger-v2.onnx").write_bytes(b"x") + (camie / "camie-tagger-v2-metadata.json").write_text("{}") with patch.object(dm, "_snapshot") as snap: dm.ensure_camie() snap.assert_not_called() From 9f54efdedfc81645525ee79ee61ff31dfb16e469 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 25 May 2026 11:25:45 -0400 Subject: [PATCH 6/9] fix(migrators): tag_apply phase 4 now covers deviantart + pixiv (was silently dropping IR PostMetadata from those platforms) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _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) --- backend/app/services/migrators/tag_apply.py | 18 +++++++++-- tests/test_tag_apply.py | 36 +++++++++++++++++++++ 2 files changed, 51 insertions(+), 3 deletions(-) diff --git a/backend/app/services/migrators/tag_apply.py b/backend/app/services/migrators/tag_apply.py index 5db92ff..cfcb585 100644 --- a/backend/app/services/migrators/tag_apply.py +++ b/backend/app/services/migrators/tag_apply.py @@ -37,13 +37,25 @@ from ...utils.slug import slugify from .ir_ingest import manifest_path # Per-platform artist-profile URL — used as Source.url when restoring -# IR PostMetadata into FC. Keep this table in sync with -# backend/app/services/extension_service.py:_PLATFORM_PATTERNS and -# extension/lib/platforms.js. +# IR PostMetadata into FC. Must cover every platform that +# backend/app/services/extension_service.py:_PLATFORM_PATTERNS +# 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 = { "patreon": "https://www.patreon.com/{slug}", "subscribestar": "https://www.subscribestar.com/{slug}", "hentaifoundry": "https://www.hentai-foundry.com/user/{slug}", + "deviantart": "https://www.deviantart.com/{slug}", + "pixiv": "https://www.pixiv.net/users/{slug}", } diff --git a/tests/test_tag_apply.py b/tests/test_tag_apply.py index cee72c0..cf1a5c2 100644 --- a/tests/test_tag_apply.py +++ b/tests/test_tag_apply.py @@ -281,6 +281,42 @@ async def test_image_posts_unknown_platform_skipped(db, tmp_path): 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 async def test_image_posts_dry_run_makes_no_writes(db, tmp_path): sha = "2" * 64 From c36103255475a39bee2ce666b991b9a02d012136 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 25 May 2026 12:30:18 -0400 Subject: [PATCH 7/9] =?UTF-8?q?feat(gallery):=20sort/group/jump=20by=20COA?= =?UTF-8?q?LESCE(post.post=5Fdate,=20image=5Frecord.created=5Fat)=20?= =?UTF-8?q?=E2=80=94=20surface=20migrated=20content=20at=20its=20original?= =?UTF-8?q?=20publish=20date,=20not=20FC=20scan=20date?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- backend/app/api/gallery.py | 2 + backend/app/services/gallery_service.py | 167 +++++++++++++------- backend/app/services/migrators/tag_apply.py | 21 ++- tests/test_gallery_service.py | 132 ++++++++++++++++ tests/test_tag_apply.py | 61 +++++++ 5 files changed, 328 insertions(+), 55 deletions(-) diff --git a/backend/app/api/gallery.py b/backend/app/api/gallery.py index b14d414..8de2ed8 100644 --- a/backend/app/api/gallery.py +++ b/backend/app/api/gallery.py @@ -42,6 +42,8 @@ async def scroll(): "width": i.width, "height": i.height, "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, "artist": i.artist, } diff --git a/backend/app/services/gallery_service.py b/backend/app/services/gallery_service.py index 8337390..0a46fec 100644 --- a/backend/app/services/gallery_service.py +++ b/backend/app/services/gallery_service.py @@ -1,26 +1,34 @@ """Cursor-paginated gallery queries. -Cursor format: opaque base64-encoded ":". -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 -a ValueError; the API layer translates that to HTTP 400. +Cursor format: opaque base64-encoded ":". + +Pagination key is (effective_date DESC, id DESC) where effective_date is +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 from dataclasses import dataclass 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 ..models import Artist, ImageProvenance, ImageRecord, Source, Tag +from ..models import Artist, ImageProvenance, ImageRecord, Post, Source, Tag from ..models.tag import image_tag CURSOR_SEPARATOR = "|" -def encode_cursor(created_at: datetime, image_id: int) -> str: - raw = f"{created_at.isoformat()}{CURSOR_SEPARATOR}{image_id}" +def encode_cursor(effective_date: datetime, image_id: int) -> str: + raw = f"{effective_date.isoformat()}{CURSOR_SEPARATOR}{image_id}" 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 +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) class GalleryImage: id: int @@ -41,7 +69,9 @@ class GalleryImage: mime: str width: 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 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): """Correlated EXISTS clause (NOT a join) so an image with multiple 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: return exists().where( ImageProvenance.image_record_id == ImageRecord.id, @@ -125,7 +155,9 @@ class GalleryService: raise ValueError("limit must be between 1 and 200") _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: stmt = stmt.join(image_tag, image_tag.c.image_record_id == ImageRecord.id).where( image_tag.c.tag_id == tag_id @@ -138,34 +170,38 @@ class GalleryService: cur_ts, cur_id = decode_cursor(cursor) stmt = stmt.where( or_( - ImageRecord.created_at < cur_ts, - and_(ImageRecord.created_at == cur_ts, ImageRecord.id < cur_id), + eff < cur_ts, + and_(eff == cur_ts, ImageRecord.id < cur_id), ) ) - stmt = stmt.order_by(ImageRecord.created_at.desc(), ImageRecord.id.desc()).limit(limit + 1) - rows = (await self.session.execute(stmt)).scalars().all() + stmt = stmt.order_by(eff.desc(), ImageRecord.id.desc()).limit(limit + 1) + rows = (await self.session.execute(stmt)).all() next_cursor = None if len(rows) > limit: - last = rows[limit - 1] - next_cursor = encode_cursor(last.created_at, last.id) + last_record, _last_posted_at, last_eff = rows[limit - 1] + next_cursor = encode_cursor(last_eff, last_record.id) 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 = [ GalleryImage( - id=r.id, - path=r.path, - sha256=r.sha256, - mime=r.mime, - width=r.width, - height=r.height, - created_at=r.created_at, - thumbnail_url=thumbnail_url(r.sha256, r.mime), - artist=artists.get(r.id), + id=record.id, + path=record.path, + sha256=record.sha256, + mime=record.mime, + width=record.width, + height=record.height, + created_at=record.created_at, + effective_date=eff_date, + 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( images=images, @@ -179,11 +215,13 @@ class GalleryService: post_id: int | None = None, artist_id: int | None = None, ) -> list[TimelineBucket]: - year_col = func.date_part("year", ImageRecord.created_at).label("yr") - month_col = func.date_part("month", ImageRecord.created_at).label("mo") + eff = _effective_date_col() + year_col = func.date_part("year", eff).label("yr") + month_col = func.date_part("month", eff).label("mo") stmt = select( 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) if tag_id is not None: 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, ) -> str | None: """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 - stmt = select(ImageRecord).where( - extract("year", ImageRecord.created_at) == year, - extract("month", ImageRecord.created_at) == month, + eff = _effective_date_col() + stmt = select(ImageRecord, eff.label("eff")).where( + extract("year", eff) == year, + extract("month", eff) == month, ) + stmt = _outer_join_primary_post(stmt) _require_single_filter(tag_id, post_id, artist_id) if tag_id is not None: 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) if prov is not None: stmt = stmt.where(prov) - stmt = stmt.order_by(ImageRecord.created_at.desc(), ImageRecord.id.desc()).limit(1) - first = (await self.session.execute(stmt)).scalar_one_or_none() + stmt = stmt.order_by(eff.desc(), ImageRecord.id.desc()).limit(1) + first = (await self.session.execute(stmt)).first() if first is None: return None + record, eff_date = first # Cursor is exclusive; we encode a cursor with id+1 so the row itself # 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: record = await self.session.get(ImageRecord, image_id) @@ -236,6 +278,13 @@ class GalleryService: .order_by(Tag.kind.asc(), Tag.name.asc()) ) 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 " 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) # Direct artist FK — used by the modal's ProvenancePanel as a # fallback when ImageProvenance is empty (i.e., filesystem- @@ -256,6 +305,7 @@ class GalleryService: "size_bytes": record.size_bytes, "integrity_status": record.integrity_status, "created_at": record.created_at.isoformat(), + "posted_at": posted_at.isoformat() if posted_at else None, "thumbnail_url": thumbnail_url(record.sha256, record.mime), "image_url": f"/images/{record.path.split('/images/', 1)[-1]}", "artist": ( @@ -275,34 +325,41 @@ class GalleryService: } async def _neighbors(self, record: ImageRecord) -> dict: - prev_stmt = ( - select(ImageRecord.id) - .where( + # Compute the boundary image's effective_date in Python (one query + # below + the SELECT we already have on `record`) and use it for + # 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_( - ImageRecord.created_at > record.created_at, + eff > boundary_eff, and_( - ImageRecord.created_at == record.created_at, + eff == boundary_eff, ImageRecord.id > record.id, ), ) ) - .order_by(ImageRecord.created_at.asc(), ImageRecord.id.asc()) - .limit(1) - ) - next_stmt = ( - select(ImageRecord.id) - .where( + ).order_by(eff.asc(), ImageRecord.id.asc()).limit(1) + next_stmt = _outer_join_primary_post( + select(ImageRecord.id).where( or_( - ImageRecord.created_at < record.created_at, + eff < boundary_eff, and_( - ImageRecord.created_at == record.created_at, + eff == boundary_eff, ImageRecord.id < record.id, ), ) ) - .order_by(ImageRecord.created_at.desc(), ImageRecord.id.desc()) - .limit(1) - ) + ).order_by(eff.desc(), ImageRecord.id.desc()).limit(1) prev_id = (await self.session.execute(prev_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} @@ -311,9 +368,11 @@ class GalleryService: def _group_by_year_month( images: list[GalleryImage], ) -> 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]]] = [] 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: groups[-1][2].append(img.id) else: diff --git a/backend/app/services/migrators/tag_apply.py b/backend/app/services/migrators/tag_apply.py index cfcb585..3f15c88 100644 --- a/backend/app/services/migrators/tag_apply.py +++ b/backend/app/services/migrators/tag_apply.py @@ -122,7 +122,14 @@ async def _ensure_provenance( db: AsyncSession, *, image_id: int, post_id: int, source_id: int, dry_run: 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( select(ImageProvenance.id).where( ImageProvenance.image_record_id == image_id, @@ -130,6 +137,18 @@ async def _ensure_provenance( ImageProvenance.source_id == source_id, ) )).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: return False if dry_run: diff --git a/tests/test_gallery_service.py b/tests/test_gallery_service.py index ecba6bb..705747c 100644 --- a/tests/test_gallery_service.py +++ b/tests/test_gallery_service.py @@ -133,3 +133,135 @@ async def test_get_image_with_tags_includes_integrity_status(db): svc = GalleryService(db) payload = await svc.get_image_with_tags(img.id) 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"] diff --git a/tests/test_tag_apply.py b/tests/test_tag_apply.py index cf1a5c2..d1385c1 100644 --- a/tests/test_tag_apply.py +++ b/tests/test_tag_apply.py @@ -227,6 +227,67 @@ async def test_image_posts_creates_source_post_provenance(db, tmp_path): )).scalar_one() 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 async def test_image_posts_idempotent_on_rerun(db, tmp_path): From b6a917ac81be24a2fb384a02cfa15e6286d90d52 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 25 May 2026 12:37:07 -0400 Subject: [PATCH 8/9] =?UTF-8?q?feat(import):=20/api/import/clear-stuck=20e?= =?UTF-8?q?ndpoint=20+=20Clear=20stuck=20UI=20button=20=E2=80=94=20escape?= =?UTF-8?q?=20hatch=20for=20the=20autoretry-loop=20case=20the=20automatic?= =?UTF-8?q?=20sweep=20can't=20break?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- backend/app/api/import_admin.py | 77 +++++++++++++++++++ .../components/settings/ImportTaskList.vue | 40 ++++++++++ frontend/src/stores/import.js | 9 ++- tests/test_api_import_admin.py | 55 +++++++++++++ 4 files changed, 180 insertions(+), 1 deletion(-) diff --git a/backend/app/api/import_admin.py b/backend/app/api/import_admin.py index 6a16bd3..f4ecf07 100644 --- a/backend/app/api/import_admin.py +++ b/backend/app/api/import_admin.py @@ -120,6 +120,83 @@ async def retry_failed(): 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"]) async def clear_completed(): body = await request.get_json(silent=True) or {} diff --git a/frontend/src/components/settings/ImportTaskList.vue b/frontend/src/components/settings/ImportTaskList.vue index a78562d..c3edbd1 100644 --- a/frontend/src/components/settings/ImportTaskList.vue +++ b/frontend/src/components/settings/ImportTaskList.vue @@ -17,6 +17,12 @@ > Retry failed + + Clear stuck… + + + + + Clear stuck tasks + + + Force every pending / queued / processing task to + failed 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). + +

+ Tasks remain in the database with status=failed; + click Retry failed once the underlying cause is + resolved to re-queue them. +

+ + + + Cancel + Clear stuck + + + @@ -80,6 +111,7 @@ const store = useImportStore() const statusFilter = ref(null) const clearDialog = ref(false) const clearAgeDays = ref(7) +const clearStuckDialog = ref(false) const statusOptions = [ { title: 'All', value: null }, @@ -100,6 +132,9 @@ const headers = [ ] 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) { return { @@ -138,4 +173,9 @@ async function onClearConfirm() { await store.clearCompleted(clearAgeDays.value) clearDialog.value = false } +function onClearStuckOpen() { clearStuckDialog.value = true } +async function onClearStuckConfirm() { + await store.clearStuck() + clearStuckDialog.value = false +} diff --git a/frontend/src/stores/import.js b/frontend/src/stores/import.js index 861db25..9d4832e 100644 --- a/frontend/src/stores/import.js +++ b/frontend/src/stores/import.js @@ -92,6 +92,13 @@ export const useImportStore = defineStore('import', () => { 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) return { @@ -101,6 +108,6 @@ export const useImportStore = defineStore('import', () => { triggerError, loadSettings, patchSettings, refreshStatus, triggerScan, - loadTasks, setStatusFilter, retryFailed, clearCompleted + loadTasks, setStatusFilter, retryFailed, clearCompleted, clearStuck } }) diff --git a/tests/test_api_import_admin.py b/tests/test_api_import_admin.py index 7e15f4b..f5512aa 100644 --- a/tests/test_api_import_admin.py +++ b/tests/test_api_import_admin.py @@ -86,6 +86,61 @@ async def test_clear_completed(client, db): 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 async def test_trigger_accepts_deep(client, monkeypatch): # Stub the task dispatch — assert the API accepts 'deep' and forwards From 3a359f6c5ed1041be096152088eccdb8f2e0c72b Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 25 May 2026 12:43:37 -0400 Subject: [PATCH 9/9] =?UTF-8?q?fix(ui):=20Quick=20scan=20button=20always?= =?UTF-8?q?=20visible=20(disabled=20when=20active=20batch=20present)=20+?= =?UTF-8?q?=20inline=20Clear=20stuck=20action=20=E2=80=94=20operator=20was?= =?UTF-8?q?=20clicking=20a=20spinner=20area=20thinking=20it=20was=20the=20?= =?UTF-8?q?button=20because=20activeBatch=20hid=20the=20Quick=20scan=20but?= =?UTF-8?q?ton=20entirely?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.7 (1M context) --- .../settings/ImportTriggerPanel.vue | 56 +++++++++++++++---- 1 file changed, 44 insertions(+), 12 deletions(-) diff --git a/frontend/src/components/settings/ImportTriggerPanel.vue b/frontend/src/components/settings/ImportTriggerPanel.vue index 232df86..957aaf6 100644 --- a/frontend/src/components/settings/ImportTriggerPanel.vue +++ b/frontend/src/components/settings/ImportTriggerPanel.vue @@ -2,7 +2,7 @@ Trigger scan -
+
@@ -13,20 +13,40 @@ failed {{ store.activeBatch.failed }} / {{ store.activeBatch.total_files }} files + + + Clear stuck +
-
-

+ +

+ Run a quick scan of the import directory. Deep scan (pHash dedup, archives) lands in FC-2d. -

- - mdi-magnify-scan - Quick scan - - - {{ store.triggerError }} - -
+ + + An active batch is in progress. Wait for it to finish, or click + Clear stuck above if it has been wedged with no + measurable progress. + +

+ + + mdi-magnify-scan + Quick scan + + + + {{ store.triggerError }} + @@ -37,9 +57,21 @@ import { useImportStore } from '../../stores/import.js' const store = useImportStore() const busy = ref(false) +const clearing = ref(false) async function trigger() { busy.value = true 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 + } +}