nambrogi
/
pomice
mirror of https://github.com/cloudwithax/pomice.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

[pre-commit.ci] auto fixes from pre-commit.com hooks 

for more information, see https://pre-commit.ci
This commit is contained in:
pre-commit-ci[bot] 2025-12-28 07:58:12 +00:00
parent 0e7473a807
commit e5dd3aec86
7 changed files with 340 additions and 370 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

198
examples/advanced_features.py
View File

@ -7,14 +7,15 @@ This example demonstrates:
- Playlist Export/Import
- Track Filtering and Search
"""
import asyncio
import discord
from discord.ext import commands
import pomice
# Initialize bot
bot = commands.Bot(command_prefix='!', intents=discord.Intents.all())
bot = commands.Bot(command_prefix="!", intents=discord.Intents.all())
class AdvancedMusic(commands.Cog):
@ -31,10 +32,10 @@ class AdvancedMusic(commands.Cog):
"""Start Lavalink nodes."""
await self.pomice.create_node(
bot=self.bot,
host='127.0.0.1',
port='3030',
password='youshallnotpass',
identifier='MAIN'
host="127.0.0.1",
port="3030",
password="youshallnotpass",
identifier="MAIN",
)
@commands.Cog.listener()
@ -45,7 +46,7 @@ class AdvancedMusic(commands.Cog):
self.track_histories[player.guild.id].add(track)
@commands.command(name='play')
@commands.command(name="play")
async def play(self, ctx, *, search: str):
"""Play a track."""
if not ctx.voice_client:
@ -55,133 +56,111 @@ class AdvancedMusic(commands.Cog):
results = await player.get_tracks(query=search, ctx=ctx)
if not results:
return await ctx.send('No results found.')
return await ctx.send("No results found.")
if isinstance(results, pomice.Playlist):
await player.queue.put(results.tracks)
await ctx.send(f'Added playlist **{results.name}** with {len(results.tracks)} tracks.')
await ctx.send(f"Added playlist **{results.name}** with {len(results.tracks)} tracks.")
else:
track = results[0]
await player.queue.put(track)
await ctx.send(f'Added **{track.title}** to queue.')
await ctx.send(f"Added **{track.title}** to queue.")
if not player.is_playing:
await player.do_next()
@commands.command(name='history')
@commands.command(name="history")
async def history(self, ctx, limit: int = 10):
"""Show recently played tracks."""
if ctx.guild.id not in self.track_histories:
return await ctx.send('No history available.')
return await ctx.send("No history available.")
history = self.track_histories[ctx.guild.id]
if history.is_empty:
return await ctx.send('No tracks in history.')
return await ctx.send("No tracks in history.")
tracks = history.get_last(limit)
embed = discord.Embed(
title='🎵 Recently Played',
color=discord.Color.blue()
)
embed = discord.Embed(title="🎵 Recently Played", color=discord.Color.blue())
for i, track in enumerate(tracks, 1):
embed.add_field(
name=f'{i}. {track.title}',
value=f'by {track.author}',
inline=False
)
embed.add_field(name=f"{i}. {track.title}", value=f"by {track.author}", inline=False)
await ctx.send(embed=embed)
@commands.command(name='stats')
@commands.command(name="stats")
async def queue_stats(self, ctx):
"""Show detailed queue statistics."""
if not ctx.voice_client:
return await ctx.send('Not connected to voice.')
return await ctx.send("Not connected to voice.")
player = ctx.voice_client
stats = pomice.QueueStats(player.queue)
summary = stats.get_summary()
embed = discord.Embed(
title='📊 Queue Statistics',
color=discord.Color.green()
embed = discord.Embed(title="📊 Queue Statistics", color=discord.Color.green())
embed.add_field(name="Total Tracks", value=summary["total_tracks"], inline=True)
embed.add_field(
name="Total Duration", value=summary["total_duration_formatted"], inline=True,
)
embed.add_field(
name="Average Duration", value=summary["average_duration_formatted"], inline=True,
)
embed.add_field(
name='Total Tracks',
value=summary['total_tracks'],
inline=True
)
embed.add_field(
name='Total Duration',
value=summary['total_duration_formatted'],
inline=True
)
embed.add_field(
name='Average Duration',
value=summary['average_duration_formatted'],
inline=True
)
if summary['longest_track']:
if summary["longest_track"]:
embed.add_field(
name='Longest Track',
name="Longest Track",
value=f"{summary['longest_track'].title} ({stats.format_duration(summary['longest_track'].length)})",
inline=False
inline=False,
)
# Duration breakdown
breakdown = summary['duration_breakdown']
breakdown = summary["duration_breakdown"]
embed.add_field(
name='Duration Breakdown',
name="Duration Breakdown",
value=f"Short (<3min): {breakdown['short']}\n"
f"Medium (3-6min): {breakdown['medium']}\n"
f"Long (6-10min): {breakdown['long']}\n"
f"Very Long (>10min): {breakdown['very_long']}",
inline=False
f"Medium (3-6min): {breakdown['medium']}\n"
f"Long (6-10min): {breakdown['long']}\n"
f"Very Long (>10min): {breakdown['very_long']}",
inline=False,
)
# Top requesters
top_requesters = stats.get_top_requesters(3)
if top_requesters:
requesters_text = '\n'.join(
f'{i}. {req.display_name}: {count} tracks'
requesters_text = "\n".join(
f"{i}. {req.display_name}: {count} tracks"
for i, (req, count) in enumerate(top_requesters, 1)
)
embed.add_field(
name='Top Requesters',
value=requesters_text,
inline=False
)
embed.add_field(name="Top Requesters", value=requesters_text, inline=False)
await ctx.send(embed=embed)
@commands.command(name='export')
async def export_queue(self, ctx, filename: str = 'playlist.json'):
@commands.command(name="export")
async def export_queue(self, ctx, filename: str = "playlist.json"):
"""Export current queue to a file."""
if not ctx.voice_client:
return await ctx.send('Not connected to voice.')
return await ctx.send("Not connected to voice.")
player = ctx.voice_client
if player.queue.is_empty:
return await ctx.send('Queue is empty.')
return await ctx.send("Queue is empty.")
try:
pomice.PlaylistManager.export_queue(
player.queue,
f'playlists/{filename}',
f"playlists/{filename}",
name=f"{ctx.guild.name}'s Playlist",
description=f'Exported from {ctx.guild.name}'
description=f"Exported from {ctx.guild.name}",
)
await ctx.send(f'✅ Queue exported to `playlists/{filename}`')
await ctx.send(f"✅ Queue exported to `playlists/{filename}`")
except Exception as e:
await ctx.send(f'❌ Error exporting queue: {e}')
await ctx.send(f"❌ Error exporting queue: {e}")
@commands.command(name='import')
@commands.command(name="import")
async def import_playlist(self, ctx, filename: str):
"""Import a playlist from a file."""
if not ctx.voice_client:
@ -190,10 +169,10 @@ class AdvancedMusic(commands.Cog):
player = ctx.voice_client
try:
data = pomice.PlaylistManager.import_playlist(f'playlists/{filename}')
data = pomice.PlaylistManager.import_playlist(f"playlists/{filename}")
# Get URIs and search for tracks
uris = [track['uri'] for track in data['tracks'] if track.get('uri')]
uris = [track["uri"] for track in data["tracks"] if track.get("uri")]
added = 0
for uri in uris:
@ -215,11 +194,11 @@ class AdvancedMusic(commands.Cog):
await player.do_next()
except FileNotFoundError:
await ctx.send(f'❌ Playlist file `{filename}` not found.')
await ctx.send(f"❌ Playlist file `{filename}` not found.")
except Exception as e:
await ctx.send(f'❌ Error importing playlist: {e}')
await ctx.send(f"❌ Error importing playlist: {e}")
@commands.command(name='filter')
@commands.command(name="filter")
async def filter_queue(self, ctx, filter_type: str, *, value: str):
"""Filter queue by various criteria.
@ -229,55 +208,52 @@ class AdvancedMusic(commands.Cog):
!filter title Thunder
"""
if not ctx.voice_client:
return await ctx.send('Not connected to voice.')
return await ctx.send("Not connected to voice.")
player = ctx.voice_client
queue_tracks = list(player.queue)
if filter_type == 'author':
if filter_type == "author":
filtered = pomice.TrackFilter.by_author(queue_tracks, value)
elif filter_type == 'title':
elif filter_type == "title":
filtered = pomice.TrackFilter.by_title(queue_tracks, value)
elif filter_type == 'duration':
elif filter_type == "duration":
# Parse duration range (e.g., "180000-300000")
if '-' in value:
min_dur, max_dur = map(int, value.split('-'))
if "-" in value:
min_dur, max_dur = map(int, value.split("-"))
filtered = pomice.TrackFilter.by_duration(
queue_tracks,
min_duration=min_dur,
max_duration=max_dur
queue_tracks, min_duration=min_dur, max_duration=max_dur,
)
else:
return await ctx.send('Duration format: min-max (in milliseconds)')
return await ctx.send("Duration format: min-max (in milliseconds)")
else:
return await ctx.send('Valid filters: author, title, duration')
return await ctx.send("Valid filters: author, title, duration")
if not filtered:
return await ctx.send('No tracks match the filter.')
return await ctx.send("No tracks match the filter.")
embed = discord.Embed(
title=f'🔍 Filtered Results ({len(filtered)} tracks)',
color=discord.Color.purple()
title=f"🔍 Filtered Results ({len(filtered)} tracks)", color=discord.Color.purple(),
)
for i, track in enumerate(filtered[:10], 1):
stats = pomice.QueueStats(player.queue)
embed.add_field(
name=f'{i}. {track.title}',
value=f'by {track.author} - {stats.format_duration(track.length)}',
inline=False
name=f"{i}. {track.title}",
value=f"by {track.author} - {stats.format_duration(track.length)}",
inline=False,
)
if len(filtered) > 10:
embed.set_footer(text=f'Showing 10 of {len(filtered)} results')
embed.set_footer(text=f"Showing 10 of {len(filtered)} results")
await ctx.send(embed=embed)
@commands.command(name='search_history')
@commands.command(name="search_history")
async def search_history(self, ctx, *, query: str):
"""Search through play history."""
if ctx.guild.id not in self.track_histories:
return await ctx.send('No history available.')
return await ctx.send("No history available.")
history = self.track_histories[ctx.guild.id]
results = history.search(query)
@ -287,63 +263,59 @@ class AdvancedMusic(commands.Cog):
embed = discord.Embed(
title=f'🔍 History Search: "{query}"',
description=f'Found {len(results)} tracks',
color=discord.Color.gold()
description=f"Found {len(results)} tracks",
color=discord.Color.gold(),
)
for i, track in enumerate(results[:10], 1):
embed.add_field(
name=f'{i}. {track.title}',
value=f'by {track.author}',
inline=False
)
embed.add_field(name=f"{i}. {track.title}", value=f"by {track.author}", inline=False)
if len(results) > 10:
embed.set_footer(text=f'Showing 10 of {len(results)} results')
embed.set_footer(text=f"Showing 10 of {len(results)} results")
await ctx.send(embed=embed)
@commands.command(name='sort')
async def sort_queue(self, ctx, sort_by: str = 'duration'):
@commands.command(name="sort")
async def sort_queue(self, ctx, sort_by: str = "duration"):
"""Sort the queue.
Options: duration, title, author
"""
if not ctx.voice_client:
return await ctx.send('Not connected to voice.')
return await ctx.send("Not connected to voice.")
player = ctx.voice_client
if player.queue.is_empty:
return await ctx.send('Queue is empty.')
return await ctx.send("Queue is empty.")
queue_tracks = list(player.queue)
if sort_by == 'duration':
if sort_by == "duration":
sorted_tracks = pomice.SearchHelper.sort_by_duration(queue_tracks)
elif sort_by == 'title':
elif sort_by == "title":
sorted_tracks = pomice.SearchHelper.sort_by_title(queue_tracks)
elif sort_by == 'author':
elif sort_by == "author":
sorted_tracks = pomice.SearchHelper.sort_by_author(queue_tracks)
else:
return await ctx.send('Valid options: duration, title, author')
return await ctx.send("Valid options: duration, title, author")
# Clear and refill queue
player.queue._queue.clear()
for track in sorted_tracks:
await player.queue.put(track)
await ctx.send(f'✅ Queue sorted by {sort_by}')
await ctx.send(f"✅ Queue sorted by {sort_by}")
@bot.event
async def on_ready():
print(f'{bot.user} is ready!')
await bot.get_cog('AdvancedMusic').start_nodes()
print(f"{bot.user} is ready!")
await bot.get_cog("AdvancedMusic").start_nodes()
# Add cog
bot.add_cog(AdvancedMusic(bot))
# Run bot
bot.run('YOUR_BOT_TOKEN')
bot.run("YOUR_BOT_TOKEN")

6
pomice/history.py
View File

@ -143,7 +143,8 @@ class TrackHistory:
"""
query_lower = query.lower()
return [
track for track in reversed(self._history)
track
for track in reversed(self._history)
if query_lower in track.title.lower() or query_lower in track.author.lower()
]
@ -177,7 +178,8 @@ class TrackHistory:
Tracks requested by the user (most recent first)
"""
return [
track for track in reversed(self._history)
track
for track in reversed(self._history)
if track.requester and track.requester.id == requester_id
]

116
pomice/playlist_manager.py
View File

@ -55,42 +55,42 @@ class PlaylistManager:
tracks_data = []
for track in queue:
track_dict = {
'title': track.title,
'author': track.author,
'uri': track.uri,
'identifier': track.identifier,
'length': track.length,
'is_stream': track.is_stream,
"title": track.title,
"author": track.author,
"uri": track.uri,
"identifier": track.identifier,
"length": track.length,
"is_stream": track.is_stream,
}
if include_metadata:
track_dict['requester_id'] = track.requester.id if track.requester else None
track_dict['requester_name'] = str(track.requester) if track.requester else None
track_dict['timestamp'] = track.timestamp
track_dict["requester_id"] = track.requester.id if track.requester else None
track_dict["requester_name"] = str(track.requester) if track.requester else None
track_dict["timestamp"] = track.timestamp
if track.thumbnail:
track_dict['thumbnail'] = track.thumbnail
track_dict["thumbnail"] = track.thumbnail
if track.isrc:
track_dict['isrc'] = track.isrc
track_dict["isrc"] = track.isrc
if track.playlist:
track_dict['playlist_name'] = track.playlist.name
track_dict["playlist_name"] = track.playlist.name
tracks_data.append(track_dict)
playlist_data = {
'name': name,
'description': description,
'created_at': datetime.utcnow().isoformat(),
'track_count': len(tracks_data),
'total_duration': sum(t['length'] for t in tracks_data),
'tracks': tracks_data,
'version': '1.0',
"name": name,
"description": description,
"created_at": datetime.utcnow().isoformat(),
"track_count": len(tracks_data),
"total_duration": sum(t["length"] for t in tracks_data),
"tracks": tracks_data,
"version": "1.0",
}
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, 'w', encoding='utf-8') as f:
with open(path, "w", encoding="utf-8") as f:
json.dump(playlist_data, f, indent=2, ensure_ascii=False)
@staticmethod
@ -115,7 +115,7 @@ class PlaylistManager:
"""
path = Path(filepath)
with open(path, 'r', encoding='utf-8') as f:
with open(path, encoding="utf-8") as f:
data = json.load(f)
return data
@ -148,29 +148,29 @@ class PlaylistManager:
tracks_data = [
{
'title': track.title,
'author': track.author,
'uri': track.uri,
'identifier': track.identifier,
'length': track.length,
'thumbnail': track.thumbnail,
'isrc': track.isrc,
"title": track.title,
"author": track.author,
"uri": track.uri,
"identifier": track.identifier,
"length": track.length,
"thumbnail": track.thumbnail,
"isrc": track.isrc,
}
for track in tracks
]
playlist_data = {
'name': name,
'description': description,
'created_at': datetime.utcnow().isoformat(),
'track_count': len(tracks_data),
'total_duration': sum(t['length'] for t in tracks_data),
'tracks': tracks_data,
'version': '1.0',
"name": name,
"description": description,
"created_at": datetime.utcnow().isoformat(),
"track_count": len(tracks_data),
"total_duration": sum(t["length"] for t in tracks_data),
"tracks": tracks_data,
"version": "1.0",
}
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, 'w', encoding='utf-8') as f:
with open(path, "w", encoding="utf-8") as f:
json.dump(playlist_data, f, indent=2, ensure_ascii=False)
@staticmethod
@ -188,7 +188,7 @@ class PlaylistManager:
List of track URIs
"""
data = PlaylistManager.import_playlist(filepath)
return [track['uri'] for track in data['tracks'] if track.get('uri')]
return [track["uri"] for track in data["tracks"] if track.get("uri")]
@staticmethod
def merge_playlists(
@ -220,8 +220,8 @@ class PlaylistManager:
for filepath in filepaths:
data = PlaylistManager.import_playlist(filepath)
for track in data['tracks']:
uri = track.get('uri', '')
for track in data["tracks"]:
uri = track.get("uri", "")
if remove_duplicates:
if uri and uri in seen_uris:
@ -232,18 +232,18 @@ class PlaylistManager:
all_tracks.append(track)
merged_data = {
'name': name or 'Merged Playlist',
'description': description or f'Merged from {len(filepaths)} playlists',
'created_at': datetime.utcnow().isoformat(),
'track_count': len(all_tracks),
'total_duration': sum(t['length'] for t in all_tracks),
'tracks': all_tracks,
'version': '1.0',
"name": name or "Merged Playlist",
"description": description or f"Merged from {len(filepaths)} playlists",
"created_at": datetime.utcnow().isoformat(),
"track_count": len(all_tracks),
"total_duration": sum(t["length"] for t in all_tracks),
"tracks": all_tracks,
"version": "1.0",
}
output = Path(output_path)
output.parent.mkdir(parents=True, exist_ok=True)
with open(output, 'w', encoding='utf-8') as f:
with open(output, "w", encoding="utf-8") as f:
json.dump(merged_data, f, indent=2, ensure_ascii=False)
@staticmethod
@ -267,16 +267,16 @@ class PlaylistManager:
path = Path(filepath)
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, 'w', encoding='utf-8') as f:
f.write('#EXTM3U\n')
with open(path, "w", encoding="utf-8") as f:
f.write("#EXTM3U\n")
if name:
f.write(f'#PLAYLIST:{name}\n')
f.write(f"#PLAYLIST:{name}\n")
for track in tracks:
# Duration in seconds
duration = track.length // 1000
f.write(f'#EXTINF:{duration},{track.author} - {track.title}\n')
f.write(f'{track.uri}\n')
f.write(f"#EXTINF:{duration},{track.author} - {track.title}\n")
f.write(f"{track.uri}\n")
@staticmethod
def get_playlist_info(filepath: str) -> Dict[str, Any]:
@ -295,10 +295,10 @@ class PlaylistManager:
data = PlaylistManager.import_playlist(filepath)
return {
'name': data.get('name'),
'description': data.get('description'),
'track_count': data.get('track_count'),
'total_duration': data.get('total_duration'),
'created_at': data.get('created_at'),
'version': data.get('version'),
"name": data.get("name"),
"description": data.get("description"),
"track_count": data.get("track_count"),
"total_duration": data.get("total_duration"),
"created_at": data.get("created_at"),
"version": data.get("version"),
}

62
pomice/queue_stats.py
View File

@ -100,15 +100,15 @@ class QueueStats:
user_id = track.requester.id
if user_id not in stats:
stats[user_id] = {
'count': 0,
'total_duration': 0,
'tracks': [],
'requester': track.requester
"count": 0,
"total_duration": 0,
"tracks": [],
"requester": track.requester,
}
stats[user_id]['count'] += 1
stats[user_id]['total_duration'] += track.length
stats[user_id]['tracks'].append(track)
stats[user_id]["count"] += 1
stats[user_id]["total_duration"] += track.length
stats[user_id]["tracks"].append(track)
return stats
@ -125,14 +125,12 @@ class QueueStats:
List[tuple]
List of (requester, count) tuples sorted by count (descending)
"""
requester_counts = Counter(
track.requester.id for track in self._queue if track.requester
)
requester_counts = Counter(track.requester.id for track in self._queue if track.requester)
# Get requester objects
stats = self.get_requester_stats()
return [
(stats[user_id]['requester'], count)
(stats[user_id]["requester"], count)
for user_id, count in requester_counts.most_common(limit)
]
@ -202,23 +200,23 @@ class QueueStats:
- 'very_long' (> 10 min)
"""
breakdown = {
'short': 0, # < 3 minutes
'medium': 0, # 3-6 minutes
'long': 0, # 6-10 minutes
'very_long': 0 # > 10 minutes
"short": 0, # < 3 minutes
"medium": 0, # 3-6 minutes
"long": 0, # 6-10 minutes
"very_long": 0, # > 10 minutes
}
for track in self._queue:
duration_minutes = track.length / 60000 # Convert ms to minutes
if duration_minutes < 3:
breakdown['short'] += 1
breakdown["short"] += 1
elif duration_minutes < 6:
breakdown['medium'] += 1
breakdown["medium"] += 1
elif duration_minutes < 10:
breakdown['long'] += 1
breakdown["long"] += 1
else:
breakdown['very_long'] += 1
breakdown["very_long"] += 1
return breakdown
@ -252,19 +250,19 @@ class QueueStats:
Dictionary containing various queue statistics
"""
return {
'total_tracks': len(self._queue),
'total_duration': self.total_duration,
'total_duration_formatted': self.format_duration(self.total_duration),
'average_duration': self.average_duration,
'average_duration_formatted': self.format_duration(int(self.average_duration)),
'longest_track': self.longest_track,
'shortest_track': self.shortest_track,
'stream_count': self.get_stream_count(),
'unique_authors': len(self.get_author_distribution()),
'unique_requesters': len(self.get_requester_stats()),
'duration_breakdown': self.get_duration_breakdown(),
'loop_mode': self._queue.loop_mode,
'is_looping': self._queue.is_looping,
"total_tracks": len(self._queue),
"total_duration": self.total_duration,
"total_duration_formatted": self.format_duration(self.total_duration),
"average_duration": self.average_duration,
"average_duration_formatted": self.format_duration(int(self.average_duration)),
"longest_track": self.longest_track,
"shortest_track": self.shortest_track,
"stream_count": self.get_stream_count(),
"unique_authors": len(self.get_author_distribution()),
"unique_requesters": len(self.get_requester_stats()),
"duration_breakdown": self.get_duration_breakdown(),
"loop_mode": self._queue.loop_mode,
"is_looping": self._queue.is_looping,
}
def __repr__(self) -> str:

6
pomice/track_utils.py
View File

@ -133,10 +133,7 @@ class TrackFilter:
Filtered tracks
"""
playlist_lower = playlist_name.lower()
return [
t for t in tracks
if t.playlist and playlist_lower in t.playlist.name.lower()
]
return [t for t in tracks if t.playlist and playlist_lower in t.playlist.name.lower()]
@staticmethod
def streams_only(tracks: List[Track]) -> List[Track]:
@ -404,4 +401,5 @@ class SearchHelper:
Random tracks (without replacement)
"""
import random
return random.sample(tracks, min(count, len(tracks)))