feat(maintenance): add library-validation sweep task and API

Walks the downloads tree and reports files that fail magic-byte
validation — the "how many pre-existing files in my library are
silently truncated?" question, answered. Skips the _quarantine
subtree (already known-bad) and caps suspect_paths at 500 with a
truncated flag so the persisted report can't grow unbounded.

On-demand only: a full filesystem walk on a NAS-mounted library is
expensive and the user should choose when to pay for it. Pre-existing
files are reported, not quarantined — they may be the only copy and
the user decides what to do.

Adds GET /api/settings/library-validation (most recent report) and
POST /api/settings/library-validation/run (queue a fresh sweep).
Cached under cache.library_validation_report.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

backend/app/api/settings.py

@@ -6,7 +6,14 @@ from quart import Blueprint, request, jsonify, current_app
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.models.setting import Setting, DEFAULT_SETTINGS, EXTENSION_API_KEY_SETTING, STORAGE_STATS_SETTING, generate_api_key
+from app.models.setting import (
+    Setting,
+    DEFAULT_SETTINGS,
+    EXTENSION_API_KEY_SETTING,
+    STORAGE_STATS_SETTING,
+    LIBRARY_VALIDATION_REPORT_SETTING,
+    generate_api_key,
+)
 from app.config import get_settings
 from app.services.gallery_dl import GalleryDLService
 
@@ -270,6 +277,32 @@ async def refresh_storage_stats():
         return jsonify({"error": str(e)}), 500
 
 
+@bp.route("/library-validation", methods=["GET"])
+async def get_library_validation_report():
+    """Return the most recent library-validation sweep report (or null)."""
+    async with current_app.db_engine.connect() as conn:
+        async with AsyncSession(bind=conn) as session:
+            result = await session.execute(
+                select(Setting).where(Setting.key == LIBRARY_VALIDATION_REPORT_SETTING)
+            )
+            setting = result.scalar_one_or_none()
+            return jsonify(setting.value if setting and setting.value else None)
+
+
+@bp.route("/library-validation/run", methods=["POST"])
+async def run_library_validation():
+    """Queue a library-validation sweep (walks the library, reports truncated files)."""
+    from app.tasks.maintenance import validate_library
+
+    try:
+        validate_library.delay()
+        current_app.logger.info("Queued library validation sweep")
+        return jsonify({"message": "Library validation queued"})
+    except Exception as e:
+        current_app.logger.error(f"Failed to queue library validation: {e}")
+        return jsonify({"error": str(e)}), 500
+
+
 @bp.route("/logs", methods=["GET"])
 async def get_logs():
     """Get recent download logs from download metadata.

backend/app/tasks/maintenance.py

@@ -11,7 +11,11 @@ from sqlalchemy import select, update
 
 from app.tasks.db import get_async_session, cleanup_engine as _cleanup_engine
 from app.tasks.celery_app import celery_app
-from app.models.setting import Setting, STORAGE_STATS_SETTING
+from app.models.setting import (
+    Setting,
+    STORAGE_STATS_SETTING,
+    LIBRARY_VALIDATION_REPORT_SETTING,
+)
 from app.models.download import Download, DownloadStatus
 from app.models.base import utcnow
 
@@ -151,6 +155,126 @@ def update_storage_stats() -> dict:
     return result
 
 
+def scan_library_for_corruption(
+    root: Path,
+    max_suspect_paths: int = 500,
+) -> dict:
+    """Walk the library and report files that fail magic-byte validation.
+
+    The user-facing case for this is "I just deployed the validator. How many
+    pre-existing files in my library are silently truncated?" — running this
+    once gives that number plus a list of paths to act on. We report only;
+    the live download path quarantines, but pre-existing files are left in
+    place because they may be the only copy and the user should decide.
+
+    Skips the `_quarantine` subtree (those files are already known-bad).
+    `suspect_paths` is capped at `max_suspect_paths` so the persisted report
+    can't grow unbounded; `suspect_count` always reflects the true total.
+    """
+    # Local import keeps this module's import cost low (filesystem walk
+    # tasks shouldn't drag the validator into every Celery worker process
+    # unconditionally).
+    from app.services.file_validator import is_validatable, validate_file
+
+    started_at = utcnow().isoformat()
+    suspect_paths: list[dict] = []
+    suspect_count = 0
+    scanned = 0
+    by_format: dict[str, int] = {}
+
+    if not root.exists():
+        return {
+            "started_at": started_at,
+            "completed_at": utcnow().isoformat(),
+            "scanned": 0,
+            "suspect_count": 0,
+            "by_format": {},
+            "suspect_paths": [],
+            "root": str(root),
+            "truncated": False,
+        }
+
+    quarantine_root = root / "_quarantine"
+
+    for dirpath, dirnames, filenames in os.walk(root):
+        # Don't re-scan the quarantine directory.
+        dirnames[:] = [d for d in dirnames if Path(dirpath, d) != quarantine_root]
+
+        for name in filenames:
+            p = Path(dirpath, name)
+            if not is_validatable(p):
+                continue
+            scanned += 1
+            try:
+                result = validate_file(p)
+            except Exception as e:
+                logger.warning(f"Validator raised on {p}: {e}")
+                continue
+            if result.ok:
+                continue
+            suspect_count += 1
+            fmt = result.format or "unknown"
+            by_format[fmt] = by_format.get(fmt, 0) + 1
+            if len(suspect_paths) < max_suspect_paths:
+                suspect_paths.append(
+                    {
+                        "path": str(p),
+                        "format": result.format,
+                        "reason": result.reason,
+                        "size": result.size,
+                    }
+                )
+
+    return {
+        "started_at": started_at,
+        "completed_at": utcnow().isoformat(),
+        "scanned": scanned,
+        "suspect_count": suspect_count,
+        "by_format": by_format,
+        "suspect_paths": suspect_paths,
+        "root": str(root),
+        "truncated": suspect_count > len(suspect_paths),
+    }
+
+
+async def _validate_library_async() -> dict:
+    """Async wrapper that runs the sweep and persists the report."""
+    session_factory, engine = get_async_session()
+    try:
+        report = scan_library_for_corruption(Path(settings.download_path))
+        async with session_factory() as session:
+            result = await session.execute(
+                select(Setting).where(Setting.key == LIBRARY_VALIDATION_REPORT_SETTING)
+            )
+            setting = result.scalar_one_or_none()
+            if setting:
+                setting.value = report
+            else:
+                setting = Setting(key=LIBRARY_VALIDATION_REPORT_SETTING, value=report)
+                session.add(setting)
+            await session.commit()
+        return report
+    finally:
+        await _cleanup_engine(engine)
+
+
+@celery_app.task(name="tasks.validate_library")
+def validate_library() -> dict:
+    """Celery task: walk the library and report truncated files.
+
+    On-demand (UI button or manual trigger). Not in the periodic schedule —
+    a full filesystem walk on a large NAS-mounted library is expensive and
+    the user should choose when to pay for it.
+    """
+    logger.info("Starting library validation sweep...")
+    report = asyncio.run(_validate_library_async())
+    logger.info(
+        f"Library validation: scanned {report['scanned']}, "
+        f"suspect={report['suspect_count']} ({report['by_format']})"
+    )
+    return report
+
+
 async def _reset_orphaned_running_jobs_async(threshold_minutes: int = None) -> dict:
     """Reset jobs stuck in 'running' status back to 'queued'.
 

