囁き声のように優しい読み上げbot
import os, sys, platform, subprocess
from util import download


def sdk_url(version: str, os: str, arch: str | None, device: str = "cpu"):
    if arch is not None:
        arch = f"_{arch}"

    return f"https://github.com/VOICEVOX/voicevox_core/releases/download/{version}/voicevox_core-{version}+{device}-cp38-abi3-{os}{arch}.whl"


voicevox_version = (
    os.environ["VOICEVOX_VERSION"] if "VOICEVOX_VERSION" in os.environ else "0.14.4"
)

device_type = sys.argv[1] if len(sys.argv) >= 2 else "cpu"
if not device_type in ("cpu", "cuda", "directml"):
    raise Exception(f"Invalid device type {device_type}.")

os_name = platform.system()
match os_name:
    case "Windows":
        os_name = "win"

    case "Linux":
        os_name = "linux"

    case "Darwin":
        os_name = "macosx"

arch = platform.machine().lower()

print(
    "Download voicevox core binding.\n\n"
    + f"version: {voicevox_version}\n"
    + f"os_name: {os_name}\n"
    + f"device_type: {device_type}\n",
    file=sys.stderr,
)

# 得た情報からライブラリをダウンロードする
match (device_type, os_name, arch):
    # GPU
    case ("cuda", "linux", "amd64" | "x86_64"):
        filename = download(sdk_url(voicevox_version, os_name, "x86_64", device_type))

    case ("directml" | "cuda", "win", "amd64" | "x86_64"):
        filename = download(sdk_url(voicevox_version, os_name, "amd64", device_type))

    # CPU
    case ("cpu", "win", "amd64" | "x86_64"):
        filename = download(sdk_url(voicevox_version, os_name, "amd64", device_type))

    case ("cpu", "win", "x86" | "i386"):
        filename = download(sdk_url(voicevox_version, "win32", None, device_type))

    case ("cpu", "macosx", "amd64" | "x86_64"):
        filename = download(
            sdk_url(voicevox_version, f"{os_name}_10_7", "amd64", device_type),
        )

    case ("cpu", "macosx", "aarch64" | "arm64"):
        filename = download(
            sdk_url(voicevox_version, f"{os_name}_11_0", "arm64", device_type),
        )

    case ("cpu", "linux", "amd64" | "x86_64"):
        filename = download(sdk_url(voicevox_version, os_name, "x86_64", device_type))

    case ("cpu", "linux", "aarch64" | "arm64"):
        filename = download(sdk_url(voicevox_version, os_name, "aarch64", device_type))

    case _ as t:
        raise Exception(f"Invalid library version {t}.")

subprocess.run(["python", "-m", "pip", "install", filename])

scripts/install-bind.py
murmur-develop/murmur
import requests, urllib.parse, os.path


def download(url: str):
    res = requests.get(url)
    if not (res.status_code >= 200 and res.status_code < 300):
        raise Exception(
            f"Failed to download file from url.\nstatus_code: {res.status_code}.\nurl: {url}"
        )

    filename = os.path.basename(urllib.parse.urlparse(url).path)

    with open(filename, "w+b") as f:
        f.write(res.content)

    return filename

scripts/util.py
murmur-develop/murmur
import discord
from discord.ext import commands
from preprocessor import preprocess_text
from voicevox import Voicevox
from textToSpeechQueue import Text2SpeechQueue, SpeakTask

counter = 0

description = """
voicevoxの読み上げbot
"""

intents = discord.Intents.default()
intents.message_content = True

bot = commands.Bot(
    command_prefix=commands.when_mentioned_or("$"),
    description=description,
    intents=intents,
)

voicevox = Voicevox()
voice_queue = Text2SpeechQueue(bot, voicevox)


@bot.event
async def on_ready():
    print(f"Logged in as {bot.user} (ID: {bot.user.id})")
    print("------")
    await bot.tree.sync()
    voice_queue.run()


@bot.listen()
async def on_message(message: discord.Message):
    if (
        type(message.channel) is discord.channel.VoiceChannel
        and bot.user in message.channel.members
        and message.author != bot.user
        and not message.content.startswith("$")
        and message.guild
        and type(message.guild.voice_client) is discord.VoiceClient
        and message.guild.voice_client.is_connected()
    ):
        buf = await voicevox.genarete_sound(
            text=preprocess_text(message.content), guild=message.guild
        )
        if buf is None:
            await message.reply("audio failed")
            return

        src = discord.FFmpegOpusAudio(source=buf, pipe=True)
        voice_queue.add(SpeakTask(message=message, source=src))

