From 2fc71cfe4e4033dcf0d5e520f6c6376aba34d38c Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Fri, 6 Nov 2026 10:31:37 -0500 Subject: [PATCH] feat: Cleanup can import files that are on disk but not in the library A download writes its files into the artist's folder and imports them in a later step. A run that stopped in between left complete files no record points at, and a walk that had already marked them seen never returned for them; on the live instance that is about 95 GB of art. "Import files the library is missing" (Cleanup, Duplicates & leftovers) walks every folder named for an artist, matched by slug so a differently cased folder still matches, and lists the images, videos and archives no image, attachment or queued import points at. Stores, sidecars, staging files and files changed in the last day are left alone. Preview shows the counts per artist; Import queues the same list as a library import batch, whose tasks file the folder's post record, attach each file where it lies with its folder's artist and source, and remove the ones the library already holds or the import filters reject. A file whose import crashed is kept. Co-Authored-By: Claude Opus 5.5 --- backend/app/api/admin.py | 15 ++ backend/app/services/library_scan.py | 181 ++++++++++++++++++ backend/app/tasks/admin.py | 26 +++ backend/app/tasks/import_file.py | 29 ++- backend/app/tasks/maintenance.py | 3 + .../components/settings/LibraryImportCard.vue | 150 +++++++++++++++ frontend/src/views/CleanupView.vue | 5 +- tests/test_api_admin.py | 19 ++ tests/test_library_scan.py | 123 ++++++++++++ tests/test_maintenance.py | 11 ++ 10 files changed, 559 insertions(+), 3 deletions(-) create mode 100644 backend/app/services/library_scan.py create mode 100644 frontend/src/components/settings/LibraryImportCard.vue create mode 100644 tests/test_library_scan.py diff --git a/backend/app/api/admin.py b/backend/app/api/admin.py index 41b865f..8457256 100644 --- a/backend/app/api/admin.py +++ b/backend/app/api/admin.py @@ -475,6 +475,21 @@ async def trigger_reclaim_attachments(): return _queued(async_result) +@admin_bp.route("/maintenance/import-unimported", methods=["POST"]) +async def trigger_import_unimported(): + """Import files in the artists' folders that the library has no record of. + Body {"dry_run": bool}; dry_run (the DEFAULT) lists what was found, per + artist, and changes nothing. dry_run=false queues them as an import batch + that attaches each file where it lies. Returns the Celery task id — poll + /maintenance/task-result/ for the summary.""" + from ..tasks.admin import import_unimported_files_task + + body = await request.get_json(silent=True) or {} + dry_run = bool(body.get("dry_run", True)) + async_result = import_unimported_files_task.delay(dry_run=dry_run) + return _queued(async_result) + + @admin_bp.route("/maintenance/repair-discord-downloads", methods=["POST"]) async def trigger_repair_discord_downloads(): """Clean re-download of the Discord files broken by the `None` naming diff --git a/backend/app/services/library_scan.py b/backend/app/services/library_scan.py new file mode 100644 index 0000000..010c99d --- /dev/null +++ b/backend/app/services/library_scan.py @@ -0,0 +1,181 @@ +"""Files in the artists' folders that the library has no record of. + +A download writes its files into the artist's folder and imports them in a +later step. A run that stops between the two (a restart, a time limit) leaves +complete files on disk that no image or attachment points at, and a walk that +has already marked them seen never comes back for them. `find_unimported` +lists those files; `queue_unimported` turns the same list into an import batch +that attaches each file where it lies (`scan_mode="library"`), so a file the +library already holds, or one the import filters reject, is removed instead. + +Only files under a folder named for an existing artist (matched by slug, so a +differently-cased folder still matches) are considered, and only images, +videos and archives. Files changed in the last day are left for the download +that may still be writing them. +""" +from __future__ import annotations + +import os +import stat +import time +from collections import defaultdict +from dataclasses import dataclass +from pathlib import Path + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from ..models import Artist, ImageRecord, ImportBatch, ImportTask, PostAttachment, Source +from ..utils.slug import slugify +from ..utils.staging import is_temp_staging_file +from .archive_extractor import is_archive +from .importer import is_supported + +# Top-level folders that hold the library's own stores, not an artist's files. +_STORE_DIRS = frozenset({"thumbs", "attachments", "cookies", "secrets"}) +MIN_AGE_SECONDS = 24 * 3600 +_ACTIVE_TASK_STATUSES = ("pending", "queued", "processing") +_SAMPLES_PER_ARTIST = 3 +_ARTISTS_LISTED = 50 + + +@dataclass(frozen=True) +class Unimported: + path: Path + artist_slug: str + task_type: str # "media" or "archive", as ImportTask.task_type + size: int + + +def _referenced_paths(session: Session) -> set[str]: + paths: set[str] = set() + for p, thumb in session.execute(select(ImageRecord.path, ImageRecord.thumbnail_path)): + paths.add(p) + if thumb: + paths.add(thumb) + paths.update(session.execute(select(PostAttachment.path)).scalars()) + paths.update(session.execute( + select(ImportTask.source_path).where(ImportTask.status.in_(_ACTIVE_TASK_STATUSES)) + ).scalars()) + return paths + + +def find_unimported( + session: Session, images_root: Path, *, now: float | None = None, +) -> tuple[list[Unimported], dict]: + """Walk the artists' folders and return (files to import, summary). + + The summary counts what was found and what was passed over, and lists the + artists with the most to import (bytes first) with a few sample paths each: + {"files", "bytes", "archives", "artists": [{slug, files, bytes, + samples}], "recent", "not_media", "no_artist"}""" + now = time.time() if now is None else now + referenced = _referenced_paths(session) + slugs = set(session.execute(select(Artist.slug)).scalars()) + found: list[Unimported] = [] + passed = {"recent": 0, "not_media": 0, "no_artist": 0} + root = Path(images_root) + for top in sorted(os.listdir(root)): + if top in _STORE_DIRS or top.startswith(("_", ".")): + continue + top_dir = root / top + if not top_dir.is_dir(): + continue + slug = slugify(top) + known = slug in slugs + for dirpath, dirnames, filenames in os.walk(top_dir): + dirnames[:] = sorted(d for d in dirnames if not d.startswith(".")) + for name in sorted(filenames): + path = Path(dirpath) / name + if ( + name.startswith(".") or path.suffix.lower() == ".json" + or is_temp_staging_file(path) + ): + continue + if str(path) in referenced: + continue + try: + st = path.lstat() + except OSError: + continue + if not stat.S_ISREG(st.st_mode): + continue + kind = "archive" if is_archive(path) else ("media" if is_supported(path) else None) + if kind is None: + passed["not_media"] += 1 + elif now - st.st_mtime < MIN_AGE_SECONDS: + passed["recent"] += 1 + elif not known: + passed["no_artist"] += 1 + else: + found.append(Unimported(path, slug, kind, st.st_size)) + + per_artist: dict[str, dict] = defaultdict(lambda: {"files": 0, "bytes": 0, "samples": []}) + for u in found: + a = per_artist[u.artist_slug] + a["files"] += 1 + a["bytes"] += u.size + if len(a["samples"]) < _SAMPLES_PER_ARTIST: + a["samples"].append(str(u.path.relative_to(root))) + artists = sorted( + ({"slug": s, **v} for s, v in per_artist.items()), key=lambda a: -a["bytes"], + )[:_ARTISTS_LISTED] + summary = { + "files": len(found), + "bytes": sum(u.size for u in found), + "archives": sum(1 for u in found if u.task_type == "archive"), + "artists": artists, + **passed, + } + return found, summary + + +def queue_unimported( + session: Session, images_root: Path, *, triggered_by: str = "manual", +) -> dict: + """Create a `library` import batch holding every file `find_unimported` + lists, and queue it. Returns the same summary plus `batch_id` (None when + there was nothing to import).""" + from ..tasks.import_file import enqueue_import # tasks import services + + found, summary = find_unimported(session, images_root) + summary["batch_id"] = None + if not found: + return summary + batch = ImportBatch( + triggered_by=triggered_by, source_path=str(images_root), + scan_mode="library", status="running", total_files=len(found), + ) + session.add(batch) + session.flush() + tasks = [ + ImportTask( + batch_id=batch.id, source_path=str(u.path), task_type=u.task_type, + status="queued", size_bytes=u.size, + ) + for u in found + ] + session.add_all(tasks) + session.commit() + for t in tasks: + enqueue_import(t.id, t.task_type) + summary["batch_id"] = batch.id + return summary + + +def folder_owner( + session: Session, path: Path, images_root: Path, +) -> tuple[Artist | None, Source | None]: + """The artist whose folder holds `path`, and that artist's source for the + platform folder below it when there is exactly one (a Discord artist has a + source per channel, so there the post's sidecar decides).""" + parts = Path(path).relative_to(images_root).parts + artist = session.execute( + select(Artist).where(Artist.slug == slugify(parts[0])) + ).scalar_one_or_none() + if artist is None or len(parts) < 3: + return artist, None + sources = session.execute( + select(Source).where(Source.artist_id == artist.id, Source.platform == parts[1]) + ).scalars().all() + return artist, (sources[0] if len(sources) == 1 else None) diff --git a/backend/app/tasks/admin.py b/backend/app/tasks/admin.py index 571da4a..64f9c6a 100644 --- a/backend/app/tasks/admin.py +++ b/backend/app/tasks/admin.py @@ -459,3 +459,29 @@ def reclaim_orphaned_attachments_task(self, dry_run: bool = True) -> dict: return cleanup_service.reclaim_orphaned_attachments( session, images_root=IMAGES_ROOT, dry_run=dry_run, ) + + +@celery.task( + name="backend.app.tasks.admin.import_unimported_files_task", + bind=True, + autoretry_for=(OperationalError, DBAPIError), + retry_backoff=15, retry_backoff_max=180, max_retries=1, + # One walk of the artists' folders: a stat per file, so minutes on a large + # library on network storage. The limits are the backstop for a wedged + # filesystem, not the expected exit. + soft_time_limit=1800, time_limit=2400, # 30 min / 40 min +) +def import_unimported_files_task(self, dry_run: bool = True) -> dict: + """Find files in the artists' folders that the library has no record of. + dry_run (the default) returns what was found, per artist, without changing + anything; apply queues them as a `library` import batch, which attaches each + file where it lies and removes the ones the library already holds or the + import filters reject.""" + from ..services import library_scan + + SessionLocal = _sync_session_factory() + with SessionLocal() as session: + if dry_run: + _found, summary = library_scan.find_unimported(session, IMAGES_ROOT) + return {**summary, "batch_id": None} + return library_scan.queue_unimported(session, IMAGES_ROOT) diff --git a/backend/app/tasks/import_file.py b/backend/app/tasks/import_file.py index fcdf78e..4a46a86 100644 --- a/backend/app/tasks/import_file.py +++ b/backend/app/tasks/import_file.py @@ -19,7 +19,8 @@ from sqlalchemy.exc import DBAPIError, OperationalError from ..celery_app import celery from ..models import ImportBatch, ImportSettings, ImportTask -from ..services.importer import Importer +from ..services.importer import Importer, ImportResult +from ..services.library_scan import folder_owner from ..services.thumbnailer import Thumbnailer from ._sync_engine import sync_session_factory as _sync_session_factory @@ -165,6 +166,27 @@ def enqueue_import(task_id: int, task_type: str) -> None: import_media_file.delay(task_id) +def _attach_where_it_lies(session, importer: Importer, path: Path) -> ImportResult: + """Import a file that already sits in its artist's folder (a `library` + batch, from services.library_scan). It is attached in place with that + artist and source. The post record its folder holds (`_post.json`) is filed + first, so the post gets its title and date. Afterwards the file is removed + if the library didn't keep it: a copy of something it already holds, or a + file the import filters reject. A file whose import crashed stays, with the + task marked failed: it may be the only copy of an archive the extractor + can't read yet.""" + if not path.is_file(): + return ImportResult(status="skipped", error="no longer on disk") + artist, source = folder_owner(session, path, IMAGES_ROOT) + record = path.parent / "_post.json" + if artist is not None and record.is_file(): + importer.upsert_post_record(record, artist=artist, source=source) + result = importer.attach_in_place(path, artist=artist, source=source) + if result.status != "failed": + importer.discard_after_attach(path, result) + return result + + def _do_import(session, task, import_task_id: int) -> dict: """Actual work, called from inside the resilience wrapper.""" settings = ImportSettings.load_sync(session) @@ -180,7 +202,10 @@ def _do_import(session, task, import_task_id: int) -> dict: deep=deep, ) - result = importer.import_one(Path(task.source_path)) + if batch is not None and batch.scan_mode == "library": + result = _attach_where_it_lies(session, importer, Path(task.source_path)) + else: + result = importer.import_one(Path(task.source_path)) if result.status in ("imported", "superseded"): task.status = "complete" diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 3ac6f78..767065a 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -189,6 +189,9 @@ TASK_STUCK_THRESHOLD_MINUTES: dict[str, int] = { # external-fetch entry above — without an override a healthy in-flight walk # is swept 'RecoverySweep' at the bare 5-min default. 30 = 25 + 5. "backend.app.tasks.admin.reclaim_orphaned_attachments_task": 30, + # Importing what's on disk stats every file in the artists' folders, and + # its hard limit is 40 min for the same wedged-filesystem case. 45 = 40 + 5. + "backend.app.tasks.admin.import_unimported_files_task": 45, # Head training and the manual head apply run to 65 min (3900s) — past the # ml queue's threshold (#4432). 70 = 65 + 5. "backend.app.tasks.ml.train_heads": 70, diff --git a/frontend/src/components/settings/LibraryImportCard.vue b/frontend/src/components/settings/LibraryImportCard.vue new file mode 100644 index 0000000..b5c036f --- /dev/null +++ b/frontend/src/components/settings/LibraryImportCard.vue @@ -0,0 +1,150 @@ + + + + + diff --git a/frontend/src/views/CleanupView.vue b/frontend/src/views/CleanupView.vue index 72a3c6f..7acbcd3 100644 --- a/frontend/src/views/CleanupView.vue +++ b/frontend/src/views/CleanupView.vue @@ -22,13 +22,15 @@

