alien_py_cloud/alien_util/celery_app.py

from celery import Celery
from celery.schedules import crontab
from alien_gateway.config import settings

# 创建 Celery 应用：broker / backend 由 settings 自动适配单机或 Sentinel
celery_app = Celery(
    "alien_cloud",
    broker=settings.CELERY_BROKER_URL,
    backend=settings.CELERY_BACKEND_URL,
    include=["alien_util.tasks.contract_tasks"],
)

_celery_conf = dict(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="Asia/Shanghai",
    enable_utc=True,
    beat_schedule={
        "check-contract-expiry": {
            "task": "alien_util.tasks.contract_tasks.check_contract_expiry",
            "schedule": crontab(hour=0, minute=1),  # 每天凌晨 0:01 执行
        },
    },
)

# 哨兵模式下必须传 transport_options 让 kombu 感知 master_name 等参数
if settings.CELERY_TRANSPORT_OPTIONS is not None:
    _celery_conf["broker_transport_options"] = settings.CELERY_TRANSPORT_OPTIONS
    _celery_conf["result_backend_transport_options"] = settings.CELERY_TRANSPORT_OPTIONS

celery_app.conf.update(**_celery_conf)