feat: add command to change all items transmog to the currently equip…

…ped stuff

Backend/bungio/manifest.py

@@ -14,6 +14,7 @@
     DestinyPresentationNodeDefinition,
     DestinyRecordDefinition,
     DestinySeasonPassDefinition,
+    DestinySocketTypeDefinition,
 )
 
 from Backend.bungio.client import get_bungio_client
@@ -39,6 +40,7 @@
 get_item_lock = asyncio.Lock()
 get_all_weapons_lock = asyncio.Lock()
 get_seals_lock = asyncio.Lock()
+get_sockets_lock = asyncio.Lock()
 
 
 class CRUDManifest:
@@ -56,6 +58,9 @@ class CRUDManifest:
     # DestinyLoreModel
     _manifest_lore: dict[int, DestinyLoreModel] = {}
 
+    # DestinySocketTypeDefinition
+    _manifest_sockets: dict[int, DestinySocketTypeDefinition] = {}
+
     # DestinyRecordDefinition
     _manifest_triumphs: dict[int, DestinyRecordDefinition] = {}
     _manifest_seals: dict[DestinyPresentationNodeDefinition, list[DestinyRecordDefinition]] = {}
@@ -66,44 +71,58 @@ class CRUDManifest:
     _manifest_grandmasters: list[DestinyActivityModel] = []
     _manifest_interesting_solos: dict[str, list[DestinyActivityModel]] = {}  # Key: activity_category
 
-    async def reset(self):
+    async def reset(self, soft: bool = False):
         """Reset the caches after a manifest update"""
 
         self._manifest_weapons = {}
-        await destiny_manifest.get_all_weapons()
+        if not soft:
+            await destiny_manifest.get_all_weapons()
+
+        self._manifest_sockets = {}
+        if not soft:
+            await destiny_manifest.get_all_sockets()
 
         self._manifest_season_pass_definition = None  # noqa
         await destiny_manifest.get_current_season_pass()
 
         self._manifest_seasonal_challenges_definition = None  # noqa
-        await self.get_seasonal_challenges_definition()
+        if not soft:
+            await self.get_seasonal_challenges_definition()
 
         self._manifest_items = {}
         # per item, so not populated here
 
         self._manifest_collectibles = {}
-        await destiny_manifest.get_all_collectibles()
+        if not soft:
+            await destiny_manifest.get_all_collectibles()
 
         self._manifest_triumphs = {}
-        await destiny_manifest.get_all_triumphs()
+        if not soft:
+            await destiny_manifest.get_all_triumphs()
 
         self._manifest_seals = {}
-        await destiny_manifest.get_seals()
+        if not soft:
+            await destiny_manifest.get_seals()
 
         self._manifest_lore = {}
-        await destiny_manifest.get_all_lore()
+        if not soft:
+            await destiny_manifest.get_all_lore()
 
         self._manifest_catalysts = []
-        await destiny_manifest.get_catalysts()
+        if not soft:
+            await destiny_manifest.get_catalysts()
 
         self._manifest_activities = {}
-        await destiny_manifest.get_all_activities()
+        if not soft:
+            await destiny_manifest.get_all_activities()
 
         self._manifest_grandmasters = []
-        await destiny_manifest.get_grandmaster_nfs()
+        if not soft:
+            await destiny_manifest.get_grandmaster_nfs()
 
         self._manifest_interesting_solos = {}
-        await destiny_manifest.get_challenging_solo_activities()
+        if not soft:
+            await destiny_manifest.get_challenging_solo_activities()
 
     async def get_all_weapons(self) -> dict[int, DestinyInventoryItemDefinition]:
         """Return all weapons"""
@@ -119,6 +138,18 @@ async def get_all_weapons(self) -> dict[int, DestinyInventoryItemDefinition]:
                     self._manifest_items[result.hash] = result
         return self._manifest_weapons
 
