【Python】VOICE BOXを用いたDiscord読み上げBOTの作成

import discord
from discord.ext import commands
import aiohttp
import asyncio
import os
import json
import datetime
import re
import sys

CONFIG_PATH = "config.json"
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
    config = json.load(f)

DICTIONARY_PATH = "data/dictionary.json"
TTS_CACHE_DIR = "data/tts_cache"
SECURITY_MESSAGE_FILE = "data/messages.json"

os.makedirs("data", exist_ok=True)
os.makedirs(TTS_CACHE_DIR, exist_ok=True)

if not os.path.exists(DICTIONARY_PATH):
    with open(DICTIONARY_PATH, "w", encoding="utf-8") as f:
        json.dump({}, f, ensure_ascii=False, indent=4)

if not os.path.exists(SECURITY_MESSAGE_FILE):
    with open(SECURITY_MESSAGE_FILE, "w", encoding="utf-8") as f:
        json.dump([], f, ensure_ascii=False, indent=4)

TOKEN = config["discord_token"]
VOICE_CHANNEL_ID = config["voice_channel_id"]
TEXT_CHANNEL_ID = config["text_channel_id"]
VOICEVOX_API_BASE_URL = config.get("voicevox_url", "http://localhost:50021")
SPEAKER_ID = config.get("speaker_id", 1)
FFMPEG_PATH = config.get("ffmpeg_path", "ffmpeg")

intents = discord.Intents.default()
intents.messages = True
intents.guilds = True
intents.voice_states = True
intents.message_content = True
bot = commands.Bot(command_prefix="!", intents=intents)

def load_dictionary():
    with open(DICTIONARY_PATH, "r", encoding="utf-8") as f:
        return json.load(f)

def save_dictionary(data):
    with open(DICTIONARY_PATH, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=4)

@bot.event
async def on_ready():
    print(f"Logged in as {bot.user}")

@bot.event
async def on_voice_state_update(member, before, after):
    if member.bot:
        return

    channel = bot.get_channel(VOICE_CHANNEL_ID)

    if before.channel != after.channel:
        if after.channel == channel:
            await announce_voice(f"{member.display_name} joined.")
        elif before.channel == channel:
            non_bot_members = [m for m in before.channel.members if not m.bot]
            if non_bot_members:
                await announce_voice(f"{member.display_name} left.")
            else:
                await asyncio.sleep(3)
                for vc in bot.voice_clients:
                    if vc.channel == channel:
                        await vc.disconnect()
                        await asyncio.sleep(1)
                        os.execv(sys.executable, ["python"] + sys.argv)

@bot.event
async def on_message(message):
    if message.author.bot:
        return

    if message.channel.id == TEXT_CHANNEL_ID:
        with open(SECURITY_MESSAGE_FILE, "r", encoding="utf-8") as f:
            logs = json.load(f)

        logs.append({
            "timestamp": datetime.datetime.now().isoformat(),
            "author": message.author.name,
            "content": message.content
        })

        with open(SECURITY_MESSAGE_FILE, "w", encoding="utf-8") as f:
            json.dump(logs, f, ensure_ascii=False, indent=4)

        if re.search(r'https?://\S+', message.content):
            await read_text("URL")
        elif len(message.content) >= 60:
            await read_text(message.content[:30] + "…")
        else:
            await read_text(message.content)

    await bot.process_commands(message)

async def get_tts_audio(text):
    dictionary = load_dictionary()
    for key, value in dictionary.items():
        text = text.replace(key, value)

    filename = os.path.join(TTS_CACHE_DIR, f"{hash(text)}.wav")
    if os.path.exists(filename):
        return filename

    try:
        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{VOICEVOX_API_BASE_URL}/audio_query",
                params={"text": text, "speaker": SPEAKER_ID}
            ) as query_resp:

                if query_resp.status == 200:
                    audio_query = await query_resp.json()
                    async with session.post(
                        f"{VOICEVOX_API_BASE_URL}/synthesis",
                        params={"speaker": SPEAKER_ID},
                        json=audio_query
                    ) as synthesis_resp:

                        if synthesis_resp.status == 200:
                            with open(filename, "wb") as f:
                                f.write(await synthesis_resp.read())
                            return filename
    except Exception as e:
        print(f"TTS error: {e}")

    return None

async def read_text(text):
    channel = bot.get_channel(VOICE_CHANNEL_ID)
    if not channel:
        return

    if bot.voice_clients:
        vc = bot.voice_clients[0]
    else:
        vc = await channel.connect()

    audio_file = await get_tts_audio(text)
    if not audio_file:
        return

    try:
        vc.play(discord.FFmpegPCMAudio(audio_file, executable=FFMPEG_PATH))
        while vc.is_playing():
            await asyncio.sleep(0.5)
        await asyncio.sleep(1)
        if os.path.exists(audio_file):
            os.remove(audio_file)
    except Exception as e:
        print(f"Playback error: {e}")

async def announce_voice(text):
    await read_text(text)

@bot.command(name="add_word")
async def add_word(ctx, key: str, value: str):
    data = load_dictionary()
    data[key] = value
    save_dictionary(data)
    await ctx.send(f"Added: {key} → {value}")

@bot.command(name="remove_word")
async def remove_word(ctx, key: str):
    data = load_dictionary()
    if key in data:
        del data[key]
        save_dictionary(data)
        await ctx.send(f"Removed: {key}")
    else:
        await ctx.send("Not found.")

@bot.command(name="show_dictionary")
async def show_dictionary(ctx):
    data = load_dictionary()
    if data:
        content = "\n".join([f"{k}: {v}" for k, v in data.items()])
        await ctx.send(content)
    else:
        await ctx.send("Dictionary is empty.")

bot.run(TOKEN)