1

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",

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,