1

video-gen-api/app/tasks/celery_app.py

@@ -57,8 +57,103 @@ def _derive_redis_db(url: str, db_no: int) -> str:
     return url.rstrip("/") + f"/{db_no}"
 
 
+def _parse_schedule_to_celery(schedule_str: str):
+    """将 schedule 字符串解析为 Celery 可识别的调度值。
+
+    - 纯数字：视为间隔秒数（返回 int）
+    - cron 表达式 (5 字段空格分隔)：返回 crontab 对象
+    """
+    from celery.schedules import crontab
+
+    s = (schedule_str or "").strip()
+    if not s:
+        return None
+    # 纯数字 → 间隔秒数
+    if s.isdigit():
+        return int(s)
+    # cron 表达式 (分 时 日 月 周)
+    parts = s.split()
+    if len(parts) == 5:
+        try:
+            return crontab(
+                minute=parts[0],
+                hour=parts[1],
+                day_of_month=parts[2],
+                month_of_year=parts[3],
+                day_of_week=parts[4],
+            )
+        except Exception:
+            logger.exception("解析 cron 表达式失败: %s", s)
+            return None
+    logger.warning("无法解析 schedule 表达式: %s", s)
+    return None
+
+
+# 同步引擎（用于 Beat 启动时加载数据库定时任务）
+_sync_engine = None
+
+
+def _get_sync_engine():
+    global _sync_engine
+    if _sync_engine is None:
+        from sqlalchemy import create_engine
+        from app.config import settings
+
+        db_url = settings.DATABASE_URL
+        # 异步 URL → 同步 URL（asyncpg → psycopg2 / aiosqlite → sqlite3）
+        if db_url.startswith("postgresql+asyncpg"):
+            db_url = db_url.replace("postgresql+asyncpg", "postgresql+psycopg2", 1)
+        elif db_url.startswith("sqlite+aiosqlite"):
+            db_url = db_url.replace("sqlite+aiosqlite", "sqlite", 1)
+        elif db_url.startswith("mysql+aiomysql"):
+            db_url = db_url.replace("mysql+aiomysql", "mysql+pymysql", 1)
+
+        _sync_engine = create_engine(db_url, pool_pre_ping=True)
+    return _sync_engine
+
+
+def _load_dynamic_beat_tasks() -> dict:
+    """从数据库加载活跃定时任务，返回 beat_schedule 格式的字典。
+
+    在 Celery 配置阶段同步调用，确保 Beat 启动时能读取到动态任务。
+    """
+    from sqlalchemy import select
+    from sqlalchemy.orm import Session
+
+    from app.models.scheduled_task import ScheduledTask
+
+    dynamic_schedule = {}
+    try:
+        engine = _get_sync_engine()
+        with Session(engine) as session:
+            result = session.execute(
+                select(ScheduledTask).where(ScheduledTask.is_active.is_(True))
+            )
+            tasks = result.scalars().all()
+
+            for task in tasks:
+                schedule_val = _parse_schedule_to_celery(task.schedule)
+                if schedule_val is None:
+                    logger.warning("定时任务 %s schedule 无效，跳过注册: %s", task.id, task.schedule)
+                    continue
+                beat_key = f"dynamic-scheduled-task-{task.id}"
+                dynamic_schedule[beat_key] = {
+                    "task": "execute_scheduled_task",
+                    "schedule": schedule_val,
+                    "args": (task.id,),
+                    "options": {"queue": RECOVERY_QUEUE},
+                }
+                logger.info(
+                    "动态注册定时任务到 Beat: %s (%s) schedule=%s",
+                    task.name, task.id, task.schedule,
+                )
+    except Exception:
+        logger.exception("加载动态定时任务失败，跳过")
+    return dynamic_schedule
+
+
 def _beat_schedule() -> dict:
