advtech/Amber
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: advtech:Dev-user-preference
Branches Tags
advtech:main
advtech:dev-tts
advtech:Dev-user-preference
advtech:Dev
..
compare: advtech:main
Branches Tags
advtech:dev-tts
advtech:Dev-user-preference
advtech:Dev
advtech:main

No commits in common. "Dev-user-preference" and "main" have entirely different histories.
Dev-user-p ... main

4 changed files with 0 additions and 408 deletions
Show Stats Expand all files Collapse all files

1
.gitignore vendored
View File

@ -169,4 +169,3 @@ cython_debug/
# PyPI configuration file
.pypirc
/user_data.json

283
audio.py
View File

@ -1,283 +0,0 @@
import discord
import yt_dlp
import asyncio
import os
import json
import random
import requests
from dotenv import load_dotenv
load_dotenv()
voice_clients = {} # Track active voice connections
music_queues = {} # Per-guild song queue
current_tracks = {} # Currently playing tracks
volumes = {} # Volume levels per guild
repeat_modes = {} # Tracks repeat mode per guild: "one", "all", or "off"
default_volume = 0.05 # Default volume (5%)
USER_DATA_FILE = 'user_data.json'
YOUTUBE_API_KEY = os.getenv("YOUTUBE_API_KEY") # Replace with your API key
# ---------- Utility Functions ----------
def load_user_data():
if not os.path.exists(USER_DATA_FILE):
return {}
with open(USER_DATA_FILE, 'r') as f:
return json.load(f)
def save_user_data(data):
with open(USER_DATA_FILE, 'w') as f:
json.dump(data, f, indent=4)
def update_user_history(user_id, song_title, artist):
user_data = load_user_data()
if str(user_id) not in user_data:
user_data[str(user_id)] = {"history": []}
# Add the song to the user's history
user_data[str(user_id)]["history"].append({"title": song_title, "artist": artist})
save_user_data(user_data)
# ---------- Playback Functions ----------
async def join_voice(interaction: discord.Interaction):
if interaction.user.voice is None or interaction.user.voice.channel is None:
await interaction.response.send_message("You need to be in a voice channel for me to join.")
return None
channel = interaction.user.voice.channel
guild_id = interaction.guild.id
if guild_id not in voice_clients:
voice_clients[guild_id] = await channel.connect()
# Ensure the response is completed
if interaction.response.is_done():
await interaction.followup.send(f"✅ **Joined {channel.name}.**")
else:
await interaction.response.send_message(f"✅ **Joined {channel.name}.**")
return voice_clients[guild_id]
async def play_audio(interaction: discord.Interaction, query: str):
guild_id = interaction.guild.id
# Ensure Amber is connected to a voice channel
if guild_id not in voice_clients or not voice_clients[guild_id].is_connected():
await interaction.response.defer() # Defer the response early
voice_client = await join_voice(interaction)
if voice_client is None: # If join failed
return
else:
voice_client = voice_clients[guild_id]
# Defer response if not already done
if not interaction.response.is_done():
await interaction.response.defer()
# Search for the song on YouTube
ydl_opts = {
'format': 'bestaudio/best',
'quiet': True,
}
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
try:
info = ydl.extract_info(f"ytsearch:{query}", download=False)['entries'][0]
except Exception as e:
await interaction.followup.send(f"Failed to find or play the requested audio. Error: {str(e)}")
return
song_url = info['url']
title = info.get('title', 'Unknown Title')
artist = info.get('uploader', 'Unknown Artist')
# Log the song in the user's history
update_user_history(interaction.user.id, title, artist)
# Add the song to the queue
if guild_id not in music_queues:
music_queues[guild_id] = []
music_queues[guild_id].append((song_url, title))
await interaction.followup.send(f"✅ **Added to queue:** {title}")
# If nothing is playing, start playback
if not voice_client.is_playing():
await play_next(interaction)
async def play_next(interaction: discord.Interaction):
guild_id = interaction.guild.id
if guild_id not in music_queues:
music_queues[guild_id] = []
# Handle Repeat One
if repeat_modes.get(guild_id) == "one" and current_tracks.get(guild_id):
song_url, title = current_tracks[guild_id]
music_queues[guild_id].insert(0, (song_url, title)) # Re-add the current song to the front of the queue
# Handle Repeat All
elif repeat_modes.get(guild_id) == "all" and not music_queues[guild_id]:
# Move completed songs back to the queue
song_url, title = current_tracks.get(guild_id, (None, None))
if song_url and title:
music_queues[guild_id].append((song_url, title))
music_queues[guild_id].extend(current_tracks.values())
# Proceed to play the next song
if not music_queues[guild_id]:
await interaction.followup.send("❌ **No more songs in the queue.**")
return
voice_client = voice_clients[guild_id]
song_url, title = music_queues[guild_id].pop(0)
current_tracks[guild_id] = (song_url, title)
# Prepare FFmpeg options
ffmpeg_options = {
'before_options': '-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5',
'options': '-vn',
}
try:
source = discord.FFmpegPCMAudio(song_url, **ffmpeg_options)
volume = volumes.get(guild_id, default_volume)
source = discord.PCMVolumeTransformer(source, volume=volume)
# Play the audio
voice_client.play(
source,
after=lambda e: asyncio.run_coroutine_threadsafe(
play_next(interaction), interaction.client.loop
),
)
await interaction.followup.send(f"🎵 **Now playing:** {title}")
except Exception as e:
await interaction.followup.send(f"Failed to play the next song. Error: {str(e)}")
async def stop_audio(interaction: discord.Interaction):
guild_id = interaction.guild.id
if guild_id in voice_clients:
voice_client = voice_clients[guild_id]
if voice_client.is_playing():
voice_client.stop()
music_queues[guild_id] = [] # Clear the queue
current_tracks.pop(guild_id, None) # Remove the current track
await interaction.response.send_message("🛑 **Playback stopped and queue cleared.**")
else:
await interaction.response.send_message("❌ **No music is playing.**")
async def pause_audio(interaction: discord.Interaction):
"""Pauses the currently playing song."""
guild_id = interaction.guild.id
if guild_id in voice_clients and voice_clients[guild_id].is_playing():
voice_clients[guild_id].pause() # Pause playback
await interaction.response.send_message("⏸️ **Music paused.**")
else:
await interaction.response.send_message("❌ **No music is currently playing.**")
async def resume_audio(interaction: discord.Interaction):
"""Resumes playback of a paused song."""
guild_id = interaction.guild.id
if guild_id in voice_clients and voice_clients[guild_id].is_paused():
voice_clients[guild_id].resume() # Resume playback
await interaction.response.send_message("▶️ **Music resumed.**")
else:
await interaction.response.send_message("❌ **No paused music to resume.**")
async def set_volume(interaction: discord.Interaction, level: int):
guild_id = interaction.guild.id
if level < 0 or level > 100:
await interaction.response.send_message("❌ **Volume must be between 0 and 100.**")
return
# Set the volume
volume = level / 100
volumes[guild_id] = volume
# Adjust volume for the current source if playing
if guild_id in voice_clients and voice_clients[guild_id].is_playing():
source = voice_clients[guild_id].source
if isinstance(source, discord.PCMVolumeTransformer):
source.volume = volume
await interaction.response.send_message(f"🔊 **Volume set to {level}%**.")
async def leave_voice(interaction: discord.Interaction):
guild_id = interaction.guild.id
if guild_id in voice_clients:
voice_client = voice_clients[guild_id]
if voice_client.is_connected():
await voice_client.disconnect()
voice_clients.pop(guild_id, None)
await interaction.response.send_message("👋 **Left the voice channel.**")
else:
await interaction.response.send_message("❌ **I am not connected to any voice channel.**")
else:
await interaction.response.send_message("❌ **I am not connected to any voice channel.**")
# ---------- Recommendations ----------
def fetch_related_songs(song_title, artist):
"""
Fetch related songs using the YouTube Data API v3.
Returns a list of recommendations in the format:
[ "Title - URL" ]
"""
query = f"{song_title} {artist} related songs"
url = "https://www.googleapis.com/youtube/v3/search"
params = {
"q": query,
"key": YOUTUBE_API_KEY,
"type": "video",
"part": "snippet",
"maxResults": 5
}
response = requests.get(url, params=params)
if response.status_code == 200:
data = response.json()
return [
f'{item["snippet"]["title"]} - https://www.youtube.com/watch?v={item["id"]["videoId"]}'
for item in data.get("items", [])
]
else:
print(f"Error fetching related songs: {response.status_code} {response.text}")
return []
def generate_recommendations(user_id):
"""
Generate recommendations using the last song the user listened to.
"""
user_data = load_user_data()
user_history = user_data.get(str(user_id), {}).get("history", [])
if not user_history:
return ["No recommendations yet. Start playing some songs!"]
# Use the last played song to fetch recommendations
last_song = user_history[-1]
title = last_song["title"]
artist = last_song["artist"]
recommendations = fetch_related_songs(title, artist)
if recommendations:
return recommendations
else:
return ["No recommendations could be fetched at this time."]

