Merge 9b6ed59e26 into 42af7b5db7

legendary/api/egs.py

@@ -250,6 +250,28 @@ class EPCAPI:
 
         return records
 
+    def get_game_achievements(self, namespace):
+        r = self.session.post(f'https://{self._store_gql_host}/graphql',
+                              headers={'user-agent': self._store_user_agent},
+                              json=dict(query=egl_game_achievements_query,
+                                        variables=dict(sandboxId=namespace,
+                                                       locale=self.language_code)),
+                              timeout=self.request_timeout)
+        r.raise_for_status()
+        return r.json()
+
+    def get_game_achievements_user(self, namespace):
+        user_id = self.user.get('account_id')
+        r = self.session.post(f'https://{self._store_gql_host}/graphql',
+                              headers={'user-agent': self._store_user_agent},
+                              json=dict(query=egl_game_achievements_user_query,
+                                        variables=dict(sandboxId=namespace,
+                                                       epicAccountId=user_id,
+                                                       locale=self.language_code)),
+                              timeout=self.request_timeout)
+        r.raise_for_status()
+        return r.json()
+
     def get_user_cloud_saves(self, app_name='', manifests=False, filenames=None):
         if app_name:
             app_name += '/manifests/' if manifests else '/'

legendary/cli.py

@@ -66,9 +66,9 @@ class LegendaryCLI:
     @staticmethod
     def _print_json(data, pretty=False):
         if pretty:
-            print(json.dumps(data, indent=2, sort_keys=True))
+            print(json.dumps(data, indent=2, sort_keys=True, default=str))
         else:
-            print(json.dumps(data))
+            print(json.dumps(data, default=str))
 
     def auth(self, args):
         if args.auth_delete:
@@ -2626,6 +2626,53 @@ class LegendaryCLI:
         self.core.install_game(igame)
         logger.info('Finished.')
 
+    def achievements(self, args):
+        if not self.core.login():
+            logger.error('Login failed! Unable to check for EULAs.')
+            exit(1)
+
+        app_name = self._resolve_aliases(args.app_name)
+        game = self.core.get_game(app_name, update_meta=True)
+        if not game:
+            logger.error(f'No game found for "{app_name}"')
+            return
+
+        achievements = self.core.get_achievements(game, update=True)
+        if not achievements:
+            logger.info(f'No achievements found for "{game.app_name}"')
+            return
+
+        if args.json:
+            self._print_json(achievements, args.pretty_json)
+            return
+
+        print(f'* Achievements for "{game.app_title}"')
+        print(f'  Total achievements: {achievements["total_achievements"]}')
+        print(f'  Completed achievements: {achievements["user_unlocked"]}')
+        print(f'  Total XP: {achievements["total_product_xp"]}')
+        print(f'  Player XP: {achievements["user_xp"]}')
+        print(f'  Player awards: {achievements["user_awards"]}')
+
+        for group, title in zip(
+            (achievements['completed'], achievements['in_progress'], achievements['uninitiated']),
+            ('Completed', 'In progress', 'Uninitiated')
+        ):
+            print(f'* {title}')
+            for a in group:
+                print(' - {display_name} | {xp}XP | {description} | Progress: {progress:.1%} | Completed on: {unlock_date}'.format(**a))
+
+        if args.show_hidden:
+            print('* Hidden')
+            for a in achievements['hidden']:
+                print(' - {display_name} | {xp}XP | {description} | Progress: {progress:.1%} | Completed on: {unlock_date}'.format(**a))
+
+        count = sum(
+            map(len, (achievements['completed'], achievements['in_progress'], achievements['uninitiated'], achievements['hidden']))
+        )
+        logger.info(f'Found {count} achievements')
+
+        return
+
 
 def main():
     # Set output encoding to UTF-8 if not outputting to a terminal
@@ -2680,6 +2727,7 @@ def main():
     uninstall_parser = subparsers.add_parser('uninstall', help='Uninstall (delete) a game')
     verify_parser = subparsers.add_parser('verify', help='Verify a game\'s local files',
                                           aliases=('verify-game',), hide_aliases=True)
+    achievements_parser = subparsers.add_parser('achievements', help='List achievement status for a given game')
 
     # hidden commands have no help text
     get_token_parser = subparsers.add_parser('get-token')
