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