src/bot.py
murmur-develop/murmur
import discord
from discord import app_commands
from discord.ext import commands


class ReadingAloud(commands.Cog):
    def __init__(self, bot: commands.Bot):
        self.bot = bot

    @app_commands.command(name="join", description="botをボイスチャンネルに参加させます")
    @app_commands.describe(channel="参加させるボイスチャンネル(何も入力しない場合は現在のボイスチャンネルに参加します)")
    async def join(
        self,
        interaction: discord.Interaction,
        *,
        channel: discord.VoiceChannel | None = None,
    ):
        """join voice channel"""
        target_channel = channel
        if target_channel is None:
            if type(interaction.channel) is not discord.VoiceChannel:
                embed = discord.Embed(
                    title="Error",
                    color=discord.Color.brand_red(),
                    description=(
                        """Not participating in voice channel.
Please join the voice channel or specify a valid channel as an argument."""
                    ),
                )
                embed.set_author(name=self.bot.user)
                await interaction.response.send_message(embed=embed)
                return

            target_channel = interaction.channel
        if (
            interaction.guild
            and type(interaction.guild.voice_client) is discord.VoiceClient
        ):
            return await interaction.guild.voice_client.move_to(target_channel)

        await target_channel.connect()
        embed = discord.Embed(
            title="Connect",
            color=discord.Color.brand_green(),
            description=f"{self.bot.user}が接続しました",
        )
        embed.set_author(name=self.bot.user)
        await interaction.response.send_message(embed=embed)

    @app_commands.command(name="bye", description="botをボイスチャンネルから退出させます")
    async def bye(self, interaction: discord.Interaction):
        """leave voice channel"""
        if (
            interaction.guild
            and type(interaction.guild.voice_client) is not discord.VoiceClient
        ):
            embed = discord.Embed(
                title="Error",
                color=discord.Color.brand_red(),
                description="Already left the audio channel.",
            )
            embed.set_author(name=self.bot.user)
            await interaction.response.send_message(embed=embed)
            return
        if interaction.guild and interaction.guild.voice_client:
            await interaction.guild.voice_client.disconnect(force=True)
        embed = discord.Embed(
            title="Disconnect",
            color=discord.Color.greyple(),
            description=f"{self.bot.user}が退出しました",
        )
        embed.set_author(name=self.bot.user)
        await interaction.response.send_message(embed=embed)

src/commands.py
murmur-develop/murmur
from dotenv import load_dotenv
import os
import discord
from commands import ReadingAloud
import asyncio
import logging
from bot import bot

discord.utils.setup_logging(level=logging.INFO, root=False)

token = os.environ.get("BOT_TOKEN", "DISCORD_BOT_TOKEN")


async def main():
    async with bot:
        await bot.add_cog(ReadingAloud(bot))
        await bot.start(token=token)


asyncio.run(main())

src/main.py
murmur-develop/murmur
import re, json
from os import path
from functools import reduce


def load_romanization_table(paths: list[str]):
    tables: dict[str, str] = {}
    for path in paths:
        with open(path) as f:
            for k, v in json.load(f).items():
                tables[k] = v

    tables = dict(sorted(tables.items(), key=lambda x: -len(x[0])))

    return tables


romanization_table = load_romanization_table(
    [
        path.join(path.dirname(__file__), "romanization/hepburn.json"),
        path.join(path.dirname(__file__), "romanization/japanese.json"),
    ]
)

# duplicated consonant adds "ッ" at the beginning
dup_consonant_table: dict[str, str] = {}
for k, v in romanization_table.items():
    if len(k) > 1:
        dup_consonant_table[k[0] + k] = f"っ{v}"


def preprocess_url(text: str):
    return re.sub(
        r"https?://(([a-zA-Z0-9]-)*[a-zA-Z0-9]+.)*[a-zA-Z0-9](-[a-zA-Z0-9]+)*(/[a-zA-Z0-9-.]+)*(\?[a-zA-Z0-9%=-]*)?/?",
        "URL省略",
        text,
    )


def preprocess_emoji(text: str):
    return re.sub(r"<a?:([_a-zA-Z0-9-]+):[0-9]+>", r"\1", text)


