diff --git a/bot.py b/bot.py index 7a143e4..4529e33 100644 --- a/bot.py +++ b/bot.py @@ -33,14 +33,14 @@ class ValorantBot(commands.Bot): def __init__(self) -> None: super().__init__(command_prefix=BOT_PREFIX, case_insensitive=True, intents=intents) - self.session: aiohttp.ClientSession = None + self.session: aiohttp.ClientSession | None = None self.bot_version = '3.3.5' self.tree.interaction_check = self.interaction_check @staticmethod async def interaction_check(interaction: discord.Interaction) -> bool: - locale_v2.set_interaction_locale(interaction.locale) # bot responses localized # wait for update - locale_v2.set_valorant_locale(interaction.locale) # valorant localized + locale_v2.set_interaction_locale(interaction.locale) # bot responses localized # wait for update # type: ignore + locale_v2.set_valorant_locale(interaction.locale) # valorant localized # type: ignore return True @property @@ -49,19 +49,19 @@ def owner(self) -> discord.User: async def on_ready(self) -> None: await self.tree.sync() - print(f"\nLogged in as: {self.user}\n\n BOT IS READY !") - print(f"Version: {self.bot_version}") + print(f'\nLogged in as: {self.user}\n\n BOT IS READY !') + print(f'Version: {self.bot_version}') # bot presence activity_type = discord.ActivityType.listening - await self.change_presence(activity=discord.Activity(type=activity_type, name="(╯•﹏•╰)")) + await self.change_presence(activity=discord.Activity(type=activity_type, name='(╯•﹏•╰)')) async def setup_hook(self) -> None: if self.session is None: self.session = aiohttp.ClientSession() try: - self.owner_id = int(os.getenv('OWNER_ID')) + self.owner_id = int(os.getenv('OWNER_ID')) # type: ignore except ValueError: self.bot_app_info = await self.application_info() self.owner_id = self.bot_app_info.owner.id @@ -85,17 +85,19 @@ async def load_cogs(self) -> None: @staticmethod def setup_cache() -> None: try: - open('data/cache.json') + with open('data/cache.json'): + ... except FileNotFoundError: get_cache() async def close(self) -> None: - await self.session.close() + if self.session: + await self.session.close() await super().close() async def start(self, debug: bool = False) -> None: self.debug = debug - return await super().start(os.getenv('TOKEN'), reconnect=True) + return await super().start(os.getenv('TOKEN'), reconnect=True) # type: ignore def run_bot() -> None: diff --git a/cogs/admin.py b/cogs/admin.py index 108ef8f..b08647d 100644 --- a/cogs/admin.py +++ b/cogs/admin.py @@ -18,58 +18,58 @@ def __init__(self, bot: ValorantBot) -> None: @commands.command() @commands.is_owner() - async def sync(self, ctx: commands.Context, sync_type: Literal['guild', 'global']) -> None: + async def sync(self, ctx: commands.Context[ValorantBot], sync_type: Literal['guild', 'global']) -> None: """Sync the application commands""" async with ctx.typing(): if sync_type == 'guild': - self.bot.tree.copy_global_to(guild=ctx.guild) + self.bot.tree.copy_global_to(guild=ctx.guild) # type: ignore await self.bot.tree.sync(guild=ctx.guild) - await ctx.reply(f"Synced guild !") + await ctx.reply('Synced guild !') return await self.bot.tree.sync() - await ctx.reply(f"Synced global !") + await ctx.reply('Synced global !') @commands.command() @commands.is_owner() - async def unsync(self, ctx: commands.Context, unsync_type: Literal['guild', 'global']) -> None: + async def unsync(self, ctx: commands.Context[ValorantBot], unsync_type: Literal['guild', 'global']) -> None: """Unsync the application commands""" async with ctx.typing(): if unsync_type == 'guild': self.bot.tree.clear_commands(guild=ctx.guild) await self.bot.tree.sync(guild=ctx.guild) - await ctx.reply(f"Un-Synced guild !") + await ctx.reply('Un-Synced guild !') return - self.bot.tree.clear_commands() + self.bot.tree.clear_commands() # type: ignore await self.bot.tree.sync() - await ctx.reply(f"Un-Synced global !") + await ctx.reply('Un-Synced global !') @app_commands.command(description='Shows basic information about the bot.') async def about(self, interaction: Interaction) -> None: """Shows basic information about the bot.""" - owner_url = f'https://discord.com/users/240059262297047041' + owner_url = 'https://discord.com/users/240059262297047041' github_project = 'https://github.com/staciax/Valorant-DiscordBot' support_url = 'https://discord.gg/FJSXPqQZgz' embed = discord.Embed(color=0xFFFFFF) embed.set_author(name='VALORANT BOT PROJECT', url=github_project) embed.set_thumbnail(url='https://i.imgur.com/ZtuNW0Z.png') - embed.add_field(name='DEV:', value=f"[ꜱᴛᴀᴄɪᴀ.#7475]({owner_url})", inline=False) + embed.add_field(name='DEV:', value=f'[ꜱᴛᴀᴄɪᴀ.#7475]({owner_url})', inline=False) embed.add_field( name='ᴄᴏɴᴛʀɪʙᴜᴛᴏʀꜱ:', - value=f"[kiznick](https://github.com/kiznick)\n" - "[KANATAISGOD](https://github.com/KANATAISGOD)\n" + value='[kiznick](https://github.com/kiznick)\n' + '[KANATAISGOD](https://github.com/KANATAISGOD)\n' "[TMADZ2007](https://github.com/KANATAISGOD')\n" - "[sevzin](https://github.com/sevzin)\n" - "[MIIGØ](https://github.com/miigo-dev)\n" - "[Connor](https://github.com/ConnorDoesDev)\n" - "[KohanaSann](https://github.com/KohanaSann)\n" - "[RyugaXhypeR](https://github.com/RyugaXhypeR)\n" - "[Austin Hornhead](https://github.com/marchingon12)\n", + '[sevzin](https://github.com/sevzin)\n' + '[MIIGØ](https://github.com/miigo-dev)\n' + '[Connor](https://github.com/ConnorDoesDev)\n' + '[KohanaSann](https://github.com/KohanaSann)\n' + '[RyugaXhypeR](https://github.com/RyugaXhypeR)\n' + '[Austin Hornhead](https://github.com/marchingon12)\n', inline=False, ) view = ui.View() diff --git a/cogs/errors.py b/cogs/errors.py index afcfcfd..a501235 100644 --- a/cogs/errors.py +++ b/cogs/errors.py @@ -1,7 +1,7 @@ from __future__ import annotations import traceback -from typing import TYPE_CHECKING, Union +from typing import TYPE_CHECKING import discord from discord import Interaction @@ -44,56 +44,56 @@ async def on_app_command_error(self, interaction: Interaction, error: AppCommand if self.bot.debug is True: traceback.print_exception(type(error), error, error.__traceback__) + error_message = 'An unknown error occurred, sorry' # if isinstance(error, CommandInvokeError): # error = error.original if isinstance(error, NotOwner): - error = "You are not the owner of this bot." + error_message = 'You are not the owner of this bot.' elif isinstance(error, BadArgument): - error = "Bad argument." - elif isinstance(error, (ValorantBotError, ResponseError, HandshakeError, DatabaseError, AuthenticationError)): - error = error + error_message = 'Bad argument.' + elif isinstance( + error, (ValorantBotError | ResponseError | HandshakeError | DatabaseError | AuthenticationError) + ): + error_message = error elif isinstance(error, ResponseError): - error = "Empty response from Riot server." + error_message = 'Empty response from Riot server.' elif isinstance(error, HandshakeError): - error = "Could not connect to Riot server." - elif isinstance(error, CommandOnCooldown): - error = error - elif isinstance(error, Union[AppCommandNotFound, MissingPermissions, BotMissingPermissions]): + error_message = 'Could not connect to Riot server.' + elif isinstance(error, (CommandOnCooldown | AppCommandNotFound | MissingPermissions | BotMissingPermissions)): error = error - else: - error = f"An unknown error occurred, sorry" - traceback.print_exception(type(error), error) + # else: + # traceback.print_exception(type(error), error) - embed = discord.Embed(description=f'{str(error)[:2000]}', color=0xFE676E) + embed = discord.Embed(description=f'{str(error_message)[:2000]}', color=0xFE676E) if interaction.response.is_done(): return await interaction.followup.send(embed=embed, ephemeral=True) await interaction.response.send_message(embed=embed, ephemeral=True) @commands.Cog.listener() - async def on_command_error(self, ctx: commands.Context, error: Exception) -> None: + async def on_command_error(self, ctx: commands.Context[ValorantBot], error: Exception) -> None: embed = discord.Embed(color=0xFE676E) if isinstance(error, CommandNotFound): return elif isinstance(error, CheckFailure): - cm_error = "Only owners can run this command!" + cm_error = 'Only owners can run this command!' elif isinstance(error, MissingRequiredArgument): - cm_error = f"You didn't pass a required argument!" - if ctx.command.name in ['sync', 'unsync']: - cm_error = f"You need to specify a sync type: `guild` or `global`" - elif hasattr(error, "original"): - if isinstance(error.original, discord.Forbidden): - cm_error = f"Bot don't have permission to run this command." - if ctx.command.name in ['sync', 'unsync']: - cm_error = f"Bot don't have permission `applications.commands` to sync." + cm_error = "You didn't pass a required argument!" + if ctx.command and ctx.command.name in ['sync', 'unsync']: + cm_error = 'You need to specify a sync type: `guild` or `global`' + elif hasattr(error, 'original'): + if isinstance(error.original, discord.Forbidden): # type: ignore + cm_error = "Bot don't have permission to run this command." + if ctx.command and ctx.command.name in ['sync', 'unsync']: + cm_error = "Bot don't have permission `applications.commands` to sync." embed.set_image(url=app_cmd_scope) - elif isinstance(error.original, discord.HTTPException): - cm_error = f"An error occurred while processing your request." + elif isinstance(error.original, discord.HTTPException): # type: ignore + cm_error = 'An error occurred while processing your request.' elif isinstance(error, BadLiteralArgument): cm_error = f"**Invalid literal:** {', '.join(error.literals)}" else: traceback.print_exception(type(error), error, error.__traceback__) - cm_error = f"An unknown error occurred, sorry" + cm_error = 'An unknown error occurred, sorry' embed.description = cm_error await ctx.send(embed=embed, delete_after=30, ephemeral=True) diff --git a/cogs/notify.py b/cogs/notify.py index 04a7312..74ad4fc 100644 --- a/cogs/notify.py +++ b/cogs/notify.py @@ -3,7 +3,7 @@ import traceback from datetime import datetime, time, timedelta from difflib import get_close_matches -from typing import TYPE_CHECKING, Any, Literal, Tuple +from typing import TYPE_CHECKING, Any, Literal # Standard import discord @@ -29,8 +29,8 @@ class Notify(commands.Cog): def __init__(self, bot: ValorantBot) -> None: self.bot: ValorantBot = bot - self.endpoint: API_ENDPOINT = None - self.db: DATABASE = None + self.endpoint: API_ENDPOINT = None # type: ignore + self.db: DATABASE = None # type: ignore self.notifys.start() def cog_unload(self) -> None: @@ -41,10 +41,10 @@ async def on_ready(self) -> None: self.db = DATABASE() self.endpoint = API_ENDPOINT() - async def get_endpoint_and_data(self, user_id: int) -> Tuple[API_ENDPOINT, Any]: + async def get_endpoint_and_data(self, user_id: int) -> tuple[API_ENDPOINT, Any]: data = await self.db.is_data(user_id, 'en-US') endpoint = self.endpoint - endpoint.activate(data) + endpoint.activate(data) # type: ignore return endpoint, data async def send_notify(self) -> None: @@ -53,14 +53,13 @@ async def send_notify(self) -> None: for user_id in notify_users: try: - # endpoint endpoint, data = await self.get_endpoint_and_data(int(user_id)) # offer offer = endpoint.store_fetch_storefront() - skin_offer_list = offer["SkinsPanelLayout"]["SingleItemOffers"] - duration = offer["SkinsPanelLayout"]["SingleItemOffersRemainingDurationInSeconds"] + skin_offer_list = offer['SkinsPanelLayout']['SingleItemOffers'] + duration = offer['SkinsPanelLayout']['SingleItemOffersRemainingDurationInSeconds'] # author author = self.bot.get_user(int(user_id)) or await self.bot.fetch_user(int(user_id)) @@ -68,36 +67,40 @@ async def send_notify(self) -> None: # get guild language guild_locale = 'en-US' - get_guild_locale = [guild.preferred_locale for guild in self.bot.guilds if channel_send in guild.channels] + get_guild_locale = [ + guild.preferred_locale for guild in self.bot.guilds if channel_send in guild.channels + ] if len(get_guild_locale) > 0: guild_locale = guild_locale[0] response = ResponseLanguage('notify_send', guild_locale) - user_skin_list = [skin for skin in notify_data if skin['id'] == str(user_id)] - user_skin_list_uuid = [skin['uuid'] for skin in notify_data if skin['id'] == str(user_id)] + user_skin_list = [skin for skin in notify_data if skin['id'] == str(user_id)] # type: ignore + user_skin_list_uuid = [skin['uuid'] for skin in notify_data if skin['id'] == str(user_id)] # type: ignore if data['notify_mode'] == 'Specified': skin_notify_list = list(set(skin_offer_list).intersection(set(user_skin_list_uuid))) for noti in user_skin_list: - if noti['uuid'] in skin_notify_list: - uuid = noti['uuid'] + if noti['uuid'] in skin_notify_list: # type: ignore + uuid = noti['uuid'] # type: ignore skin = GetItems.get_skin(uuid) name = skin['names'][guild_locale] icon = skin['icon'] emoji = GetEmoji.tier_by_bot(uuid, self.bot) - notify_send: str = response.get('RESPONSE_SPECIFIED') - duration = format_relative(datetime.utcnow() + timedelta(seconds=duration)) + notify_send: str = response.get('RESPONSE_SPECIFIED') # type: ignore + duration = format_relative(datetime.utcnow() + timedelta(seconds=duration)) # type: ignore embed = Embed(notify_send.format(emoji=emoji, name=name, duration=duration), color=0xFD4554) embed.set_thumbnail(url=icon) view = View.NotifyView(user_id, uuid, name, ResponseLanguage('notify_add', guild_locale)) - view.message = await channel_send.send(content=f'||{author.mention}||', embed=embed, view=view) + view.message = await channel_send.send( # type: ignore + content=f'||{author.mention}||', embed=embed, view=view + ) elif data['notify_mode'] == 'All': embeds = GetEmbed.notify_all_send(endpoint.player, offer, response, self.bot) - await channel_send.send(content=f'||{author.mention}||', embeds=embeds) + await channel_send.send(content=f'||{author.mention}||', embeds=embeds) # type: ignore except (KeyError, FileNotFoundError): print(f'{user_id} is not in notify list') @@ -130,20 +133,19 @@ async def before_daily_send(self) -> None: @app_commands.guild_only() # @dynamic_cooldown(cooldown_5s) async def notify_add(self, interaction: Interaction, skin: str) -> None: - await interaction.response.defer() - await self.db.is_data(interaction.user.id, interaction.locale) # check if user is in db + await self.db.is_data(interaction.user.id, interaction.locale) # type: ignore # language - response = ResponseLanguage('notify_add', interaction.locale) + response = ResponseLanguage('notify_add', interaction.locale) # type: ignore # # setup emoji # await setup_emoji(self.bot, interaction.guild, interaction.locale) # check file whether - create_json('notifys', []) + create_json('notifys', []) # type: ignore # get cache skin_data = self.db.read_cache() @@ -168,18 +170,18 @@ async def notify_add(self, interaction: Interaction, skin: str) -> None: emoji = GetEmoji.tier_by_bot(skin_uuid, self.bot) for skin in notify_data: - if skin['id'] == str(interaction.user.id) and skin['uuid'] == skin_uuid: + if skin['id'] == str(interaction.user.id) and skin['uuid'] == skin_uuid: # type: ignore skin_already = response.get('SKIN_ALREADY_IN_LIST') - raise ValorantBotError(skin_already.format(emoji=emoji, skin=name)) + raise ValorantBotError(skin_already.format(emoji=emoji, skin=name)) # type: ignore - payload = dict(id=str(interaction.user.id), uuid=skin_uuid) + payload = {'id': str(interaction.user.id), 'uuid': skin_uuid} try: - notify_data.append(payload) + notify_data.append(payload) # type: ignore JSON.save('notifys', notify_data) except AttributeError: notify_data = [payload] - JSON.save('notifys', notify_data) + JSON.save('notifys', notify_data) # type: ignore # check if user is notify is on userdata = JSON.read('users') @@ -190,7 +192,7 @@ async def notify_add(self, interaction: Interaction, skin: str) -> None: JSON.save('users', userdata) success = response.get('SUCCESS') - embed = Embed(success.format(emoji=emoji, skin=name)) + embed = Embed(success.format(emoji=emoji, skin=name)) # type: ignore embed.set_thumbnail(url=icon) view = View.NotifyView(interaction.user.id, uuid, name, response) @@ -202,36 +204,34 @@ async def notify_add(self, interaction: Interaction, skin: str) -> None: @notify.command(name='list', description='View skins you have set a for notification.') # @dynamic_cooldown(cooldown_5s) async def notify_list(self, interaction: Interaction) -> None: - await interaction.response.defer(ephemeral=True) - response = ResponseLanguage('notify_list', interaction.locale) + response = ResponseLanguage('notify_list', interaction.locale) # type: ignore - await self.db.is_data(interaction.user.id, interaction.locale) # check if user is in db - view = View.NotifyViewList(interaction, response) + await self.db.is_data(interaction.user.id, interaction.locale) # check if user is in db # type: ignore + view = View.NotifyViewList(interaction, response) # type: ignore await view.start() @notify.command(name='mode', description='Change notification mode/channel.') @app_commands.describe(mode='Select the mode you want to change.') # @dynamic_cooldown(cooldown_5s) async def notify_mode(self, interaction: Interaction, mode: Literal['Specified Skin', 'All Skin', 'Off']) -> None: - await interaction.response.defer(ephemeral=True) # language - response = ResponseLanguage('notify_mode', interaction.locale) + response = ResponseLanguage('notify_mode', interaction.locale) # type: ignore - await self.db.is_data(interaction.user.id, interaction.locale) # check if user is in db + await self.db.is_data(interaction.user.id, interaction.locale) # type: ignore if mode == 'Specified Skin': # Check notify list if use mode specified skin self.db.check_notify_list(interaction.user.id) # check total notify list self.db.change_notify_mode(interaction.user.id, mode) # change notify mode - success = response.get("SUCCESS") - turn_off = response.get("TURN_OFF") + success = response.get('SUCCESS') + turn_off = response.get('TURN_OFF') - embed = Embed(success.format(mode=mode)) + embed = Embed(success.format(mode=mode)) # type: ignore if mode == 'Specified Skin': embed.set_image(url='https://i.imgur.com/RF6fHRY.png') elif mode == 'All Skin': @@ -245,33 +245,31 @@ async def notify_mode(self, interaction: Interaction, mode: Literal['Specified S @app_commands.describe(channel='Select the channel you want to change.') # @dynamic_cooldown(cooldown_5s) async def notify_channel(self, interaction: Interaction, channel: Literal['DM Message', 'Channel']) -> None: - await interaction.response.defer(ephemeral=True) # language - response = ResponseLanguage('notify_channel', interaction.locale) + response = ResponseLanguage('notify_channel', interaction.locale) # type: ignore - await self.db.is_data(interaction.user.id, interaction.locale) # check if user is in db + await self.db.is_data(interaction.user.id, interaction.locale) # type: ignore self.db.check_notify_list(interaction.user.id) # check total notify list self.db.change_notify_channel(interaction.user.id, channel, interaction.channel_id) # change notify channel - channel = '**DM Message**' if channel == 'DM Message' else f'{interaction.channel.mention}' + channel = '**DM Message**' if channel == 'DM Message' else f'{interaction.channel.mention}' # type: ignore - embed = discord.Embed(description=response.get('SUCCESS').format(channel=channel), color=0x77DD77) + embed = discord.Embed(description=response.get('SUCCESS').format(channel=channel), color=0x77DD77) # type: ignore await interaction.followup.send(embed=embed, ephemeral=True) @notify.command(name='test', description='Testing notification') # @dynamic_cooldown(cooldown_5s) async def notify_test(self, interaction: Interaction) -> None: - await interaction.response.defer(ephemeral=True) # language - response_test = ResponseLanguage('notify_test', interaction.locale) - response_send = ResponseLanguage('notify_send', interaction.locale) - response_add = ResponseLanguage('notify_add', interaction.locale) + response_test = ResponseLanguage('notify_test', interaction.locale) # type: ignore + response_send = ResponseLanguage('notify_send', interaction.locale) # type: ignore + response_add = ResponseLanguage('notify_add', interaction.locale) # type: ignore # notify list notify_data = JSON.read('notifys') @@ -281,8 +279,8 @@ async def notify_test(self, interaction: Interaction) -> None: offer = endpoint.store_fetch_storefront() # offer data - duration = offer["SkinsPanelLayout"]["SingleItemOffersRemainingDurationInSeconds"] - user_skin_list = [skin for skin in notify_data if skin['id'] == str(interaction.user.id)] + duration = offer['SkinsPanelLayout']['SingleItemOffersRemainingDurationInSeconds'] + user_skin_list = [skin for skin in notify_data if skin['id'] == str(interaction.user.id)] # type: ignore if len(user_skin_list) == 0: empty_list = response_test.get('EMPTY_LIST') @@ -293,38 +291,38 @@ async def notify_test(self, interaction: Interaction) -> None: try: if data['notify_mode'] == 'Specified': for notify in user_skin_list: - uuid = notify['uuid'] + uuid = notify['uuid'] # type: ignore skin = GetItems.get_skin(uuid) name = skin['names'][str(VLR_locale)] icon = skin['icon'] emoji = GetEmoji.tier_by_bot(uuid, self.bot) - notify_send: str = response_send.get('RESPONSE_SPECIFIED') + notify_send: str = response_send.get('RESPONSE_SPECIFIED') # type: ignore duration = format_relative(datetime.utcnow() + timedelta(seconds=duration)) embed = Embed(notify_send.format(emoji=emoji, name=name, duration=duration), color=0xFD4554) embed.set_thumbnail(url=icon) view = View.NotifyView(interaction.user.id, uuid, name, response_add) - view.message = await channel_send.send(embed=embed, view=view) + view.message = await channel_send.send(embed=embed, view=view) # type: ignore break elif data['notify_mode'] == 'All': embeds = GetEmbed.notify_all_send(endpoint.player, offer, response_send, self.bot) - await channel_send.send(embeds=embeds) + await channel_send.send(embeds=embeds) # type: ignore else: raise ValorantBotError(response_test.get('NOTIFY_TURN_OFF')) - except Forbidden: + except Forbidden as e: if channel_send == interaction.user: - raise ValorantBotError(response_test.get('PLEASE_ALLOW_DM_MESSAGE')) - raise ValorantBotError(response_test.get('BOT_MISSING_PERM')) - except HTTPException: - raise ValorantBotError(response_test.get('FAILED_SEND_NOTIFY')) + raise ValorantBotError(response_test.get('PLEASE_ALLOW_DM_MESSAGE')) from e + raise ValorantBotError(response_test.get('BOT_MISSING_PERM')) from e + except HTTPException as e: + raise ValorantBotError(response_test.get('FAILED_SEND_NOTIFY')) from e except Exception as e: print(e) - raise ValorantBotError(f"{response_test.get('FAILED_SEND_NOTIFY')} - {e}") + raise ValorantBotError(f"{response_test.get('FAILED_SEND_NOTIFY')} - {e}") from e else: await interaction.followup.send( embed=Embed(response_test.get('NOTIFY_IS_WORKING'), color=0x77DD77), ephemeral=True diff --git a/cogs/valorant.py b/cogs/valorant.py index 8012464..7960457 100644 --- a/cogs/valorant.py +++ b/cogs/valorant.py @@ -1,7 +1,7 @@ from __future__ import annotations import contextlib -from typing import TYPE_CHECKING, Literal # noqa: F401 +from typing import TYPE_CHECKING, Literal from discord import Interaction, app_commands, ui from discord.ext import commands, tasks @@ -28,14 +28,14 @@ class ValorantCog(commands.Cog, name='Valorant'): def __init__(self, bot: ValorantBot) -> None: self.bot: ValorantBot = bot - self.endpoint: API_ENDPOINT = None - self.db: DATABASE = None + self.endpoint: API_ENDPOINT = MISSING + self.db: DATABASE = MISSING self.reload_cache.start() def cog_unload(self) -> None: self.reload_cache.cancel() - def funtion_reload_cache(self, force=False) -> None: + def funtion_reload_cache(self, force: bool = False) -> None: """Reload the cache""" with contextlib.suppress(Exception): cache = self.db.read_cache() @@ -64,88 +64,85 @@ async def on_ready(self) -> None: self.endpoint = API_ENDPOINT() async def get_endpoint( - self, user_id: int, locale_code: str = None, username: str = None, password: str = None + self, + user_id: int, + locale_code: str | None = None, + username: str | None = None, + password: str | None = None, ) -> API_ENDPOINT: """Get the endpoint for the user""" if username is not None and password is not None: auth = self.db.auth - auth.locale_code = locale_code + auth.locale_code = locale_code # type: ignore data = await auth.temp_auth(username, password) elif username or password: - raise ValorantBotError(f"Please provide both username and password!") + raise ValorantBotError('Please provide both username and password!') else: - data = await self.db.is_data(user_id, locale_code) - data['locale_code'] = locale_code + data = await self.db.is_data(user_id, locale_code) # type: ignore + data['locale_code'] = locale_code # type: ignore endpoint = self.endpoint - endpoint.activate(data) + endpoint.activate(data) # type: ignore return endpoint @app_commands.command(description='Log in with your Riot acoount') @app_commands.describe(username='Input username', password='Input password') # @dynamic_cooldown(cooldown_5s) - async def login(self, interaction: Interaction, username: str, password: str) -> None: - - response = ResponseLanguage(interaction.command.name, interaction.locale) + async def login(self, interaction: Interaction[ValorantBot], username: str, password: str) -> None: + response = ResponseLanguage(interaction.command.name, interaction.locale) # type: ignore user_id = interaction.user.id auth = self.db.auth - auth.locale_code = interaction.locale + auth.locale_code = interaction.locale # type: ignore authenticate = await auth.authenticate(username, password) - if authenticate['auth'] == 'response': + if authenticate['auth'] == 'response': # type: ignore await interaction.response.defer(ephemeral=True) - login = await self.db.login(user_id, authenticate, interaction.locale) + login = await self.db.login(user_id, authenticate, interaction.locale) # type: ignore - if login['auth']: - embed = Embed(f"{response.get('SUCCESS')} **{login['player']}!**") + if login['auth']: # type: ignore + embed = Embed(f"{response.get('SUCCESS')} **{login['player']}!**") # type: ignore return await interaction.followup.send(embed=embed, ephemeral=True) raise ValorantBotError(f"{response.get('FAILED')}") - elif authenticate['auth'] == '2fa': - cookies = authenticate['cookie'] - message = authenticate['message'] - label = authenticate['label'] + elif authenticate['auth'] == '2fa': # type: ignore + cookies = authenticate['cookie'] # type: ignore + message = authenticate['message'] # type: ignore + label = authenticate['label'] # type: ignore modal = View.TwoFA_UI(interaction, self.db, cookies, message, label, response) await interaction.response.send_modal(modal) @app_commands.command(description='Logout and Delete your account from database') # @dynamic_cooldown(cooldown_5s) - async def logout(self, interaction: Interaction) -> None: - + async def logout(self, interaction: Interaction[ValorantBot]) -> None: await interaction.response.defer(ephemeral=True) - response = ResponseLanguage(interaction.command.name, interaction.locale) + response = ResponseLanguage(interaction.command.name, interaction.locale) # type: ignore user_id = interaction.user.id - if logout := self.db.logout(user_id, interaction.locale): + if logout := self.db.logout(user_id, interaction.locale): # type: ignore if logout: embed = Embed(response.get('SUCCESS')) return await interaction.followup.send(embed=embed, ephemeral=True) raise ValorantBotError(response.get('FAILED')) - @app_commands.command(description="Shows your daily store in your accounts") - @app_commands.describe(username='Input username (without login)', password='password (without login)') + @app_commands.command(description='Shows your daily store in your accounts') @app_commands.guild_only() # @dynamic_cooldown(cooldown_5s) - async def store(self, interaction: Interaction, username: str = None, password: str = None) -> None: + async def store(self, interaction: Interaction[ValorantBot]) -> None: + await interaction.response.defer() # language - response = ResponseLanguage(interaction.command.name, interaction.locale) - - # check if user is logged in - is_private_message = True if username is not None or password is not None else False - - await interaction.response.defer(ephemeral=is_private_message) + response = ResponseLanguage(interaction.command.name, interaction.locale) # type: ignore if not interaction.guild: raise ValorantBotError('This command can only be used in a server') # setup emoji - await setup_emoji(self.bot, interaction.guild, interaction.locale) + await setup_emoji(self.bot, interaction.guild, interaction.locale) # type: ignore # get endpoint - endpoint = await self.get_endpoint(interaction.user.id, interaction.locale, username, password) + endpoint = await self.get_endpoint(interaction.user.id, interaction.locale) # type: ignore # fetch skin price skin_price = endpoint.store_fetch_offers() @@ -154,82 +151,70 @@ async def store(self, interaction: Interaction, username: str = None, password: # data data = endpoint.store_fetch_storefront() embeds = GetEmbed.store(endpoint.player, data, response, self.bot) - await interaction.followup.send( - embeds=embeds, view=View.share_button(interaction, embeds) if is_private_message else MISSING - ) + await interaction.followup.send(embeds=embeds, view=View.share_button(interaction, embeds)) @app_commands.command(description='View your remaining Valorant and Riot Points (VP/RP)') @app_commands.guild_only() # @dynamic_cooldown(cooldown_5s) - async def point(self, interaction: Interaction, username: str = None, password: str = None) -> None: - + async def point(self, interaction: Interaction[ValorantBot]) -> None: # check if user is logged in - is_private_message = True if username is not None or password is not None else False - await interaction.response.defer(ephemeral=is_private_message) + await interaction.response.defer() - response = ResponseLanguage(interaction.command.name, interaction.locale) + response = ResponseLanguage(interaction.command.name, interaction.locale) # type: ignore if not interaction.guild: raise ValorantBotError('This command can only be used in a server') # setup emoji - await setup_emoji(self.bot, interaction.guild, interaction.locale) + await setup_emoji(self.bot, interaction.guild, interaction.locale.value) # endpoint - endpoint = await self.get_endpoint(interaction.user.id, locale_code=interaction.locale) + endpoint = await self.get_endpoint(interaction.user.id, locale_code=interaction.locale.value) # data data = endpoint.store_fetch_wallet() embed = GetEmbed.point(endpoint.player, data, response, self.bot) - await interaction.followup.send( - embed=embed, view=View.share_button(interaction, [embed]) if is_private_message else MISSING - ) + await interaction.followup.send(embed=embed, view=View.share_button(interaction, [embed])) @app_commands.command(description='View your daily/weekly mission progress') # @dynamic_cooldown(cooldown_5s) - async def mission(self, interaction: Interaction, username: str = None, password: str = None) -> None: - + async def mission(self, interaction: Interaction[ValorantBot]) -> None: # check if user is logged in - is_private_message = True if username is not None or password is not None else False - await interaction.response.defer(ephemeral=is_private_message) + await interaction.response.defer() - response = ResponseLanguage(interaction.command.name, interaction.locale) + response = ResponseLanguage(interaction.command.name, interaction.locale) # type: ignore # endpoint - endpoint = await self.get_endpoint(interaction.user.id, interaction.locale, username, password) + endpoint = await self.get_endpoint(interaction.user.id, interaction.locale) # type: ignore # data data = endpoint.fetch_contracts() embed = GetEmbed.mission(endpoint.player, data, response) - await interaction.followup.send( - embed=embed, view=View.share_button(interaction, [embed]) if is_private_message else MISSING - ) + await interaction.followup.send(embed=embed, view=View.share_button(interaction, [embed])) @app_commands.command(description='Show skin offers on the nightmarket') @app_commands.guild_only() # @dynamic_cooldown(cooldown_5s) - async def nightmarket(self, interaction: Interaction, username: str = None, password: str = None) -> None: - + async def nightmarket(self, interaction: Interaction[ValorantBot]) -> None: # check if user is logged in - is_private_message = True if username is not None or password is not None else False - await interaction.response.defer(ephemeral=is_private_message) + await interaction.response.defer() if not interaction.guild: raise ValorantBotError('This command can only be used in a server') # setup emoji - await setup_emoji(self.bot, interaction.guild, interaction.locale) + await setup_emoji(self.bot, interaction.guild, interaction.locale) # type: ignore # language - response = ResponseLanguage(interaction.command.name, interaction.locale) + response = ResponseLanguage(interaction.command.name, interaction.locale) # type: ignore # endpoint - endpoint = await self.get_endpoint(interaction.user.id, interaction.locale, username, password) + endpoint = await self.get_endpoint(interaction.user.id, interaction.locale) # type: ignore # fetch skin price skin_price = endpoint.store_fetch_offers() @@ -239,23 +224,19 @@ async def nightmarket(self, interaction: Interaction, username: str = None, pass data = endpoint.store_fetch_storefront() embeds = GetEmbed.nightmarket(endpoint.player, data, self.bot, response) - await interaction.followup.send( - embeds=embeds, view=View.share_button(interaction, embeds) if is_private_message else MISSING - ) + await interaction.followup.send(embeds=embeds, view=View.share_button(interaction, embeds)) # type: ignore @app_commands.command(description='View your battlepass current tier') # @dynamic_cooldown(cooldown_5s) - async def battlepass(self, interaction: Interaction, username: str = None, password: str = None) -> None: - + async def battlepass(self, interaction: Interaction[ValorantBot]) -> None: # check if user is logged in - is_private_message = True if username is not None or password is not None else False - await interaction.response.defer(ephemeral=is_private_message) + await interaction.response.defer() - response = ResponseLanguage(interaction.command.name, interaction.locale) + response = ResponseLanguage(interaction.command.name, interaction.locale) # type: ignore # endpoint - endpoint = await self.get_endpoint(interaction.user.id, interaction.locale, username, password) + endpoint = await self.get_endpoint(interaction.user.id, interaction.locale) # type: ignore # data data = endpoint.fetch_contracts() @@ -264,26 +245,23 @@ async def battlepass(self, interaction: Interaction, username: str = None, passw embed = GetEmbed.battlepass(endpoint.player, data, season, response) - await interaction.followup.send( - embed=embed, view=View.share_button(interaction, [embed]) if is_private_message else MISSING - ) + await interaction.followup.send(embed=embed, view=View.share_button(interaction, [embed])) # inspired by https://github.com/giorgi-o - @app_commands.command(description="inspect a specific bundle") - @app_commands.describe(bundle="The name of the bundle you want to inspect!") + @app_commands.command(description='inspect a specific bundle') + @app_commands.describe(bundle='The name of the bundle you want to inspect!') @app_commands.guild_only() # @dynamic_cooldown(cooldown_5s) - async def bundle(self, interaction: Interaction, bundle: str) -> None: - + async def bundle(self, interaction: Interaction[ValorantBot], bundle: str) -> None: await interaction.response.defer() - response = ResponseLanguage(interaction.command.name, interaction.locale) + response = ResponseLanguage(interaction.command.name, interaction.locale.value) # type: ignore if not interaction.guild: raise ValorantBotError('This command can only be used in a server') # setup emoji - await setup_emoji(self.bot, interaction.guild, interaction.locale) + await setup_emoji(self.bot, interaction.guild, interaction.locale.value) # cache cache = self.db.read_cache() @@ -305,27 +283,26 @@ async def bundle(self, interaction: Interaction, bundle: str) -> None: find_bundle = (find_bundle_en_US if len(find_bundle_en_US) > 0 else find_bundle_locale)[:25] # bundle view - view = View.BaseBundle(interaction, find_bundle, response) + view = View.BaseBundle(interaction, find_bundle, response) # type: ignore await view.start() # inspired by https://github.com/giorgi-o - @app_commands.command(description="Show the current featured bundles") + @app_commands.command(description='Show the current featured bundles') @app_commands.guild_only() # @dynamic_cooldown(cooldown_5s) - async def bundles(self, interaction: Interaction) -> None: - + async def bundles(self, interaction: Interaction[ValorantBot]) -> None: await interaction.response.defer() - response = ResponseLanguage(interaction.command.name, interaction.locale) + response = ResponseLanguage(interaction.command.name, interaction.locale.value) # type: ignore if not interaction.guild: raise ValorantBotError('This command can only be used in a server') # setup emoji - await setup_emoji(self.bot, interaction.guild, interaction.locale) + await setup_emoji(self.bot, interaction.guild, interaction.locale.value) # endpoint - endpoint = await self.get_endpoint(interaction.user.id, interaction.locale) + endpoint = await self.get_endpoint(interaction.user.id, interaction.locale.value) # data bundle_entries = endpoint.store_fetch_storefront() @@ -337,24 +314,24 @@ async def bundles(self, interaction: Interaction) -> None: # credit https://github.com/giorgi-o # https://github.com/giorgi-o/SkinPeek/wiki/How-to-get-your-Riot-cookies @app_commands.command() - @app_commands.describe(cookie="Your cookie") - async def cookies(self, interaction: Interaction, cookie: str) -> None: + @app_commands.describe(cookie='Your cookie') + async def cookies(self, interaction: Interaction[ValorantBot], cookie: str) -> None: """Login to your account with a cookie""" await interaction.response.defer(ephemeral=True) # language - response = ResponseLanguage(interaction.command.name, interaction.locale) + response = ResponseLanguage(interaction.command.name, interaction.locale.value) # type: ignore - login = await self.db.cookie_login(interaction.user.id, cookie, interaction.locale) + login = await self.db.cookie_login(interaction.user.id, cookie, interaction.locale.value) - if login['auth']: - embed = Embed(f"{response.get('SUCCESS')} **{login['player']}!**") + if login['auth']: # type: ignore + embed = Embed(f"{response.get('SUCCESS')} **{login['player']}!**") # type: ignore await interaction.followup.send(embed=embed, ephemeral=True) return view = ui.View() - view.add_item(ui.Button(label="Tutorial", emoji="🔗", url="https://youtu.be/cFMNHEHEp2A")) + view.add_item(ui.Button(label='Tutorial', emoji='🔗', url='https://youtu.be/cFMNHEHEp2A')) await interaction.followup.send(f"{response.get('FAILURE')}", view=view, ephemeral=True) # ---------- ROAD MAP ---------- # @@ -376,20 +353,21 @@ async def cookies(self, interaction: Interaction, cookie: str) -> None: # ---------- DEBUGs ---------- # @app_commands.command(description='The command debug for the bot') - @app_commands.describe(bug="The bug you want to fix") + @app_commands.describe(bug='The bug you want to fix') @app_commands.guild_only() @owner_only() async def debug( - self, interaction: Interaction, bug: Literal['Skin price not loading', 'Emoji not loading', 'Cache not loading'] + self, + interaction: Interaction[ValorantBot], + bug: Literal['Skin price not loading', 'Emoji not loading', 'Cache not loading'], ) -> None: - await interaction.response.defer(ephemeral=True) - response = ResponseLanguage(interaction.command.name, interaction.locale) + response = ResponseLanguage(interaction.command.name, interaction.locale.value) # type: ignore if bug == 'Skin price not loading': # endpoint - endpoint = await self.get_endpoint(interaction.user.id, interaction.locale) + endpoint = await self.get_endpoint(interaction.user.id, interaction.locale.value) # fetch skin price skin_price = endpoint.store_fetch_offers() @@ -399,12 +377,12 @@ async def debug( if not interaction.guild: raise ValorantBotError('This command can only be used in a server') - await setup_emoji(self.bot, interaction.guild, interaction.locale, force=True) + await setup_emoji(self.bot, interaction.guild, interaction.locale.value, force=True) elif bug == 'Cache not loading': self.funtion_reload_cache(force=True) - success = response.get('SUCCESS') + success: str = response.get('SUCCESS', 'success') await interaction.followup.send(embed=Embed(success.format(bug=bug))) diff --git a/pyproject.toml b/pyproject.toml index 8f4a375..46c2052 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,32 +1,40 @@ -[tool.black] -line-length = 125 -skip-string-normalization = true +[project] +name = "valorant-discord-bot" +requires-python = ">=3.12" -[tool.isort] -profile = "black" -combine_as_imports = true -combine_star = true -line_length = 125 +[tool.pyright] +include = ["cogs"] +exclude = ["**/__pycache__", ".git", ".ruff_cache", ".venv"] +pythonVersion = "3.12" +typeCheckingMode = "basic" +reportUnnecessaryTypeIgnoreComment = "warning" +reportMissingParameterType = "error" +reportUnusedVariable = "warning" -[tool.poetry] -name = "repl_nix_Valorant-DiscordBot" -version = "0.1.0" -description = "" -authors = ["Replit user <<>>"] +[tool.ruff] +line-length = 120 +target-version = "py312" +exclude = [".git", ".ruff_cache", ".venv"] -[tool.poetry.dependencies] -python = "^3.8" -aiohttp = "^3.8.1" -discord = "^2.0.0" -multidict = "^6.0.2" -python-dotenv = "^0.20.0" -requests = "^2.28.1" -urllib3 = "^1.26.12" -replit = "^3.2.4" -Flask = "^2.2.2" +[tool.ruff.lint] +preview = true +select = [ + "E", # pycodestyle errors + "W", # pycodestyle warnings + "F", # pyflakes + "I", # isort + "UP", # pyupgrade, + "B", # flake8-bugbear + "C4", # flake8-comprehensions + "TCH", # flake8-type-checking +] +ignore = [ + "E501", # line-too-long +] -[tool.poetry.dev-dependencies] +[tool.ruff.format] +quote-style = "single" +skip-magic-trailing-comma = false -[build-system] -requires = ["poetry-core>=1.0.0"] -build-backend = "poetry.core.masonry.api" +[tool.ruff.lint.isort] +combine-as-imports = true \ No newline at end of file diff --git a/replit_db_clean.py b/replit_db_clean.py index f9f131b..9897e8a 100644 --- a/replit_db_clean.py +++ b/replit_db_clean.py @@ -1,4 +1,4 @@ from replit import db -for _ in db.keys(): +for _ in db: del db[_] diff --git a/replit_main.py b/replit_main.py index 807ddb1..6bc7425 100644 --- a/replit_main.py +++ b/replit_main.py @@ -13,7 +13,7 @@ def main(): def run(): - app.run(host="0.0.0.0", port=8080) + app.run(host='0.0.0.0', port=8080) def keep_alive(): diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..b4d3a06 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,3 @@ +-r requirements.txt + +ruff==0.3.7 \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 0284745..52ba569 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,6 +1,3 @@ -discord.py~=2.0.0 -requests~=2.27.1 -python-dotenv~=0.19.2 -aiohttp~=3.8.1 -urllib3~=1.26.8 -multidict~=6.0.2 \ No newline at end of file +discord.py>=2.3.2 +python-dotenv>=1.0.1 +requests>=2.31.0 \ No newline at end of file diff --git a/utils/checks.py b/utils/checks.py index ff8ac00..52d1bfb 100644 --- a/utils/checks.py +++ b/utils/checks.py @@ -1,6 +1,6 @@ from __future__ import annotations -from typing import TYPE_CHECKING, Optional +from typing import TYPE_CHECKING import discord from discord import Interaction, app_commands @@ -10,7 +10,7 @@ def _self_bot(interaction: Interaction) -> ValorantBot: - bot: ValorantBot = getattr(interaction, "client", interaction._state._get_client()) + bot: ValorantBot = getattr(interaction, 'client', interaction._state._get_client()) return bot @@ -29,7 +29,7 @@ async def predicate(interaction: Interaction): return app_commands.check(predicate) -def cooldown_5s(interaction: discord.Interaction) -> Optional[app_commands.Cooldown]: +def cooldown_5s(interaction: discord.Interaction) -> app_commands.Cooldown | None: """ Example cooldown: from discord.app_commands.checks import dynamic_cooldown diff --git a/utils/locale_v2.py b/utils/locale_v2.py index 65ac477..f4f020c 100644 --- a/utils/locale_v2.py +++ b/utils/locale_v2.py @@ -1,11 +1,11 @@ """ DEMO TRANSLATION """ + from __future__ import annotations import os from contextvars import ContextVar -from typing import Optional discord_locale = [ 'da', # Danish @@ -59,8 +59,8 @@ 'vi': 'vi-VN', # vietnamese } -_current_locale = ContextVar("_current_locale", default="en-US") -_valorant_current_locale = ContextVar("_valorant_current_locale", default="en-US") +_current_locale = ContextVar('_current_locale', default='en-US') +_valorant_current_locale = ContextVar('_valorant_current_locale', default='en-US') def get_interaction_locale() -> str: @@ -68,24 +68,24 @@ def get_interaction_locale() -> str: return str(_current_locale.get()) -def set_interaction_locale(locale: Optional[str]) -> None: +def set_interaction_locale(locale: str | None) -> None: """Set the locale for bot""" _current_locale.set(locale) def get_valorant_locale() -> str: """Get the locale for valorant api""" - valorant_locale = valorant_locale_overwrite.get(str(_valorant_current_locale.get()), "en-US") + valorant_locale = valorant_locale_overwrite.get(str(_valorant_current_locale.get()), 'en-US') return valorant_locale -def set_valorant_locale(locale: Optional[str]) -> None: +def set_valorant_locale(locale: str | None) -> None: """Set the locale for valorant api""" language_files = os.listdir('languages') locale_json = str(locale) + '.json' if locale_json not in language_files: - _valorant_current_locale.set("en-US") + _valorant_current_locale.set('en-US') _valorant_current_locale.set(locale) diff --git a/utils/valorant/auth.py b/utils/valorant/auth.py index 1504035..e32ba29 100644 --- a/utils/valorant/auth.py +++ b/utils/valorant/auth.py @@ -5,12 +5,10 @@ import re import ssl from datetime import datetime, timedelta -from typing import Any, Dict, Optional, Tuple +from typing import Any # Third import aiohttp -import urllib3 -from multidict import MultiDict from ..errors import AuthenticationError from ..locale_v2 import ValorantTranslator @@ -20,25 +18,24 @@ vlr_locale = ValorantTranslator() -# disable urllib3 warnings that might arise from making requests to 127.0.0.1 -urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) - def _extract_tokens(data: str) -> str: """Extract tokens from data""" - pattern = re.compile('access_token=((?:[a-zA-Z]|\d|\.|-|_)*).*id_token=((?:[a-zA-Z]|\d|\.|-|_)*).*expires_in=(\d*)') - response = pattern.findall(data['response']['parameters']['uri'])[0] + pattern = re.compile( + r'access_token=((?:[a-zA-Z]|\d|\.|-|_)*).*id_token=((?:[a-zA-Z]|\d|\.|-|_)*).*expires_in=(\d*)' + ) + response = pattern.findall(data['response']['parameters']['uri'])[0] # type: ignore return response -def _extract_tokens_from_uri(url: str) -> Optional[Tuple[str, Any]]: +def _extract_tokens_from_uri(url: str) -> tuple[str, str]: try: - access_token = url.split("access_token=")[1].split("&scope")[0] - token_id = url.split("id_token=")[1].split("&")[0] + access_token = url.split('access_token=')[1].split('&scope')[0] + token_id = url.split('id_token=')[1].split('&')[0] return access_token, token_id - except IndexError: - raise AuthenticationError('Cookies Invalid') + except IndexError as e: + raise AuthenticationError('Cookies Invalid') from e # https://developers.cloudflare.com/ssl/ssl-tls/cipher-suites/ @@ -65,7 +62,7 @@ def _extract_tokens_from_uri(url: str) -> Optional[Tuple[str, Any]]: class ClientSession(aiohttp.ClientSession): - def __init__(self, *args, **kwargs): + def __init__(self, *args: Any, **kwargs: Any) -> None: ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH) ctx.minimum_version = ssl.TLSVersion.TLSv1_3 ctx.set_ciphers(':'.join(FORCED_CIPHERS)) @@ -73,10 +70,10 @@ def __init__(self, *args, **kwargs): class Auth: - RIOT_CLIENT_USER_AGENT = "RiotClient/60.0.6.4770705.4749685 rso-auth (Windows;10;;Professional, x64)" - + RIOT_CLIENT_USER_AGENT = 'RiotClient/60.0.6.4770705.4749685 rso-auth (Windows;10;;Professional, x64)' + def __init__(self) -> None: - self._headers: Dict = { + self._headers: dict = { 'Content-Type': 'application/json', 'User-Agent': Auth.RIOT_CLIENT_USER_AGENT, 'Accept': 'application/json, text/plain, */*', @@ -86,12 +83,12 @@ def __init__(self) -> None: self.locale_code = 'en-US' # default language self.response = {} # prepare response for local response - def local_response(self) -> LocalErrorResponse: + def local_response(self) -> dict[str, Any]: """This function is used to check if the local response is enabled.""" self.response = LocalErrorResponse('AUTH', self.locale_code) return self.response - async def authenticate(self, username: str, password: str) -> Optional[Dict[str, Any]]: + async def authenticate(self, username: str, password: str) -> dict[str, Any] | None: """This function is used to authenticate the user.""" # language @@ -100,10 +97,10 @@ async def authenticate(self, username: str, password: str) -> Optional[Dict[str, session = ClientSession() data = { - "client_id": "play-valorant-web-prod", - "nonce": "1", - "redirect_uri": "https://playvalorant.com/opt_in", - "response_type": "token id_token", + 'client_id': 'play-valorant-web-prod', + 'nonce': '1', + 'redirect_uri': 'https://playvalorant.com/opt_in', + 'response_type': 'token id_token', 'scope': 'account openid', } @@ -116,9 +113,11 @@ async def authenticate(self, username: str, password: str) -> Optional[Dict[str, for cookie in r.cookies.items(): cookies['cookie'][cookie[0]] = str(cookie).split('=')[1].split(';')[0] - data = {"type": "auth", "username": username, "password": password, "remember": True} + data = {'type': 'auth', 'username': username, 'password': password, 'remember': True} - async with session.put('https://auth.riotgames.com/api/v1/authorization', json=data, headers=self._headers) as r: + async with session.put( + 'https://auth.riotgames.com/api/v1/authorization', json=data, headers=self._headers + ) as r: data = await r.json() for cookie in r.cookies.items(): cookies['cookie'][cookie[0]] = str(cookie).split('=')[1].split(';')[0] @@ -134,22 +133,21 @@ async def authenticate(self, username: str, password: str) -> Optional[Dict[str, token_id = response[1] expiry_token = datetime.now() + timedelta(minutes=59) - cookies['expiry_token'] = int(datetime.timestamp(expiry_token)) + cookies['expiry_token'] = int(datetime.timestamp(expiry_token)) # type: ignore return {'auth': 'response', 'data': {'cookie': cookies, 'access_token': access_token, 'token_id': token_id}} elif data['type'] == 'multifactor': - if r.status == 429: raise AuthenticationError(local_response.get('RATELIMIT', 'Please wait a few minutes and try again.')) label_modal = local_response.get('INPUT_2FA_CODE') - WaitFor2FA = {"auth": "2fa", "cookie": cookies, 'label': label_modal} + WaitFor2FA = {'auth': '2fa', 'cookie': cookies, 'label': label_modal} if data['multifactor']['method'] == 'email': - WaitFor2FA[ - 'message' - ] = f"{local_response.get('2FA_TO_EMAIL', 'Riot sent a code to')} {data['multifactor']['email']}" + WaitFor2FA['message'] = ( + f"{local_response.get('2FA_TO_EMAIL', 'Riot sent a code to')} {data['multifactor']['email']}" + ) return WaitFor2FA WaitFor2FA['message'] = local_response.get('2FA_ENABLE', 'You have 2FA enabled!') @@ -157,7 +155,7 @@ async def authenticate(self, username: str, password: str) -> Optional[Dict[str, raise AuthenticationError(local_response.get('INVALID_PASSWORD', 'Your username or password may be incorrect!')) - async def get_entitlements_token(self, access_token: str) -> Optional[str]: + async def get_entitlements_token(self, access_token: str) -> str: """This function is used to get the entitlements token.""" # language @@ -173,12 +171,14 @@ async def get_entitlements_token(self, access_token: str) -> Optional[str]: await session.close() try: entitlements_token = data['entitlements_token'] - except KeyError: - raise AuthenticationError(local_response.get('COOKIES_EXPIRED', 'Cookies is expired, plz /login again!')) + except KeyError as e: + raise AuthenticationError( + local_response.get('COOKIES_EXPIRED', 'Cookies is expired, plz /login again!') + ) from e else: return entitlements_token - async def get_userinfo(self, access_token: str) -> Tuple[str, str, str]: + async def get_userinfo(self, access_token: str) -> tuple[str, str, str]: """This function is used to get the user info.""" # language @@ -196,8 +196,10 @@ async def get_userinfo(self, access_token: str) -> Tuple[str, str, str]: puuid = data['sub'] name = data['acct']['game_name'] tag = data['acct']['tag_line'] - except KeyError: - raise AuthenticationError(local_response.get('NO_NAME_TAG', 'This user hasn\'t created a name or tagline yet.')) + except KeyError as e: + raise AuthenticationError( + local_response.get('NO_NAME_TAG', "This user hasn't created a name or tagline yet.") + ) from e else: return puuid, name, tag @@ -211,7 +213,7 @@ async def get_region(self, access_token: str, token_id: str) -> str: headers = {'Content-Type': 'application/json', 'Authorization': f'Bearer {access_token}'} - body = {"id_token": token_id} + body = {'id_token': token_id} async with session.put( 'https://riot-geo.pas.si.riotgames.com/pas/v1/product/valorant', headers=headers, json=body @@ -221,14 +223,14 @@ async def get_region(self, access_token: str, token_id: str) -> str: await session.close() try: region = data['affinities']['live'] - except KeyError: + except KeyError as e: raise AuthenticationError( local_response.get('REGION_NOT_FOUND', 'An unknown error occurred, plz `/login` again') - ) + ) from e else: return region - async def give2facode(self, code: str, cookies: Dict) -> Dict[str, Any]: + async def give2facode(self, code: str, cookies: dict[str, Any]) -> dict[str, Any]: """This function is used to give the 2FA code.""" # language @@ -238,10 +240,13 @@ async def give2facode(self, code: str, cookies: Dict) -> Dict[str, Any]: # headers = {'Content-Type': 'application/json', 'User-Agent': self.user_agent} - data = {"type": "multifactor", "code": code, "rememberDevice": True} + data = {'type': 'multifactor', 'code': code, 'rememberDevice': True} async with session.put( - 'https://auth.riotgames.com/api/v1/authorization', headers=self._headers, json=data, cookies=cookies['cookie'] + 'https://auth.riotgames.com/api/v1/authorization', + headers=self._headers, + json=data, + cookies=cookies['cookie'], ) as r: data = await r.json() @@ -258,7 +263,7 @@ async def give2facode(self, code: str, cookies: Dict) -> Dict[str, Any]: return {'auth': 'failed', 'error': local_response.get('2FA_INVALID_CODE')} - async def redeem_cookies(self, cookies: Dict) -> Tuple[Dict[str, Any], str, str]: + async def redeem_cookies(self, cookies: dict) -> tuple[dict[str, Any], str, str]: """This function is used to redeem the cookies.""" # language @@ -273,8 +278,8 @@ async def redeem_cookies(self, cookies: Dict) -> Tuple[Dict[str, Any], str, str] cookies = cookies['cookie'] async with session.get( - "https://auth.riotgames.com/authorize?redirect_uri=https%3A%2F%2Fplayvalorant.com%2Fopt_in&client_id=play" - "-valorant-web-prod&response_type=token%20id_token&scope=account%20openid&nonce=1", + 'https://auth.riotgames.com/authorize?redirect_uri=https%3A%2F%2Fplayvalorant.com%2Fopt_in&client_id=play' + '-valorant-web-prod&response_type=token%20id_token&scope=account%20openid&nonce=1', cookies=cookies, allow_redirects=False, ) as r: @@ -294,17 +299,16 @@ async def redeem_cookies(self, cookies: Dict) -> Tuple[Dict[str, Any], str, str] await session.close() - accessToken, tokenId = _extract_tokens_from_uri(data) - entitlements_token = await self.get_entitlements_token(accessToken) - - return new_cookies, accessToken, entitlements_token + access_token, _token_id = _extract_tokens_from_uri(data) + entitlements_token = await self.get_entitlements_token(access_token) - async def temp_auth(self, username: str, password: str) -> Optional[Dict[str, Any]]: + return new_cookies, access_token, entitlements_token + async def temp_auth(self, username: str, password: str) -> dict[str, Any] | None: authenticate = await self.authenticate(username, password) - if authenticate['auth'] == 'response': - access_token = authenticate['data']['access_token'] - token_id = authenticate['data']['token_id'] + if authenticate['auth'] == 'response': # type: ignore + access_token = authenticate['data']['access_token'] # type: ignore + token_id = authenticate['data']['token_id'] # type: ignore entitlements_token = await self.get_entitlements_token(access_token) puuid, name, tag = await self.get_userinfo(access_token) @@ -323,25 +327,25 @@ async def temp_auth(self, username: str, password: str) -> Optional[Dict[str, An # next update - async def login_with_cookie(self, cookies: Dict) -> Dict[str, Any]: + async def login_with_cookie(self, cookies: dict[str, Any] | str) -> dict[str, Any]: """This function is used to log in with cookie.""" # language local_response = ResponseLanguage('cookies', self.locale_code) - cookie_payload = f'ssid={cookies};' if cookies.startswith('e') else cookies + cookie_payload = f'ssid={cookies};' if isinstance(cookies, str) and cookies.startswith('e') else cookies self._headers['cookie'] = cookie_payload session = ClientSession() r = await session.get( - "https://auth.riotgames.com/authorize" - "?redirect_uri=https%3A%2F%2Fplayvalorant.com%2Fopt_in" - "&client_id=play-valorant-web-prod" - "&response_type=token%20id_token" - "&scope=account%20openid" - "&nonce=1", + 'https://auth.riotgames.com/authorize' + '?redirect_uri=https%3A%2F%2Fplayvalorant.com%2Fopt_in' + '&client_id=play-valorant-web-prod' + '&response_type=token%20id_token' + '&scope=account%20openid' + '&nonce=1', allow_redirects=False, headers=self._headers, ) @@ -365,5 +369,5 @@ async def login_with_cookie(self, cookies: Dict) -> Dict[str, Any]: data = {'cookies': new_cookies, 'AccessToken': accessToken, 'token_id': tokenID, 'emt': entitlements_token} return data - async def refresh_token(self, cookies: Dict) -> Tuple[Dict[str, Any], str, str]: + async def refresh_token(self, cookies: dict) -> tuple[dict[str, Any], str, str]: return await self.redeem_cookies(cookies) diff --git a/utils/valorant/cache.py b/utils/valorant/cache.py index c742445..8cc2c65 100644 --- a/utils/valorant/cache.py +++ b/utils/valorant/cache.py @@ -2,32 +2,30 @@ import json import os -from typing import Dict, Optional +from typing import Any -# Standard import requests -# Local from .useful import JSON, on_replit -def create_json(filename: str, formats: Dict) -> None: +def create_json(filename: str, formats: dict[str, Any]) -> None: """Create a json file""" if on_replit: - from replit import db + from replit import db # type: ignore db[filename] = formats else: - file_path = f"data/" + filename + ".json" + file_path = 'data/' + filename + '.json' file_dir = os.path.dirname(file_path) os.makedirs(file_dir, exist_ok=True) if not os.path.exists(file_path): - with open(file_path, "w") as fp: + with open(file_path, 'w') as fp: json.dump(formats, fp, indent=2) -def get_valorant_version() -> Optional[str]: +def get_valorant_version() -> str | None: """Get the valorant version from valorant-api.com""" print('Fetching Valorant version !') @@ -43,7 +41,7 @@ def fetch_skin() -> None: data = JSON.read('cache') print('Fetching weapons skin !') - resp = requests.get(f'https://valorant-api.com/v1/weapons/skins?language=all') + resp = requests.get('https://valorant-api.com/v1/weapons/skins?language=all') if resp.status_code == 200: json = {} for skin in resp.json()['data']: @@ -95,7 +93,7 @@ def fetch_mission() -> None: data = JSON.read('cache') print('Fetching mission !') - resp = requests.get(f'https://valorant-api.com/v1/missions?language=all') + resp = requests.get('https://valorant-api.com/v1/missions?language=all') if resp.status_code == 200: json = {} # json['version'] = get_valorant_version() @@ -116,7 +114,7 @@ def fetch_playercard() -> None: data = JSON.read('cache') print('Fetching Player cards !') - resp = requests.get(f'https://valorant-api.com/v1/playercards?language=all') + resp = requests.get('https://valorant-api.com/v1/playercards?language=all') if resp.status_code == 200: payload = {} # json['version'] = get_valorant_version() @@ -140,7 +138,7 @@ def fetch_titles() -> None: data = JSON.read('cache') print('Fetching Player titles !') - resp = requests.get(f'https://valorant-api.com/v1/playertitles?language=all') + resp = requests.get('https://valorant-api.com/v1/playertitles?language=all') if resp.status_code == 200: payload = {} for title in resp.json()['data']: @@ -153,9 +151,8 @@ def fetch_spray() -> None: """Fetch the spray from valorant-api.com""" data = JSON.read('cache') - session = requests.session() print('Fetching Sprays !') - resp = requests.get(f'https://valorant-api.com/v1/sprays?language=all') + resp = requests.get('https://valorant-api.com/v1/sprays?language=all') if resp.status_code == 200: payload = {} for spray in resp.json()['data']: @@ -173,7 +170,7 @@ def fetch_bundles() -> None: data = JSON.read('cache') print('Fetching bundles !') - resp = requests.get(f'https://valorant-api.com/v1/bundles?language=all') + resp = requests.get('https://valorant-api.com/v1/bundles?language=all') if resp.status_code == 200: bundles = {} for bundle in resp.json()['data']: @@ -189,9 +186,9 @@ def fetch_bundles() -> None: 'expires': None, } - #resp2 = requests.get(f'https://api.valtracker.gg/bundles') + # resp2 = requests.get(f'https://api.valtracker.gg/bundles') - #for bundle2 in resp2.json()['data']: + # for bundle2 in resp2.json()['data']: # if bundle2['uuid'] in bundles: # bundle = bundles[bundle2.get('uuid')] # items = [] @@ -245,7 +242,7 @@ def fetch_contracts() -> None: data = JSON.read('cache') print('Fetching Contracts !') - resp = requests.get(f'https://valorant-api.com/v1/contracts?language=all') + resp = requests.get('https://valorant-api.com/v1/contracts?language=all') # IGNOR OLD BATTLE_PASS ignor_contract = [ @@ -258,14 +255,14 @@ def fetch_contracts() -> None: 'de37c775-4017-177a-8c64-a8bb414dae1f', # BP EP 3 ACT 1 'b0bd7062-4d62-1ff1-7920-b39622ee926b', # BP EP 3 ACT 2 'be540721-4d60-0675-a586-ecb14adcb5f7', # BP EP 3 ACT 3 - '60f2e13a-4834-0a18-5f7b-02b1a97b7adb' '60f2e13a-4834-0a18-5f7b-02b1a97b7adb' # BP EP 4 ACT 1 # BP EP 4 ACT 1 + '60f2e13a-4834-0a18-5f7b-02b1a97b7adb' '60f2e13a-4834-0a18-5f7b-02b1a97b7adb', # BP EP 4 ACT 1 # BP EP 4 ACT 1 # 'c1cd8895-4bd2-466d-e7ff-b489e3bc3775', # BP EP 4 ACT 2 ] if resp.status_code == 200: json = {} for contract in resp.json()['data']: - if not contract['uuid'] in ignor_contract: + if contract['uuid'] not in ignor_contract: json[contract['uuid']] = { 'uuid': contract['uuid'], 'free': contract['shipIt'], @@ -306,7 +303,7 @@ def fetch_currencies() -> None: data = JSON.read('cache') print('Fetching currencies !') - resp = requests.get(f'https://valorant-api.com/v1/currencies?language=all') + resp = requests.get('https://valorant-api.com/v1/currencies?language=all') if resp.status_code == 200: payload = {} for currencie in resp.json()['data']: @@ -326,7 +323,7 @@ def fetch_buddies() -> None: print('Fetching buddies !') - resp = requests.get(f'https://valorant-api.com/v1/buddies?language=all') + resp = requests.get('https://valorant-api.com/v1/buddies?language=all') if resp.status_code == 200: payload = {} for buddy in resp.json()['data']: @@ -340,14 +337,14 @@ def fetch_buddies() -> None: JSON.save('cache', data) -def fetch_price(data_price: Dict) -> None: +def fetch_price(data_price: dict) -> None: """Fetch the price of a skin""" data = JSON.read('cache') payload = {} for skin in data_price['Offers']: - if skin["OfferID"] in data['skins']: - (*cost,) = skin["Cost"].values() + if skin['OfferID'] in data['skins']: + (*cost,) = skin['Cost'].values() payload[skin['OfferID']] = cost[0] # prices['is_price'] = True data['prices'] = payload @@ -387,7 +384,7 @@ def fetch_price(data_price: Dict) -> None: def get_cache() -> None: """Get all cache from valorant-api.com""" - create_json('cache', {"valorant_version": get_valorant_version()}) + create_json('cache', {'valorant_version': get_valorant_version()}) fetch_skin() fetch_tier() diff --git a/utils/valorant/db.py b/utils/valorant/db.py index f1ffda9..96be0bf 100644 --- a/utils/valorant/db.py +++ b/utils/valorant/db.py @@ -1,7 +1,7 @@ from __future__ import annotations from datetime import datetime, timedelta -from typing import Any, Dict, Optional +from typing import Any from ..errors import DatabaseError from .auth import Auth @@ -10,7 +10,7 @@ from .useful import JSON -def timestamp_utc() -> datetime: +def timestamp_utc() -> float: return datetime.timestamp(datetime.utcnow()) @@ -21,25 +21,25 @@ def __init__(self) -> None: """Initialize database""" self.auth = Auth() - def insert_user(self, data: Dict) -> None: + def insert_user(self, data: dict[str, Any]) -> None: """Insert user""" JSON.save('users', data) - def read_db(self) -> Dict: + def read_db(self) -> dict[str, Any]: """Read database""" data = JSON.read('users') return data - def read_cache(self) -> Dict: + def read_cache(self) -> dict[str, Any]: """Read database""" data = JSON.read('cache') return data - def insert_cache(self, data: Dict) -> None: + def insert_cache(self, data: dict[str, Any]) -> None: """Insert cache""" JSON.save('cache', data) - async def is_login(self, user_id: int, response: Dict) -> Optional[Dict[str, Any]]: + async def is_login(self, user_id: int, response: dict[str, Any]) -> dict[str, Any] | bool | None: """Check if user is logged in""" db = self.read_db() @@ -53,7 +53,7 @@ async def is_login(self, user_id: int, response: Dict) -> Optional[Dict[str, Any return False return data - async def login(self, user_id: int, data: dict, locale_code: str) -> Optional[Dict[str, Any]]: + async def login(self, user_id: int, data: dict[str, Any], locale_code: str) -> dict[str, Any] | None: """Login to database""" # language @@ -75,18 +75,18 @@ async def login(self, user_id: int, data: dict, locale_code: str) -> Optional[Di expiry_token = datetime.timestamp(datetime.utcnow() + timedelta(minutes=59)) - data = dict( - cookie=cookie, - access_token=access_token, - token_id=token_id, - emt=entitlements_token, - puuid=puuid, - username=player_name, - region=region, - expiry_token=expiry_token, - notify_mode=None, - DM_Message=True, - ) + data = { + 'cookie': cookie, + 'access_token': access_token, + 'token_id': token_id, + 'emt': entitlements_token, + 'puuid': puuid, + 'username': player_name, + 'region': region, + 'expiry_token': expiry_token, + 'notify_mode': None, + 'DM_Message': True, + } db[str(user_id)] = data @@ -94,11 +94,11 @@ async def login(self, user_id: int, data: dict, locale_code: str) -> Optional[Di except Exception as e: print(e) - raise DatabaseError(response.get('LOGIN_ERROR')) + raise DatabaseError(response.get('LOGIN_ERROR')) from e else: return {'auth': True, 'player': player_name} - def logout(self, user_id: int, locale_code: str) -> Optional[bool]: + def logout(self, user_id: int, locale_code: str) -> bool | None: """Logout from database""" # language @@ -108,49 +108,49 @@ def logout(self, user_id: int, locale_code: str) -> Optional[bool]: db = self.read_db() del db[str(user_id)] self.insert_user(db) - except KeyError: - raise DatabaseError(response.get('LOGOUT_ERROR')) + except KeyError as e: + raise DatabaseError(response.get('LOGOUT_ERROR')) from e except Exception as e: print(e) - raise DatabaseError(response.get('LOGOUT_EXCEPT')) + raise DatabaseError(response.get('LOGOUT_EXCEPT')) from e else: return True - async def is_data(self, user_id: int, locale_code: str = 'en-US') -> Optional[Dict[str, Any]]: + async def is_data(self, user_id: int, locale_code: str = 'en-US') -> dict[str, Any] | None: """Check if user is registered""" response = LocalErrorResponse('DATABASE', locale_code) auth = await self.is_login(user_id, response) - puuid = auth['puuid'] - region = auth['region'] - username = auth['username'] - access_token = auth['access_token'] - entitlements_token = auth['emt'] - notify_mode = auth['notify_mode'] - expiry_token = auth['expiry_token'] - cookie = auth['cookie'] - notify_channel = auth.get('notify_channel', None) - dm_message = auth.get('DM_Message', None) + puuid = auth['puuid'] # type: ignore + region = auth['region'] # type: ignore + username = auth['username'] # type: ignore + access_token = auth['access_token'] # type: ignore + entitlements_token = auth['emt'] # type: ignore + notify_mode = auth['notify_mode'] # type: ignore + expiry_token = auth['expiry_token'] # type: ignore + cookie = auth['cookie'] # type: ignore + notify_channel = auth.get('notify_channel', None) # type: ignore + dm_message = auth.get('DM_Message', None) # type: ignore if timestamp_utc() > expiry_token: - access_token, entitlements_token = await self.refresh_token(user_id, auth) + access_token, entitlements_token = await self.refresh_token(user_id, auth) # type: ignore headers = {'Authorization': f'Bearer {access_token}', 'X-Riot-Entitlements-JWT': entitlements_token} - data = dict( - puuid=puuid, - region=region, - headers=headers, - player_name=username, - notify_mode=notify_mode, - cookie=cookie, - notify_channel=notify_channel, - dm_message=dm_message, - ) + data = { + 'puuid': puuid, + 'region': region, + 'headers': headers, + 'player_name': username, + 'notify_mode': notify_mode, + 'cookie': cookie, + 'notify_channel': notify_channel, + 'dm_message': dm_message, + } return data - async def refresh_token(self, user_id: int, data: Dict) -> Optional[Dict]: + async def refresh_token(self, user_id: int, data: dict[str, Any]) -> tuple[str, str]: """Refresh token""" auth = self.auth @@ -169,17 +169,17 @@ async def refresh_token(self, user_id: int, data: Dict) -> Optional[Dict]: return access_token, entitlements_token - def change_notify_mode(self, user_id: int, mode: str = None) -> None: + def change_notify_mode(self, user_id: int, mode: str | None = None) -> None: """Change notify mode""" db = self.read_db() overite_mode = {'All Skin': 'All', 'Specified Skin': 'Specified', 'Off': None} - db[str(user_id)]['notify_mode'] = overite_mode[mode] + db[str(user_id)]['notify_mode'] = overite_mode[mode] # type: ignore self.insert_user(db) - def change_notify_channel(self, user_id: int, channel: str, channel_id: int = None) -> None: + def change_notify_channel(self, user_id: int, channel: str, channel_id: int | None = None) -> None: """Change notify mode""" db = self.read_db() @@ -195,18 +195,18 @@ def change_notify_channel(self, user_id: int, channel: str, channel_id: int = No def check_notify_list(self, user_id: int) -> None: database = JSON.read('notifys') - notify_skin = [x for x in database if x['id'] == str(user_id)] + notify_skin = [x for x in database if x['id'] == str(user_id)] # type: ignore if len(notify_skin) == 0: raise DatabaseError("You're notification list is empty!") - def get_user_is_notify(self) -> Dict[str, Any]: + def get_user_is_notify(self) -> list[Any]: """Get user is notify""" database = JSON.read('users') notifys = [user_id for user_id in database if database[user_id]['notify_mode'] is not None] return notifys - def insert_skin_price(self, skin_price: Dict, force=False) -> None: + def insert_skin_price(self, skin_price: dict[str, Any], force: bool = False) -> None: """Insert skin price to cache""" cache = self.read_cache() @@ -215,7 +215,7 @@ def insert_skin_price(self, skin_price: Dict, force=False) -> None: if check_price is False or force: fetch_price(skin_price) - async def cookie_login(self, user_id: int, cookie: Optional[str], locale_code: str) -> Optional[Dict[str, Any]]: + async def cookie_login(self, user_id: int, cookie: dict[str, Any] | str, locale_code: str) -> dict[str, Any] | None: """Login with cookie""" db = self.read_db() @@ -236,18 +236,18 @@ async def cookie_login(self, user_id: int, cookie: Optional[str], locale_code: s expiry_token = datetime.timestamp(datetime.utcnow() + timedelta(minutes=59)) try: - data = dict( - cookie=cookie, - access_token=access_token, - token_id=token_id, - emt=entitlements_token, - puuid=puuid, - username=player_name, - region=region, - expiry_token=expiry_token, - notify_mode=None, - DM_Message=True, - ) + data = { + 'cookie': cookie, + 'access_token': access_token, + 'token_id': token_id, + 'emt': entitlements_token, + 'puuid': puuid, + 'username': player_name, + 'region': region, + 'expiry_token': expiry_token, + 'notify_mode': None, + 'DM_Message': True, + } db[str(user_id)] = data self.insert_user(db) diff --git a/utils/valorant/embed.py b/utils/valorant/embed.py index 2d93fb8..41e884b 100644 --- a/utils/valorant/embed.py +++ b/utils/valorant/embed.py @@ -2,7 +2,7 @@ import contextlib from datetime import datetime, timedelta -from typing import TYPE_CHECKING, Any, Dict, List, Union +from typing import TYPE_CHECKING, Any import discord @@ -16,12 +16,13 @@ class Embed(discord.Embed): # Custom Embed - def __init__(self, description: str = None, color: Union[discord.Color, int] = 0xFD4554, **kwargs: Any) -> None: + def __init__(self, description: str | None = None, color: discord.Color | int = 0xFD4554, **kwargs: Any) -> None: super().__init__(description=description, color=color, **kwargs) class GetEmbed: - def __giorgio_embed(skin: Dict, bot: ValorantBot) -> discord.Embed: + @staticmethod + def __giorgio_embed(skin: dict[str, Any], bot: ValorantBot) -> discord.Embed: """EMBED DESIGN Giorgio""" uuid, name, price, icon = skin['uuid'], skin['name'], skin['price'], skin['icon'] @@ -29,12 +30,18 @@ def __giorgio_embed(skin: Dict, bot: ValorantBot) -> discord.Embed: vp_emoji = GetEmoji.point_by_bot('ValorantPointIcon', bot) - embed = Embed(f"{emoji} **{name}**\n{vp_emoji} {price}", color=0x0F1923) + embed = Embed(f'{emoji} **{name}**\n{vp_emoji} {price}', color=0x0F1923) embed.set_thumbnail(url=icon) return embed @classmethod - def store(cls, player: str, offer: Dict, response: Dict, bot: ValorantBot) -> List[discord.Embed]: + def store( + cls, + player: str, + offer: dict[str, Any], + response: dict[str, Any], + bot: ValorantBot, + ) -> list[discord.Embed]: """Embed Store""" store_esponse = response.get('RESPONSE') @@ -43,19 +50,20 @@ def store(cls, player: str, offer: Dict, response: Dict, bot: ValorantBot) -> Li duration = data.pop('duration') - description = store_esponse.format( + description = store_esponse.format( # type: ignore username=player, duration=format_relative(datetime.utcnow() + timedelta(seconds=duration)) ) embed = Embed(description) embeds = [embed] - [embeds.append(cls.__giorgio_embed(data[skin], bot)) for skin in data] + [embeds.append(cls.__giorgio_embed(data[skin], bot)) for skin in data] # type: ignore - return embeds + return embeds # type: ignore # ---------- MISSION EMBED ---------- # - def mission(player: str, mission: Dict, response: Dict) -> discord.Embed: + @staticmethod + def mission(player: str, mission: dict[str, Any], response: dict[str, Any]) -> discord.Embed: """Embed Mission""" # language @@ -82,20 +90,20 @@ def mission(player: str, mission: Dict, response: Dict) -> discord.Embed: weekly_end_time = '' with contextlib.suppress(Exception): - weekly_end_time = f"{refill_in.format(duration=format_relative(iso_to_time(weekly_end)))}" + weekly_end_time = f'{refill_in.format(duration=format_relative(iso_to_time(weekly_end)))}' # type: ignore - embed = Embed(title=f"**{title_mission}**") + embed = Embed(title=f'**{title_mission}**') embed.set_footer(text=player) if len(daily) != 0: embed.add_field( - name=f"**{title_daily}**", - value=f"{daily}\n{reset_in.format(duration=format_relative(iso_to_time(daily_end)))}", + name=f'**{title_daily}**', + value=f'{daily}\n{reset_in.format(duration=format_relative(iso_to_time(daily_end)))}', # type: ignore inline=False, ) if len(weekly) != 0: - embed.add_field(name=f"**{title_weekly}**", value=f"{weekly}\n\n{weekly_end_time}", inline=False) + embed.add_field(name=f'**{title_weekly}**', value=f'{weekly}\n\n{weekly_end_time}', inline=False) if len(new_player) != 0: - embed.add_field(name=f"**{title_new_player}**", value=f"{new_player}", inline=False) + embed.add_field(name=f'**{title_new_player}**', value=f'{new_player}', inline=False) if len(embed.fields) == 0: embed.color = 0x77DD77 embed.description = clear_all_mission @@ -104,7 +112,8 @@ def mission(player: str, mission: Dict, response: Dict) -> discord.Embed: # ---------- POINT EMBED ---------- # - def point(player: str, wallet: Dict, response: Dict, bot: ValorantBot) -> discord.Embed: + @staticmethod + def point(player: str, wallet: dict[str, Any], response: dict[str, Any], bot: ValorantBot) -> discord.Embed: """Embed Point""" # language @@ -124,20 +133,21 @@ def point(player: str, wallet: Dict, response: Dict, bot: ValorantBot) -> discor if vp == 'VP': vp = 'Valorant Points' - embed = Embed(title=f"{title_point}:") + embed = Embed(title=f'{title_point}:') vp_emoji = GetEmoji.point_by_bot('ValorantPointIcon', bot) rad_emoji = GetEmoji.point_by_bot('RadianitePointIcon', bot) - embed.add_field(name=vp, value=f"{vp_emoji} {valorant_point}") - embed.add_field(name=rad, value=f"{rad_emoji} {radiant_point}") + embed.add_field(name=vp, value=f'{vp_emoji} {valorant_point}') + embed.add_field(name=rad, value=f'{rad_emoji} {radiant_point}') embed.set_footer(text=player) return embed # ---------- NIGHT MARKET EMBED ---------- # - def __nightmarket_embed(skins: Dict, bot: ValorantBot) -> discord.Embed: + @staticmethod + def __nightmarket_embed(skins: dict[str, Any], bot: ValorantBot) -> discord.Embed: """Generate Embed Night Market""" uuid, name, icon, price, dpice = skins['uuid'], skins['name'], skins['icon'], skins['price'], skins['disprice'] @@ -145,12 +155,14 @@ def __nightmarket_embed(skins: Dict, bot: ValorantBot) -> discord.Embed: emoji = GetEmoji.tier_by_bot(uuid, bot) vp_emoji = GetEmoji.point_by_bot('ValorantPointIcon', bot) - embed = Embed(f"{emoji} **{name}**\n{vp_emoji} {dpice} ~~{price}~~", color=0x0F1923) + embed = Embed(f'{emoji} **{name}**\n{vp_emoji} {dpice} ~~{price}~~', color=0x0F1923) embed.set_thumbnail(url=icon) return embed @classmethod - def nightmarket(cls, player: str, offer: Dict, bot: ValorantBot, response: Dict) -> discord.Embed: + def nightmarket( + cls, player: str, offer: dict[str, Any], bot: ValorantBot, response: dict[str, Any] + ) -> discord.Embed: """Embed Night Market""" # language @@ -160,27 +172,34 @@ def nightmarket(cls, player: str, offer: Dict, bot: ValorantBot, response: Dict) skins = night_mk['nightmarket'] duration = night_mk['duration'] - description = msg_response.format( - username=player, duration=format_relative(datetime.utcnow() + timedelta(seconds=duration)) + description = msg_response.format( # type: ignore + username=player, + duration=format_relative(datetime.utcnow() + timedelta(seconds=duration)), ) embed = Embed(description) embeds = [embed] - [embeds.append(cls.__nightmarket_embed(skins[skin], bot)) for skin in skins] + [embeds.append(cls.__nightmarket_embed(skins[skin], bot)) for skin in skins] # type: ignore - return embeds + return embeds # type: ignore # ---------- BATTLEPASS EMBED ---------- # - def battlepass(player: str, data: Dict, season: Dict, response: Dict) -> discord.Embed: + @staticmethod + def battlepass( + player: str, + data: dict[str, Any], + season: dict[str, Any], + response: dict[str, Any], + ) -> discord.Embed: """Embed Battle-pass""" # language MSG_RESPONSE = response.get('RESPONSE') MSG_TIER = response.get('TIER') - BTP = GetFormat.battlepass_format(data, season, response) + BTP = GetFormat.battlepass_format(data, season, response) # type: ignore item = BTP['data'] reward = item['reward'] @@ -192,14 +211,14 @@ def battlepass(player: str, data: Dict, season: Dict, response: Dict) -> discord item_type = item['type'] original_type = item['original_type'] - description = MSG_RESPONSE.format( + description = MSG_RESPONSE.format( # type: ignore next=f'`{reward}`', type=f'`{item_type}`', xp=f'`{xp:,}/{calculate_level_xp(tier + 1):,}`', end=format_relative(season_end), ) - embed = Embed(description, title=f"BATTLEPASS") + embed = Embed(description, title='BATTLEPASS') if icon: if original_type in ['PlayerCard', 'EquippableSkinLevel']: @@ -213,29 +232,35 @@ def battlepass(player: str, data: Dict, season: Dict, response: Dict) -> discord if tier == 55: embed.description = str(reward) - embed.set_footer(text=f"{MSG_TIER} {tier} | {act}\n{player}") + embed.set_footer(text=f'{MSG_TIER} {tier} | {act}\n{player}') return embed # ---------- NOTIFY EMBED ---------- # - def notify_specified_send(uuid: str) -> discord.Embed: - ... + @staticmethod + def notify_specified_send(uuid: str) -> discord.Embed: ... @classmethod - def notify_all_send(cls, player: str, offer: Dict, response: Dict, bot: ValorantBot) -> discord.Embed: - + def notify_all_send( + cls, + player: str, + offer: dict[str, Any], + response: dict[str, Any], + bot: ValorantBot, + ) -> list[discord.Embed]: description_format = response.get('RESPONSE_ALL') data = GetFormat.offer_format(offer) duration = data.pop('duration') - description = description_format.format( - username=player, duration=format_relative(datetime.utcnow() + timedelta(seconds=duration)) + description = description_format.format( # type: ignore + username=player, + duration=format_relative(datetime.utcnow() + timedelta(seconds=duration)), ) embed = Embed(description) embeds = [embed] - [embeds.append(cls.__giorgio_embed(data[skin], bot)) for skin in data] + [embeds.append(cls.__giorgio_embed(data[skin], bot)) for skin in data] # type: ignore - return embeds + return embeds # type: ignore diff --git a/utils/valorant/endpoint.py b/utils/valorant/endpoint.py index 560b68e..a872935 100644 --- a/utils/valorant/endpoint.py +++ b/utils/valorant/endpoint.py @@ -4,18 +4,21 @@ # Standard import json -from typing import Any, Dict, Mapping +from typing import Any import requests -import urllib3 from ..errors import HandshakeError, ResponseError from .local import LocalErrorResponse # Local -from .resources import base_endpoint, base_endpoint_glz, base_endpoint_shared, region_shard_override, shard_region_override - -urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) +from .resources import ( + base_endpoint, + base_endpoint_glz, + base_endpoint_shared, + region_shard_override, + shard_region_override, +) class API_ENDPOINT: @@ -39,7 +42,7 @@ def __init__(self) -> None: # language self.locale_code = 'en-US' - def activate(self, auth: Mapping[str, Any]) -> None: + def activate(self, auth: dict[str, Any]) -> None: """activate api""" try: @@ -54,9 +57,9 @@ def activate(self, auth: Mapping[str, Any]) -> None: self.__build_urls() except Exception as e: print(e) - raise HandshakeError(self.locale_response().get('FAILED_ACTIVE')) + raise HandshakeError(self.locale_response().get('FAILED_ACTIVE')) from e - def locale_response(self) -> LocalErrorResponse: + def locale_response(self) -> dict[str, Any]: """This function is used to check if the local response is enabled.""" self.response = LocalErrorResponse('API', self.locale_code) return self.response @@ -67,7 +70,7 @@ def locale_response(self) -> LocalErrorResponse: # self.__build_headers() - def fetch(self, endpoint: str = '/', url: str = 'pd', errors: Dict = {}) -> Dict: + def fetch(self, endpoint: str = '/', url: str = 'pd', errors: dict[str, Any] | None = None) -> dict[str, Any]: """fetch data from the api""" self.locale_response() @@ -78,41 +81,45 @@ def fetch(self, endpoint: str = '/', url: str = 'pd', errors: Dict = {}) -> Dict r = requests.get(f'{endpoint_url}{endpoint}', headers=self.headers) - try: + try: # noqa: SIM105 data = json.loads(r.text) - except: # as no data is set, an exception will be raised later in the method + except Exception: pass - if "httpStatus" not in data: - return data + if 'httpStatus' not in data: # type: ignore + return data # type: ignore - if data["httpStatus"] == 400: + if r.status_code == 400: response = LocalErrorResponse('AUTH', self.locale_code) raise ResponseError(response.get('COOKIES_EXPIRED')) # await self.refresh_token() # return await self.fetch(endpoint=endpoint, url=url, errors=errors) - - def put(self, endpoint: str = "/", url: str = 'pd', data: Dict = {}, errors: Dict = {}) -> Dict: + return {} + + def put( + self, + endpoint: str = '/', + url: str = 'pd', + data: dict[str, Any] | list[Any] | None = None, + errors: dict[str, Any] | None = None, + ) -> Any: """put data to the api""" self.locale_response() - data = data if type(data) is list else json.dumps(data) - endpoint_url = getattr(self, url) - data = None r = requests.put(f'{endpoint_url}{endpoint}', headers=self.headers, data=data) data = json.loads(r.text) - if data is not None: - return data - else: + if data is None: raise ResponseError(self.response.get('REQUEST_FAILED')) + return data + # contracts endpoints - def fetch_contracts(self) -> Mapping[str, Any]: + def fetch_contracts(self) -> dict[str, Any]: """ Contracts_Fetch Get a list of contracts and completion status including match history @@ -122,7 +129,7 @@ def fetch_contracts(self) -> Mapping[str, Any]: # PVP endpoints - def fetch_content(self) -> Mapping[str, Any]: + def fetch_content(self) -> dict[str, Any]: """ Content_FetchContent Get names and ids for game content such as agents, maps, guns, etc. @@ -130,7 +137,7 @@ def fetch_content(self) -> Mapping[str, Any]: data = self.fetch(endpoint='/content-service/v3/content', url='shared') return data - def fetch_account_xp(self) -> Mapping[str, Any]: + def fetch_account_xp(self) -> dict[str, Any]: """ AccountXP_GetPlayer Get the account level, XP, and XP history for the active player @@ -138,12 +145,12 @@ def fetch_account_xp(self) -> Mapping[str, Any]: data = self.fetch(endpoint=f'/account-xp/v1/players/{self.puuid}', url='pd') return data - def fetch_player_mmr(self, puuid: str = None) -> Mapping[str, Any]: + def fetch_player_mmr(self, puuid: str | None = None) -> dict[str, Any]: puuid = self.__check_puuid(puuid) data = self.fetch(endpoint=f'/mmr/v1/players/{puuid}', url='pd') return data - def fetch_name_by_puuid(self, puuid: str = None) -> Mapping[str, Any]: + def fetch_name_by_puuid(self, puuid: str | None = None) -> dict[str, Any]: """ Name_service get player name tag by puuid @@ -151,13 +158,13 @@ def fetch_name_by_puuid(self, puuid: str = None) -> Mapping[str, Any]: format ['PUUID'] """ if puuid is None: - puuid = [self.__check_puuid()] + puuids = [self.__check_puuid()] elif puuid is not None and type(puuid) is str: - puuid = [puuid] - data = self.put(endpoint='/name-service/v2/players', url='pd', body=puuid) + puuids = [puuid] + data = self.put(endpoint='/name-service/v2/players', url='pd', data=puuids) return data - def fetch_player_loadout(self) -> Mapping[str, Any]: + def fetch_player_loadout(self) -> dict[str, Any]: """ playerLoadoutUpdate Get the player's current loadout @@ -165,17 +172,17 @@ def fetch_player_loadout(self) -> Mapping[str, Any]: data = self.fetch(endpoint=f'/personalization/v2/players/{self.puuid}/playerloadout', url='pd') return data - def put_player_loadout(self, loadout: Mapping) -> Mapping[str, Any]: + def put_player_loadout(self, loadout: dict[str, Any]) -> dict[str, Any]: """ playerLoadoutUpdate Use the values from `fetch_player_loadout` excluding properties like `subject` and `version.` Loadout changes take effect when starting a new game """ - data = self.put(endpoint=f'/personalization/v2/players/{self.puuid}/playerloadout', url='pd', body=loadout) + data = self.put(endpoint=f'/personalization/v2/players/{self.puuid}/playerloadout', url='pd', data=loadout) return data # store endpoints - def store_fetch_offers(self) -> Mapping[str, Any]: + def store_fetch_offers(self) -> dict[str, Any]: """ Store_GetOffers Get prices for all store items @@ -183,7 +190,7 @@ def store_fetch_offers(self) -> Mapping[str, Any]: data = self.fetch('/store/v1/offers/', url='pd') return data - def store_fetch_storefront(self) -> Mapping[str, Any]: + def store_fetch_storefront(self) -> dict[str, Any]: """ Store_GetStorefrontV2 Get the currently available items in the store @@ -191,7 +198,7 @@ def store_fetch_storefront(self) -> Mapping[str, Any]: data = self.fetch(f'/store/v2/storefront/{self.puuid}', url='pd') return data - def store_fetch_wallet(self) -> Mapping[str, Any]: + def store_fetch_wallet(self) -> dict[str, Any]: """ Store_GetWallet Get amount of Valorant points and Radiant points the player has @@ -200,7 +207,7 @@ def store_fetch_wallet(self) -> Mapping[str, Any]: data = self.fetch(f'/store/v1/wallet/{self.puuid}', url='pd') return data - def store_fetch_order(self, order_id: str) -> Mapping[str, Any]: + def store_fetch_order(self, order_id: str) -> dict[str, Any]: """ Store_GetOrder {order id}: The ID of the order. Can be obtained when creating an order. @@ -208,7 +215,7 @@ def store_fetch_order(self, order_id: str) -> Mapping[str, Any]: data = self.fetch(f'/store/v1/order/{order_id}', url='pd') return data - def store_fetch_entitlements(self, item_type: Mapping) -> Mapping[str, Any]: + def store_fetch_entitlements(self, item_type: dict) -> dict[str, Any]: """ Store_GetEntitlements List what the player owns (agents, skins, buddies, ect.) @@ -225,27 +232,27 @@ def store_fetch_entitlements(self, item_type: Mapping) -> Mapping[str, Any]: '3ad1b2b2-acdb-4524-852f-954a76ddae0a': 'Skins chroma',\n 'de7caa6b-adf7-4588-bbd1-143831e786c6': 'Player titles',\n """ - data = self.fetch(endpoint=f"/store/v1/entitlements/{self.puuid}/{item_type}", url="pd") + data = self.fetch(endpoint=f'/store/v1/entitlements/{self.puuid}/{item_type}', url='pd') return data # useful endpoints - def fetch_mission(self) -> Mapping[str, Any]: + def fetch_mission(self) -> dict[str, Any]: """ Get player daily/weekly missions """ data = self.fetch_contracts() - mission = data["Missions"] + mission = data['Missions'] return mission - def get_player_level(self) -> Mapping[str, Any]: + def get_player_level(self) -> dict[str, Any]: """ Aliases `fetch_account_xp` but received a level """ data = self.fetch_account_xp()['Progress']['Level'] return data - def get_player_tier_rank(self, puuid: str = None) -> str: + def get_player_tier_rank(self, puuid: str | None = None) -> str: """ get player current tier rank """ @@ -253,7 +260,7 @@ def get_player_tier_rank(self, puuid: str = None) -> str: season_id = data['LatestCompetitiveUpdate']['SeasonID'] if len(season_id) == 0: season_id = self.__get_live_season() - current_season = data["QueueSkills"]['competitive']['SeasonalInfoBySeasonID'] + current_season = data['QueueSkills']['competitive']['SeasonalInfoBySeasonID'] current_Tier = current_season[season_id]['CompetitiveTier'] return current_Tier @@ -262,16 +269,16 @@ def get_player_tier_rank(self, puuid: str = None) -> str: def __get_live_season(self) -> str: """Get the UUID of the live competitive season""" content = self.fetch_content() - season_id = [season["ID"] for season in content["Seasons"] if season["IsActive"] and season["Type"] == "act"] + season_id = [season['ID'] for season in content['Seasons'] if season['IsActive'] and season['Type'] == 'act'] if not season_id: - return self.fetch_player_mmr()["LatestCompetitiveUpdate"]["SeasonID"] + return self.fetch_player_mmr()['LatestCompetitiveUpdate']['SeasonID'] return season_id[0] - def __check_puuid(self, puuid: str) -> str: + def __check_puuid(self, puuid: str | None = None) -> str: """If puuid passed into method is None make it current user's puuid""" return self.puuid if puuid is None else puuid - def __build_urls(self) -> str: + def __build_urls(self) -> None: """ generate URLs based on region/shard """ @@ -279,9 +286,8 @@ def __build_urls(self) -> str: self.shared = base_endpoint_shared.format(shard=self.shard) self.glz = base_endpoint_glz.format(region=self.region, shard=self.shard) - def __build_headers(self, headers: Mapping) -> Mapping[str, Any]: + def __build_headers(self, headers: dict[str, Any]) -> dict[str, Any]: """build headers""" - headers['X-Riot-ClientPlatform'] = self.client_platform headers['X-Riot-ClientVersion'] = self._get_client_version() return headers @@ -290,9 +296,9 @@ def __format_region(self) -> None: """Format region to match from user input""" self.shard = self.region - if self.region in region_shard_override.keys(): + if self.region in region_shard_override: self.shard = region_shard_override[self.region] - if self.shard in shard_region_override.keys(): + if self.shard in shard_region_override: self.region = shard_region_override[self.shard] def _get_client_version(self) -> str: @@ -301,10 +307,10 @@ def _get_client_version(self) -> str: data = r.json()['data'] return f"{data['branch']}-shipping-{data['buildVersion']}-{data['version'].split('.')[3]}" # return formatted version string - def _get_valorant_version(self) -> str: + def _get_valorant_version(self) -> str | None: """Get the valorant version""" r = requests.get('https://valorant-api.com/v1/version') - if r.status != 200: + if r.status_code != 200: return None data = r.json()['data'] return data['version'] diff --git a/utils/valorant/local.py b/utils/valorant/local.py index 8ade931..2e08f14 100644 --- a/utils/valorant/local.py +++ b/utils/valorant/local.py @@ -6,7 +6,7 @@ import contextlib import json -from typing import Any, Dict +from typing import Any # credit by /giorgi-o/ @@ -30,21 +30,21 @@ } -def InteractionLanguage(local_code: str) -> Dict[str, Any]: +def InteractionLanguage(local_code: str) -> str: return Locale.get(str(local_code), 'en-US') -def __LocalRead(filename: str) -> Dict: +def __LocalRead(filename: str) -> dict: data = {} try: - with open(f"languages/{filename}.json", "r", encoding='utf-8') as json_file: + with open(f'languages/{filename}.json', encoding='utf-8') as json_file: data = json.load(json_file) except FileNotFoundError: return __LocalRead('en-US') return data -def ResponseLanguage(command_name: str, local_code: str) -> Dict[str, Any]: +def ResponseLanguage(command_name: str, local_code: str) -> dict[str, Any]: local_code = __verify_localcode(local_code) local = {} with contextlib.suppress(KeyError): @@ -53,7 +53,7 @@ def ResponseLanguage(command_name: str, local_code: str) -> Dict[str, Any]: return local -def LocalErrorResponse(value: str, local_code: str) -> Dict[str, Any]: +def LocalErrorResponse(value: str, local_code: str) -> dict[str, Any]: local_code = __verify_localcode(local_code) local = {} with contextlib.suppress(KeyError): diff --git a/utils/valorant/resources.py b/utils/valorant/resources.py index cd09561..0ecac5d 100644 --- a/utils/valorant/resources.py +++ b/utils/valorant/resources.py @@ -1,7 +1,7 @@ from __future__ import annotations from io import BytesIO -from typing import TYPE_CHECKING, Optional +from typing import TYPE_CHECKING import discord import requests @@ -15,16 +15,16 @@ # ------------------- # # credit https://github.com/colinhartigan/ -base_endpoint = "https://pd.{shard}.a.pvp.net" -base_endpoint_glz = "https://glz-{region}-1.{shard}.a.pvp.net" -base_endpoint_shared = "https://shared.{shard}.a.pvp.net" +base_endpoint = 'https://pd.{shard}.a.pvp.net' +base_endpoint_glz = 'https://glz-{region}-1.{shard}.a.pvp.net' +base_endpoint_shared = 'https://shared.{shard}.a.pvp.net' -regions: list = ["na", "eu", "latam", "br", "ap", "kr", "pbe"] +regions: list = ['na', 'eu', 'latam', 'br', 'ap', 'kr', 'pbe'] region_shard_override = { - "latam": "na", - "br": "na", + 'latam': 'na', + 'br': 'na', } -shard_region_override = {"pbe": "na"} +shard_region_override = {'pbe': 'na'} # ------------------- # @@ -62,16 +62,20 @@ 'emoji': '<:Select:950376833982021662>', 'color': 0x5A9FE2, }, - '411e4a55-4e59-7757-41f0-86a53f101bb5': {'name': 'UltraTier', 'emoji': '<:Ultra:950376896745586719>', 'color': 0xEFEB65}, + '411e4a55-4e59-7757-41f0-86a53f101bb5': { + 'name': 'UltraTier', + 'emoji': '<:Ultra:950376896745586719>', + 'color': 0xEFEB65, + }, } points = { - 'ValorantPointIcon': f'<:ValorantPoint:950365917613817856>', - 'RadianitePointIcon': f'<:RadianitePoint:950365909636235324>', + 'ValorantPointIcon': '<:ValorantPoint:950365917613817856>', + 'RadianitePointIcon': '<:RadianitePoint:950365909636235324>', } -def get_item_type(uuid: str) -> Optional[str]: +def get_item_type(uuid: str) -> str | None: """Get item type""" item_type = { '01bb38e1-da47-4e6a-9b3d-945fe4655707': 'Agents', @@ -83,10 +87,10 @@ def get_item_type(uuid: str) -> Optional[str]: '3ad1b2b2-acdb-4524-852f-954a76ddae0a': 'Skins chroma', 'de7caa6b-adf7-4588-bbd1-143831e786c6': 'Player titles', } - return item_type.get(uuid, None) + return item_type.get(uuid) -def __url_to_image(url) -> Optional[bytes]: +def __url_to_image(url: str) -> bytes | None: session = requests.session() r = session.get(url) @@ -104,10 +108,10 @@ async def setup_emoji(bot: ValorantBot, guild: discord.Guild, local_code: str, f emoji = discord.utils.get(bot.emojis, name=name) if not emoji: try: - emoji = await guild.create_custom_emoji(name=name, image=__url_to_image(emoji_url)) - except discord.Forbidden: + emoji = await guild.create_custom_emoji(name=name, image=__url_to_image(emoji_url)) # type: ignore + except discord.Forbidden as e: if force: - raise ValorantBotError(response.get('MISSING_PERM')) + raise ValorantBotError(response.get('MISSING_PERM')) from e continue except discord.HTTPException: print(response.get('FAILED_CREATE_EMOJI')) diff --git a/utils/valorant/useful.py b/utils/valorant/useful.py index 7bbf5e8..5473c96 100644 --- a/utils/valorant/useful.py +++ b/utils/valorant/useful.py @@ -4,8 +4,8 @@ import json import os import uuid -from datetime import datetime, timezone -from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple +from datetime import UTC, datetime +from typing import TYPE_CHECKING, Any import discord from dotenv import load_dotenv @@ -16,7 +16,7 @@ load_dotenv() global on_replit -on_replit = True if os.getenv('ON_REPLIT') else False +on_replit = bool(os.getenv('ON_REPLIT')) VLR_locale = ValorantTranslator() @@ -41,11 +41,11 @@ def is_valid_uuid(value: str) -> bool: # ---------- ACT SEASON ---------- # -def get_season_by_content(content: Dict) -> Tuple[str, str]: +def get_season_by_content(content: dict) -> dict[str, Any]: """Get season id by content""" try: - season_data = [season for season in content["Seasons"] if season["IsActive"] and season["Type"] == "act"] + season_data = [season for season in content['Seasons'] if season['IsActive'] and season['Type'] == 'act'] season_id = season_data[0]['ID'] season_end = iso_to_time(season_data[0]['EndTime']) @@ -71,18 +71,18 @@ def calculate_level_xp(level: int) -> int: # https://github.com/giorgi-o # ---------- TIME UTILS ---------- # -def iso_to_time(iso: datetime) -> datetime: +def iso_to_time(iso: str) -> datetime: """Convert ISO time to datetime""" - timestamp = datetime.strptime(iso, "%Y-%m-%dT%H:%M:%S%z").timestamp() + timestamp = datetime.strptime(iso, '%Y-%m-%dT%H:%M:%S%z').timestamp() time = datetime.utcfromtimestamp(timestamp) return time -def format_dt(dt: datetime, style: str = None) -> str: # style 'R' or 'd' +def format_dt(dt: datetime, style: str | None = None) -> str: """datatime to time format""" if dt.tzinfo is None: - dt = dt.replace(tzinfo=timezone.utc) + dt = dt.replace(tzinfo=UTC) if style is None: return f'' @@ -107,15 +107,16 @@ def data_folder() -> None: class JSON: - def read(filename: str, force: bool = True) -> Dict: + @staticmethod + def read(filename: str, force: bool = True) -> dict[str, Any]: """Read json file""" try: if on_replit: - from replit import db + from replit import db # type: ignore data = db[filename] else: - with open("data/" + filename + ".json", "r", encoding='utf-8') as json_file: + with open('data/' + filename + '.json', encoding='utf-8') as json_file: data = json.load(json_file) except (FileNotFoundError, KeyError): from .cache import create_json @@ -125,15 +126,16 @@ def read(filename: str, force: bool = True) -> Dict: return JSON.read(filename, False) return data - def save(filename: str, data: Dict) -> None: + @staticmethod + def save(filename: str, data: dict[str, Any]) -> None: """Save data to json file""" try: if on_replit: - from replit import db + from replit import db # type: ignore db[filename] = data else: - with open("data/" + filename + ".json", 'w', encoding='utf-8') as json_file: + with open('data/' + filename + '.json', 'w', encoding='utf-8') as json_file: json.dump(data, json_file, indent=2, ensure_ascii=False) except (FileNotFoundError, KeyError): from .cache import create_json @@ -147,7 +149,7 @@ def save(filename: str, data: Dict) -> None: class GetItems: @classmethod - def get_item_by_type(cls, Itemtype: str, uuid: str) -> Dict[str, Any]: + def get_item_by_type(cls, Itemtype: str, uuid: str) -> dict[str, Any]: # type: ignore """Get item by type""" item_type = get_item_type(Itemtype) @@ -168,110 +170,120 @@ def get_item_by_type(cls, Itemtype: str, uuid: str) -> Dict[str, Any]: elif item_type == 'Player titles': return cls.get_title(uuid) - def get_skin(uuid: str) -> Dict[str, Any]: + @staticmethod + def get_skin(uuid: str) -> dict[str, Any]: """Get Skin data""" try: - skin_data = JSON.read('cache') - skin = skin_data["skins"][uuid] - except KeyError: - raise ValorantBotError('Some skin data is missing, plz use `/debug cache`') + skin = skin_data['skins'][uuid] + except KeyError as e: + raise ValorantBotError('Some skin data is missing, plz use `/debug cache`') from e return skin + @staticmethod def get_skin_price(uuid: str) -> str: """Get Skin price by skin uuid""" data = JSON.read('cache') - price = data["prices"] + price = data['prices'] try: cost = price[uuid] - except: + except Exception: cost = '-' return cost + @staticmethod def get_skin_tier_icon(skin: str) -> str: """Get Skin skin tier image""" skindata = JSON.read('cache') - tier_uuid = skindata["skins"][skin]['tier'] - tier = skindata['tiers'][tier_uuid]["icon"] + tier_uuid = skindata['skins'][skin]['tier'] + tier = skindata['tiers'][tier_uuid]['icon'] return tier - def get_spray(uuid: str) -> Dict[str, Any]: + @staticmethod + def get_spray(uuid: str) -> Any: """Get Spray""" data = JSON.read('cache') spray = None with contextlib.suppress(Exception): - spray = data["sprays"][uuid] + spray = data['sprays'][uuid] return spray - def get_title(uuid: str) -> Dict[str, Any]: + @staticmethod + def get_title(uuid: str) -> Any: """Get Title""" data = JSON.read('cache') title = None with contextlib.suppress(Exception): - title = data["titles"][uuid] + title = data['titles'][uuid] return title - def get_playercard(uuid: str) -> Dict[str, Any]: + @staticmethod + def get_playercard(uuid: str) -> Any: """Get Player card""" data = JSON.read('cache') title = None with contextlib.suppress(Exception): - title = data["playercards"][uuid] + title = data['playercards'][uuid] return title - def get_buddie(uuid: str) -> Dict: + @staticmethod + def get_buddie(uuid: str) -> Any: """Get Buddie""" data = JSON.read('cache') title = None with contextlib.suppress(Exception): - title = data["buddies"][uuid] + title = data['buddies'][uuid] return title - def get_skin_lvl_or_name(name: str, uuid: str) -> Dict[str, Any]: + @staticmethod + def get_skin_lvl_or_name(name: str, uuid: str) -> Any: """Get Skin uuid by name""" data = JSON.read('cache') skin = None with contextlib.suppress(Exception): - skin = data["skins"][uuid] + skin = data['skins'][uuid] with contextlib.suppress(Exception): if skin is None: - skin = [data["skins"][x] for x in data["skins"] if data["skins"][x]['name'] in name][0] + skin = [data['skins'][x] for x in data['skins'] if data['skins'][x]['name'] in name][0] return skin - def get_tier_name(skin_uuid: str) -> Optional[str]: + @staticmethod + def get_tier_name(skin_uuid: str) -> str | None: """Get tier name by skin uuid""" try: data = JSON.read('cache') uuid = data['skins'][skin_uuid]['tier'] name = data['tiers'][uuid]['name'] - except KeyError: - raise ValorantBotError('Some skin data is missing, plz use `/debug cache`') + except KeyError as e: + raise ValorantBotError('Some skin data is missing, plz use `/debug cache`') from e return name - def get_contract(uuid: str) -> Dict[str, Any]: + @staticmethod + def get_contract(uuid: str) -> Any: """Get contract by uuid""" data = JSON.read('cache') contract = None with contextlib.suppress(Exception): - contract = data["contracts"][uuid] + contract = data['contracts'][uuid] return contract - def get_bundle(uuid: str) -> Dict[str, Any]: + @staticmethod + def get_bundle(uuid: str) -> Any: """Get bundle by uuid""" data = JSON.read('cache') bundle = None with contextlib.suppress(Exception): - bundle = data["bundles"][uuid] + bundle = data['bundles'][uuid] return bundle @@ -279,6 +291,7 @@ def get_bundle(uuid: str) -> Dict[str, Any]: class GetEmoji: + @staticmethod def tier(skin_uuid: str) -> discord.Emoji: """Get tier emoji""" @@ -292,12 +305,13 @@ def tier(skin_uuid: str) -> discord.Emoji: def tier_by_bot(cls, skin_uuid: str, bot: ValorantBot) -> discord.Emoji: """Get tier emoji from bot""" - emoji = discord.utils.get(bot.emojis, name=GetItems.get_tier_name(skin_uuid) + 'Tier') + emoji = discord.utils.get(bot.emojis, name=GetItems.get_tier_name(skin_uuid) + 'Tier') # type: ignore if emoji is None: return cls.tier(skin_uuid) return emoji - def point_by_bot(point: str, bot: ValorantBot) -> discord.Emoji: + @staticmethod + def point_by_bot(point: str, bot: ValorantBot) -> discord.Emoji | str | None: """Get point emoji from bot""" emoji = discord.utils.get(bot.emojis, name=point) @@ -310,31 +324,31 @@ def point_by_bot(point: str, bot: ValorantBot) -> discord.Emoji: class GetFormat: - def offer_format(data: Dict) -> Dict: + @staticmethod + def offer_format(data: dict[str, Any]) -> dict[str, Any]: """Get skins list""" - offer_list = data["SkinsPanelLayout"]["SingleItemOffers"] - duration = data["SkinsPanelLayout"]["SingleItemOffersRemainingDurationInSeconds"] + offer_list = data['SkinsPanelLayout']['SingleItemOffers'] + duration = data['SkinsPanelLayout']['SingleItemOffersRemainingDurationInSeconds'] skin_count = 0 skin_source = {} - for uuid in offer_list: - skin = GetItems.get_skin(uuid) + for skin_count, skin_id in enumerate(offer_list): + skin = GetItems.get_skin(skin_id) name, icon = skin['names'][str(VLR_locale)], skin['icon'] - price = GetItems.get_skin_price(uuid) - tier_icon = GetItems.get_skin_tier_icon(uuid) + price = GetItems.get_skin_price(skin_id) + tier_icon = GetItems.get_skin_tier_icon(skin_id) if skin_count == 0: - skin1 = dict(name=name, icon=icon, price=price, tier=tier_icon, uuid=uuid) + skin1 = {'name': name, 'icon': icon, 'price': price, 'tier': tier_icon, 'uuid': skin_id} elif skin_count == 1: - skin2 = dict(name=name, icon=icon, price=price, tier=tier_icon, uuid=uuid) + skin2 = {'name': name, 'icon': icon, 'price': price, 'tier': tier_icon, 'uuid': skin_id} elif skin_count == 2: - skin3 = dict(name=name, icon=icon, price=price, tier=tier_icon, uuid=uuid) + skin3 = {'name': name, 'icon': icon, 'price': price, 'tier': tier_icon, 'uuid': skin_id} elif skin_count == 3: - skin4 = dict(name=name, icon=icon, price=price, tier=tier_icon, uuid=uuid) - skin_count += 1 + skin4 = {'name': name, 'icon': icon, 'price': price, 'tier': tier_icon, 'uuid': skin_id} skin_source = {'skin1': skin1, 'skin2': skin2, 'skin3': skin3, 'skin4': skin4, 'duration': duration} @@ -342,10 +356,11 @@ def offer_format(data: Dict) -> Dict: # ---------- UTILS FOR MISSION EMBED ---------- # - def mission_format(data: Dict) -> Dict[str, Any]: + @staticmethod + def mission_format(data: dict[str, Any]) -> dict[str, Any]: """Get mission format""" - mission = data["Missions"] + mission = data['Missions'] weekly = [] daily = [] @@ -356,7 +371,7 @@ def mission_format(data: Dict) -> Dict[str, Any]: except KeyError: weekly_end = '' - def get_mission_by_id(ID) -> Optional[str]: + def get_mission_by_id(ID: str) -> str | None: data = JSON.read('cache') mission = data['missions'][ID] return mission @@ -364,37 +379,43 @@ def get_mission_by_id(ID) -> Optional[str]: for m in mission: mission = get_mission_by_id(m['ID']) (*complete,) = m['Objectives'].values() - title = mission['titles'][str(VLR_locale)] - progress = mission['progress'] - xp = mission['xp'] + title = mission['titles'][str(VLR_locale)] # type: ignore + progress = mission['progress'] # type: ignore + xp = mission['xp'] # type: ignore - format_m = f"\n{title} | **+ {xp:,} XP**\n- **`{complete[0]}/{progress}`**" + format_m = f'\n{title} | **+ {xp:,} XP**\n- **`{complete[0]}/{progress}`**' - if mission['type'] == 'EAresMissionType::Weekly': + if mission['type'] == 'EAresMissionType::Weekly': # type: ignore weekly.append(format_m) - if mission['type'] == 'EAresMissionType::Daily': + if mission['type'] == 'EAresMissionType::Daily': # type: ignore daily_end = m['ExpirationTime'] daily.append(format_m) - if mission['type'] == 'EAresMissionType::NPE': + if mission['type'] == 'EAresMissionType::NPE': # type: ignore newplayer.append(format_m) - misson_data = dict(daily=daily, weekly=weekly, daily_end=daily_end, weekly_end=weekly_end, newplayer=newplayer) + misson_data = { + 'daily': daily, + 'weekly': weekly, + 'daily_end': daily_end, + 'weekly_end': weekly_end, + 'newplayer': newplayer, + } return misson_data # ---------- UTILS FOR NIGHTMARKET EMBED ---------- # - def nightmarket_format(offer: Dict, response: Dict) -> Dict[str, Any]: + @staticmethod + def nightmarket_format(offer: dict[str, Any], response: dict[str, Any]) -> dict[str, Any]: """Get Nightmarket offers""" try: night_offer = offer['BonusStore']['BonusStoreOffers'] - except KeyError: - raise ValorantBotError(response.get('NIGMARKET_HAS_END', 'Nightmarket has been ended')) + except KeyError as e: + raise ValorantBotError(response.get('NIGMARKET_HAS_END', 'Nightmarket has been ended')) from e duration = offer['BonusStore']['BonusStoreRemainingDurationInSeconds'] night_market = {} - count = 0 - for x in night_offer: + for count, x in enumerate(night_offer): count += 1 price = (*x['Offer']['Cost'].values(),) Disprice = (*x['DiscountCosts'].values(),) @@ -405,7 +426,7 @@ def nightmarket_format(offer: Dict, response: Dict) -> Dict[str, Any]: icon = skin['icon'] tier = GetItems.get_skin_tier_icon(uuid) - night_market['skin' + f'{count}'] = { + night_market['skin' + f'{count + 1}'] = { 'uuid': uuid, 'name': name, 'tier': tier, @@ -418,7 +439,8 @@ def nightmarket_format(offer: Dict, response: Dict) -> Dict[str, Any]: # ---------- UTILS FOR BATTLEPASS EMBED ---------- # - def __get_item_battlepass(type: str, uuid: str, response: Dict) -> Dict[str, Any]: + @staticmethod + def __get_item_battlepass(type: str, uuid: str, response: dict[str, Any]) -> dict[str, Any]: """Get item battle pass by type and uuid""" if type == 'Currency': @@ -426,52 +448,53 @@ def __get_item_battlepass(type: str, uuid: str, response: Dict) -> Dict[str, Any name = data['currencies'][uuid]['names'][str(VLR_locale)] icon = data['currencies'][uuid]['icon'] item_type = response.get('POINT', 'Point') - return {"success": True, "data": {'type': item_type, 'name': '10 ' + name, 'icon': icon}} + return {'success': True, 'data': {'type': item_type, 'name': '10 ' + name, 'icon': icon}} elif type == 'PlayerCard': data = JSON.read('cache') name = data['playercards'][uuid]['names'][str(VLR_locale)] icon = data['playercards'][uuid]['icon']['wide'] item_type = response.get('PLAYER_CARD', 'Player Card') - return {"success": True, "data": {'type': item_type, 'name': name, 'icon': icon}} + return {'success': True, 'data': {'type': item_type, 'name': name, 'icon': icon}} elif type == 'Title': data = JSON.read('cache') name = data['titles'][uuid]['names'][str(VLR_locale)] item_type = response.get('PLAYER_TITLE', 'Title') - return {"success": True, "data": {'type': item_type, 'name': name, 'icon': False}} + return {'success': True, 'data': {'type': item_type, 'name': name, 'icon': False}} elif type == 'Spray': data = JSON.read('cache') name = data['sprays'][uuid]['names'][str(VLR_locale)] icon = data['sprays'][uuid]['icon'] item_type = response.get('SPRAY', 'Spray') - return {"success": True, "data": {'type': item_type, 'name': name, 'icon': icon}} + return {'success': True, 'data': {'type': item_type, 'name': name, 'icon': icon}} elif type == 'EquippableSkinLevel': data = JSON.read('cache') name = data['skins'][uuid]['names'][str(VLR_locale)] icon = data['skins'][uuid]['icon'] item_type = response.get('SKIN', 'Skin') - return {"success": True, "data": {'type': item_type, 'name': name, 'icon': icon}} + return {'success': True, 'data': {'type': item_type, 'name': name, 'icon': icon}} elif type == 'EquippableCharmLevel': data = JSON.read('cache') name = data['buddies'][uuid]['names'][str(VLR_locale)] icon = data['buddies'][uuid]['icon'] item_type = response.get('BUDDY', 'Buddie') - return {"success": True, "data": {'type': item_type, 'name': name, 'icon': icon}} + return {'success': True, 'data': {'type': item_type, 'name': name, 'icon': icon}} - return {"success": False, "error": f"Failed to get : {type}"} + return {'success': False, 'error': f'Failed to get : {type}'} - def __get_contract_tier_reward(tier: int, reward: List[Dict]) -> Dict[str, Any]: + @staticmethod + def __get_contract_tier_reward(tier: int, reward: list[dict[str, Any]]) -> dict[str, Any]: """Get tier reward""" data = {} count = 0 for lvl in reward: - for rw in lvl["levels"]: + for rw in lvl['levels']: count += 1 data[count] = rw['reward'] @@ -482,33 +505,38 @@ def __get_contract_tier_reward(tier: int, reward: List[Dict]) -> Dict[str, Any]: return current_reward - def __get_contracts_by_season_id(contracts: Dict, data_contracts: Dict, season_id: str) -> Dict[str, Any]: + @staticmethod + def __get_contracts_by_season_id( + contracts: dict[str, Any], data_contracts: dict[str, Any], season_id: str + ) -> dict[str, Any]: """Get battle pass info""" contracts_uuid = [ - x for x in data_contracts['contracts'] if data_contracts['contracts'][x]['reward']['relationUuid'] == season_id + x + for x in data_contracts['contracts'] + if data_contracts['contracts'][x]['reward']['relationUuid'] == season_id ] if contracts_uuid: - battlepass = [x for x in contracts if x['ContractDefinitionID'] == contracts_uuid[0]] - TIER = battlepass[0]['ProgressionLevelReached'] - XP = battlepass[0]['ProgressionTowardsNextLevel'] + battlepass = [x for x in contracts if x['ContractDefinitionID'] == contracts_uuid[0]] # type: ignore + TIER = battlepass[0]['ProgressionLevelReached'] # type: ignore + XP = battlepass[0]['ProgressionTowardsNextLevel'] # type: ignore REWARD = data_contracts['contracts'][contracts_uuid[0]]['reward']['chapters'] ACT = data_contracts['contracts'][contracts_uuid[0]]['names'][str(VLR_locale)] - return {"success": True, 'tier': TIER, 'xp': XP, 'reward': REWARD, 'act': ACT} + return {'success': True, 'tier': TIER, 'xp': XP, 'reward': REWARD, 'act': ACT} - return {"success": False, "error": "Failed to get battlepass info"} + return {'success': False, 'error': 'Failed to get battlepass info'} @classmethod - def battlepass_format(cls, data: Dict, season: str, response: Dict) -> Dict[str, Any]: + def battlepass_format(cls, data: dict[str, Any], season: str, response: dict[str, Any]) -> dict[str, Any]: """Get battle pass format""" data = data['Contracts'] contracts = JSON.read('cache') # data_contracts['contracts'].pop('version') - season_id = season['id'] - season_end = season['end'] + season_id = season['id'] # type: ignore + season_end = season['end'] # type: ignore btp = cls.__get_contracts_by_season_id(data, contracts, season_id) if btp['success']: @@ -521,17 +549,17 @@ def battlepass_format(cls, data: Dict, season: str, response: Dict) -> Dict[str, item_type = item['data']['type'] item_icon = item['data']['icon'] - return dict( - data=dict( - tier=tier, - act=act, - xp=xp, - reward=item_name, - type=item_type, - icon=item_icon, - end=season_end, - original_type=item_reward['type'], - ) - ) - - raise ValorantBotError(f"Failed to get battlepass info") + return { + 'data': { + 'tier': tier, + 'act': act, + 'xp': xp, + 'reward': item_name, + 'type': item_type, + 'icon': item_icon, + 'end': season_end, + 'original_type': item_reward['type'], + } + } + + raise ValorantBotError('Failed to get battlepass info') diff --git a/utils/valorant/view.py b/utils/valorant/view.py index 74a13f2..bb24fe7 100644 --- a/utils/valorant/view.py +++ b/utils/valorant/view.py @@ -2,7 +2,7 @@ import contextlib from datetime import datetime, timedelta -from typing import TYPE_CHECKING, Awaitable, Dict, List, Union +from typing import TYPE_CHECKING, Any # Standard import discord @@ -24,7 +24,7 @@ class share_button(ui.View): - def __init__(self, interaction: Interaction, embeds: List[discord.Embed]) -> None: + def __init__(self, interaction: Interaction, embeds: list[discord.Embed]) -> None: self.interaction: Interaction = interaction self.embeds = embeds super().__init__(timeout=300) @@ -35,12 +35,12 @@ async def on_timeout(self) -> None: @ui.button(label='Share to friends', style=ButtonStyle.primary) async def button_callback(self, interaction: Interaction, button: ui.Button): - await interaction.channel.send(embeds=self.embeds) + await interaction.channel.send(embeds=self.embeds) # type: ignore await self.interaction.edit_original_response(content='\u200b', embed=None, view=None) class NotifyView(discord.ui.View): - def __init__(self, user_id: int, uuid: str, name: str, response: Dict) -> None: + def __init__(self, user_id: int, uuid: str, name: str, response: dict) -> None: self.user_id = user_id self.uuid = uuid self.name = name @@ -51,23 +51,25 @@ def __init__(self, user_id: int, uuid: str, name: str, response: Dict) -> None: async def interaction_check(self, interaction: Interaction) -> bool: if interaction.user.id == int(self.user_id): return True - await interaction.response.send_message('This pagination menu cannot be controlled by you, sorry!', ephemeral=True) + await interaction.response.send_message( + 'This pagination menu cannot be controlled by you, sorry!', ephemeral=True + ) return False async def on_timeout(self) -> None: """Called when the view times out""" with contextlib.suppress(Exception): - self.remve_notify.disabled = True - await self.message.edit_original_response(view=self) + self.remve_notify.disabled = True # type: ignore + await self.message.edit_original_response(view=self) # type: ignore @discord.ui.button(label='Remove Notify', emoji='✖️', style=ButtonStyle.red) async def remove_notify(self, interaction: Interaction, button: ui.Button): data = JSON.read('notifys') for i in range(len(data)): - if data[i]['uuid'] == self.uuid and data[i]['id'] == str(self.user_id): - data.pop(i) + if data[i]['uuid'] == self.uuid and data[i]['id'] == str(self.user_id): # type: ignore + data.pop(i) # type: ignore break JSON.save('notifys', data) @@ -76,38 +78,37 @@ async def remove_notify(self, interaction: Interaction, button: ui.Button): await interaction.response.edit_message(view=self) removed_notify = self.response.get('REMOVED_NOTIFY') - await interaction.followup.send(removed_notify.format(skin=self.name), ephemeral=True) + await interaction.followup.send(removed_notify.format(skin=self.name), ephemeral=True) # type: ignore class _NotifyListButton(ui.Button): - def __init__(self, label, custom_id) -> None: + def __init__(self, label: str, custom_id: str) -> None: super().__init__(label=label, style=ButtonStyle.red, custom_id=str(custom_id)) async def callback(self, interaction: Interaction) -> None: - await interaction.response.defer() - data: list = JSON.read('notifys') + data: dict[str, Any] = JSON.read('notifys') for i in range(len(data)): - if data[i]['uuid'] == self.custom_id and data[i]['id'] == str(self.view.interaction.user.id): - data.pop(i) + if data[i]['uuid'] == self.custom_id and data[i]['id'] == str(self.view.interaction.user.id): # type: ignore + data.pop(i) # type: ignore break JSON.save('notifys', data) - del self.view.skin_source[self.custom_id] - self.view.update_button() - embed = self.view.main_embed() - await self.view.interaction.edit_original_response(embed=embed, view=self.view) + del self.view.skin_source[self.custom_id] # type: ignore + self.view.update_button() # type: ignore + embed = self.view.main_embed() # type: ignore + await self.view.interaction.edit_original_response(embed=embed, view=self.view) # type: ignore class NotifyViewList(ui.View): - skin_source: Dict + skin_source: dict - def __init__(self, interaction: Interaction, response: Dict) -> None: + def __init__(self, interaction: Interaction[ValorantBot], response: dict[str, Any]) -> None: self.interaction: Interaction = interaction self.response = response - self.bot: ValorantBot = getattr(interaction, "client", interaction._state._get_client()) + self.bot: ValorantBot = interaction.client self.default_language = 'en-US' super().__init__(timeout=600) @@ -119,7 +120,9 @@ async def on_timeout(self) -> None: async def interaction_check(self, interaction: Interaction) -> bool: if interaction.user == self.interaction.user: return True - await interaction.response.send_message('This pagination menu cannot be controlled by you, sorry!', ephemeral=True) + await interaction.response.send_message( + 'This pagination menu cannot be controlled by you, sorry!', ephemeral=True + ) return False def update_button(self) -> None: @@ -129,13 +132,13 @@ def update_button(self) -> None: def create_button(self) -> None: data = self.skin_source for index, skin in enumerate(data, start=1): - self.add_item(_NotifyListButton(label=index, custom_id=skin)) + self.add_item(_NotifyListButton(label=str(index), custom_id=skin)) def get_data(self) -> None: """Gets the data from the cache.""" database = JSON.read('notifys') - notify_skin = [x['uuid'] for x in database if x['id'] == str(self.interaction.user.id)] + notify_skin = [x['uuid'] for x in database if x['id'] == str(self.interaction.user.id)] # type: ignore skin_source = {} for uuid in notify_skin: @@ -169,13 +172,12 @@ def main_embed(self) -> discord.Embed: embed.set_footer(text=click_for_remove) count = 0 text_format = [] - for skin in skin_list: + for count, skin in enumerate(skin_list): name = skin_list[skin]['name'] icon = skin_list[skin]['icon'] price = skin_list[skin]['price'] emoji = skin_list[skin]['emoji'] - count += 1 - text_format.append(f"**{count}.** {emoji} **{name}**\n{vp_emoji} {price}") + text_format.append(f'**{count + 1}.** {emoji} **{name}**\n{vp_emoji} {price}') else: embed.description = '\n'.join(text_format) if len(skin_list) == 1: @@ -183,7 +185,7 @@ def main_embed(self) -> discord.Embed: return embed - async def start(self) -> Awaitable[None]: + async def start(self) -> None: """Starts the view.""" self.get_data() self.create_button() @@ -195,7 +197,13 @@ class TwoFA_UI(ui.Modal, title='Two-factor authentication'): """Modal for riot login with multifactorial authentication""" def __init__( - self, interaction: Interaction, db: DATABASE, cookie: dict, message: str, label: str, response: Dict + self, + interaction: Interaction, + db: DATABASE, + cookie: dict[str, Any], + message: str, + label: str, + response: dict[str, Any], ) -> None: super().__init__(timeout=600) self.interaction: Interaction = interaction @@ -215,52 +223,53 @@ async def on_submit(self, interaction: Interaction) -> None: cookie = self.cookie user_id = self.interaction.user.id auth = self.db.auth - auth.locale_code = self.interaction.locale + auth.locale_code = self.interaction.locale # type: ignore - async def send_embed(content: str) -> Awaitable[None]: + async def send_embed(content: str) -> None: embed = discord.Embed(description=content, color=0xFD4554) if interaction.response.is_done(): return await interaction.followup.send(embed=embed, ephemeral=True) await interaction.response.send_message(embed=embed, ephemeral=True) if not code.isdigit(): - return await send_embed(f"`{code}` is not a number") + return await send_embed(f'`{code}` is not a number') auth = await auth.give2facode(code, cookie) if auth['auth'] == 'response': + login = await self.db.login(user_id, auth, self.interaction.locale) # type: ignore + if login['auth']: # type: ignore + return await send_embed(f"{self.response.get('SUCCESS')} **{login['player']}!**") # type: ignore - login = await self.db.login(user_id, auth, self.interaction.locale) - if login['auth']: - return await send_embed(f"{self.response.get('SUCCESS')} **{login['player']}!**") - - return await send_embed(login['error']) + return await send_embed(login['error']) # type: ignore elif auth['auth'] == 'failed': return await send_embed(auth['error']) async def on_error(self, interaction: Interaction, error: Exception) -> None: """Called when the user submits the modal with an error.""" - print("TwoFA_UI:", error) + print('TwoFA_UI:', error) embed = discord.Embed(description='Oops! Something went wrong.', color=0xFD4554) await interaction.response.send_message(embed=embed, ephemeral=True) # inspired by https://github.com/giorgi-o class BaseBundle(ui.View): - def __init__(self, interaction: Interaction, entries: Dict, response: Dict) -> None: + def __init__( + self, interaction: Interaction[ValorantBot], entries: dict[str, Any], response: dict[str, Any] + ) -> None: self.interaction: Interaction = interaction self.entries = entries self.response = response self.language = str(VLR_locale) - self.bot: ValorantBot = getattr(interaction, "client", interaction._state._get_client()) + self.bot: ValorantBot = interaction.client self.current_page: int = 0 - self.embeds: List[List[discord.Embed]] = [] + self.embeds: list[list[discord.Embed]] = [] self.page_format = {} super().__init__() self.clear_items() - def fill_items(self, force=False) -> None: + def fill_items(self, force: bool = False) -> None: self.clear_items() if len(self.embeds) > 1 or force: self.add_item(self.back_button) @@ -283,26 +292,29 @@ def build_embeds(self, selected_bundle: int = 1) -> None: collection_title = self.response.get('TITLE') - for index, bundle in enumerate(sorted(self.entries, key=lambda c: c['names'][self.language]), start=1): + for index, bundle in enumerate(sorted(self.entries, key=lambda c: c['names'][self.language]), start=1): # type: ignore if index == selected_bundle: embeds.append( discord.Embed( - title=bundle['names'][self.language] + f" {collection_title}", - description=f"{vp_emoji} {bundle['price']}", + title=bundle['names'][self.language] + f' {collection_title}', # type: ignore + description=f"{vp_emoji} {bundle['price']}", # type: ignore color=0xFD4554, - ).set_image(url=bundle['icon']) + ).set_image(url=bundle['icon']) # type: ignore ) - for items in sorted(bundle['items'], key=lambda x: x['price'], reverse=True): - item = GetItems.get_item_by_type(items['type'], items['uuid']) - item_type = get_item_type(items['type']) + for items in sorted(bundle['items'], key=lambda x: x['price'], reverse=True): # type: ignore + item = GetItems.get_item_by_type(items['type'], items['uuid']) # type: ignore + item_type = get_item_type(items['type']) # type: ignore - emoji = GetEmoji.tier_by_bot(items['uuid'], self.bot) if item_type == 'Skins' else '' + emoji = GetEmoji.tier_by_bot(items['uuid'], self.bot) if item_type == 'Skins' else '' # type: ignore icon = item['icon'] if item_type != 'Player Cards' else item['icon']['large'] color = 0xFD4554 if item_type == 'Skins' else 0x0F1923 embed = self.base_embed( - f"{emoji} {item['names'][self.language]}", f"{vp_emoji} {items['price']}", icon, color + f"{emoji} {item['names'][self.language]}", + f"{vp_emoji} {items['price']}", # type: ignore + icon, + color, # type: ignore ) embeds.append(embed) @@ -315,37 +327,38 @@ def build_embeds(self, selected_bundle: int = 1) -> None: self.embeds = embeds_list - def build_featured_bundle(self, bundle: List[Dict]) -> List[discord.Embed]: + def build_featured_bundle(self, bundle: list[dict]) -> list[discord.Embed]: """Builds the featured bundle embeds""" vp_emoji = discord.utils.get(self.bot.emojis, name='ValorantPointIcon') - name = bundle['names'][self.language] + name = bundle['names'][self.language] # type: ignore featured_bundle_title = self.response.get('TITLE') - duration = bundle['duration'] - duration_text = self.response.get('DURATION').format( + duration = bundle['duration'] # type: ignore + duration_text = self.response.get('DURATION').format( # type: ignore duration=format_relative(datetime.utcnow() + timedelta(seconds=duration)) ) - bundle_price = bundle['price'] - bundle_base_price = bundle['base_price'] - bundle_price_text = f"**{bundle_price}** {(f'~~{bundle_base_price}~~' if bundle_base_price != bundle_price else '')}" + bundle_price = bundle['price'] # type: ignore + bundle_base_price = bundle['base_price'] # type: ignore + bundle_price_text = ( + f"**{bundle_price}** {(f'~~{bundle_base_price}~~' if bundle_base_price != bundle_price else '')}" + ) embed = discord.Embed( - title=featured_bundle_title.format(bundle=name), - description=f"{vp_emoji} {bundle_price_text}" f" ({duration_text})", + title=featured_bundle_title.format(bundle=name), # type: ignore + description=f'{vp_emoji} {bundle_price_text}' f' ({duration_text})', color=0xFD4554, ) - embed.set_image(url=bundle['icon']) + embed.set_image(url=bundle['icon']) # type: ignore embed_list = [] embeds = [embed] - for items in sorted(bundle['items'], reverse=True, key=lambda c: c['base_price']): - + for items in sorted(bundle['items'], reverse=True, key=lambda c: c['base_price']): # type: ignore item = GetItems.get_item_by_type(items['type'], items['uuid']) item_type = get_item_type(items['type']) emoji = GetEmoji.tier_by_bot(items['uuid'], self.bot) if item_type == 'Skins' else '' @@ -357,7 +370,7 @@ def build_featured_bundle(self, bundle: List[Dict]) -> List[discord.Embed]: item_price_text = f"**{item_price}** {(f'~~{item_base_price}~~' if item_base_price != item_price else '')}" embed = self.base_embed( - f"{emoji} {item['names'][self.language]}", f"**{vp_emoji}** {item_price_text}", icon, color + f"{emoji} {item['names'][self.language]}", f'**{vp_emoji}** {item_price_text}', icon, color ) embeds.append(embed) @@ -373,8 +386,8 @@ def build_featured_bundle(self, bundle: List[Dict]) -> List[discord.Embed]: def build_select(self) -> None: """Builds the select bundle""" - for index, bundle in enumerate(sorted(self.entries, key=lambda c: c['names']['en-US']), start=1): - self.select_bundle.add_option(label=bundle['names'][self.language], value=index) + for index, bundle in enumerate(sorted(self.entries, key=lambda c: c['names']['en-US']), start=1): # type: ignore + self.select_bundle.add_option(label=bundle['names'][self.language], value=index) # type: ignore @ui.select(placeholder='Select a bundle:') async def select_bundle(self, interaction: Interaction, select: ui.Select): @@ -410,7 +423,7 @@ async def interaction_check(self, interaction: Interaction) -> bool: await interaction.response.send_message('This menus cannot be controlled by you, sorry!', ephemeral=True) return False - async def start(self) -> Awaitable[None]: + async def start(self) -> None: """Starts the bundle view""" if len(self.entries) == 1: @@ -418,32 +431,34 @@ async def start(self) -> Awaitable[None]: self.fill_items() self.update_button() embeds = self.embeds[0] - return await self.interaction.followup.send(embeds=embeds, view=self) + await self.interaction.followup.send(embeds=embeds, view=self) + return elif len(self.entries) != 0: self.add_item(self.select_bundle) placeholder = self.response.get('DROPDOWN_CHOICE_TITLE') self.select_bundle.placeholder = placeholder self.build_select() - return await self.interaction.followup.send('\u200b', view=self) + await self.interaction.followup.send('\u200b', view=self) + return not_found_bundle = self.response.get('NOT_FOUND_BUNDLE') raise ValorantBotError(not_found_bundle) - async def start_furture(self) -> Awaitable[None]: + async def start_furture(self) -> None: """Starts the featured bundle view""" BUNDLES = [] FBundle = self.entries['FeaturedBundle']['Bundles'] for fbd in FBundle: - get_bundle = GetItems.get_bundle(fbd["DataAssetID"]) + get_bundle = GetItems.get_bundle(fbd['DataAssetID']) bundle_payload = { - "uuid": fbd["DataAssetID"], - "icon": get_bundle['icon'], - "names": get_bundle['names'], - "duration": fbd["DurationRemainingInSeconds"], - "items": [], + 'uuid': fbd['DataAssetID'], + 'icon': get_bundle['icon'], + 'names': get_bundle['names'], + 'duration': fbd['DurationRemainingInSeconds'], + 'items': [], } price = 0 @@ -451,16 +466,16 @@ async def start_furture(self) -> Awaitable[None]: for items in fbd['Items']: item_payload = { - "uuid": items["Item"]["ItemID"], - "type": items["Item"]["ItemTypeID"], - "item": GetItems.get_item_by_type(items["Item"]["ItemTypeID"], items["Item"]["ItemID"]), - "amount": items["Item"]["Amount"], - "price": items["DiscountedPrice"], - "base_price": items["BasePrice"], - "discount": items["DiscountPercent"], + 'uuid': items['Item']['ItemID'], + 'type': items['Item']['ItemTypeID'], + 'item': GetItems.get_item_by_type(items['Item']['ItemTypeID'], items['Item']['ItemID']), + 'amount': items['Item']['Amount'], + 'price': items['DiscountedPrice'], + 'base_price': items['BasePrice'], + 'discount': items['DiscountPercent'], } - price += int(items["DiscountedPrice"]) - baseprice += int(items["BasePrice"]) + price += int(items['DiscountedPrice']) + baseprice += int(items['BasePrice']) bundle_payload['items'].append(item_payload) bundle_payload['price'] = price @@ -471,30 +486,30 @@ async def start_furture(self) -> Awaitable[None]: if len(BUNDLES) > 1: return await self.interaction.followup.send('\u200b', view=SelectionFeaturedBundleView(BUNDLES, self)) - self.embeds = self.build_featured_bundle(BUNDLES[0]) + self.embeds = self.build_featured_bundle(BUNDLES[0]) # type: ignore self.fill_items() self.update_button() - await self.interaction.followup.send(embeds=self.embeds[0], view=self) + await self.interaction.followup.send(embeds=self.embeds[0], view=self) # type: ignore class SelectionFeaturedBundleView(ui.View): - def __init__(self, bundles: Dict, other_view: Union[ui.View, BaseBundle] = None): + def __init__(self, bundles: list[dict[str, Any]], other_view: ui.View | BaseBundle | None = None): # type: ignore self.bundles = bundles self.other_view = other_view super().__init__(timeout=120) self.__build_select() - self.select_bundle.placeholder = self.other_view.response.get('DROPDOWN_CHOICE_TITLE') + self.select_bundle.placeholder = self.other_view.response.get('DROPDOWN_CHOICE_TITLE') # type: ignore def __build_select(self) -> None: """Builds the select bundle""" for index, bundle in enumerate(self.bundles): - self.select_bundle.add_option(label=bundle['names'][str(VLR_locale)], value=index) + self.select_bundle.add_option(label=bundle['names'][str(VLR_locale)], value=str(index)) @ui.select(placeholder='Select a bundle:') async def select_bundle(self, interaction: Interaction, select: ui.Select): value = select.values[0] bundle = self.bundles[int(value)] - embeds = self.other_view.build_featured_bundle(bundle) - self.other_view.fill_items() - self.other_view.update_button() - await interaction.response.edit_message(content=None, embeds=embeds[0], view=self.other_view) + embeds = self.other_view.build_featured_bundle(bundle) # type: ignore + self.other_view.fill_items() # type: ignore + self.other_view.update_button() # type: ignore + await interaction.response.edit_message(content=None, embeds=embeds[0], view=self.other_view) # type: ignore