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 08:10:51 +00:00
parent 7551362b2b
commit 77d1e3fcbc
4 changed files with 52 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
FEATURES.md
View File

@ -169,22 +169,22 @@ from discord.ext import commands
class Music(commands.Cog): class Music(commands.Cog):
def __init__(self, bot): def __init__(self, bot):
self.bot = bot self.bot = bot
@commands.command() @commands.command()
async def play(self, ctx, *, search: str): async def play(self, ctx, *, search: str):
if not ctx.voice_client: if not ctx.voice_client:
await ctx.author.voice.channel.connect(cls=pomice.Player) await ctx.author.voice.channel.connect(cls=pomice.Player)
player = ctx.voice_client player = ctx.voice_client
results = await player.get_tracks(query=search, ctx=ctx) results = await player.get_tracks(query=search, ctx=ctx)
if not results: if not results:
return await ctx.send("No results.") return await ctx.send("No results.")
track = results[0] track = results[0]
player.queue.put(track) 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: if not player.is_playing:
await player.do_next() await player.do_next()
@ -193,7 +193,7 @@ class Music(commands.Cog):
"""Show recently played songs.""" """Show recently played songs."""
player = ctx.voice_client player = ctx.voice_client
recent = player.history.get_last(5) recent = player.history.get_last(5)
msg = "\n".join(f"{i}. {t.title}" for i, t in enumerate(recent, 1)) msg = "\n".join(f"{i}. {t.title}" for i, t in enumerate(recent, 1))
await ctx.send(f"**Recently Played:**\n{msg}") await ctx.send(f"**Recently Played:**\n{msg}")
@ -202,7 +202,7 @@ class Music(commands.Cog):
"""Show queue analytics.""" """Show queue analytics."""
stats = ctx.voice_client.get_stats() stats = ctx.voice_client.get_stats()
summary = stats.get_summary() summary = stats.get_summary()
await ctx.send( await ctx.send(
f"**Queue Stats**\n" f"**Queue Stats**\n"
f"Tracks: {summary['total_tracks']}\n" f"Tracks: {summary['total_tracks']}\n"

74
examples/advanced_features.py
View File

@ -7,13 +7,13 @@ This example shows how easy it is to use:
- Integrated Analytics with player.get_stats() - Integrated Analytics with player.get_stats()
- Playlist Import/Export - Playlist Import/Export
""" """
import discord import discord
from discord.ext import commands from discord.ext import commands
import pomice import pomice
# Initialize bot # Initialize bot
bot = commands.Bot(command_prefix='!', intents=discord.Intents.all()) bot = commands.Bot(command_prefix="!", intents=discord.Intents.all())
class IntegratedMusic(commands.Cog): class IntegratedMusic(commands.Cog):
@ -27,13 +27,13 @@ class IntegratedMusic(commands.Cog):
"""Start Lavalink nodes.""" """Start Lavalink nodes."""
await self.pomice.create_node( await self.pomice.create_node(
bot=self.bot, bot=self.bot,
host='127.0.0.1', host="127.0.0.1",
port='3030', port="3030",
password='youshallnotpass', password="youshallnotpass",
identifier='MAIN' identifier="MAIN",
) )
@commands.command(name='play') @commands.command(name="play")
async def play(self, ctx, *, search: str): async def play(self, ctx, *, search: str):
"""Play a track using the integrated queue.""" """Play a track using the integrated queue."""
if not ctx.voice_client: if not ctx.voice_client:
@ -43,91 +43,89 @@ class IntegratedMusic(commands.Cog):
results = await player.get_tracks(query=search, ctx=ctx) results = await player.get_tracks(query=search, ctx=ctx)
if not results: if not results:
return await ctx.send('No results found.') return await ctx.send("No results found.")
if isinstance(results, pomice.Playlist): if isinstance(results, pomice.Playlist):
player.queue.extend(results.tracks) player.queue.extend(results.tracks)
await ctx.send(f'Added playlist **{results.name}** ({len(results.tracks)} tracks).') await ctx.send(f"Added playlist **{results.name}** ({len(results.tracks)} tracks).")
else: else:
track = results[0] track = results[0]
player.queue.put(track) 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: if not player.is_playing:
await player.do_next() await player.do_next()
@commands.command(name='history') @commands.command(name="history")
async def history(self, ctx, limit: int = 10): async def history(self, ctx, limit: int = 10):
"""Show recently played tracks (tracked automatically!).""" """Show recently played tracks (tracked automatically!)."""
player: pomice.Player = ctx.voice_client player: pomice.Player = ctx.voice_client
if not player: if not player:
return await ctx.send('Not connected.') return await ctx.send("Not connected.")
if player.history.is_empty: if player.history.is_empty:
return await ctx.send('No tracks in history.') return await ctx.send("No tracks in history.")
tracks = player.history.get_last(limit) tracks = player.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): 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) await ctx.send(embed=embed)
@commands.command(name='stats') @commands.command(name="stats")
async def queue_stats(self, ctx): async def queue_stats(self, ctx):
"""Show detailed queue statistics via integrated get_stats().""" """Show detailed queue statistics via integrated get_stats()."""
player: pomice.Player = ctx.voice_client player: pomice.Player = ctx.voice_client
if not player: if not player:
return await ctx.send('Not connected.') return await ctx.send("Not connected.")
stats = player.get_stats() stats = player.get_stats()
summary = stats.get_summary() 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="Tracks", value=summary["total_tracks"], inline=True)
embed.add_field(name='Duration', value=summary['total_duration_formatted'], inline=True) embed.add_field(name="Duration", value=summary["total_duration_formatted"], inline=True)
# Who added the most? # Who added the most?
top_requesters = stats.get_top_requesters(3) top_requesters = stats.get_top_requesters(3)
if top_requesters: if top_requesters:
text = '\n'.join(f'{u.display_name}: {c} tracks' for u, c in top_requesters) 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) embed.add_field(name="Top Requesters", value=text, inline=False)
await ctx.send(embed=embed) await ctx.send(embed=embed)
@commands.command(name='export') @commands.command(name="export")
async def export_queue(self, ctx, filename: str = 'my_playlist.json'): async def export_queue(self, ctx, filename: str = "my_playlist.json"):
"""Export current integrated queue.""" """Export current integrated queue."""
player: pomice.Player = ctx.voice_client player: pomice.Player = ctx.voice_client
if not player or player.queue.is_empty: if not player or player.queue.is_empty:
return await ctx.send('Queue is empty.') return await ctx.send("Queue is empty.")
pomice.PlaylistManager.export_queue( pomice.PlaylistManager.export_queue(
player.queue, player.queue, f"playlists/{filename}", name=f"{ctx.guild.name}'s Playlist",
f'playlists/{filename}',
name=f"{ctx.guild.name}'s Playlist"
) )
await ctx.send(f'✅ Queue exported to `playlists/{filename}`') await ctx.send(f"✅ Queue exported to `playlists/{filename}`")
@commands.command(name='sort') @commands.command(name="sort")
async def sort_queue(self, ctx): async def sort_queue(self, ctx):
"""Sort the queue using integrated utilities.""" """Sort the queue using integrated utilities."""
player: pomice.Player = ctx.voice_client player: pomice.Player = ctx.voice_client
if not player or player.queue.is_empty: if not player or player.queue.is_empty:
return await ctx.send('Queue is empty.') return await ctx.send("Queue is empty.")
# Use SearchHelper to sort the queue list # Use SearchHelper to sort the queue list
sorted_tracks = pomice.SearchHelper.sort_by_title(list(player.queue)) sorted_tracks = pomice.SearchHelper.sort_by_title(list(player.queue))
player.queue.clear() player.queue.clear()
player.queue.extend(sorted_tracks) player.queue.extend(sorted_tracks)
await ctx.send('✅ Queue sorted alphabetically.') await ctx.send("✅ Queue sorted alphabetically.")
@bot.event @bot.event
async def on_ready(): async def on_ready():
print(f'{bot.user} is ready!') print(f"{bot.user} is ready!")
if __name__ == "__main__": if __name__ == "__main__":

12
pomice/player.py
View File

@ -26,11 +26,11 @@ from .exceptions import TrackInvalidPosition
from .exceptions import TrackLoadError from .exceptions import TrackLoadError
from .filters import Filter from .filters import Filter
from .filters import Timescale from .filters import Timescale
from .history import TrackHistory
from .objects import Playlist from .objects import Playlist
from .objects import Track from .objects import Track
from .history import TrackHistory
from .queue_stats import QueueStats
from .queue import Queue from .queue import Queue
from .queue_stats import QueueStats
if TYPE_CHECKING: if TYPE_CHECKING:
from discord.types.voice import VoiceServerUpdate from discord.types.voice import VoiceServerUpdate
@ -192,7 +192,7 @@ class Player(VoiceProtocol):
self._voice_state: dict = {} self._voice_state: dict = {}
self._player_endpoint_uri: str = f"sessions/{self._node._session_id}/players" self._player_endpoint_uri: str = f"sessions/{self._node._session_id}/players"
self.queue: Queue = Queue() self.queue: Queue = Queue()
self.history: TrackHistory = TrackHistory() self.history: TrackHistory = TrackHistory()
@ -773,7 +773,7 @@ class Player(VoiceProtocol):
async def do_next(self) -> Optional[Track]: async def do_next(self) -> Optional[Track]:
"""Automatically plays the next track from the queue. """Automatically plays the next track from the queue.
Returns Returns
------- -------
Optional[Track] Optional[Track]
@ -781,14 +781,14 @@ class Player(VoiceProtocol):
""" """
if self.queue.is_empty: if self.queue.is_empty:
return None return None
track = self.queue.get() track = self.queue.get()
await self.play(track) await self.play(track)
return track return track
def get_stats(self) -> QueueStats: def get_stats(self) -> QueueStats:
"""Get detailed statistics for the current player and queue. """Get detailed statistics for the current player and queue.
Returns Returns
------- -------
QueueStats QueueStats

5
pomice/queue.py
View File

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