🎤【Python】TikTok Liveのコメントをリアルタイムで音声読み上げするボットを作ってみた【tiktok-voice-bot】

こんにちは、@milkc0deです！

TikTok Liveで流れるコメント・フォロー・ギフトをリアルタイムで音声読み上げしてくれるボットをPythonで作りました。
ライブラリや非同期・音声処理などを組み合わせて、Windows環境でサクッと動きます！

GitHubにも公開しています👇
🔗 https://github.com/milkc0de/tiktok-voice-bot

📌 このスクリプトでできること

TikTokのライブコメントを取得（TikTokLiveライブラリ使用）
コメント・フォロー・ギフトを自動で日本語音声に変換（gTTS）
音声ファイルを作ってWAV形式で即時再生（pydub + playsound）
Queue + スレッド + asyncio で安定再生＆同時処理を実現

💻 動作環境

Windows 10 / 11
Python 3.9以上
ffmpeg（pydubのMP3→WAV変換に必要）

🔧 インストール

pip install TikTokLive gTTS pydub playsound

Windowsユーザーは ffmpeg を準備してください：

https://ffmpeg.org/download.html から Windows 版をダウンロード
ffmpeg.exe のあるフォルダを環境変数 PATH に追加
または pydub.AudioSegment.converter に直接パスを指定してもOK

🧠 ソースコード全体

import asyncio
import os
import threading
import queue
from TikTokLive import TikTokLiveClient
from TikTokLive.events import CommentEvent as EventComment
from TikTokLive.events import FollowEvent as EventFollow
from TikTokLive.events import GiftEvent as EventGift
from gtts import gTTS
from pydub import AudioSegment
from playsound import playsound
import tempfile
import traceback

class TikTokLiveClientWithQueue():
    def __init__(self, max_queue_size=20):
        self._queue = queue.Queue(maxsize=max_queue_size)
        self._loop = asyncio.new_event_loop()
        self._thread = threading.Thread(target=self._run_event_loop, daemon=True)
        self._thread.start()

    def add_to_queue(self, text):
        try:
            self._queue.put(text, timeout=5)
        except queue.Full:
            print("Queue is full. Dropping message:", text)

    def _run_event_loop(self):
        asyncio.set_event_loop(self._loop)
        try:
            self._loop.run_until_complete(self._process_queue())
        except Exception as e:
            print(f"Error in event loop: {e}")
            traceback.print_exc()

    async def _process_queue(self):
        while True:
            try:
                text = self._queue.get()
                print(f"Processing text: {text}")
                await speak_sync(text)
                self._queue.task_done()
            except Exception as e:
                print(f"Error processing queue: {e}")
                traceback.print_exc()

async def speak_sync(text, lang='ja'):
    try:
        with tempfile.NamedTemporaryFile(delete=False, suffix='.mp3') as temp_mp3_file:
            temp_mp3_path = temp_mp3_file.name
        with tempfile.NamedTemporaryFile(delete=False, suffix='.wav') as temp_wav_file:
            temp_wav_path = temp_wav_file.name

        tts = gTTS(text=text, lang=lang)
        tts.save(temp_mp3_path)

        audio = AudioSegment.from_mp3(temp_mp3_path)
        audio.export(temp_wav_path, format="wav")

        print("Playing WAV file...")
        playsound(temp_wav_path)
        print("Audio playback completed.")
    except Exception as e:
        print(f"Error during speech synthesis or playback: {e}")
        traceback.print_exc()
    finally:
        if os.path.exists(temp_mp3_path):
            os.remove(temp_mp3_path)
        if os.path.exists(temp_wav_path):
            os.remove(temp_wav_path)

USERNAME = ""
client = TikTokLiveClient(unique_id=USERNAME)
th = TikTokLiveClientWithQueue()

@client.on(EventComment)
async def on_comment(event: EventComment):
    text = f"{event.user.nickname}さんのコメント。{event.comment}"
    print(f"Received comment: {text}")
    th.add_to_queue(text)

@client.on(EventFollow)
async def on_follow(event: EventFollow):
    text = f"{event.user.nickname}さんがフォローしました。ありがとうございます。"
    print(f"Received follow: {text}")
    th.add_to_queue(text)

@client.on(EventGift)
async def on_gift(event: EventGift):
    text = f"{event.user.nickname}さんがギフトをくれました。ありがとうございます。"
    print(f"Received gift: {text}")
    th.add_to_queue(text)

async def main():
    try:
        if not client.connected:
            print("Connecting to TikTokLive...")
            await client.start()
        await asyncio.sleep(1)
    except Exception as e:
        print("Stopping...")
        if client.connected:
            await client.disconnect()
        raise e

if __name__ == "__main__":
    asyncio.run(main())

📌 補足

playsound はブロッキングなため、スレッドで非同期処理と分離
コメント読み上げは Queue を使って順番制御（暴走防止）
gTTS → MP3 → WAV にしているのは playsound の安定性のため（MP3だと一部環境で問題が出やすい）

💡 発展アイデア

読み上げ音声をボイスチェンジャーで加工
自分の配信キャラの声に変換（RVCなど）
コメントの内容によって効果音やBGMを追加

📦 GitHub

GitHubでも公開しています！
👉 https://github.com/milkc0de/tiktok-voice-bot

🙋‍♀️さいごに

配信の臨場感をグッと上げてくれる音声ボット、試してみませんか？
コメントを見逃さず、耳からキャッチできるのでとても便利です！

🏷️ タグ

#Python #TikTokLive #TTS #音声読み上げ #非同期処理 #配信支援