@@ -3013,6 +3061,12 @@ def main():
     move_parser.add_argument('--skip-move', dest='skip_move', action='store_true',
                              help='Only change legendary database, do not move files (e.g. if already moved)')
 
+    achievements_parser.add_argument('app_name', metavar='<App Name>', help='Name of the app')
+    achievements_parser.add_argument('--hidden', dest='show_hidden', action='store_true',
+                                     help='Show undiscovered achievements (may contain spoilers)')
+    achievements_parser.add_argument('--json', dest='json', action='store_true',
+                                     help='Output information in JSON format')
+
     args, extra = parser.parse_known_args()
 
     if args.version:
@@ -3113,6 +3167,8 @@ def main():
             cli.crossover_setup(args)
         elif args.subparser_name == 'move':
             cli.move(args)
+        elif args.subparser_name == 'achievements':
+            cli.achievements(args)
     except KeyboardInterrupt:
         logger.info('Command was aborted via KeyboardInterrupt, cleaning up...')
 

legendary/core.py

@@ -314,6 +314,95 @@ class LegendaryCore:
 
         return update_info.get('game_wiki', {}).get(app_name, {}).get(sys_platform)
 
+    def get_user_achievements(self, namespace: str, update: bool = False):
+        if not (achievements := self.lgd.achievements):
+            achievements = {}
+
+        if not achievements or not achievements.get(namespace, None) or update:
+            response = self.egs.get_game_achievements_user(namespace)
+            records = response['data']['PlayerAchievement']['playerAchievementGameRecordsBySandbox']['records']
+            achievements[namespace] = None
+            if records:
+                achievements[namespace] = records[0]
+            self.lgd.achievements = achievements
+
+        return self.lgd.achievements[namespace]
+
+    def get_achievements(self, game: Game, update: bool = False):
+        if not game.achievements.achievements:
+            return None
+
+        user_achievements = self.get_user_achievements(game.namespace, update)
+        user_unlocked = {}
+        if user_achievements:
+            user_unlocked = {
+                ach['playerAchievement']['achievementName']: ach['playerAchievement'] for ach in
+                user_achievements['playerAchievements']
+            }
+
+        achievements = {
+            'total_achievements': game.achievements.total_achievements,
+            'total_product_xp': game.achievements.total_product_xp,
+            'achievement_sets': game.achievements.achievement_sets,
+            'platinum_rarity': game.achievements.platinum_rarity,
+            'completed': [],
+            'in_progress': [],
+            'uninitiated': [],
+            'hidden': [],
+        }
+        achievements.update({
+            'user_unlocked': user_achievements['totalUnlocked'] if user_achievements else 0,
+            'user_xp': user_achievements['totalXP'] if user_achievements else 0,
+            'user_awards': user_achievements['playerAwards'] if user_achievements else [],
+        })
+
+        for item in game.achievements.achievements:
+            game_ach = item['achievement']
+            is_unlocked = game_ach['name'] in user_unlocked
+
+            _unlocked = False
+            _progress = 0.0
+            _unlock_date = None
+            if is_unlocked:
+                user_ach = user_unlocked[game_ach['name']]
+                _unlocked = user_ach['unlocked']
+                _progress = float(user_ach['progress'])
+                _unlock_date = user_ach['unlockDate']
+                _unlock_date = datetime.fromisoformat(_unlock_date[:-1]).replace(
+                    tzinfo=timezone.utc) if _unlock_date != "N/A" else None
+
+            data = {
+                'name': game_ach['name'],
+                'is_base': game_ach['isBase'],
+                'hidden': False if is_unlocked else game_ach['hidden'],
+                'xp': game_ach['XP'],
+                'unlocked': _unlocked,
+                'progress': _progress,
+                'unlock_date': _unlock_date if _unlock_date else None,
+                'display_name': game_ach['unlockedDisplayName'] if is_unlocked else game_ach['lockedDisplayName'],
+                'description': game_ach['unlockedDescription'] if is_unlocked else game_ach['lockedDescription'],
+                'icon_id': game_ach['unlockedIconId'] if is_unlocked else game_ach['lockedIconId'],
+                'icon_link': game_ach['unlockedIconLink'] if is_unlocked else game_ach['lockedIconLink'],
+                'tier': game_ach['tier'],
+                'rarity': game_ach['rarity'],
+            }
+
+            if data['unlocked']:
+                achievements['completed'].append(data)
+            elif 0.0 < data['progress'] < 1.0:
+                achievements['in_progress'].append(data)
+            elif not data['hidden'] and data['progress'] == 0.0:
+                achievements['uninitiated'].append(data)
+            elif data['hidden']:
+                achievements['hidden'].append(data)
+
+        achievements['completed'] = sorted(achievements['completed'], key=lambda a: a['unlock_date'], reverse=True)
+        achievements['in_progress'] = sorted(achievements['in_progress'], key=lambda a: a['progress'], reverse=True)
+        achievements['uninitiated'] = sorted(achievements['uninitiated'], key=lambda a: a['xp'], reverse=False)
+        achievements['hidden'] = sorted(achievements['hidden'], key=lambda a: a['xp'], reverse=False)
+
+        return achievements
+
     def get_sdl_data(self, app_name, platform='Windows'):
         if platform not in ('Win32', 'Windows'):
             app_name = f'{app_name}_{platform}'