70
commands.py
View File

@ -1,70 +0,0 @@
import discord
from discord import app_commands
from audio import (
play_audio,
stop_audio,
set_volume,
join_voice,
leave_voice,
pause_audio,
resume_audio,
repeat_modes,
generate_recommendations
)
async def setup_commands(client, guild_id=None):
@client.tree.command(name="join", description="Join the user's current voice channel.")
async def join(interaction: discord.Interaction):
await join_voice(interaction)
@client.tree.command(name="leave", description="Leave the current voice channel.")
async def leave(interaction: discord.Interaction):
await leave_voice(interaction)
@client.tree.command(name="play", description="Play a song by title or artist.")
async def play(interaction: discord.Interaction, query: str):
await play_audio(interaction, query)
@client.tree.command(name="stop", description="Stop playback and clear the queue.")
async def stop(interaction: discord.Interaction):
await stop_audio(interaction)
@client.tree.command(name="pause", description="Pause the current song.")
async def pause(interaction: discord.Interaction):
await pause_audio(interaction)
@client.tree.command(name="resume", description="Resume the paused song.")
async def resume(interaction: discord.Interaction):
await resume_audio(interaction)
@client.tree.command(name="repeat", description="Set the repeat mode for playback.")
@app_commands.describe(mode="Repeat mode: 'one', 'all', or 'off'")
async def repeat(interaction: discord.Interaction, mode: str):
guild_id = interaction.guild.id
valid_modes = ["one", "all", "off"]
if mode not in valid_modes:
await interaction.response.send_message(f"❌ **Invalid mode. Use one of: {', '.join(valid_modes)}.**")
return
repeat_modes[guild_id] = mode
await interaction.response.send_message(f"🔄 **Repeat mode set to:** {mode.capitalize()}.")
@client.tree.command(name="volume", description="Set playback volume.")
async def volume(interaction: discord.Interaction, level: int):
await set_volume(interaction, level)
@client.tree.command(name="recommend", description="Get song recommendations based on your listening history.")
async def recommend(interaction: discord.Interaction):
user_id = interaction.user.id
recommendations = generate_recommendations(user_id)
embed = discord.Embed(
title="🎵 Recommended Songs",
description="\n".join(recommendations),
color=discord.Color.green()
)
await interaction.response.send_message(embed=embed)
await client.tree.sync()