backend/tests/tasks/test_validate_library.py

@@ -0,0 +1,81 @@
+"""Tests for the library-validation sweep.
+
+Covers: scanning a fake library, ignoring the _quarantine subtree,
+counting suspect files by format, and capping suspect_paths.
+"""
+from pathlib import Path
+
+import pytest
+
+from app.services.file_validator import (
+    GIF_HEAD_89A,
+    JPEG_HEAD,
+    JPEG_TAIL,
+    PNG_HEAD,
+    PNG_TAIL,
+)
+from app.tasks.maintenance import scan_library_for_corruption
+
+
+def _good_jpeg(p: Path) -> None:
+    p.parent.mkdir(parents=True, exist_ok=True)
+    p.write_bytes(JPEG_HEAD + b"\x00" * 32 + JPEG_TAIL)
+
+
+def _bad_jpeg(p: Path) -> None:
+    p.parent.mkdir(parents=True, exist_ok=True)
+    p.write_bytes(JPEG_HEAD + b"\x00" * 32)  # no EOI
+
+
+def _bad_png(p: Path) -> None:
+    p.parent.mkdir(parents=True, exist_ok=True)
+    p.write_bytes(PNG_HEAD + b"\x00" * 32)  # no IEND
+
+
+def test_scan_finds_truncated_files(tmp_path):
+    _good_jpeg(tmp_path / "Artist" / "patreon" / "01.jpg")
+    _bad_jpeg(tmp_path / "Artist" / "patreon" / "02.jpg")
+    _bad_png(tmp_path / "Artist" / "patreon" / "03.png")
+    # JSON sidecars must not be scanned
+    (tmp_path / "Artist" / "patreon" / "01.json").write_bytes(b"{}")
+
+    report = scan_library_for_corruption(tmp_path)
+
+    assert report["scanned"] == 3
+    assert report["suspect_count"] == 2
+    assert report["by_format"] == {"jpeg": 1, "png": 1}
+    paths = {entry["path"] for entry in report["suspect_paths"]}
+    assert any("02.jpg" in p for p in paths)
+    assert any("03.png" in p for p in paths)
+    assert not report["truncated"]
+
+
+def test_scan_skips_quarantine_subtree(tmp_path):
+    """Files already moved to _quarantine are known-bad; don't re-scan them."""
+    _bad_jpeg(tmp_path / "_quarantine" / "Artist" / "patreon" / "01.jpg")
+    _good_jpeg(tmp_path / "Artist" / "patreon" / "02.jpg")
+
+    report = scan_library_for_corruption(tmp_path)
+
+    assert report["scanned"] == 1
+    assert report["suspect_count"] == 0
+
+
+def test_scan_caps_suspect_paths(tmp_path):
+    """suspect_count is the true total; suspect_paths is capped + truncated flag set."""
+    for i in range(7):
+        _bad_jpeg(tmp_path / "Artist" / "patreon" / f"{i:02d}.jpg")
+
+    report = scan_library_for_corruption(tmp_path, max_suspect_paths=3)
+
+    assert report["scanned"] == 7
+    assert report["suspect_count"] == 7
+    assert len(report["suspect_paths"]) == 3
+    assert report["truncated"] is True
+
+
+def test_scan_missing_root_returns_empty_report(tmp_path):
+    report = scan_library_for_corruption(tmp_path / "does_not_exist")
+    assert report["scanned"] == 0
+    assert report["suspect_count"] == 0
+    assert report["suspect_paths"] == []