FabledCurator/backend/app/tasks/scan.py

"""scan tasks:

* ``scan_directory`` — walks the configured import path, creates an
  ImportBatch, enumerates files into ImportTasks, and enqueues
  import_media_file for each.
* ``tick_due_sources`` (FC-3d) — periodic Beat-fired scan that fires
  ``download_source.delay()`` for sources due for a check.
"""

import asyncio
from datetime import UTC, datetime
from pathlib import Path

from sqlalchemy import select

from ..celery_app import celery
from ..models import DownloadEvent, ImportBatch, ImportSettings, ImportTask
from ..services.archive_extractor import is_archive
from ..services.scheduler_service import record_tick, select_due_sources
from ._async_session import async_session_factory
from ._sync_engine import sync_session_factory as _sync_session_factory


def _iter_import_files(import_root: Path):
    """Every regular file except sidecar .json, dotfiles, and .partial
    temp files. The Importer dispatches by kind (media / archive /
    other) — scan no longer filters to media (FC-2d-iii)."""
    for entry in import_root.rglob("*"):
        if not entry.is_file():
            continue
        if entry.name.startswith("."):
            continue
        if entry.suffix.lower() in (".json", ".partial"):
            continue
        yield entry


@celery.task(
    name="backend.app.tasks.scan.scan_directory",
    bind=True,
    # Audit 2026-06-02 — large libraries make the scan legitimately long.
    # Hard cap at 70 min so the corresponding QUEUE_STUCK_THRESHOLD_MINUTES
    # ("scan") of 75 min always wins; soft limit gives the task a clean
    # exit window before SIGKILL.
    soft_time_limit=3600, time_limit=4200,
)
def scan_directory(self, triggered_by: str = "manual",
                   mode: str = "quick") -> int:
    """Walks the import root and creates ImportTasks. `mode` is 'quick'
    or 'deep'; deep also enqueues a library-wide phash backfill and (via
    Importer.deep) re-derives on already-imported files. Returns the
    batch id."""
    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
        settings = ImportSettings.load_sync(session)
        import_root = Path(settings.import_scan_path)

        batch = ImportBatch(
            triggered_by=triggered_by,
            source_path=str(import_root),
            scan_mode=mode,
            status="running",
        )
        session.add(batch)
        session.flush()
        batch_id = batch.id

        # Skip-set behavior splits by mode (operator-flagged 2026-05-25):
        #
        # quick: any non-failed prior ImportTask (active OR finished) is
        #   skipped — quick scan only does new-file enqueue, so re-touching
        #   already-imported files is wasted work.
        #
        # deep: ONLY currently-in-flight tasks (pending/queued/processing)
        #   are skipped. Completed and skipped tasks ARE re-queued because
        #   deep scan exists precisely to re-touch already-imported files
        #   (refresh sidecar metadata, fill NULL phash, fill NULL artist
        #   via Importer._deep_rederive). Matches IR's deep-scan behavior.
        active_statuses = ["pending", "queued", "processing"]
        if mode == "deep":
            skip_statuses = active_statuses
        else:
            skip_statuses = active_statuses + ["complete", "skipped"]
        non_failed_existing = set(session.execute(
            select(ImportTask.source_path).where(
                ImportTask.status.in_(skip_statuses),
            )
        ).scalars().all())

        # Walk and enumerate.
        files_seen = 0
        files_skipped_existing = 0
        for entry in _iter_import_files(import_root):
            entry_str = str(entry)
            if entry_str in non_failed_existing:
                files_skipped_existing += 1
                continue
            try:
                size = entry.stat().st_size
            except OSError:
                size = None
            task = ImportTask(
                batch_id=batch_id,
                source_path=entry_str,
                # Archives route to import_archive_file (larger time
                # budget) — they run the per-member pipeline inline.
                task_type="archive" if is_archive(entry) else "media",
                status="pending",
                size_bytes=size,
            )
            session.add(task)
            files_seen += 1

        batch.total_files = files_seen
        # If the walk enqueued nothing (every file was already on a
        # non-failed ImportTask from a prior scan), there's no
        # import_media_file message that would ever flip this batch to
        # 'complete' — finalize it now so the active-batch query in
        # /api/system/stats doesn't get stuck reporting all-zero
        # counters from an empty scan.
        if files_seen == 0:
            batch.status = "complete"
            batch.finished_at = datetime.now(UTC)
        session.commit()

        # Now enqueue each pending task on the right Celery task
        # (media vs archive) via the shared router.
        from .import_file import enqueue_import

        for task in session.execute(
            select(ImportTask).where(ImportTask.batch_id == batch_id)
        ).scalars():
            task.status = "queued"
            session.add(task)
            enqueue_import(task.id, task.task_type)
        session.commit()

        if mode == "deep":
            from .maintenance import backfill_phash

            backfill_phash.delay()

    return batch_id


# --- FC-3d: periodic source-check tick ------------------------------------


async def _tick_due_sources_async() -> dict:
    factory, engine = async_session_factory()
    try:
        async with factory() as session:
            # Prove the scheduler is alive even on empty ticks (UI reads this).
            await record_tick(session)
            due = await select_due_sources(session)
            if not due:
                return {
                    "due": 0, "queued": 0, "skipped_in_flight": 0,
                    "tick_at": datetime.now(UTC).isoformat(),
                }
            due_ids = [s.id for s in due]
            in_flight_rows = (await session.execute(
                select(DownloadEvent.source_id).where(
                    DownloadEvent.source_id.in_(due_ids),
                    DownloadEvent.status.in_(["pending", "running"]),
                )
            )).scalars().all()
            in_flight = set(in_flight_rows)

            # Local import: keep top of module light, avoid circular import
            # at celery task discovery time.
            from .download import download_source

            queued = 0
            for s in due:
                if s.id in in_flight:
                    continue
                session.add(DownloadEvent(source_id=s.id, status="pending"))
                await session.commit()
                download_source.delay(s.id)
                queued += 1
            return {
                "due": len(due),
                "queued": queued,
                "skipped_in_flight": len(in_flight),
                "tick_at": datetime.now(UTC).isoformat(),
            }
    finally:
        await engine.dispose()


@celery.task(name="backend.app.tasks.scan.tick_due_sources", bind=True)
def tick_due_sources(self) -> dict:
    """FC-3d: scan for sources due for a periodic check and fire
    ``download_source.delay()`` for each."""
    return asyncio.run(_tick_due_sources_async())