From bedf39d4f746a617e0e159d983f078b2082971f0 Mon Sep 17 00:00:00 2001 From: wwwwwwwww <526125649@qq.com> Date: Fri, 14 Aug 2026 15:35:54 +0800 Subject: [PATCH 1/5] 1 --- video-gen-api/.env | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/video-gen-api/.env b/video-gen-api/.env index dfc96f81..ae0a2d19 100644 --- a/video-gen-api/.env +++ b/video-gen-api/.env @@ -1,7 +1,7 @@ # App APP_NAME=VideoGen API APP_VERSION=1.0.0 -DEBUG=true +DEBUG=false SECRET_KEY=local-dev-secret-key-not-for-production # Database (PostgreSQL) From 556ea4dbd24fa0509ab7863ebf4feba711f26c05 Mon Sep 17 00:00:00 2001 From: wwwwwwwww <526125649@qq.com> Date: Fri, 14 Aug 2026 15:39:22 +0800 Subject: [PATCH 2/5] 1 --- video-gen-api/app/tasks/scheduled_tasks.py | 29 +++++++++------------- 1 file changed, 12 insertions(+), 17 deletions(-) diff --git a/video-gen-api/app/tasks/scheduled_tasks.py b/video-gen-api/app/tasks/scheduled_tasks.py index eb70cc58..47960446 100644 --- a/video-gen-api/app/tasks/scheduled_tasks.py +++ b/video-gen-api/app/tasks/scheduled_tasks.py @@ -31,21 +31,17 @@ from app.tasks.celery_app import celery_app logger = logging.getLogger("video_gen") -def _update_task_status(task_id: str, status: str, error_msg: str | None = None) -> None: +async def _update_task_status(task_id: str, status: str, error_msg: str | None = None) -> None: """更新任务最后执行状态。""" - - async def _do(): - async with async_session() as db: - result = await db.execute(select(ScheduledTask).where(ScheduledTask.id == task_id)) - task = result.scalar_one_or_none() - if task is None: - return - task.last_run_at = datetime.now(timezone.utc).isoformat() - task.last_status = status - task.last_error = error_msg - await db.commit() - - run_async(_do()) + async with async_session() as db: + result = await db.execute(select(ScheduledTask).where(ScheduledTask.id == task_id)) + task = result.scalar_one_or_none() + if task is None: + return + task.last_run_at = datetime.now(timezone.utc).isoformat() + task.last_status = status + task.last_error = error_msg + await db.commit() def _execute_internal_method(config: str | None) -> dict: @@ -91,16 +87,15 @@ def execute_scheduled_task(self, task_id: str): if not task.is_active: logger.info("定时任务已禁用，跳过执行: %s", task_id) return - task_config = task.config try: exec_result = _execute_internal_method(task_config) - _update_task_status(task_id, "success") + await _update_task_status(task_id, "success") logger.info("定时任务执行成功: %s -> %s", task_id, exec_result) except Exception as e: error_msg = str(e) - _update_task_status(task_id, "error", error_msg) + await _update_task_status(task_id, "error", error_msg) logger.exception("定时任务执行失败: %s", task_id) run_async(_run()) From 7ab5c9a5c9ecb6a3975dcab7882388815ac18165 Mon Sep 17 00:00:00 2001 From: wwwwwwwww <526125649@qq.com> Date: Fri, 14 Aug 2026 15:42:12 +0800 Subject: [PATCH 3/5] 1 --- .../app/services/bank/sync_service.py | 30 +++++++------------ video-gen-api/app/tasks/scheduled_tasks.py | 13 ++++++-- 2 files changed, 21 insertions(+), 22 deletions(-) diff --git a/video-gen-api/app/services/bank/sync_service.py b/video-gen-api/app/services/bank/sync_service.py index cc4db468..7d74874c 100644 --- a/video-gen-api/app/services/bank/sync_service.py +++ b/video-gen-api/app/services/bank/sync_service.py @@ -34,7 +34,7 @@ logger = logging.getLogger("video_gen") _PAGE_SIZE = 100 -def sync_bank_transactions( +async def sync_bank_transactions( url: str = "", api_key: str = "", acct_no: str = "", @@ -85,24 +85,16 @@ def sync_bank_transactions( "errors": [], } - import asyncio - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - try: - loop.run_until_complete( - _do_sync( - url=url, - api_key=api_key, - acct_no=acct_no, - start_date=start_date, - end_date=end_date, - dc_flag=dc_flag, - sync_batch=sync_batch, - stats=stats, - ) - ) - finally: - loop.close() + await _do_sync( + url=url, + api_key=api_key, + acct_no=acct_no, + start_date=start_date, + end_date=end_date, + dc_flag=dc_flag, + sync_batch=sync_batch, + stats=stats, + ) logger.info( "银行流水同步完成: batch=%s, acct=%s, 翻页=%d, 获取=%d, 新增=%d, 重复=%d", diff --git a/video-gen-api/app/tasks/scheduled_tasks.py b/video-gen-api/app/tasks/scheduled_tasks.py index 47960446..04a4daab 100644 --- a/video-gen-api/app/tasks/scheduled_tasks.py +++ b/video-gen-api/app/tasks/scheduled_tasks.py @@ -48,7 +48,12 @@ def _execute_internal_method(config: str | None) -> dict: """执行内部方法调用。 配置 JSON 中 module 和 function 指定要调用的函数，其余字段作为 kwargs 传入。 + 支持同步函数和异步函数（async def）。 """ + import asyncio + import importlib + from inspect import iscoroutinefunction + cfg = json.loads(config or "{}") module_path = cfg.pop("module", "").strip() function_name = cfg.pop("function", "").strip() @@ -56,8 +61,6 @@ def _execute_internal_method(config: str | None) -> dict: if not module_path or not function_name: raise ValueError("内部方法需要指定 module 和 function") - import importlib - module = importlib.import_module(module_path) func = getattr(module, function_name, None) if func is None or not callable(func): @@ -65,7 +68,11 @@ def _execute_internal_method(config: str | None) -> dict: # 剩余字段作为 kwargs 传给函数 start = time.monotonic() - result = func(**cfg) + if iscoroutinefunction(func): + # 异步函数：在当前事件循环中 await + result = asyncio.get_event_loop().run_until_complete(func(**cfg)) + else: + result = func(**cfg) duration_ms = int((time.monotonic() - start) * 1000) return { "duration_ms": duration_ms, From 4a2355a9029433cf8cf88b0a271381259d40ffb6 Mon Sep 17 00:00:00 2001 From: wwwwwwwww <526125649@qq.com> Date: Fri, 14 Aug 2026 15:47:51 +0800 Subject: [PATCH 4/5] 1 --- video-gen-api/app/tasks/scheduled_tasks.py | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/video-gen-api/app/tasks/scheduled_tasks.py b/video-gen-api/app/tasks/scheduled_tasks.py index 04a4daab..375aaf6f 100644 --- a/video-gen-api/app/tasks/scheduled_tasks.py +++ b/video-gen-api/app/tasks/scheduled_tasks.py @@ -44,13 +44,12 @@ async def _update_task_status(task_id: str, status: str, error_msg: str | None = await db.commit() -def _execute_internal_method(config: str | None) -> dict: +async def _execute_internal_method(config: str | None) -> dict: """执行内部方法调用。 配置 JSON 中 module 和 function 指定要调用的函数，其余字段作为 kwargs 传入。 支持同步函数和异步函数（async def）。 """ - import asyncio import importlib from inspect import iscoroutinefunction @@ -69,8 +68,8 @@ def _execute_internal_method(config: str | None) -> dict: # 剩余字段作为 kwargs 传给函数 start = time.monotonic() if iscoroutinefunction(func): - # 异步函数：在当前事件循环中 await - result = asyncio.get_event_loop().run_until_complete(func(**cfg)) + # 异步函数：直接 await + result = await func(**cfg) else: result = func(**cfg) duration_ms = int((time.monotonic() - start) * 1000) @@ -97,7 +96,7 @@ def execute_scheduled_task(self, task_id: str): task_config = task.config try: - exec_result = _execute_internal_method(task_config) + exec_result = await _execute_internal_method(task_config) await _update_task_status(task_id, "success") logger.info("定时任务执行成功: %s -> %s", task_id, exec_result) except Exception as e: From 5eed7686c98ddc186941ee9c3e27f26a323168db Mon Sep 17 00:00:00 2001 From: wwwwwwwww <526125649@qq.com> Date: Fri, 14 Aug 2026 15:52:52 +0800 Subject: [PATCH 5/5] 1 --- video-gen-api/app/admin_api/scheduled_tasks/routes.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/video-gen-api/app/admin_api/scheduled_tasks/routes.py b/video-gen-api/app/admin_api/scheduled_tasks/routes.py index d10cb1a7..5baed0e3 100644 --- a/video-gen-api/app/admin_api/scheduled_tasks/routes.py +++ b/video-gen-api/app/admin_api/scheduled_tasks/routes.py @@ -148,9 +148,10 @@ async def run_task( if task is None: raise HTTPException(status_code=404, detail="任务不存在") + from app.tasks.celery_app import RECOVERY_QUEUE from app.tasks.scheduled_tasks import execute_scheduled_task - execute_scheduled_task.apply_async(args=[task_id]) + execute_scheduled_task.apply_async(args=[task_id], queue=RECOVERY_QUEUE) return {"message": "任务已提交执行"}