+    async def get_all_sockets(self) -> dict[int, DestinySocketTypeDefinition]:
+        """Return all sockets"""
+
+        async with get_sockets_lock:
+            if not self._manifest_sockets:
+                results: list[DestinySocketTypeDefinition] = await get_bungio_client().manifest.fetch_all(
+                    manifest_class=DestinySocketTypeDefinition
+                )
+                for result in results:
+                    self._manifest_sockets[result.hash] = result
+        return self._manifest_sockets
+
     async def get_weapon(self, weapon_id: int) -> DestinyInventoryItemDefinition:
         """Gets weapon"""
 

Backend/core/destiny/profile.py

@@ -7,22 +7,31 @@
 
 from anyio import to_thread
 from bungio.models import (
+    MISSING,
+    DestinyCharacter,
     DestinyClass,
     DestinyCollectibleComponent,
     DestinyCollectibleState,
     DestinyCraftableComponent,
     DestinyHistoricalStatsAccountResult,
+    DestinyInsertPlugsActionRequest,
+    DestinyInsertPlugsFreeActionRequest,
+    DestinyInsertPlugsRequestEntry,
+    DestinyInventoryItemDefinition,
     DestinyItemComponent,
     DestinyMetricComponent,
     DestinyProfileResponse,
     DestinyRecordComponent,
     DestinyRecordState,
+    DestinySocketArrayType,
     DestinyStatsGroupType,
     GroupsForMemberFilter,
     GroupType,
+    ItemState,
 )
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from Backend.bungio.client import get_bungio_client
 from Backend.bungio.manifest import destiny_manifest
 from Backend.core.errors import CustomException
 from Backend.crud import crud_activities, discord_users
@@ -81,6 +90,111 @@ def __post_init__(self):
         self.destiny_id = self.user.destiny_id
         self.system = self.user.system
 
