From 37cdca09ecb7521591ff0fc3116927f1afdeb814 Mon Sep 17 00:00:00 2001 From: SeaswimmerTheFsh Date: Thu, 29 Feb 2024 22:15:44 -0500 Subject: [PATCH] fix(pterodactyl): pylint fixes --- pterodactyl/pterodactyl.py | 89 +++++++++++++++++++------------------- 1 file changed, 44 insertions(+), 45 deletions(-) diff --git a/pterodactyl/pterodactyl.py b/pterodactyl/pterodactyl.py index bd7b5c7..6993f8f 100644 --- a/pterodactyl/pterodactyl.py +++ b/pterodactyl/pterodactyl.py @@ -60,52 +60,52 @@ class Pterodactyl(commands.Cog): return self.logger.error('Failed to retrieve Pterodactyl websocket: %s', e) async with websockets.connect(websocket_credentials['data']['socket'], origin=base_url, ping_timeout=60) as websocket: - self.logger.info("WebSocket connection established") + self.logger.info("WebSocket connection established") - auth_message = json.dumps({"event": "auth", "args": [websocket_credentials['data']['token']]}) - await websocket.send(auth_message) - self.logger.debug("Authentication message sent") + auth_message = json.dumps({"event": "auth", "args": [websocket_credentials['data']['token']]}) + await websocket.send(auth_message) + self.logger.debug("Authentication message sent") - self.websocket = websocket - current_status = '' + self.websocket = websocket + current_status = '' - while True: - message = await websocket.recv() - if json.loads(message)['event'] in ['token expiring', 'token expired']: - self.logger.debug("Received token expiring/expired event. Refreshing token.") - websocket_credentials = client.servers.get_websocket(server_id) - auth_message = json.dumps({"event": "auth", "args": [websocket_credentials['data']['token']]}) - await websocket.send(auth_message) - self.logger.debug("Authentication message sent") + while True: + message = await websocket.recv() + if json.loads(message)['event'] in ('token expiring', 'token expired'): + self.logger.debug("Received token expiring/expired event. Refreshing token.") + websocket_credentials = client.servers.get_websocket(server_id) + auth_message = json.dumps({"event": "auth", "args": [websocket_credentials['data']['token']]}) + await websocket.send(auth_message) + self.logger.debug("Authentication message sent") - if json.loads(message)['event'] == 'auth success': - self.logger.info("WebSocket authentication successful") + if json.loads(message)['event'] == 'auth success': + self.logger.info("WebSocket authentication successful") - if json.loads(message)['event'] == 'console output' and await self.config.console_channel() is not None: - if current_status == 'running' or current_status == 'offline' or current_status == '': - content = self.remove_ansi_escape_codes(json.loads(message)['args'][0]) + if json.loads(message)['event'] == 'console output' and await self.config.console_channel() is not None: + if current_status in ('running', 'offline', ''): + content = self.remove_ansi_escape_codes(json.loads(message)['args'][0]) - channel = self.bot.get_channel(await self.config.console_channel()) - if channel is not None: - if content.startswith('['): - pagified_content = pagify(content, delims=[" ", "\n"]) - for page in pagified_content: - await channel.send(content=page) + channel = self.bot.get_channel(await self.config.console_channel()) + if channel is not None: + if content.startswith('['): + pagified_content = pagify(content, delims=[" ", "\n"]) + for page in pagified_content: + await channel.send(content=page) - chat_message = await self.check_if_chat_message(content) - if chat_message: - info = await self.get_info(chat_message['username']) - if info is not None: - await self.send_chat_discord(chat_message['username'], chat_message['message'], info['data']['player']['avatar']) - else: - await self.send_chat_discord(chat_message['username'], chat_message['message'], 'https://seafsh.cc/u/j3AzqQ.png') + chat_message = await self.check_if_chat_message(content) + if chat_message: + info = await self.get_info(chat_message['username']) + if info is not None: + await self.send_chat_discord(chat_message['username'], chat_message['message'], info['data']['player']['avatar']) + else: + await self.send_chat_discord(chat_message['username'], chat_message['message'], 'https://seafsh.cc/u/j3AzqQ.png') - if json.loads(message)['event'] == 'status': - current_status = json.loads(message)['args'][0] - if await self.config.console_channel() is not None: - console = self.bot.get_channel(await self.config.console_channel()) - if console is not None: - await console.send(f"Server status changed! `{json.loads(message)['args'][0]}`") + if json.loads(message)['event'] == 'status': + current_status = json.loads(message)['args'][0] + if await self.config.console_channel() is not None: + console = self.bot.get_channel(await self.config.console_channel()) + if console is not None: + await console.send(f"Server status changed! `{json.loads(message)['args'][0]}`") def remove_ansi_escape_codes(self, text: str) -> str: ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])') @@ -117,9 +117,9 @@ class Pterodactyl(commands.Cog): regex = await self.config.chat_regex() match: Optional[re.Match[str]] = re.match(regex, text) if match: - dict = {"time": match.group(1), "username": match.group(2), "message": match.group(3)} - self.logger.debug("Message is a chat message\n%s", json.dumps(dict)) - return dict + groups = {"time": match.group(1), "username": match.group(2), "message": match.group(3)} + self.logger.debug("Message is a chat message\n%s", json.dumps(groups)) + return groups self.logger.debug("Message is not a chat message") return False @@ -131,9 +131,8 @@ class Pterodactyl(commands.Cog): if response.status == 200: self.logger.debug("Player info retrieved for %s\n%s", username, json.dumps(await response.json())) return await response.json() - else: - self.logger.error("Failed to retrieve player info for %s: %s", username, response.status) - return None + self.logger.error("Failed to retrieve player info for %s: %s", username, response.status) + return None async def send_chat_discord(self, username: str, message: str, avatar_url: str) -> None: self.logger.debug("Sending chat message to Discord") @@ -165,7 +164,7 @@ class Pterodactyl(commands.Cog): fut.result() except asyncio.CancelledError: pass - except Exception as e: + except Exception as e: # pylint: disable=broad-exception-caught self.logger.error("WebSocket task has failed: %s", e, exc_info=e) self.task.cancel() self.task = self.get_task()