autoclip/backend/core/celery_app.py

"""
Celery应用配置
任务队列配置和初始化
"""

import os
from celery import Celery
from celery.schedules import crontab
from pathlib import Path

# 设置默认配置模块
# os.environ.setdefault('CELERY_CONFIG_MODULE', 'backend.core.celery_app')

# 创建Celery应用
celery_app = Celery('autoclip')

# 配置Celery
class CeleryConfig:
    """Celery配置类"""
    
    # 任务序列化格式
    task_serializer = 'json'
    accept_content = ['json']
    result_serializer = 'json'
    timezone = 'Asia/Shanghai'
    enable_utc = True
    
    # Redis配置
    broker_url = os.getenv('REDIS_URL', 'redis://localhost:6379/0')
    result_backend = os.getenv('REDIS_URL', 'redis://localhost:6379/0')
    
    # 任务配置
    task_always_eager = os.getenv('CELERY_ALWAYS_EAGER', 'False').lower() == 'true'  # 生产环境异步执行
    task_eager_propagates = True
    
    # 工作进程配置
    worker_prefetch_multiplier = 1
    worker_max_tasks_per_child = 1000
    worker_disable_rate_limits = True
    worker_concurrency = 1  # 强制设置并发数为1，防止重复处理
    
    # 任务路由
    task_routes = {
        'backend.tasks.processing.*': {'queue': 'processing'},
        'backend.tasks.video.*': {'queue': 'video'},
        'backend.tasks.notification.*': {'queue': 'notification'},
        'backend.tasks.upload.*': {'queue': 'upload'},  # 添加upload任务路由
        'backend.tasks.import_processing.*': {'queue': 'processing'},  # 导入任务路由
    }
    
    # 定时任务配置
    beat_schedule = {
        'cleanup-expired-tasks': {
            'task': 'backend.tasks.maintenance.cleanup_expired_tasks',
            'schedule': crontab(hour=2, minute=0),  # 每天凌晨2点
        },
        'health-check': {
            'task': 'backend.tasks.maintenance.health_check',
            'schedule': crontab(minute='*/5'),  # 每5分钟
        },
    }
    
    # 结果配置
    result_expires = 3600  # 1小时
    task_ignore_result = False
    
    # 日志配置
    worker_log_format = '[%(asctime)s: %(levelname)s/%(processName)s] %(message)s'
    worker_task_log_format = '[%(asctime)s: %(levelname)s/%(processName)s] [%(task_name)s(%(task_id)s)] %(message)s'

# 应用配置
celery_app.config_from_object(CeleryConfig)


def _is_desktop_mode() -> bool:
    return os.getenv("AUTOCLIP_DESKTOP_MODE", "").lower() in {"1", "true", "yes"}


class _LocalAsyncResult:
    """轻量级 AsyncResult 替身，桌面模式本地线程执行时返回。"""

    def __init__(self, task_id: str):
        self.id = task_id
        self.task_id = task_id
        self.state = "PENDING"

    def get(self, *args, **kwargs):
        return None

    def ready(self) -> bool:
        return False


class DesktopAwareTask(celery_app.Task):
    """桌面安装包里没有 Redis broker，生产 core.celery_app 又指向 redis://localhost。

    所有端点都用 `task.delay(...)` / `apply_async(...)` 派发任务，默认会把任务塞进
    Redis 队列 —— 桌面模式下没人消费，于是永远卡在 0%「初始化中」。

    这里在桌面模式下把 apply_async 改成「在后台守护线程里同步执行 apply()」：
    不依赖任何 broker，立即返回，进度照常写库供前端轮询。生产模式行为不变。
    """

    def apply_async(self, args=None, kwargs=None, task_id=None, **options):
        if _is_desktop_mode():
            import threading
            import uuid

            tid = task_id or str(uuid.uuid4())
            call_args = list(args) if args else []
            call_kwargs = dict(kwargs) if kwargs else {}

            def _run():
                try:
                    self.apply(args=call_args, kwargs=call_kwargs, task_id=tid)
                except Exception as exc:  # noqa: BLE001
                    import logging
                    logging.getLogger(__name__).error(
                        f"桌面模式本地执行任务失败 {self.name} ({tid}): {exc}", exc_info=True
                    )

            threading.Thread(target=_run, name=f"task-{self.name}", daemon=True).start()
            return _LocalAsyncResult(tid)

        return super().apply_async(args=args, kwargs=kwargs, task_id=task_id, **options)


# 桌面模式下让所有 @celery_app.task 使用上面的本地执行基类
celery_app.Task = DesktopAwareTask

# 自动发现任务
celery_app.autodiscover_tasks([
    'backend.tasks.processing',
    'backend.tasks.video', 
    'backend.tasks.notification',
    'backend.tasks.maintenance',
    'backend.tasks.import_processing'  # 添加导入处理任务
])

if __name__ == '__main__':
    celery_app.start()