feat: A.B.I.N.A.S.H Discord bot with GIF emotes, slash commands, and music
- Core bot with slash command tree (discord.py 2.7.x compatible) - Commands: /ping, /info, /hello, /emoji, /app-emoji - Admin: /upload-emoji, /list-emojis, /delete-emoji - GIF emote detection and reaction handling - Message reactions with percentage-based triggers and cooldowns - Voice/music cog with YTDLSource, skip, volume, pause, resume, stop - Global error handler and voice state change logging - Environment config via .env (DISCORD_TOKEN, COMMAND_PREFIX, etc.)
This commit is contained in:
Benjamyn committed
commit
43ca943cb8
26 files changed
+1546
No files matched your search
Whitespace-only changes.
@@ -0,0 +1,124 @@
|
||||
"""GIF emote handler — parses emoji strings, fetches from CDN, sends reactions."""
|
||||
|
||||
import re
|
||||
import logging
|
||||
from typing import Optional
|
||||
|
||||
import discord
|
||||
from discord.ext import commands
|
||||
import aiohttp
|
||||
import discordemojiparser as edp
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ─── Configuration ───────────────────────────────────────────────────────────
|
||||
|
||||
GIF_EMOTE_CDN = "https://media.discordapp.net/gifs/"
|
||||
REACTION_CHANCE = 0.05 # 5% chance to trigger on GIF emote mention
|
||||
REACTION_COOLDOWN = 60 # seconds per channel
|
||||
|
||||
# ─── Emoji Parser ─────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def parse_emoji(emoji_str: str) -> Optional[discord.Emoji]:
|
||||
"""Parse an emoji string (e.g., :cat:) into a discord.Emoji object."""
|
||||
emoji_str = emoji_str.strip()
|
||||
if not emoji_str.startswith(":") or not emoji_str.endswith(":"):
|
||||
return None
|
||||
|
||||
emoji_name = emoji_str[1:-1]
|
||||
|
||||
try:
|
||||
emoji_obj = edp.parse(emoji_name)
|
||||
except Exception as e:
|
||||
logger.debug(f"Failed to parse emoji '{emoji_name}': {e}")
|
||||
return None
|
||||
|
||||
if not emoji_obj:
|
||||
return None
|
||||
|
||||
guild_id = emoji_obj.get("guild_id")
|
||||
emoji_id = emoji_obj.get("id")
|
||||
|
||||
if guild_id is None or emoji_id is None:
|
||||
return None
|
||||
|
||||
try:
|
||||
guild = bot.get_guild(guild_id)
|
||||
if guild is None:
|
||||
return None
|
||||
return guild.get_emoji(emoji_id)
|
||||
except (discord.NotFound, AttributeError) as e:
|
||||
logger.debug(f"Emoji not found: guild={guild_id}, emoji={emoji_id}")
|
||||
return None
|
||||
|
||||
|
||||
# ─── Reaction Handler ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class GIFReactionHandler:
|
||||
"""Handles GIF emote reactions on messages."""
|
||||
|
||||
def __init__(self, bot: commands.Bot) -> None:
|
||||
self.bot = bot
|
||||
self._cooldowns: dict[int, float] = {} # channel_id -> next_available
|
||||
|
||||
async def handle_reaction(self, message: discord.Message) -> None:
|
||||
"""Check if a message mentions a GIF emote and react accordingly."""
|
||||
channel_id = message.channel.id
|
||||
|
||||
# Enforce cooldown
|
||||
now = discord.utils.utcnow().timestamp()
|
||||
if channel_id in self._cooldowns:
|
||||
if now < self._cooldowns[channel_id]:
|
||||
return
|
||||
|
||||
# Check if message contains a GIF emote
|
||||
if await self._check_and_react(message):
|
||||
self._cooldown(channel_id)
|
||||
|
||||
async def _check_and_react(self, message: discord.Message) -> bool:
|
||||
"""Check if message mentions a GIF emote and add reaction."""
|
||||
# Look for emoji mentions in the message
|
||||
emoji_mentions = re.findall(r":(\w+):", message.content)
|
||||
|
||||
for name in emoji_mentions:
|
||||
emoji_obj = parse_emoji(f":{name}:")
|
||||
|
||||
if emoji_obj is None:
|
||||
continue
|
||||
|
||||
# Determine the GIF URL
|
||||
gif_url = f"{GIF_EMOTE_CDN}{emoji_obj.id}.gif"
|
||||
|
||||
# Add reaction
|
||||
try:
|
||||
await message.add_reaction(emoji_obj)
|
||||
return True
|
||||
except discord.Forbidden:
|
||||
pass
|
||||
|
||||
return False
|
||||
|
||||
def _cooldown(self, channel_id: int) -> None:
|
||||
"""Set cooldown for a channel."""
|
||||
self._cooldowns[channel_id] = discord.utils.utcnow().timestamp() + REACTION_COOLDOWN
|
||||
|
||||
|
||||
# ─── Bot Integration ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
async def setup(bot: commands.Bot) -> None:
|
||||
"""Register the GIF reaction handler with the bot."""
|
||||
handler = GIFReactionHandler(bot)
|
||||
|
||||
@bot.event
|
||||
async def on_message(message: discord.Message) -> None:
|
||||
"""Check for GIF emote mentions in messages."""
|
||||
if message.author == bot.user:
|
||||
return
|
||||
|
||||
if message.author.bot:
|
||||
return
|
||||
|
||||
await handler.handle_reaction(message)
|
||||
@@ -0,0 +1,109 @@
|
||||
"""Reaction handler with percentage-based triggers, cooldowns, and mention reactions."""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from typing import Optional
|
||||
|
||||
import discord
|
||||
from discord.ext import commands
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ─── Configuration ───────────────────────────────────────────────────────────
|
||||
|
||||
REACTION_CHANCE = 0.05 # 5% chance to react to a message
|
||||
REACTION_COOLDOWN = 60 # seconds per channel
|
||||
|
||||
# ─── Reaction Handler ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class ReactionHandler:
|
||||
"""Handles message reactions with cooldowns and mention triggers."""
|
||||
|
||||
def __init__(self, bot: commands.Bot) -> None:
|
||||
self.bot = bot
|
||||
self._cooldowns: dict[int, float] = {} # channel_id -> next_available
|
||||
self._reaction_emojis: list[discord.Emoji] = []
|
||||
|
||||
async def handle_reactions(self, message: discord.Message) -> None:
|
||||
"""Check if a message should be reacted to and handle reactions."""
|
||||
channel_id = message.channel.id
|
||||
|
||||
# Enforce cooldown
|
||||
now = discord.utils.utcnow().timestamp()
|
||||
if channel_id in self._cooldowns:
|
||||
if now < self._cooldowns[channel_id]:
|
||||
return
|
||||
|
||||
# Check if message mentions the bot
|
||||
if self._is_mention(message):
|
||||
await self._handle_mention(message)
|
||||
else:
|
||||
# Random chance to react
|
||||
if self._should_react():
|
||||
await self._add_reaction(message)
|
||||
|
||||
# Enforce cooldown
|
||||
self._cooldown(channel_id)
|
||||
|
||||
def _is_mention(self, message: discord.Message) -> bool:
|
||||
"""Check if the message mentions the bot."""
|
||||
return f"<@{self.bot.user.id}>" in message.content or f"<@!{self.bot.user.id}>" in message.content
|
||||
|
||||
def _should_react(self) -> bool:
|
||||
"""Determine if we should react (random chance)."""
|
||||
import random
|
||||
return random.random() < REACTION_CHANCE
|
||||
|
||||
async def _handle_mention(self, message: discord.Message) -> None:
|
||||
"""Handle a mention — react with a random emoji."""
|
||||
if self._reaction_emojis:
|
||||
emoji = self._reaction_emojis[0]
|
||||
try:
|
||||
await message.add_reaction(emoji)
|
||||
except discord.Forbidden:
|
||||
pass
|
||||
|
||||
async def _add_reaction(self, message: discord.Message) -> None:
|
||||
"""Add a reaction to a message."""
|
||||
if self._reaction_emojis:
|
||||
emoji = self._reaction_emojis[0]
|
||||
try:
|
||||
await message.add_reaction(emoji)
|
||||
except discord.Forbidden:
|
||||
pass
|
||||
|
||||
def _cooldown(self, channel_id: int) -> None:
|
||||
"""Set cooldown for a channel."""
|
||||
self._cooldowns[channel_id] = discord.utils.utcnow().timestamp() + REACTION_COOLDOWN
|
||||
|
||||
def register_emoji(self, emoji: discord.Emoji) -> None:
|
||||
"""Register an emoji for reactions."""
|
||||
self._reaction_emojis.append(emoji)
|
||||
|
||||
|
||||
# ─── Bot Integration ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
async def setup(bot: commands.Bot) -> None:
|
||||
"""Register the reaction handler with the bot."""
|
||||
handler = ReactionHandler(bot)
|
||||
|
||||
@bot.event
|
||||
async def on_message(message: discord.Message) -> None:
|
||||
"""Handle reactions on messages."""
|
||||
if message.author == bot.user:
|
||||
return
|
||||
|
||||
if message.author.bot:
|
||||
return
|
||||
|
||||
await handler.handle_reactions(message)
|
||||
|
||||
|
||||
def register_emoji(bot: commands.Bot, emoji: discord.Emoji) -> None:
|
||||
"""Register an emoji for reactions."""
|
||||
handler = ReactionHandler.__new__(ReactionHandler)
|
||||
handler.bot = bot
|
||||
handler._reaction_emojis = []
|
||||
bot._reaction_handler = handler
|
||||
@@ -0,0 +1,296 @@
|
||||
"""Voice channel music handler with YTDLSource, FFmpegOpusAudio, queue, and controls."""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from typing import Optional
|
||||
|
||||
import discord
|
||||
from discord.ext import commands
|
||||
from discord.ext.commands import Cog, Context, has_permissions
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ─── Configuration ───────────────────────────────────────────────────────────
|
||||
|
||||
PREFIX = "!"
|
||||
GUILD_IDS = [] # Set in bot.json or env var
|
||||
CHANNEL_ID = None # Set in bot.json or env var
|
||||
|
||||
# ─── Music Cog ───────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class MusicCog(Cog):
|
||||
"""Music commands for voice channels."""
|
||||
|
||||
def __init__(self, bot: commands.Bot) -> None:
|
||||
self.bot = bot
|
||||
self._current_queue: list[discord.AudioTrack] = []
|
||||
self._is_playing: bool = False
|
||||
self._voice_channel: Optional[discord.VoiceChannel] = None
|
||||
self._playing_lock: asyncio.Lock = asyncio.Lock()
|
||||
self._stop_requested: bool = False
|
||||
self._current_volume: int = 100
|
||||
self._is_paused: bool = False
|
||||
|
||||
def cog_unload(self) -> None:
|
||||
"""Clean up on cog unload."""
|
||||
if self._voice_channel:
|
||||
self._voice_channel.disconnect()
|
||||
|
||||
async def _ensure_voice(self, ctx: Context) -> None:
|
||||
"""Ensure the bot is in a voice channel."""
|
||||
if self._voice_channel is None:
|
||||
channel = await ctx.author.voice.channel.connect()
|
||||
self._voice_channel = channel
|
||||
self._is_playing = True
|
||||
elif not self._voice_channel.is_connected():
|
||||
self._voice_channel = await ctx.author.voice.channel.connect()
|
||||
|
||||
async def _disconnect(self) -> None:
|
||||
"""Disconnect from the voice channel."""
|
||||
if self._voice_channel:
|
||||
try:
|
||||
await self._voice_channel.disconnect()
|
||||
except discord.DiscordError:
|
||||
pass
|
||||
self._voice_channel = None
|
||||
self._is_playing = False
|
||||
self._stop_requested = False
|
||||
|
||||
async def play(self, ctx: Context, source: str) -> None:
|
||||
"""Play a track from a source (URL or file)."""
|
||||
if not self._is_playing:
|
||||
await self._ensure_voice(ctx)
|
||||
|
||||
try:
|
||||
track = await YTDLSource.from_url(source, source_type="youtube", loop=False)
|
||||
self._current_queue.append(track)
|
||||
await self._play_next()
|
||||
except Exception as e:
|
||||
logger.error(f"Error playing source {source}: {e}")
|
||||
await ctx.send(f"Error playing: {e}")
|
||||
|
||||
async def _play_next(self) -> None:
|
||||
"""Play the next track in the queue."""
|
||||
if not self._current_queue:
|
||||
await self._disconnect()
|
||||
return
|
||||
|
||||
track = self._current_queue.pop(0)
|
||||
|
||||
try:
|
||||
if self._voice_channel:
|
||||
await self._voice_channel.play(track, volume=self._current_volume / 100.0)
|
||||
self._is_playing = True
|
||||
except Exception as e:
|
||||
logger.error(f"Error playing track: {e}")
|
||||
await self._play_next()
|
||||
|
||||
async def skip(self, ctx: Context) -> None:
|
||||
"""Skip the current track."""
|
||||
if not self._is_playing:
|
||||
await ctx.send("Not playing anything.")
|
||||
return
|
||||
|
||||
try:
|
||||
await self._voice_channel.stop()
|
||||
self._current_queue.clear()
|
||||
self._is_playing = False
|
||||
await ctx.send("Skipped.")
|
||||
except Exception as e:
|
||||
logger.error(f"Error skipping: {e}")
|
||||
await ctx.send(f"Error: {e}")
|
||||
|
||||
async def volume(self, ctx: Context, *, volume: int = 100) -> None:
|
||||
"""Set the volume (0-100)."""
|
||||
if not self._is_playing:
|
||||
await ctx.send("Not playing anything.")
|
||||
return
|
||||
|
||||
if volume < 0 or volume > 100:
|
||||
await ctx.send("Volume must be between 0 and 100.")
|
||||
return
|
||||
|
||||
self._current_volume = volume
|
||||
await ctx.send(f"Volume set to {volume}%")
|
||||
|
||||
async def pause(self, ctx: Context) -> None:
|
||||
"""Pause playback."""
|
||||
if not self._is_playing:
|
||||
await ctx.send("Not playing anything.")
|
||||
return
|
||||
|
||||
self._is_paused = True
|
||||
await ctx.send("Playback paused.")
|
||||
|
||||
async def resume(self, ctx: Context) -> None:
|
||||
"""Resume playback."""
|
||||
if self._is_paused:
|
||||
self._is_paused = False
|
||||
await ctx.send("Playback resumed.")
|
||||
else:
|
||||
await ctx.send("Already playing.")
|
||||
|
||||
async def stop(self, ctx: Context) -> None:
|
||||
"""Stop and disconnect."""
|
||||
await self._disconnect()
|
||||
await ctx.send("Stopped and disconnected.")
|
||||
|
||||
async def queue(self, ctx: Context) -> None:
|
||||
"""Show the current queue."""
|
||||
if not self._current_queue:
|
||||
await ctx.send("Queue is empty.")
|
||||
return
|
||||
|
||||
tracks = []
|
||||
for i, track in enumerate(self._current_queue):
|
||||
tracks.append(f"{i+1}. {track.title} — {track.author}")
|
||||
await ctx.send("\n".join(tracks))
|
||||
|
||||
|
||||
# ─── YTDLSource ──────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class YTDLSource(discord.PCMVolumeTransformer):
|
||||
"""Audio source from youtube-dl."""
|
||||
|
||||
def __init__(self, data, *, is_live=False, loop=False):
|
||||
super().__init__()
|
||||
self.title = data["title"]
|
||||
self.url = data["url"]
|
||||
self.is_live = is_live
|
||||
self.duration = data["duration"] if "duration" in data else None
|
||||
self.author = data.get("uploader", "Unknown")
|
||||
self.loop = loop
|
||||
self._duration = data["duration"] if "duration" in data else None
|
||||
|
||||
@classmethod
|
||||
async def from_url(cls, url, source_type="youtube", loop=False):
|
||||
"""Create a source from a URL."""
|
||||
import youtube_dl
|
||||
|
||||
ydl_opts = {
|
||||
"format": "bestaudio/best",
|
||||
"noplaylist": True,
|
||||
"quiet": True,
|
||||
"no_warnings": True,
|
||||
}
|
||||
|
||||
with youtube_dl.YoutubeDL(ydl_opts) as ydl:
|
||||
info = ydl.extract_info(url, download=False)
|
||||
|
||||
if info is None:
|
||||
raise ValueError("No video info found.")
|
||||
|
||||
if info.get("url") is None:
|
||||
raise ValueError("No URL found in video info.")
|
||||
|
||||
if info.get("duration") is None:
|
||||
raise ValueError("No duration found in video info.")
|
||||
|
||||
data = {
|
||||
"url": info["url"],
|
||||
"title": info["title"],
|
||||
"duration": info["duration"],
|
||||
"uploader": info.get("uploader", "Unknown"),
|
||||
}
|
||||
|
||||
return cls(data, is_live=info.get("is_live", False), loop=loop)
|
||||
|
||||
@property
|
||||
def duration(self):
|
||||
"""Return the duration in seconds."""
|
||||
return self._duration
|
||||
|
||||
@duration.setter
|
||||
def duration(self, value):
|
||||
"""Set the duration."""
|
||||
self._duration = value
|
||||
|
||||
@property
|
||||
def is_live(self):
|
||||
"""Return whether this is a live stream."""
|
||||
return self._is_live
|
||||
|
||||
@is_live.setter
|
||||
def is_live(self, value):
|
||||
"""Set whether this is a live stream."""
|
||||
self._is_live = value
|
||||
|
||||
|
||||
# ─── FFmpegOpusAudio ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class FFmpegOpusAudio(discord.PCMVolumeTransformer):
|
||||
"""Audio source from FFmpegOpusAudio."""
|
||||
|
||||
def __init__(self, data, *, is_live=False, loop=False):
|
||||
super().__init__()
|
||||
self.title = data["title"]
|
||||
self.url = data["url"]
|
||||
self.is_live = is_live
|
||||
self.duration = data["duration"] if "duration" in data else None
|
||||
self.author = data.get("uploader", "Unknown")
|
||||
self.loop = loop
|
||||
self._duration = data["duration"] if "duration" in data else None
|
||||
|
||||
@classmethod
|
||||
async def from_url(cls, url, source_type="youtube", loop=False):
|
||||
"""Create a source from a URL using FFmpegOpusAudio."""
|
||||
import youtube_dl
|
||||
|
||||
ydl_opts = {
|
||||
"format": "bestaudio/best",
|
||||
"noplaylist": True,
|
||||
"quiet": True,
|
||||
"no_warnings": True,
|
||||
}
|
||||
|
||||
with youtube_dl.YoutubeDL(ydl_opts) as ydl:
|
||||
info = ydl.extract_info(url, download=False)
|
||||
|
||||
if info is None:
|
||||
raise ValueError("No video info found.")
|
||||
|
||||
if info.get("url") is None:
|
||||
raise ValueError("No URL found in video info.")
|
||||
|
||||
if info.get("duration") is None:
|
||||
raise ValueError("No duration found in video info.")
|
||||
|
||||
data = {
|
||||
"url": info["url"],
|
||||
"title": info["title"],
|
||||
"duration": info["duration"],
|
||||
"uploader": info.get("uploader", "Unknown"),
|
||||
}
|
||||
|
||||
return cls(data, is_live=info.get("is_live", False), loop=loop)
|
||||
|
||||
@property
|
||||
def duration(self):
|
||||
"""Return the duration in seconds."""
|
||||
return self._duration
|
||||
|
||||
@duration.setter
|
||||
def duration(self, value):
|
||||
"""Set the duration."""
|
||||
self._duration = value
|
||||
|
||||
@property
|
||||
def is_live(self):
|
||||
"""Return whether this is a live stream."""
|
||||
return self._is_live
|
||||
|
||||
@is_live.setter
|
||||
def is_live(self, value):
|
||||
"""Set whether this is a live stream."""
|
||||
self._is_live = value
|
||||
|
||||
|
||||
# ─── Bot Integration ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
async def setup(bot: commands.Bot) -> None:
|
||||
"""Register the music cog with the bot."""
|
||||
await bot.add_cog(MusicCog(bot))
|
||||
Reference in new issue
Block a user