-    schedule: dict = {}
+    schedule: dict = _load_dynamic_beat_tasks()
     if bool(getattr(settings, "POLL_DUE_DISPATCH_ENABLED", True)):
         schedule["dispatch-due-poll-tasks-every-minute"] = {
             "task": CeleryTaskName.DISPATCH_DUE_POLL.value,
@@ -293,122 +388,6 @@ else:
     celery_app = None
 
 
-def _parse_schedule_to_celery(schedule_str: str):
-    """将 schedule 字符串解析为 Celery 可识别的调度值。
-
-    - 纯数字：视为间隔秒数（返回 int）
-    - cron 表达式 (5 字段空格分隔)：返回 crontab 对象
-    """
-    from celery.schedules import crontab
-
-    s = (schedule_str or "").strip()
-    if not s:
-        return None
-    # 纯数字 → 间隔秒数
-    if s.isdigit():
-        return int(s)
-    # cron 表达式 (分 时 日 月 周)
-    parts = s.split()
-    if len(parts) == 5:
-        try:
-            return crontab(
-                minute=parts[0],
-                hour=parts[1],
-                day_of_month=parts[2],
-                month_of_year=parts[3],
-                day_of_week=parts[4],
-            )
-        except Exception:
-            logger.exception("解析 cron 表达式失败: %s", s)
-            return None
-    logger.warning("无法解析 schedule 表达式: %s", s)
-    return None
-
-
-def _setup_dynamic_beat_tasks(sender, **kwargs):
-    """从数据库加载活跃定时任务并注册到 Beat 调度。
-
-    通过 @celery_app.on_after_configure.connect 在 Celery 配置完成后执行，
-    适用于 Worker 和 Beat 启动场景。
-
-    注意：此信号可能运行在已有 event loop 的上下文中（Celery Beat 主 loop），
-    不能使用 run_until_complete 或 run_async，改用同步数据库查询。
-    """
-    if celery_app is None:
-        return
-
-    try:
-        active_tasks = _load_active_tasks_sync()
-    except Exception:
-        logger.exception("加载定时任务失败，跳过动态 Beat 注册")
-        return
-
-    for task in active_tasks:
-        schedule_val = _parse_schedule_to_celery(task.schedule)
-        if schedule_val is None:
-            logger.warning("定时任务 %s schedule 无效，跳过注册: %s", task.id, task.schedule)
-            continue
-        beat_key = f"dynamic-scheduled-task-{task.id}"
-        sender.conf.beat_schedule[beat_key] = {
-            "task": "execute_scheduled_task",
-            "schedule": schedule_val,
-            "args": (task.id,),
-            "options": {"queue": RECOVERY_QUEUE},
-        }
-        logger.info(
-            "动态注册定时任务到 Beat: %s (%s) schedule=%s",
-            task.name, task.id, task.schedule,
-        )
-
-
-# 同步引擎（仅在 Beat 动态注册定时任务时使用）
-_sync_engine = None
-
-
-def _get_sync_engine():
-    global _sync_engine
-    if _sync_engine is None:
-        from sqlalchemy import create_engine
-        from app.config import settings
-
-        db_url = settings.DATABASE_URL
-        # 异步 URL → 同步 URL（asyncpg → psycopg2 / aiosqlite → sqlite3）
-        if db_url.startswith("postgresql+asyncpg"):
-            db_url = db_url.replace("postgresql+asyncpg", "postgresql+psycopg2", 1)
-        elif db_url.startswith("sqlite+aiosqlite"):
-            db_url = db_url.replace("sqlite+aiosqlite", "sqlite", 1)
-        elif db_url.startswith("mysql+aiomysql"):
-            db_url = db_url.replace("mysql+aiomysql", "mysql+pymysql", 1)
-
-        _sync_engine = create_engine(db_url, pool_pre_ping=True)
-    return _sync_engine
-
-
-def _load_active_tasks_sync():
-    """同步查询活跃的定时任务。"""
-    from sqlalchemy import select
-    from sqlalchemy.orm import Session
-
-    from app.models.scheduled_task import ScheduledTask
-
-    engine = _get_sync_engine()
-    with Session(engine) as session:
-        result = session.execute(
-            select(ScheduledTask).where(ScheduledTask.is_active.is_(True))
-        )
-        #  detach objects so they can be used after session closes
-        tasks = result.scalars().all()
-        # Expunge all to detach from session
-        for t in tasks:
-            session.expunge(t)
-        return tasks
-
-
-# 注册信号（放在函数定义之后，确保函数已存在）
-if celery_app is not None:
-    celery_app.on_after_configure.connect(_setup_dynamic_beat_tasks)
-
-
 async def _try_acquire_startup_recovery_lock() -> bool:
     """任意 worker 启动时都可尝试抢恢复投递锁，避免依赖 hostname 命名。"""
     from app.services.redis_registry_service import redis_acquire_lock