FabledCurator/backend/app/tasks/import_file.py

"""import_media_file task: imports one file via the Importer service and
updates the ImportTask state machine + ImportBatch counters atomically.

Resilience contract (2026-05-24, operator-mandated): once a row has been
flipped to 'processing' inside this task, EVERY exit path MUST flip it
to a terminal state (complete / skipped / failed) or rely on Celery's
autoretry to attempt the work again. No exit path is allowed to leave
the row stuck in 'processing'. The `recover_interrupted_tasks`
maintenance sweep is the safety net (5 min threshold) for the case
where even the failure-marking commit can't be written.
"""

from datetime import UTC, datetime
from pathlib import Path

from celery.exceptions import SoftTimeLimitExceeded
from sqlalchemy import select, update
from sqlalchemy.exc import DBAPIError, OperationalError

from ..celery_app import celery
from ..models import ImportBatch, ImportSettings, ImportTask
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

IMAGES_ROOT = Path("/images")


def _map_result_to_status(result):
    """(ImportTask.status, should_requeue_ml_and_thumb) for an ImportResult.
    'superseded' = the kept row's file/ML changed → complete + re-derive.
    'attached' = a non-art file preserved → complete, no ML/thumb.
    'refreshed' = deep scan refreshed sidecar/phash on an existing row →
    complete, no ML/thumb re-derive (file/pixels unchanged)."""
    if result.status in ("imported", "superseded"):
        return ("complete", True)
    if result.status in ("attached", "refreshed"):
        return ("complete", False)
    if result.status == "skipped":
        return ("skipped", False)
    return ("failed", False)


def _mark_failed(session, task, error_msg: str) -> None:
    """Best-effort flip of a 'processing' row to 'failed' + batch counter
    increment. Wrapped in its own try because if the DB is what just
    broke, this commit will also fail — that's why the maintenance sweep
    exists as a backstop."""
    try:
        task.status = "failed"
        task.error = error_msg
        task.finished_at = datetime.now(UTC)
        session.add(task)
        session.execute(
            update(ImportBatch)
            .where(ImportBatch.id == task.batch_id)
            .values(failed=ImportBatch.failed + 1)
        )
        session.commit()
    except Exception:  # noqa: BLE001 — best-effort, sweep catches the rest
        try:
            session.rollback()
        except Exception:  # noqa: BLE001
            pass


def _run_import_task(import_task_id: int) -> dict:
    """Shared body for import_media_file + import_archive_file. The two
    tasks differ ONLY in their Celery time limits (a single media file
    is sub-second; an archive runs the full per-member pipeline inline
    for every member and can take many minutes). Both flip the row to
    'processing', dispatch to `_do_import`, and honor the
    flip-to-terminal resilience contract.
    """
    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
        task = session.get(ImportTask, import_task_id)
        if task is None:
            return {"status": "missing", "task_id": import_task_id}

        task.status = "processing"
        task.started_at = datetime.now(UTC)
        session.add(task)
        session.commit()

        try:
            return _do_import(session, task, import_task_id)
        except SoftTimeLimitExceeded:
            _mark_failed(session, task, "soft_time_limit exceeded")
            raise
        except (OperationalError, DBAPIError, OSError):
            # Retryable per the decorator; do NOT mark failed (let
            # autoretry have a clean go at it). If autoretry exhausts,
            # the row stays 'processing' and the maintenance sweep
            # flips it.
            raise
        except Exception as exc:  # noqa: BLE001 — pipeline crash, mark + re-raise
            _mark_failed(session, task, f"{type(exc).__name__}: {exc}")
            raise


@celery.task(
    name="backend.app.tasks.import_file.import_media_file",
    bind=True,
    autoretry_for=(OperationalError, DBAPIError, OSError),
    retry_backoff=5,
    retry_backoff_max=60,
    retry_jitter=True,
    max_retries=3,
    soft_time_limit=300,
    time_limit=360,
)
def import_media_file(self, import_task_id: int) -> dict:
    """Import ONE media file (or non-media → PostAttachment). Sub-second
    for the common case; the tight 5-min soft limit keeps a genuinely
    stuck single-file import detectable fast.

    Decorator notes:
      - autoretry_for: transient DB / filesystem errors retry with
        exponential backoff (5s base, jitter, max 3 attempts). On final
        give-up the task raises and acks_late=True (set globally on the
        Celery app) does NOT redeliver — the recovery sweep catches the
        row instead.
      - soft_time_limit (300s) raises SoftTimeLimitExceeded in-process
        so the task can mark its row failed before being killed.
      - time_limit (360s) is the hard SIGKILL cap.
    """
    return _run_import_task(import_task_id)


