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

Final integration of advanced features into core classes

This commit is contained in:
wizardoesmagic 2025-12-28 08:10:42 +00:00
parent 012bea6a19
commit 7551362b2b
4 changed files with 245 additions and 793 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

643
FEATURES.md
View File

@ -2,167 +2,103 @@
## 🎉 Overview
This guide covers the advanced features added to Pomice to enhance your music bot capabilities. These features include track history, queue statistics, playlist management, and advanced track utilities.
Pomice now comes with built-in advanced features to help you build powerful music bots. These features are **integrated directly into the Player and Queue classes**, providing a "batteries-included" experience.
### What's New
### Key Enhancements
- **Track History**: Keep track of previously played songs with navigation and search
- **Queue Statistics**: Detailed analytics about queue contents (duration, requesters, etc.)
- **Playlist Manager**: Export/import playlists to JSON and M3U formats
- **Track Utilities**: Advanced filtering, searching, and sorting capabilities
All features are **fully backward compatible** and **optional** - use what you need!
- **Integrated Queue & History**: Every `Player` now has its own `queue` and `history` automatically.
- **Auto-History**: Tracks are automatically added to history when they finish playing.
- **Advanced Analytics**: Detailed statistics available directly via `player.get_stats()` or `queue.get_stats()`.
- **Integrated Utilities**: Filtering, sorting, and playlist management.
---
## 📚 Table of Contents
1. [Track History](#-track-history)
2. [Queue Statistics](#-queue-statistics)
3. [Playlist Manager](#-playlist-manager)
4. [Track Utilities](#-track-utilities)
5. [Complete Examples](#-complete-examples)
6. [Quick Reference](#-quick-reference)
1. [Integrated Features](#-integrated-features)
2. [Track History](#-track-history)
3. [Queue Statistics](#-queue-statistics)
4. [Playlist Manager](#-playlist-manager)
5. [Track Utilities](#-track-utilities)
6. [Complete Examples](#-complete-examples)
---
## 🚀 Integrated Features
Since these features are now part of the core classes, usage is extremely simple:
```python
# Every player now has a queue and history by default
player = ctx.voice_client
# Access the queue
player.queue.put(track)
# Play the next track from the queue
await player.do_next()
# Access the history (automatically updated)
last_song = player.history.current
# Get real-time statistics
stats = player.get_stats()
print(f"Queue Duration: {stats.format_duration(stats.total_duration)}")
```
---
## 🕐 Track History
Keep track of previously played songs with navigation and search capabilities.
The `player.history` object automatically tracks every song that finishes playing.
### Features
- Configurable maximum history size
- Navigation (previous/next)
- Search through history
- Filter by requester
- Get unique tracks (remove duplicates)
### Basic Usage
- Configurable maximum history size (default: 100)
- Navigation: `get_previous()`, `get_next()`
- Search: `history.search("query")`
- Filter: `get_by_requester(user_id)`
- Unique tracks: `get_unique_tracks()`
### Usage
```python
import pomice
# Create a history tracker
history = pomice.TrackHistory(max_size=100)
# Add tracks as they play
history.add(track)
# Get last 10 played tracks
recent = history.get_last(10)
# Show last 10 songs
recent = player.history.get_last(10)
# Search history
results = history.search("Imagine Dragons")
results = player.history.search("Imagine Dragons")
# Get tracks by specific user
user_tracks = history.get_by_requester(user_id=123456789)
# Navigate through history
previous_track = history.get_previous()
next_track = history.get_next()
# Get all unique tracks (removes duplicates)
unique = history.get_unique_tracks()
# Clear history
history.clear()
# Play previous track
prev = player.history.get_previous()
if prev:
await player.play(prev)
```
### Properties
- `is_empty` - Check if history is empty
- `current` - Get current track in navigation
### Use Cases
- "What was that song that just played?"
- "Show me the last 10 songs"
- "Play the previous track"
- "Show all songs requested by User X"
---
## 📊 Queue Statistics
Get detailed analytics about your queue contents.
Access advanced analytics via `player.get_stats()` or `player.queue.get_stats()`.
### Features
- Total and average duration
- Longest/shortest tracks
- Requester statistics
- Total/Average duration
- Longest/Shortest tracks
- Requester analytics (who added what)
- Author distribution
- Duration breakdown
- Stream detection
- Playlist distribution
### Basic Usage
- Duration breakdown (short/medium/long)
### Usage
```python
import pomice
# Create stats for a queue
stats = pomice.QueueStats(player.queue)
# Get total duration
total_ms = stats.total_duration
formatted = stats.format_duration(total_ms) # "1:23:45"
# Get average duration
avg_ms = stats.average_duration
# Find longest and shortest tracks
longest = stats.longest_track
shortest = stats.shortest_track
# Get requester statistics
requester_stats = stats.get_requester_stats()
# Returns: {user_id: {'count': 5, 'total_duration': 900000, 'tracks': [...]}}
# Get top requesters
top_requesters = stats.get_top_requesters(limit=5)
# Returns: [(requester, count), ...]
# Get author distribution
authors = stats.get_author_distribution()
# Returns: {'Artist Name': track_count, ...}
# Get top authors
top_authors = stats.get_top_authors(limit=10)
# Returns: [('Artist Name', count), ...]
# Get duration breakdown
breakdown = stats.get_duration_breakdown()
# Returns: {'short': 10, 'medium': 25, 'long': 5, 'very_long': 2}
# Get stream count
streams = stats.get_stream_count()
# Get comprehensive summary
stats = player.get_stats()
summary = stats.get_summary()
```
### Summary Dictionary
```python
{
'total_tracks': 42,
'total_duration': 7200000, # milliseconds
'total_duration_formatted': '2:00:00',
'average_duration': 171428.57,
'average_duration_formatted': '2:51',
'longest_track': Track(...),
'shortest_track': Track(...),
'stream_count': 3,
'unique_authors': 15,
'unique_requesters': 5,
'duration_breakdown': {...},
'loop_mode': LoopMode.QUEUE,
'is_looping': True
}
```
print(f"Total Tracks: {summary['total_tracks']}")
print(f"Total Duration: {summary['total_duration_formatted']}")
### Use Cases
- "How long is the queue?"
- "Who added the most songs?"
- "What's the longest track?"
- "Show me queue statistics"
# Who added the most songs?
top = stats.get_top_requesters(3)
for user, count in top:
print(f"{user.display_name}: {count} tracks")
```
---
@ -170,230 +106,61 @@ summary = stats.get_summary()
Export and import playlists to/from JSON and M3U formats.
### Features
- Export queue to JSON
- Import playlists from JSON
- Export to M3U format
- Merge multiple playlists
- Remove duplicates
- Playlist metadata
### Export Queue
### Usage
```python
import pomice
# Export current queue
# Export current queue to file
pomice.PlaylistManager.export_queue(
player.queue,
filepath='playlists/my_playlist.json',
name='My Awesome Playlist',
description='Best songs ever',
include_metadata=True # Include requester info
filepath='playlists/party.json',
name='Party Mix'
)
```
### Import Playlist
```python
# Import playlist data
data = pomice.PlaylistManager.import_playlist('playlists/my_playlist.json')
# Import a playlist
data = pomice.PlaylistManager.import_playlist('playlists/rock.json')
uris = pomice.PlaylistManager.get_track_uris('playlists/rock.json')
# Get just the URIs
uris = pomice.PlaylistManager.get_track_uris('playlists/my_playlist.json')
# Load tracks into queue
for uri in uris:
results = await player.get_tracks(query=uri)
if results:
await player.queue.put(results[0])
player.queue.put(results[0])
```
### Export Track List
```python
# Export a list of tracks (not from queue)
tracks = [track1, track2, track3]
pomice.PlaylistManager.export_track_list(
tracks,
filepath='playlists/favorites.json',
name='Favorites',
description='My favorite tracks'
)
```
### Merge Playlists
```python
# Merge multiple playlists into one
pomice.PlaylistManager.merge_playlists(
filepaths=['playlist1.json', 'playlist2.json', 'playlist3.json'],
output_path='merged_playlist.json',
name='Mega Playlist',
remove_duplicates=True # Remove duplicate tracks
)
```
### Export to M3U
```python
# Export to M3U format (compatible with many players)
tracks = list(player.queue)
pomice.PlaylistManager.export_to_m3u(
tracks,
filepath='playlists/my_playlist.m3u',
name='My Playlist'
)
```
### Get Playlist Info
```python
# Get metadata without loading all tracks
info = pomice.PlaylistManager.get_playlist_info('playlists/my_playlist.json')
# Returns: {'name': '...', 'track_count': 42, 'total_duration': 7200000, ...}
```
### JSON Format
```json
{
"name": "My Playlist",
"description": "Best songs",
"created_at": "2024-01-15T12:30:00",
"track_count": 10,
"total_duration": 1800000,
"version": "1.0",
"tracks": [
{
"title": "Song Title",
"author": "Artist Name",
"uri": "https://...",
"identifier": "abc123",
"length": 180000,
"thumbnail": "https://...",
"isrc": "USRC12345678",
"requester_id": 123456789,
"requester_name": "User#1234"
}
]
}
```
### Use Cases
- "Save this queue for later"
- "Load my favorite playlist"
- "Merge all my playlists"
- "Export to M3U for my media player"
---
## 🔧 Track Utilities
Advanced filtering, searching, and sorting utilities for tracks.
### TrackFilter
Filter tracks by various criteria.
Advanced filtering and sorting.
### Filtering
```python
import pomice
tracks = list(player.queue)
# Filter by duration (milliseconds)
short_tracks = pomice.TrackFilter.by_duration(
tracks,
min_duration=60000, # 1 minute
max_duration=300000 # 5 minutes
)
# Get tracks under 5 minutes
short = pomice.TrackFilter.by_duration(tracks, max_duration=300000)
# Filter by author
artist_tracks = pomice.TrackFilter.by_author(
tracks,
author='Imagine Dragons',
exact=False # Case-insensitive contains
)
# Filter by title
title_tracks = pomice.TrackFilter.by_title(
tracks,
title='Thunder',
exact=True # Exact match
)
# Filter by requester
user_tracks = pomice.TrackFilter.by_requester(tracks, requester_id=123456789)
# Filter by playlist
playlist_tracks = pomice.TrackFilter.by_playlist(tracks, playlist_name='Rock Hits')
# Get only streams
streams = pomice.TrackFilter.streams_only(tracks)
# Get only non-streams
non_streams = pomice.TrackFilter.non_streams_only(tracks)
# Custom filter with lambda
long_tracks = pomice.TrackFilter.custom(
tracks,
predicate=lambda t: t.length > 600000 # > 10 minutes
)
# Get tracks by a specific artist
artist_songs = pomice.TrackFilter.by_author(tracks, "Artist Name")
```
### SearchHelper
Search, sort, and organize tracks.
### Sorting
```python
import pomice
# Sort queue by title
sorted_tracks = pomice.SearchHelper.sort_by_title(list(player.queue))
tracks = list(player.queue)
# Search tracks
results = pomice.SearchHelper.search_tracks(
tracks,
query='imagine',
search_title=True,
search_author=True,
case_sensitive=False
)
# Sort by duration
sorted_tracks = pomice.SearchHelper.sort_by_duration(
tracks,
reverse=True # Longest first
)
# Sort by title (alphabetically)
sorted_tracks = pomice.SearchHelper.sort_by_title(tracks)
# Sort by author
sorted_tracks = pomice.SearchHelper.sort_by_author(tracks)
# Remove duplicates
unique_tracks = pomice.SearchHelper.remove_duplicates(
tracks,
by_uri=True, # Remove by URI
by_title_author=False # Or by title+author combo
)
# Group by author
grouped = pomice.SearchHelper.group_by_author(tracks)
# Returns: {'Artist Name': [track1, track2, ...], ...}
# Group by playlist
grouped = pomice.SearchHelper.group_by_playlist(tracks)
# Get random tracks
random_tracks = pomice.SearchHelper.get_random_tracks(tracks, count=5)
# Clear and refill with sorted tracks
player.queue.clear()
player.queue.extend(sorted_tracks)
```
### Use Cases
- "Show me all songs by Artist X"
- "Find tracks between 3-5 minutes"
- "Sort queue by duration"
- "Remove duplicate songs"
- "Play 5 random tracks"
---
## 🎯 Complete Examples
### Example 1: Basic Music Bot with History
### Integrated Music Cog
```python
import pomice
@ -402,221 +169,57 @@ from discord.ext import commands
class Music(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.history = pomice.TrackHistory(max_size=100)
@commands.Cog.listener()
async def on_pomice_track_end(self, player, track, _):
# Add to history when track ends
self.history.add(track)
@commands.command()
async def history(self, ctx, limit: int = 10):
"""Show recently played tracks."""
recent = self.history.get_last(limit)
async def play(self, ctx, *, search: str):
if not ctx.voice_client:
await ctx.author.voice.channel.connect(cls=pomice.Player)
tracks_list = '\n'.join(
f"{i}. {track.title} by {track.author}"
for i, track in enumerate(recent, 1)
)
await ctx.send(f"**Recently Played:**\n{tracks_list}")
```
### Example 2: Queue Statistics Command
```python
@commands.command()
async def stats(self, ctx):
"""Show queue statistics."""
player = ctx.voice_client
stats = pomice.QueueStats(player.queue)
results = await player.get_tracks(query=search, ctx=ctx)
if not results:
return await ctx.send("No results.")
track = results[0]
player.queue.put(track)
await ctx.send(f"Added **{track.title}** to queue.")
if not player.is_playing:
await player.do_next()
@commands.command()
async def history(self, ctx):
"""Show recently played songs."""
player = ctx.voice_client
recent = player.history.get_last(5)
msg = "\n".join(f"{i}. {t.title}" for i, t in enumerate(recent, 1))
await ctx.send(f"**Recently Played:**\n{msg}")
@commands.command()
async def stats(self, ctx):
"""Show queue analytics."""
stats = ctx.voice_client.get_stats()
summary = stats.get_summary()
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)
if summary['longest_track']:
embed.add_field(
name='Longest Track',
value=f"{summary['longest_track'].title} ({stats.format_duration(summary['longest_track'].length)})",
inline=False
await ctx.send(
f"**Queue Stats**\n"
f"Tracks: {summary['total_tracks']}\n"
f"Duration: {summary['total_duration_formatted']}"
)
# Top requesters
top_requesters = stats.get_top_requesters(3)
if top_requesters:
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)
await ctx.send(embed=embed)
```
### Example 3: Export/Import Playlists
```python
@commands.command()
async def export(self, ctx, filename: str = 'playlist.json'):
"""Export current queue to a file."""
player = ctx.voice_client
pomice.PlaylistManager.export_queue(
player.queue,
f'playlists/{filename}',
name=f"{ctx.guild.name}'s Playlist",
description=f'Exported from {ctx.guild.name}'
)
await ctx.send(f'✅ Queue exported to `playlists/{filename}`')
@commands.command()
async def import_playlist(self, ctx, filename: str):
"""Import a playlist from a file."""
player = ctx.voice_client
data = pomice.PlaylistManager.import_playlist(f'playlists/{filename}')
uris = [track['uri'] for track in data['tracks'] if track.get('uri')]
added = 0
for uri in uris:
results = await player.get_tracks(query=uri, ctx=ctx)
if results:
await player.queue.put(results[0])
added += 1
await ctx.send(f'✅ Imported {added} tracks from `{data["name"]}`')
```
### Example 4: Filter and Sort Queue
```python
@commands.command()
async def filter_short(self, ctx):
"""Show tracks shorter than 3 minutes."""
player = ctx.voice_client
tracks = list(player.queue)
short_tracks = pomice.TrackFilter.by_duration(
tracks,
max_duration=180000 # 3 minutes in ms
)
await ctx.send(f'Found {len(short_tracks)} tracks under 3 minutes!')
@commands.command()
async def sort_queue(self, ctx, sort_by: str = 'duration'):
"""Sort the queue by duration, title, or author."""
player = ctx.voice_client
queue_tracks = list(player.queue)
if sort_by == 'duration':
sorted_tracks = pomice.SearchHelper.sort_by_duration(queue_tracks)
elif sort_by == 'title':
sorted_tracks = pomice.SearchHelper.sort_by_title(queue_tracks)
elif sort_by == 'author':
sorted_tracks = pomice.SearchHelper.sort_by_author(queue_tracks)
else:
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}')
```
---
## 📖 Quick Reference
### Track History
```python
history = pomice.TrackHistory(max_size=100)
history.add(track)
recent = history.get_last(10)
results = history.search("query")
previous = history.get_previous()
unique = history.get_unique_tracks()
```
### Queue Statistics
```python
stats = pomice.QueueStats(queue)
total = stats.total_duration
formatted = stats.format_duration(total)
top_users = stats.get_top_requesters(5)
summary = stats.get_summary()
```
### Playlist Manager
```python
# Export
pomice.PlaylistManager.export_queue(queue, 'playlist.json')
# Import
data = pomice.PlaylistManager.import_playlist('playlist.json')
# Merge
pomice.PlaylistManager.merge_playlists(['p1.json', 'p2.json'], 'merged.json')
# M3U
pomice.PlaylistManager.export_to_m3u(tracks, 'playlist.m3u')
```
### Track Utilities
```python
# Filter
short = pomice.TrackFilter.by_duration(tracks, max_duration=180000)
artist = pomice.TrackFilter.by_author(tracks, "Artist Name")
# Search & Sort
results = pomice.SearchHelper.search_tracks(tracks, "query")
sorted_tracks = pomice.SearchHelper.sort_by_duration(tracks)
unique = pomice.SearchHelper.remove_duplicates(tracks)
random = pomice.SearchHelper.get_random_tracks(tracks, 5)
```
---
## 📝 Notes
- All duration values are in **milliseconds**
- History should be maintained per-guild
- Exported playlists are in JSON format by default
- M3U export is compatible with most media players
- All utilities work with standard Pomice Track objects
---
## 🚀 Getting Started
1. **Import the features you need**:
```python
import pomice
```
2. **Use them in your commands**:
```python
history = pomice.TrackHistory()
stats = pomice.QueueStats(player.queue)
```
3. **Check the examples** in `examples/advanced_features.py` for a complete bot
4. **Experiment** and customize to fit your needs!
---
## 🎓 Additional Resources
- **Full Example Bot**: See `examples/advanced_features.py`
- **Main Documentation**: See the main Pomice README
- **Discord Support**: Join the Pomice Discord server
| Feature | Integrated Access |
| :--- | :--- |
| **Queue** | `player.queue` |
| **History** | `player.history` |
| **Statistics** | `player.get_stats()` |
| **Next Track** | `await player.do_next()` |
---

326
examples/advanced_features.py
View File

@ -1,328 +1,134 @@
"""
Example usage of Pomice's new advanced features.
Example usage of Pomice's integrated advanced features.
This example demonstrates:
- Track History
- Queue Statistics
- Playlist Export/Import
- Track Filtering and Search
This example shows how easy it is to use:
- Integrated Track History (auto-tracking)
- Integrated Player Queue
- Integrated Analytics with player.get_stats()
- Playlist Import/Export
"""
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):
"""Music cog with advanced features."""
class IntegratedMusic(commands.Cog):
"""Music cog with integrated advanced features."""
def __init__(self, bot):
self.bot = bot
self.pomice = pomice.NodePool()
# Track history for each guild
self.track_histories = {}
async def start_nodes(self):
"""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()
async def on_pomice_track_end(self, player, track, _):
"""Add track to history when it ends."""
if player.guild.id not in self.track_histories:
self.track_histories[player.guild.id] = pomice.TrackHistory(max_size=100)
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."""
"""Play a track using the integrated queue."""
if not ctx.voice_client:
await ctx.author.voice.channel.connect(cls=pomice.Player)
player = ctx.voice_client
player: pomice.Player = ctx.voice_client
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.")
player.queue.extend(results.tracks)
await ctx.send(f'Added playlist **{results.name}** ({len(results.tracks)} tracks).')
else:
track = results[0]
await player.queue.put(track)
await ctx.send(f"Added **{track.title}** to queue.")
player.queue.put(track)
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.")
"""Show recently played tracks (tracked automatically!)."""
player: pomice.Player = ctx.voice_client
if not player:
return await ctx.send('Not connected.')
history = self.track_histories[ctx.guild.id]
if player.history.is_empty:
return await ctx.send('No tracks in history.')
if history.is_empty:
return await ctx.send("No tracks in history.")
tracks = history.get_last(limit)
embed = discord.Embed(title="🎵 Recently Played", color=discord.Color.blue())
tracks = player.history.get_last(limit)
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.")
"""Show detailed queue statistics via integrated get_stats()."""
player: pomice.Player = ctx.voice_client
if not player:
return await ctx.send('Not connected.')
player = ctx.voice_client
stats = pomice.QueueStats(player.queue)
stats = player.get_stats()
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='Tracks', value=summary['total_tracks'], inline=True)
embed.add_field(name='Duration', value=summary['total_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"]:
embed.add_field(
name="Longest Track",
value=f"{summary['longest_track'].title} ({stats.format_duration(summary['longest_track'].length)})",
inline=False,
)
# Duration breakdown
breakdown = summary["duration_breakdown"]
embed.add_field(
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,
)
# Top requesters
# Who added the most?
top_requesters = stats.get_top_requesters(3)
if top_requesters:
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)
text = '\n'.join(f'{u.display_name}: {c} tracks' for u, c in top_requesters)
embed.add_field(name='Top Requesters', value=text, inline=False)
await ctx.send(embed=embed)
@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.")
@commands.command(name='export')
async def export_queue(self, ctx, filename: str = 'my_playlist.json'):
"""Export current integrated queue."""
player: pomice.Player = ctx.voice_client
if not player or player.queue.is_empty:
return await ctx.send('Queue is empty.')
player = ctx.voice_client
if player.queue.is_empty:
return await ctx.send("Queue is empty.")
try:
pomice.PlaylistManager.export_queue(
player.queue,
f"playlists/{filename}",
name=f"{ctx.guild.name}'s Playlist",
description=f"Exported from {ctx.guild.name}",
f'playlists/{filename}',
name=f"{ctx.guild.name}'s Playlist"
)
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'✅ Queue exported to `playlists/{filename}`')
@commands.command(name="import")
async def import_playlist(self, ctx, filename: str):
"""Import a playlist from a file."""
if not ctx.voice_client:
await ctx.author.voice.channel.connect(cls=pomice.Player)
@commands.command(name='sort')
async def sort_queue(self, ctx):
"""Sort the queue using integrated utilities."""
player: pomice.Player = ctx.voice_client
if not player or player.queue.is_empty:
return await ctx.send('Queue is empty.')
player = ctx.voice_client
# Use SearchHelper to sort the queue list
sorted_tracks = pomice.SearchHelper.sort_by_title(list(player.queue))
try:
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")]
added = 0
for uri in uris:
try:
results = await player.get_tracks(query=uri, ctx=ctx)
if results:
if isinstance(results, pomice.Playlist):
await player.queue.put(results.tracks)
added += len(results.tracks)
else:
await player.queue.put(results[0])
added += 1
except:
continue
await ctx.send(f'✅ Imported {added} tracks from `{data["name"]}`')
if not player.is_playing:
await player.do_next()
except FileNotFoundError:
await ctx.send(f"❌ Playlist file `{filename}` not found.")
except Exception as e:
await ctx.send(f"❌ Error importing playlist: {e}")
@commands.command(name="filter")
async def filter_queue(self, ctx, filter_type: str, *, value: str):
"""Filter queue by various criteria.
Examples:
!filter author Imagine Dragons
!filter duration 180000-300000 (3-5 minutes in ms)
!filter title Thunder
"""
if not ctx.voice_client:
return await ctx.send("Not connected to voice.")
player = ctx.voice_client
queue_tracks = list(player.queue)
if filter_type == "author":
filtered = pomice.TrackFilter.by_author(queue_tracks, value)
elif filter_type == "title":
filtered = pomice.TrackFilter.by_title(queue_tracks, value)
elif filter_type == "duration":
# Parse duration range (e.g., "180000-300000")
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,
)
else:
return await ctx.send("Duration format: min-max (in milliseconds)")
else:
return await ctx.send("Valid filters: author, title, duration")
if not filtered:
return await ctx.send("No tracks match the filter.")
embed = discord.Embed(
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,
)
if len(filtered) > 10:
embed.set_footer(text=f"Showing 10 of {len(filtered)} results")
await ctx.send(embed=embed)
@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.")
history = self.track_histories[ctx.guild.id]
results = history.search(query)
if not results:
return await ctx.send(f'No tracks found matching "{query}"')
embed = discord.Embed(
title=f'🔍 History Search: "{query}"',
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)
if len(results) > 10:
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"):
"""Sort the queue.
Options: duration, title, author
"""
if not ctx.voice_client:
return await ctx.send("Not connected to voice.")
player = ctx.voice_client
if player.queue.is_empty:
return await ctx.send("Queue is empty.")
queue_tracks = list(player.queue)
if sort_by == "duration":
sorted_tracks = pomice.SearchHelper.sort_by_duration(queue_tracks)
elif sort_by == "title":
sorted_tracks = pomice.SearchHelper.sort_by_title(queue_tracks)
elif sort_by == "author":
sorted_tracks = pomice.SearchHelper.sort_by_author(queue_tracks)
else:
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}")
player.queue.clear()
player.queue.extend(sorted_tracks)
await ctx.send('✅ Queue sorted alphabetically.')
@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!')
# Add cog
bot.add_cog(AdvancedMusic(bot))
# Run bot
bot.run("YOUR_BOT_TOKEN")
if __name__ == "__main__":
print("Example script ready for use!")

38
pomice/player.py
View File

@ -28,9 +28,9 @@ from .filters import Filter
from .filters import Timescale
from .objects import Playlist
from .objects import Track
from .pool import Node
from .pool import NodePool
from pomice.utils import LavalinkVersion
from .history import TrackHistory
from .queue_stats import QueueStats
from .queue import Queue
if TYPE_CHECKING:
from discord.types.voice import VoiceServerUpdate
@ -154,6 +154,8 @@ class Player(VoiceProtocol):
"_log",
"_voice_state",
"_player_endpoint_uri",
"queue",
"history",
)
def __call__(self, client: Client, channel: VoiceChannel) -> Player:
@ -191,6 +193,9 @@ class Player(VoiceProtocol):
self._player_endpoint_uri: str = f"sessions/{self._node._session_id}/players"
self.queue: Queue = Queue()
self.history: TrackHistory = TrackHistory()
def __repr__(self) -> str:
return (
f"<Pomice.player bot={self.bot} guildId={self.guild.id} "
@ -358,6 +363,8 @@ class Player(VoiceProtocol):
event: PomiceEvent = getattr(events, event_type)(data, self)
if isinstance(event, TrackEndEvent) and event.reason not in ("REPLACED", "replaced"):
if self._current:
self.history.add(self._current)
self._current = None
event.dispatch(self._bot)
@ -763,3 +770,28 @@ class Player(VoiceProtocol):
if self._log:
self._log.debug(f"Fast apply passed, now removing all filters instantly.")
await self.seek(self.position)
async def do_next(self) -> Optional[Track]:
"""Automatically plays the next track from the queue.
Returns
-------
Optional[Track]
The track that is now playing, or None if the queue is empty.
"""
if self.queue.is_empty:
return None
track = self.queue.get()
await self.play(track)
return track
def get_stats(self) -> QueueStats:
"""Get detailed statistics for the current player and queue.
Returns
-------
QueueStats
A QueueStats object containing detailed analytics.
"""
return QueueStats(self.queue)

11
pomice/queue.py
View File

@ -372,3 +372,14 @@ class Queue(Iterable[Track]):
else:
new_queue = self._queue[index : self.size]
self._queue = new_queue
def get_stats(self) -> "pomice.QueueStats":
"""Get detailed statistics for this queue.
Returns
-------
QueueStats
A QueueStats object containing detailed analytics.
"""
from .queue_stats import QueueStats
return QueueStats(self)