feat(fc2a): add Celery tasks — scan_directory, import_media_file, generate_thumbnail

scan_directory walks ImportSettings.import_scan_path, creates an ImportBatch, enumerates supported files into ImportTasks, and enqueues import_media_file per task. import_media_file moves the task through its state machine (pending → queued → processing → complete/skipped/failed), updates ImportBatch counters atomically (UPDATE ... SET col = col + 1), enqueues a thumbnail task on success, and marks the batch complete when the last task drains. generate_thumbnail runs on its own queue (thumbnail) so big imports don't starve thumbnail throughput; failure here is logged and does not fail the import. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-14 12:08:37 -04:00
parent f38a1d48c5
commit d3bb8c509a
5 changed files with 272 additions and 1 deletions
@@ -0,0 +1,122 @@
+"""import_media_file task: imports one file via the Importer service and
+updates the ImportTask state machine + ImportBatch counters atomically.
+"""
+
+from datetime import datetime, timezone
+from pathlib import Path
+
+from sqlalchemy import create_engine, select, update
+from sqlalchemy.orm import sessionmaker
+
+from ..celery_app import celery
+from ..config import get_config
+from ..models import ImportBatch, ImportSettings, ImportTask
+from ..services.importer import Importer
+from ..services.thumbnailer import Thumbnailer
+
+
+def _sync_session_factory():
+    cfg = get_config()
+    engine = create_engine(cfg.database_url_sync, future=True, pool_pre_ping=True)
+    return sessionmaker(engine, expire_on_commit=False)
+
+
+IMAGES_ROOT = Path("/images")
+
+
+@celery.task(name="backend.app.tasks.import_file.import_media_file", bind=True)
+def import_media_file(self, import_task_id: int) -> dict:
+    """Returns a dict so the eager-mode tests can assert without DB."""
+    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(timezone.utc)
+        session.add(task)
+        session.commit()
+
+        settings = session.execute(
+            select(ImportSettings).where(ImportSettings.id == 1)
+        ).scalar_one()
+        import_root = Path(settings.import_scan_path)
+        importer = Importer(
+            session=session,
+            images_root=IMAGES_ROOT,
+            import_root=import_root,
+            thumbnailer=Thumbnailer(images_root=IMAGES_ROOT),
+            settings=settings,
+        )
+
+        try:
+            result = importer.import_one(Path(task.source_path))
+        except Exception as exc:  # pragma: no cover — pipeline crash
+            task.status = "failed"
+            task.error = f"{type(exc).__name__}: {exc}"
+            task.finished_at = datetime.now(timezone.utc)
+            session.execute(
+                update(ImportBatch)
+                .where(ImportBatch.id == task.batch_id)
+                .values(failed=ImportBatch.failed + 1)
+            )
+            session.add(task)
+            session.commit()
+            raise
+
+        if result.status == "imported":
+            task.status = "complete"
+            task.result_image_id = result.image_id
+            counter_col_name = "imported"
+            counter_col = ImportBatch.imported
+        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(timezone.utc)
+        session.execute(
+            update(ImportBatch)
+            .where(ImportBatch.id == task.batch_id)
+            .values({counter_col_name: counter_col + 1})
+        )
+        session.add(task)
+        session.commit()
+
+        # Enqueue the thumbnail task for newly imported images.
+        if result.status == "imported" and result.image_id is not None:
+            from .thumbnail import generate_thumbnail
+
+            generate_thumbnail.delay(result.image_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(timezone.utc),
+                )
+            )
+            session.commit()
+
+        return {"task_id": import_task_id, "status": task.status}