AiOpsWechat/celery_app.py

from celery import Celery
import celery.schedules
from redbeat import RedBeatSchedulerEntry
from datetime import timedelta


from services.redis_service import RedisService
from services.kafka_service import KafkaService
from services.biz_service import BizService

from config import load_config,conf
from urllib.parse import quote
import asyncio,os


load_config()

KAFKA_BOOTSTRAP_SERVERS = conf().get("kafka_bootstrap_servers")
KAFKA_TOPIC = 'topic.ai.ops.wx'
KAFKA_GROUP_ID = 'ai-ops-wx'

redis_host=conf().get("redis_host")
redis_port=conf().get("redis_port")
redis_password=conf().get("redis_password") 
redis_db=conf().get("redis_db") 
encoded_password = quote(redis_password) 
# 配置 Celery
celery_app = Celery(
    "worker",
    broker=f"redis://:{encoded_password}@{redis_host}:{redis_port}/{redis_db}",
    backend=f"redis://:{encoded_password}@{redis_host}:{redis_port}/{redis_db}",
    include=['tasks']
)

# 配置 redbeat 作为 Celery Beat 调度器
celery_app.conf.update(
    timezone="Asia/Shanghai",  # 设定时区
    beat_scheduler="redbeat.RedBeatScheduler",  # 使用 RedBeat 作为调度器
    redbeat_redis_url=f"redis://:{encoded_password}@{redis_host}:{redis_port}/{redis_db}"  # redbeat 存储任务调度信息的 Redis
    ,
    redbeat_lock_timeout=60,  # 避免多个 Beat 实例冲突
    beat_max_loop_interval=5  # 让 Celery Beat 每 5 秒检查一次任务
)

# task_name = "tasks.scheduled_task"
# # 任务执行间隔（每 10 秒执行一次）
# schedule = celery.schedules.schedule(timedelta(seconds=3))

# # RedBeat 任务唯一 ID
# redbeat_entry = RedBeatSchedulerEntry(
#     name="redbeat:scheduled_task",  # 任务 ID
#     task=task_name,  # 任务名称
#     schedule=schedule,  # 任务调度时间
#     args=[],
#     app=celery_app
        
# )

# # 保存任务到 Redis
# redbeat_entry.save()


# tasks_schedule = [
#     ("redbeat:scheduled_task", "tasks.scheduled_task", 3),
#     ("redbeat:scheduled_task_sync_wx", "tasks.scheduled_task_sync_wx", 15),
# ]

# # 创建并保存 RedBeat 任务
# for task_id, task_name, interval in tasks_schedule:
#     redbeat_entry = RedBeatSchedulerEntry(
#         name=task_id,
#         task=task_name,
#         schedule=celery.schedules.schedule(timedelta(seconds=interval)),
#         args=[],
#         app=celery_app
#     )
#     redbeat_entry.save()

# 获取配置文件中的 redis_config、kafka_config、gewe_config
redis_config = {
    'host': redis_host,
    'port': redis_port,
    'password': redis_password,
    'db': redis_db,
}


kafka_config = {
    'bootstrap_servers': KAFKA_BOOTSTRAP_SERVERS,
    'topic': KAFKA_TOPIC,
    'group_id': KAFKA_GROUP_ID,
}
gewe_config = {
    'api_url': "http://api.geweapi.com/gewe",
}

scheduled_task_sync_wx_info_interval = 10
environment = os.environ.get('environment', 'default') 
if environment != 'default':
    scheduled_task_sync_wx_info_interval = 60*11

# 定义定时任务列表 (任务 ID, 任务名称, 执行间隔秒, 任务参数)
tasks_schedule = [
    #("redbeat:scheduled_task", "tasks.scheduled_task", 3, []),
    #("redbeat:scheduled_task_sync_wx", "tasks.scheduled_task_sync_wx", 15, []),
    ("redbeat:scheduled_task_sync_wx_info", "tasks.scheduled_task_sync_wx_info", scheduled_task_sync_wx_info_interval, [redis_config, kafka_config, gewe_config]),

]



# 注册 RedBeat 任务
for task_id, task_name, interval, task_args in tasks_schedule:
    redbeat_entry = RedBeatSchedulerEntry(
        name=task_id,
        task=task_name,
        schedule=celery.schedules.schedule(timedelta(seconds=interval)),
        args=task_args, 
        app=celery_app
    )
    redbeat_entry.save()