@celery.task(
    name="backend.app.tasks.import_file.import_archive_file",
    bind=True,
    autoretry_for=(OperationalError, DBAPIError, OSError),
    retry_backoff=5,
    retry_backoff_max=60,
    retry_jitter=True,
    max_retries=3,
    # Archives run the full per-member pipeline (sha256 + pHash + dedup
    # query + copy + provenance) for EVERY media member inline, under a
    # single task budget. A multi-hundred-member archive blows the
    # 5-min media limit. soft=30min / hard=35min sizes for a large
    # archive. Operator-flagged 2026-05-28 (target 1645019 hit the old
    # shared 300s soft limit). The recovery sweep gives this task its
    # own 40-min threshold via maintenance.TASK_STUCK_THRESHOLD_MINUTES
    # so it isn't preempted while legitimately grinding through members.
    soft_time_limit=1800,
    time_limit=2100,
)
def import_archive_file(self, import_task_id: int) -> dict:
    """Import an archive: extract + run the per-member media pipeline for
    every member inline, then preserve the archive as a PostAttachment.
    Same body as import_media_file (dispatch is by file kind inside
    Importer.import_one); split out purely for the larger time budget."""
    return _run_import_task(import_task_id)


def enqueue_import(task_id: int, task_type: str) -> None:
    """Route an ImportTask to the right Celery task by its task_type.
    Single source of truth for the media-vs-archive dispatch so the
    scan, retry, and recovery-requeue paths stay in sync."""
    if task_type == "archive":
        import_archive_file.delay(task_id)
    else:
        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)
    import_root = Path(settings.import_scan_path)
    batch = session.get(ImportBatch, task.batch_id)
    deep = bool(batch and batch.scan_mode == "deep")
    importer = Importer(
        session=session,
        images_root=IMAGES_ROOT,
        import_root=import_root,
        thumbnailer=Thumbnailer(images_root=IMAGES_ROOT),
        settings=settings,
        deep=deep,
    )

    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"
        task.result_image_id = result.image_id
        counter_col_name = "imported"
        counter_col = ImportBatch.imported
    elif result.status == "refreshed":
        # Deep-scan rederive: existing row got phash/artist/sidecar
        # refreshed. Task is complete (no further work), but counted in
        # `refreshed` not `imported` so the UI can surface the actual
        # work done. operator-flagged 2026-05-25.
        task.status = "complete"
        task.result_image_id = result.image_id
        counter_col_name = "refreshed"
        counter_col = ImportBatch.refreshed
    elif result.status == "attached":
        task.status = "complete"
        counter_col_name = "attachments"
        counter_col = ImportBatch.attachments
    elif result.status == "skipped":
        task.status = "skipped"
        task.error = (
            f"{result.skip_reason.value}: {result.error}"
            if result.skip_reason
            else result.error
        )
        task.result_image_id = result.image_id
        counter_col_name = "skipped"
        counter_col = ImportBatch.skipped
    else:
        task.status = "failed"
        task.error = result.error
        counter_col_name = "failed"
        counter_col = ImportBatch.failed

    task.finished_at = datetime.now(UTC)
    session.execute(
        update(ImportBatch)
        .where(ImportBatch.id == task.batch_id)
        .values({counter_col_name: counter_col + 1})
    )
    session.add(task)
    session.commit()

    # Enqueue thumbnail + ML for newly imported AND superseded images
    # (a superseded row has cleared ML + no thumbnail).
    if result.status in ("imported", "superseded"):
        from .ml import cpu_embed_enabled, embed_image
        from .thumbnail import generate_thumbnail

        do_embed = cpu_embed_enabled()
        ids = list(result.member_image_ids)
        if result.image_id is not None and result.image_id not in ids:
            ids.append(result.image_id)
        for img_id in ids:
            generate_thumbnail.delay(img_id)
            if do_embed:
                embed_image.delay(img_id)

    # If this was the last task in the batch, mark the batch complete.
    remaining = session.execute(
        select(ImportTask.id)
        .where(ImportTask.batch_id == task.batch_id)
        .where(ImportTask.status.in_(["pending", "queued", "processing"]))
        .limit(1)
    ).scalar_one_or_none()
    if remaining is None:
        session.execute(
            update(ImportBatch)
            .where(ImportBatch.id == task.batch_id)
            .values(
                status="complete",
                finished_at=datetime.now(UTC),
            )
        )
        session.commit()

    return {"task_id": import_task_id, "status": task.status}