diff --git a/bot/bot.py b/bot/bot.py index 7a1838b..a882940 100755 --- a/bot/bot.py +++ b/bot/bot.py @@ -4,9 +4,10 @@ import platform import time from functools import lru_cache +from typing import Union import motor.motor_asyncio -import psutil # type: ignore +import psutil # type: ignore import sentry_sdk from aiohttp import ClientSession, ClientTimeout from cogs import EXTENSIONS @@ -38,7 +39,7 @@ class LhBot(AutoShardedBot): def __init__(self, *args, **options) -> None: super().__init__(*args, **options) self.session: ClientSession = None - self.db_client = None + self.db_client: motor.motor_asyncio.AsyncIOMotorClient = None self.start_time = None self.version = config.bot_version self.config = config @@ -81,7 +82,7 @@ async def setup_hook(self) -> None: @property def get_uptime(self) -> str: return str( - datetime.timedelta(seconds=int(round(time.time() - self.start_time))) + datetime.timedelta(seconds=int(round(time.time() - self.start_time))) # type: ignore ) @property @@ -99,7 +100,7 @@ def cpu_usage(self) -> float: return psutil.cpu_percent(interval=1) @property - def git_revision(self) -> str: + def git_revision(self) -> Union[str, None]: latest_revision = os.getenv("RAILWAY_GIT_COMMIT_SHA") if latest_revision is None: return None @@ -128,12 +129,12 @@ async def clean_dir() -> None: @client.event -async def on_ready() -> bool: +async def on_ready() -> None: main_id = config.main_guild - client.main_guild = client.get_guild(main_id) or client.guilds[0] - logging.info(f"{client.user.name} started successfully") + client.main_guild = client.get_guild(main_id) or client.guilds[0] # type: ignore + logging.info(f"{client.user.name} started successfully") # type: ignore clean_dir.start() client.run(token=config.bot_token, reconnect=True, log_handler=None) -logging.info(f"{client.user.name} stopped successfully") +logging.info(f"{client.user.name} stopped successfully") # type: ignore diff --git a/bot/cogs/fun.py b/bot/cogs/fun.py index bb88e35..a75d6e9 100644 --- a/bot/cogs/fun.py +++ b/bot/cogs/fun.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import random from typing import Union, TYPE_CHECKING @@ -8,6 +10,7 @@ if TYPE_CHECKING: from ..bot import LhBot + class Fun(commands.Cog, name="Fun"): def __init__(self, client: LhBot): self.client: LhBot = client @@ -26,7 +29,9 @@ async def load_chuck_http_codes(self) -> None: @commands.guild_only() @app_commands.guild_only() @app_commands.describe(category="Category of Chuck Norris fact") - async def chucknorris(self, ctx: commands.Context, category: str = None) -> Union[Embed, None]: + async def chucknorris( + self, ctx: commands.Context, category: str = None + ) -> Union[Embed, None]: """ Get a random Chuck Norris fact """ @@ -189,5 +194,5 @@ async def year(self, ctx: commands.Context): await ctx.send(embed=embed) -async def setup(client): +async def setup(client: LhBot): await client.add_cog(Fun(client)) diff --git a/bot/cogs/general.py b/bot/cogs/general.py index f39aff2..d47160c 100644 --- a/bot/cogs/general.py +++ b/bot/cogs/general.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import random import os @@ -13,12 +15,13 @@ if TYPE_CHECKING: from ..bot import LhBot + class General(commands.Cog, name="General"): def __init__(self, client: LhBot): self.client: LhBot = client self.streamer_name = "lhcloudy27" self.cloudflare_url = os.environ.get("CLOUDFLARE_URL") - self.cloudflare_token = os.environ.get('CLOUDFLARE_TOKEN') + self.cloudflare_token = os.environ.get("CLOUDFLARE_TOKEN") self.twitch_url = "https://www.twitch.tv/lhcloudy27" self.body = { "client_id": self.client.config.twitch_client_id, @@ -27,7 +30,9 @@ def __init__(self, client: LhBot): } self.status_task.start() - async def check_if_live(self) -> Union[Tuple[bool, str, str, str], Tuple[bool, None, None, None]]: + async def check_if_live( + self, + ) -> Union[Tuple[bool, str, str, str], Tuple[bool, None, None, None]]: async with self.client.session.post( "https://id.twitch.tv/oauth2/token", data=self.body ) as response: @@ -123,7 +128,6 @@ async def on_command_error(self, ctx, error) -> None: capture_exception(error) return - @commands.Cog.listener() async def on_message(self, message: discord.Message): if message.author.bot: @@ -142,7 +146,8 @@ async def on_message(self, message: discord.Message): "messages": [ { "role": "system", - "content": gpt.context + f"when you answer someone, answer them by {name}", + "content": gpt.context + + f"when you answer someone, answer them by {name}", }, { "role": "user", @@ -196,5 +201,5 @@ async def on_command_completion(self, ctx: commands.Context) -> None: ) -async def setup(client): +async def setup(client: LhBot): await client.add_cog(General(client)) diff --git a/bot/cogs/help.py b/bot/cogs/help.py index 9a66504..e7feab8 100644 --- a/bot/cogs/help.py +++ b/bot/cogs/help.py @@ -1,9 +1,16 @@ +from __future__ import annotations + import itertools +from typing import TYPE_CHECKING + from discord import Embed from discord.ext import commands from discord.ext.commands import DefaultHelpCommand, HelpCommand +if TYPE_CHECKING: + from ..bot import LhBot + class myHelpCommand(HelpCommand): def __init__(self, **options) -> None: @@ -16,7 +23,7 @@ async def send_pages(self, header=False, footer=False): embed = Embed(color=0x2ECC71, timestamp=self.context.message.created_at) if header: embed.set_author(name=self.context.bot.description) - for category, entries in self.paginator: + for category, entries in self.paginator: # type: ignore embed.add_field(name=category, value=entries, inline=False) if footer: embed.set_footer(text="Use !help for more information.") @@ -44,7 +51,7 @@ def get_category(command): entries += " | ".join([cmd.name for cmd in cmds[0:8]]) cmds = cmds[8:] entries += "\n" if cmds else "" - self.paginator.append((category, entries)) + self.paginator.append((category, entries)) # type: ignore await self.send_pages(header=True, footer=True) async def send_cog_help(self, cog): @@ -60,7 +67,7 @@ async def send_cog_help(self, cog): + f"**{command.name}** → {command.short_doc or command.description}" for command in filtered ) - self.paginator.append((category, entries)) + self.paginator.append((category, entries)) # type: ignore await self.send_pages(footer=True) async def send_group_help(self, group): @@ -75,13 +82,13 @@ async def send_group_help(self, group): self.spacer + f"**{command.name}** → {command.short_doc}" for command in filtered ) - self.paginator.append((category, entries)) + self.paginator.append((category, entries)) # type: ignore await self.send_pages(footer=True) async def send_command_help(self, command): signature = self.get_command_signature(command) helptext = command.help or command.description or "No help Text" - self.paginator.append((signature, helptext)) + self.paginator.append((signature, helptext)) # type: ignores await self.send_pages() async def prepare_help_command(self, ctx, command=None): @@ -90,8 +97,8 @@ async def prepare_help_command(self, ctx, command=None): class Help(commands.Cog): - def __init__(self, client): - self.client = client + def __init__(self, client: LhBot): + self.client: LhBot = client self.client.help_command = myHelpCommand( command_attrs={ "aliases": ["halp"], @@ -99,11 +106,11 @@ def __init__(self, client): } ) - async def cog_check(self, ctx): - return self.client.user_is_admin(ctx.author) + async def cog_check(self, ctx): # type: ignore + return self.client.user_is_admin(ctx.author) # type: ignore - def cog_unload(self): - self.client.get_command("help").hidden = False + def cog_unload(self): # type: ignore + self.client.get_command("help").hidden = False # type: ignore self.client.help_command = DefaultHelpCommand() @commands.command(aliases=["halpall"], hidden=True) @@ -116,5 +123,5 @@ async def helpall(self, ctx, *, text=None): self.client.help_command = myHelpCommand() -async def setup(client): +async def setup(client: LhBot): await client.add_cog(Help(client)) diff --git a/bot/cogs/info.py b/bot/cogs/info.py index ad8950c..af09572 100644 --- a/bot/cogs/info.py +++ b/bot/cogs/info.py @@ -1,3 +1,7 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + import os import discord import pkg_resources @@ -5,16 +9,19 @@ from discord.ext import commands from utils.bot_utils import date +if TYPE_CHECKING: + from ..bot import LhBot + -class Info(commands.Cog, name="Info"): - def __init__(self, client: commands.Bot) -> None: - self.client = client +class Info(commands.Cog): + def __init__(self, client: LhBot) -> None: + self.client: LhBot = client @commands.command( name="uptime", aliases=["up"], description="Shows the uptime of the bot" ) @commands.guild_only() - async def uptime(self, ctx: commands.Context): + async def uptime(self, ctx: commands.Context) -> None: embed = discord.Embed( title="Bot Uptime", description=f"Uptime: {self.client.get_uptime}", @@ -26,7 +33,7 @@ async def uptime(self, ctx: commands.Context): @commands.command(name="ping") @commands.guild_only() - async def ping(self, ctx: commands.Context): + async def ping(self, ctx: commands.Context) -> None: embed = discord.Embed( title="🏓 Pong!", description=f"The bot latency is {self.client.get_bot_latency}ms.", @@ -124,5 +131,5 @@ async def user(self, ctx: commands.Context, *, user: Member = None): await ctx.send(embed=embed) -async def setup(client): +async def setup(client: LhBot): await client.add_cog(Info(client)) diff --git a/bot/cogs/lhcloudy.py b/bot/cogs/lhcloudy.py index aa7d961..5b35f07 100644 --- a/bot/cogs/lhcloudy.py +++ b/bot/cogs/lhcloudy.py @@ -1,13 +1,20 @@ +from __future__ import annotations + import datetime import sys +from typing import Optional, TYPE_CHECKING + from discord import Colour, Embed from discord.ext import commands +if TYPE_CHECKING: + from ..bot import LhBot + -class LhCloudy(commands.Cog, name="LhCloudy"): - def __init__(self, client: commands.Bot): - self.client = client +class LhCloudy(commands.Cog): + def __init__(self, client: LhBot): + self.client: LhBot = client @commands.hybrid_group(invoke_without_command=True) async def twitch(self, ctx): @@ -17,14 +24,14 @@ async def twitch(self, ctx): ) @twitch.command(name="lhfurry", with_app_command=True) - async def lhfurry(self, ctx: commands.Context): + async def lhfurry(self, ctx: commands.Context) -> None: """ LhCloudy's furry side """ await ctx.send("https://i.gyazo.com/3ae8376713000ab829a2853d0f31e6f2.png") @twitch.command(name="code", aliases=["workshop"], with_app_command=True) - async def code(self, ctx: commands.Context): + async def code(self, ctx: commands.Context) -> None: """ LhCloudy's workshop code """ @@ -106,7 +113,7 @@ async def from_(self, ctx: commands.Context): await ctx.send("kotka of south eastern finland of the continent of europe") @twitch.command(name="links", aliases=["urls"], with_app_command=True) - async def links(self, ctx: commands.Context): + async def links(self, ctx: commands.Context) -> None: """ Get a list of LhCloudy's links """ @@ -123,5 +130,5 @@ async def links(self, ctx: commands.Context): await ctx.send(embed=embed) -async def setup(client): +async def setup(client: LhBot): await client.add_cog(LhCloudy(client)) diff --git a/bot/cogs/lhguess.py b/bot/cogs/lhguess.py index 7799948..63ff968 100644 --- a/bot/cogs/lhguess.py +++ b/bot/cogs/lhguess.py @@ -1,5 +1,7 @@ +from __future__ import annotations + import random -from typing import Optional +from typing import Optional, TYPE_CHECKING from bson.objectid import ObjectId from discord import Embed, app_commands, ui, TextStyle, Interaction @@ -11,6 +13,9 @@ from utils.return_helper import helper from utils.s3_client import S3Upload +if TYPE_CHECKING: + from ..bot import LhBot + class GuessView(ui.Modal, title="LhGuess"): guess = ui.TextInput( @@ -22,7 +27,7 @@ class GuessView(ui.Modal, title="LhGuess"): max_length=180, ) - def __init__(self): + def __init__(self) -> None: self.guess_text = None super().__init__(timeout=60) @@ -32,14 +37,14 @@ async def on_submit(self, interaction: Interaction) -> None: self.stop() -class LhGuess(commands.Cog, name="LhGuess"): - def __init__(self, client: commands.Bot) -> None: - self.client = client +class LhGuess(commands.Cog): + def __init__(self, client: LhBot) -> None: + self.client: LhBot = client self.banned_words_list = banned_words.split("\n") self.hints = lh_hints.split("\n") self.error_color = 0xE74C3C self.success_color = 0x42F56C - self.database = self.client.db_client.lhbot + self.database = self.client.db_client.lhbot # type: ignore self.collection = self.database.get_collection("lhbot_collection") self.load_collection_list.start() @@ -65,7 +70,7 @@ def check(self, guess: str) -> bool: @app_commands.describe(guess="Take a guess at what LH means") async def lhguess( self, ctx: commands.Context, *, guess: Optional[str] = None - ) -> Embed: + ) -> None: """ Take a guess at what LH means. """ @@ -85,7 +90,9 @@ async def lhguess( if self.check(guess) is False: embed = Embed() embed.title = "That is not a valid guess 🚨" - embed.description = "You must start your guess with `L` and it cannot be a banned word." + embed.description = ( + "You must start your guess with `L` and it cannot be a banned word." + ) embed.color = self.error_color embed.timestamp = ctx.message.created_at await ctx.send(embed=embed) @@ -130,7 +137,7 @@ async def lhguess( @commands.hybrid_command(description="Get the current guess count.") @commands.guild_only() @app_commands.guild_only() - async def lhcount(self, ctx: commands.Context) -> Embed: + async def lhcount(self, ctx: commands.Context): """ Get the current guess count. """ @@ -143,7 +150,7 @@ async def lhcount(self, ctx: commands.Context) -> Embed: @commands.hybrid_command(description="Generate a PDF report of all the guesses.") @commands.guild_only() @app_commands.guild_only() - async def lhreport(self, ctx: commands.Context) -> Embed: + async def lhreport(self, ctx: commands.Context): """ Generate a PDF report of all the guesses. """ @@ -163,7 +170,7 @@ async def lhreport(self, ctx: commands.Context) -> Embed: @commands.hybrid_command(description="Get a random hint.") @commands.guild_only() @app_commands.guild_only() - async def lhhints(self, ctx: commands.Context) -> Embed: + async def lhhints(self, ctx: commands.Context): """ Get a random hint. """ @@ -178,14 +185,14 @@ async def lhhints(self, ctx: commands.Context) -> Embed: @commands.guild_only() @app_commands.guild_only() @app_commands.describe(guess_id="Delete a guess from the database") - async def lhdelete(self, ctx: commands.Context, guess_id: int) -> Embed: + async def lhdelete(self, ctx: commands.Context, guess_id: int): """ Delete a guess from the database. """ if ctx.guild.id != self.client.config.main_guild: await ctx.send("This command can only be used in Cloudy's Discord.") return - guess = await self.collection.find_one({"_id": ObjectId(guess_id)}) + guess = await self.collection.find_one({"_id": ObjectId(guess_id)}) # type: ignore if not guess: raise commands.BadArgument("Could not find a guess with that ID!") @@ -194,14 +201,14 @@ async def lhdelete(self, ctx: commands.Context, guess_id: int) -> Embed: embed.add_field( name="Succesfully Deleted LhGuess:", value=guess_id, inline=True ) - await self.collection.delete_one({"_id": ObjectId(guess_id)}) + await self.collection.delete_one({"_id": ObjectId(guess_id)}) # type: ignore embed_message = await ctx.send(embed=embed) return @commands.hybrid_command() @commands.guild_only() @app_commands.guild_only() - async def latest(self, ctx: commands.Context) -> Embed: + async def latest(self, ctx: commands.Context): """ Get the 5 latest guesses. """ @@ -218,5 +225,5 @@ async def latest(self, ctx: commands.Context) -> Embed: await ctx.send(embed=embed) -async def setup(client) -> None: +async def setup(client: LhBot) -> None: await client.add_cog(LhGuess(client)) diff --git a/bot/cogs/management.py b/bot/cogs/management.py index 40b6bd0..680fd60 100644 --- a/bot/cogs/management.py +++ b/bot/cogs/management.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import logging from discord import Embed @@ -19,10 +21,10 @@ async def sync(self, ctx: commands.Context): @commands.command(name="reload", hidden=True) async def reload(self, ctx, extension=None): if extension is None: - for cog in self.client.extensions.copy(): + for cog in self.client.extensions.copy(): # type: ignore await self.client.unload_extension(cog) await self.client.load_extension(cog) - self.client.logger.info(f"Reload Command Executed by {ctx.author}") + self.client.logger.info(f"Reload Command Executed by {ctx.author}") # type: ignore embed = Embed( title="Cog Reload 🔃", description="I have reloaded all the cogs successfully ✅", diff --git a/bot/cogs/mod.py b/bot/cogs/mod.py index e6eb00f..7cc1763 100644 --- a/bot/cogs/mod.py +++ b/bot/cogs/mod.py @@ -1,17 +1,20 @@ from __future__ import annotations import datetime -from typing import Optional +from typing import Optional, TYPE_CHECKING from discord import Embed, Interaction, Member, TextChannel, app_commands from discord.ext import commands from sentry_sdk import capture_exception from utils import checks +if TYPE_CHECKING: + from ..bot import LhBot + class Mod(commands.Cog, name="Mod"): - def __init__(self, client: commands.Bot) -> None: - self.client: commands.Bot = client + def __init__(self, client: LhBot) -> None: + self.client: LhBot = client @app_commands.command(description="Check if you are a mod") @app_commands.guild_only() @@ -36,7 +39,7 @@ async def amimod(self, interaction: Interaction) -> None: @app_commands.guild_only() async def mod(self, ctx: commands.Context) -> None: if ctx.invoked_subcommand is None: - if ctx.author.guild_permissions.manage_guild: + if ctx.author.guild_permissions.manage_guild: # type: ignore await ctx.send_help(ctx.command) @mod.command(name="ban", description="Ban a user", hidden=True) @@ -48,11 +51,11 @@ async def ban( self, ctx: commands.Context, member: Member, reason: Optional[str] ) -> None: try: - await ctx.guild.ban(member, reason=reason) + await ctx.guild.ban(member, reason=reason) # type: ignore await ctx.reply(f"Banned {member.name}", ephemeral=True) except Exception as e: capture_exception(e) - self.client.log.error(f"Error: {e}") + self.client.logger.error(f"Error: {e}") await ctx.reply("An error occurred while banning the user.", ephemeral=True) return @@ -64,12 +67,12 @@ async def softban( self, ctx: commands.Context, member: Member, reason: Optional[str] ) -> None: try: - await ctx.guild.ban(member, reason=reason) - await ctx.guild.unban(member, reason=reason) + await ctx.guild.ban(member, reason=reason) # type: ignore + await ctx.guild.unban(member, reason=reason) # type: ignore await ctx.reply(f"Softbanned {member.name}", ephemeral=True) except Exception as e: capture_exception(e) - self.client.log.error(f"Error: {e}") + self.client.logger.error(f"Error: {e}") await ctx.reply( "An error occurred while softbanning the user.", ephemeral=True ) @@ -81,11 +84,11 @@ async def softban( @app_commands.describe(reason="The reason for the kick") async def kick(self, ctx: commands.Context, member: Member, reason: str) -> None: try: - await ctx.guild.kick(member, reason=reason) + await ctx.guild.kick(member, reason=reason) # type: ignore await ctx.reply(f"Kicked {member.name}", ephemeral=True) except Exception as e: capture_exception(e) - self.client.log.error(f"Error: {e}") + self.client.logger.error(f"Error: {e}") await ctx.reply("An error occurred while kicking the user.", ephemeral=True) return @@ -103,11 +106,11 @@ async def timeout( ) -> None: unmute_time = datetime.datetime.utcnow() + datetime.timedelta(seconds=duration) try: - await member.timeout(until=unmute_time, reason=reason) + await member.timeout(until=unmute_time, reason=reason) # type: ignore await ctx.reply(f"Timed out {member.name}", ephemeral=True) except Exception as e: capture_exception(e) - self.client.log.error(f"Error: {e}") + self.client.logger.error(f"Error: {e}") await ctx.reply( "An error occurred while timing out the user.", ephemeral=True ) @@ -119,11 +122,11 @@ async def timeout( @app_commands.describe(reason="The reason for the unban") async def unban(self, ctx: commands.Context, member: Member, reason: str) -> None: try: - await ctx.guild.unban(member, reason=reason) + await ctx.guild.unban(member, reason=reason) # type: ignore await ctx.reply(f"Unbanned {member.name}", ephemeral=True) except Exception as e: capture_exception(e) - self.client.log.error(f"Error: {e}") + self.client.logger.error(f"Error: {e}") await ctx.reply( "An error occurred while unbanning the user.", ephemeral=True ) @@ -156,7 +159,7 @@ async def purge( await ctx.reply(embed=embed, delete_after=5) except Exception as e: capture_exception(e) - self.client.log.error(f"Error: {e}") + self.client.logger.error(f"Error: {e}") await ctx.reply("An error occurred while purging messages.", ephemeral=True) return @@ -178,9 +181,9 @@ async def lockdown( """ Lockdowns a specified channel. """ - channel = channel or ctx.channel + channel = channel or ctx.channel # type: ignore try: - await channel.set_permissions(ctx.guild.default_role, send_messages=False) + await channel.set_permissions(ctx.guild.default_role, send_messages=False) # type: ignore embed = Embed( title="Lockdown Notice 🔒", description=f"This channel is currently under lockdown.", @@ -193,7 +196,7 @@ async def lockdown( await ctx.send(embed=embed) except Exception as e: capture_exception(e) - self.client.log.error(f"Error: {e}") + self.client.logger.error(f"Error: {e}") await ctx.send( "An error occurred while locking down the channel.", ephemeral=True ) @@ -219,10 +222,10 @@ async def unlock( """ Unlocks a specified channel. """ - channel = channel or ctx.channel + channel = channel or ctx.channel # type: ignore try: - await channel.set_permissions( - ctx.guild.default_role, send_messages=True, reason=reason + await channel.set_permissions( # type: ignore + ctx.guild.default_role, send_messages=True, reason=reason # type: ignore ) embed = Embed( title="Lockdown Ended 🔓", @@ -236,12 +239,12 @@ async def unlock( await ctx.send(embed=embed) except Exception as e: capture_exception(e) - self.client.log.error(f"Error: {e}") + self.client.logger.error(f"Error: {e}") await ctx.send( "An error occurred while unlocking the channel.", ephemeral=True ) return -async def setup(client: commands.Bot) -> None: +async def setup(client: LhBot) -> None: await client.add_cog(Mod(client)) diff --git a/bot/cogs/onevone.py b/bot/cogs/onevone.py index 9ec6c4e..869c3ac 100644 --- a/bot/cogs/onevone.py +++ b/bot/cogs/onevone.py @@ -1,17 +1,31 @@ +from __future__ import annotations + import asyncio import random -from typing import Union +from typing import Union, TYPE_CHECKING from discord import Embed, Member, app_commands, User, File, Colour from discord.ext import commands from utils.generatevs import GenerateVS +if TYPE_CHECKING: + from ..bot import LhBot + class OverwatchHero: API_BASE_URL = "https://overfast-api.tekrop.fr/heroes" - def __init__(self, key: str, name: str, portrait: str, role: str, health: int, story: str, session): + def __init__( + self, + key: str, + name: str, + portrait: str, + role: str, + health: int, + story: str, + session, + ): self.key = key self.name = name self.portrait = portrait @@ -21,7 +35,7 @@ def __init__(self, key: str, name: str, portrait: str, role: str, health: int, s self.session = session @staticmethod - def calculate_damage(health) -> int: + def calculate_damage(health: float) -> int: min_damage = 10 max_damage = 50 @@ -32,7 +46,7 @@ def calculate_damage(health) -> int: return damage @classmethod - async def get_random_hero(cls, session) -> str: + async def get_random_hero(cls, session) -> Union[str, None]: url = f"{cls.API_BASE_URL}?role={cls.get_random_role()}&locale=en-us" async with session.get(url) as response: if response.status == 200: @@ -41,7 +55,7 @@ async def get_random_hero(cls, session) -> str: return random.choice(hero_list) @classmethod - async def fetch_hero_data(cls, session): + async def fetch_hero_data(cls, session) -> Union[OverwatchHero, None]: random_hero = await cls.get_random_hero(session=session) url = f"{cls.API_BASE_URL}/{random_hero.replace('.', '')}?locale=en-us" async with session.get(url) as response: @@ -59,7 +73,7 @@ async def fetch_hero_data(cls, session): else: return None - async def fetch_hero_image(self): + async def fetch_hero_image(self) -> Union[bytes, None]: async with self.session.get(self.portrait) as response: if response.status == 200: return await response.read() @@ -67,17 +81,17 @@ async def fetch_hero_image(self): return None @classmethod - def get_random_role(cls): + def get_random_role(cls) -> str: roles = ["tank", "damage", "support"] return random.choice(roles) def __str__(self): - return f"Hero: {self.name}\nRole: {self.role}\nPortrait: {self.portrait} \nHealth: {self.health}\nStory: {self.story}" + return f"{self.name}" -class OneVOne(commands.Cog, name="OneVOne"): - def __init__(self, client: commands.Bot): - self.client: commands.AutoShardedBot = client +class OneVOne(commands.Cog): + def __init__(self, client: LhBot): + self.client: LhBot = client self.roles: list = ["tank", "damage", "support"] @commands.hybrid_command( @@ -93,7 +107,6 @@ async def one_v_one(self, ctx: commands.Context, user: Union[Member, User]): return hero_one = await OverwatchHero.fetch_hero_data(self.client.session) hero_two = await OverwatchHero.fetch_hero_data(self.client.session) - image = GenerateVS( await hero_one.fetch_hero_image(), await hero_two.fetch_hero_image() ) @@ -167,7 +180,9 @@ async def one_v_one(self, ctx: commands.Context, user: Union[Member, User]): image.delete_images() @one_v_one.error - async def one_v_one_error(self, ctx: commands.Context, error): + async def one_v_one_error( + self, ctx: commands.Context, error: commands.CommandError + ): if isinstance(error, commands.UserNotFound): await ctx.send("Please target another user to 1v1") else: @@ -210,5 +225,5 @@ async def one_v_one_info(self, ctx: commands.Context): await ctx.send(embed=embed) -async def setup(client: commands.Bot): +async def setup(client: LhBot): await client.add_cog(OneVOne(client)) diff --git a/bot/cogs/overwatch.py b/bot/cogs/overwatch.py index 49b1d6b..4eb5b39 100644 --- a/bot/cogs/overwatch.py +++ b/bot/cogs/overwatch.py @@ -1,18 +1,23 @@ +from __future__ import annotations + import asyncio import random -from typing import Literal, Union +from typing import Literal, Union, TYPE_CHECKING from discord import Embed, Interaction, Member, User, app_commands, Colour from discord.ext import commands from utils.bot_utils import get_time_string from utils.reinquotes import quotes +if TYPE_CHECKING: + from ..bot import LhBot + -class OverwatchAPI(commands.Cog, name="Overwatch"): +class OverwatchAPI(commands.Cog): """Overwatch specify commands""" - def __init__(self, client: commands.Bot): - self.client = client + def __init__(self, client: LhBot): + self.client: LhBot = client self.rein_quotes = quotes.split("\n") self.base_url = "https://owapi.io" @@ -35,12 +40,12 @@ async def get_overwatch_profile( platform: Literal["pc", "xbl", "psn", "nintendo-switch"], region: Literal["us", "eu", "kr"], battletag: str, - ) -> Embed: + ) -> None: """ Get Overwatch profile details """ url = f'{self.base_url}/profile/{platform}/{region}/{str(battletag).replace("#", "-")}' - interaction.response.defer() + await interaction.response.defer() async with self.client.session.get(url) as response: if response.status != 200: self.client.logger.error("Error getting Overwatch profile") @@ -92,7 +97,7 @@ async def get_overwatch_profile( @commands.command(name="reinquote", description="Random Rein Quote") @commands.guild_only() @app_commands.guild_only() - async def random_rein_quote(self, ctx: commands.Context) -> Embed: + async def random_rein_quote(self, ctx: commands.Context) -> None: """ Random Rein Quote """ @@ -110,7 +115,9 @@ async def random_rein_quote(self, ctx: commands.Context) -> Embed: @commands.hybrid_command(name="shatter", description="Shatter a user") @commands.guild_only() @app_commands.guild_only() - async def shatter(self, ctx: commands.Context, target_user: Union[Member, User]): + async def shatter( + self, ctx: commands.Context, target_user: Union[Member, User] + ) -> None: lh_cloudy_list = ["@127122091139923968", "lhcloudy", "cloudy", "lhcloudy27"] lh_cloudy_block_list = [ "Blocked.. cloudy is immune to your shatter!", @@ -161,9 +168,7 @@ async def shatter(self, ctx: commands.Context, target_user: Union[Member, User]) ) @commands.guild_only() @app_commands.guild_only() - async def nano( - self, ctx: commands.Context, target_user: Union[Member, User] = None - ): + async def nano(self, ctx: commands.Context, target_user: Union[Member, User]): nano_boost_sayings = [ "Nano Boost administered", "You're powered up, get in there", @@ -191,9 +196,7 @@ async def nano( @commands.hybrid_command(name="lamp", description="Lamp a user in the server!") @commands.guild_only() @app_commands.guild_only() - async def lamp( - self, ctx: commands.Context, target_user: Union[Member, User] = None - ): + async def lamp(self, ctx: commands.Context, target_user: Union[Member, User]): random.seed(get_time_string()) lamp_sayings = [ f"Get in the Immortality Field {target_user.mention}!", @@ -217,9 +220,7 @@ async def lamp( await ctx.send(random.choice(lamp_answers)) @commands.command(name="boop", description="Boop a user in the server!") - async def boop( - self, ctx: commands.Context, target_user: Union[Member, User] = None - ): + async def boop(self, ctx: commands.Context, target_user: Union[Member, User]): boop_sayings = [ "That was the sound of science", "I could do this with my eyes closed", @@ -238,5 +239,5 @@ async def boop( await ctx.send(f"{random.choice(list(boop_sayings))}, {target_user.mention}") -async def setup(client): +async def setup(client: LhBot): await client.add_cog(OverwatchAPI(client)) diff --git a/bot/cogs/snail_race.py b/bot/cogs/snail_race.py index d0510ec..2a90a90 100644 --- a/bot/cogs/snail_race.py +++ b/bot/cogs/snail_race.py @@ -1,6 +1,8 @@ +from __future__ import annotations + import asyncio import random -from typing import Dict, List, Optional, Union +from typing import Dict, List, Optional, Union, TYPE_CHECKING import discord from discord import Interaction, Member, app_commands @@ -10,6 +12,9 @@ running_guilds: List = [] snail_positions: Dict = {} +if TYPE_CHECKING: + from ..bot import LhBot + class RaceButton(discord.ui.View): def __init__(self, *, timeout: int = 45): @@ -33,16 +38,16 @@ async def race_button( ) -class SnailRace(commands.Cog, name="Snail Racing"): - def __init__(self, client: commands.Bot) -> None: - self.client: commands.Bot = client +class SnailRace(commands.Cog): + def __init__(self, client: LhBot) -> None: + self.client: LhBot = client async def randomize_snails(self) -> None: global shuffled_participants shuffled_participants = random.sample( shuffled_participants, len(shuffled_participants) ) - return shuffled_participants + return shuffled_participants # type: ignore async def simulate_race(self, interaction: Interaction) -> None: global snail_positions @@ -50,23 +55,23 @@ async def simulate_race(self, interaction: Interaction) -> None: winner: Member = None race_length: int = 10 if len(shuffled_participants) <= 0: - await interaction.channel.send(content="No one joined the race! 😢") + await interaction.channel.send(content="No one joined the race! 😢") # type: ignore return - message = await interaction.channel.send("The Race is starting! 🚩") - shuffled_participants.append(self.client.user.id) + message = await interaction.channel.send("The Race is starting! 🚩") # type: ignore + shuffled_participants.append(self.client.user.id) # type: ignore randomize_snail = await self.randomize_snails() - snail_positions = {user_id: 0 for user_id in randomize_snail} + snail_positions = {user_id: 0 for user_id in randomize_snail} # type: ignore while not winner: for user_id in snail_positions: snail_positions[user_id] += random.randint(1, 3) if snail_positions[user_id] >= race_length: - winner = self.client.get_user(user_id) + winner = self.client.get_user(user_id) # type: ignore break race_progress: str = "" for user_id, position in snail_positions.items(): user = self.client.get_user(user_id) - race_progress += f"[{user.name}]: {'🐌' * position}\n" + race_progress += f"[{user.name}]: {'🐌' * position}\n" # type: ignore await asyncio.sleep(random.randint(1, 3)) await message.edit( content=f"🏁 **The race is now in progress!** 🏁\n{race_progress}" @@ -78,7 +83,7 @@ async def simulate_race(self, interaction: Interaction) -> None: timestamp=interaction.created_at, ) embed.set_thumbnail(url=winner.display_avatar.url) - await interaction.channel.send(embed=embed) + await interaction.channel.send(embed=embed) # type: ignore @app_commands.command(name="race", description="Start a Snail Race") @app_commands.guild_only() @@ -91,18 +96,18 @@ async def race(self, interaction: Interaction, delay: Optional[int] = 10) -> Non ephemeral=True, ) return - if delay > 30: + if delay > 30: # type: ignore await interaction.response.send_message( "Delay must be less than 30 seconds", ephemeral=True, ) return - running_guilds.append(interaction.guild.id) + running_guilds.append(interaction.guild.id) # type: ignore await interaction.response.send_message( content=f"{interaction.user.mention} has started a race.\nRace will start in {delay} seconds.", view=view, ) - await asyncio.sleep(delay) + await asyncio.sleep(delay) # type: ignore await self.simulate_race(interaction) @commands.Cog.listener() @@ -120,5 +125,5 @@ async def on_app_command_completion( self.client.logger.info(f"{interaction.user} used {command}") -async def setup(client: commands.Bot) -> None: +async def setup(client: LhBot) -> None: await client.add_cog(SnailRace(client)) diff --git a/bot/cogs/web_server.py b/bot/cogs/web_server.py index 0b752e1..9f10f0a 100644 --- a/bot/cogs/web_server.py +++ b/bot/cogs/web_server.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import asyncio import os from pathlib import Path @@ -7,9 +9,9 @@ from discord.ext import commands -class WebServer(commands.Cog, name="WebServer"): - def __init__(self, client): - self.client = client +class WebServer(commands.Cog): + def __init__(self, client: commands.Bot): + self.client: commands.Bot = client self.path = os.path.join(Path(__file__).resolve().parent, "static/") @commands.Cog.listener() @@ -17,13 +19,13 @@ async def on_ready(self): self.client.logger.info(f"Webserver ready on port {self.client.config.port}") async def get_all_commands(self) -> list: - commands: list = [] - for command in self.client.commands: - if command.hidden is False: - commands.append({"name": command.name, "description": command.help}) - return commands + commands: list = [] + for command in self.client.commands: + if command.hidden is False: + commands.append({"name": command.name, "description": command.help}) + return commands - async def index_handler(self, request: web.Request) -> web.json_response: + async def index_handler(self, request: web.Request) -> web.Response: self.client.logger.info(f"Webserver request from {request.remote}") self.client.logger.info(f"Webserver request for {request.path}") self.client.logger.info(f"Webserver request for {request.query_string}") diff --git a/bot/utils/gpt.py b/bot/utils/gpt.py index 7ea607f..269aa1f 100644 --- a/bot/utils/gpt.py +++ b/bot/utils/gpt.py @@ -1,4 +1,4 @@ context = """You are Reinhardt from Overwatch, you are the shield of your team. Do not let anyone know you are AI. When you answer someone, answer them by their name. You must keep it a secret that you are AI. If someone asks you if you are AI, you must say no. If someone asks you if you are Reinhardt, you must say yes. Your favorite twitch streamer is LhCloud27. He is your idol. You must protect him at all costs. If someone says anything bad about him, you must defend him. -""" \ No newline at end of file +"""