112 lines
4.9 KiB
Python
112 lines
4.9 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
import discord
|
|
from discord import app_commands
|
|
from discord.ext import commands
|
|
|
|
from zyrarbot.audio.track_factory import TrackFactory
|
|
from zyrarbot.music.music_player import MusicPlayer
|
|
from zyrarbot.storage.uploads_repository import UploadsRepository
|
|
from zyrarbot.ui.bot.cogs.common import MusicCogBase
|
|
|
|
logger = logging.getLogger(__name__)
|
|
AUTOCOMPLETE_LIMIT = 25
|
|
|
|
|
|
class PlaybackCog(MusicCogBase, name="Playback"):
|
|
"""Komendy dodawania i ponownego odtwarzania utworów."""
|
|
|
|
def __init__(
|
|
self,
|
|
bot: commands.Bot,
|
|
player: MusicPlayer,
|
|
tracks: TrackFactory,
|
|
uploads: UploadsRepository,
|
|
):
|
|
super().__init__(bot, player)
|
|
self._tracks = tracks
|
|
self._uploads = uploads
|
|
|
|
async def _uploads_autocomplete(
|
|
self, interaction: discord.Interaction, current: str
|
|
) -> list[app_commands.Choice[str]]:
|
|
names = self._uploads.search(current, limit=AUTOCOMPLETE_LIMIT)
|
|
return [app_commands.Choice(name=name, value=name) for name in names]
|
|
|
|
async def _history_autocomplete(
|
|
self, interaction: discord.Interaction, current: str
|
|
) -> list[app_commands.Choice[str]]:
|
|
entries = self.player.history(interaction.guild.id)
|
|
matches = [entry for entry in entries if current.lower() in entry.title.lower()]
|
|
return [
|
|
app_commands.Choice(name=entry.title[:100], value=entry.id)
|
|
for entry in matches[:AUTOCOMPLETE_LIMIT]
|
|
]
|
|
|
|
@app_commands.command(name="play", description="Odtwórz utwór z YouTube")
|
|
@app_commands.describe(query="Link YouTube lub zapytanie wyszukiwania")
|
|
async def play(self, interaction: discord.Interaction, query: str) -> None:
|
|
if not await self._ensure_voice(interaction):
|
|
return
|
|
await interaction.response.defer()
|
|
try:
|
|
track = await self._tracks.resolve(query, loop=self.bot.loop)
|
|
await self._enqueue_and_reply(interaction, track)
|
|
except Exception as error: # noqa: BLE001
|
|
logger.error("Komenda /play nie powiodła się: %s", error)
|
|
await interaction.followup.send(f"Wystąpił błąd: {error}", ephemeral=True)
|
|
|
|
@app_commands.command(name="playfile", description="Odtwórz plik przesłany na serwer")
|
|
@app_commands.describe(filename="Wybierz plik z listy przesłanych")
|
|
@app_commands.autocomplete(filename=_uploads_autocomplete)
|
|
async def playfile(self, interaction: discord.Interaction, filename: str) -> None:
|
|
if not await self._ensure_voice(interaction):
|
|
return
|
|
await interaction.response.defer()
|
|
if not self._uploads.exists(filename):
|
|
await interaction.followup.send("Nie znaleziono pliku na serwerze.", ephemeral=True)
|
|
return
|
|
try:
|
|
track = await self._tracks.from_file(self._uploads.resolve(filename))
|
|
await self._enqueue_and_reply(interaction, track)
|
|
except Exception as error: # noqa: BLE001
|
|
logger.error("Komenda /playfile nie powiodła się: %s", error)
|
|
await interaction.followup.send(f"Wystąpił błąd: {error}", ephemeral=True)
|
|
|
|
@app_commands.command(name="replay", description="Odtwórz ponownie utwór z historii")
|
|
@app_commands.describe(track_id="Wybierz utwór z historii odtwarzania")
|
|
@app_commands.autocomplete(track_id=_history_autocomplete)
|
|
async def replay(self, interaction: discord.Interaction, track_id: str) -> None:
|
|
if not await self._ensure_voice(interaction):
|
|
return
|
|
await interaction.response.defer()
|
|
guild_id = interaction.guild.id
|
|
entries = self.player.history(guild_id)
|
|
entry = self.player.find_history_entry(guild_id, track_id)
|
|
if entry is None and track_id.isdigit():
|
|
position = int(track_id) - 1
|
|
if 0 <= position < len(entries):
|
|
entry = entries[position]
|
|
if entry is None:
|
|
await interaction.followup.send("Nie znaleziono utworu w historii.", ephemeral=True)
|
|
return
|
|
try:
|
|
if entry.is_local:
|
|
filepath = entry.replay_source or ""
|
|
if not self._uploads.exists(filepath):
|
|
await interaction.followup.send(
|
|
"Plik lokalny z historii już nie istnieje na serwerze.", ephemeral=True
|
|
)
|
|
return
|
|
track = await self._tracks.from_file(
|
|
self._uploads.resolve(filepath), title=entry.title
|
|
)
|
|
else:
|
|
track = await self._tracks.from_url(entry.url or "", loop=self.bot.loop)
|
|
await self._enqueue_and_reply(interaction, track, prefix=" z historii")
|
|
except Exception as error: # noqa: BLE001
|
|
logger.error("Komenda /replay nie powiodła się: %s", error)
|
|
await interaction.followup.send(f"Wystąpił błąd: {error}", ephemeral=True)
|