Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,7 @@ A fully-featured Discord music bot built with Python. Supports YouTube and Sound
| `!8d` | Apply 8D audio (use headphones!) |
| `!karaoke` | Remove center vocals |
| `!reset` | Remove all audio effects |
| `!effect` | Show which effect is currently active |
| `!effects` | List all available effects |

### 🎤 Extras
Expand Down
149 changes: 51 additions & 98 deletions cogs/effects.py
Original file line number Diff line number Diff line change
@@ -1,138 +1,92 @@
import time
from dataclasses import dataclass

import discord
from discord.ext import commands

from utils.player import players
from utils.embeds import success_embed, error_embed
from utils.checks import same_voice_channel
from utils.embeds import BLURPLE, error_embed, success_embed
from utils.player import players

# Minimum seconds between effect changes per guild — each one restarts the
# FFmpeg process, so rapid toggling is throttled to protect CPU/memory.
_EFFECT_COOLDOWN = 3.0


# FFmpeg audio-filter presets. Applying one respawns the stream through
# ``-af <filter>``, resuming at the current playback position — see
# ``MusicPlayer.apply_effect``.
EFFECTS = {
"bass": "equalizer=f=54:width_type=o:width=2:g=5", # gentle low-end lift
"bassboost": "equalizer=f=54:width_type=o:width=2:g=10", # heavy low-end lift
@dataclass(frozen=True)
class Effect:
"""An FFmpeg filter chain and what the bot says when it is applied."""
filter: str
label: str


# The single source of truth for effects: the filter, the reply, and — via the
# command aliases built from these keys below — the command name itself. Adding
# an effect is one entry here and nothing else.
#
# Applying one respawns the stream through ``-af <filter>``, resuming at the
# current playback position — see ``MusicPlayer.apply_effect``.
EFFECTS: dict[str, Effect] = {
# gentle / heavy low-end lift
"bass": Effect("equalizer=f=54:width_type=o:width=2:g=5", "Bass boost applied 🔊"),
"bassboost": Effect("equalizer=f=54:width_type=o:width=2:g=10", "Heavy bass boost applied 💥"),
# The leading `aresample=48000` is load-bearing: `asetrate` *reinterprets* a
# stream's declared rate instead of scaling it, so without normalising first
# the speed factor becomes 48000*N/<source rate> and differs per track. A
# 44.1 kHz upload ran at 1.36x and a 22 kHz one at 2.72x, not 1.25x.
"nightcore": "aresample=48000,asetrate=48000*1.25,aresample=48000", # +pitch, +speed
"vaporwave": "aresample=48000,asetrate=48000*0.8,aresample=48000", # -pitch, -speed
"treble": "equalizer=f=8000:width_type=o:width=2:g=5", # high-end lift
"echo": "aecho=0.8:0.88:60:0.4", # short echo
"karaoke": "pan=stereo|c0=c0-c1|c1=c1-c0", # cancel centre vocals
"8d": "apulsator=hz=0.08", # rotating stereo
"nightcore": Effect("aresample=48000,asetrate=48000*1.25,aresample=48000",
"Nightcore effect applied 🌙✨"),
"vaporwave": Effect("aresample=48000,asetrate=48000*0.8,aresample=48000",
"Vaporwave effect applied 🌊🎶"),
"treble": Effect("equalizer=f=8000:width_type=o:width=2:g=5", "Treble boost applied 🎵"),
"echo": Effect("aecho=0.8:0.88:60:0.4", "Echo effect applied 🔔"),
"karaoke": Effect("pan=stereo|c0=c0-c1|c1=c1-c0", "Karaoke mode on 🎤"),
"8d": Effect("apulsator=hz=0.08", "8D audio applied 🎧 *Use headphones!*"),
}

_LABELS = {
"bass": "Bass boost applied 🔊",
"bassboost": "Heavy bass boost applied 💥",
"nightcore": "Nightcore effect applied 🌙✨",
"vaporwave": "Vaporwave effect applied 🌊🎶",
"treble": "Treble boost applied 🎵",
"echo": "Echo effect applied 🔔",
"8d": "8D audio applied 🎧 *Use headphones!*",
"karaoke": "Karaoke mode on 🎤",
}
# discord.py's CogMeta collects commands when the class body is executed, so a
# command cannot be registered per effect after the fact. One command carrying
# every effect name as an alias gets the same result from a single definition:
# `ctx.invoked_with` says which name the user actually typed.
_EFFECT_NAMES = list(EFFECTS)


class Effects(commands.Cog):
class Effects(commands.Cog, name="🎛️ Audio Effects"):
def __init__(self, bot: commands.Bot):
self.bot = bot
self._last_change: dict[int, float] = {} # guild_id → monotonic time

def _throttled(self, ctx) -> bool:
now = time.monotonic()
last = self._last_change.get(ctx.guild.id, 0.0)
if now - last < _EFFECT_COOLDOWN:
if now - self._last_change.get(ctx.guild.id, 0.0) < _EFFECT_COOLDOWN:
return True
self._last_change[ctx.guild.id] = now
return False

def _apply(self, ctx, name: str, filter_str: str) -> bool:
player = players.get(ctx.guild.id)
if not player:
return False
return player.apply_effect(name, filter_str)

async def _run(self, ctx, name: str):
async def _switch_to(self, ctx, name: str | None, filter_str: str, label: str) -> None:
"""Throttle, apply, and report — the whole path every effect command takes."""
if self._throttled(ctx):
return await ctx.send(embed=error_embed(
f"Easy — wait {_EFFECT_COOLDOWN:.0f}s between effect changes."
))
if self._apply(ctx, name, EFFECTS[name]):
await ctx.send(embed=success_embed(_LABELS[name]))
player = players.get(ctx.guild.id)
if player and player.apply_effect(name, filter_str):
await ctx.send(embed=success_embed(label))
else:
await ctx.send(embed=error_embed("Nothing is playing."))

# ── Effect commands ───────────────────────────────────────────────

@commands.command()
@same_voice_channel()
async def bass(self, ctx):
"""Add a light bass boost."""
await self._run(ctx, "bass")

@commands.command()
@same_voice_channel()
async def bassboost(self, ctx):
"""Add a heavy bass boost."""
await self._run(ctx, "bassboost")

@commands.command()
@same_voice_channel()
async def nightcore(self, ctx):
"""Apply nightcore (faster + higher pitch)."""
await self._run(ctx, "nightcore")

@commands.command()
@commands.command(name=_EFFECT_NAMES[0], aliases=_EFFECT_NAMES[1:],
help="Apply an audio effect. Use !effects to see them all.")
@same_voice_channel()
async def vaporwave(self, ctx):
"""Apply vaporwave (slower + lower pitch)."""
await self._run(ctx, "vaporwave")

@commands.command()
@same_voice_channel()
async def treble(self, ctx):
"""Boost treble frequencies."""
await self._run(ctx, "treble")

@commands.command()
@same_voice_channel()
async def echo(self, ctx):
"""Add an echo effect."""
await self._run(ctx, "echo")

@commands.command(name="8d")
@same_voice_channel()
async def eight_d(self, ctx):
"""Apply 8D audio (rotating stereo)."""
await self._run(ctx, "8d")

@commands.command()
@same_voice_channel()
async def karaoke(self, ctx):
"""Remove centre vocals."""
await self._run(ctx, "karaoke")
async def apply_effect(self, ctx):
effect = EFFECTS[ctx.invoked_with.lower()]
await self._switch_to(ctx, ctx.invoked_with.lower(), effect.filter, effect.label)

@commands.command(name="reset", aliases=["fxreset", "noeffect"])
@same_voice_channel()
async def reset_effect(self, ctx):
"""Remove all audio effects."""
if self._throttled(ctx):
return await ctx.send(embed=error_embed(
f"Easy — wait {_EFFECT_COOLDOWN:.0f}s between effect changes."
))
if self._apply(ctx, None, ""):
await ctx.send(embed=success_embed("Audio effects removed ✅"))
else:
await ctx.send(embed=error_embed("Nothing is playing."))
await self._switch_to(ctx, None, "", "Audio effects removed ✅")

@commands.command(name="effect")
async def current_effect(self, ctx):
Expand All @@ -144,13 +98,12 @@ async def current_effect(self, ctx):
@commands.command(name="effects")
async def list_effects(self, ctx):
"""List all available audio effects."""
names = ", ".join(f"`!{k}`" for k in EFFECTS)
embed = discord.Embed(
await ctx.send(embed=discord.Embed(
title="🎛️ Available Effects",
description=names + "\n\nUse `!reset` to remove all effects.",
color=0x5865F2,
)
await ctx.send(embed=embed)
description=", ".join(f"`!{name}`" for name in EFFECTS)
+ "\n\nUse `!reset` to remove all effects.",
color=BLURPLE,
))


async def setup(bot):
Expand Down
54 changes: 25 additions & 29 deletions cogs/lyrics.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,29 @@
from discord.ext import commands

from services import lyrics_api
from utils.embeds import error_embed, lyrics_embed
from utils.player import players
from utils.embeds import lyrics_embed, error_embed


class Lyrics(commands.Cog):
class Lyrics(commands.Cog, name="🎤 Lyrics"):
def __init__(self, bot):
self.bot = bot

@staticmethod
def _split_query(query: str) -> tuple[str, str]:
"""Split a ``title - artist`` query. A query without the separator is all title."""
if " - " in query:
title, artist = query.split(" - ", 1)
return title.strip(), artist.strip()
return query.strip(), ""

def _current_track(self, ctx) -> tuple[str, str]:
"""Title and artist of whatever is playing, or ``("", "")`` if nothing is."""
player = players.get(ctx.guild.id)
if not player or not player.current:
return "", ""
return player.current["title"], player.current.get("uploader", "")

@commands.command(aliases=["ly"])
async def lyrics(self, ctx, *, query: str = None):
"""
Expand All @@ -17,42 +33,22 @@ async def lyrics(self, ctx, *, query: str = None):
!lyrics <title> - <artist> → title + artist
"""
async with ctx.typing():
title, artist = "", ""

if query:
# Support "title - artist" format
if " - " in query:
parts = query.split(" - ", 1)
title, artist = parts[0].strip(), parts[1].strip()
else:
title = query.strip()
else:
# Fall back to the currently playing track.
player = players.get(ctx.guild.id)
if not player or not player.current:
return await ctx.send(embed=error_embed(
"Nothing is playing. Provide a song name: `!lyrics <title>`"
))
current = player.current
# Spotify titles are "Song — Artist"; split for a cleaner search.
title = current["title"].split(" — ")[0].strip()
artist = current.get("uploader", "")
title, artist = (self._split_query(query) if query
else self._current_track(ctx))
if not title:
return await ctx.send(embed=error_embed(
"Nothing is playing. Provide a song name: `!lyrics <title>`"
))

result = await lyrics_api.fetch(title, artist, loop=self.bot.loop)

if not result:
return await ctx.send(embed=error_embed(
f"Couldn't find lyrics for **{title}**."
))

embeds = lyrics_embed(result["title"], result["artist"], result["lyrics"])
for embed in embeds:
for embed in lyrics_embed(result["title"], result["artist"], result["lyrics"]):
await ctx.send(embed=embed)

@lyrics.error
async def lyrics_error(self, ctx, error):
await ctx.send(embed=error_embed(str(error)))


async def setup(bot):
await bot.add_cog(Lyrics(bot))
20 changes: 5 additions & 15 deletions cogs/music.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,8 @@

from services import media
from utils.player import players, MusicPlayer, MAX_QUEUE
from utils.embeds import queue_embed, now_playing_embed, error_embed, success_embed
from utils.embeds import (added_embed, error_embed, now_playing_embed,
queue_embed, success_embed)
from utils.checks import user_in_voice, same_voice_channel

log = logging.getLogger("loopify.music")
Expand All @@ -25,7 +26,7 @@ def _is_playlist_url(query: str) -> bool:
return "/sets/" in q or "/album/" in q # SoundCloud set / Bandcamp album


class Music(commands.Cog):
class Music(commands.Cog, name="🎵 Music & Queue"):
def __init__(self, bot: commands.Bot):
self.bot = bot

Expand Down Expand Up @@ -55,7 +56,7 @@ async def _ensure_voice(self, ctx) -> bool:
return False

def _player(self, ctx) -> MusicPlayer:
return players.get_or_create(self.bot, ctx)
return players.get_or_create(self.bot, ctx.guild, ctx.channel)

# ── Playback commands ─────────────────────────────────────────────

Expand Down Expand Up @@ -98,7 +99,7 @@ async def _enqueue(self, ctx, player: MusicPlayer, tracks: list[dict], batch_lab
f"Queue is full (max {MAX_QUEUE} tracks)."
))
if not was_idle:
await ctx.send(embed=self._added_embed(tracks[0]))
await ctx.send(embed=added_embed(tracks[0]))
else:
added = player.add_many(tracks)
if added == 0:
Expand All @@ -109,16 +110,6 @@ async def _enqueue(self, ctx, player: MusicPlayer, tracks: list[dict], batch_lab
await ctx.send(embed=success_embed(
f"Added **{added} tracks** from {batch_label} to the queue.{skipped}"
))
# When idle, the player loop picks up the newly-added track automatically.

@staticmethod
def _added_embed(track: dict) -> discord.Embed:
url = track.get("url") or track.get("spotify_url")
title = f"[{track['title']}]({url})" if url else track["title"]
embed = discord.Embed(description=f"➕ Added to queue: **{title}**", color=0x5865F2)
if track.get("thumbnail"):
embed.set_thumbnail(url=track["thumbnail"])
return embed

@commands.command()
@same_voice_channel()
Expand Down Expand Up @@ -280,7 +271,6 @@ async def on_voice_state_update(self, member, before, after):
vc = member.guild.voice_client
if not vc:
return
# Only react to people leaving the bot's own channel.
if before.channel != vc.channel:
return
if len([m for m in vc.channel.members if not m.bot]) == 0:
Expand Down
12 changes: 3 additions & 9 deletions config.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,18 +20,12 @@
PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
load_dotenv(os.path.join(PROJECT_ROOT, ".env"))

# ── Bot ───────────────────────────────────────────────────────────────
DISCORD_TOKEN = os.getenv("DISCORD_TOKEN")
COMMAND_PREFIX = os.getenv("COMMAND_PREFIX", "!")
GENIUS_TOKEN = os.getenv("GENIUS_TOKEN")
COOKIES_PATH = os.getenv("COOKIES_PATH")
LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO").upper()

# ── Genius (Lyrics) ───────────────────────────────────────────────────
GENIUS_TOKEN = os.getenv("GENIUS_TOKEN")

# ── Misc ──────────────────────────────────────────────────────────────
COOKIES_PATH = os.getenv("COOKIES_PATH")
LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO").upper()

# ── Cogs to load ──────────────────────────────────────────────────────
COGS = [
"cogs.music",
"cogs.effects",
Expand Down
Loading
Loading