70 lines
2.3 KiB
Python
70 lines
2.3 KiB
Python
import os
|
|
import random
|
|
import string
|
|
from typing import Callable, AsyncGenerator, AsyncIterable, Optional, Any
|
|
|
|
import discord
|
|
import gtts
|
|
|
|
|
|
async def async_filter(fun: Callable, iterable: AsyncIterable) -> AsyncGenerator:
|
|
async for val in iterable:
|
|
if fun(val):
|
|
yield val
|
|
|
|
|
|
def find_category(guild: discord.Guild, group: str) -> Optional[discord.CategoryChannel]:
|
|
group = group.lower()
|
|
for cat in guild.categories:
|
|
if cat.name.lower() == group:
|
|
return cat
|
|
return None
|
|
|
|
|
|
async def find_role_case_insensitive(guild: discord.Guild, name: str, prefix: str = '') -> Optional[discord.Role]:
|
|
for role in await guild.fetch_roles():
|
|
if name.strip().lower() == role.name[len(prefix):].lower():
|
|
return role
|
|
return None
|
|
|
|
|
|
def link_channel(channel: discord.abc.GuildChannel, italic: bool = False) -> str:
|
|
if italic:
|
|
return '[*' + channel.name + '*](https://discord.com/channels/' + str(channel.guild.id) + '/' + str(
|
|
channel.id) + ')'
|
|
return '[' + channel.name + '](https://discord.com/channels/' + str(channel.guild.id) + '/' + str(channel.id) + ')'
|
|
|
|
|
|
def text_to_speech(text: str, lang: str = "de") -> (discord.AudioSource, Callable[[], None]):
|
|
tts = gtts.gTTS(text, lang=lang)
|
|
os.makedirs('temp', exist_ok=True)
|
|
file_name = 'temp/' + ''.join(random.choice(string.ascii_lowercase) for i in range(12)) + '.mp3'
|
|
tts.save(file_name)
|
|
source = discord.PCMVolumeTransformer(
|
|
discord.FFmpegPCMAudio(source=file_name, before_options='-v quiet', options='-filter:a "atempo=1.5"')
|
|
)
|
|
|
|
def destroy():
|
|
os.remove(file_name)
|
|
|
|
return source, destroy
|
|
|
|
|
|
async def connect_and_play(
|
|
channel: discord.VoiceChannel, source: discord.AudioSource,
|
|
after_play: Optional[Callable[[discord.DiscordException, discord.VoiceProtocol], Any]] = None
|
|
) -> Optional[discord.VoiceClient]:
|
|
# noinspection PyTypeChecker
|
|
client: discord.VoiceClient = await channel.connect()
|
|
after_callback: Optional[Callable[[discord.DiscordException], Any]]
|
|
if after_play is None:
|
|
after_callback = None
|
|
else:
|
|
def callback(exc: discord.DiscordException) -> Any:
|
|
after_play(exc, client)
|
|
|
|
after_callback = callback
|
|
|
|
client.play(source, after=after_callback)
|
|
return client
|