异步函数安全调用方式, 可以当成库或者当成main

def run_async_safely(coro): """安全执行协程，自动适配已有事件循环的场景""" try: # 检查是否已有运行中的事件循环 loop = asyncio.get_running_loop() except RuntimeError: # 无运行循环，直接用 asyncio.run return asyncio.run(coro) else: # 已有运行循环(如Jupyter/IPython) if loop.is_running(): # 提交任务并等待完成(兼容运行中的循环) task = loop.create_task(coro) # 等待任务完成(仅在支持 await 的环境中生效，或用同步方式) return asyncio.run_coroutine_threadsafe(coro, loop).result() else: # 循环存在但未运行，直接执行 return loop.run_until_complete(coro)