def preprocess_dup_consonant_alphabet(text: str):
    dup_consonant_table_key_regexp = "|".join(
        map(re.escape, dup_consonant_table.keys())
    )
    return re.sub(
        dup_consonant_table_key_regexp, lambda k: dup_consonant_table[k.group()], text
    )


def preprocess_alphabet(text: str):
    romanization_table_key_regexp = "|".join(map(re.escape, romanization_table.keys()))
    return re.sub(
        romanization_table_key_regexp, lambda k: romanization_table[k.group()], text
    )


def preprocess_text(text: str):
    processors = [
        preprocess_url,
        preprocess_emoji,
        preprocess_dup_consonant_alphabet,
        preprocess_alphabet,
    ]

    return reduce(lambda p, c: c(p), processors, text)

src/preprocessor.py
murmur-develop/murmur
import discord
from discord.ext import commands
from typing import cast, Coroutine
from voicevox import Voicevox
import asyncio


class SpeakTask:
    message: discord.Message
    source: discord.FFmpegOpusAudio

    def __init__(self, message: discord.Message, source: discord.FFmpegOpusAudio):
        self.message = message
        self.source = source

    @property
    def channel(self):
        return cast(discord.VoiceChannel, self.message.channel)

    @property
    def guild(self):
        return self.message.guild


class Text2SpeechQueue:
    bot: commands.Bot
    voicevox: Voicevox

    queue: list[SpeakTask] = []

    _tasks: set[asyncio.Task] = set()

    def __init__(self, bot: commands.Bot, voicevox: Voicevox):
        self.bot = bot
        self.voicevox = voicevox

    def add(self, task: SpeakTask):
        self.queue.append(task)

    def shift(self) -> SpeakTask | None:
        if len(self.queue) <= 0:
            return None
        return self.queue.pop(0)

    def create_task(self, c: Coroutine):
        task = self.bot.loop.create_task(c)
        self._tasks.add(task)

        task.add_done_callback(self._tasks.discard)

    def run(self):
        def after_fn(prev_result: Exception | None = None):
            if prev_result is not None:
                raise prev_result

            self.create_task(next_fn())

        async def next_fn():
            await asyncio.sleep(0.05)

            task = self.shift()
            if task is not None:
                if task.source and type(task.guild.voice_client) is discord.VoiceClient:
                    task.guild.voice_client.play(task.source, after=after_fn)
                else:
                    await task.message.reply("audio failed")
                    after_fn(None)
            else:
                await asyncio.sleep(0.1)
                after_fn(None)

        self.create_task(next_fn())

src/textToSpeechQueue.py
murmur-develop/murmur
from io import BytesIO
from dataclasses import dataclass
import re
from discord.message import Guild
import os
from os import path
import asyncio
from concurrent.futures import ThreadPoolExecutor

voicevox_path = (
    os.environ["VOICEVOX_PATH"]
    if "VOICEVOX_PATH" in os.environ
    else path.join(path.dirname(__file__), "../voicevox_core/")
)

from voicevox_core import VoicevoxCore


class Voicevox:
    core: VoicevoxCore
    thread_pool: ThreadPoolExecutor

    def __init__(
        self,
        dict_dir=path.join(voicevox_path, "open_jtalk_dic_utf_8-1.11"),
        thread_pool=ThreadPoolExecutor(max_workers=8),
    ) -> None:
        self.core = VoicevoxCore(open_jtalk_dict_dir=dict_dir, load_all_models=True)
        self.thread_pool = thread_pool

    async def genarete_sound(
        self, text: str, guild: Guild, speaker=1, speed=100, pitch=0
    ) -> BytesIO | None:
        # Mention先のUserID取得
        mention_id = re.search("[0-9]+", text)
        if mention_id is not None:
            member = guild.get_member(int(mention_id.group()))
            if member is not None:
                user_name = member.display_name
                # @{user_id} を @{display_name}さん に置き換える
                text = text.replace(mention_id.group(), user_name + "さん")

        query = self.core.audio_query(text, speaker)
        query.speed_scale = speed / 100
        query.pitch_scale = pitch / 100

        if len(text) < 25:
            return BytesIO(self.core.synthesis(query, speaker))
        else:

            def _fn():
                return BytesIO(self.core.synthesis(query, speaker))

            try:
                return await asyncio.wait_for(
                    asyncio.wrap_future(self.thread_pool.submit(_fn)), timeout=120
                )
            except:
                return None

src/voicevox.py
murmur-develop/murmur