diff --git a/cogs/basic.py b/cogs/basic.py index a8f6d70..55d2e24 100644 --- a/cogs/basic.py +++ b/cogs/basic.py @@ -865,6 +865,9 @@ class Basic(commands.Cog): if not player.is_playing: await player.do_next() + + if player.is_ipc_connected: + await player.send_ws({"op": "toggleAutoplay", "status": check}) @commands.hybrid_command(name="help", aliases=get_aliases("help")) @app_commands.autocomplete(category=help_autocomplete) diff --git a/ipc/__init__.py b/ipc/__init__.py new file mode 100644 index 0000000..f29640f --- /dev/null +++ b/ipc/__init__.py @@ -0,0 +1 @@ +from .client import IPCClient \ No newline at end of file diff --git a/ipc/client.py b/ipc/client.py new file mode 100644 index 0000000..862d178 --- /dev/null +++ b/ipc/client.py @@ -0,0 +1,101 @@ +import aiohttp +import asyncio +import traceback +import logging + +from discord.ext import commands +from typing import Optional + +from .methods import process_methods + +class IPCClient: + def __init__( + self, + bot: commands.Bot, + host: str, + port: int, + password: str, + heartbeat: int = 30 + ) -> None: + + self._bot: commands.Bot = bot + self._host: str = host + self._port: int = port + self._password: str = password + self._heartbeat: int = heartbeat + self._is_connected: bool = False + self._is_connecting: bool = False + self._logger: logging.Logger = logging.getLogger("ipc_client") + + self._websocket_url: str = f"ws://{self._host}:{self._port}/ws_bot" + self._session: Optional[aiohttp.ClientSession] = None + self._websocket: Optional[aiohttp.ClientWebSocketResponse] = None + self._task: Optional[asyncio.Task] = None + + self._heanders = { + "Authorization": self._password, + "User-Id": str(bot.user.id), + } + + async def _listen(self) -> None: + while True: + try: + msg = await self._websocket.receive() + self._logger.debug(f"Receive Message: {msg}") + except: + break + + if msg.type in [aiohttp.WSMsgType.CLOSE, aiohttp.WSMsgType.CLOSED]: + self._is_connected = False + + self._logger.info("Trying to reconnect dashboard in 10s") + await asyncio.sleep(10) + if not self._is_connected: + try: + await self.connect() + except: + pass + else: + self._bot.loop.create_task(process_methods(self, self._bot, msg.json())) + + async def send(self, data: dict): + if self._is_connected: + self._logger.debug(f"Send Message: {data}") + await self._websocket.send_json(data) + + async def connect(self): + try: + if not self._session: + self._session = aiohttp.ClientSession() + + if self._is_connecting: + return + + self._is_connecting = True + self._websocket = await self._session.ws_connect( + self._websocket_url, headers=self._heanders, heartbeat=self._heartbeat + ) + + self._task = self._bot.loop.create_task(self._listen()) + self._is_connected = True + + self._logger.info("Connected to dashboard!") + + except aiohttp.ClientConnectorError: + raise Exception("The connection is failed.") + + except aiohttp.WSServerHandshakeError: + raise Exception("The password is invalid.") + + except Exception as e: + self._logger.error("Error occured while connecting to dashboard!", exc_info=e) + + finally: + self._is_connecting = False + + return self + + async def disconnect(self) -> None: + self._is_connected = False + self._task.cancel() + self._logger.info("Disconnected to dashboard!") \ No newline at end of file diff --git a/ipc/methods.py b/ipc/methods.py new file mode 100644 index 0000000..f07f79c --- /dev/null +++ b/ipc/methods.py @@ -0,0 +1,452 @@ +import function as func + +from typing import ( + List, + Dict +) + +from discord import Member, VoiceChannel +from discord.ext import commands +from voicelink import Player, Track, Playlist, NodePool, decode, LoopType + +class TempCtx(): + def __init__(self, author: Member, channel: VoiceChannel) -> None: + self.author = author + self.channel = channel + self.guild = channel.guild + +class SystemMethod: + def __init__(self, function): + self.function: callable = function + self.params: List[str] = ["bot", "data"] + +class PlayerMethod(SystemMethod): + def __init__(self, function, *, auto_connect: bool = False): + super().__init__(function) + self.params: List[str] = ["player", "member", "data"] + self.auto_connect: bool = auto_connect + +class UserMethod(SystemMethod): + def __init__(self, function): + super().__init__(function) + self.params: List[str] = ["member", "data"] + +def missingPermission(user_id:int): + payload = {"op": "errorMsg", "level": "info", "msg": "Only the DJ or admins may use this funciton!"} + payload["user_id"] = user_id + return payload + +def error_msg(msg: str, *, user_id: int = None, guild_id: int = None, level: str = "info"): + payload = {"op": "errorMsg", "level": level, "msg": msg} + if user_id: + payload["user_id"] = user_id + if guild_id: + payload["guild_id"] = guild_id + + return payload + +async def connect_channel(member: Member, bot: commands.Bot): + if not member.voice: + return + + channel = member.voice.channel + try: + settings = await func.get_settings(channel.guild.id) + player: Player = await channel.connect(cls=Player(bot, channel, TempCtx(member, channel), settings)) + await player.send_ws({"op": "createPlayer", "member_ids": [member.id for member in channel.members]}) + return player + except: + return + +async def initUser(bot: commands.Bot, data: Dict) -> Dict: + user_id = data.get("user_id") + user = bot.get_user(user_id) + if not user: + return + + guild_id = None + for guild in user.mutual_guilds: + member = guild.get_member(user.id) + if member.voice and member.voice.channel: + guild_id = guild.id + + return { + "op": "initUser", + "user_id": user_id, + "guild_id": guild_id + } + +async def initPlayer(player: Player, member: Member, data: Dict): + player._ipc_connection = True + return { + "op": "initPlayer", + "guild_id": player.guild.id, + "user_id": data.get("user_id"), + "users": [{ + "user_id": member.id, + "avatar_url": member.display_avatar.url, + "name": member.name + } for member in player.channel.members ], + "tracks": [ track.track_id for track in player.queue._queue ], + "repeat_mode": player.queue.repeat.lower(), + "channel_name": player.channel.name, + "current_queue_position": player.queue._position if player._current else player.queue._position + 1, + "current_position": 0 or player.position if player.is_playing else 0, + "is_playing": player.is_playing, + "is_paused": player.is_paused, + "is_dj": player.is_privileged(member, check_user_join=False), + "autoplay": player.settings.get("autoplay", False) + } + +async def skipTo(player: Player, member: Member, data: Dict): + if not player.is_privileged(member): + if player.current and member == player.current.requester: + pass + + elif member in player.skip_votes: + return error_msg(player.get_msg('voted'), user_id=member.id) + else: + player.skip_votes.add(member) + if len(player.skip_votes) >= (required := player.required()): + pass + else: + return error_msg(player.get_msg('skipVote').format(member, len(player.skip_votes), required), guild_id=player.guild.id) + + index = data.get("index", 1) + if index > 1: + player.queue.skipto(index) + + if player.queue._repeat.mode == LoopType.track: + await player.set_repeat(LoopType.off.name) + await player.stop() + +async def backTo(player: Player, member: Member, data: Dict): + if not player.is_privileged(member): + if player.current and member == player.current.requester: + pass + + elif member in player.skip_votes: + return error_msg(player.get_msg('voted'), user_id=member.id) + else: + player.skip_votes.add(member) + if len(player.skip_votes) >= (required := player.required()): + pass + else: + return error_msg(player.get_msg('backVote').format(member, len(player.skip_votes), required), guild_id=player.guild.id) + + index = data.get("index", 1) + if not player.is_playing: + player.queue.backto(index) + await player.do_next() + else: + player.queue.backto(index + 1) + await player.stop() + +async def moveTrack(player: Player, member: Member, data: Dict): + if not player.is_privileged(member): + return missingPermission(member.id) + + c = player.queue._position - 1 + position = data.get("position") + new_position = data.get("newPosition") + + moveItem = player.queue._queue[position] + player.queue._queue.remove(moveItem) + player.queue._queue.insert(new_position, moveItem) + + if position > c and new_position <= c: + player.queue._position += 1 + + elif position < c and new_position >= c: + player.queue._position -= 1 + + elif position == c: + player.queue._position = new_position + 1 + + return { + "op": "moveTrack", + "position": { + "index": position - c, + "track_id": moveItem.track_id + }, + "newPosition": { + "index": new_position - c + }, + "guild_id": player.guild.id, + "requester_id": member.id, + "skip_users": [member.id] + } + +async def addTracks(player: Player, member: Member, data: Dict): + raw_tracks = data.get("tracks", []) + tracks = [Track( + track_id=track_id, + info=decode(track_id), + requester=member + ) for track_id in raw_tracks] + + await player.add_track(tracks) + + if not player.is_playing: + await player.do_next() + +async def getTracks(player: Player, member: Member, data: Dict): + query = data.get("query", None) + + if query: + payload = {"op": "getTracks", "user_id": member.id} + tracks = await player.get_tracks(query, requester=member) + if not tracks: + return payload + + if isinstance(tracks, Playlist): + tracks = [ track for track in tracks.tracks[:50] ] + + payload["tracks"] = [ track.track_id for track in tracks ] + return payload + +async def shuffleTrack(player: Player, member: Member, data: Dict): + if not player.is_privileged(member): + + if member in player.shuffle_votes: + return error_msg(player.get_msg('voted'), user_id=member.id) + else: + player.shuffle_votes.add(member) + if len(player.shuffle_votes) >= (required := player.required()): + pass + else: + return error_msg(player.get_msg('shuffleVote').format(member, len(player.skip_votes), required), guild_id=player.guild.id) + + await player.shuffle(data.get("type", "queue"), member) + +async def repeatTrack(player: Player, member: Member, data: Dict): + if not player.is_privileged(member): + return missingPermission(member.id) + + await player.set_repeat() + +async def removeTrack(player: Player, member: Member, data: Dict): + if not player.is_privileged(member): + return missingPermission(member.id) + + position = data.get("position") + verify_id = data.get("track_id") + + track = player.queue._queue[position] + if track.track_id == verify_id: + player.queue._queue.remove(track) + + if position < player.queue._position: + player.queue._position -= 1 + + return { + "op": "removeTrack", + "positions": [position], + "track_ids": [track.track_id], + "current_queue_position": player.queue._position, + "requester_id": member.id, + "guild_id": player.guild.id + } + +async def updatePause(player: Player, member: Member, data: Dict): + pause = data.get("pause", True) + if not player.is_privileged(member): + if pause: + if member in player.pause_votes: + return error_msg(player.get_msg('voted'), user_id=member.id) + else: + player.pause_votes.add(member) + if len(player.pause_votes) >= (required := player.required()): + pass + else: + return error_msg(player.get_msg('pauseVote').format(member, len(player.pause_votes), required), guild_id=player.guild.id) + else: + if member in player.resume_votes: + return error_msg(player.get_msg('voted'), user_id=member.id) + else: + player.resume_votes.add(member) + if len(player.resume_votes) >= (required := player.required()): + pass + else: + return error_msg(player.get_msg('resumeVote').format(member, len(player.resume_votes), required), guild_id=player.guild.id) + + if pause: + player.pause_votes.clear() + else: + player.resume_votes.clear() + + await player.set_pause(pause, member) + +async def updatePosition(player: Player, member: Member, data: Dict): + if not player.is_privileged(member): + return missingPermission(member.id) + + position = data.get("position"); + await player.seek(position, member); + +async def toggleAutoplay(player: Player, member: Member, data: Dict): + if not player.is_privileged(member): + return error_msg(player.get_msg('missingPerms_autoplay')) + + check = data.get("status", False) + player.settings['autoplay'] = check + + if not player.is_playing: + await player.do_next() + + return { + "op": "toggleAutoplay", + "status": check, + "guild_id": player.guild.id, + "requester_id": member.id + } + +async def closeConnection(player: Player, member: Member, data: Dict): + player._ipc_connection = False + +async def getPlaylists(member: Member, data: Dict): + playlists: Dict = await func.get_user(member.id, "playlist") + if not playlists: + return + + for pId, pList in playlists.copy().items(): + if "type" in pList: + if pList["type"] == "link": + tracks: Playlist = await NodePool.get_node().get_tracks(pList["uri"], requester=member) + if tracks: + playlists[pId]["tracks"] = [ track.track_id for track in tracks.tracks ] + + elif pList["type"] == "share": + playlist = await func.get_user(pList["user"], "playlist") + playlist = playlist.get(pList["referId"]) + if playlist: + if member.id not in playlist["perms"]["read"]: + await func.update_user(member.id, {"$unset": {f"playlist.{pId}": 1}}) + del playlists[pId] + continue + + if playlist['type'] == 'link': + tracks: Playlist = await NodePool.get_node().get_tracks(playlist["uri"], requester=member) + playlists[pId]["tracks"] = [ track.track_id for track in tracks.tracks ] + else: + playlists[pId]["tracks"] = playlist["tracks"] + + return { + "op": "getPlaylists", + "playlists": playlists, + "user_id": member.id + } + +async def removePlaylist(member: Member, data: Dict): + pId = data.get("pId") + isShare = data.get("isShare", False) + + if pId == 200: + return + + if isShare: + refer_user = data.get("refer_user") + await func.update_user(refer_user, {"$pull": {f"playlist.{pId}.perms.read": member.id}}) + + await func.update_user(member.id, {"$unset": {f'playlist.{pId}': 1}}) + +async def addPlaylistTrack(member: Member, data: Dict): + track_id = data.get("track_id") + pId = data.get("pId") + if not track_id or not pId: + return + + playlist: Dict = await func.get_user(member.id, 'playlist') + playlist = playlist.get(pId) + if not playlist: + return + + if playlist["type"] != "playlist": + return error_msg(func.get_lang(member.guild.id, 'playlistNotAllow'), user_id=member.id) + + rank, max_p, max_t = func.check_roles() + if len(playlist["tracks"]) >= max_t: + return error_msg(func.get_lang(member.guild.id, "playlistlimited").format(max_t), user_id=member.id) + + if track_id in playlist['tracks']: + return error_msg(func.get_lang(member.guild.id, "playlistrepeated"), user_id=member.id) + + await func.update_user(member.id, {"$push": {f'playlist.{pId}.tracks': track_id}}) + +async def removePlaylistTrack(member: Member, data: Dict): + track_id = data.get("track_id") + pId = data.get("pId") + if not track_id or not pId: + return + + await func.update_user(member.id, {"$pull": {f'playlist.{pId}.tracks': track_id }}) + +methods: Dict[str, SystemMethod] = { + "initUser": SystemMethod(initUser), + "closeConnection": PlayerMethod(closeConnection), + "initPlayer": PlayerMethod(initPlayer), + "skipTo": PlayerMethod(skipTo), + "backTo": PlayerMethod(backTo), + "moveTrack": PlayerMethod(moveTrack), + "addTracks": PlayerMethod(addTracks, auto_connect=True), + "getTracks": PlayerMethod(getTracks, auto_connect=True), + "shuffleTrack": PlayerMethod(shuffleTrack), + "repeatTrack": PlayerMethod(repeatTrack), + "removeTrack": PlayerMethod(removeTrack), + "updatePause": PlayerMethod(updatePause), + "updatePosition": PlayerMethod(updatePosition), + "toggleAutoplay": PlayerMethod(toggleAutoplay), + "getPlaylists": UserMethod(getPlaylists), + "removePlaylist": UserMethod(removePlaylist), + "addPlaylistTrack": UserMethod(addPlaylistTrack), + "removePlaylistTrack": UserMethod(removePlaylistTrack) +} + +async def process_methods(ipc_client, bot: commands.Bot, data: Dict) -> None: + op: str = data.get("op", "") + method = methods.get(op) + if not method or not (user_id := data.get("user_id")): + return + + try: + env: Dict = {} + args: List = [] + + if guild_id := data.get("guild_id"): + if not (guild := bot.get_guild(guild_id)): + return + env["guild"] = guild + + params = method.params + if "member" in params: + if not (guild := env.get("guild")) or not (member := guild.get_member(user_id)): + return + env["member"] = member + + if "player" in params: + if not (guild := env.get("guild")) or not (player := guild.voice_client): + if not method.auto_connect or not (member := env.get("member")): + return + player = await connect_channel(member, bot) + env["player"] = player + + if "bot" in params: + env["bot"] = bot + + if "data" in params: + env["data"] = data + + for param in params: + args.append(env.get(param)) + + if resp := await method.function(*args): + await ipc_client.send(resp) + + except Exception as e: + payload = { + "op": "errorMsg", + "level": "error", + "msg": str(e), + "user_id": user_id + } + await ipc_client.send(payload) \ No newline at end of file diff --git a/main.py b/main.py index 92a8dfe..b65a380 100644 --- a/main.py +++ b/main.py @@ -7,7 +7,7 @@ import logging import function as func from discord.ext import commands -from web import IPCServer +from ipc import IPCClient from motor.motor_asyncio import AsyncIOMotorClient from logging.handlers import TimedRotatingFileHandler from voicelink import VoicelinkException @@ -29,12 +29,7 @@ class Vocard(commands.Bot): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) - self.ipc = IPCServer( - self, - host=func.settings.ipc_server["host"], - port=func.settings.ipc_server["port"], - sercet_key=func.tokens.sercet_key - ) + self.ipc: IPCClient async def on_message(self, message: discord.Message, /) -> None: if message.author.bot or not message.guild: @@ -49,7 +44,7 @@ class Vocard(commands.Bot): await self.process_commands(message) async def connect_db(self) -> None: - if not ((db_name := func.tokens.mongodb_name) and (db_url := func.tokens.mongodb_url)): + if not ((db_name := func.settings.mongodb_name) and (db_url := func.settings.mongodb_url)): raise Exception("MONGODB_NAME and MONGODB_URL can't not be empty in settings.json") try: @@ -79,8 +74,17 @@ class Vocard(commands.Bot): except Exception as e: func.logger.error(f"Something went wrong while loading {module[:-3]} cog.", exc_info=e) - if func.settings.ipc_server.get("enable", False): - await self.ipc.start() + if func.settings.ipc_client.get("enable", False): + try: + self.ipc = IPCClient( + self, + host=func.settings.ipc_client["host"], + port=func.settings.ipc_client["port"], + password=func.settings.ipc_client["password"] + ) + await self.ipc.connect() + except Exception as e: + func.logger.error(f"Cannot connected to dashboard! - Reason: {e}") if not func.settings.version or func.settings.version != update.__version__: func.update_json("settings.json", new_data={"version": update.__version__}) diff --git a/voicelink/player.py b/voicelink/player.py index ce33dff..8448595 100644 --- a/voicelink/player.py +++ b/voicelink/player.py @@ -76,8 +76,7 @@ async def connect_channel(ctx: Union[commands.Context, Interaction], channel: Vo channel, ctx, settings )) - # if player.client.ipc._is_connected: - # await player.send_ws({"op": "createPlayer", "members_id": [member.id for member in channel.members]}) + await player.send_ws({"op": "createPlayer", "members_id": [member.id for member in channel.members]}) return player @@ -104,11 +103,13 @@ class Player(VoiceProtocol): ): self.client: Client = client self._bot: Client = client + self._ipc = self._bot.ipc + self._ipc_connection = False + self.context = ctx self.dj: Member = ctx.user if isinstance(ctx, Interaction) else ctx.author self.channel: VoiceChannel = channel self._guild = channel.guild if channel else None - self._ipc_connection: bool = False self.settings: dict = settings self.joinTime: float = round(time.time()) @@ -240,7 +241,7 @@ class Player(VoiceProtocol): @property def is_ipc_connected(self) -> bool: - return bool(self._ipc_connection and len(self.bot.ipc.connections)) + return self._ipc._is_connected and self._ipc_connection def is_user_join(self, user: Member): if user not in self.channel.members: @@ -278,6 +279,7 @@ class Player(VoiceProtocol): async def _dispatch_voice_update(self, voice_data: Dict[str, Any] = None): if {"sessionId", "event"} != self._voice_state.keys(): + self._logger.debug(f"Player in {self.guild.name}({self.guild.id}) dispatched voice update failed {voice_data}") return state = voice_data or self._voice_state @@ -548,7 +550,7 @@ class Player(VoiceProtocol): return self._current async def add_track(self, raw_tracks: Union[Track, List[Track]], *, at_font: bool = False, duplicate: bool = True) -> int: - tracks = [] + tracks: List[Track] = [] _duplicate_tracks = () if self.queue._allow_duplicate and duplicate else (track.uri for track in self.queue._queue) @@ -565,6 +567,7 @@ class Player(VoiceProtocol): position = self.queue.put_at_front(raw_tracks) if at_font else self.queue.put(raw_tracks) tracks.append(raw_tracks) + finally: if tracks: if self.is_ipc_connected: