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 <noreply@anthropic.com>

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/<id> 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

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)

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)

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"

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,