54
main.py
View File

@ -1,54 +0,0 @@
import os
import logging
import discord
from dotenv import load_dotenv
from commands import setup_commands
# Load environment variables
load_dotenv()
TOKEN = os.getenv("DISCORD_TOKEN")
GUILD_ID = os.getenv("GUILD_ID")
# Validate Guild ID
if GUILD_ID:
try:
GUILD_ID = int(GUILD_ID)
except ValueError:
logging.error("Invalid GUILD_ID in .env file. It must be a numeric value.")
GUILD_ID = None
intents = discord.Intents.default()
intents.messages = True
intents.message_content = True
intents.guilds = True
intents.voice_states = True
class AmberClient(discord.Client):
def __init__(self):
super().__init__(intents=intents)
self.tree = discord.app_commands.CommandTree(self)
async def on_ready(self):
logging.info(f"Amber is online as {self.user}")
# Sync commands after the bot is fully ready
if GUILD_ID:
logging.info(f"Setting up commands for guild: {GUILD_ID}")
else:
logging.info("Setting up global commands (no guild ID specified).")
try:
await setup_commands(self, guild_id=GUILD_ID)
except Exception as e:
logging.error(f"Failed to setup commands: {e}")
logging.basicConfig(level=logging.INFO)
client = AmberClient()
if TOKEN:
client.run(TOKEN)
else:
logging.error("Bot token not found. Check your .env file.")