Duplicates & leftovers

Tidy post records, duplicates, locked-preview leftovers and attachments - that outlived what they belonged to. + that outlived what they belonged to, and import downloads that never + made it into the library.

+
@@ -63,6 +65,7 @@ import PostMaintenanceCard from '../components/settings/PostMaintenanceCard.vue' import VideoDedupCard from '../components/settings/VideoDedupCard.vue' import GatedPurgeCard from '../components/settings/GatedPurgeCard.vue' import AttachmentReclaimCard from '../components/settings/AttachmentReclaimCard.vue' +import LibraryImportCard from '../components/settings/LibraryImportCard.vue' import TagMaintenanceCard from '../components/settings/TagMaintenanceCard.vue' import DangerZoneCard from '../components/settings/DangerZoneCard.vue' diff --git a/tests/test_api_admin.py b/tests/test_api_admin.py index 0ff580f..e3bce33 100644 --- a/tests/test_api_admin.py +++ b/tests/test_api_admin.py @@ -708,3 +708,22 @@ async def test_trigger_reclaim_attachments_threads_apply(client, monkeypatch): ) assert resp.status_code == 202 assert calls[0][1] == {"dry_run": False} + + +@pytest.mark.asyncio +async def test_trigger_import_unimported_defaults_to_preview(client, monkeypatch): + """An empty body previews: the apply removes files the library rejects.""" + from backend.app.tasks import admin as admin_tasks + + calls = [] + monkeypatch.setattr( + admin_tasks.import_unimported_files_task, "delay", _fake_delay(calls) + ) + resp = await client.post("/api/admin/maintenance/import-unimported", json={}) + assert resp.status_code == 202 + assert calls[0][1] == {"dry_run": True} + + resp = await client.post( + "/api/admin/maintenance/import-unimported", json={"dry_run": False}, + ) + assert calls[1][1] == {"dry_run": False} diff --git a/tests/test_library_scan.py b/tests/test_library_scan.py new file mode 100644 index 0000000..c6355d6 --- /dev/null +++ b/tests/test_library_scan.py @@ -0,0 +1,123 @@ +"""Finding files in the artists' folders that the library has no record of, and +importing them where they lie (services.library_scan + the `library` batch).""" + +import io +import os +import time + +import pytest +from PIL import Image +from sqlalchemy import select + +import backend.app.tasks.import_file as import_file +from backend.app.models import Artist, ImageRecord, ImportSettings, ImportTask +from backend.app.services import library_scan +from backend.app.services.importer import Importer +from backend.app.services.thumbnailer import Thumbnailer + +pytestmark = pytest.mark.integration + + +def _jpeg(seed: int) -> bytes: + im = Image.new("RGB", (256, 256)) + px = im.load() + for y in range(256): + for x in range(256): + px[x, y] = ((x * seed) % 256, (y * (seed + 1)) % 256, (x ^ y) % 256) + buf = io.BytesIO() + im.save(buf, "JPEG") + return buf.getvalue() + + +def _write(path, data: bytes, *, age_days: float = 2): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(data) + t = time.time() - age_days * 86400 + os.utime(path, (t, t)) + return path + + +@pytest.fixture +def layout(db_sync, tmp_path): + db_sync.add(Artist(name="Libscan Maewix", slug="libscan-maewix")) + db_sync.commit() + post = tmp_path / "Libscan-Maewix" / "patreon" / "2026-05-01_123_Pack" + files = { + "missing": _write(post / "01_art.jpg", _jpeg(1)), + "owned": _write(post / "02_owned.jpg", _jpeg(2)), + "sidecar": _write(post / "01_art.json", b'{"category": "patreon", "id": "123"}'), + "staging": _write(post / "03_big.zip.part", b"partial"), + "recent": _write(post / "04_new.jpg", _jpeg(4), age_days=0), + "not_media": _write(post / "05_game.exe", b"MZ"), + "no_artist": _write(tmp_path / "libscan-nobody" / "x" / "a.jpg", _jpeg(5)), + "store": _write(tmp_path / "thumbs" / "abc" / "f.jpg", _jpeg(6)), + } + db_sync.add(ImageRecord( + path=str(files["owned"]), sha256="libscan-owned".ljust(64, "x"), + size_bytes=files["owned"].stat().st_size, mime="image/jpeg", + width=256, height=256, origin="downloaded", integrity_status="ok", + )) + db_sync.commit() + return tmp_path, files + + +def test_finds_only_unrecorded_media_in_an_artists_folder(db_sync, layout): + """A differently-cased folder still matches its artist. Recorded files, + sidecars, staging files, the library's own stores, folders that match no + artist, files changed in the last day and non-media are not listed.""" + root, files = layout + found, summary = library_scan.find_unimported(db_sync, root) + + assert [u.path for u in found] == [files["missing"]] + assert found[0].artist_slug == "libscan-maewix" + assert found[0].task_type == "media" + assert summary["files"] == 1 + assert summary["recent"] == 1 + assert summary["not_media"] == 1 + assert summary["no_artist"] == 1 + assert summary["artists"][0]["slug"] == "libscan-maewix" + assert summary["artists"][0]["samples"] == [ + "Libscan-Maewix/patreon/2026-05-01_123_Pack/01_art.jpg", + ] + + +def test_queue_makes_a_library_batch_of_what_find_lists(db_sync, layout, monkeypatch): + """The import queued is exactly the previewed list.""" + root, files = layout + queued = [] + monkeypatch.setattr(import_file, "enqueue_import", lambda tid, tt: queued.append((tid, tt))) + + summary = library_scan.queue_unimported(db_sync, root) + + tasks = db_sync.execute( + select(ImportTask).where(ImportTask.batch_id == summary["batch_id"]) + ).scalars().all() + assert [t.source_path for t in tasks] == [str(files["missing"])] + assert tasks[0].batch.scan_mode == "library" + assert queued == [(tasks[0].id, "media")] + + +def test_a_library_import_attaches_in_place_and_removes_a_copy(db_sync, layout, monkeypatch): + """The file stays where it is and becomes the record's file. A byte-identical + copy found later is removed from disk, since the library already holds it.""" + root, files = layout + monkeypatch.setattr(import_file, "IMAGES_ROOT", root) + importer = Importer( + session=db_sync, images_root=root, import_root=root, + thumbnailer=Thumbnailer(images_root=root), + settings=ImportSettings.load_sync(db_sync), + ) + + result = import_file._attach_where_it_lies(db_sync, importer, files["missing"]) + + assert result.status == "imported" + record = db_sync.get(ImageRecord, result.image_id) + assert record.path == str(files["missing"]) + assert files["missing"].exists() + assert db_sync.get(Artist, record.artist_id).slug == "libscan-maewix" + + copy = _write(files["missing"].parent / "09_copy.jpg", files["missing"].read_bytes()) + again = import_file._attach_where_it_lies(db_sync, importer, copy) + assert again.status == "skipped" + assert not copy.exists() + assert files["missing"].exists() diff --git a/tests/test_maintenance.py b/tests/test_maintenance.py index 8982c96..cae9705 100644 --- a/tests/test_maintenance.py +++ b/tests/test_maintenance.py @@ -889,3 +889,14 @@ def test_reclaim_attachments_stuck_threshold_exceeds_hard_time_limit(): "backend.app.tasks.admin.reclaim_orphaned_attachments_task" ] assert override >= hard_minutes + + +def test_import_unimported_stuck_threshold_exceeds_hard_time_limit(): + """The same invariant for the walk that imports files on disk.""" + from backend.app.tasks.admin import import_unimported_files_task + from backend.app.tasks.maintenance import TASK_STUCK_THRESHOLD_MINUTES + + override = TASK_STUCK_THRESHOLD_MINUTES[ + "backend.app.tasks.admin.import_unimported_files_task" + ] + assert override >= import_unimported_files_task.time_limit / 60