wehub-resource-sync
141 lines
4.6 KiB
Python
141 lines
4.6 KiB
Python
"""
|
|
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() |