@@ -431,15 +520,16 @@ class LegendaryCore:
                 continue
 
             game = self.lgd.get_game_meta(app_name)
-            asset_updated = sidecar_updated = False
+            asset_updated = sidecar_updated = achievements_updated = False
             if game:
                 asset_updated = any(game.app_version(_p) != app_assets[_p].build_version for _p in app_assets.keys())
                 # assuming sidecar data is the same for all platforms, just check the baseline (Windows) for updates.
                 sidecar_updated = (app_assets['Windows'].sidecar_rev > 0 and
                                    (not game.sidecar or game.sidecar.rev != app_assets['Windows'].sidecar_rev))
+                achievements_updated = not game.achievements or asset_updated
                 games[app_name] = game
 
-            if update_assets and (not game or force_refresh or (game and (asset_updated or sidecar_updated))):
+            if update_assets and (not game or force_refresh or (game and (asset_updated or sidecar_updated or achievements_updated))):
                 self.log.debug(f'Scheduling metadata update for {app_name}')
                 # namespace/catalog item are the same for all platforms, so we can just use the first one
                 _ga = next(iter(app_assets.values()))
@@ -463,8 +553,12 @@ class LegendaryCore:
                     sidecar_json = json.loads(manifest_info['sidecar']['config'])
                     sidecar = Sidecar(config=sidecar_json, rev=manifest_info['sidecar']['rvn'])
 
+            self.log.debug(f'Updating achivement information for {app_name}...')
+            achievements_api_response = self.egs.get_game_achievements(namespace)
+            achievements = Achievements.from_egs_json(achievements_api_response)
+
             game = Game(app_name=app_name, app_title=eg_meta['title'], metadata=eg_meta, asset_infos=assets[app_name],
-                        sidecar=sidecar)
+                        sidecar=sidecar, achievements=achievements)
             self.lgd.set_game_meta(game.app_name, game)
             games[app_name] = game
             try:

legendary/lfs/lgndry.py

@@ -37,6 +37,8 @@ class LGDLFS:
         self._user_data = None
         # EGS entitlements
         self._entitlements = None
+        # EGS achievements
+        self._achievements = None
         # EGS asset data
         self._assets = None
         # EGS metadata
@@ -196,6 +198,27 @@ class LGDLFS:
         json.dump(entitlements, open(os.path.join(self.path, 'entitlements.json'), 'w'),
                   indent=2, sort_keys=True)
 
+    @property
+    def achievements(self):
+        if self._achievements is not None:
+            return self._achievements
+
+        try:
+            self._achievements = json.load(open(os.path.join(self.path, 'achievements.json')))
+            return self._achievements
+        except Exception as e:
+            self.log.debug(f'Failed to load achievements data: {e!r}')
+            return None
+
+    @achievements.setter
+    def achievements(self, achievements):
+        if achievements is None:
+            raise ValueError('Achievements is none!')
+
+        self._achievements = achievements
+        json.dump(achievements, open(os.path.join(self.path, 'achievements.json'), 'w'),
+                  indent=2, sort_keys=True)
+
     @property
     def assets(self):
         if self._assets is None:

legendary/models/game.py

@@ -63,6 +63,41 @@ class Sidecar:
         )
 
 