+    async def set_transmog(self, character_id: int):
+        """Sets the transmog and the shader off all items to the one currently equiped"""
+
+        def get_socket_indexes(
+            definition: DestinyInventoryItemDefinition, transmog_socket_hash: int
+        ) -> tuple[int, int]:
+            shader_index, transmog_index = -1, -1
+            for i, socket in enumerate(definition.sockets.socket_entries):
+                if socket.socket_type_hash == shader_socket_hash:
+                    shader_index = i
+                if socket.socket_type_hash == transmog_socket_hash:
+                    transmog_index = i
+            return shader_index, transmog_index
+
+        char_ids = await self.get_character_ids()
+        if character_id not in char_ids:
+            raise CustomException("CharacterIdNotFound")
+        bungio_client = get_bungio_client()
+        bungio_user = self.user.bungio_user
+        auth_data = self.user.auth
+
+        shader_socket_hash = 2321980680
+
+        # get all the items and instances
+        items = await self.get_character_items(character_id)
+        sockets = self._profile.item_components.sockets.data
+        character = self._profile.characters.data[character_id]
+
+        # get all the correct sockets
+        all_sockets = await destiny_manifest.get_all_sockets()
+        correct_sockets = {"helmet": None, "gauntlets": None, "chest": None, "leg": None, "class": None}
+        for socket in all_sockets.values():
+            # <SocketCategory "ARMOR COSMETICS">
+            if socket.socket_category_hash == 1926152773:
+                # categoryIdentifier:"armor_skins_empty"
+                if socket.plug_whitelist[0].category_identifier == "armor_skins_empty":
+                    for plug in socket.plug_whitelist:
+                        if plug.category_identifier.startswith(f"armor_skins_{character.class_type.name.lower()}"):
+                            if plug.category_identifier.endswith("head"):
+                                correct_sockets["helmet"] = socket
+                            elif plug.category_identifier.endswith("arms"):
+                                correct_sockets["gauntlets"] = socket
+                            elif plug.category_identifier.endswith("chest"):
+                                correct_sockets["chest"] = socket
+                            elif plug.category_identifier.endswith("legs"):
+                                correct_sockets["leg"] = socket
+                            elif plug.category_identifier.endswith("class"):
+                                correct_sockets["class"] = socket
+                            break
+
+        for slot_name, data in items.items():
+            equipped = data["equipped"][0]
+            equipped_definition = await destiny_manifest.get_item(item_id=equipped.item_hash)
+
+            # skip non legendaries
+            if equipped_definition.inventory.tier_type_name != "Legendary":
+                continue
+
+            equipped_sockets = sockets[equipped.item_instance_id].sockets
+            shader_index, transmog_index = get_socket_indexes(
+                equipped_definition, transmog_socket_hash=correct_sockets[slot_name].hash
+            )
+
+            equipped_shader = equipped_sockets[shader_index].plug_hash
+            equipped_transmog = equipped_sockets[transmog_index].plug_hash
+
+            inventory = data["inventory"]
+
+            for item in inventory:
+                item_definition = await destiny_manifest.get_item(item_id=item.item_hash)
+
+                # skip non legendaries
+                if item_definition.inventory.tier_type_name != "Legendary":
+                    continue
+
+                item_sockets = sockets[item.item_instance_id].sockets
+                shader_index, transmog_index = get_socket_indexes(
+                    item_definition, transmog_socket_hash=correct_sockets[slot_name].hash
+                )
+                if shader_index == -1 or transmog_index == -1:
+                    continue
+
+                data = DestinyInsertPlugsFreeActionRequest(
+                    character_id=character_id,
+                    item_id=item.item_instance_id,
+                    membership_type=bungio_user.membership_type,
+                    plug=DestinyInsertPlugsRequestEntry(
+                        plug_item_hash=equipped_shader,
+                        socket_array_type=DestinySocketArrayType.DEFAULT,
+                        socket_index=shader_index,
+                    ),
+                )
+
+                # shader
+                if item_sockets[shader_index].plug_hash != equipped_shader:
+                    await bungio_client.api.insert_socket_plug_free(data=data, auth=auth_data)
+                # transmog
+                if (
+                    item_sockets[transmog_index].plug_hash != equipped_transmog
+                    and item_definition.hash != equipped_transmog
+                ):
+                    data.plug.plug_item_hash = equipped_transmog
+                    data.plug.socket_index = transmog_index
+                    await bungio_client.api.insert_socket_plug_free(data=data, auth=auth_data)
+
     async def get_clan(self) -> DestinyClanModel:
         """Return the user's clan"""
 
@@ -715,6 +829,83 @@ async def get_time_played(
             character_class=character_class,
         )
 
