Siphalor/och-bot
1
0
Fork 0
Code Releases Activity

Implement och command to timeout anyone

Browse Source
This commit is contained in:
Siphalor
2021-01-18 22:51:00 +01:00
parent 4cb20aa771
commit 103e1fa354
3 changed files with 84 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
bot.py
View File

@@ -6,6 +6,7 @@ import time
from typing import Optional
import discord
import gtts
from discord.ext import commands
from dotenv import load_dotenv
@@ -18,6 +19,7 @@ TOKEN = os.getenv('DISCORD_TOKEN')
PING_REGEX = re.compile(r'\b(?:ge)?ping', re.IGNORECASE)
TOPFIT_REGEX = re.compile(r'\b(topfit|fit|top|micro|microsoft|virtual reality|vr|ä+h*m*|hä*)\b', re.IGNORECASE)
LOEH_REGEX = re.compile(r'Och L(?:oe|ö)h!?', re.IGNORECASE)
OCH_ANYONE_REGEX = re.compile(r'^Och\s*(\w.*?)\s*[,.!?:]*$', re.IGNORECASE)
TOPFIT_WORDS = (
"äh",
@@ -42,7 +44,7 @@ OWNER_ID = 327126546970312739
OCH_LOEH_SOUND = "assets/och_loeh.mp3"
last_loeh = 0
last_och = 0
config_load()
bot = commands.Bot(command_prefix=config.get('prefix'))
@@ -65,19 +67,17 @@ async def on_ready():
@bot.event
async def on_message(message: discord.Message):
global last_loeh
global last_och
if message.guild is not None and not message.author.bot:
if config_get('loeh-enable', message.guild.id) and LOEH_REGEX.match(message.content):
if message.author.id == LOEH_ID:
await message.channel.send("https://siphalor.de/img/spidy-is-that-you.jpg")
else:
t = time.time()
if t - config_get('loeh-cooldown', message.guild.id) <= last_loeh:
await message.channel.send('Don\'t try this to often, **' + message.author.nick +
'**. That might backfire.')
last_loeh = t
if t - config_get('och-cooldown', message.guild.id) <= last_och:
await message.channel.send('429: Too many requests')
return
last_loeh = t
last_och = t
loeh = await get_loeh(message.guild)
if loeh is None:
await message.channel.send('404: Löh not found!')
@@ -92,7 +92,7 @@ async def on_message(message: discord.Message):
source = discord.PCMVolumeTransformer(discord.FFmpegPCMAudio(source=OCH_LOEH_SOUND))
voice_protocol.play(source)
await loeh.edit(mute=True)
sleeper = asyncio.sleep(config_get('loeh-timeout', message.guild.id))
sleeper = asyncio.sleep(config_get('och-timeout', message.guild.id))
message: Optional[discord.Message] = await message.channel.send('Zu Befehl!')
await sleeper
@@ -104,6 +104,63 @@ async def on_message(message: discord.Message):
except (asyncio.TimeoutError, discord.Forbidden, discord.HTTPException, discord.ClientException):
await message.channel.send('Failed to complete your command, Sir')
return
elif config_get('och-anyone-enable', message.guild.id) and (
match := OCH_ANYONE_REGEX.search(message.content)) is not None:
t = time.time()
if t - config_get('och-cooldown', message.guild.id) <= last_och:
await message.channel.send('429: Too many requests')
return
last_och = t
guild: discord.Guild = message.guild
bot_member: discord.Member = guild.get_member(bot.user.id)
search: str = match.group(1).lower()
matches: list = []
voice: Optional[discord.VoiceChannel] = None
async for member in guild.fetch_members(limit=100):
if not member.bot and member.voice is not None and search in member.display_name.lower().split(' '):
perms: discord.Permissions = member.voice.channel.permissions_for(bot_member)
if perms.connect and perms.speak:
voice = member.voice.channel
matches.append(member)
if matches and voice is not None:
voice_protocol: discord.VoiceProtocol = await voice.connect()
if type(voice_protocol) is discord.VoiceClient:
tts = gtts.gTTS(message.content, lang='de')
os.makedirs('temp', exist_ok=True)
tts.save('temp/och.mp3')
source = discord.PCMVolumeTransformer(
discord.FFmpegPCMAudio(source='temp/och.mp3', before_options='-v quiet')
)
voice_protocol.play(source)
async def _mute(m: discord.Member):
await m.edit(mute=True)
async def _unmute(m: discord.Member):
await m.edit(mute=False)
await asyncio.gather(*map(_mute, matches))
sleeper = asyncio.sleep(config_get('och-timeout', message.guild.id))
message: Optional[discord.Message] = await message.channel.send('Auf gehts!')
await sleeper
await asyncio.gather(*map(_unmute, matches))
waiter = None
if message is not None:
waiter = message.edit(content='~~Auf gehts!~~\nGeschafft!')
if type(voice_protocol) is discord.VoiceClient:
await voice_protocol.disconnect()
if waiter is not None:
await waiter
os.remove('temp/och.mp3')
else:
await message.channel.send('404: No users found!')
elif config_get('inf19x-insiders-enable', message.guild.id):
if PING_REGEX.search(message.content):
embed = discord.Embed(
@@ -125,7 +182,7 @@ async def on_message(message: discord.Message):
text = ''
for i in range(0, random.randint(7, 13)):
text += random.choice(TOPFIT_WORDS) + " "
text += match.group()
text += match.group(1)
await message.channel.send(text)
await bot.process_commands(message)
@@ -136,7 +193,8 @@ async def random_message_command(ctx: commands.Context, channel: Optional[discor
typing = ctx.channel.trigger_typing()
if channel is None:
channel = ctx.channel
messages = async_filter(lambda m: _is_message_valid_for_selection(m, reaction_filter), channel.history(limit=max_cnt))
messages = async_filter(lambda m: _is_message_valid_for_selection(m, reaction_filter),
channel.history(limit=max_cnt))
messages = [item async for item in messages]
if not messages:
await typing
@@ -159,7 +217,8 @@ async def collect_messages_command(ctx: commands.Context, channel: Optional[disc
typing = ctx.channel.trigger_typing()
if channel is None:
channel = ctx.channel
messages = async_filter(lambda m: _is_message_valid_for_selection(m, reaction_filter), channel.history(limit=max_cnt))
messages = async_filter(lambda m: _is_message_valid_for_selection(m, reaction_filter),
channel.history(limit=max_cnt))
messages = [item async for item in messages]
embed = discord.Embed(
title="List of matching messages",