+@dataclass
+class Achievements:
+    namespace: str
+    total_achievements: int
+    total_product_xp: int
+    achievement_sets: List = field(default_factory=list)
+    platinum_rarity: Dict = field(default_factory=dict)
+    achievements: List = field(default_factory=list)
+
+    @classmethod
+    def from_egs_json(cls, json):
+        json = json['data']['Achievement']['productAchievementsRecordBySandbox']
+        tmp = cls(
+            namespace=json.get('sandboxId', ''),
+            total_achievements=json.get('totalAchievements', 0),
+            total_product_xp=json.get('totalProductXP', 0),
+            achievement_sets=json.get('achievementSets', []),
+            platinum_rarity=json.get('platinumRarity', {}),
+            achievements=json.get('achievements', []),
+        )
+        return tmp
+
+    @classmethod
+    def from_json(cls, json):
+        tmp = cls(
+            namespace=json.get('namespace', ''),
+            total_achievements=json.get('total_achievements', 0),
+            total_product_xp=json.get('total_product_xp', 0),
+            achievement_sets=json.get('achievement_sets', []),
+            platinum_rarity=json.get('platinum_rarity', {}),
+            achievements=json.get('achievements', []),
+        )
+        return tmp
+
+
 @dataclass
 class Game:
     """
@@ -75,6 +110,7 @@ class Game:
     base_urls: List[str] = field(default_factory=list)
     metadata: Dict = field(default_factory=dict)
     sidecar: Optional[Sidecar] = None
+    achievements: Optional[Achievements] = None
 
     def app_version(self, platform='Windows'):
         if platform not in self.asset_infos:
@@ -155,6 +191,9 @@ class Game:
         if sidecar := json.get('sidecar', None):
             tmp.sidecar = Sidecar.from_json(sidecar)
 
+        if achievements := json.get('achievements', None):
+            tmp.achievements = Achievements.from_json(achievements)
+
         tmp.base_urls = json.get('base_urls', list())
         return tmp
 
@@ -163,8 +202,10 @@ class Game:
         """This is just here so asset_infos gets turned into a dict as well"""
         assets_dictified = {k: v.__dict__ for k, v in self.asset_infos.items()}
         sidecar_dictified = self.sidecar.__dict__ if self.sidecar else None
+        achievements_dictified = self.achievements.__dict__ if self.achievements else None
         return dict(metadata=self.metadata, asset_infos=assets_dictified, app_name=self.app_name,
-                    app_title=self.app_title, base_urls=self.base_urls, sidecar=sidecar_dictified)
+                    app_title=self.app_title, base_urls=self.base_urls, sidecar=sidecar_dictified,
+                    achievements=achievements_dictified)
 
 
 @dataclass

legendary/models/gql.py

@@ -58,4 +58,88 @@ mutation claimUplayCode($accountId: String!, $uplayAccountId: String!, $gameId:
     }
   }
 }
-'''
+'''
+
+egl_game_achievements_query = '''
+query Achievement($sandboxId: String!, $locale: String!) {
+  Achievement {
+    productAchievementsRecordBySandbox(sandboxId: $sandboxId, locale: $locale) {
+      sandboxId
+      totalAchievements
+      totalProductXP
+      achievementSets {
+        achievementSetId
+        isBase
+        totalAchievements
+        totalXP
+      }
+      platinumRarity {
+        percent
+      }
+      achievements {
+        achievement {
+          name
+          hidden
+          isBase
+          unlockedDisplayName
+          lockedDisplayName
+          unlockedDescription
+          lockedDescription
+          unlockedIconId
+          lockedIconId
+          XP
+          flavorText
+          unlockedIconLink
+          lockedIconLink
+          tier {
+            name
+            hexColor
+            min
+            max
+          }
+          rarity {
+            percent
+          }
+        }
+      }
+    }
+  }
+}
+'''
+
+egl_game_achievements_user_query = '''
+query PlayerAchievement($epicAccountId: String!, $sandboxId: String!) {
+  PlayerAchievement {
+    playerAchievementGameRecordsBySandbox(epicAccountId: $epicAccountId, sandboxId: $sandboxId) {
+      records {
+        totalXP
+        totalUnlocked
+        playerAwards {
+          awardType
+          unlockedDateTime
+          achievementSetId
+        }
+        achievementSets {
+          achievementSetId
+          isBase
+          totalUnlocked
+          totalXP
+        }
+        playerAchievements {
+          playerAchievement {
+            sandboxId
+            epicAccountId
+            unlocked
+            progress
+            XP
+            unlockDate
+            achievementName
+            isBase
+            achievementSetId
+          }
+        }
+      }
+    }
+  }
+}
+'''