"""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 sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from ..celery_app import celery
from ..config import get_config
from ..models import DownloadEvent, ImportBatch, ImportSettings, ImportTask
from ..services.scheduler_service import select_due_sources
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)
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 = session.execute(
            select(ImportSettings).where(ImportSettings.id == 1)
        ).scalar_one()
        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: any source_path that already has a non-failed ImportTask
        # row. Re-running scan_directory must not re-enqueue files the
        # importer has already handled (or is currently handling); doing so
        # creates duplicate work and inflates the queue. Failed prior tasks
        # are eligible for retry.
        non_failed_existing = set(session.execute(
            select(ImportTask.source_path).where(
                ImportTask.status.in_(
                    ["pending", "queued", "processing", "complete", "skipped"]
                ),
            )
        ).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,
                task_type="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 import_media_file for each pending task.
        for task in session.execute(
            select(ImportTask).where(ImportTask.batch_id == batch_id)
        ).scalars():
            task.status = "queued"
            session.add(task)
            from .import_file import import_media_file

            import_media_file.delay(task.id)
        session.commit()

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

            backfill_phash.delay()

    return batch_id


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


def _async_session_factory():
    cfg = get_config()
    engine = create_async_engine(cfg.database_url, future=True, pool_pre_ping=True)
    return async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False), engine


async def _tick_due_sources_async() -> dict:
    factory, engine = _async_session_factory()
    try:
        async with factory() as 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())