feat(repo): make all cogs pylance-typechecking compliant
Some checks failed
Actions / Lint Code (Ruff & Pylint) (push) Failing after 43s
Actions / Build Documentation (MkDocs) (push) Failing after 24s

at `basic` level, does not include Aurora as it's being rewritten in the `aurora/v3` branch
This commit is contained in:
cswimr 2025-02-01 16:57:45 +00:00
parent ea0b7937f8
commit 2a5b924409
Signed by: cswimr
GPG key ID: 0EC431A8DA8F8087
11 changed files with 184 additions and 139 deletions

View file

@ -2,7 +2,7 @@
import json
import re
from pathlib import Path
from typing import Optional, Tuple, Union
from typing import Any, Optional, Tuple, Union
import aiohttp
import discord
@ -56,7 +56,9 @@ async def establish_websocket_connection(coginstance: Pterodactyl) -> None:
content = mask_ip(content)
console_channel = coginstance.bot.get_channel(await config.console_channel())
assert isinstance(console_channel, discord.abc.Messageable)
chat_channel = coginstance.bot.get_channel(await config.chat_channel())
assert isinstance(chat_channel, discord.abc.Messageable)
if console_channel is not None:
if content.startswith("["):
pagified_content = pagify(content, delims=[" ", "\n"])
@ -83,7 +85,7 @@ async def establish_websocket_connection(coginstance: Pterodactyl) -> None:
embed, img = await generate_join_leave_embed(coginstance=coginstance, username=join_message, join=True)
if img:
with open(img, "rb") as file:
await chat_channel.send(embed=embed, file=file)
await chat_channel.send(embed=embed, file=discord.File(fp=file))
else:
await chat_channel.send(embed=embed)
else:
@ -96,7 +98,7 @@ async def establish_websocket_connection(coginstance: Pterodactyl) -> None:
embed, img = await generate_join_leave_embed(coginstance=coginstance, username=leave_message, join=False)
if img:
with open(img, "rb") as file:
await chat_channel.send(embed=embed, file=file)
await chat_channel.send(embed=embed, file=discord.File(fp=file))
else:
await chat_channel.send(embed=embed)
else:
@ -106,7 +108,11 @@ async def establish_websocket_connection(coginstance: Pterodactyl) -> None:
if achievement_message:
if chat_channel is not None:
if coginstance.bot.embed_requested(chat_channel):
await chat_channel.send(embed=await generate_achievement_embed(coginstance, achievement_message["username"], achievement_message["achievement"], achievement_message["challenge"]))
embed, img = await generate_achievement_embed(coginstance, achievement_message["username"], achievement_message["achievement"], achievement_message["challenge"])
if img:
await chat_channel.send(embed=embed, file=discord.File(fp=img))
else:
await chat_channel.send(embed=embed)
else:
await chat_channel.send(f"{achievement_message['username']} has {'completed the challenge' if achievement_message['challenge'] else 'made the advancement'} {achievement_message['achievement']}")
@ -130,24 +136,27 @@ async def establish_websocket_connection(coginstance: Pterodactyl) -> None:
await chat.send(await config.shutdown_msg())
async def retrieve_websocket_credentials(coginstance: Pterodactyl) -> Optional[dict]:
async def retrieve_websocket_credentials(coginstance: Pterodactyl) -> dict:
pterodactyl_keys = await coginstance.bot.get_shared_api_tokens("pterodactyl")
api_key = pterodactyl_keys.get("api_key")
if api_key is None:
coginstance.task.cancel()
coginstance.maybe_cancel_task()
raise ValueError("Pterodactyl API key not set. Please set it using `[p]set api`.")
base_url = await config.base_url()
if base_url is None:
coginstance.task.cancel()
coginstance.maybe_cancel_task()
raise ValueError("Pterodactyl base URL not set. Please set it using `[p]pterodactyl config url`.")
server_id = await config.server_id()
if server_id is None:
coginstance.task.cancel()
coginstance.maybe_cancel_task()
raise ValueError("Pterodactyl server ID not set. Please set it using `[p]pterodactyl config serverid`.")
client = PterodactylClient(base_url, api_key).client
coginstance.client = client
websocket_credentials = client.servers.get_websocket(server_id)
websocket_credentials: dict[str, Any] = client.servers.get_websocket(server_id).json()
if not websocket_credentials:
coginstance.maybe_cancel_task()
raise ValueError("Failed to retrieve websocket credentials. Please ensure the API details are correctly configured.")
logger.debug(
"""Websocket connection details retrieved:
Socket: %s
@ -165,44 +174,44 @@ def remove_ansi_escape_codes(text: str) -> str:
return ansi_escape.sub("", text)
async def check_if_server_message(text: str) -> Union[bool, str]:
async def check_if_server_message(text: str) -> Optional[str]:
regex = await config.server_regex()
match: Optional[re.Match[str]] = re.match(regex, text)
if match:
logger.trace("Message is a server message")
return match.group(1)
return False
return None
async def check_if_chat_message(text: str) -> Union[bool, dict]:
async def check_if_chat_message(text: str) -> Optional[dict]:
regex = await config.chat_regex()
match: Optional[re.Match[str]] = re.match(regex, text)
if match:
groups = {"username": match.group(1), "message": match.group(2)}
logger.trace("Message is a chat message\n%s", json.dumps(groups))
return groups
return False
return None
async def check_if_join_message(text: str) -> Union[bool, str]:
async def check_if_join_message(text: str) -> Optional[str]:
regex = await config.join_regex()
match: Optional[re.Match[str]] = re.match(regex, text)
if match:
logger.trace("Message is a join message")
return match.group(1)
return False
return None
async def check_if_leave_message(text: str) -> Union[bool, str]:
async def check_if_leave_message(text: str) -> Optional[str]:
regex = await config.leave_regex()
match: Optional[re.Match[str]] = re.match(regex, text)
if match:
logger.trace("Message is a leave message")
return match.group(1)
return False
return None
async def check_if_achievement_message(text: str) -> Union[bool, dict]:
async def check_if_achievement_message(text: str) -> Optional[dict]:
regex = await config.achievement_regex()
match: Optional[re.Match[str]] = re.match(regex, text)
if match:
@ -213,7 +222,7 @@ async def check_if_achievement_message(text: str) -> Union[bool, dict]:
groups["challenge"] = False
logger.trace("Message is an achievement message")
return groups
return False
return None
async def get_info(username: str) -> Optional[dict]: