From 31edbd9ad60e3824a4f9e0dd2f71dcfe1b99f7a2 Mon Sep 17 00:00:00 2001 From: TrustyJAID Date: Sun, 5 May 2024 14:23:40 -0600 Subject: [PATCH] [Destiny] 2.0.0 - Separate API class into its own class so it's easier to use for other projects. - Improve craftables command by separating weapons into their respective subtypes. - Migrate API tokens to core's token manager. User OAuth tokens are still stored within the cog. - Make raw URL strings into URL objects. - Replace my old Custom Typing object with d.py's ctx.typing since it supports slash commands better. - Cleanup some old code. - Improve authorization flow for smaller bots. It should now respond immediately if dashboard authorization has been given instead of waiting until the bot sees a new message. - Add Crota's End to raids command. - Prevent running commands in DM that should only be run in a server. - Improve typehinting. --- README.md | 2 +- destiny/api.py | 475 ++++++++++++++--------- destiny/converter.py | 412 +++++++++----------- destiny/destiny.py | 883 +++++++++++++++++++++++++------------------ destiny/menus.py | 12 +- 5 files changed, 982 insertions(+), 802 deletions(-) diff --git a/README.md b/README.md index 8ed3f24351..9bc1a4a823 100644 --- a/README.md +++ b/README.md @@ -16,7 +16,7 @@ TrustyJAID's Cogs for [Red-DiscordBot](https://github.com/Cog-Creators/Red-Disc | Compliment | 1.0.0 |
Compliment people in a creative wayCompliment people in a creative way
| Airen, JennJenn, and TrustyJAID | | Conversions | 1.3.2 |
Conversions for currencies, crypto-currencies, and stocks.Conversions for currencies, crypto-currencies, and stocks.
| TrustyJAID | | CrabRave | 1.1.3 |
Make Crab rave videos, in discord!Create your very own Crab Rave videos with custom text! This cog requires FFMPEG, moviepy (https://github.com/Zulko/moviepy), and imagemagick to work. This cog downloads a template video and font file which is then saved locally and generates crab rave videos from the template. Old videos are deleted after uploading. This cog may consume heavy resources rendering videos.
| DankMemer Team, TrustyJAID, and thisisjvgrace | -| | 1.9.1 |
Thanks for installing
| | +| | 2.0.0 |
Thanks for installing
| | | Elements | 1.1.1 |
Periodic table of elementsGet a plethora of information about elements on the periodic table.
| TrustyJAID | | Encoding | 1.3.1 |
Encode messages into various types of encoding.Encode messages into various types of encoding. Encoding types include: DNA, binary, Caeser cipher, hex, base 64, character, and braille.
| TrustyJAID | | EventPoster | 2.1.3 |
Admin approved announcments/eventsAllow users to setup and host events to be approved by admins.
| TrustyJAID | diff --git a/destiny/api.py b/destiny/api.py index bfab37bdda..525970458b 100644 --- a/destiny/api.py +++ b/destiny/api.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import asyncio import functools import json @@ -5,7 +7,7 @@ from base64 import b64encode from datetime import datetime from pathlib import Path -from typing import Dict, List, Optional, Tuple, Union +from typing import TYPE_CHECKING, List, NamedTuple, Optional, Tuple, Union import aiohttp import discord @@ -13,12 +15,14 @@ from redbot.core import Config, commands from redbot.core.bot import Red from redbot.core.data_manager import cog_data_path -from redbot.core.i18n import Translator, cog_i18n, get_locale +from redbot.core.i18n import Translator, get_locale from redbot.core.utils.menus import start_adding_reactions from redbot.core.utils.predicates import ReactionPredicate +from yarl import URL from .converter import ( STRING_VAR_RE, + BungieMembershipType, BungieTweet, BungieXAccount, DestinyActivityModeGroup, @@ -26,7 +30,6 @@ DestinyComponents, DestinyComponentType, DestinyStatsGroup, - DestinyStatsGroupType, NewsArticles, PeriodType, ) @@ -40,27 +43,16 @@ ServersUnavailable, ) +if TYPE_CHECKING: + from .destiny import Destiny + DEV_BOTS = [552261846951002112] # If you want parsing the manifest data to be easier add your # bots ID to this list otherwise this should help performance # on bots that are just running the cog like normal -DESTINY1_BASE_URL = "https://www.bungie.net/d1/Platform/Destiny/" -BASE_URL = "https://www.bungie.net/Platform" -IMAGE_URL = "https://www.bungie.net" -AUTH_URL = "https://www.bungie.net/en/oauth/authorize" -TOKEN_URL = "https://www.bungie.net/platform/app/oauth/token/" -BUNGIE_MEMBERSHIP_TYPES = { - 0: "None", - 1: "Xbox", - 2: "Playstation", - 3: "Steam", - 4: "Blizzard", - 5: "Stadia", - 6: "Epic Games", - 10: "Demon", - 254: "BungieNext", -} +BASE_URL = URL("https://www.bungie.net") +BASE_HEADERS = {"User-Agent": "Red-TrustyCogs-DestinyCog"} COMPONENTS = DestinyComponents( DestinyComponentType.profiles, @@ -98,45 +90,115 @@ log = getLogger("red.trusty-cogs.Destiny") -class MyTyping(discord.ext.commands.context.DeferTyping): - async def __aenter__(self): - if self.ctx.interaction and not self.ctx.interaction.response.is_done(): - await self.ctx.defer(ephemeral=self.ephemeral) - else: - await self.ctx.typing() +class UserAccount(NamedTuple): + LastSeenDisplayName: str + LastSeendDisplayNameType: int + iconPath: str + crossSaveOverride: BungieMembershipType + applicableMembershipTypes: List[BungieMembershipType] + isPublic: bool + membershipType: BungieMembershipType + membershipId: str + displayName: str + bungieGlobalDisplayName: str + bungieGlobalDisplayNameCode: int + + @property + def id(self) -> int: + return int(self.membershipId) + + @property + def platform(self) -> int: + return int(self.membershipType) + + @classmethod + def from_json(cls, data: dict) -> UserAccount: + known_keys = { + "LastSeenDisplayName": None, + "LastSeendDisplayNameType": None, + "iconPath": None, + "crossSaveOverride": None, + "applicableMembershipTypes": [], + "isPublic": None, + "membershipType": None, + "membershipId": None, + "displayName": None, + "bungieGlobalDisplayName": None, + "bungieGlobalDisplayNameCode": None, + } + for k, v in data.items(): + if k in known_keys: + if k in ["crossSaveOverride", "membershipType"]: + try: + known_keys[k] = BungieMembershipType(v) + except ValueError: + known_keys[k] = BungieMembershipType.Unknown + elif k == "applicableMembershipTypes": + to_add = [] + for i in v: + try: + to_add.append(BungieMembershipType(i)) + except ValueError: + to_add.append(BungieMembershipType.Unknown) + known_keys[k] = to_add + else: + known_keys[k] = v + return cls(**known_keys) -@cog_i18n(_) class DestinyAPI: - config: Config - bot: Red - throttle: float - dashboard_authed: Dict[int, dict] - session: aiohttp.ClientSession - _manifest: dict - _ready: asyncio.Event - - async def load_cache(self): - if await self.config.cache_manifest() > 1: - self._ready.set() - return - loop = asyncio.get_running_loop() - for file in cog_data_path(self).iterdir(): - if not file.is_file(): - continue - task = functools.partial(self.load_file, file=file) - name = file.name.replace(".json", "") - try: - self._manifest[name] = await asyncio.wait_for( - loop.run_in_executor(None, task), timeout=60 - ) - except asyncio.TimeoutError: - log.info("Error loading manifest data") - continue - self._ready.set() + def __init__( + self, + cog: Destiny, + *, + api_key: Optional[str] = None, + client_id: Optional[str] = None, + client_secret: Optional[str] = None, + ): + self.cog: Destiny = cog + self.config: Config = cog.config + self.bot: Red = cog.bot + headers = BASE_HEADERS.copy() + if api_key is not None: + headers.update( + { + "X-API-Key": api_key, + "Content-Type": "application/x-www-form-urlencoded", + "cache-control": "no-cache", + } + ) + self.session = aiohttp.ClientSession( + base_url=URL("https://www.bungie.net"), + headers=headers, + ) + self._manifest: dict = {} + self.throttle: float = 0.0 + self.extra_session = aiohttp.ClientSession(headers=BASE_HEADERS) + # extra session for anything not bungie.net based + self._api_key: Optional[str] = api_key + self._client_id: Optional[str] = client_id + self._client_secret: Optional[str] = client_secret + + @property + def api_key(self): + return self._api_key + + @property + def client_id(self): + return self._client_id + + @api_key.setter + def api_key(self, other: str): + self._api_key = other + self.session.headers.update({"X-API-Key": other}) + # update the session headers if the api_key has been changed + + async def close(self): + await self.session.close() + await self.extra_session.close() async def request_url( - self, url: str, params: Optional[dict] = None, headers: Optional[dict] = None + self, url: URL, params: Optional[dict] = None, headers: Optional[dict] = None ) -> dict: """ Helper to make requests from formed headers and params elsewhere @@ -168,14 +230,14 @@ async def request_url( raise Destiny2APIError async def bungie_tweets(self, account: BungieXAccount) -> List[BungieTweet]: - url = f"https://bungiehelp.org/data/{account.path}" - async with self.session.get(url) as resp: + url = URL(f"https://bungiehelp.org/data/{account.path}") + async with self.extra_session.get(url) as resp: data = await resp.json() return [BungieTweet(**i) for i in data] async def post_url( self, - url: str, + url: URL, params: Optional[dict] = None, headers: Optional[dict] = None, body: Optional[dict] = None, @@ -227,7 +289,7 @@ async def pull_from_postmaster( if item_instance: data["itemId"] = item_instance return await self.post_url( - f"{BASE_URL}/Destiny2/Actions/Items/PullFromPostmaster/", headers=headers, body=data + URL("/Platform/Destiny2/Actions/Items/PullFromPostmaster/"), headers=headers, body=data ) async def transfer_item( @@ -251,22 +313,24 @@ async def transfer_item( if item_instance: data["itemId"] = item_instance return await self.post_url( - f"{BASE_URL}/Destiny2/Actions/Items/TransferItem/", headers=headers, body=data + URL("/Platform/Destiny2/Actions/Items/TransferItem/"), headers=headers, body=data ) async def get_access_token(self, code: str) -> dict: """ Called once the OAuth flow is complete and acquires an access token """ - client_id = await self.config.api_token.client_id() - client_secret = await self.config.api_token.client_secret() - tokens = b64encode(f"{client_id}:{client_secret}".encode("ascii")).decode("utf8") + tokens = b64encode(f"{self.client_id}:{self._client_secret}".encode("ascii")).decode( + "utf8" + ) header = { "Authorization": "Basic {0}".format(tokens), "Content-Type": "application/x-www-form-urlencoded", } data = f"grant_type=authorization_code&code={code}" - async with self.session.post(TOKEN_URL, data=data, headers=header) as resp: + async with self.session.post( + URL("/platform/app/oauth/token/"), data=data, headers=header + ) as resp: if resp.status == 200: data = await resp.json() if "error" in data: @@ -280,16 +344,24 @@ async def get_refresh_token(self, user: discord.abc.User) -> dict: """ Generate a refresh token if the token is expired """ - client_id = await self.config.api_token.client_id() - client_secret = await self.config.api_token.client_secret() - tokens = b64encode(f"{client_id}:{client_secret}".encode("ascii")).decode("utf8") + tokens = b64encode(f"{self.client_id}:{self._client_secret}".encode("ascii")).decode( + "utf8" + ) header = { "Authorization": "Basic {0}".format(tokens), "Content-Type": "application/x-www-form-urlencoded", } - refresh_token = await self.config.user(user).oauth.refresh_token() + oauth = await self.config.user(user).oauth() + try: + refresh_token = oauth["refresh_token"] + except KeyError: + raise Destiny2RefreshTokenError( + "The user has some tokens saved but the refresh token is missing." + ) data = f"grant_type=refresh_token&refresh_token={refresh_token}" - async with self.session.post(TOKEN_URL, data=data, headers=header) as resp: + async with self.session.post( + URL("/platform/app/oauth/token/"), data=data, headers=header + ) as resp: if resp.status == 200: data = await resp.json() if "error" in data: @@ -301,29 +373,18 @@ async def get_refresh_token(self, user: discord.abc.User) -> dict: raise Destiny2RefreshTokenError(_("The refresh token is invalid.")) async def wait_for_oauth_code(self, ctx: commands.Context) -> Optional[str]: - wait_msg = None author = ctx.author code = None - - def check(message): - return (author.id in self.dashboard_authed) or ( - message.author.id == author.id - and re.search(r"\?code=([a-z0-9]+)|(exit|stop)", message.content, flags=re.I) - ) + self.cog.waiting_auth[author.id] = asyncio.Event() try: - wait_msg = await self.bot.wait_for("message", check=check, timeout=180) + await asyncio.wait_for(self.cog.wait_for_auth(author.id), timeout=180) except asyncio.TimeoutError: pass - if author.id in self.dashboard_authed: - code = self.dashboard_authed[author.id]["code"] - elif wait_msg is not None: - code_check = re.compile(r"\?code=([a-z0-9]+)", flags=re.I) - find = code_check.search(wait_msg.content) - if find: - code = find.group(1) - else: - code = wait_msg.content + if author.id in self.cog.dashboard_authed: + code = self.cog.dashboard_authed[author.id]["code"] + elif author.id in self.cog.message_authed: + code = self.cog.message_authed[author.id]["code"] if code not in ["exit", "stop"]: return code @@ -335,58 +396,60 @@ async def get_o_auth(self, ctx: commands.Context) -> Optional[dict]: """ is_slash = ctx.interaction is not None author = ctx.author - client_id = await self.config.api_token.client_id() - if not client_id: + if not self.client_id: raise Destiny2MissingAPITokens( _("The bot owner needs to provide some API tokens first.") ) - url = AUTH_URL + f"?client_id={client_id}&response_type=code" + params = {"client_id": self.client_id, "response_type": "code"} + url = BASE_URL.join(URL("/en/oauth/authorize")).with_query(params) msg = _( - "Go to the following url authorize " "this application and provide the final URL.\n" - ) + "Go to the following url authorize this application and provide the final URL.\n{url}" + ).format(url=url) if is_slash: - await ctx.send(msg + url, ephemeral=True) + await ctx.send(msg, ephemeral=True) else: try: - await author.send(msg + url) + await author.send(msg) except discord.errors.Forbidden: - await ctx.channel.send(msg + url) + await ctx.channel.send(msg) code = await self.wait_for_oauth_code(ctx) - if author.id in self.dashboard_authed: - del self.dashboard_authed[author.id] + if author.id in self.cog.dashboard_authed: + del self.cog.dashboard_authed[author.id] if code is None: return None return await self.get_access_token(code) - async def build_headers(self, user: discord.abc.User = None) -> dict: + async def build_headers(self, user: Optional[discord.abc.User] = None) -> dict: """ Build the headers for each API call from a discord User if present, if a function doesn't require OAuth it won't pass the user object """ - if not await self.config.api_token.api_key(): + if not self.api_key: raise Destiny2MissingAPITokens("The Bot owner needs to set an API Key first.") - header = { - "X-API-Key": await self.config.api_token.api_key(), - "Content-Type": "application/x-www-form-urlencoded", - "cache-control": "no-cache", - } - if not user: - return header + if user is None: + return {} try: await self.check_expired_token(user) except Destiny2RefreshTokenError as e: log.error(e, exc_info=True) raise - access_token = await self.config.user(user).oauth.access_token() - token_type = await self.config.user(user).oauth.token_type() + header = {} + oauth = await self.config.user(user).oauth() + try: + access_token = oauth["access_token"] + token_type = oauth["token_type"] + except KeyError: + raise Destiny2MissingAPITokens( + "The user has some tokens saved but the access token is missing." + ) header["Authorization"] = "{} {}".format(token_type, access_token) return header async def get_user_profile(self, user: discord.abc.User) -> dict: headers = await self.build_headers(user) return await self.request_url( - BASE_URL + "/User/GetMembershipsForCurrentUser/", headers=headers + URL("/Platform/User/GetMembershipsForCurrentUser/"), headers=headers ) async def check_expired_token(self, user: discord.abc.User): @@ -414,7 +477,7 @@ async def check_expired_token(self, user: discord.abc.User): await self.config.user(user).oauth.set(refresh) return if user_oauth["refresh_expires_at"] < now: - await self.config.user(user).clear() + await self.config.user(user).oauth.clear() # We know we have to refresh the oauth after a certain time # So we'll clear the scope so the user can supply it again raise Destiny2RefreshTokenError @@ -440,16 +503,23 @@ async def get_variables(self, user: discord.abc.User) -> dict: components = DestinyComponents(DestinyComponentType.string_variables) params = components.to_dict() - platform = await self.config.user(user).account.membershipType() - user_id = await self.config.user(user).account.membershipId() - url = BASE_URL + f"/Destiny2/{platform}/Profile/{user_id}/" + account = await self.get_user_account(user) + if account is None: + raise Destiny2MissingAPITokens("This user does not have a valid account saved.") + url = URL(f"/Platform/Destiny2/{account.platform}/Profile/{account.id}/") return await self.request_url(url, params=params, headers=headers) + async def get_user_account(self, user: discord.abc.User) -> Optional[UserAccount]: + account = await self.config.user(user).account() + if not account: + return None + return UserAccount.from_json(account) + async def replace_string( self, user: discord.abc.User, text: str, - character: Optional[int] = None, + character: Optional[str] = None, variables: Optional[dict] = None, ) -> str: """ @@ -493,9 +563,10 @@ async def get_characters( components.add(DestinyComponentType.characters) components.add(DestinyComponentType.profiles) params = components.to_dict() - platform = await self.config.user(user).account.membershipType() - user_id = await self.config.user(user).account.membershipId() - url = BASE_URL + f"/Destiny2/{platform}/Profile/{user_id}/" + account = await self.get_user_account(user) + if account is None: + raise Destiny2MissingAPITokens("This user does not have a valid account saved.") + url = URL(f"/Platform/Destiny2/{account.platform}/Profile/{account.id}/") try: chars = await self.request_url(url, params=params, headers=headers) except Exception: @@ -525,9 +596,12 @@ async def get_character( components = COMPONENTS params = components.to_dict() - platform = await self.config.user(user).account.membershipType() - user_id = await self.config.user(user).account.membershipId() - url = BASE_URL + f"/Destiny2/{platform}/Profile/{user_id}/Character/{character_id}" + account = await self.get_user_account(user) + if account is None: + raise Destiny2MissingAPITokens("This user does not have a valid account saved.") + url = URL( + f"/Platform/Destiny2/{account.platform}/Profile/{account.id}/Character/{character_id}" + ) return await self.request_url(url, params=params, headers=headers) async def get_instanced_item( @@ -545,9 +619,12 @@ async def get_instanced_item( components = COMPONENTS params = components.to_dict() - platform = await self.config.user(user).account.membershipType() - user_id = await self.config.user(user).account.membershipId() - url = BASE_URL + f"/Destiny2/{platform}/Profile/{user_id}/Item/{instanced_item}/" + account = await self.get_user_account(user) + if account is None: + raise Destiny2MissingAPITokens("This user does not have a valid account saved.") + url = URL( + f"/Platform/Destiny2/{account.platform}/Profile/{account.id}/Item/{instanced_item}/" + ) return await self.request_url(url, params=params, headers=headers) def _get_entities(self, entity: str, d1: bool = False, *, cache: bool = False) -> dict: @@ -555,9 +632,9 @@ def _get_entities(self, entity: str, d1: bool = False, *, cache: bool = False) - This loads the entity from the saved manifest """ if d1: - path = cog_data_path(self) / f"d1/{entity}.json" + path = cog_data_path(self.cog) / f"d1/{entity}.json" else: - path = cog_data_path(self) / f"{entity}.json" + path = cog_data_path(self.cog) / f"{entity}.json" data = self.load_file(path) if cache: self._manifest[path.name.replace(".json", "")] = data @@ -614,7 +691,7 @@ async def get_definition_from_api( raise Destiny2APIError items = {} for hashes in entity_hash: - url = f"{BASE_URL}/Destiny2/Manifest/{entity}/{hashes}/" + url = URL(f"/Platform/Destiny2/Manifest/{entity}/{hashes}/") data = await self.request_url(url, headers=headers) items[str(hashes)] = data # items.append(data) @@ -670,9 +747,12 @@ async def get_vendor( DestinyComponentType.vendor_sales, ) params = components.to_dict() - platform = await self.config.user(user).account.membershipType() - user_id = await self.config.user(user).account.membershipId() - url = f"{BASE_URL}/Destiny2/{platform}/Profile/{user_id}/Character/{character}/Vendors/{vendor}/" + account = await self.get_user_account(user) + if account is None: + raise Destiny2MissingAPITokens("This user does not have a valid account saved.") + url = URL( + f"/Platform/Destiny2/{account.platform}/Profile/{account.id}/Character/{character}/Vendors/{vendor}/" + ) return await self.request_url(url, params=params, headers=headers) async def get_vendors( @@ -696,9 +776,12 @@ async def get_vendors( DestinyComponentType.vendor_sales, ) params = components.to_dict() - platform = await self.config.user(user).account.membershipType() - user_id = await self.config.user(user).account.membershipId() - url = f"{BASE_URL}/Destiny2/{platform}/Profile/{user_id}/Character/{character}/Vendors/" + account = await self.get_user_account(user) + if account is None: + raise Destiny2MissingAPITokens("This user does not have a valid account saved.") + url = URL( + f"/Platform/Destiny2/{account.platform}/Profile/{account.id}/Character/{character}/Vendors/" + ) return await self.request_url(url, params=params, headers=headers) async def get_clan_members(self, user: discord.abc.User, clan_id: str) -> dict: @@ -709,7 +792,7 @@ async def get_clan_members(self, user: discord.abc.User, clan_id: str) -> dict: headers = await self.build_headers(user) except Exception: raise Destiny2RefreshTokenError - url = f"{BASE_URL}/GroupV2/{clan_id}/Members/" + url = URL(f"/Platform/GroupV2/{clan_id}/Members/") return await self.request_url(url, headers=headers) async def get_bnet_user(self, user: discord.abc.User, membership_id: str) -> dict: @@ -720,7 +803,7 @@ async def get_bnet_user(self, user: discord.abc.User, membership_id: str) -> dic headers = await self.build_headers(user) except Exception: raise Destiny2RefreshTokenError - url = f"{BASE_URL}/User/GetBungieNetUserById/{membership_id}/" + url = URL(f"/Platform/User/GetBungieNetUserById/{membership_id}/") return await self.request_url(url, headers=headers) async def get_bnet_user_credentials(self, user: discord.abc.User, membership_id: str) -> dict: @@ -731,7 +814,7 @@ async def get_bnet_user_credentials(self, user: discord.abc.User, membership_id: headers = await self.build_headers(user) except Exception: raise Destiny2RefreshTokenError - url = f"{BASE_URL}/User/GetCredentialTypesForTargetAccount/{membership_id}/" + url = URL(f"/Platform/User/GetCredentialTypesForTargetAccount/{membership_id}/") return await self.request_url(url, headers=headers) async def get_clan_pending(self, user: discord.abc.User, clan_id: str) -> dict: @@ -742,7 +825,7 @@ async def get_clan_pending(self, user: discord.abc.User, clan_id: str) -> dict: headers = await self.build_headers(user) except Exception: raise Destiny2RefreshTokenError - url = f"{BASE_URL}/GroupV2/{clan_id}/Members/Pending" + url = URL(f"/Platform/GroupV2/{clan_id}/Members/Pending") return await self.request_url(url, headers=headers) async def approve_clan_pending( @@ -760,7 +843,9 @@ async def approve_clan_pending( headers = await self.build_headers(user) except Exception: raise Destiny2RefreshTokenError - url = f"{BASE_URL}/GroupV2/{clan_id}/Members/Approve/{membership_type}/{membership_id}/" + url = URL( + f"/Platform/GroupV2/{clan_id}/Members/Approve/{membership_type}/{membership_id}/" + ) return await self.post_url(url, headers=headers, body=member_data) async def kick_clan_member( @@ -770,11 +855,11 @@ async def kick_clan_member( headers = await self.build_headers(user) except Exception: raise Destiny2RefreshTokenError - url = f"{BASE_URL}/GroupV2/GroupV2/{clan_id}/Members/{membership_type}/{user_id}/Kick/" + url = URL(f"/Platform/GroupV2/GroupV2/{clan_id}/Members/{membership_type}/{user_id}/Kick/") return await self.post_url(url, headers=headers) async def equip_loadout( - self, user: discord.abc.User, loadout_index: int, character_id: int, membership_type: int + self, user: discord.abc.User, loadout_index: int, character_id: str, membership_type: int ) -> dict: """ Equip a Destiny 2 loadout @@ -783,7 +868,7 @@ async def equip_loadout( headers = await self.build_headers(user) except Exception: raise Destiny2RefreshTokenError - url = f"{BASE_URL}/Destiny2/Actions/Loadouts/EquipLoadout/" + url = URL("/Platform/Destiny2/Actions/Loadouts/EquipLoadout/") return await self.post_url( url, headers=headers, @@ -802,7 +887,7 @@ async def get_clan_info(self, user: discord.abc.User, clan_id: str) -> dict: headers = await self.build_headers(user) except Exception: raise Destiny2RefreshTokenError - url = f"{BASE_URL}/GroupV2/{clan_id}/" + url = URL(f"/Platform/GroupV2/{clan_id}/") return await self.request_url(url, headers=headers) async def get_clan_weekly_reward_state(self, user: discord.abc.User, clan_id: str) -> dict: @@ -813,7 +898,7 @@ async def get_clan_weekly_reward_state(self, user: discord.abc.User, clan_id: st headers = await self.build_headers(user) except Exception: raise Destiny2RefreshTokenError - url = f"{BASE_URL}/Destiny2/Clan/{clan_id}/WeeklyRewardState" + url = URL(f"/Platform/Destiny2/Clan/{clan_id}/WeeklyRewardState") return await self.request_url(url, headers=headers) async def get_milestones(self, user: discord.abc.User) -> dict: @@ -824,7 +909,7 @@ async def get_milestones(self, user: discord.abc.User) -> dict: headers = await self.build_headers(user) except Exception: raise Destiny2RefreshTokenError - url = f"{BASE_URL}/Destiny2/Milestones/" + url = URL("/Platform/Destiny2/Milestones/") return await self.request_url(url, headers=headers) async def get_milestone_content(self, user: discord.abc.User, milestone_hash: str) -> dict: @@ -835,7 +920,7 @@ async def get_milestone_content(self, user: discord.abc.User, milestone_hash: st headers = await self.build_headers(user) except Exception: raise Destiny2RefreshTokenError - url = f"{BASE_URL}/Destiny2/Milestones/{milestone_hash}/Content/" + url = URL(f"/Platform/Destiny2/Milestones/{milestone_hash}/Content/") return await self.request_url(url, headers=headers) async def get_post_game_carnage_report(self, user: discord.abc.User, activity_id: int) -> dict: @@ -843,7 +928,7 @@ async def get_post_game_carnage_report(self, user: discord.abc.User, activity_id headers = await self.build_headers(user) except Exception: raise Destiny2RefreshTokenError - url = f"{BASE_URL}/Destiny2/Stats/PostGameCarnageReport/{activity_id}/" + url = URL(f"/Platform/Destiny2/Stats/PostGameCarnageReport/{activity_id}/") return await self.request_url(url, headers=headers) async def get_news(self, page_number: int = 0) -> NewsArticles: @@ -851,7 +936,7 @@ async def get_news(self, page_number: int = 0) -> NewsArticles: headers = await self.build_headers() except Exception: raise Destiny2RefreshTokenError - url = f"{BASE_URL}/Content/Rss/NewsArticles/{page_number}" + url = URL(f"/Platform/Content/Rss/NewsArticles/{page_number}") data = NewsArticles.from_json(await self.request_url(url, headers=headers)) return data @@ -879,9 +964,12 @@ async def get_activity_history( mode_value = mode.value params = {"count": 5, "mode": mode_value, "groups": groups.to_str()} - platform = await self.config.user(user).account.membershipType() - user_id = await self.config.user(user).account.membershipId() - url = f"{BASE_URL}/Destiny2/{platform}/Account/{user_id}/Character/{character}/Stats/Activities/" + account = await self.get_user_account(user) + if account is None: + raise Destiny2MissingAPITokens("This user does not have a valid account saved.") + url = URL( + f"/Platform/Destiny2/{account.platform}/Account/{account.id}/Character/{character}/Stats/Activities/" + ) return await self.request_url(url, params=params, headers=headers) async def get_aggregate_activity_history( @@ -900,10 +988,13 @@ async def get_aggregate_activity_history( raise Destiny2RefreshTokenError if groups is None: groups = DestinyStatsGroup.all() - platform = await self.config.user(user).account.membershipType() - user_id = await self.config.user(user).account.membershipId() + account = await self.get_user_account(user) + if account is None: + raise Destiny2MissingAPITokens("This user does not have a valid account saved.") params = groups.to_dict() - url = f"{BASE_URL}/Destiny2/{platform}/Account/{user_id}/Character/{character}/Stats/AggregateActivityStats/" + url = URL( + f"/Platform/Destiny2/{account.platform}/Account/{account.id}/Character/{character}/Stats/AggregateActivityStats/" + ) return await self.request_url(url, params=params, headers=headers) async def get_weapon_history( @@ -922,10 +1013,13 @@ async def get_weapon_history( raise Destiny2RefreshTokenError if groups is None: groups = DestinyStatsGroup.all() - platform = await self.config.user(user).account.membershipType() - user_id = await self.config.user(user).account.membershipId() + account = await self.get_user_account(user) + if account is None: + raise Destiny2MissingAPITokens("This user does not have a valid account saved.") params = groups.to_dict() - url = f"{BASE_URL}/Destiny2/{platform}/Account/{user_id}/Character/{character}/Stats/UniqueWeapons/" + url = URL( + f"/Platform/Destiny2/{account.platform}/Account/{account.id}/Character/{character}/Stats/UniqueWeapons/" + ) return await self.request_url(url, params=params, headers=headers) async def get_historical_stats( @@ -960,9 +1054,12 @@ async def get_historical_stats( params["dayend"] = dayend if daystart: params["daystart"] = daystart - platform = await self.config.user(user).account.membershipType() - user_id = await self.config.user(user).account.membershipId() - url = f"{BASE_URL}/Destiny2/{platform}/Account/{user_id}/Character/{character}/Stats/" + account = await self.get_user_account(user) + if account is None: + raise Destiny2MissingAPITokens("This user does not have a valid account saved.") + url = URL( + f"/Platform/Destiny2/{account.platform}/Account/{account.id}/Character/{character}/Stats/" + ) return await self.request_url(url, params=params, headers=headers) async def get_historical_stats_account(self, user: discord.abc.User) -> dict: @@ -977,12 +1074,15 @@ async def get_historical_stats_account(self, user: discord.abc.User) -> dict: except Exception: raise Destiny2RefreshTokenError params = {"groups": "1,2,3,101,102,103"} - platform = await self.config.user(user).account.membershipType() - user_id = await self.config.user(user).account.membershipId() - url = f"{BASE_URL}/Destiny2/{platform}/Account/{user_id}/Stats/" + account = await self.get_user_account(user) + if account is None: + raise Destiny2MissingAPITokens("This user does not have a valid account saved.") + url = URL(f"/Platform/Destiny2/{account.platform}/Account/{account.id}/Stats/") return await self.request_url(url, params=params, headers=headers) - async def has_oauth(self, ctx: commands.Context, user: discord.Member = None) -> bool: + async def has_oauth( + self, ctx: commands.Context, user: Optional[discord.Member] = None + ) -> bool: """ Basic checks to see if the user has OAuth setup if not or the OAuth keys are expired this will call the refresh @@ -1038,7 +1138,7 @@ async def has_oauth(self, ctx: commands.Context, user: discord.Member = None) -> return False await self.config.user(author).account.set(datas) else: - platform = BUNGIE_MEMBERSHIP_TYPES[data["destinyMemberships"][0]["membershipType"]] + platform = BungieMembershipType(data["destinyMemberships"][0]["membershipType"]) await self.config.user(author).account.set(data["destinyMemberships"][0]) name = data["bungieNetUser"]["uniqueName"] await ctx.channel.send(_("Account set to {name}").format(name=name)) @@ -1067,8 +1167,12 @@ async def pick_account(self, ctx: commands.Context, profile: dict) -> tuple: ) count = 1 membership = None + membership_name = None for memberships in profile["destinyMemberships"]: - platform = BUNGIE_MEMBERSHIP_TYPES[memberships["membershipType"]] + try: + platform = BungieMembershipType(memberships["membershipType"]) + except ValueError: + platform = BungieMembershipType.Unknown account_name = memberships["displayName"] msg += f"**{count}. {account_name} {platform}**\n" count += 1 @@ -1092,9 +1196,12 @@ async def pick_account(self, ctx: commands.Context, profile: dict) -> tuple: return None, None membership = profile["destinyMemberships"][int(pred.result)] - membership_name = BUNGIE_MEMBERSHIP_TYPES[ - profile["destinyMemberships"][int(pred.result)]["membershipType"] - ] + try: + membership_name = BungieMembershipType( + profile["destinyMemberships"][int(pred.result)]["membershipType"] + ) + except ValueError: + membership_name = BungieMembershipType.Unknown return (membership, membership_name) async def save(self, data: dict, loc: str = "sample.json"): @@ -1108,7 +1215,7 @@ async def save(self, data: dict, loc: str = "sample.json"): def save_manifest(self, data: dict, d1: bool = False): simple_items = {} for key, value in data.items(): - path = cog_data_path(self) / f"{key}.json" + path = cog_data_path(self.cog) / f"{key}.json" if key in self._manifest: self._manifest[key] = value with path.open(encoding="utf-8", mode="w") as f: @@ -1130,7 +1237,7 @@ def save_manifest(self, data: dict, d1: bool = False): "hash": int(item_hash), "loreHash": item_data.get("loreHash", None), } - path = cog_data_path(self) / "simpleitems.json" + path = cog_data_path(self.cog) / "simpleitems.json" with path.open(encoding="utf-8", mode="w") as f: if self.bot.user.id in DEV_BOTS: json.dump( @@ -1143,6 +1250,20 @@ def save_manifest(self, data: dict, d1: bool = False): else: json.dump(simple_items, f) + async def get_manifest_data(self) -> Optional[dict]: + try: + headers = await self.build_headers() + except Destiny2MissingAPITokens: + return + return await self.request_url(URL("/Platform/Destiny2/Manifest/"), headers=headers) + + async def get_d1_manifest_data(self) -> Optional[dict]: + try: + headers = await self.build_headers() + except Destiny2MissingAPITokens: + return + return await self.request_url(URL("/d1/Platform/Destiny/Manifest"), headers=headers) + async def get_manifest(self, d1: bool = False) -> None: """ Checks if the manifest is up to date and downloads if it's not @@ -1152,9 +1273,9 @@ async def get_manifest(self, d1: bool = False) -> None: except Destiny2MissingAPITokens: return if d1: - manifest_data = await self.request_url( - f"{DESTINY1_BASE_URL}/Manifest/", headers=headers - ) + manifest_data = await self.get_d1_manifest_data() + if not manifest_data: + return locale = get_locale() if locale in manifest_data: manifest = manifest_data["mobileWorldContentPaths"][locale] @@ -1163,9 +1284,9 @@ async def get_manifest(self, d1: bool = False) -> None: else: manifest = manifest_data["mobileWorldContentPaths"]["en"] else: - manifest_data = await self.request_url( - f"{BASE_URL}/Destiny2/Manifest/", headers=headers - ) + manifest_data = await self.get_manifest_data() + if manifest_data is None: + return locale = get_locale() if locale in manifest_data: manifest = manifest_data["jsonWorldContentPaths"][locale] @@ -1173,9 +1294,7 @@ async def get_manifest(self, d1: bool = False) -> None: manifest = manifest_data["jsonWorldContentPaths"][locale[:-3]] else: manifest = manifest_data["jsonWorldContentPaths"]["en"] - async with self.session.get( - f"https://bungie.net/{manifest}", headers=headers, timeout=None - ) as resp: + async with self.session.get(URL(manifest), headers=headers, timeout=None) as resp: if d1: data = await resp.read() task = functools.partial(self.download_d1_manifest, data) @@ -1192,7 +1311,7 @@ async def get_manifest(self, d1: bool = False) -> None: return manifest_data["version"] def download_d1_manifest(self, data): - directory = cog_data_path(self) / "d1/" + directory = cog_data_path(self.cog) / "d1/" if not directory.is_dir(): log.debug("Creating guild folder") directory.mkdir(exist_ok=True, parents=True) @@ -1205,7 +1324,7 @@ def download_d1_manifest(self, data): db_name = None with zipfile.ZipFile(str(path), "r") as zip_ref: - zip_ref.extractall(str(cog_data_path(self) / "d1/")) + zip_ref.extractall(str(cog_data_path(self.cog) / "d1/")) db_name = zip_ref.namelist() conn = sqlite3.connect(directory / db_name[0]) @@ -1237,7 +1356,7 @@ def download_d1_manifest(self, data): hash_id = _id json_data[str(hash_id)] = json.loads(datas) # log.debug(dict(row)) - path = cog_data_path(self) / f"d1/{name}.json" + path = cog_data_path(self.cog) / f"d1/{name}.json" with path.open(encoding="utf-8", mode="w") as f: if self.bot.user.id in DEV_BOTS: json.dump( diff --git a/destiny/converter.py b/destiny/converter.py index 3ddd465741..272d223430 100644 --- a/destiny/converter.py +++ b/destiny/converter.py @@ -20,6 +20,31 @@ STRING_VAR_RE = re.compile(r"{var:(?P\d+)}") +LOADOUT_COLOURS = { + "3871954967": discord.Colour(2171690), + "3871954966": discord.Colour(9867402), + "3871954965": discord.Colour(3947839), + "3871954964": discord.Colour(506554), + "3871954963": discord.Colour(4285049), + "3871954962": discord.Colour(6723509), + "3871954961": discord.Colour(3893443), + "3871954960": discord.Colour(1978973), + "3871954975": discord.Colour(10388775), + "3871954974": discord.Colour(4074770), + "1693821586": discord.Colour(2445870), + "1693821587": discord.Colour(3241571), + "1693821584": discord.Colour(7317564), + "1693821585": discord.Colour(12277017), + "1693821590": discord.Colour(11813737), + "1693821591": discord.Colour(3942220), + "1693821588": discord.Colour(9004445), + "1693821589": discord.Colour(3483255), + "1693821594": discord.Colour(8333347), + "1693821595": discord.Colour(5248791), +} +# Generated from the dominant colour of each image using +# this algorithm https://stackoverflow.com/a/61730849 + class BungieTweet: def __init__(self, **kwargs): @@ -144,90 +169,91 @@ class DestinyComponentType(Enum): class DestinyActivityModeType(Enum): - none = 0 - story = 2 - strike = 3 - raid = 4 - allpvp = 5 - patrol = 6 - allpve = 7 - reserved9 = 9 - control = 10 - reserved11 = 11 - clash = 12 - reserved13 = 13 - crimsondoubles = 15 - nightfall = 16 - heroicnightfall = 17 - allstrikes = 18 - ironbanner = 10 - reserved20 = 20 - reserved21 = 21 - reserved22 = 22 - reserved24 = 24 - allmayhem = 25 - reserved26 = 26 - reserved27 = 27 - reserved28 = 28 - reserved29 = 29 - reserved30 = 30 - supremacy = 31 - privatematchesall = 32 - survival = 37 - countdown = 38 - trialsofthenine = 39 - social = 40 - trialscountdown = 31 - trialssurvival = 42 - ironbannercontrol = 43 - ironbannerclash = 44 - ironbannersupremacy = 45 - scorednightfall = 46 - scoredheroicnightfall = 47 - rumble = 48 - alldoubles = 49 - doubles = 50 - privatematchesclash = 51 - privatematchescontrol = 52 - privatematchessupremacy = 53 - privatematchescountdown = 54 - privatematchesruvival = 55 - privatematchesmayhem = 56 - privatematchesrumble = 57 - heroicadventure = 58 - showdown = 59 - lockdown = 60 - scorched = 61 - scorchedteam = 62 - gambit = 63 - allpvecompetitive = 64 - breakthrough = 65 - blackarmoryrun = 66 - salvage = 67 - ironbannersalvage = 68 - pvpcompetitive = 69 - pvpquickplay = 70 - clashquickplay = 71 - clashcompetitive = 72 - controlquickplay = 73 - controlcompetitive = 74 - gambitprime = 75 - reckoning = 76 - menagerie = 77 - vexoffensive = 78 - nightmarehunt = 79 - elimination = 80 - momentum = 81 - dungeon = 82 - sundial = 83 - trialsofosiris = 84 - dares = 85 - offensive = 86 - lostsector = 87 - rift = 88 - zonecontrol = 89 - ironbannerrift = 90 - ironbannerzonecontrol = 91 + Unknown = 0 + Story = 2 + Strike = 3 + Raid = 4 + AllPvP = 5 + Patrol = 6 + AllPvE = 7 + Reserved9 = 9 + Control = 10 + Reserved11 = 11 + Clash = 12 + # Clash -> Destiny's name for Team Deathmatch. 4v4 combat, the team with the highest kills at the end of time wins. + Reserved13 = 13 + CrimsonDoubles = 15 + Nightfall = 16 + HeroicNightfall = 17 + AllStrikes = 18 + IronBanner = 19 + Reserved20 = 20 + Reserved21 = 21 + Reserved22 = 22 + Reserved24 = 24 + AllMayhem = 25 + Reserved26 = 26 + Reserved27 = 27 + Reserved28 = 28 + Reserved29 = 29 + Reserved30 = 30 + Supremacy = 31 + PrivateMatchesAll = 32 + Survival = 37 + Countdown = 38 + TrialsOfTheNine = 39 + Social = 40 + TrialsCountdown = 41 + TrialsSurvival = 42 + IronBannerControl = 43 + IronBannerClash = 44 + IronBannerSupremacy = 45 + ScoredNightfall = 46 + ScoredHeroicNightfall = 47 + Rumble = 48 + AllDoubles = 49 + Doubles = 50 + PrivateMatchesClash = 51 + PrivateMatchesControl = 52 + PrivateMatchesSupremacy = 53 + PrivateMatchesCountdown = 54 + PrivateMatchesSurvival = 55 + PrivateMatchesMayhem = 56 + PrivateMatchesRumble = 57 + HeroicAdventure = 58 + Showdown = 59 + Lockdown = 60 + Scorched = 61 + ScorchedTeam = 62 + Gambit = 63 + AllPvECompetitive = 64 + Breakthrough = 65 + BlackArmoryRun = 66 + Salvage = 67 + IronBannerSalvage = 68 + PvPCompetitive = 69 + PvPQuickplay = 70 + ClashQuickplay = 71 + ClashCompetitive = 72 + ControlQuickplay = 73 + ControlCompetitive = 74 + GambitPrime = 75 + Reckoning = 76 + Menagerie = 77 + VexOffensive = 78 + NightmareHunt = 79 + Elimination = 80 + Momentum = 81 + Dungeon = 82 + Sundial = 83 + TrialsOfOsiris = 84 + Dares = 85 + Offensive = 86 + LostSector = 87 + Rift = 88 + ZoneControl = 89 + IronBannerRift = 90 + IronBannerZoneControl = 91 class DestinyEnumGroup: @@ -307,6 +333,36 @@ def all(cls): return cls(*DestinyStatsGroupType) +class BungieMembershipType(Enum): + All = -1 + Unknown = 0 + TigerXbox = 1 + TigerPsn = 2 + TigerSteam = 3 + TigerBlizzard = 4 + TigerStadia = 5 + TigerEgs = 6 + TigerDemon = 10 + BungieNext = 254 + + def __int__(self) -> int: + return self.value + + def __str__(self) -> str: + return { + BungieMembershipType.All: _("All"), + BungieMembershipType.Unknown: _("Unknown"), + BungieMembershipType.TigerXbox: _("Xbox"), + BungieMembershipType.TigerPsn: _("Playstation"), + BungieMembershipType.TigerSteam: _("Steam"), + BungieMembershipType.TigerBlizzard: _("Blizzard"), + BungieMembershipType.TigerStadia: _("Stadia"), + BungieMembershipType.TigerEgs: _("Epic Games"), + BungieMembershipType.TigerDemon: _("Demon"), + BungieMembershipType.BungieNext: _("BungieNext"), + }[self] + + class PeriodType(Enum): none = 0 daily = 1 @@ -394,164 +450,42 @@ class DestinyItemType(Enum): finisher = 29 -class DestinyActivity(Converter): +class DestinyActivity(discord.app_commands.Transformer): """Returns the correct history code if provided a named one""" - CHOICES: List[dict] = [ - {"name": "all", "value": "0"}, - {"name": "story", "value": "2"}, - {"name": "strike", "value": "3"}, - {"name": "raid", "value": "4"}, - {"name": "allpvp", "value": "5"}, - {"name": "patrol", "value": "6"}, - {"name": "allpve", "value": "7"}, - {"name": "control", "value": "10"}, - {"name": "clash", "value": "12"}, - {"name": "crimsondoubles", "value": "15"}, - {"name": "nightfall", "value": "16"}, - {"name": "heroicnightfall", "value": "17"}, - {"name": "allstrikes", "value": "18"}, - {"name": "ironbanner", "value": "19"}, - {"name": "allmayhem", "value": "25"}, - {"name": "supremacy", "value": "31"}, - {"name": "privatematchesall", "value": "32"}, - {"name": "survival", "value": "37"}, - {"name": "countdown", "value": "38"}, - {"name": "trialsofthenine", "value": "39"}, - {"name": "social", "value": "40"}, - {"name": "trialscountdown", "value": "41"}, - {"name": "trialssurvival", "value": "42"}, - {"name": "ironbannercontrol", "value": "43"}, - {"name": "ironbannerclash", "value": "44"}, - {"name": "ironbannersupremacy", "value": "45"}, - {"name": "scorednightfall", "value": "46"}, - {"name": "scoredheroicnightfall", "value": "47"}, - {"name": "rumble", "value": "48"}, - {"name": "alldoubles", "value": "49"}, - {"name": "doubles", "value": "50"}, - {"name": "privatematchesclash", "value": "51"}, - {"name": "privatematchescontrol", "value": "52"}, - {"name": "privatematchessupremacy", "value": "53"}, - {"name": "privatematchescountdown", "value": "54"}, - {"name": "privatematchessurvival", "value": "55"}, - {"name": "privatematchesmayhem", "value": "56"}, - {"name": "privatematchesrumble", "value": "57"}, - {"name": "heroicadventure", "value": "58"}, - {"name": "showdown", "value": "59"}, - {"name": "lockdown", "value": "60"}, - {"name": "scorched", "value": "61"}, - {"name": "scorchedteam", "value": "62"}, - {"name": "gambit", "value": "63"}, - {"name": "allpvecompetitive", "value": "64"}, - {"name": "breakthrough", "value": "65"}, - {"name": "blackarmoryrun", "value": "66"}, - {"name": "salvage", "value": "67"}, - {"name": "ironbannersalvage", "value": "68"}, - {"name": "pvpcompetitive", "value": "69"}, - {"name": "pvpquickplay", "value": "70"}, - {"name": "clashquickplay", "value": "71"}, - {"name": "clashcompetitive", "value": "72"}, - {"name": "controlquickplay", "value": "73"}, - {"name": "controlcompetitive", "value": "74"}, - {"name": "gambirprime", "value": "75"}, - {"name": "reckoning", "value": "76"}, - {"name": "menagerie", "value": "77"}, - {"name": "vexoffensive", "value": "78"}, - {"name": "nightmarehunt", "value": "79"}, - {"name": "elimination", "value": "80"}, - {"name": "momentum", "value": "81"}, - {"name": "dungeon", "value": "82"}, - {"name": "sundial", "value": "83"}, - {"name": "trialsofosiris", "value": "84"}, - ] - - async def convert(self, ctx: commands.Context, argument: str) -> int: - possible_results: dict = { - "all": {"code": 0, "alt": ["none"]}, - "story": {"code": 2, "alt": []}, - "strike": {"code": 3, "alt": []}, - "raid": {"code": 4, "alt": []}, - "allpvp": {"code": 5, "alt": ["pvp"]}, - "patrol": {"code": 6, "alt": []}, - "allpve": {"code": 7, "alt": ["pve"]}, - "control": {"code": 10, "alt": []}, - "clash": {"code": 12, "alt": []}, - "crimsondoubles": {"code": 15, "alt": []}, - "nightfall": {"code": 16, "alt": []}, - "heroicnightfall": {"code": 17, "alt": []}, - "allstrikes": {"code": 18, "alt": []}, - "ironbanner": {"code": 19, "alt": []}, - "allmayhem": {"code": 25, "alt": []}, - "supremacy": {"code": 31, "alt": []}, - "privatematchesall": {"code": 32, "alt": ["private"]}, - "survival": {"code": 37, "alt": []}, - "countdown": {"code": 38, "alt": []}, - "trialsofthenine": {"code": 39, "alt": ["9"]}, - "social": {"code": 40, "alt": []}, - "trialscountdown": {"code": 41, "alt": []}, - "trialssurvival": {"code": 42, "alt": []}, - "ironbannercontrol": {"code": 43, "alt": []}, - "ironbannerclash": {"code": 44, "alt": []}, - "ironbannersupremacy": {"code": 45, "alt": []}, - "scorednightfall": {"code": 46, "alt": []}, - "scoredheroicnightfall": {"code": 47, "alt": []}, - "rumble": {"code": 48, "alt": []}, - "alldoubles": {"code": 49, "alt": []}, - "doubles": {"code": 50, "alt": []}, - "privatematchesclash": {"code": 51, "alt": ["privateclash"]}, - "privatematchescontrol": {"code": 52, "alt": ["privatecontrol"]}, - "privatematchessupremacy": {"code": 53, "alt": ["privatesupremacy"]}, - "privatematchescountdown": {"code": 54, "alt": ["privatecountdown"]}, - "privatematchessurvival": {"code": 55, "alt": ["privatesurvival"]}, - "privatematchesmayhem": {"code": 56, "alt": ["privatemayhem"]}, - "privatematchesrumble": {"code": 57, "alt": ["privaterumble"]}, - "heroicadventure": {"code": 58, "alt": []}, - "showdown": {"code": 59, "alt": []}, - "lockdown": {"code": 60, "alt": []}, - "scorched": {"code": 61, "alt": []}, - "scorchedteam": {"code": 62, "alt": []}, - "gambit": {"code": 63, "alt": []}, - "allpvecompetitive": {"code": 64, "alt": ["pvecomp"]}, - "breakthrough": {"code": 65, "alt": []}, - "blackarmoryrun": {"code": 66, "alt": ["blackarmory", "armory"]}, - "salvage": {"code": 67, "alt": []}, - "ironbannersalvage": {"code": 68, "alt": []}, - "pvpcompetitive": {"code": 69, "alt": ["pvpcomp", "comp"]}, - "pvpquickplay": {"code": 70, "alt": ["pvpqp", "qp"]}, - "clashquickplay": {"code": 71, "alt": ["clashqp"]}, - "clashcompetitive": {"code": 72, "alt": ["clashcomp"]}, - "controlquickplay": {"code": 73, "alt": ["controlqp"]}, - "controlcompetitive": {"code": 74, "alt": ["controlcomp"]}, - "gambirprime": {"code": 75, "alt": []}, - "reckoning": {"code": 76, "alt": []}, - "menagerie": {"code": 77, "alt": []}, - "vexoffensive": {"code": 78, "alt": []}, - "nightmarehunt": {"code": 79, "alt": []}, - "elimination": {"code": 80, "alt": ["elim"]}, - "momentum": {"code": 81, "alt": []}, - "dungeon": {"code": 82, "alt": []}, - "sundial": {"code": 83, "alt": []}, - "trialsofosiris": {"code": 84, "alt": ["trials"]}, - } - result = None - argument = argument.lower() - if argument.isdigit() and int(argument) in [ - v["code"] for k, v in possible_results.items() - ]: - result = int(argument) - elif argument in possible_results: - result = possible_results[argument]["code"] - else: - for k, v in possible_results.items(): - if argument in v["alt"]: - result = v["code"] - if not result: - raise BadArgument( - _("That is not an available activity, pick from these: {activity_list}").format( - activity_list=humanize_list(list(possible_results.keys())) + async def convert(self, ctx: commands.Context, argument: str) -> DestinyActivityModeType: + if argument.isdigit(): + try: + return DestinyActivityModeType(int(argument)) + except ValueError: + raise BadArgument( + _( + "That is not an available activity, pick from these: {activity_list}" + ).format( + activity_list=humanize_list(list(i.name for i in DestinyActivityModeType)) + ) ) - ) - return result + for activity in DestinyActivityModeType: + if activity.name.lower() == argument.lower(): + return activity + return DestinyActivityModeType.Unknown + + async def transform( + self, interaction: discord.Interaction, argument: str + ) -> DestinyActivityModeType: + ctx = await interaction.client.get_context(interaction) + return await self.convert(ctx, argument) + + async def autocomplete(self, interaction: discord.Interaction, current: str): + possible_options = [ + discord.app_commands.Choice(name=i.name, value=str(i.value)) + for i in DestinyActivityModeType + ] + choices = [] + for choice in possible_options: + if current.lower() in choice.name.lower(): + choices.append(discord.app_commands.Choice(name=choice.name, value=choice.value)) + return choices[:25] class StatsPage(discord.app_commands.Transformer): diff --git a/destiny/destiny.py b/destiny/destiny.py index fa5d2cba5f..0a28029470 100644 --- a/destiny/destiny.py +++ b/destiny/destiny.py @@ -1,32 +1,37 @@ import asyncio import csv import datetime +import functools import random import re -from io import BytesIO, StringIO +from io import StringIO from typing import Dict, List, Literal, Optional -import aiohttp import discord from discord import app_commands from discord.ext import tasks from red_commons.logging import getLogger from redbot.core import Config, commands +from redbot.core.data_manager import cog_data_path from redbot.core.i18n import Translator, cog_i18n from redbot.core.utils.chat_formatting import ( box, humanize_number, humanize_timedelta, pagify, + text_to_file, ) +from redbot.core.utils.views import SetApiView from tabulate import tabulate +from yarl import URL -from .api import DestinyAPI, MyTyping +from .api import DestinyAPI from .converter import ( + LOADOUT_COLOURS, BungieXAccount, DestinyActivity, + DestinyActivityModeType, DestinyCharacter, - DestinyClassType, DestinyComponents, DestinyComponentType, DestinyItemType, @@ -52,37 +57,35 @@ # bots ID to this list otherwise this should help performance # on bots that are just running the cog like normal -BASE_URL = "https://www.bungie.net/Platform" -IMAGE_URL = "https://www.bungie.net" -AUTH_URL = "https://www.bungie.net/en/oauth/authorize" -TOKEN_URL = "https://www.bungie.net/platform/app/oauth/token/" +IMAGE_URL = URL("https://www.bungie.net") + _ = Translator("Destiny", __file__) log = getLogger("red.trusty-cogs.Destiny") @cog_i18n(_) -class Destiny(DestinyAPI, commands.Cog): +class Destiny(commands.Cog): """ Get information from the Destiny 2 API """ - __version__ = "1.9.1" + __version__ = "2.0.0" __author__ = "TrustyJAID" def __init__(self, bot): self.bot = bot - default_global = { - "api_token": {"api_key": "", "client_id": "", "client_secret": ""}, - "manifest_version": "", - "enable_slash": False, - "manifest_channel": None, - "manifest_guild": None, - "manifest_notified_version": None, - "cache_manifest": 0, - "manifest_auto": False, - } self.config = Config.get_conf(self, 35689771456) - self.config.register_global(**default_global) + self.config.register_global( + api_token={"api_key": "", "client_id": "", "client_secret": ""}, + manifest_version="", + enable_slash=False, + manifest_channel=None, + manifest_guild=None, + manifest_notified_version=None, + cache_manifest=0, + manifest_auto=False, + cog_version="0", + ) self.config.register_user(oauth={}, account={}, characters={}) self.config.register_guild( clan_id=None, @@ -92,9 +95,9 @@ def __init__(self, bot): posted_tweets=[], tweets_channel=None, ) - self.throttle: float = 0 self.dashboard_authed: Dict[int, dict] = {} - self.session = aiohttp.ClientSession(headers={"User-Agent": "Red-TrustyCogs-DestinyCog"}) + self.message_authed: Dict[int, dict] = {} + self.waiting_auth: Dict[int, asyncio.Event] = {} self.manifest_check_loop.start() self.news_checker.start() self.tweet_checker.start() @@ -103,18 +106,39 @@ def __init__(self, bot): self._repo = "" self._commit = "" self._ready = asyncio.Event() + self.api: DestinyAPI async def cog_unload(self): try: self.bot.remove_dev_env_value("destiny") except Exception: pass - if not self.session.closed: - await self.session.close() + await self.api.close() self.manifest_check_loop.cancel() self.news_checker.cancel() self.tweet_checker.cancel() + async def load_cache(self): + tokens = await self.bot.get_shared_api_tokens("bungie") + self.api = DestinyAPI(self, **tokens) + if await self.config.cache_manifest() > 1: + self._ready.set() + return + loop = asyncio.get_running_loop() + for file in cog_data_path(self).iterdir(): + if not file.is_file(): + continue + task = functools.partial(self.api.load_file, file=file) + name = file.name.replace(".json", "") + try: + self.api._manifest[name] = await asyncio.wait_for( + loop.run_in_executor(None, task), timeout=60 + ) + except asyncio.TimeoutError: + log.info("Error loading manifest data") + continue + self._ready.set() + async def cog_load(self): if self.bot.user.id in DEV_BOTS: try: @@ -122,9 +146,29 @@ async def cog_load(self): except Exception: pass loop = asyncio.get_running_loop() + await self._migrate_v1_v2() loop.create_task(self.load_cache()) loop.create_task(self._get_commit()) + async def _migrate_v1_v2(self): + if await self.config.cog_version() < "1": + tokens = await self.config.api_token() + await self.bot.set_shared_api_tokens("bungie", **tokens) + await self.config.api_token.clear() + await self.config.cog_version.set("1") + + @commands.Cog.listener() + async def on_red_api_tokens_update(self, service_name: str, tokens: dict): + if service_name != "bungie": + return + for key, value in tokens.items(): + if key == "api_key": + self.api.api_key = value + if key == "client_id": + self.api._client_id = value + if key == "client_secret": + self.api._client_secret = value + def format_help_for_context(self, ctx: commands.Context) -> str: """ Thanks Sinbad! @@ -165,6 +209,11 @@ async def cog_before_invoke(self, ctx: commands.Context): await self._ready.wait() return True + async def wait_for_auth(self, user_id: int): + if user_id not in self.waiting_auth: + raise RuntimeError("Tried to wait for a user's auth but they're not expecting it.") + await self.waiting_auth[user_id].wait() + @commands.Cog.listener() async def on_oauth_receive(self, user_id: int, payload: dict): if payload["provider"] != "destiny": @@ -173,13 +222,23 @@ async def on_oauth_receive(self, user_id: int, payload: dict): log.error("Received Destiny OAuth without a code parameter %s - %s", user_id, payload) return self.dashboard_authed[int(user_id)] = payload + self.waiting_auth[int(user_id)].set() + + @commands.Cog.listener() + async def on_message(self, message: discord.Message): + if message.author.id not in self.waiting_auth: + return + match = re.search(r"\?code=(?P[a-z0-9]+)|(exit|stop)", message.content, flags=re.I) + if match: + self.message_authed[message.author.id] = {"code": match.group(1)} + self.waiting_auth[message.author.id].set() @tasks.loop(seconds=300) async def tweet_checker(self): all_tweets = [] for account in BungieXAccount: try: - all_tweets.extend(await self.bungie_tweets(account)) + all_tweets.extend(await self.api.bungie_tweets(account)) except Exception: log.exception("Error Checking bungiehelp.org") continue @@ -214,11 +273,12 @@ async def tweet_checker(self): @tweet_checker.before_loop async def before_tweet_checker(self): await self.bot.wait_until_red_ready() + await self._ready.wait() @tasks.loop(seconds=300) async def news_checker(self): try: - news = await self.get_news() + news = await self.api.get_news() except Destiny2APIError as e: log.error("Error checking Destiny news sources: %s", e) return @@ -261,6 +321,7 @@ async def news_checker(self): @news_checker.before_loop async def before_news_checker(self): await self.bot.wait_until_red_ready() + await self._ready.wait() @tasks.loop(seconds=3600) async def manifest_check_loop(self): @@ -275,22 +336,18 @@ async def manifest_check_loop(self): # ignore if the manifest has never been downloaded return try: - headers = await self.build_headers() + manifest_data = await self.api.get_manifest_data() + if manifest_data is None: + return except Exception: - return - try: - manifest_data = await self.request_url( - f"{BASE_URL}/Destiny2/Manifest/", headers=headers - ) - except Exception as e: - log.error("Error getting manifest data", exc_info=e) + log.exception("Error getting manifest data") return notify_version = await self.config.manifest_notified_version() if manifest_data["version"] != notify_version: await self.config.manifest_notified_version.set(manifest_data["version"]) if await self.config.manifest_auto(): try: - await self.get_manifest() + await self.api.get_manifest() except Exception: return msg = _("I have downloaded the latest Destiny Manifest: {version}").format( @@ -321,6 +378,9 @@ async def destiny_set_news( - `` The channel you want news articles posted in. """ + if ctx.guild is None: + await ctx.send(_("This command can only be run inside a server.")) + return if channel is not None: if not channel.permissions_for(ctx.me).send_messages: await ctx.send( @@ -330,7 +390,7 @@ async def destiny_set_news( ) return try: - news = await self.get_news() + news = await self.api.get_news() except Destiny2APIError: await ctx.send( _("There was an error getting the current news posts. Please try again later.") @@ -359,6 +419,9 @@ async def destiny_set_tweets( - `` The channel you want tweets posted in. """ + if ctx.guild is None: + await ctx.send(_("This command can only be run inside a server.")) + return if channel is not None: if not channel.permissions_for(ctx.me).send_messages: await ctx.send( @@ -407,7 +470,7 @@ async def forgetme(self, ctx: commands.Context) -> None: """ Remove your authorization to the destiny API on the bot """ - async with MyTyping(ctx, ephemeral=False): + async with ctx.typing(ephemeral=False): await self.red_delete_data_for_user(requester="user", user_id=ctx.author.id) msg = _("Your authorization has been reset.") await ctx.send(msg) @@ -432,7 +495,7 @@ async def items( using `details`, `true`, or `stats` will show the weapons stat bars using `lore` here will instead display the weapons lore card instead if it exists. """ - async with MyTyping(ctx, ephemeral=False): + async with ctx.typing(ephemeral=False): show_lore = True if details_or_lore is False else False if search.startswith("lore "): search = search.replace("lore ", "") @@ -440,15 +503,17 @@ async def items( try: if search.isdigit() or isinstance(search, int): try: - items = await self.get_definition( + items = await self.api.get_definition( "DestinyInventoryItemDefinition", [search] ) except Exception: - items = await self.search_definition( + items = await self.api.search_definition( "DestinyInventoryItemDefinition", search ) else: - items = await self.search_definition("DestinyInventoryItemDefinition", search) + items = await self.api.search_definition( + "DestinyInventoryItemDefinition", search + ) except Destiny2MissingManifest as e: await ctx.send(e) return @@ -457,13 +522,13 @@ async def items( return embeds = [] log.trace("Item: %s", items[0]) - for item_hash, item in items.items(): + for item in items.values(): embed = discord.Embed() damage_type = "" try: damage_data = ( - await self.get_definition( + await self.api.get_definition( "DestinyDamageTypeDefinition", [item["defaultDamageTypeHash"]] ) )[str(item["defaultDamageTypeHash"])] @@ -488,7 +553,7 @@ async def items( continue stat_info = ( - await self.get_definition("DestinyStatDefinition", [stat_hash]) + await self.api.get_definition("DestinyStatDefinition", [stat_hash]) )[str(stat_hash)] stat_name = stat_info["displayProperties"]["name"] if not stat_name: @@ -510,12 +575,12 @@ async def items( stats_str += rpm + recoil + magazine description += stats_str embed.description = description - perks = await self.get_weapon_possible_perks(item) + perks = await self.api.get_weapon_possible_perks(item) for key, value in perks.items(): embed.add_field(name=key, value=value[:1024]) if "loreHash" in item and (show_lore or item["itemType"] in [2]): lore = ( - await self.get_definition("DestinyLoreDefinition", [item["loreHash"]]) + await self.api.get_definition("DestinyLoreDefinition", [item["loreHash"]]) )[str(item["loreHash"])] description += _("Lore: \n\n") + lore["displayProperties"]["description"] if len(description) > 2048: @@ -531,11 +596,11 @@ async def items( name = item["displayProperties"]["name"] embed.title = name - icon_url = IMAGE_URL + item["displayProperties"]["icon"] + icon_url = IMAGE_URL.join(URL(item["displayProperties"]["icon"])) embed.set_author(name=name, icon_url=icon_url) embed.set_thumbnail(url=icon_url) if item.get("screenshot", False): - embed.set_image(url=IMAGE_URL + item["screenshot"]) + embed.set_image(url=IMAGE_URL.join(URL(item["screenshot"]))) embeds.append(embed) if not embeds: await ctx.send(_("That item search could not be found.")) @@ -550,7 +615,7 @@ async def items( @items.autocomplete("search") async def parse_search_items(self, interaction: discord.Interaction, current: str): - possible_options = await self.search_definition("simpleitems", current) + possible_options = await self.api.search_definition("simpleitems", current) choices = [] for hash_key, data in possible_options.items(): name = data["displayProperties"]["name"] @@ -564,11 +629,11 @@ async def destiny_join_command(self, ctx: commands.Context) -> None: """ Get your Steam ID to give people to join your in-game fireteam """ - if not await self.has_oauth(ctx): + if not await self.api.has_oauth(ctx): return - async with MyTyping(ctx, ephemeral=False): + async with ctx.typing(ephemeral=False): bungie_id = await self.config.user(ctx.author).oauth.membership_id() - creds = await self.get_bnet_user(ctx.author, bungie_id) + creds = await self.api.get_bnet_user(ctx.author, bungie_id) bungie_name = creds.get("uniqueName", "") join_code = f"\n```css\n/join {bungie_name}\n```" msg = _( @@ -590,9 +655,9 @@ async def show_clan_info(self, ctx: commands.Context, clan_id: Optional[str] = N """ Display basic information about the clan set in this server """ - if not await self.has_oauth(ctx): + if not await self.api.has_oauth(ctx): return - async with MyTyping(ctx, ephemeral=False): + async with ctx.typing(ephemeral=False): if clan_id: clan_re = re.compile( r"(https:\/\/)?(www\.)?bungie\.net\/.*(groupid=(\d+))", flags=re.I @@ -600,7 +665,7 @@ async def show_clan_info(self, ctx: commands.Context, clan_id: Optional[str] = N clan_invite = clan_re.search(clan_id) if clan_invite: clan_id = clan_invite.group(4) - else: + elif ctx.guild is not None: clan_id = await self.config.guild(ctx.guild).clan_id() if not clan_id: prefix = ctx.clean_prefix @@ -611,22 +676,29 @@ async def show_clan_info(self, ctx: commands.Context, clan_id: Optional[str] = N await ctx.send(msg) return try: - clan_info = await self.get_clan_info(ctx.author, clan_id) - rewards = await self.get_clan_weekly_reward_state(ctx.author, clan_id) + clan_info = await self.api.get_clan_info(ctx.author, clan_id) + rewards = await self.api.get_clan_weekly_reward_state(ctx.author, clan_id) embed = await self.make_clan_embed(clan_info, rewards) except Exception: log.exception("Error getting clan info") msg = _("I could not find any information about this servers clan.") await ctx.send(msg) return - await ctx.send(embed=embed) + view = discord.ui.View() + view.add_item( + discord.ui.Button( + label=_("Request to join the clan"), + url=f"https://www.bungie.net/en/ClanV2?groupid={clan_id}", + ) + ) + await ctx.send(embed=embed, view=view) async def make_clan_embed(self, clan_info: dict, rewards: dict) -> discord.Embed: milestone_hash = rewards.get("milestoneHash", {}) reward_string = "" if milestone_hash: emojis = {True: "✅", False: "❌"} - milestones = await self.get_definition( + milestones = await self.api.get_definition( "DestinyMilestoneDefinition", [str(milestone_hash)] ) milestone_info = milestones[str(milestone_hash)] @@ -686,16 +758,19 @@ async def set_clan_id(self, ctx: commands.Context, clan_id: str) -> None: example link: `https://www.bungie.net/en/ClanV2?groupid=1234567` the numbers after `groupid=` is the clan ID. """ + if ctx.guild is None: + await ctx.send(_("This command can only be run inside a server.")) + return await ctx.defer() - if not await self.has_oauth(ctx): + if not await self.api.has_oauth(ctx): return clan_re = re.compile(r"(https:\/\/)?(www\.)?bungie\.net\/.*(groupid=(\d+))", flags=re.I) clan_invite = clan_re.search(clan_id) if clan_invite: clan_id = clan_invite.group(4) try: - clan_info = await self.get_clan_info(ctx.author, clan_id) - embed = await self.make_clan_embed(clan_info) + clan_info = await self.api.get_clan_info(ctx.author, clan_id) + embed = await self.make_clan_embed(clan_info, {}) except Exception: log.exception("Error getting clan info") msg = _("I could not find a clan with that ID.") @@ -714,8 +789,11 @@ async def clan_pending(self, ctx: commands.Context) -> None: Clan admin can further approve specified clan members by reacting to the resulting message. """ + if ctx.guild is None: + await ctx.send(_("This command can only be run inside a server.")) + return await ctx.defer() - if not await self.has_oauth(ctx): + if not await self.api.has_oauth(ctx): return clan_id = await self.config.guild(ctx.guild).clan_id() if not clan_id: @@ -725,7 +803,7 @@ async def clan_pending(self, ctx: commands.Context) -> None: "Use `{prefix}destiny clan set` to set one." ).format(prefix=prefix) await ctx.send(msg) - clan_pending = await self.get_clan_pending(ctx.author, clan_id) + clan_pending = await self.api.get_clan_pending(ctx.author, clan_id) if not clan_pending["results"]: msg = _("There is no one pending clan approval.") await ctx.send(msg) @@ -737,7 +815,7 @@ async def clan_pending(self, ctx: commands.Context) -> None: @commands.bot_has_permissions(embed_links=True) @commands.mod_or_permissions(manage_messages=True) async def get_clan_roster( - self, ctx: commands.Context, output_format: Optional[Literal["csv", "md"]] = None + self, ctx: commands.Context, output_format: Optional[Literal["csv", "md", "raw"]] = None ) -> None: """ Get the full clan roster @@ -745,7 +823,10 @@ async def get_clan_roster( `[output_format]` if `csv` is provided this will upload a csv file of the clan roster instead of displaying the output. """ - if not await self.has_oauth(ctx): + if ctx.guild is None: + await ctx.send(_("This command can only be run inside a server.")) + return + if not await self.api.has_oauth(ctx): return clan_id = await self.config.guild(ctx.guild).clan_id() if not clan_id: @@ -756,8 +837,8 @@ async def get_clan_roster( ).format(prefix=prefix) await ctx.send(msg) return - async with MyTyping(ctx, ephemeral=False): - clan = await self.get_clan_members(ctx.author, clan_id) + async with ctx.typing(ephemeral=False): + clan = await self.api.get_clan_members(ctx.author, clan_id) headers = [ "Discord Name", "Discord ID", @@ -794,7 +875,7 @@ async def get_clan_roster( try: bungie_id = member["bungieNetUserInfo"]["membershipId"] # bungie_name = member["bungieNetUserInfo"]["displayName"] - creds = await self.get_bnet_user_credentials(ctx.author, bungie_id) + creds = await self.api.get_bnet_user_credentials(ctx.author, bungie_id) steam_id = "" for cred in creds: if "credentialAsString" in cred: @@ -829,11 +910,11 @@ async def get_clan_roster( for row in rows: employee_writer.writerow(row) outfile.seek(0) - file = discord.File(outfile, filename="clan_roster.csv") + file = text_to_file(outfile.getvalue(), filename="clan_roster.csv") await ctx.send(file=file) elif output_format == "md": data = tabulate(rows, headers=headers, tablefmt="github") - file = discord.File(BytesIO(data.encode()), filename="clan_roster.md") + file = text_to_file(data, filename="clan_roster.md") await ctx.send(file=file) else: await ctx.send(_("Displaying member roster for the servers clan.")) @@ -846,7 +927,7 @@ async def destiny_reset_time(self, ctx: commands.Context): """ Show exactly when Weekyl and Daily reset is """ - async with MyTyping(ctx, ephemeral=False): + async with ctx.typing(ephemeral=False): today = datetime.datetime.now(datetime.timezone.utc) tuesday = today + datetime.timedelta(days=((1 - today.weekday()) % 7)) weekly = datetime.datetime( @@ -878,9 +959,9 @@ async def destiny_news(self, ctx: commands.Context) -> None: """ Get the latest news articles from Bungie.net """ - async with MyTyping(ctx, ephemeral=False): + async with ctx.typing(ephemeral=False): try: - news = await self.get_news() + news = await self.api.get_news() except Destiny2APIError as e: return await self.send_error_msg(ctx, e) source = BungieNewsSource(news) @@ -894,23 +975,23 @@ async def destiny_tweets( """ Get the latest news articles from Bungie.net """ - async with MyTyping(ctx, ephemeral=False): + async with ctx.typing(ephemeral=False): try: if account is None: all_tweets = [] for account in BungieXAccount: - all_tweets.extend(await self.bungie_tweets(account)) + all_tweets.extend(await self.api.bungie_tweets(account)) all_tweets.sort(key=lambda x: x.time, reverse=True) else: - all_tweets = await self.bungie_tweets(account) + all_tweets = await self.api.bungie_tweets(account) except Destiny2APIError as e: return await self.send_error_msg(ctx, e) source = BungieTweetsSource(all_tweets) await BaseMenu(source=source, cog=self).start(ctx=ctx) - async def get_seal_icon(self, record: dict) -> str: + async def get_seal_icon(self, record: dict) -> Optional[str]: if record["parentNodeHashes"]: - node_defs = await self.get_definition( + node_defs = await self.api.get_definition( "DestinyPresentationNodeDefinition", record["parentNodeHashes"] ) for key, data in node_defs.items(): @@ -922,7 +1003,7 @@ async def get_seal_icon(self, record: dict) -> str: return dp["iconSequences"][0]["frames"][1] return dp["displayProperties"]["icon"] else: - pres_node = await self.get_entities("DestinyPresentationNodeDefinition") + pres_node = await self.api.get_entities("DestinyPresentationNodeDefinition") node = None for key, data in pres_node.items(): if "completionRecordHash" not in data: @@ -942,15 +1023,15 @@ async def get_seal_icon(self, record: dict) -> str: async def get_character_description(self, char: dict) -> str: info = "" - race = (await self.get_definition("DestinyRaceDefinition", [char["raceHash"]]))[ + race = (await self.api.get_definition("DestinyRaceDefinition", [char["raceHash"]]))[ str(char["raceHash"]) ] - gender = (await self.get_definition("DestinyGenderDefinition", [char["genderHash"]]))[ + gender = (await self.api.get_definition("DestinyGenderDefinition", [char["genderHash"]]))[ str(char["genderHash"]) ] - char_class = (await self.get_definition("DestinyClassDefinition", [char["classHash"]]))[ - str(char["classHash"]) - ] + char_class = ( + await self.api.get_definition("DestinyClassDefinition", [char["classHash"]]) + )[str(char["classHash"])] info += "{race} {gender} {char_class} ".format( race=race["displayProperties"]["name"], gender=gender["displayProperties"]["name"], @@ -959,9 +1040,11 @@ async def get_character_description(self, char: dict) -> str: return info async def get_engram_tracker(self, user: discord.abc.User, char_id: str, chars: dict) -> str: - engram_tracker = await self.get_definition("DestinyInventoryItemDefinition", [1624697519]) + engram_tracker = await self.api.get_definition( + "DestinyInventoryItemDefinition", [1624697519] + ) eg = engram_tracker["1624697519"] - return await self.replace_string( + return await self.api.replace_string( user, eg["displayProperties"]["description"], char_id, chars ) @@ -971,17 +1054,18 @@ async def make_character_embed( char = chars["characters"]["data"][char_id] info = await self.get_character_description(char) titles = "" + title_name = "" embed = discord.Embed(title=info) if "titleRecordHash" in char: # TODO: Add fetch for Destiny.Definitions.Records.DestinyRecordDefinition char_title = ( - await self.get_definition("DestinyRecordDefinition", [char["titleRecordHash"]]) + await self.api.get_definition("DestinyRecordDefinition", [char["titleRecordHash"]]) )[str(char["titleRecordHash"])] icon_url = await self.get_seal_icon(char_title) title_info = "**{title_name}**\n{title_desc}\n" try: gilded = "" - is_gilded, count = await self.check_gilded_title(chars, char_title) + is_gilded, count = await self.api.check_gilded_title(chars, char_title) if is_gilded: gilded = _("Gilded ") title_name = ( @@ -991,7 +1075,8 @@ async def make_character_embed( ) title_desc = char_title["displayProperties"]["description"] titles += title_info.format(title_name=title_name, title_desc=title_desc) - embed.set_thumbnail(url=IMAGE_URL + icon_url) + if icon_url is not None: + embed.set_thumbnail(url=IMAGE_URL.join(URL(icon_url))) except KeyError: pass bnet_display_name = chars["profile"]["data"]["userInfo"]["bungieGlobalDisplayName"] @@ -999,9 +1084,9 @@ async def make_character_embed( bnet_name = f"{bnet_display_name}#{bnet_code}" embed.set_author(name=bnet_name, icon_url=user.display_avatar) # if "emblemPath" in char: - # embed.set_thumbnail(url=IMAGE_URL + char["emblemPath"]) + # embed.set_thumbnail(url=IMAGE_URL.join(URL(char["emblemPath"])) if "emblemBackgroundPath" in char: - embed.set_image(url=IMAGE_URL + char["emblemBackgroundPath"]) + embed.set_image(url=IMAGE_URL.join(URL(char["emblemBackgroundPath"]))) if titles: # embed.add_field(name=_("Titles"), value=titles) embed.set_author(name=f"{bnet_name} ({title_name})", icon_url=user.display_avatar) @@ -1020,7 +1105,7 @@ async def make_character_embed( char["dateLastPlayed"], "%Y-%m-%dT%H:%M:%SZ" ).replace(tzinfo=datetime.timezone.utc) for stat_hash, value in char["stats"].items(): - stat_info = (await self.get_definition("DestinyStatDefinition", [stat_hash]))[ + stat_info = (await self.api.get_definition("DestinyStatDefinition", [stat_hash]))[ str(stat_hash) ] stat_name = stat_info["displayProperties"]["name"] @@ -1047,7 +1132,7 @@ async def make_character_embed( ).format(active=active_score, legacy=legacy_score, lifetime=lifetime_score) embed.add_field(name=_("Triumphs"), value=triumph_str) embed.description = stats_str - embed = await self.get_char_colour(embed, char) + embed = await self.api.get_char_colour(embed, char) if titles: embed.add_field(name=_("Titles"), value=titles) # embed.add_field(name=_("Current Currencies"), value=player_currency, inline=False) @@ -1061,17 +1146,17 @@ async def postmaster(self, ctx: commands.Context): """ View and pull from the postmaster """ - if not await self.has_oauth(ctx): + if not await self.api.has_oauth(ctx): return - async with MyTyping(ctx, ephemeral=False): + async with ctx.typing(ephemeral=False): try: - chars = await self.get_characters( + chars = await self.api.get_characters( ctx.author, components=DestinyComponents( DestinyComponentType.characters, DestinyComponentType.character_inventories ), ) - await self.save(chars) + await self.api.save(chars) except Destiny2APIError as e: log.exception(e) await self.send_error_msg(ctx, e) @@ -1087,7 +1172,7 @@ async def postmaster(self, ctx: commands.Context): postmaster_items = [ i for i in items["items"] if "bucketHash" in i and i["bucketHash"] == 215593132 ] - pm = await self.get_definition( + pm = await self.api.get_definition( "DestinyInventoryItemDefinition", list(set(i["itemHash"] for i in postmaster_items)), ) @@ -1102,13 +1187,13 @@ async def commendations(self, ctx: commands.Context): """ Show your commendation scores """ - if not await self.has_oauth(ctx): + if not await self.api.has_oauth(ctx): return - async with MyTyping(ctx, ephemeral=False): + async with ctx.typing(ephemeral=False): try: components = DestinyComponents(DestinyComponentType.social_commendations) - chars = await self.get_characters(ctx.author, components=components) - await self.save(chars, "character.json") + chars = await self.api.get_characters(ctx.author, components=components) + await self.api.save(chars, "character.json") except Destiny2APIError as e: log.error(e, exc_info=True) await self.send_error_msg(ctx, e) @@ -1121,8 +1206,8 @@ async def make_commendations_embed(self, chars: dict) -> discord.Embed: bnet_code = chars["profile"]["data"]["userInfo"]["bungieGlobalDisplayNameCode"] bnet_name = f"{bnet_display_name}#{bnet_code}" profile_commendations = chars["profileCommendations"]["data"] - commendation_nodes = await self.get_entities("DestinySocialCommendationNodeDefinition") - commendations = await self.get_entities("DestinySocialCommendationDefinition") + commendation_nodes = await self.api.get_entities("DestinySocialCommendationNodeDefinition") + commendations = await self.api.get_entities("DestinySocialCommendationDefinition") bar = await self.get_commendation_scores(chars["profileCommendations"]["data"]) given, received = profile_commendations["scoreDetailValues"] total_score = profile_commendations["totalScore"] @@ -1143,8 +1228,8 @@ async def make_commendations_embed(self, chars: dict) -> discord.Embed: } data = {} for commendation_hash, number in profile_commendations["commendationScoresByHash"].items(): - commendation = commendations.get(commendation_hash) - parent_hash = str(commendation["parentCommendationNodeHash"]) + commendation = commendations.get(commendation_hash, {}) + parent_hash = str(commendation.get("parentCommendationNodeHash")) parent = commendation_nodes.get(parent_hash) if parent is None: continue @@ -1190,25 +1275,23 @@ async def get_commendation_scores(self, data: dict) -> str: @destiny.command() @commands.bot_has_permissions(embed_links=True) - async def user(self, ctx: commands.Context, user: discord.Member = None) -> None: + async def user(self, ctx: commands.Context, user: discord.Member = commands.Author) -> None: """ Display a menu of your basic character's info `[user]` A member on the server who has setup their account on this bot. """ - if not await self.has_oauth(ctx, user): + if not await self.api.has_oauth(ctx, user): return - if not user: - user = ctx.author - async with MyTyping(ctx, ephemeral=False): + async with ctx.typing(ephemeral=False): try: - chars = await self.get_characters(user) - await self.save(chars, "character.json") + chars = await self.api.get_characters(user) + await self.api.save(chars, "character.json") except Destiny2APIError as e: log.error(e, exc_info=True) await self.send_error_msg(ctx, e) return embeds = [] - currency_datas = await self.get_definition( + currency_datas = await self.api.get_definition( "DestinyInventoryItemLiteDefinition", [v["itemHash"] for v in chars["profileCurrencies"]["data"]["items"]], ) @@ -1234,9 +1317,9 @@ async def lore(self, ctx: commands.Context, entry: str = None) -> None: """ Find Destiny Lore """ - async with MyTyping(ctx, ephemeral=False): + async with ctx.typing(ephemeral=False): try: - data = await self.get_entities("DestinyLoreDefinition") + data = await self.api.get_entities("DestinyLoreDefinition") except Exception: msg = _("The manifest needs to be downloaded for this to work.") await ctx.send(msg) @@ -1256,7 +1339,7 @@ async def lore(self, ctx: commands.Context, entry: str = None) -> None: if entries["displayProperties"]["hasIcon"]: icon = entries["displayProperties"]["icon"] - em.set_thumbnail(url=f"{IMAGE_URL}{icon}") + em.set_thumbnail(url=IMAGE_URL.join(URL(icon))) lore.append(em) if entry: for t in lore: @@ -1273,9 +1356,9 @@ async def lore(self, ctx: commands.Context, entry: str = None) -> None: @lore.autocomplete("entry") async def parse_search_lore(self, interaction: discord.Interaction, current: str): - possible_options = self.get_entities("DestinyLoreDefinition") + possible_options: dict = await self.api.get_entities("DestinyLoreDefinition") choices = [] - for hash_key, data in possible_options.items(): + for data in possible_options.values(): name = data["displayProperties"]["name"] if current.lower() in name.lower(): choices.append(app_commands.Choice(name=name, value=name)) @@ -1288,41 +1371,39 @@ async def whereisxur(self, ctx: commands.Context) -> None: """ Display Xûr's current location """ - if not await self.has_oauth(ctx): + if not await self.api.has_oauth(ctx): return - async with MyTyping(ctx, ephemeral=False): + async with ctx.typing(ephemeral=False): try: - chars = await self.get_characters(ctx.author) + chars = await self.api.get_characters(ctx.author) # await self.save(chars, "characters.json") except Destiny2APIError as e: # log.debug(e) await self.send_error_msg(ctx, e) return - for char_id, char in chars["characters"]["data"].items(): - # log.debug(char) - try: - xur = await self.get_vendor(ctx.author, char_id, "2190858386") - xur_def = ( - await self.get_definition("DestinyVendorDefinition", ["2190858386"]) - )["2190858386"] - except Destiny2APIError: - log.error("I can't seem to see Xûr at the moment") - today = datetime.datetime.now(tz=datetime.timezone.utc) - friday = today.replace(hour=17, minute=0, second=0) + datetime.timedelta( - (4 - today.weekday()) % 7 - ) - msg = _("Xûr's not around, come back {next_xur}.").format( - next_xur=discord.utils.format_dt(friday, "R") - ) - await ctx.send(msg) - return - break + char_id = list(chars["characters"]["data"].keys())[0] + try: + xur = await self.api.get_vendor(ctx.author, char_id, "2190858386") + xur_def = ( + await self.api.get_definition("DestinyVendorDefinition", ["2190858386"]) + )["2190858386"] + except Destiny2APIError: + log.error("I can't seem to see Xûr at the moment") + today = datetime.datetime.now(tz=datetime.timezone.utc) + friday = today.replace(hour=17, minute=0, second=0) + datetime.timedelta( + (4 - today.weekday()) % 7 + ) + msg = _("Xûr's not around, come back {next_xur}.").format( + next_xur=discord.utils.format_dt(friday, "R") + ) + await ctx.send(msg) + return try: loc_index = xur["vendor"]["data"]["vendorLocationIndex"] loc = xur_def["locations"][loc_index].get("destinationHash") location_data = ( - await self.get_definition("DestinyDestinationDefinition", [loc]) + await self.api.get_definition("DestinyDestinationDefinition", [loc]) ).get(str(loc), None) location_name = location_data.get("displayProperties", {}).get("name", "") except Exception: @@ -1343,7 +1424,7 @@ async def xur( `[character_class]` Which class you want to see the inventory for. """ - if not await self.has_oauth(ctx): + if not await self.api.has_oauth(ctx): return await self.vendor_menus(ctx, "2190858386", character) @@ -1353,25 +1434,26 @@ async def vendor_menus( vendor_id: str, character: Optional[str] = None, ): - async with MyTyping(ctx, ephemeral=False): - char_id = character + async with ctx.typing(ephemeral=False): if character is None: try: - chars = await self.get_characters(ctx.author) - char_id = list(chars["characters"]["data"].keys())[0] + chars = await self.api.get_characters(ctx.author) + char_id: str = list(chars["characters"]["data"].keys())[0] - await self.save(chars, "characters.json") + await self.api.save(chars, "characters.json") except Destiny2APIError as e: # log.debug(e) await self.send_error_msg(ctx, e) return + else: + char_id: str = character try: - vendor = await self.get_vendor(ctx.author, char_id, vendor_id) - vendor_def = (await self.get_definition("DestinyVendorDefinition", [vendor_id]))[ - vendor_id - ] - await self.save(vendor, "vendor.json") - await self.save(vendor_def, "vendor_def.json") + vendor = await self.api.get_vendor(ctx.author, char_id, vendor_id) + vendor_def = ( + await self.api.get_definition("DestinyVendorDefinition", [vendor_id]) + )[vendor_id] + await self.api.save(vendor, "vendor.json") + await self.api.save(vendor_def, "vendor_def.json") except Destiny2APIError: if vendor_id == "2190858386": log.error("I can't seem to see Xûr at the moment") @@ -1389,7 +1471,7 @@ async def vendor_menus( loc_index = vendor["vendor"]["data"]["vendorLocationIndex"] loc = vendor_def["locations"][loc_index].get("destinationHash") location_data = ( - await self.get_definition("DestinyDestinationDefinition", [loc]) + await self.api.get_definition("DestinyDestinationDefinition", [loc]) ).get(str(loc), None) location = location_data.get("displayProperties", {}).get("name", "") except Exception: @@ -1412,17 +1494,19 @@ async def vendor_menus( ) if "largeTransparentIcon" in vendor_def["displayProperties"]: embed.set_thumbnail( - url=IMAGE_URL + vendor_def["displayProperties"]["largeTransparentIcon"] + url=IMAGE_URL.join( + URL(vendor_def["displayProperties"]["largeTransparentIcon"]) + ) ) # embed.set_author(name=_("Xûr's current wares")) # location = xur_def["locations"][0]["destinationHash"] # log.debug(await self.get_definition("DestinyDestinationDefinition", [location])) all_hashes = [i["itemHash"] for i in vendor["sales"]["data"].values()] - all_items = await self.get_definition("DestinyInventoryItemDefinition", all_hashes) + all_items = await self.api.get_definition("DestinyInventoryItemDefinition", all_hashes) item_costs = [ c["itemHash"] for k, i in vendor["sales"]["data"].items() for c in i["costs"] ] - item_cost_defs = await self.get_definition( + item_cost_defs = await self.api.get_definition( "DestinyInventoryItemLiteDefinition", item_costs ) stat_hashes = [] @@ -1439,8 +1523,10 @@ async def vendor_menus( ].items(): for plug in plugs: perk_hashes.append(plug["plugItemHash"]) - all_perks = await self.get_definition("DestinyInventoryItemDefinition", perk_hashes) - all_stats = await self.get_definition("DestinyStatDefinition", stat_hashes) + all_perks = await self.api.get_definition( + "DestinyInventoryItemDefinition", perk_hashes + ) + all_stats = await self.api.get_definition("DestinyStatDefinition", stat_hashes) main_page = {} for index, item_base in vendor["sales"]["data"].items(): item = all_items[str(item_base["itemHash"])] @@ -1449,9 +1535,11 @@ async def vendor_menus( item_hash = item_base["itemHash"] url = f"https://www.light.gg/db/items/{item_hash}" item_embed = discord.Embed(title=item["displayProperties"]["name"], url=url) - item_embed.set_thumbnail(url=IMAGE_URL + item["displayProperties"]["icon"]) + item_embed.set_thumbnail( + url=IMAGE_URL.join(URL(item["displayProperties"]["icon"])) + ) if "screenshot" in item: - item_embed.set_image(url=IMAGE_URL + item["screenshot"]) + item_embed.set_image(url=IMAGE_URL.join(URL(item["screenshot"]))) for perk_index in vendor["itemComponents"]["reusablePlugs"]["data"].get( index, {"plugs": []} )["plugs"]: @@ -1520,7 +1608,7 @@ async def vendor_menus( cost_str = "" item_description = item["displayProperties"]["description"] + "\n" msg = f"{tier_type_url}{item_description}{stats_str}{perks}{cost_str}{refresh_str}" - msg = await self.replace_string(ctx.author, msg) + msg = await self.api.replace_string(ctx.author, msg) item_embed.description = msg[:4096] if item_type.value not in main_page: main_page[item_type.value] = {} @@ -1565,7 +1653,7 @@ async def eververse( `[character_class]` Which class you want to see the inventory for. """ - if not await self.has_oauth(ctx): + if not await self.api.has_oauth(ctx): return await self.vendor_menus(ctx, "3361454721", character) @@ -1581,7 +1669,7 @@ async def rahool( `[character_class]` Which class you want to see the inventory for. """ - if not await self.has_oauth(ctx): + if not await self.api.has_oauth(ctx): return await self.vendor_menus(ctx, "2255782930", character) @@ -1597,7 +1685,7 @@ async def banshee( `[character_class]` Which class you want to see the inventory for. """ - if not await self.has_oauth(ctx): + if not await self.api.has_oauth(ctx): return await self.vendor_menus(ctx, "672118013", character) @@ -1613,7 +1701,7 @@ async def ada_1_inventory( `[character_class]` Which class you want to see the inventory for. """ - if not await self.has_oauth(ctx): + if not await self.api.has_oauth(ctx): return await self.vendor_menus(ctx, "350061650", character) @@ -1629,7 +1717,7 @@ async def saint_14_inventory( `[character_class]` Which class you want to see the inventory for. """ - if not await self.has_oauth(ctx): + if not await self.api.has_oauth(ctx): return await self.vendor_menus(ctx, "765357505", character) @@ -1646,16 +1734,16 @@ async def vendor_search( `` - The vendor whose inventory you want to see. """ - if not await self.has_oauth(ctx): + if not await self.api.has_oauth(ctx): return if not vendor.isdigit(): - possible_options = await self.search_definition("DestinyVendorDefinition", vendor) + possible_options = await self.api.search_definition("DestinyVendorDefinition", vendor) vendor = list(possible_options.keys())[0] await self.vendor_menus(ctx, vendor, character) @vendor_search.autocomplete("vendor") async def find_vendor(self, interaction: discord.Interaction, current: str): - possible_options = await self.search_definition("DestinyVendorDefinition", current) + possible_options = await self.api.search_definition("DestinyVendorDefinition", current) choices = [] for key, choice in possible_options.items(): name = choice["displayProperties"]["name"] @@ -1673,7 +1761,7 @@ async def destiny_random(self, ctx: commands.Context): pass async def get_random_item(self, weapons_or_class: DestinyRandomConverter, tier_type: int): - data = await self.get_entities("DestinyInventoryItemDefinition") + data = await self.api.get_entities("DestinyInventoryItemDefinition") pool = [] for key, value in data.items(): if value["inventory"]["tierType"] != tier_type: @@ -1698,15 +1786,15 @@ async def random_exotic( Get a random Exotic Weapon or choose a specific Class to get a random armour piece """ - async with MyTyping(ctx, ephemeral=False): + async with ctx.typing(ephemeral=False): item = await self.get_random_item(weapons_or_class, 6) em = discord.Embed(title=item["displayProperties"]["name"], colour=0xF1C40F) if "flavorText" in item: em.description = item["flavorText"] if item["displayProperties"]["hasIcon"]: - em.set_thumbnail(url=IMAGE_URL + item["displayProperties"]["icon"]) + em.set_thumbnail(url=IMAGE_URL.join(URL(item["displayProperties"]["icon"]))) if "screenshot" in item: - em.set_image(url=IMAGE_URL + item["screenshot"]) + em.set_image(url=IMAGE_URL.join(URL(item["screenshot"]))) await ctx.send(embed=em) @destiny.command(name="nightfall", aliases=["nf"]) @@ -1719,21 +1807,21 @@ async def d2_nightfall( Get information about this weeks Nightfall activity """ user = ctx.author - if not await self.has_oauth(ctx): + if not await self.api.has_oauth(ctx): return - async with MyTyping(ctx, ephemeral=False): + async with ctx.typing(ephemeral=False): embeds = [] try: - milestones = await self.get_milestones(user) - chars = await self.get_characters(user) + milestones = await self.api.get_milestones(user) + chars = await self.api.get_characters(user) - await self.save(milestones, "milestones.json") + await self.api.save(milestones, "milestones.json") except Destiny2APIError as e: await self.send_error_msg(ctx, e) return # nightfalls = milestones["1942283261"] - activities = {nf["activityHash"]: nf for nf in milestones["1942283261"]["activities"]} + activities = {nf["activityHash"]: nf for nf in milestones["2029743966"]["activities"]} nf_hashes = {} for char_id, av in chars["characterActivities"]["data"].items(): if character and character != char_id: @@ -1769,15 +1857,15 @@ async def d2_activities( Get information about available activities """ user = ctx.author - if not await self.has_oauth(ctx): + if not await self.api.has_oauth(ctx): return - async with MyTyping(ctx, ephemeral=False): + async with ctx.typing(ephemeral=False): embeds = [] try: - milestones = await self.get_milestones(user) - chars = await self.get_characters(user) + milestones = await self.api.get_milestones(user) + chars = await self.api.get_characters(user) - await self.save(milestones, "milestones.json") + await self.api.save(milestones, "milestones.json") except Destiny2APIError as e: await self.send_error_msg(ctx, e) return @@ -1803,12 +1891,13 @@ async def d2_activities( async def check_character_completion(self, activity_hash: int, chars: dict) -> str: msg = "" - class_info = await self.get_entities("DestinyClassDefinition") + class_info = await self.api.get_entities("DestinyClassDefinition") emojis = { - True: "✅", - False: "❌", + True: "\N{WHITE HEAVY CHECK MARK}", + False: "\N{CROSS MARK}", } reset = "" + ms_group = {} for char_id, data in chars["characterProgressions"]["data"].items(): class_name = ( class_info[str(chars["characters"]["data"][char_id]["classHash"])] @@ -1816,10 +1905,16 @@ async def check_character_completion(self, activity_hash: int, chars: dict) -> s .get("name", "") ) done = "" + ms_data = await self.api.get_entities("DestinyMilestoneDefinition") for ms_hash, ms in data["milestones"].items(): for activity in ms.get("activities", []): if activity["activityHash"] != activity_hash: continue + ms_name = ( + ms_data.get(str(ms_hash), {}).get("displayProperties", {}).get("name") + ) + if ms_hash not in ms_group: + ms_group[ms_hash] = {"name": ms_name, "msg": ""} if "endDate" in ms: reset_dt = datetime.datetime.strptime( ms["endDate"], "%Y-%m-%dT%H:%M:%SZ" @@ -1827,16 +1922,20 @@ async def check_character_completion(self, activity_hash: int, chars: dict) -> s reset = discord.utils.format_dt(reset_dt, "R") if "phases" in activity: done = "-".join(emojis[phase["complete"]] for phase in activity["phases"]) - msg += f"{class_name}\n{done}\n" + ms_group[ms_hash]["msg"] += f"- {class_name}\n{done}\n" elif "challenges" in activity: - is_complete = activity["challenges"][0]["objective"]["complete"] - done = emojis[is_complete] - msg += f"{class_name} - {done}\n" + done = "-".join( + emojis[challenge["objective"]["complete"]] + for challenge in activity["challenges"] + ) + # is_complete = activity["challenges"][0]["objective"]["complete"] + # done = emojis[is_complete] + ms_group[ms_hash]["msg"] += f"- {class_name} - {done}\n" else: - msg += f"{class_name} - ✅" - - # done = emojis[activity["isCompleted"]] + ms_group[ms_hash]["msg"] += f"- {class_name} - \N{WHITE HEAVY CHECK MARK}" + # done = emojis[activity["isCompleted"]] + msg = "\n".join(f"{k['name']}\n{k['msg']}" for k in ms_group.values()) if reset: msg += _("Resets {reset}\n").format(reset=reset) return msg @@ -1844,12 +1943,12 @@ async def check_character_completion(self, activity_hash: int, chars: dict) -> s async def make_activity_embed( self, ctx: commands.Context, activity_hash: int, activity_data: dict, chars: dict ) -> Optional[discord.Embed]: - activity = await self.get_definition("DestinyActivityDefinition", [activity_hash]) + activity = await self.api.get_definition("DestinyActivityDefinition", [activity_hash]) activity = activity[str(activity_hash)] mod_hashes = activity_data.get("modifierHashes", []) mods = None if mod_hashes: - mods = await self.get_definition("DestinyActivityModifierDefinition", mod_hashes) + mods = await self.api.get_definition("DestinyActivityModifierDefinition", mod_hashes) ssdp = activity.get("selectionScreenDisplayProperties", {}).get("description", "") + "\n" mod_string = "" if ssdp: @@ -1865,12 +1964,12 @@ async def make_activity_embed( if activity["displayProperties"]["hasIcon"]: embed.set_author( name=name, - icon_url=IMAGE_URL + activity["displayProperties"]["icon"], + icon_url=IMAGE_URL.join(URL(activity["displayProperties"]["icon"])), ) else: embed.set_author(name=name) if "pgcrImage" in activity: - embed.set_image(url=IMAGE_URL + activity["pgcrImage"]) + embed.set_image(url=IMAGE_URL.join(URL(activity["pgcrImage"]))) if mods: for mod in mods.values(): mod_name = mod["displayProperties"]["name"] @@ -1879,23 +1978,25 @@ async def make_activity_embed( if not mod["displayInActivitySelection"] and mod["displayInNavMode"]: continue mod_desc = re.sub(r"\W?\[[^\[\]]+\]", "", mod["displayProperties"]["description"]) - mod_desc = await self.replace_string(ctx.author, mod_desc) - mod_icon = IMAGE_URL + mod_desc = await self.api.replace_string(ctx.author, mod_desc) + mod_desc = re.sub(r"\n\n", "\n", mod_desc) + mod_desc = re.sub(r"\n", "\n - ", mod_desc) + # mod_icon = IMAGE_URL + mod.get("displayProperties", {}).get("icon", '') - mod_string += f"- [{mod_name}]({mod_icon} '{mod_desc}')\n" + mod_string += f"- {mod_name}\n - {mod_desc}\n" # mod_string += f"> {mod_name}\n {mod_desc}\n\n" if activity["rewards"]: reward_hashes = set() for reward_type in activity["rewards"]: for reward in reward_type["rewardItems"]: reward_hashes.add(reward["itemHash"]) - rewards = await self.get_definition( + rewards = await self.api.get_definition( "DestinyInventoryItemLiteDefinition", list(reward_hashes) ) msg = "" - for reward_hash, data in rewards.items(): + for data in rewards.values(): msg += data["displayProperties"]["name"] + "\n" - msg = await self.replace_string(ctx.author, msg) + msg = await self.api.replace_string(ctx.author, msg) embed.add_field(name=_("Rewards"), value=msg) completion = await self.check_character_completion(activity_hash, chars) if completion: @@ -1908,31 +2009,33 @@ async def dungeons(self, ctx: commands.Context): """ Show your current Dungeon completion state """ - if not await self.has_oauth(ctx): + if not await self.api.has_oauth(ctx): return raid_milestones = { - "422102671", # Shattered Throne - "1742973996", # Pit of Heresy + "1742973996", # Shattered Throne + "422102671", # Pit of Heresy "478604913", # Prophecy "1092691445", # Grasp of Avarice "3618845105", # Duality "526718853", # Spire of the Watcher + "390471874", # Ghosts of the Deep + "3921784328", # Warlord's Ruin } - async with MyTyping(ctx, ephemeral=False): - ms_defs = await self.get_definition( + async with ctx.typing(ephemeral=False): + ms_defs = await self.api.get_definition( "DestinyMilestoneDefinition", list(raid_milestones) ) try: - chars = await self.get_characters(ctx.author) + chars = await self.api.get_characters(ctx.author) except Destiny2APIError as e: await self.send_error_msg(ctx, e) return em = discord.Embed() - em.set_author(name=_("Raid Completion State")) + em.set_author(name=_("Dungeon Completion State")) act_hashes = [ r["activityHash"] for act in ms_defs.values() for r in act.get("activities", []) ] - activities = await self.get_definition("DestinyActivityDefinition", act_hashes) + activities = await self.api.get_definition("DestinyActivityDefinition", act_hashes) embeds = [] for raid in ms_defs.values(): msg = "" @@ -1943,24 +2046,32 @@ async def dungeons(self, ctx: commands.Context): if not completion: continue current_act = act + for char_id, data in chars["characterProgressions"]["data"].items(): + ms = data["milestones"].get(str(raid["hash"])) + if not ms: + continue + for actual_act in ms.get("activities", []): + if act["activityHash"] == actual_act["activityHash"]: + current_act = actual_act + break embeds.append( await self.make_activity_embed(ctx, raid_hash, current_act, chars) ) activity_name = ( - activities.get(str(raid_hash)).get("displayProperties", {}).get("name") + activities.get(str(raid_hash), {}).get("displayProperties", {}).get("name") ) activity_description = ( - activities.get(str(raid_hash)) + activities.get(str(raid_hash), {}) .get("displayProperties", {}) .get("description") ) - if activities.get(str(raid_hash)).get("tier") == -1: + if activities.get(str(raid_hash), {}).get("tier") == -1: activity_description = activity_name.split(":")[-1] msg += f"**{activity_description}**\n{completion}" em.add_field(name=raid_name, value=msg) icon = raid.get("displayProperties", {}).get("icon") if icon: - em.set_thumbnail(url=f"{IMAGE_URL}{icon}") + em.set_thumbnail(url=IMAGE_URL.join(URL(icon))) embeds.insert(0, em) await BaseMenu( source=BasePages( @@ -1976,7 +2087,7 @@ async def raids(self, ctx: commands.Context): """ Show your current raid completion state """ - if not await self.has_oauth(ctx): + if not await self.api.has_oauth(ctx): return raid_milestones = { "3181387331", # Last Wish @@ -1986,13 +2097,14 @@ async def raids(self, ctx: commands.Context): "2136320298", # Vow of the Disciple "292102995", # King's Fall "3699252268", # Root of Nightmares + "540415767", # Crota's End } - async with MyTyping(ctx, ephemeral=False): - ms_defs = await self.get_definition( + async with ctx.typing(ephemeral=False): + ms_defs = await self.api.get_definition( "DestinyMilestoneDefinition", list(raid_milestones) ) try: - chars = await self.get_characters(ctx.author) + chars = await self.api.get_characters(ctx.author) except Destiny2APIError as e: await self.send_error_msg(ctx, e) return @@ -2001,7 +2113,7 @@ async def raids(self, ctx: commands.Context): act_hashes = [ r["activityHash"] for act in ms_defs.values() for r in act.get("activities", []) ] - activities = await self.get_definition("DestinyActivityDefinition", act_hashes) + activities = await self.api.get_definition("DestinyActivityDefinition", act_hashes) embeds = [] for raid in ms_defs.values(): msg = "" @@ -2022,20 +2134,20 @@ async def raids(self, ctx: commands.Context): await self.make_activity_embed(ctx, raid_hash, current_act, chars) ) activity_name = ( - activities.get(str(raid_hash)).get("displayProperties", {}).get("name") + activities.get(str(raid_hash), {}).get("displayProperties", {}).get("name") ) activity_description = ( - activities.get(str(raid_hash)) + activities.get(str(raid_hash), {}) .get("displayProperties", {}) .get("description") ) - if activities.get(str(raid_hash)).get("tier") == -1: + if activities.get(str(raid_hash), {}).get("tier") == -1: activity_description = activity_name.split(":")[-1] msg += f"**{activity_description}**\n{completion}" em.add_field(name=raid_name, value=msg) icon = raid.get("displayProperties", {}).get("icon") if icon: - em.set_thumbnail(url=f"{IMAGE_URL}{icon}") + em.set_thumbnail(url=IMAGE_URL.join(URL(icon))) embeds.insert(0, em) await BaseMenu( source=BasePages( @@ -2052,17 +2164,17 @@ async def d2_dares(self, ctx: commands.Context): Get information about this weeks Nightfall activity """ user = ctx.author - if not await self.has_oauth(ctx): + if not await self.api.has_oauth(ctx): return - async with MyTyping(ctx, ephemeral=False): + async with ctx.typing(ephemeral=False): try: - chars = await self.get_characters(user) + chars = await self.api.get_characters(user) except Destiny2APIError as e: await self.send_error_msg(ctx, e) return acts = None dares_hash = 1030714181 - activity = await self.get_definition("DestinyActivityDefinition", [dares_hash]) + activity = await self.api.get_definition("DestinyActivityDefinition", [dares_hash]) activity = activity[str(dares_hash)] for char_id, av in chars["characterActivities"]["data"].items(): for act in av["availableActivities"]: @@ -2080,14 +2192,14 @@ async def d2_craftables(self, ctx: commands.Context): Show which weapons you're missing deepsight resonance for crafting """ user = ctx.author - if not await self.has_oauth(ctx): + if not await self.api.has_oauth(ctx): return embeds = [] - async with MyTyping(ctx, ephemeral=False): + async with ctx.typing(ephemeral=False): try: - chars = await self.get_characters(user) + chars = await self.api.get_characters(user) - await self.save(chars, "character.json") + await self.api.save(chars, "character.json") except Destiny2APIError as e: log.error(e, exc_info=True) await self.send_error_msg(ctx, e) @@ -2095,7 +2207,7 @@ async def d2_craftables(self, ctx: commands.Context): bnet_display_name = chars["profile"]["data"]["userInfo"]["bungieGlobalDisplayName"] bnet_code = chars["profile"]["data"]["userInfo"]["bungieGlobalDisplayNameCode"] bnet_name = f"{bnet_display_name}#{bnet_code}" - await self.save(chars, "characters.json") + await self.api.save(chars, "characters.json") craftables = chars["profileRecords"]["data"] hashes = [] for record_hash, data in craftables["records"].items(): @@ -2106,40 +2218,63 @@ async def d2_craftables(self, ctx: commands.Context): objective = data["objectives"][0] if not objective["complete"]: hashes.append(record_hash) - weapon_info = await self.get_definition("DestinyRecordDefinition", hashes) - entity = await self.get_entities("DestinyPresentationNodeDefinition") + weapon_info = await self.api.get_definition("DestinyRecordDefinition", hashes) + entity = await self.api.get_entities("DestinyPresentationNodeDefinition") presentation_node_hashes = set() weapon_slots = { 127506319, # Primary Weapon Patterns 3289524180, # Special Weapon Patterns 1464475380, # Heavy Weapon Patterns } + weapon_types = {} for k, v in entity.items(): if not any(i in v.get("parentNodeHashes", []) for i in weapon_slots): continue presentation_node_hashes.add(int(k)) + weapon_types[int(k)] = { + "name": v.get("displayProperties", {}).get("name"), + "value": "", + } log.trace("Presentation Node %s", presentation_node_hashes) - msg = "" for r_hash, i in weapon_info.items(): if not any(h in i.get("parentNodeHashes", []) for h in presentation_node_hashes): continue - state = craftables["records"][str(r_hash)] - objective = state.get("objectives", []) - if not objective: - continue - progress = objective[0]["progress"] - completion = objective[0]["completionValue"] - state_str = f"{progress}/{completion}" - msg += f"{state_str} - {i['displayProperties']['name']}\n" - - for page in pagify(msg): - em = discord.Embed( - title=_("Missing Craftable Weapons"), - description=page, - colour=await self.bot.get_embed_colour(ctx), - ) - em.set_author(name=bnet_name, icon_url=ctx.author.display_avatar) - embeds.append(em) + for h in i.get("parentNodeHashes", []): + if h not in weapon_types: + continue + state = craftables["records"][str(r_hash)] + objective = state.get("objectives", []) + if not objective: + continue + progress = objective[0]["progress"] + completion = objective[0]["completionValue"] + state_str = f"{progress}/{completion}" + weapon_types[h]["value"] += f"{state_str} - {i['displayProperties']['name']}\n" + + em = discord.Embed( + title=_("Missing Craftable Weapons"), + colour=await self.bot.get_embed_colour(ctx), + ) + + for r_hash, field in weapon_types.items(): + if len(em.fields) > 20 or len(em) >= 4000: + embeds.append(em) + em = discord.Embed( + title=_("Missing Craftable Weapons"), + colour=await self.bot.get_embed_colour(ctx), + ) + if field["value"] and len(field["value"]) < 1024: + em.add_field(name=field["name"], value=field["value"]) + elif len(field["value"]) > 1024: + for page in pagify(field["value"], page_length=1024): + if len(em.fields) > 20 or len(em) >= 4000: + embeds.append(em) + em = discord.Embed( + title=_("Missing Craftable Weapons"), + colour=await self.bot.get_embed_colour(ctx), + ) + em.add_field(name=field["name"], value=page) + embeds.append(em) if not embeds: await ctx.send("You have all craftable weapons available! :)") return @@ -2153,9 +2288,8 @@ async def d2_craftables(self, ctx: commands.Context): ).start(ctx=ctx) async def make_loadout_embeds(self, chars: dict) -> Dict[int, discord.Embed]: - colours = await self.get_entities("DestinyLoadoutColorDefinition") - icons = await self.get_entities("DestinyLoadoutIconDefinition") - names = await self.get_entities("DestinyLoadoutNameDefinition") + icons = await self.api.get_entities("DestinyLoadoutIconDefinition") + names = await self.api.get_entities("DestinyLoadoutNameDefinition") bnet_display_name = chars["profile"]["data"]["userInfo"]["bungieGlobalDisplayName"] bnet_code = chars["profile"]["data"]["userInfo"]["bungieGlobalDisplayNameCode"] bnet_name = f"{bnet_display_name}#{bnet_code}" @@ -2164,14 +2298,14 @@ async def make_loadout_embeds(self, chars: dict) -> Dict[int, discord.Embed]: for char_id, loadouts in chars["characterLoadouts"]["data"].items(): ret[char_id] = {"embeds": [], "char_info": ""} char = chars["characters"]["data"][char_id] - race = (await self.get_definition("DestinyRaceDefinition", [char["raceHash"]]))[ + race = (await self.api.get_definition("DestinyRaceDefinition", [char["raceHash"]]))[ str(char["raceHash"]) ] - gender = (await self.get_definition("DestinyGenderDefinition", [char["genderHash"]]))[ - str(char["genderHash"]) - ] + gender = ( + await self.api.get_definition("DestinyGenderDefinition", [char["genderHash"]]) + )[str(char["genderHash"])] char_class = ( - await self.get_definition("DestinyClassDefinition", [char["classHash"]]) + await self.api.get_definition("DestinyClassDefinition", [char["classHash"]]) )[str(char["classHash"])] info = "{race} {gender} {char_class} ".format( race=race["displayProperties"]["name"], @@ -2182,9 +2316,11 @@ async def make_loadout_embeds(self, chars: dict) -> Dict[int, discord.Embed]: for loadout in loadouts["loadouts"]: name = names.get(str(loadout["nameHash"])) icon = icons.get(str(loadout["iconHash"])) - icon_url = IMAGE_URL + icon["iconImagePath"] if icon else None + icon_url = IMAGE_URL.join(URL(icon["iconImagePath"])) if icon else None loadout_name = name["name"] if name else _("Empty Loadout") - embed = discord.Embed(title=loadout_name, description=info) + colour_hash = loadout["colorHash"] + colour = LOADOUT_COLOURS.get(str(colour_hash)) + embed = discord.Embed(title=loadout_name, description=info, colour=colour) embed.set_author( name=_("{name} Loadouts").format(name=bnet_name), icon_url=icon_url ) @@ -2208,7 +2344,7 @@ async def make_loadout_embeds(self, chars: dict) -> Dict[int, discord.Embed]: all_items.add(item_hash) for p in i["perk_hashes"]: all_items.add(p) - inventory = await self.get_definition( + inventory = await self.api.get_definition( "DestinyInventoryItemDefinition", list(all_items) ) for data in items.values(): @@ -2231,10 +2367,8 @@ async def make_loadout_embeds(self, chars: dict) -> Dict[int, discord.Embed]: ret[char_id]["embeds"].append(embed) return ret - @destiny.group() - @commands.bot_has_permissions( - embed_links=True, - ) + @destiny.group(aliases=["loadouts"]) + @commands.bot_has_permissions(embed_links=True) async def loadout(self, ctx: commands.Context) -> None: """ Commands for interacting with your loadouts @@ -2254,10 +2388,10 @@ async def loadout_equip( `` The character you want to select a loadout for `` The loadout you want to equip """ - if not await self.has_oauth(ctx): + if not await self.api.has_oauth(ctx): return loadout -= 1 - async with MyTyping(ctx, ephemeral=False): + async with ctx.typing(ephemeral=False): try: components = DestinyComponents( DestinyComponentType.characters, DestinyComponentType.character_loadouts @@ -2271,8 +2405,8 @@ async def loadout_equip( chars = None if chars is None: try: - chars = await self.get_characters(ctx.author, components) - await self.save(chars, "character.json") + chars = await self.api.get_characters(ctx.author, components) + await self.api.save(chars, "character.json") except Destiny2APIError as e: log.error(e, exc_info=True) await self.send_error_msg(ctx, e) @@ -2284,13 +2418,13 @@ async def loadout_equip( await self.send_error_msg(ctx, e) return try: - await self.equip_loadout(ctx.author, loadout, character_id, membership_type) + await self.api.equip_loadout(ctx.author, loadout, character_id, membership_type) except Destiny2APIError as e: if ctx.author.id in self._loadout_temp: del self._loadout_temp[ctx.author.id] await ctx.send(f"There was an error equipping that loadout: {e}") return - loadout_names = await self.get_entities("DestinyLoadoutNameDefinition") + loadout_names = await self.api.get_entities("DestinyLoadoutNameDefinition") loadout_name_hash = chars["characterLoadouts"]["data"][character_id]["loadouts"][ loadout ]["nameHash"] @@ -2303,7 +2437,7 @@ async def loadout_equip( @loadout_equip.autocomplete("loadout") async def find_loadout(self, interaction: discord.Interaction, current: str): - loadout_names = await self.get_entities("DestinyLoadoutNameDefinition") + loadout_names = await self.api.get_entities("DestinyLoadoutNameDefinition") components = DestinyComponents( DestinyComponentType.characters, DestinyComponentType.character_loadouts ) @@ -2316,7 +2450,7 @@ async def find_loadout(self, interaction: discord.Interaction, current: str): chars = None if chars is None: try: - chars = await self.get_characters(interaction.user, components) + chars = await self.api.get_characters(interaction.user, components) except Destiny2APIError: return [ app_commands.Choice( @@ -2345,11 +2479,11 @@ async def loadout_view(self, ctx: commands.Context) -> None: `[user]` A member on the server who has setup their account on this bot. """ user = ctx.author - if not await self.has_oauth(ctx, user): + if not await self.api.has_oauth(ctx, user): return - async with MyTyping(ctx, ephemeral=False): + async with ctx.typing(ephemeral=False): try: - chars = await self.get_characters(user) + chars = await self.api.get_characters(user) except Destiny2APIError as e: # log.debug(e) @@ -2373,11 +2507,11 @@ async def loadout_equipped(self, ctx: commands.Context, full: Optional[bool] = F `[user]` A member on the server who has setup their account on this bot. """ user = ctx.author - if not await self.has_oauth(ctx, user): + if not await self.api.has_oauth(ctx, user): return - async with MyTyping(ctx, ephemeral=False): + async with ctx.typing(ephemeral=False): try: - chars = await self.get_characters(user) + chars = await self.api.get_characters(user) except Destiny2APIError as e: # log.debug(e) @@ -2389,14 +2523,14 @@ async def loadout_equipped(self, ctx: commands.Context, full: Optional[bool] = F bnet_name = f"{bnet_display_name}#{bnet_code}" for char_id, char in chars["characters"]["data"].items(): info = "" - race = (await self.get_definition("DestinyRaceDefinition", [char["raceHash"]]))[ - str(char["raceHash"]) - ] + race = ( + await self.api.get_definition("DestinyRaceDefinition", [char["raceHash"]]) + )[str(char["raceHash"])] gender = ( - await self.get_definition("DestinyGenderDefinition", [char["genderHash"]]) + await self.api.get_definition("DestinyGenderDefinition", [char["genderHash"]]) )[str(char["genderHash"])] char_class = ( - await self.get_definition("DestinyClassDefinition", [char["classHash"]]) + await self.api.get_definition("DestinyClassDefinition", [char["classHash"]]) )[str(char["classHash"])] info += "{race} {gender} {char_class} ".format( race=race["displayProperties"]["name"], @@ -2404,6 +2538,7 @@ async def loadout_equipped(self, ctx: commands.Context, full: Optional[bool] = F char_class=char_class["displayProperties"]["name"], ) titles = "" + title_name = "" if "titleRecordHash" in char: # TODO: Add fetch for Destiny.Definitions.Records.DestinyRecordDefinition char_title = ( @@ -2414,7 +2549,7 @@ async def loadout_equipped(self, ctx: commands.Context, full: Optional[bool] = F title_info = "**{title_name}**\n{title_desc}\n" try: gilded = "" - is_gilded, count = await self.check_gilded_title(chars, char_title) + is_gilded, count = await self.api.check_gilded_title(chars, char_title) if is_gilded: gilded = _("Gilded ") title_name = ( @@ -2431,7 +2566,7 @@ async def loadout_equipped(self, ctx: commands.Context, full: Optional[bool] = F embed = discord.Embed(title=info) embed.set_author(name=bnet_name, icon_url=user.display_avatar) if "emblemPath" in char: - embed.set_thumbnail(url=IMAGE_URL + char["emblemPath"]) + embed.set_thumbnail(url=IMAGE_URL.join(URL(char["emblemPath"]))) if titles: # embed.add_field(name=_("Titles"), value=titles) embed.set_author( @@ -2440,11 +2575,12 @@ async def loadout_equipped(self, ctx: commands.Context, full: Optional[bool] = F char_items = chars["characterEquipment"]["data"][char_id]["items"] item_list = [i["itemHash"] for i in char_items] # log.debug(item_list) - items = await self.get_definition("DestinyInventoryItemDefinition", item_list) + items = await self.api.get_definition("DestinyInventoryItemDefinition", item_list) # log.debug(items) weapons = "" for item_hash, data in items.items(): # log.debug(data) + instance_id = None for item in char_items: # log.debug(item) if data["hash"] == item["itemHash"]: @@ -2462,7 +2598,7 @@ async def loadout_equipped(self, ctx: commands.Context, full: Optional[bool] = F light = "" perk_list = chars["itemComponents"]["perks"]["data"][instance_id]["perks"] perk_hashes = [p["perkHash"] for p in perk_list] - perk_data = await self.get_definition( + perk_data = await self.api.get_definition( "DestinySandboxPerkDefinition", perk_hashes ) perks = "" @@ -2482,7 +2618,7 @@ async def loadout_equipped(self, ctx: commands.Context, full: Optional[bool] = F "sockets" ] mod_hashes = [p["plugHash"] for p in mod_list if "plugHash" in p] - mod_data = await self.get_definition( + mod_data = await self.api.get_definition( "DestinyInventoryItemDefinition", mod_hashes ) mods = "" @@ -2499,9 +2635,9 @@ async def loadout_equipped(self, ctx: commands.Context, full: Optional[bool] = F # log.debug(data) stats_str = "" for stat_hash, value in char["stats"].items(): - stat_info = (await self.get_definition("DestinyStatDefinition", [stat_hash]))[ - str(stat_hash) - ] + stat_info = ( + await self.api.get_definition("DestinyStatDefinition", [stat_hash]) + )[str(stat_hash)] stat_name = stat_info["displayProperties"]["name"] prog = "█" * int(value / 10) empty = "░" * int((100 - value) / 10) @@ -2513,7 +2649,7 @@ async def loadout_equipped(self, ctx: commands.Context, full: Optional[bool] = F bar = _("Artifact Bonus: {bonus}").format(bonus=artifact_bonus) stats_str += f"{stat_name}: **{value}** \n{bar}\n" embed.description = stats_str - embed = await self.get_char_colour(embed, char) + embed = await self.api.get_char_colour(embed, char) embeds.append(embed) await BaseMenu( @@ -2531,7 +2667,7 @@ async def loadout_equipped(self, ctx: commands.Context, full: Optional[bool] = F async def history( self, ctx: commands.Context, - activity: str, + activity: discord.app_commands.Transform[DestinyActivityModeType, DestinyActivity], character: Optional[discord.app_commands.Transform[str, DestinyCharacter]] = None, ) -> None: """ @@ -2552,18 +2688,12 @@ async def history( reckoning, menagerie, vexoffensive, nightmarehunt, elimination, momentum, dungeon, sundial, trialsofosiris """ - if not await self.has_oauth(ctx): + if not await self.api.has_oauth(ctx): return - async with MyTyping(ctx, ephemeral=False): - if not activity.isdigit(): - try: - activity = await DestinyActivity().convert(ctx, activity) - except commands.BadArgument as e: - await ctx.send(e) - return + async with ctx.typing(ephemeral=False): user = ctx.author try: - chars = await self.get_characters(user) + chars = await self.api.get_characters(user) except Destiny2APIError as e: # log.debug(e) @@ -2590,14 +2720,14 @@ async def history( # log.debug(char) char_info = "" char_class = ( - await self.get_definition("DestinyClassDefinition", [char["classHash"]]) + await self.api.get_definition("DestinyClassDefinition", [char["classHash"]]) )[str(char["classHash"])] char_info += "{user} - {char_class} ".format( user=user.display_name, char_class=char_class["displayProperties"]["name"], ) try: - data = await self.get_activity_history(user, char_id, activity) + data = await self.api.get_activity_history(user, char_id, activity) except Exception: log.error( "Something went wrong I couldn't get info on character %s for activity %s", @@ -2611,7 +2741,7 @@ async def history( for activities in data["activities"]: activity_hash = str(activities["activityDetails"]["directorActivityHash"]) activity_data = ( - await self.get_definition("DestinyActivityDefinition", [activity_hash]) + await self.api.get_definition("DestinyActivityDefinition", [activity_hash]) )[str(activity_hash)] embed = discord.Embed( title=activity_data["displayProperties"]["name"] + f"- {char_info}", @@ -2624,10 +2754,10 @@ async def history( embed.timestamp = date if activity_data["displayProperties"]["hasIcon"]: embed.set_thumbnail( - url=IMAGE_URL + activity_data["displayProperties"]["icon"] + url=IMAGE_URL.join(URL(activity_data["displayProperties"]["icon"])) ) if activity_data.get("pgcrImage", None) is not None: - embed.set_image(url=IMAGE_URL + activity_data["pgcrImage"]) + embed.set_image(url=IMAGE_URL.join(URL(activity_data["pgcrImage"]))) embed.set_author(name=char_info, icon_url=user.display_avatar) for attr, name in RAID.items(): if activities["values"][attr]["basic"]["value"] < 0: @@ -2636,7 +2766,7 @@ async def history( name=name, value=str(activities["values"][attr]["basic"]["displayValue"]), ) - embed = await self.get_char_colour(embed, char) + embed = await self.api.get_char_colour(embed, char) embeds.append(embed) await BaseMenu( @@ -2647,17 +2777,6 @@ async def history( page_start=0, ).start(ctx=ctx) - @history.autocomplete("activity") - async def parse_history(self, interaction: discord.Interaction, current: str): - possible_options = [ - app_commands.Choice(name=i["name"], value=i["value"]) for i in DestinyActivity.CHOICES - ] - choices = [] - for choice in possible_options: - if current.lower() in choice.name.lower(): - choices.append(app_commands.Choice(name=choice.name, value=choice.value)) - return choices[:25] - @staticmethod async def get_extra_attrs(stat_type: str, attrs: dict) -> dict: """Helper function to receive the total attributes we care about""" @@ -2694,11 +2813,11 @@ async def build_character_stats( aggregate = {} acts = {} try: - data = await self.get_historical_stats(user, char_id, 0) + data = await self.api.get_historical_stats(user, char_id, 0) if stat_type == "raid": - aggregate = await self.get_aggregate_activity_history(user, char_id) + aggregate = await self.api.get_aggregate_activity_history(user, char_id) agg_hashes = [a["activityHash"] for a in aggregate["activities"]] - acts = await self.get_definition("DestinyActivityDefinition", agg_hashes) + acts = await self.api.get_definition("DestinyActivityDefinition", agg_hashes) except Exception: log.error("Something went wrong I couldn't get info on character %s", char_id) continue @@ -2734,15 +2853,15 @@ async def build_stat_embed_char_basic( acts: dict, ) -> discord.Embed: char_info = "" - race = (await self.get_definition("DestinyRaceDefinition", [char["raceHash"]]))[ + race = (await self.api.get_definition("DestinyRaceDefinition", [char["raceHash"]]))[ str(char["raceHash"]) ] - gender = (await self.get_definition("DestinyGenderDefinition", [char["genderHash"]]))[ + gender = (await self.api.get_definition("DestinyGenderDefinition", [char["genderHash"]]))[ str(char["genderHash"]) ] - char_class = (await self.get_definition("DestinyClassDefinition", [char["classHash"]]))[ - str(char["classHash"]) - ] + char_class = ( + await self.api.get_definition("DestinyClassDefinition", [char["classHash"]]) + )[str(char["classHash"])] char_info += "{race} {gender} {char_class} ".format( race=race["displayProperties"]["name"], gender=gender["displayProperties"]["name"], @@ -2781,7 +2900,7 @@ async def build_stat_embed_char_basic( kda = f"{kills} | {deaths} | {assists}" embed.add_field(name=_("Kills | Deaths | Assists"), value=kda) if "emblemPath" in char: - embed.set_thumbnail(url=IMAGE_URL + char["emblemPath"]) + embed.set_thumbnail(url=IMAGE_URL.join(URL(char["emblemPath"]))) for stat, values in data[stat_type]["allTime"].items(): if values["basic"]["value"] < 0 or stat not in ATTRS: continue @@ -2799,21 +2918,21 @@ async def build_stat_embed_char_basic( resur = data[stat_type]["resurrectionsReceived"] if res or resur: embed.add_field(name=_("Resurrections/Received"), value=f"{res}/{resur}") - return await self.get_char_colour(embed, char) + return await self.api.get_char_colour(embed, char) async def build_stat_embed_char_gambit( self, user: discord.Member, char: dict, data: dict, stat_type: str ) -> discord.Embed: char_info = "" - race = (await self.get_definition("DestinyRaceDefinition", [char["raceHash"]]))[ + race = (await self.api.get_definition("DestinyRaceDefinition", [char["raceHash"]]))[ str(char["raceHash"]) ] - gender = (await self.get_definition("DestinyGenderDefinition", [char["genderHash"]]))[ + gender = (await self.api.get_definition("DestinyGenderDefinition", [char["genderHash"]]))[ str(char["genderHash"]) ] - char_class = (await self.get_definition("DestinyClassDefinition", [char["classHash"]]))[ - str(char["classHash"]) - ] + char_class = ( + await self.api.get_definition("DestinyClassDefinition", [char["classHash"]]) + )[str(char["classHash"])] char_info += "{race} {gender} {char_class} ".format( race=race["displayProperties"]["name"], gender=gender["displayProperties"]["name"], @@ -2885,13 +3004,13 @@ async def build_stat_embed_char_gambit( if res or resur: embed.add_field(name=_("Resurrections/Received"), value=f"{res}/{resur}") if "emblemPath" in char: - embed.set_thumbnail(url=IMAGE_URL + char["emblemPath"]) + embed.set_thumbnail(url=IMAGE_URL.join(URL(char["emblemPath"]))) for stat, values in data.items(): if values["basic"]["value"] < 0 or stat not in ATTRS: continue embed.add_field(name=ATTRS[stat], value=str(values["basic"]["displayValue"])) - return await self.get_char_colour(embed, char) + return await self.api.get_char_colour(embed, char) @destiny.command() @commands.bot_has_permissions(embed_links=True) @@ -2912,12 +3031,12 @@ async def stats(self, ctx: commands.Context, stat_type: StatsPage) -> None: `` The type of stats to display, available options are: `raid`, `pvp`, `pve`, patrol, story, gambit, and strikes """ - if not await self.has_oauth(ctx): + if not await self.api.has_oauth(ctx): return - async with MyTyping(ctx, ephemeral=False): + async with ctx.typing(ephemeral=False): user = ctx.author try: - chars = await self.get_characters(user) + chars = await self.api.get_characters(user) except Destiny2APIError as e: # log.debug(e) @@ -2944,21 +3063,23 @@ async def weapons_test(self, ctx: commands.Context) -> None: """ Get statistics about your top used weapons """ - if not await self.has_oauth(ctx): + if not await self.api.has_oauth(ctx): return - async with MyTyping(ctx, ephemeral=False): + async with ctx.typing(ephemeral=False): user = ctx.author try: - chars = await self.get_characters( + chars = await self.api.get_characters( user, components=DestinyComponents(DestinyComponentType.characters) ) char_id = list(chars["characters"]["data"].keys())[0] - weapons = await self.get_weapon_history(user, char_id) + weapons = await self.api.get_weapon_history(user, char_id) except Destiny2APIError as e: await self.send_error_msg(ctx, e) return weapon_hashes = [w["referenceId"] for w in weapons["weapons"]] - weapon_def = await self.get_definition("DestinyInventoryItemDefinition", weapon_hashes) + weapon_def = await self.api.get_definition( + "DestinyInventoryItemDefinition", weapon_hashes + ) msg = "" for we in weapon_def.values(): msg += we["displayProperties"]["name"] + "\n" @@ -3041,19 +3162,18 @@ async def manifest_download(self, ctx: commands.Context, d1: bool = False) -> No the newest one. """ if not d1: - async with MyTyping(ctx, ephemeral=False): + async with ctx.typing(ephemeral=False): + error_str = _( + "You need to set your API authentication tokens with `[p]destiny token` first." + ) try: - headers = await self.build_headers() + manifest_data = await self.api.get_manifest_data() except Exception: - await ctx.send( - _( - "You need to set your API authentication tokens with `[p]destiny token` first." - ) - ) + await ctx.send(error_str) return - manifest_data = await self.request_url( - f"{BASE_URL}/Destiny2/Manifest/", headers=headers - ) + if manifest_data is None: + return + version = await self.config.manifest_version() if not version: version = _("Not Downloaded") @@ -3070,9 +3190,9 @@ async def manifest_download(self, ctx: commands.Context, d1: bool = False) -> No _("Would you like to {redownload} manifest?").format(redownload=redownload), ) if pred: - async with MyTyping(ctx, ephemeral=False): + async with ctx.typing(ephemeral=False): try: - version = await self.get_manifest() + version = await self.api.get_manifest() response = _("Manifest Version {version} was downloaded.").format( version=version ) @@ -3084,7 +3204,7 @@ async def manifest_download(self, ctx: commands.Context, d1: bool = False) -> No await ctx.send(_("I will not download the manifest.")) else: try: - version = await self.get_manifest(d1) + version = await self.api.get_manifest(d1) except Exception: log.exception("Error getting D1 manifest") await ctx.send(_("There was an issue downloading the manifest.")) @@ -3092,9 +3212,7 @@ async def manifest_download(self, ctx: commands.Context, d1: bool = False) -> No @destiny.command(with_app_command=False) @commands.is_owner() - async def token( - self, ctx: commands.Context, api_key: str, client_id: str, client_secret: str - ) -> None: + async def token(self, ctx: commands.Context) -> None: """ Set the API tokens for Destiny 2's API @@ -3106,13 +3224,22 @@ async def token( Set the redirect URL to https://localhost/ NOTE: It is strongly recommended to use this command in DM """ - if ctx.interaction: - await ctx.defer() + message = _( + "1. Go to https://www.bungie.net/en/Application \n" + "2. select **Create New App**\n" + "3. Choose **Confidential** OAuth Client type\n" + "4. Select the scope you would like the bot to have access to\n" + "5. Set the redirect URL to https://localhost/\n" + "6. Use `{prefix}set api bungie api_key YOUR_API_KEY client_id " + "YOUR_CLIENT_ID client_secret YOUR_CLIENT_SECRET`\n" + "NOTE: It is strongly recommended to use this command in DM." + ).format(prefix=ctx.prefix) + keys = {"api_key": "", "client_id": "", "client_secret": ""} + view = SetApiView("bungie", keys) + if await ctx.embed_requested(): + em = discord.Embed(description=message, colour=await ctx.bot.get_embed_colour(ctx)) + msg = await ctx.send(embed=em, view=view) else: - await ctx.typing() - await self.config.api_token.api_key.set(api_key) - await self.config.api_token.client_id.set(client_id) - await self.config.api_token.client_secret.set(client_secret) - if ctx.channel.permissions_for(ctx.me).manage_messages: - await ctx.message.delete() - await ctx.send("Destiny 2 API credentials set!") + msg = await ctx.send(message, view=view) + await view.wait() + await msg.edit(view=None) diff --git a/destiny/menus.py b/destiny/menus.py index efe3c1788f..be9ce68658 100644 --- a/destiny/menus.py +++ b/destiny/menus.py @@ -77,7 +77,7 @@ async def callback(self, interaction: discord.Interaction): await pred.wait() if pred.result: try: - await self.view.cog.approve_clan_pending( + await self.view.cog.api.approve_clan_pending( interaction.user, self.clan_id, self.membership_type, @@ -173,7 +173,7 @@ def is_paginating(self): async def format_page(self, menu: menus.MenuPages, page): self.current_item_hash = page["itemHash"] self.current_item_instance = page.get("itemInstanceId", None) - items = await self.cog.get_definition( + items = await self.cog.api.get_definition( "DestinyInventoryItemDefinition", [self.current_item_hash] ) item_data = items[str(self.current_item_hash)] @@ -185,11 +185,11 @@ async def format_page(self, menu: menus.MenuPages, page): if item_data.get("screenshot", None): embed.set_image(url=BASE_URL + item_data["screenshot"]) if self.current_item_instance is not None: - instance_data = await self.cog.get_instanced_item( + instance_data = await self.cog.api.get_instanced_item( menu.author, self.current_item_instance ) perk_hashes = [i["perkHash"] for i in instance_data["perks"]["data"]["perks"]] - perk_info = await self.cog.get_definition( + perk_info = await self.cog.api.get_definition( "DestinyInventoryItemDefinition", perk_hashes ) perk_str = "\n".join(perk["displayProperties"]["name"] for perk in perk_info.values()) @@ -295,7 +295,7 @@ async def callback(self, interaction: discord.Interaction): membership_type = self.view.source.membership_type character_id = self.view.source.character index = self.view.source.current_index - await self.view.cog.equip_loadout( + await self.view.cog.api.equip_loadout( interaction.user, index, character_id, @@ -365,7 +365,7 @@ async def callback(self, interaction: discord.Interaction): item_name = item["displayProperties"]["name"] url = f"https://www.light.gg/db/items/{item_hash}" try: - await self.view.cog.pull_from_postmaster( + await self.view.cog.api.pull_from_postmaster( interaction.user, item_hash, char_id, membership_type, quantity, instance ) self.view.source.remove_item(char_id, item_hash, instance)