1

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

@@ -1,4 +1,3 @@
-import asyncio
 import logging
 
 from celery import Celery
@@ -326,44 +325,25 @@ def _parse_schedule_to_celery(schedule_str: str):
     return None
 
 
-@celery_app.on_after_configure.connect  # type: ignore
 def _setup_dynamic_beat_tasks(sender, **kwargs):
     """从数据库加载活跃定时任务并注册到 Beat 调度。
 
     通过 @celery_app.on_after_configure.connect 在 Celery 配置完成后执行，
     适用于 Worker 和 Beat 启动场景。
 
-    注意：此信号运行在 Beat/Worker 主线程，不能使用 run_async（single_loop 在另一线程），
+    注意：此信号可能运行在已有 event loop 的上下文中（Celery Beat 主 loop），
-    这里用独立 event loop 同步运行 async 查询，避免 asyncpg Future 跨 loop 错误。
+    不能使用 run_until_complete 或 run_async，改用同步数据库查询。
     """
     if celery_app is None:
         return
 
-    async def _load():
-        from sqlalchemy import select
-
-        from app.models.base import async_session
-        from app.models.scheduled_task import ScheduledTask
-
-        async with async_session() as db:
-            result = await db.execute(
-                select(ScheduledTask).where(ScheduledTask.is_active.is_(True))
-            )
-            return result.scalars().all()
-
-    active_tasks = None
     try:
-        # 在当前线程创建独立 loop 同步运行，不经过 single_loop
+        active_tasks = _load_active_tasks_sync()
-        loop = asyncio.new_event_loop()
-        try:
-            active_tasks = loop.run_until_complete(_load())
-        finally:
-            loop.close()
     except Exception:
         logger.exception("加载定时任务失败，跳过动态 Beat 注册")
         return
 
-    for task in active_tasks or []:
+    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)
@@ -381,6 +361,54 @@ def _setup_dynamic_beat_tasks(sender, **kwargs):
         )
 
 
+# 同步引擎（仅在 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