"""MIT License Copyright (c) 2023 Vocard Development Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. """ import time import function from math import ceil from asyncio import sleep from view import InteractiveController from typing import ( Any, Dict, Optional ) from discord import ( Client, Guild, VoiceChannel, VoiceProtocol, StageChannel, Member, Embed, ui ) from discord.ext import commands from . import events from .enums import SearchType from .events import VoicelinkEvent, TrackEndEvent, TrackStartEvent from .exceptions import VoicelinkException, FilterInvalidArgument, TrackInvalidPosition, TrackLoadError, FilterTagAlreadyInUse from .filters import Filter, Filters from .objects import Track from .pool import Node, NodePool from .queue import Queue, FairQueue class Player(VoiceProtocol): """The base player class for Voicelink. In order to initiate a player, you must pass it in as a cls when you connect to a channel. i.e: ```py await ctx.author.voice.channel.connect(cls=voicelink.Player) ``` """ def __call__(self, client: Client, channel: VoiceChannel): self.client: Client = client self.channel: VoiceChannel = channel return self def __init__( self, client: Optional[Client] = None, channel: Optional[VoiceChannel] = None, ctx: commands.Context = None, ): self.client = client self._bot = client self.context = ctx self.dj: Member = ctx.author self.channel = channel self._guild = channel.guild if channel else None self.settings: dict = function.get_settings(ctx.guild.id) self.joinTime = round(time.time()) self._volume = self.settings.get('volume', 100) self.lang = self.settings.get('lang', 'EN') if self.settings.get('lang', 'EN') in function.langs else "EN" self.queue = eval(self.settings.get("queueType", "Queue"))(5000, self.settings.get("duplicateTrack", True), self.get_msg) self._node = NodePool.get_node() self._current: Track = None self._filters: Filters = Filters() self._paused = False self._is_connected = False self._ping = 0.0 self._position = 0 self._last_position = 0 self._last_update = 0 self._ending_track: Optional[Track] = None self._voice_state = {} self.controller = None self.updating = False self.pause_votes = set() self.resume_votes = set() self.skip_votes = set() self.previous_votes = set() self.shuffle_votes = set() self.stop_votes = set() def __repr__(self): return ( f"" ) @property def position(self) -> float: """Property which returns the player's position in a track in milliseconds""" current = self._current.original if not self.is_playing or not self._current: return 0 if self.is_paused: return min(self._last_position, current.length) difference = (time.time() * 1000) - self._last_update position = self._last_position + difference if position > current.length: return 0 return min(position, current.length) @property def is_playing(self) -> bool: """Property which returns whether or not the player is actively playing a track.""" return self._is_connected and self._current is not None @property def is_connected(self) -> bool: """Property which returns whether or not the player is connected""" return self._is_connected @property def is_paused(self) -> bool: """Property which returns whether or not the player has a track which is paused or not.""" return self._is_connected and self._paused @property def current(self) -> Track: """Property which returns the currently playing track""" return self._current @property def node(self) -> Node: """Property which returns the node the player is connected to""" return self._node @property def guild(self) -> Guild: """Property which returns the guild associated with the player""" return self._guild @property def volume(self) -> int: """Property which returns the players current volume""" return self._volume @property def filters(self) -> Filter: """Property which returns the helper class for interacting with filters""" return self._filters @property def bot(self) -> Client: """Property which returns the bot associated with this player instance""" return self._bot @property def is_dead(self) -> bool: """Returns a bool representing whether the player is dead or not. A player is considered dead if it has been destroyed and removed from stored players. """ return self.guild.id not in self._node._players @property def ping(self) -> float: return round(self._ping / 1000, 2) def get_msg(self, mKey: str) -> str: return function.langs.get(self.lang, function.langs["EN"])[mKey] def required(self, leave=False): if self.settings.get('votedisable'): return 0 required = ceil((len(self.channel.members) - 1) / 2.5) if leave: if len(self.channel.members) == 3: required = 2 return required async def _update_state(self, data: dict): state: dict = data.get("state") self._last_update = time.time() * 1000 self._is_connected = state.get("connected") self._last_position = state.get("position") self._ping = state.get("ping") async def _dispatch_voice_update(self, voice_data: Dict[str, Any]): if {"sessionId", "event"} != self._voice_state.keys(): return await self._node.send( method=0, guild_id=self._guild.id, data = {"voice": { "token": voice_data['event']['token'], "endpoint": voice_data['event']['endpoint'], "sessionId": voice_data['sessionId'], }} ) async def on_voice_server_update(self, data: dict): self._voice_state.update({"event": data}) await self._dispatch_voice_update(self._voice_state) async def on_voice_state_update(self, data: dict): self._voice_state.update({"sessionId": data.get("session_id")}) if not (channel_id := data.get("channel_id")): await self.teardown() self._voice_state.clear() return self.channel = self.guild.get_channel(int(channel_id)) if not data.get("token"): return await self._dispatch_voice_update({**self._voice_state, "event": data}) async def _dispatch_event(self, data: dict): event_type = data.get("type") event: VoicelinkEvent = getattr(events, event_type)(data, self) if isinstance(event, TrackEndEvent) and event.reason != "REPLACED": self._current = None event.dispatch(self._bot) if isinstance(event, TrackStartEvent): self._ending_track = self._current async def do_next(self): if self.is_playing or not self.channel: return if self._paused: self._paused = False if not self.guild.me.voice: await self.connect(timeout=0.0, reconnect=True) self.pause_votes.clear() self.resume_votes.clear() self.skip_votes.clear() self.previous_votes.clear() self.shuffle_votes.clear() self.stop_votes.clear() track: Track = await self.queue.get() if not track: if self.settings.get("autoplay", False): if await function.similar_track(self): return await self.do_next() if self.settings.get('controller', True): await self.invoke_controller() return try: await self.play(track, start=track.position) except: await sleep(5) return await self.do_next() if self.settings.get('controller', True): await self.invoke_controller() async def invoke_controller(self): if self.updating or not self.channel: return self.updating = True if not self.controller: try: self.controller = await self.context.channel.send(embed=await self.build_embed(), view=InteractiveController(self)) except: pass elif not await self.is_position_fresh(): try: await self.controller.delete() except: ui.View.from_message(self.controller).stop() try: self.controller = await self.context.channel.send(embed= await self.build_embed(), view=InteractiveController(self)) except: pass else: embed = await self.build_embed() if embed: try: await self.controller.edit(embed=embed, view=InteractiveController(self)) except: pass self.updating = False async def build_embed(self): track = self.current if not track: embed=Embed(title=self.get_msg("noTrackPlaying"), description=f"[Vote](https://top.gg/bot/605618911471468554/vote/) | [Support]({function.invite_link}) | [Invite](https://discord.com/oauth2/authorize?client_id=605618911471468554&permissions=2184260928&scope=bot%20applications.commands) | [Questionnaire](https://forms.gle/UqeeEv4GEdCq9hi3A)", color=function.embed_color) embed.set_image(url='https://i.imgur.com/dIFBwU7.png') else: try: embed = Embed(color=function.embed_color) embed.set_author(name=self.get_msg("playerAuthor").format(self.channel.name), icon_url=self.client.user.avatar.url) embed.description = self.get_msg("playerDesc").format(track.title, track.uri, (track.requester.mention if track.requester else "<@605618911471468554>"), (f"<@&{self.settings['dj']}>" if self.settings.get('dj') else f"{self.dj.mention}")) embed.set_image(url=track.thumbnail if track.thumbnail else "https://cdn.discordapp.com/attachments/674788144931012638/823086668445384704/eq-dribbble.gif") embed.set_footer(text=self.get_msg("playerFooter").format(self.queue.count, (self.get_msg("live") if track.is_stream else function.time(track.length)), self.volume, self.get_msg("playerFooter2").format(self.queue.repeat.capitalize()) if self.queue._repeat else "")) except: embed = Embed(description=self.get_msg("missingTrackInfo"), color=function.embed_color) return embed async def is_position_fresh(self): try: async for message in self.context.channel.history(limit=5): if message.id == self.controller.id: return True except: pass return False async def teardown(self): timeNow = round(time.time()) function.update_settings(self.guild.id, {"lastActice": timeNow, "playTime": round(self.settings.get("playTime", 0) + ((timeNow - self.joinTime) / 60), 2)}) try: await self.controller.delete() except: pass try: await self.destroy() except: pass async def is_privileged(self, user: Member): if 'dj' in self.settings and self.settings['dj']: return user.guild_permissions.manage_guild or (self.settings['dj'] in [role.id for role in user.roles]) return self.dj.id == user.id or user.guild_permissions.manage_guild async def get_tracks( self, query: str, *, requester: Member, search_type: SearchType = SearchType.ytsearch ): """Fetches tracks from the node's REST api to parse into Lavalink. If you passed in Spotify API credentials when you created the node, you can also pass in a Spotify URL of a playlist, album or track and it will be parsed accordingly. You can also pass in a discord.py Context object to get a Context object on any track you search. """ return await self._node.get_tracks(query, requester=requester, search_type=search_type) async def spotifySearch(self, query: str, *, requester: Member) -> list: try: tracks = await self._node._spotify_client.trackSearch(query=query) except: raise TrackLoadError("Not able to find the provided Spotify entity, is it private?") return [ Track( track_id=track.id, requester=requester, search_type=SearchType.ytsearch, spotify=True, spotify_track=track, info={ "title": track.name, "author": track.artists, "length": track.length, "identifier": track.id, "artistId": track.artistId, "uri": track.uri, "isStream": False, "isSeekable": True, "position": 0, "thumbnail": track.image } ) for track in tracks ] async def spotifyRelatedTrack(self, seed_artists: str, seed_tracks: str): tracks = await self._node._spotify_client.similar_track(seed_artists=seed_artists, seed_tracks=seed_tracks) return [ Track( track_id=track.id, search_type=SearchType.ytsearch, spotify=True, spotify_track=track, info={ "title": track.name, "author": track.artists, "length": track.length, "identifier": track.id, "artistId": track.artistId, "uri": track.uri, "isStream": False, "isSeekable": True, "position": 0, "thumbnail": track.image }, requester=self.client.user ) for track in tracks ] async def connect(self, *, timeout: float, reconnect: bool, self_deaf: bool = True, self_mute: bool = False): await self.guild.change_voice_state(channel=self.channel, self_deaf=True, self_mute=self_mute) self._node._players[self.guild.id] = self self._is_connected = True async def stop(self): """Stops the currently playing track.""" self._current = None await self._node.send(method=0, guild_id=self._guild.id, data={'encodedTrack': None}) async def disconnect(self, *, force: bool = False): """Disconnects the player from voice.""" try: await self.guild.change_voice_state(channel=None) finally: self.cleanup() self._is_connected = False self.channel = None async def destroy(self): """Disconnects and destroys the player, and runs internal cleanup.""" try: await self.disconnect() except: # 'NoneType' has no attribute '_get_voice_client_key' raised by self.cleanup() -> # assume we're already disconnected and cleaned up assert self.channel is None and not self.is_connected self._node._players.pop(self.guild.id) await self._node.send(method=1, guild_id=self._guild.id) async def play( self, track: Track, *, start: int = 0, end: int = 0, ignore_if_playing: bool = False ) -> Track: """Plays a track. If a Spotify track is passed in, it will be handled accordingly.""" if not self._node: return track if track.spotify: if not track.original: search: Track = (await self._node.get_tracks( f"ytmsearch:{track.author} - {track.title}", requester=track.requester )) if not search: raise TrackLoadError("Can't not found a playable source!") track.original = search[0] data = { "encodedTrack": track.original.track_id if track.original else track.track_id, "position": str(start) } if end > 0: data["endTime"] = str(end) await self._node.send( method=0, guild_id=self._guild.id, data=data, query=f"noReplace={ignore_if_playing}" ) self._current = track if self.volume != 100: await self.set_volume(self.volume) return self._current async def seek(self, position: float) -> float: """Seeks to a position in the currently playing track milliseconds""" if position < 0 or position > self._current.original.length: raise TrackInvalidPosition( "Seek position must be between 0 and the track length" ) await self._node.send(method=0, guild_id=self._guild.id, data={"position": position}) return self._position async def set_pause(self, pause: bool) -> bool: """Sets the pause state of the currently playing track.""" await self._node.send(method=0, guild_id=self._guild.id, data={"paused": pause}) self._paused = pause return self._paused async def set_volume(self, volume: int) -> int: """Sets the volume of the player as an integer. Lavalink accepts values from 0 to 500.""" await self._node.send(method=0, guild_id=self._guild.id, data={"volume": volume}) self._volume = volume return self._volume async def add_filter(self, filter: Filter, fast_apply=False) -> Filters: try: self._filters.add_filter(filter=filter) except FilterTagAlreadyInUse: raise FilterTagAlreadyInUse(self.get_msg("FilterTagAlreadyInUse")) payload = self._filters.get_all_payloads() await self._node.send(method=0, guild_id=self._guild.id, data={"filters": payload}) if fast_apply: await self.seek(self.position) return self._filters async def remove_filter(self, filter_tag: str, fast_apply=False) -> Filters: self._filters.remove_filter(filter_tag=filter_tag) payload = self._filters.get_all_payloads() await self._node.send(method=0, guild_id=self._guild.id, data={"filters": payload}) if fast_apply: await self.seek(self.position) return self._filters async def reset_filter(self, *, fast_apply=False) -> None: if not self._filters: raise FilterInvalidArgument("You must have filters applied first in order to use this method.") self._filters.reset_filters() await self._node.send(method=0, guild_id=self._guild.id, data={"filters": {}}) if fast_apply: await self.seek(self.position) async def change_node(self, identifier: str = None) -> None: """Change node. """ try: node = NodePool.get_node(identifier=identifier) except: return await self.teardown() self._node._players.pop(self.guild.id) self._node = node self._node._players[self.guild.id] = self await self._dispatch_voice_update(self._voice_state) if self.current: await self.play(self.current, start=self.position) self._last_update = time.time() * 1000 if self.is_paused: await self.set_pause(True) if self.volume != 100: await self.set_volume(self.volume)