diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index 1d89c78..bf4237e 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -22,7 +22,13 @@ def make_celery() -> Celery: "fabledcurator", broker=cfg.celery_broker_url, backend=cfg.celery_result_backend, - include=["backend.app.tasks.smoke"], # FC-2/FC-3 extend this list + include=[ + "backend.app.tasks.smoke", + "backend.app.tasks.scan", + "backend.app.tasks.import_file", + "backend.app.tasks.thumbnail", + "backend.app.tasks.maintenance", + ], ) app.conf.update( task_default_queue="default", diff --git a/backend/app/tasks/import_file.py b/backend/app/tasks/import_file.py new file mode 100644 index 0000000..e9b5be2 --- /dev/null +++ b/backend/app/tasks/import_file.py @@ -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} diff --git a/backend/app/tasks/scan.py b/backend/app/tasks/scan.py new file mode 100644 index 0000000..92a1e00 --- /dev/null +++ b/backend/app/tasks/scan.py @@ -0,0 +1,78 @@ +"""scan_directory task: walks the configured import path, creates an +ImportBatch, enumerates files into ImportTasks, and enqueues +import_media_file for each. +""" + +from pathlib import Path + +from sqlalchemy import create_engine, select +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 is_supported + + +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) + + +@celery.task(name="backend.app.tasks.scan.scan_directory", bind=True) +def scan_directory(self, triggered_by: str = "manual") -> int: + """Walks the import root and creates ImportTasks. 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="quick", # FC-2a is quick only; FC-2d adds 'deep' + status="running", + ) + session.add(batch) + session.flush() + batch_id = batch.id + + # Walk and enumerate. + files_seen = 0 + for entry in import_root.rglob("*"): + if not entry.is_file(): + continue + if not is_supported(entry): + continue + try: + size = entry.stat().st_size + except OSError: + size = None + task = ImportTask( + batch_id=batch_id, + source_path=str(entry), + task_type="media", + status="pending", + size_bytes=size, + ) + session.add(task) + files_seen += 1 + + batch.total_files = files_seen + 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() + + return batch_id diff --git a/backend/app/tasks/thumbnail.py b/backend/app/tasks/thumbnail.py new file mode 100644 index 0000000..a5b9dcd --- /dev/null +++ b/backend/app/tasks/thumbnail.py @@ -0,0 +1,50 @@ +"""generate_thumbnail task: PIL/ffmpeg thumbnail generation on the thumbnail queue. + +Lives separately from import_file because thumbnails can be regenerated en +masse (FC-2c adds the 'regenerate all' admin action) and they're CPU-bound +so they deserve their own queue lane. +""" + +from pathlib import Path + +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from ..celery_app import celery +from ..config import get_config +from ..models import ImageRecord +from ..services.importer import is_video +from ..services.thumbnailer import Thumbnailer + + +IMAGES_ROOT = Path("/images") + + +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) + + +@celery.task(name="backend.app.tasks.thumbnail.generate_thumbnail", bind=True) +def generate_thumbnail(self, image_id: int) -> dict: + SessionLocal = _sync_session_factory() + with SessionLocal() as session: + record = session.get(ImageRecord, image_id) + if record is None: + return {"status": "missing", "image_id": image_id} + + thumbnailer = Thumbnailer(images_root=IMAGES_ROOT) + source = Path(record.path) + try: + if is_video(source): + result = thumbnailer.generate_video_thumbnail(source, record.sha256) + else: + result = thumbnailer.generate_image_thumbnail(source, record.sha256) + except Exception as exc: # pragma: no cover — thumbnail failure is non-fatal + return {"status": "failed", "image_id": image_id, "error": f"{type(exc).__name__}: {exc}"} + + record.thumbnail_path = str(result.path) + session.add(record) + session.commit() + return {"status": "ok", "image_id": image_id, "path": str(result.path)} diff --git a/tests/test_tasks_register.py b/tests/test_tasks_register.py new file mode 100644 index 0000000..5d4e941 --- /dev/null +++ b/tests/test_tasks_register.py @@ -0,0 +1,15 @@ +"""Confirms every FC-2a task module imports cleanly and registers with Celery.""" + +from backend.app.celery_app import celery + + +def test_scan_task_registered(): + assert "backend.app.tasks.scan.scan_directory" in celery.tasks + + +def test_import_media_file_registered(): + assert "backend.app.tasks.import_file.import_media_file" in celery.tasks + + +def test_generate_thumbnail_registered(): + assert "backend.app.tasks.thumbnail.generate_thumbnail" in celery.tasks