imagerepo/app/celery_app.py

# app/celery_app.py
"""
Celery application factory and configuration.
Provides task queue functionality for the ImageRepo import pipeline.
"""
import os
from celery import Celery

# Default broker/backend URLs
DEFAULT_BROKER = 'redis://redis:6379/0'
DEFAULT_BACKEND = 'redis://redis:6379/0'


def make_celery(app=None):
    """
    Create Celery instance with Flask app context support.

    Args:
        app: Flask application instance (optional)

    Returns:
        Configured Celery instance
    """
    # Get broker/backend from app config or environment
    if app:
        broker = app.config.get('CELERY_BROKER_URL', DEFAULT_BROKER)
        backend = app.config.get('CELERY_RESULT_BACKEND', DEFAULT_BACKEND)
    else:
        broker = os.environ.get('CELERY_BROKER_URL', DEFAULT_BROKER)
        backend = os.environ.get('CELERY_RESULT_BACKEND', DEFAULT_BACKEND)

    celery = Celery(
        'imagerepo',
        broker=broker,
        backend=backend,
        include=[
            'app.tasks.scan',
            'app.tasks.import_file',
            'app.tasks.thumbnail',
            'app.tasks.sidecar',
            'app.tasks.ml',
            'app.tasks.maintenance',
        ]
    )

    # Worker concurrency from environment
    concurrency = int(os.environ.get('CELERY_WORKER_CONCURRENCY', '2'))

    celery.conf.update(
        # Serialization
        task_serializer='json',
        accept_content=['json'],
        result_serializer='json',
        timezone='UTC',
        enable_utc=True,

        # Concurrency - intentionally low for steady background processing
        worker_concurrency=concurrency,
        worker_prefetch_multiplier=1,  # One task at a time per worker process

        # Task routing - separate queues for different task types
        # Scheduler handles: maintenance (periodic tasks) + scan (directory scans)
        # Worker handles: import, thumbnail, sidecar, default
        task_routes={
            # Scan tasks - handled by scheduler
            'app.tasks.scan.scan_directory': {'queue': 'scan'},
            'app.tasks.scan.deep_scan_directory': {'queue': 'scan'},

            # Maintenance tasks - handled by scheduler (periodic/lightweight)
            'app.tasks.scan.recover_interrupted_tasks': {'queue': 'maintenance'},
            'app.tasks.scan.cleanup_old_tasks': {'queue': 'maintenance'},
            'app.tasks.scan.update_system_stats': {'queue': 'maintenance'},
            'app.tasks.scan.update_batch_stats': {'queue': 'maintenance'},
            'app.tasks.maintenance.sweep_blocklisted_tag_from_images': {'queue': 'maintenance'},
            'app.tasks.maintenance.sync_character_fandoms_to_images': {'queue': 'maintenance'},
            'app.tasks.maintenance.verify_media_integrity': {'queue': 'maintenance'},
            'app.tasks.maintenance.verify_unverified_images': {'queue': 'maintenance'},
            'app.tasks.maintenance.apply_auto_accept_predictions': {'queue': 'maintenance'},

            # Import tasks - handled by worker (heavy processing)
            'app.tasks.import_file.*': {'queue': 'import'},
            'app.tasks.thumbnail.*': {'queue': 'thumbnail'},
            'app.tasks.sidecar.*': {'queue': 'sidecar'},

            # ML inference tasks - handled by ml-worker
            'app.tasks.ml.*': {'queue': 'ml'},
        },

        # Task default queue for unrouted tasks
        task_default_queue='default',

        # Result backend settings
        result_expires=86400,  # 24 hours

        # Task execution settings for resume capability
        task_acks_late=True,  # Acknowledge after task completes
        task_reject_on_worker_lost=True,  # Requeue if worker dies

        # Time limits (video transcoding may need longer)
        task_soft_time_limit=600,  # 10 minutes soft limit
        task_time_limit=900,  # 15 minutes hard limit

        # Periodic task schedule (Celery Beat)
        beat_schedule={
            'periodic-import-scan': {
                'task': 'app.tasks.scan.scan_directory',
                'schedule': int(os.environ.get('IMPORT_EVERY_SECONDS', '28800')),  # 8 hours default
                'args': ('/import', '/images'),
            },
            'recover-interrupted-tasks': {
                'task': 'app.tasks.scan.recover_interrupted_tasks',
                'schedule': 300,  # Every 5 minutes
            },
            'cleanup-old-tasks': {
                'task': 'app.tasks.scan.cleanup_old_tasks',
                'schedule': 86400,  # Once per day
                'args': (7,),  # Keep tasks for 7 days
            },
            'update-system-stats': {
                'task': 'app.tasks.scan.update_system_stats',
                'schedule': 21600,  # Every 6 hours
            },

            # ML-pipeline self-maintenance. Each is internally self-gating:
            # backfill returns immediately when nothing's missing,
            # recompute_all_centroids only enqueues for tags whose member
            # count changed, and apply_auto_accept_predictions short-circuits
            # when no above-threshold predictions are unattached.
            'ml-backfill-sweep': {
                'task': 'app.tasks.ml.backfill',
                'schedule': int(os.environ.get('ML_BACKFILL_EVERY_SECONDS', '86400')),  # daily
            },
            'apply-auto-accept-sweep': {
                'task': 'app.tasks.maintenance.apply_auto_accept_predictions',
                'schedule': int(os.environ.get('AUTO_ACCEPT_EVERY_SECONDS', '86400')),  # daily
            },
            'recompute-centroids-sweep': {
                'task': 'app.tasks.ml.recompute_all_centroids',
                'schedule': int(os.environ.get('CENTROIDS_EVERY_SECONDS', '604800')),  # weekly
            },
        },
    )

    if app:
        # Don't pass Flask config directly to Celery - it contains old-style keys
        # that conflict with Celery's new lowercase format.
        # The broker/backend are already set above from app.config.

        # Create a task base class that runs within Flask app context
        class ContextTask(celery.Task):
            def __call__(self, *args, **kwargs):
                with app.app_context():
                    return self.run(*args, **kwargs)

        celery.Task = ContextTask

    return celery


def create_celery_with_app():
    """
    Create Celery instance with Flask app context for standalone workers.
    This is used when running `celery -A app.celery_app:celery worker`.
    """
    from app import create_app
    flask_app = create_app()
    return make_celery(flask_app)


# Create celery instance with Flask app context
# This ensures workers have access to the database and Flask extensions
celery = create_celery_with_app()