+    async def get_character_items(
+        self, character_id: int
+    ) -> dict[
+        Literal["helmet", "gauntlets", "chest", "leg", "class"],
+        dict[Literal["equipped", "inventory"], list[DestinyItemComponent]],
+    ]:
+        """Get all items that belong to a character"""
+
+        items: dict[
+            Literal["helmet", "gauntlet", "chest", "leg", "class"],
+            dict[Literal["equipped", "inventory"], list[DestinyItemComponent]],
+        ] = {
+            "helmet": {"equipped": [], "inventory": []},
+            "gauntlets": {"equipped": [], "inventory": []},
+            "chest": {"equipped": [], "inventory": []},
+            "leg": {"equipped": [], "inventory": []},
+            "class": {"equipped": [], "inventory": []},
+        }
+
+        result = await self.__get_profile(force=True)
+        character = result.characters.data[character_id]
+
+        # get character equipped
+        for item in result.character_equipment.data[character_id].items:
+            try:
+                bucket_name = DestinyInventoryBucketEnum(item.bucket_hash).name.lower()
+            except ValueError:
+                continue
+            if bucket_name in items:
+                try:
+                    items[bucket_name]["equipped"].append(item)
+                except KeyError:
+                    continue
+
+        # get stuff from inventories
+        to_check = [
+            DestinyInventoryBucketEnum.HELMET,
+            DestinyInventoryBucketEnum.GAUNTLETS,
+            DestinyInventoryBucketEnum.CHEST,
+            DestinyInventoryBucketEnum.LEG,
+            DestinyInventoryBucketEnum.CLASS,
+        ]
+        for item in result.character_inventories.data[character_id].items:
+            try:
+                bucket = DestinyInventoryBucketEnum(item.bucket_hash)
+            except ValueError:
+                continue
+            if bucket in to_check:
+                item_definition = await destiny_manifest.get_item(item_id=item.item_hash)
+                if item_definition.class_type == character.class_type:
+                    bucket_name = bucket.name.lower()
+                    try:
+                        items[bucket_name]["inventory"].append(item)
+                    except KeyError:
+                        continue
+
+        # get stuff in vault
+        for item in result.profile_inventory.data.items:
+            try:
+                bucket = DestinyInventoryBucketEnum(item.bucket_hash)
+            except ValueError:
+                continue
+            if bucket == DestinyInventoryBucketEnum.VAULT and item.item_instance_id:
+                item_definition = await destiny_manifest.get_item(item_id=item.item_hash)
+                if item_definition.class_type == character.class_type:
+                    try:
+                        bucket = DestinyInventoryBucketEnum(item_definition.inventory.bucket_type_hash)
+                    except ValueError:
+                        continue
+                    bucket_name = bucket.name.lower()
+                    try:
+                        items[bucket_name]["inventory"].append(item)
+                    except KeyError:
+                        continue
+
+        return items
+
     async def __get_inventory_bucket(
         self, *buckets: DestinyInventoryBucketEnum
     ) -> dict[
@@ -859,12 +1050,12 @@ def add_info(
 
         return items
 
-    async def __get_profile(self) -> DestinyProfileResponse:
+    async def __get_profile(self, force: bool = False) -> DestinyProfileResponse:
         """
         Return info from the profile call
         https://bungie-net.github.io/multi/schema_Destiny-DestinyComponentType.html#schema_Destiny-DestinyComponentType
         """
-        if not self._profile:
+        if not self._profile or force:
             # just calling nearly all of them. Don't need all quite yet, but who knows what the future will bring
             components = [
                 100,
@@ -898,7 +1089,12 @@ async def __get_profile(self) -> DestinyProfileResponse:
                 1300,
             ]
 
-            self._profile = await self.user.bungio_user.get_profile(components=components, auth=self.user.auth)
+            call = self.user.bungio_user.get_profile(components=components, auth=self.user.auth)
+            if force:
+                async with get_bungio_client().http.session.disabled():
+                    self._profile = await call
+            else:
+                self._profile = await call
 
             # get bungie name
             bungie_name = self._profile.profile.data.user_info.full_bungie_name

Backend/endpoints/destiny/account.py

@@ -4,7 +4,7 @@
 from Backend.core.destiny.profile import DestinyProfile
 from Backend.crud import discord_users
 from Backend.database import acquire_db_session
-from Shared.networkingSchemas import BoolModel, DestinyAllMaterialsModel, NameModel, ValueModel
+from Shared.networkingSchemas import BoolModel, DestinyAllMaterialsModel, EmptyResponseModel, NameModel, ValueModel
 from Shared.networkingSchemas.destiny import (
     BoolModelRecord,
     DestinyCatalystsModel,
@@ -26,6 +26,18 @@
 )
 
 
+@router.post("/{character_id}/set_transmog", response_model=EmptyResponseModel)  # todo test
+async def set_transmog(guild_id: int, discord_id: int, character_id: int):
+    """Sets the transmog and the shader off all items to the one currently equiped"""
+
+    async with acquire_db_session() as db:
+        user = await discord_users.get_profile_from_discord_id(discord_id, db=db)
+        profile = DestinyProfile(db=db, user=user)
+
+        await profile.set_transmog(character_id=character_id)
+        return EmptyResponseModel()
+
+
 @router.get("/name", response_model=NameModel)  # has test
 async def destiny_name(guild_id: int, discord_id: int):
     """Return the bungie name"""