[api/cli/core/lfs] Add Legendary update check

legendary/api/lgd.py

@@ -0,0 +1,22 @@
+# !/usr/bin/env python
+# coding: utf-8
+
+import legendary
+import requests
+import logging
+
+
+class LGDAPI:
+    _user_agent = f'Legendary/{legendary.__version__}'
+    _api_host = 'legendary.rodney.io'
+
+    def __init__(self):
+        self.session = requests.session()
+        self.log = logging.getLogger('LGDAPI')
+        self.session.headers['User-Agent'] = self._user_agent
+
+    def get_version_information(self):
+        r = self.session.get(f'https://{self._api_host}/version.json',
+                             timeout=10.0)
+        r.raise_for_status()
+        return r.json()

legendary/cli.py

@@ -1065,6 +1065,8 @@ class LegendaryCLI:
                     exit(1)
             except ValueError:
                 pass
+            # if automatic checks are off force an update here
+            self.core.check_for_updates(force=True)
 
         if not self.core.lgd.userdata:
             user_name = '<not logged in>'
@@ -1089,6 +1091,16 @@ class LegendaryCLI:
         print(f'Games installed: {games_installed}')
         print(f'EGL Sync enabled: {self.core.egl_sync_enabled}')
         print(f'Config directory: {self.core.lgd.path}')
+        print(f'\nLegendary version: {__version__} - "{__codename__}"')
+        print(f'Update available: {"yes" if self.core.update_available else "no"}')
+        if self.core.update_available:
+            if update_info := self.core.get_update_info():
+                print(f'- New version: {update_info["version"]} - "{update_info["name"]}"')
+                print(f'- Release summary:\n{update_info["summary"]}\n- Release URL: {update_info["gh_url"]}')
+                if update_info['critical']:
+                    print('! This update is recommended as it fixes major issues.')
+            # prevent update message on close
+            self.core.update_available = False
 
     def cleanup(self, args):
         before = self.core.lgd.get_dir_size()
@@ -1413,6 +1425,15 @@ def main():
     except KeyboardInterrupt:
         logger.info('Command was aborted via KeyboardInterrupt, cleaning up...')
 
+    # show note if update is available
+    if cli.core.update_available:
+        if update_info := cli.core.get_update_info():
+            print(f'\nLegendary update available!')
+            print(f'- New version: {update_info["version"]} - "{update_info["name"]}"')
+            print(f'- Release summary:\n{update_info["summary"]}\n- Release URL: {update_info["gh_url"]}')
+            if update_info['critical']:
+                print('! This update is recommended as it fixes major issues.')
+
     cli.core.exit()
     ql.stop()
     exit(0)

legendary/core.py

@@ -16,7 +16,9 @@ from requests.exceptions import HTTPError
 from typing import List, Dict
 from uuid import uuid4
 
+from legendary import __version__
 from legendary.api.egs import EPCAPI
+from legendary.api.lgd import LGDAPI
 from legendary.downloader.mp.manager import DLManager
 from legendary.lfs.egl import EPCLFS
 from legendary.lfs.lgndry import LGDLFS
@@ -50,6 +52,7 @@ class LegendaryCore:
         self.egs = EPCAPI()
         self.lgd = LGDLFS()
         self.egl = EPCLFS()
+        self.lgdapi = LGDAPI()
 
         # on non-Windows load the programdata path from config
         if os.name != 'nt':
@@ -75,6 +78,8 @@ class LegendaryCore:
         else:
             self.log.warning(f'Could not determine locale, falling back to en-US')
 
+        self.update_available = False
+
     def auth(self, username, password):
         """
         Attempts direct non-web login, raises CaptchaError if manual login is required
@@ -153,6 +158,15 @@ class LegendaryCore:
         if not self.lgd.userdata:
             raise ValueError('No saved credentials')
 
+        # run update check
+        if self.update_check_enabled():
+            try:
+                self.check_for_updates()
+            except Exception as e:
+                self.log.warning(f'Checking for Legendary updates failed: {e!r}')
+        else:
+            self.apply_lgd_config()
+
         if self.lgd.userdata['expires_at']:
             dt_exp = datetime.fromisoformat(self.lgd.userdata['expires_at'][:-1])
             dt_now = datetime.utcnow()
@@ -185,6 +199,40 @@ class LegendaryCore:
         self.lgd.userdata = userdata
         return True
 
+    def update_check_enabled(self):
+        return self.lgd.config.getboolean('Legendary', 'enable_update_check',
+                                          fallback=os.name == 'nt')
+
+    def check_for_updates(self, force=False):
+        def version_tuple(v):
+            return tuple(map(int, (v.split('.'))))
+
+        cached = self.lgd.get_cached_version()
+        version_info = cached['data']
+        if force or not version_info or (datetime.now().timestamp() - cached['last_update']) > 24*3600:
+            version_info = self.lgdapi.get_version_information()
+            self.lgd.set_cached_version(version_info)
+
+        web_version = version_info['release_info']['version']
+        self.update_available = version_tuple(web_version) > version_tuple(__version__)
+        self.apply_lgd_config(version_info)
+
+    def apply_lgd_config(self, version_info=None):
+        """Applies configuration options returned by update API"""
+        if not version_info:
+            version_info = self.lgd.get_cached_version()['data']
+            # if cached data is invalid
+            if not version_info:
+                self.log.debug('No cached legendary config to apply.')
+                return
+
+        if 'egl_config' in version_info:
+            self.egs.update_egs_params(version_info['egl_config'])
+        # todo update sid auth/downloader UA and game overrides
+
+    def get_update_info(self):
+        return self.lgd.get_cached_version()['data'].get('release_info')
+
     def get_assets(self, update_assets=False, platform_override=None) -> List[GameAsset]:
         # do not save and always fetch list when platform is overridden
         if platform_override:

legendary/lfs/lgndry.py

@@ -5,6 +5,7 @@ import os
 import logging
 
 from pathlib import Path
+from time import time
 
 from legendary.models.game import *
 from legendary.utils.config import LGDConf
@@ -28,6 +29,8 @@ class LGDLFS:
         self._assets = None
         # EGS metadata
         self._game_metadata = dict()
+        # Legendary update check info
+        self._update_info = None
         # Config with game specific settings (e.g. start parameters, env variables)
         self.config = LGDConf(comment_prefixes='/', allow_no_value=True)
         self.config.optionxform = str
@@ -285,3 +288,18 @@ class LGDLFS:
 
     def get_dir_size(self):
         return sum(f.stat().st_size for f in Path(self.path).glob('**/*') if f.is_file())
+
+    def get_cached_version(self):
+        try:
+            self._update_info = json.load(open(os.path.join(self.path, 'version.json')))
+            return self._update_info
+        except Exception as e:
+            self.log.debug(f'Failed to load cached update data: {e!r}')
+            return dict(last_update=0, data=None)
+
+    def set_cached_version(self, version_data):
+        if not version_data:
+            return
+        self._update_info = dict(last_update=time(), data=version_data)
+        json.dump(self._update_info, open(os.path.join(self.path, 'version.json'), 'w'),
+                  indent=2, sort_keys=True)