diff --git a/backend/app/api/import_admin.py b/backend/app/api/import_admin.py index 2ebeb9b..7ad3fa7 100644 --- a/backend/app/api/import_admin.py +++ b/backend/app/api/import_admin.py @@ -128,6 +128,43 @@ async def retry_failed(): return jsonify({"retried": len(failed)}) +@import_admin_bp.route("/tasks//refetch", methods=["POST"]) +async def refetch_task(task_id: int): + """Layer-2 one-shot re-download: delete the (corrupt) file behind a + failed import task and re-run its source's downloader to fetch a + fresh copy. Only works for files that resolve to an enabled, + real-URL subscription Source; filesystem-only imports return + no_source. + + Returns one of: refetch_queued (+source_id) / no_source / + already_refetched / not_found / not_failed. + """ + async with get_session() as session: + result = await session.run_sync(_refetch_task_sync, task_id) + if result["status"] == "not_found": + return jsonify(result), 404 + if result["status"] == "not_failed": + return jsonify(result), 400 + return jsonify(result) + + +def _refetch_task_sync(session, task_id: int) -> dict: + from pathlib import Path + + from ..models import ImportSettings + from ..services.refetch_service import attempt_refetch + + task = session.get(ImportTask, task_id) + if task is None: + return {"status": "not_found"} + if task.status != "failed": + return {"status": "not_failed"} + settings = session.execute( + select(ImportSettings).where(ImportSettings.id == 1) + ).scalar_one() + return attempt_refetch(session, task, Path(settings.import_scan_path)) + + @import_admin_bp.route("/clear-stuck", methods=["POST"]) async def clear_stuck(): """Force any non-terminal ImportTask (status in pending/queued/ diff --git a/backend/app/services/refetch_service.py b/backend/app/services/refetch_service.py new file mode 100644 index 0000000..41832bc --- /dev/null +++ b/backend/app/services/refetch_service.py @@ -0,0 +1,106 @@ +"""Layer-2 one-shot re-download remediation for corrupt imported files. + +When an import fails on a file that came from a known, pollable +subscription Source, deleting the bad copy and re-running the source's +downloader can fetch a fresh, unblemished copy. This only helps when: + + - the corruption is in transit / on disk (not at the source), AND + - the file resolves to an ENABLED Source with a real feed URL + (a `sidecar::` synthetic anchor is not pollable), + AND + - we haven't already re-fetched this task once (bounded by + ImportTask.refetched so source-side corruption can't loop). + +Filesystem-only imports with no resolvable Source return 'no_source' — +the operator's only remediation there is to replace the file on disk. + +Operator-requested 2026-05-28 (Layer 2). +""" + +import json +import logging +from pathlib import Path + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from ..models import Artist, ImportTask, Source +from ..utils.paths import derive_top_level_artist +from ..utils.sidecar import find_sidecar, parse_sidecar +from ..utils.slug import slugify + +log = logging.getLogger(__name__) + + +def resolve_refetch_source( + session: Session, source_path: str, import_root: Path, +) -> Source | None: + """Find an enabled, real-URL Source for the file's (artist, platform), + or None when nothing re-pollable resolves.""" + path = Path(source_path) + sc = find_sidecar(path) + if sc is None: + return None + try: + data = json.loads(sc.read_text("utf-8")) + except (OSError, json.JSONDecodeError): + return None + if not isinstance(data, dict): + return None + sd = parse_sidecar(data) + if not sd.platform: + return None + artist_name = derive_top_level_artist(path, import_root) + if not artist_name: + return None + artist = session.execute( + select(Artist).where(Artist.slug == slugify(artist_name)) + ).scalar_one_or_none() + if artist is None: + return None + src = session.execute( + select(Source) + .where( + Source.artist_id == artist.id, + Source.platform == sd.platform, + Source.enabled.is_(True), + ) + .order_by(Source.id.asc()) + ).scalars().first() + if src is None: + return None + if (src.url or "").startswith("sidecar:"): + return None # synthetic anchor — not a pollable feed + return src + + +def attempt_refetch( + session: Session, task: ImportTask, import_root: Path, +) -> dict: + """Delete the corrupt file, mark the task refetched, and trigger ONE + source re-check. Idempotent/bounded: a task already refetched (or + with no resolvable Source) is a no-op. Commits.""" + if task.refetched: + return {"status": "already_refetched"} + src = resolve_refetch_source(session, task.source_path, import_root) + if src is None: + return {"status": "no_source"} + + # Remove the bad copy so gallery-dl (skip_existing) re-fetches it on + # the source re-check instead of skipping the still-present corrupt + # file. + try: + Path(task.source_path).unlink(missing_ok=True) + except OSError as exc: + log.warning("refetch unlink failed for %s: %s", task.source_path, exc) + + task.refetched = True + session.add(task) + session.commit() + + # Lazy import to avoid a tasks→services→tasks import cycle at module + # load. download_source.delay() is sync-safe in any context. + from ..tasks.download import download_source + + download_source.delay(src.id) + return {"status": "refetch_queued", "source_id": src.id} diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index f864873..4584d01 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -1,6 +1,7 @@ """Periodic maintenance: recover stuck import tasks, garbage-collect old finished tasks.""" import logging +import os import subprocess from datetime import UTC, datetime, timedelta from pathlib import Path @@ -183,6 +184,28 @@ def recover_interrupted_tasks() -> int: for tid, task_type in stuck: enqueue_import(tid, task_type) + # Layer-2 auto re-download (env-gated, default OFF). For each + # poison-pill row that resolves to a pollable Source, delete the + # bad file and trigger ONE source re-check to fetch a fresh + # copy. Bounded by ImportTask.refetched so source-side + # corruption can't loop. The 'failed' row stays as history; the + # re-downloaded file re-imports as a fresh task on the next scan. + if poison_ids and os.environ.get("FC_AUTO_REFETCH_CORRUPT", "0") == "1": + from ..models import ImportSettings + from ..services.refetch_service import attempt_refetch + import_root = Path(session.execute( + select(ImportSettings.import_scan_path) + .where(ImportSettings.id == 1) + ).scalar_one()) + for pid in poison_ids: + ptask = session.get(ImportTask, pid) + if ptask is None: + continue + try: + attempt_refetch(session, ptask, import_root) + except Exception as exc: # noqa: BLE001 — best-effort + log.warning("auto-refetch failed for task %s: %s", pid, exc) + return len(stuck) + len(poison_ids) + orphan_count diff --git a/frontend/src/components/settings/ImportTaskList.vue b/frontend/src/components/settings/ImportTaskList.vue index 97bbebf..4eadf3c 100644 --- a/frontend/src/components/settings/ImportTaskList.vue +++ b/frontend/src/components/settings/ImportTaskList.vue @@ -51,6 +51,19 @@ title="Click for full error" >{{ shorten(item.error, 60) }} +
Load more @@ -149,9 +162,29 @@ const headers = [ { title: 'Source', key: 'source_path', sortable: false }, { title: 'Size', key: 'size_bytes', sortable: false, width: 90 }, { title: 'Created', key: 'created_at', sortable: false, width: 150 }, - { title: 'Note', key: 'error', sortable: false } + { title: 'Note', key: 'error', sortable: false }, + { title: '', key: 'actions', sortable: false, width: 56 } ] +const refetching = ref(null) +const _REFETCH_MSG = { + refetch_queued: { text: 'Re-fetch queued — re-downloading from source', type: 'success' }, + no_source: { text: 'No re-fetchable source (filesystem import — replace the file manually)', type: 'info' }, + already_refetched: { text: 'Already re-fetched once', type: 'info' }, +} +async function onRefetch(item) { + refetching.value = item.id + try { + const res = await store.refetchTask(item.id) + const msg = _REFETCH_MSG[res.status] || { text: `Re-fetch: ${res.status}`, type: 'info' } + window.__fcToast?.(msg) + } catch (e) { + window.__fcToast?.({ text: `Re-fetch failed: ${e.message}`, type: 'error' }) + } finally { + refetching.value = null + } +} + 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' diff --git a/frontend/src/stores/import.js b/frontend/src/stores/import.js index 64e5146..98d4d56 100644 --- a/frontend/src/stores/import.js +++ b/frontend/src/stores/import.js @@ -146,6 +146,15 @@ export const useImportStore = defineStore('import', () => { return body } + // Layer-2 one-shot re-download for a failed task's (corrupt) file. + // Returns the endpoint's status dict (refetch_queued / no_source / + // already_refetched). Caller surfaces it as a toast. + async function refetchTask(taskId) { + const body = await api.post(`/api/import/tasks/${taskId}/refetch`) + await loadTasks(true) + return body + } + const hasMore = computed(() => tasksNextCursor.value !== null) return { @@ -155,6 +164,7 @@ export const useImportStore = defineStore('import', () => { triggerError, loadSettings, patchSettings, refreshStatus, triggerScan, - loadTasks, setStatusFilter, retryFailed, clearCompleted, clearStuck + loadTasks, setStatusFilter, retryFailed, clearCompleted, clearStuck, + refetchTask, } }) diff --git a/tests/test_api_import_admin.py b/tests/test_api_import_admin.py index f5512aa..f033fd5 100644 --- a/tests/test_api_import_admin.py +++ b/tests/test_api_import_admin.py @@ -171,6 +171,115 @@ async def test_trigger_still_rejects_unknown_mode(client): assert resp.status_code == 400 +@pytest.mark.asyncio +async def test_refetch_404_for_unknown_task(client): + resp = await client.post("/api/import/tasks/999999/refetch") + assert resp.status_code == 404 + + +@pytest.mark.asyncio +async def test_refetch_400_for_non_failed_task(client, db): + batch = ImportBatch(triggered_by="manual", source_path="/import", scan_mode="quick") + db.add(batch) + await db.flush() + task = ImportTask( + batch_id=batch.id, source_path="/x.jpg", task_type="media", + status="complete", finished_at=datetime.now(UTC), + ) + db.add(task) + await db.commit() + resp = await client.post(f"/api/import/tasks/{task.id}/refetch") + assert resp.status_code == 400 + assert (await resp.get_json())["status"] == "not_failed" + + +@pytest.mark.asyncio +async def test_refetch_no_source_when_unresolvable(client, db): + """A failed task whose file has no sidecar / no resolvable Source + returns no_source (filesystem-only import — nothing to re-poll).""" + batch = ImportBatch(triggered_by="manual", source_path="/import", scan_mode="quick") + db.add(batch) + await db.flush() + task = ImportTask( + batch_id=batch.id, source_path="/import/nowhere/x.jpg", + task_type="media", status="failed", finished_at=datetime.now(UTC), + ) + db.add(task) + await db.commit() + resp = await client.post(f"/api/import/tasks/{task.id}/refetch") + assert resp.status_code == 200 + assert (await resp.get_json())["status"] == "no_source" + + +@pytest.mark.asyncio +async def test_refetch_queued_with_resolvable_source(client, db, tmp_path, monkeypatch): + """A failed task whose file resolves (via sidecar → artist+platform) + to an enabled, real-URL Source: the file is deleted, the task is + marked refetched, and ONE source re-check is queued.""" + import json as _json + + from sqlalchemy import update as _update + + from backend.app.models import Artist, ImportSettings, Source + from backend.app.tasks import download as download_mod + + # Stub the downloader so the eager test doesn't run a real fetch. + dispatched = [] + monkeypatch.setattr(download_mod.download_source, "delay", dispatched.append) + + # import_root//post.jpg + sidecar identifying the platform. + import_root = tmp_path / "import" + artist_dir = import_root / "Maewix" + artist_dir.mkdir(parents=True) + media = artist_dir / "post.jpg" + media.write_bytes(b"corrupt-bytes") + (artist_dir / "post.jpg.json").write_text( + _json.dumps({"category": "patreon", "post_id": 123}) + ) + + # import_settings(id=1) is migration-seeded; point its scan path at + # our tmp import root rather than inserting a conflicting row. + await db.execute( + _update(ImportSettings).where(ImportSettings.id == 1) + .values(import_scan_path=str(import_root)) + ) + artist = Artist(name="Maewix", slug="maewix") + db.add(artist) + await db.flush() + db.add(Source( + artist_id=artist.id, platform="patreon", + url="https://www.patreon.com/maewix", enabled=True, + config_overrides={}, + )) + batch = ImportBatch(triggered_by="manual", source_path=str(import_root), scan_mode="quick") + db.add(batch) + await db.flush() + task = ImportTask( + batch_id=batch.id, source_path=str(media), task_type="media", + status="failed", finished_at=datetime.now(UTC), + ) + db.add(task) + await db.commit() + + resp = await client.post(f"/api/import/tasks/{task.id}/refetch") + assert resp.status_code == 200 + body = await resp.get_json() + assert body["status"] == "refetch_queued" + assert len(dispatched) == 1 + assert not media.exists() # corrupt copy removed for re-fetch + + from sqlalchemy import select as _select + refetched = (await db.execute( + _select(ImportTask.refetched).where(ImportTask.id == task.id) + )).scalar_one() + assert refetched is True + + # Second attempt is a no-op (bounded to one). + resp2 = await client.post(f"/api/import/tasks/{task.id}/refetch") + assert (await resp2.get_json())["status"] == "already_refetched" + assert len(dispatched) == 1 + + @pytest.mark.asyncio async def test_trigger_accepts_verify(client, monkeypatch): # Stub the verify task's dispatch so the API contract is asserted