RuntimeError事件循环已运行：Telegram TikTok下载Bot报错求助

错误原因 #

aiogram是异步框架，运行时已启动一个事件循环，但你使用的TikTokApi同步版本内部又调用了asyncio.get_event_loop().run_until_complete()，导致同一个事件循环被重复启动，触发RuntimeError: This event loop is already running。

修复当前TikTokApi的方案 #

方案1：改用TikTokApi的异步版本 #

TikTokApi支持异步操作，直接在aiogram的异步处理器中使用异步方法，避免循环冲突。修改后的代码如下：

# -*- coding: utf-8 -*-
from aiogram import Bot, Dispatcher, executor, types
import logging
import config
from TikTokApi import TikTokApi

logging.basicConfig(level=logging.INFO)
bot = Bot(token=config.token)
dp = Dispatcher(bot)


@dp.message_handler(commands=['start'])
async def send_welcome(message: types.Message):
    await bot.send_message(message.from_user.id,
                           f'hello',
                           parse_mode="Markdown")


@dp.message_handler(content_types=['text'])
async def text(message: types.Message):
    if message.text.startswith('https://vt.tiktok.com'):
        video_url = message.text
        # 异步初始化TikTokApi
        async with TikTokApi() as api:
            video = await api.video(url=video_url)
            # 异步获取视频字节数据
            video_data = await video.bytes()
            
            file_name = f"{message.from_user.id}.mp4"
            with open(file_name, "wb") as out_file:
                out_file.write(video_data)
            
            # 可选：将下载好的视频发回给用户
            await bot.send_video(message.chat.id, open(file_name, "rb"))


if __name__ == "__main__":
    executor.start_polling(dp, skip_updates=True)

方案2：用线程池隔离同步操作 #

如果不想修改异步代码结构，可以把TikTokApi的同步调用放到线程池中执行，避免阻塞当前事件循环：

# -*- coding: utf-8 -*-
from aiogram import Bot, Dispatcher, executor, types
import logging
import config
from TikTokApi import TikTokApi
import asyncio

logging.basicConfig(level=logging.INFO)
bot = Bot(token=config.token)
dp = Dispatcher(bot)


@dp.message_handler(commands=['start'])
async def send_welcome(message: types.Message):
    await bot.send_message(message.from_user.id,
                           f'hello',
                           parse_mode="Markdown")


@dp.message_handler(content_types=['text'])
async def text(message: types.Message):
    if message.text.startswith('https://vt.tiktok.com'):
        video_url = message.text
        
        # 定义同步下载函数
        def download_tiktok():
            with TikTokApi() as api:
                video = api.video(url=video_url)
                return video.bytes()
        
        # 在线程池中执行同步操作
        loop = asyncio.get_event_loop()
        video_data = await loop.run_in_executor(None, download_tiktok)
        
        file_name = f"{message.from_user.id}.mp4"
        with open(file_name, "wb") as out_file:
            out_file.write(video_data)
        
        await bot.send_video(message.chat.id, open(file_name, "rb"))


if __name__ == "__main__":
    executor.start_polling(dp, skip_updates=True)

其他TikTok视频下载方案 #

1. 直接解析视频真实地址 #

通过异步请求TikTok短链接，从页面内容中提取真实视频地址，无需依赖第三方库：

# -*- coding: utf-8 -*-
from aiogram import Bot, Dispatcher, executor, types
import logging
import config
import httpx
import re

logging.basicConfig(level=logging.INFO)
bot = Bot(token=config.token)
dp = Dispatcher(bot)


async def get_tiktok_real_url(short_url):
    async with httpx.AsyncClient(follow_redirects=True) as client:
        response = await client.get(short_url)
        # 从页面源码中匹配真实视频地址（逻辑可能随TikTok页面更新失效）
        video_match = re.search(r'"playAddr":"(.*?)"', response.text)
        if video_match:
            return video_match.group(1).replace('\\u002F', '/')
    return None


@dp.message_handler(commands=['start'])
async def send_welcome(message: types.Message):
    await bot.send_message(message.from_user.id,
                           f'hello',
                           parse_mode="Markdown")


@dp.message_handler(content_types=['text'])
async def text(message: types.Message):
    if message.text.startswith('https://vt.tiktok.com'):
        real_url = await get_tiktok_real_url(message.text)
        if real_url:
            async with httpx.AsyncClient() as client:
                video_data = await client.get(real_url)
                file_name = f"{message.from_user.id}.mp4"
                with open(file_name, "wb") as f:
                    f.write(video_data.content)
                await bot.send_video(message.chat.id, open(file_name, "rb"))


if __name__ == "__main__":
    executor.start_polling(dp, skip_updates=True)

2. 使用轻量异步下载库 #

比如TikTokDownloader（需先安装：pip install TikTokDownloader），适配异步框架的同时简化下载逻辑：

# -*- coding: utf-8 -*-
from aiogram import Bot, Dispatcher, executor, types
import logging
import config
from TikTokDownloader import download_video

logging.basicConfig(level=logging.INFO)
bot = Bot(token=config.token)
dp = Dispatcher(bot)


@dp.message_handler(commands=['start'])
async def send_welcome(message: types.Message):
    await bot.send_message(message.from_user.id,
                           f'hello',
                           parse_mode="Markdown")


@dp.message_handler(content_types=['text'])
async def text(message: types.Message):
    if message.text.startswith('https://vt.tiktok.com'):
        # 异步下载视频
        result = await download_video(message.text, output_path="./")
        if result:
            await bot.send_video(message.chat.id, open(result["filename"], "rb"))


if __name__ == "__main__":
    executor.start_polling(dp, skip_updates=True)

内容的提问来源于stack exchange，提问作者sokol