diff --git a/legendary/api/egs.py b/legendary/api/egs.py index f88707d..8469e2c 100644 --- a/legendary/api/egs.py +++ b/legendary/api/egs.py @@ -252,6 +252,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 '/' diff --git a/legendary/cli.py b/legendary/cli.py index 5a93a78..4d96f01 100644 --- a/legendary/cli.py +++ b/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: @@ -2642,6 +2642,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 @@ -2696,6 +2743,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') @@ -3031,6 +3079,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='', 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: @@ -3131,6 +3185,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...') diff --git a/legendary/core.py b/legendary/core.py index ec50a07..3f4492b 100644 --- a/legendary/core.py +++ b/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: diff --git a/legendary/lfs/lgndry.py b/legendary/lfs/lgndry.py index dfab258..16fcdf3 100644 --- a/legendary/lfs/lgndry.py +++ b/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: diff --git a/legendary/models/game.py b/legendary/models/game.py index c03cb95..749606e 100644 --- a/legendary/models/game.py +++ b/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: @@ -159,6 +195,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 @@ -167,8 +206,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 diff --git a/legendary/models/gql.py b/legendary/models/gql.py index d51a95a..3f8db02 100644 --- a/legendary/models/gql.py +++ b/legendary/models/gql.py @@ -58,4 +58,88 @@ mutation claimUplayCode($accountId: String!, $uplayAccountId: String!, $gameId: } } } -''' \ No newline at end of file +''' + +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 + } + } + } + } + } +} +'''