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
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."""
    if result.status in ("imported", "superseded"):
        return ("complete", True)
    if result.status == "attached":
        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


@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:
    """Returns a dict so the eager-mode tests can assert without DB.

    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 this
        process so the task can mark its row failed before being killed.
      - time_limit (360s) is the hard cap; SIGKILL if the soft signal
        was swallowed.
    """
    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 (>300s)")
            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 within 5 min.
            raise
        except Exception as exc:  # noqa: BLE001 — pipeline crash, mark + re-raise
            _mark_failed(session, task, f"{type(exc).__name__}: {exc}")
            raise


def _do_import(session, task, import_task_id: int) -> dict:
    """Actual work, called from inside the resilience wrapper."""
    settings = session.execute(
        select(ImportSettings).where(ImportSettings.id == 1)
    ).scalar_one()
    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,
    )

    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 == "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 tag_and_embed
        from .thumbnail import generate_thumbnail

        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)
            tag_and_embed.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}