diff --git a/.gitignore b/.gitignore index 480b53b7..e6c9a0ca 100644 --- a/.gitignore +++ b/.gitignore @@ -4,4 +4,6 @@ __pycache__/ venv/ .pybuild *.egg-info -*.code-workspace \ No newline at end of file +*.code-workspace +docker-compose.yaml +/.DS_Store \ No newline at end of file diff --git a/chimera_app/config.py b/chimera_app/config.py index 6a7f7e1e..7490dc64 100644 --- a/chimera_app/config.py +++ b/chimera_app/config.py @@ -13,30 +13,31 @@ def merge_single_level(src1, src2): result = {} for key in keys: - if key in src1 and key in src2: - result[key] = src1[key] | src2[key] - elif key in src1: - result[key] = src1[key] - else: - result[key] = src2[key] + if key in src1 and key in src2: + result[key] = src1[key] | src2[key] + elif key in src1: + result[key] = src1[key] + else: + result[key] = src2[key] return result -CONTENT_SHARE_ONLY = os.environ.get('CONTENT_SHARE_ONLY') == 'true' + +CONTENT_SHARE_ONLY = os.environ.get("CONTENT_SHARE_ONLY") == "true" RESOURCE_DIR = os.getcwd() -if not os.path.isfile(os.path.join(RESOURCE_DIR, 'views/base.tpl')): +if not os.path.isfile(os.path.join(RESOURCE_DIR, "views/base.tpl")): RESOURCE_DIR = "/usr/share/chimera" -BIN_PATH = os.path.abspath('libexec') +BIN_PATH = os.path.abspath("libexec") if not os.path.isdir(BIN_PATH): BIN_PATH = "/usr/libexec/chimera" -BANNER_DIR = context.DATA_HOME + '/chimera/images' -DATA_DIR = context.DATA_HOME + '/chimera/data' -CONTENT_DIR = context.DATA_HOME + '/chimera/content' -SETTINGS_DIR = context.CONFIG_HOME + '/chimera' -UPLOADS_DIR = os.path.join(context.CACHE_HOME, 'chimera', 'uploads') +BANNER_DIR = context.DATA_HOME + "/chimera/images" +DATA_DIR = context.DATA_HOME + "/chimera/data" +CONTENT_DIR = context.DATA_HOME + "/chimera/content" +SETTINGS_DIR = context.CONFIG_HOME + "/chimera" +UPLOADS_DIR = os.path.join(context.CACHE_HOME, "chimera", "uploads") SETTINGS_DEFAULT = { "enable_remote_launch": False, @@ -46,23 +47,25 @@ def merge_single_level(src1, src2): "ftp_password": generate_password(12), "ftp_port": 2121, "keep_password": False, + "steamgriddb_api_key": "423ef7be0f4b9f8cfa1a471149c5b72c", + "user_steamgriddb_key_set": False, } SESSION_OPTIONS = { - 'session.cookie_expires': True, - 'session.httponly': True, - 'session.timeout': 3600 * 2, - 'session.type': 'memory', - 'session.validate_key': True, + "session.cookie_expires": True, + "session.httponly": True, + "session.timeout": 3600 * 2, + "session.type": "memory", + "session.validate_key": True, } SETTINGS_HANDLER = Settings(SETTINGS_DIR, SETTINGS_DEFAULT) # settings overrides for exclusive content sharing mode if CONTENT_SHARE_ONLY: - SETTINGS_HANDLER.set_setting('enable_content_sharing', True) - SETTINGS_HANDLER.set_setting('enable_remote_launch', False) - SETTINGS_HANDLER.set_setting('enable_ftp_server', False) + SETTINGS_HANDLER.set_setting("enable_content_sharing", True) + SETTINGS_HANDLER.set_setting("enable_remote_launch", False) + SETTINGS_HANDLER.set_setting("enable_ftp_server", False) FTP_SERVER = None STORAGE_HANDLER = None SSH_KEY_HANDLER = None @@ -73,12 +76,13 @@ def merge_single_level(src1, src2): FTP_SERVER = FTPServer(SETTINGS_HANDLER) STORAGE_HANDLER = StorageConfig() - SSH_KEY_HANDLER = SSHKeys(os.path.expanduser('~/.ssh/authorized_keys')) + SSH_KEY_HANDLER = SSHKeys(os.path.expanduser("~/.ssh/authorized_keys")) AUTHENTICATOR = Authenticator(BIN_PATH, password_length=5) -STEAMGRID_HANDLER = Steamgrid("423ef7be0f4b9f8cfa1a471149c5b72c") +STEAMGRID_HANDLER = Steamgrid(SETTINGS_HANDLER.get_setting("steamgriddb_api_key")) + PLATFORMS_DEFAULT = { "32x": { @@ -225,42 +229,48 @@ def merge_single_level(src1, src2): # set a default cmd property on all default platforms for platform_id in PLATFORMS_DEFAULT: - PLATFORMS_DEFAULT[platform_id]['cmd'] = platform_id + PLATFORMS_DEFAULT[platform_id]["cmd"] = platform_id # merge user settings with default platforms PLATFORMS = PLATFORMS_DEFAULT -PLATFORM_SETTINGS = SETTINGS_HANDLER.get_setting('platforms') +PLATFORM_SETTINGS = SETTINGS_HANDLER.get_setting("platforms") if PLATFORM_SETTINGS: PLATFORMS = merge_single_level(PLATFORMS_DEFAULT, PLATFORM_SETTINGS) # set id property on all platforms for platform_id in PLATFORMS: - PLATFORMS[platform_id]['id'] = platform_id + PLATFORMS[platform_id]["id"] = platform_id # these platforms do not support content sharing so disable them when running in exclusive content sharing mode if CONTENT_SHARE_ONLY: - PLATFORMS['epic-store']['enabled'] = False - PLATFORMS['gog']['enabled'] = False - PLATFORMS['flathub']['enabled'] = False + PLATFORMS["epic-store"]["enabled"] = False + PLATFORMS["gog"]["enabled"] = False + PLATFORMS["flathub"]["enabled"] = False # drop disabled and invalid platforms FILTERED_PLATFORMS = {} for key, value in PLATFORMS.items(): - if 'enabled' not in value or not value['enabled'] or 'name' not in value or 'cmd' not in value: + if ( + "enabled" not in value + or not value["enabled"] + or "name" not in value + or "cmd" not in value + ): continue FILTERED_PLATFORMS[key] = value -PLATFORMS = dict(sorted(FILTERED_PLATFORMS.items(), key=lambda item: item[1]['name'])) +PLATFORMS = dict(sorted(FILTERED_PLATFORMS.items(), key=lambda item: item[1]["name"])) GAMEDB = { - 'gog' : {}, - 'epic-store' : {}, - 'flathub' : {}, - 'steam' : {}, + "gog": {}, + "epic-store": {}, + "flathub": {}, + "steam": {}, } + @dataclass class GameDbEntry: name: str @@ -284,42 +294,46 @@ class GameDbEntry: @property def status_icon(self): - if self.status == 'verified': - return '🟢' - elif self.status == 'playable': - return '🟡' - elif self.status == 'unsupported': - return '🔴' + if self.status == "verified": + return "🟢" + elif self.status == "playable": + return "🟡" + elif self.status == "unsupported": + return "🔴" else: - return '⚫' + return "⚫" try: - with open(os.path.join(DATA_DIR, 'gamedb.yaml')) as yaml_file: + with open(os.path.join(DATA_DIR, "gamedb.yaml")) as yaml_file: game_list = yaml.load(yaml_file, Loader=yaml.FullLoader) for game in game_list: - if 'id' in game: - game['id'] = str(game['id']) - GAMEDB[game['platform']][game['id']] = GameDbEntry( - name=game['name'], - platform=game['platform'], - id=game['id'], - banner=game['banner'] if 'banner' in game else None, - poster=game['poster'] if 'poster' in game else None, - background=game['background'] if 'background' in game else None, - logo=game['logo'] if 'logo' in game else None, - icon=game['icon'] if 'icon' in game else None, - compat_tool=game['compat_tool'] if 'compat_tool' in game else None, - compat_config=game['compat_config'] if 'compat_config' in game else None, - launch_options=game['launch_options'] if 'launch_options' in game else None, - status=game['status'] if 'status' in game else None, - store=game['store'] if 'store' in game else None, - steam_input=game['steam_input'] if 'steam_input' in game else None, - notes=game['notes'] if 'notes' in game else None, - priority=game['priority'] if 'priority' in game else None, - patch_dir=game['patch_dir'] if 'patch_dir' in game else None, - patches=game['patches'] if 'patches' in game else None, + if "id" in game: + game["id"] = str(game["id"]) + GAMEDB[game["platform"]][game["id"]] = GameDbEntry( + name=game["name"], + platform=game["platform"], + id=game["id"], + banner=game["banner"] if "banner" in game else None, + poster=game["poster"] if "poster" in game else None, + background=game["background"] if "background" in game else None, + logo=game["logo"] if "logo" in game else None, + icon=game["icon"] if "icon" in game else None, + compat_tool=game["compat_tool"] if "compat_tool" in game else None, + compat_config=( + game["compat_config"] if "compat_config" in game else None + ), + launch_options=( + game["launch_options"] if "launch_options" in game else None + ), + status=game["status"] if "status" in game else None, + store=game["store"] if "store" in game else None, + steam_input=game["steam_input"] if "steam_input" in game else None, + notes=game["notes"] if "notes" in game else None, + priority=game["priority"] if "priority" in game else None, + patch_dir=game["patch_dir"] if "patch_dir" in game else None, + patches=game["patches"] if "patches" in game else None, ) except: - print('WARNING: Failed to load game database') + print("WARNING: Failed to load game database") diff --git a/chimera_app/server.py b/chimera_app/server.py index f055d475..2afe21d2 100644 --- a/chimera_app/server.py +++ b/chimera_app/server.py @@ -38,6 +38,7 @@ from chimera_app.config import UPLOADS_DIR from chimera_app.config import SESSION_OPTIONS from chimera_app.config import STORAGE_HANDLER +from chimera_app.config import SETTINGS_DEFAULT from chimera_app.compat_tools import OFFICIAL_COMPAT_TOOLS from chimera_app.compat_tools import OfficialCompatTool from chimera_app.utils import sanitize @@ -55,7 +56,16 @@ import chimera_app.power as power -CONTENT_SHARE_ONLY = os.environ.get('CONTENT_SHARE_ONLY') == 'true' +# Route to display all uploaded files +@route("/uploaded-files") +@authenticate +def show_uploaded_files(): + return template( + "uploaded_files.tpl", tmpfiles=tmpfiles, content_share_only=CONTENT_SHARE_ONLY + ) + + +CONTENT_SHARE_ONLY = os.environ.get("CONTENT_SHARE_ONLY") == "true" server = SessionMiddleware(app(), SESSION_OPTIONS) @@ -79,115 +89,137 @@ def refresh_local_password(): - password = ''.join((secrets.choice(string.ascii_letters + string.digits) for i in range(100))) - f = open('/tmp/chimera-local-password', 'w') - f.write(password) + password = "".join( + (secrets.choice(string.ascii_letters + string.digits) for i in range(100)) + ) + with open("/tmp/chimera-local-password", "w") as f: + f.write(password) + return password + LOCAL_PASSWORD = refresh_local_password() + def authenticate_platform(selected_platform): if selected_platform in PLATFORM_HANDLERS: if not PLATFORM_HANDLERS[selected_platform].is_authenticated(): - redirect(f'/library/{selected_platform}') + redirect(f"/library/{selected_platform}") return False return True -@route('/') +@route("/") @authenticate def root(): - redirect('/library') + redirect("/library") + -@route('/actions') +@route("/actions") @authenticate def actions(): if CONTENT_SHARE_ONLY: - abort(404, 'Running in exclusive content sharing mode') + abort(404, "Running in exclusive content sharing mode") else: - return template('actions.tpl', audio=get_audio(), tdp=power.get_tdp(), bare=True, content_share_only=CONTENT_SHARE_ONLY) + return template( + "actions.tpl", + audio=get_audio(), + tdp=power.get_tdp(), + bare=True, + content_share_only=CONTENT_SHARE_ONLY, + ) + -@route('/emulators') +@route("/emulators") @authenticate def emulators(): - return template('emulators.tpl', bare=True, content_share_only=CONTENT_SHARE_ONLY) + return template("emulators.tpl", bare=True, content_share_only=CONTENT_SHARE_ONLY) -@route('/library') + +@route("/library") @authenticate def platforms(): - return template('platforms.tpl', platforms=PLATFORMS, content_share_only=CONTENT_SHARE_ONLY) + return template( + "platforms.tpl", platforms=PLATFORMS, content_share_only=CONTENT_SHARE_ONLY + ) -@route('/library/') +@route("/library/") @authenticate def platform_page(platform): + if platform in PLATFORM_HANDLERS: if PLATFORM_HANDLERS[platform].is_authenticated(): return template( - 'custom', + "custom", app_list=PLATFORM_HANDLERS[platform].get_installed_content(), showAll=False, isInstalledOverview=True, platform=platform, - platformName=PLATFORMS[platform]['name'], + platformName=PLATFORMS[platform]["name"], remote=False, content_share_only=CONTENT_SHARE_ONLY, ) else: - return template('custom_login', - platform=platform, - platformName=PLATFORMS[platform]['name'], - content_share_only=CONTENT_SHARE_ONLY) + return template( + "custom_login", + platform=platform, + platformName=PLATFORMS[platform]["name"], + content_share_only=CONTENT_SHARE_ONLY, + ) shortcut_file = PlatformShortcutsFile(platform) - shortcuts = sorted(shortcut_file.get_shortcuts_data(), - key=lambda s: s['name']) + shortcuts = sorted(shortcut_file.get_shortcuts_data(), key=lambda s: s["name"]) data = [] for shortcut in shortcuts: filename = None banner = None - hidden = ('hidden' - if 'hidden' in shortcut and shortcut['hidden'] - else '') - if 'banner' in shortcut: - filename = os.path.basename(shortcut['banner']) - banner = f'/images/banner/{platform}/{filename}' - if 'deleted' not in shortcut or shortcut['deleted'] != True: - data.append({'hidden': hidden, - 'filename': filename, - 'banner': banner, - 'name': shortcut['name']}) - - return template('platform.tpl', - shortcuts=data, - platform=platform, - platformName=PLATFORMS[platform]['name'], - remoteConnected=bool(REMOTE_HANDLERS), - content_share_only=CONTENT_SHARE_ONLY) - - -@route('/library//authenticate', method='POST') + hidden = "hidden" if "hidden" in shortcut and shortcut["hidden"] else "" + if "banner" in shortcut: + filename = os.path.basename(shortcut["banner"]) + banner = f"/images/banner/{platform}/{filename}" + if "deleted" not in shortcut or shortcut["deleted"] != True: + data.append( + { + "hidden": hidden, + "filename": filename, + "banner": banner, + "name": shortcut["name"], + } + ) + + return template( + "platform.tpl", + shortcuts=data, + platform=platform, + platformName=PLATFORMS[platform]["name"], + remoteConnected=bool(REMOTE_HANDLERS), + content_share_only=CONTENT_SHARE_ONLY, + ) + + +@route("/library//authenticate", method="POST") @authenticate def platform_authenticate(platform): if platform not in PLATFORM_HANDLERS: return - password = request.forms.get('password') + password = request.forms.get("password") if not password: - redirect(f'/library/{platform}') + redirect(f"/library/{platform}") PLATFORM_HANDLERS[platform].authenticate(password) - redirect(f'/library/{platform}') + redirect(f"/library/{platform}") -@route('/images/banner//') +@route("/images/banner//") @authenticate def banners(platform, filename): - base = f'{BANNER_DIR}/banner/{platform}' - return static_file(filename, root='{base}'.format(base=base)) + base = f"{BANNER_DIR}/banner/{platform}" + return static_file(filename, root="{base}".format(base=base)) -@route('/library//new') +@route("/library//new") @authenticate def new(platform): handler = None @@ -196,7 +228,7 @@ def new(platform): if platform in PLATFORM_HANDLERS: handler = PLATFORM_HANDLERS[platform] showAll = request.query.showAll - elif request.query.remote == 'true': + elif request.query.remote == "true": handler = REMOTE_HANDLERS[platform] showAll = True remote = True @@ -210,45 +242,46 @@ def new(platform): except Exception as err: print(err) if remote: - return '

Remote server was disconnected and cannot be found

' + return "

Remote server was disconnected and cannot be found

" else: - return '

Failed to get list of available content

' + return "

Failed to get list of available content

" return template( - 'custom', + "custom", app_list=app_list, showAll=showAll, isInstalledOverview=False, isNew=True, platform=platform, - platformName=PLATFORMS[platform]['name'], + platformName=PLATFORMS[platform]["name"], remote=remote, content_share_only=CONTENT_SHARE_ONLY, ) - return template('new.tpl', - isNew=True, - isEditing=False, - isEditingArtwork=False, - platform=platform, - platformName=PLATFORMS[platform]['name'], - name='', - content_id='', - hidden='', - steamShortcutID=None, - content_share_only=CONTENT_SHARE_ONLY, - ) - - -@route('/library//edit/') + return template( + "new.tpl", + isNew=True, + isEditing=False, + isEditingArtwork=False, + platform=platform, + platformName=PLATFORMS[platform]["name"], + name="", + content_id="", + hidden="", + steamShortcutID=None, + content_share_only=CONTENT_SHARE_ONLY, + ) + + +@route("/library//edit/") @authenticate def edit(platform, name): - remoteLaunchEnabled = SETTINGS_HANDLER.get_setting('enable_remote_launch') + remoteLaunchEnabled = SETTINGS_HANDLER.get_setting("enable_remote_launch") handler = None remote = False if platform in PLATFORM_HANDLERS: handler = PLATFORM_HANDLERS[platform] - elif request.query.remote == 'true': + elif request.query.remote == "true": handler = REMOTE_HANDLERS[platform] remote = True @@ -261,55 +294,63 @@ def edit(platform, name): shortcut = handler.get_shortcut(content) if content: return template( - 'custom_edit', + "custom_edit", app=content, platform=platform, - platformName=PLATFORMS[platform]['name'], + platformName=PLATFORMS[platform]["name"], name=content_id, - steamShortcutID=(get_bpmbanner_id(shortcut['cmd'], shortcut['name']) if remoteLaunchEnabled else None), + steamShortcutID=( + get_bpmbanner_id(shortcut["cmd"], shortcut["name"]) + if remoteLaunchEnabled + else None + ), remote=remote, content_share_only=CONTENT_SHARE_ONLY, ) else: - abort(404, 'Content not found') + abort(404, "Content not found") shortcuts = PlatformShortcutsFile(platform) shortcut = shortcuts.get_shortcut_match(name) banner = "" - if 'banner' in shortcut: - filename = os.path.basename(shortcut['banner']) - banner = f'/images/banner/{platform}/{filename}' + if "banner" in shortcut: + filename = os.path.basename(shortcut["banner"]) + banner = f"/images/banner/{platform}/{filename}" hidden = False - if not remote and 'hidden' in shortcut: - hidden = shortcut['hidden'] - - return template('new.tpl', - isNew=False, - isEditing=True, - isEditingArtwork=False, - platform=platform, - platformName=PLATFORMS[platform]['name'], - name=name, - content_id=name, - hidden=hidden, - banner=banner, - steamShortcutID=(get_bpmbanner_id(platform, name) if remoteLaunchEnabled else None), - content_share_only=CONTENT_SHARE_ONLY, - ) - -@route('/library//edit_artwork/') + if not remote and "hidden" in shortcut: + hidden = shortcut["hidden"] + + return template( + "new.tpl", + isNew=False, + isEditing=True, + isEditingArtwork=False, + platform=platform, + platformName=PLATFORMS[platform]["name"], + name=name, + content_id=name, + hidden=hidden, + banner=banner, + steamShortcutID=( + get_bpmbanner_id(platform, name) if remoteLaunchEnabled else None + ), + content_share_only=CONTENT_SHARE_ONLY, + ) + + +@route("/library//edit_artwork/") @authenticate def edit_artwork(platform, name): - remoteLaunchEnabled = SETTINGS_HANDLER.get_setting('enable_remote_launch') + remoteLaunchEnabled = SETTINGS_HANDLER.get_setting("enable_remote_launch") handler = None remote = False content_id = name if platform in PLATFORM_HANDLERS: handler = PLATFORM_HANDLERS[platform] - elif request.query.remote == 'true': + elif request.query.remote == "true": handler = REMOTE_HANDLERS[platform] remote = True @@ -319,174 +360,213 @@ def edit_artwork(platform, name): content = handler.get_content(content_id) shortcut = handler.get_shortcut(content) - name = shortcut['name'] + name = shortcut["name"] else: shortcuts = PlatformShortcutsFile(platform) shortcut = shortcuts.get_shortcut_match(name) - return template('new.tpl', - isNew=False, - isEditing=True, - isEditingArtwork=True, - platform=platform, - platformName=PLATFORMS[platform]['name'], - content_id=content_id, - name=name, - hidden=shortcut['hidden'], - steamShortcutID=(get_bpmbanner_id(platform, name) if remoteLaunchEnabled else None), - remote=remote, - content_share_only=CONTENT_SHARE_ONLY, - ) - -@route('/images/flathub/') + return template( + "new.tpl", + isNew=False, + isEditing=True, + isEditingArtwork=True, + platform=platform, + platformName=PLATFORMS[platform]["name"], + content_id=content_id, + name=name, + hidden=shortcut["hidden"], + steamShortcutID=( + get_bpmbanner_id(platform, name) if remoteLaunchEnabled else None + ), + remote=remote, + content_share_only=CONTENT_SHARE_ONLY, + ) + + +@route("/images/flathub/") @authenticate def flathub_images(content_id): - path = PLATFORM_HANDLERS['flathub'].get_image_file_base_dir(content_id) - return static_file(content_id + '.png', root=path) + path = PLATFORM_HANDLERS["flathub"].get_image_file_base_dir(content_id) + return static_file(content_id + ".png", root=path) -@route('/images/') +@route("/images/") def images(filename): - if os.path.isfile(os.path.join(RESOURCE_DIR, 'images', filename)): - return static_file(filename, root=os.path.join(RESOURCE_DIR, 'images')) + if os.path.isfile(os.path.join(RESOURCE_DIR, "images", filename)): + return static_file(filename, root=os.path.join(RESOURCE_DIR, "images")) else: - return static_file(filename, root=os.path.join(BANNER_DIR, 'banner')) + return static_file(filename, root=os.path.join(BANNER_DIR, "banner")) + -@route('/public/') +@route("/public/") def public(filename): - return static_file(filename, root='public') + return static_file(filename, root="public") def get_ext(url): - url_noquery = url.split('?')[0] + url_noquery = url.split("?")[0] ext = os.path.splitext(url_noquery)[1] if not ext: - ext = '.png' + ext = ".png" return ext -@route('/shortcuts/new', method='POST') + +@route("/shortcuts/auto/new", method="POST") @authenticate -def shortcut_create(): - image_urls = {} - image_paths = {} - name = sanitize(request.forms.get('name')) - platform = sanitize(request.forms.get('platform')) - hidden = sanitize(request.forms.get('hidden')) - image_urls['banner'] = request.forms.get('image-url-banner') - image_urls['poster'] = request.forms.get('image-url-poster') - image_urls['background'] = request.forms.get('image-url-background') - image_urls['logo'] = request.forms.get('image-url-logo') - image_urls['icon'] = request.forms.get('image-url-icon') - content = request.forms.get('content') - - if not name or name.strip() == '': - redirect(f'/library/{platform}/new') - return +def auto_create_rom(): + image_urls: dict = {} + game_name = request.json["game_name"] + content = request.json["content"] + platform = request.json["platform"] + + mtch: str = STEAMGRID_HANDLER.find_best_match(game_name) + name = mtch["name"] + + for img_type in ["banner", "poster", "background", "logo", "icon"]: + response_json = json.loads(STEAMGRID_HANDLER.get_images(mtch["id"], img_type))[ + "data" + ] + image_urls[img_type] = response_json[0]["url"] + + process_new_shortcut(name, platform, "off", image_urls, content) + +def process_new_shortcut( + name: str, platform: str, hidden: str, image_urls: dict, content: str +) -> bool: + + image_paths: dict = {} name = name.strip() - shortcuts = PlatformShortcutsFile(platform) + with PlatformShortcutsFile(platform) as shortcuts: - existing_shortcut = shortcuts.get_shortcut_match(name) - is_existing_shortcut_marked_deleted = 'deleted' in existing_shortcut and existing_shortcut['deleted'] == True - if existing_shortcut and not is_existing_shortcut_marked_deleted: - return 'Shortcut already exists' + existing_shortcut = shortcuts.get_shortcut_match(name) + is_existing_shortcut_marked_deleted = ( + "deleted" in existing_shortcut and existing_shortcut["deleted"] == True + ) + if existing_shortcut and not is_existing_shortcut_marked_deleted: + return "Shortcut already exists" + + for img_type in ["banner", "poster", "background", "logo", "icon"]: + if not image_urls[img_type]: + continue + ext = get_ext(image_urls[img_type]) + image_paths[img_type] = os.path.join( + BANNER_DIR, img_type, platform, f"{name}{ext}" + ) + ensure_directory_for_file(image_paths[img_type]) + download = requests.get(image_urls[img_type], timeout=20) + with open(image_paths[img_type], "wb") as image_file: + image_file.write(download.content) + + shortcut = { + "name": name, + "cmd": PLATFORMS[platform]["cmd"], + "hidden": hidden == "on", + "tags": [PLATFORMS[platform]["name"]], + } - for img_type in [ 'banner', 'poster', 'background', 'logo', 'icon' ]: - if not image_urls[img_type]: - continue - ext = get_ext(image_urls[img_type]) - image_paths[img_type] = os.path.join(BANNER_DIR, img_type, platform, f"{name}{ext}") - ensure_directory_for_file(image_paths[img_type]) - download = requests.get(image_urls[img_type], timeout=20) - with open(image_paths[img_type], "wb") as image_file: - image_file.write(download.content) + for img_type in ["banner", "poster", "background", "logo", "icon"]: + if img_type in image_paths: + shortcut[img_type] = image_paths[img_type] - shortcut = { - 'name': name, - 'cmd': PLATFORMS[platform]['cmd'], - 'hidden': hidden == 'on', - 'tags': [PLATFORMS[platform]['name']] - } + if content: + (content_src_path, content_dst_name) = tmpfiles[content] + del tmpfiles[content] + print(content_src_path, CONTENT_DIR, platform, name, content_dst_name) + content_path = upsert_file( + content_src_path, CONTENT_DIR, platform, name, content_dst_name + ) + if content_path: + shortcut["dir"] = '"' + os.path.dirname(content_path) + '"' + shortcut["params"] = '"' + os.path.basename(content_path) + '"' - for img_type in [ 'banner', 'poster', 'background', 'logo', 'icon' ]: - if img_type in image_paths: - shortcut[img_type] = image_paths[img_type] + shortcuts.add_shortcut(shortcut) - if content: - (content_src_path, content_dst_name) = tmpfiles[content] - del tmpfiles[content] - content_path = upsert_file(content_src_path, - CONTENT_DIR, - platform, - name, - content_dst_name) - if content_path: - shortcut['dir'] = '"' + os.path.dirname(content_path) + '"' - shortcut['params'] = '"' + os.path.basename(content_path) + '"' + return True - shortcuts.add_shortcut(shortcut) - shortcuts.save() - redirect(f'/library/{platform}') +@route("/shortcuts/new", method="POST") +@authenticate +def shortcut_create(): + + image_urls = {} + name = sanitize(request.forms.get("name")) + platform = sanitize(request.forms.get("platform")) + hidden = sanitize(request.forms.get("hidden")) + image_urls["banner"] = request.forms.get("image-url-banner") + image_urls["poster"] = request.forms.get("image-url-poster") + image_urls["background"] = request.forms.get("image-url-background") + image_urls["logo"] = request.forms.get("image-url-logo") + image_urls["icon"] = request.forms.get("image-url-icon") + content = request.forms.get("content") + + if not name or name.strip() == "": + redirect(f"/library/{platform}/new") + return + + name = name.strip() + + if process_new_shortcut(name, platform, hidden, image_urls, content): + redirect(f"/library/{platform}") -@route('/shortcuts/edit', method='POST') +@route("/shortcuts/edit", method="POST") @authenticate def shortcut_update(): image_urls = {} image_paths = {} - name = sanitize(request.forms.get('original_name')) # do not allow editing name - platform = sanitize(request.forms.get('platform')) - hidden = sanitize(request.forms.get('hidden')) - image_urls['banner'] = request.forms.get('image-url-banner') - image_urls['poster'] = request.forms.get('image-url-poster') - image_urls['background'] = request.forms.get('image-url-background') - image_urls['logo'] = request.forms.get('image-url-logo') - image_urls['icon'] = request.forms.get('image-url-icon') - content = request.forms.get('content') + name = sanitize(request.forms.get("original_name")) # do not allow editing name + platform = sanitize(request.forms.get("platform")) + hidden = sanitize(request.forms.get("hidden")) + image_urls["banner"] = request.forms.get("image-url-banner") + image_urls["poster"] = request.forms.get("image-url-poster") + image_urls["background"] = request.forms.get("image-url-background") + image_urls["logo"] = request.forms.get("image-url-logo") + image_urls["icon"] = request.forms.get("image-url-icon") + content = request.forms.get("content") shortcuts = PlatformShortcutsFile(platform) shortcut = shortcuts.get_shortcut_match(name) - for img_type in [ 'banner', 'poster', 'background', 'logo', 'icon' ]: + for img_type in ["banner", "poster", "background", "logo", "icon"]: if not image_urls[img_type]: continue ext = get_ext(image_urls[img_type]) - image_paths[img_type] = os.path.join(BANNER_DIR, img_type, platform, f"{name}{ext}") + image_paths[img_type] = os.path.join( + BANNER_DIR, img_type, platform, f"{name}{ext}" + ) ensure_directory_for_file(image_paths[img_type]) download = requests.get(image_urls[img_type], timeout=20) with open(image_paths[img_type], "wb") as image_file: image_file.write(download.content) - shortcut['name'] = name - shortcut['cmd'] = shortcut['cmd'] or platform - shortcut['hidden'] = hidden == 'on' + shortcut["name"] = name + shortcut["cmd"] = shortcut["cmd"] or platform + shortcut["hidden"] = hidden == "on" - for img_type in [ 'banner', 'poster', 'background', 'logo', 'icon' ]: + for img_type in ["banner", "poster", "background", "logo", "icon"]: if img_type in image_paths: shortcut[img_type] = image_paths[img_type] if content: (content_src_path, content_dst_name) = tmpfiles[content] del tmpfiles[content] - content_path = upsert_file(content_src_path, - CONTENT_DIR, - platform, - name, - content_dst_name) + content_path = upsert_file( + content_src_path, CONTENT_DIR, platform, name, content_dst_name + ) if content_path: - shortcut['dir'] = '"' + os.path.dirname(content_path) + '"' - shortcut['params'] = '"' + os.path.basename(content_path) + '"' + shortcut["dir"] = '"' + os.path.dirname(content_path) + '"' + shortcut["params"] = '"' + os.path.basename(content_path) + '"' shortcuts.save() - redirect(f'/library/{platform}') + redirect(f"/library/{platform}") -@route('/shortcuts/delete', method='POST') +@route("/shortcuts/delete", method="POST") @authenticate def shortcut_delete(): name = sanitize(request.forms.name) @@ -497,16 +577,16 @@ def shortcut_delete(): shortcuts.save() delete_file(CONTENT_DIR, platform, name) - delete_file(BANNER_DIR + '/banner', platform, name) + delete_file(BANNER_DIR + "/banner", platform, name) - redirect(f'/library/{platform}') + redirect(f"/library/{platform}") -@route('/shortcuts/file-upload', method='POST') +@route("/shortcuts/file-upload", method="POST") @authenticate def start_file_upload(): file_name = None - file_data = request.files.get('banner') or request.files.get('content') + file_data = request.files.get("banner") or request.files.get("content") if file_data: file_name = sanitize(file_data.filename) @@ -523,32 +603,34 @@ def start_file_upload(): return key -@route('/shortcuts/file-upload', method='PATCH') +@route("/shortcuts/file-upload", method="PATCH") @authenticate def upload_file_chunk(): - key = request.query.get('patch') + key = request.query.get("patch") path = tmpfiles[key][0] + if not path: abort(400) - tmpfiles[key] = (path, sanitize(request.headers.get('Upload-Name'))) + tmpfiles[key] = (path, sanitize(request.headers.get("Upload-Name"))) + + with open(path, "ab") as f: + f.seek(int(request.headers.get("Upload-Offset"))) + f.write(request.body.read()) - f = open(path, 'ab') - f.seek(int(request.headers.get('Upload-Offset'))) - f.write(request.body.read()) - f.close() + return {"path": path} -@route('/shortcuts/file-upload', method='HEAD') +@route("/shortcuts/file-upload", method="HEAD") @authenticate def check_file_upload(): return 0 -@route('/shortcuts/file-upload', method='DELETE') +@route("/shortcuts/file-upload", method="DELETE") @authenticate def delete_file_upload(): - key = request.body.read().decode('utf8') + key = request.body.read().decode("utf8") path = tmpfiles[key][0] if not path: abort(400) @@ -557,20 +639,20 @@ def delete_file_upload(): os.remove(path) -@route('//install/') +@route("//install/") @authenticate def platform_install(platform, content_id): handler = None - redirect_url = f'/library/{platform}/edit/{content_id}' + redirect_url = f"/library/{platform}/edit/{content_id}" if platform in PLATFORM_HANDLERS: handler = PLATFORM_HANDLERS[platform] else: handler = REMOTE_HANDLERS[platform] - redirect_url = f'/library/{platform}/edit/{content_id}?remote=true' + redirect_url = f"/library/{platform}/edit/{content_id}?remote=true" content = handler.get_content(content_id) if not content: - abort(404, 'Content not found') + abort(404, "Content not found") handler.install_content(content) @@ -581,9 +663,8 @@ def platform_install(platform, content_id): handler.download_images(content) - if ('compat_tool' in shortcut - and shortcut['compat_tool'] in OFFICIAL_COMPAT_TOOLS): - name = shortcut['compat_tool'] + if "compat_tool" in shortcut and shortcut["compat_tool"] in OFFICIAL_COMPAT_TOOLS: + name = shortcut["compat_tool"] tool_id = OFFICIAL_COMPAT_TOOLS[name] compat_tool = OfficialCompatTool(name, tool_id) try: @@ -594,12 +675,12 @@ def platform_install(platform, content_id): redirect(redirect_url) -@route('//uninstall/') +@route("//uninstall/") @authenticate def uninstall(platform, content_id): content = PLATFORM_HANDLERS[platform].get_content(content_id) if not content: - abort(404, 'Content not found') + abort(404, "Content not found") PLATFORM_HANDLERS[platform].uninstall_content(content_id) shortcut = PLATFORM_HANDLERS[platform].get_shortcut(content) @@ -608,21 +689,21 @@ def uninstall(platform, content_id): shortcuts.remove_shortcut(content.name) shortcuts.save() - redirect(f'/library/{platform}/edit/{content_id}') + redirect(f"/library/{platform}/edit/{content_id}") -@route('//update/') +@route("//update/") @authenticate def content_update(platform, content_id): content = PLATFORM_HANDLERS[platform].get_content(content_id) if not content: - abort(404, 'Content not found') + abort(404, "Content not found") PLATFORM_HANDLERS[platform].update_content(content_id) - redirect(f'/library/{platform}/edit/{content_id}') + redirect(f"/library/{platform}/edit/{content_id}") -@route('//progress/') +@route("//progress/") @authenticate def install_progress(platform, content_id): handler = None @@ -633,9 +714,9 @@ def install_progress(platform, content_id): content = handler.get_content(content_id) if not content: - abort(404, '{} not found'.format(content_id)) + abort(404, "{} not found".format(content_id)) - response.content_type = 'application/json' + response.content_type = "application/json" values = { "operation": content.operation, "progress": content.progress, @@ -644,19 +725,21 @@ def install_progress(platform, content_id): return json.dumps(values) -@route('/status-info') +@route("/status-info") @authenticate def status_info(): - return template('status_info.tpl', content_share_only=CONTENT_SHARE_ONLY) + return template("status_info.tpl", content_share_only=CONTENT_SHARE_ONLY) -@route('/system') +@route("/system") @authenticate def settings(): current_settings = SETTINGS_HANDLER.get_settings() - password_field = SETTINGS_HANDLER.get_setting('password') + password_field = SETTINGS_HANDLER.get_setting("password") password_is_set = password_field and len(password_field) > 7 - hostname = request.environ.get('HTTP_HOST').split(":")[0] or request.environ.get('SERVER_NAME') + hostname = request.environ.get("HTTP_HOST").split(":")[0] or request.environ.get( + "SERVER_NAME" + ) ssh_key_ids = [] if SSH_KEY_HANDLER: @@ -666,40 +749,63 @@ def settings(): if not CONTENT_SHARE_ONLY: username = pwd.getpwuid(os.getuid())[0] - return template('settings.tpl', settings=current_settings, password_is_set=password_is_set, - ssh_key_ids=ssh_key_ids, hostname=hostname, username=username, content_share_only=CONTENT_SHARE_ONLY) + return template( + "settings.tpl", + settings=current_settings, + password_is_set=password_is_set, + ssh_key_ids=ssh_key_ids, + hostname=hostname, + username=username, + content_share_only=CONTENT_SHARE_ONLY, + ) -@route('/system/update', method='POST') +@route("/system/update", method="POST") @authenticate def settings_update(): - SETTINGS_HANDLER.set_setting("enable_ftp_server", sanitize(request.forms.get('enable_ftp_server')) == 'on') - SETTINGS_HANDLER.set_setting("enable_remote_launch", sanitize(request.forms.get('enable_remote_launch')) == 'on') - SETTINGS_HANDLER.set_setting("enable_content_sharing", sanitize(request.forms.get('enable_content_sharing')) == 'on') - + SETTINGS_HANDLER.set_setting( + "enable_ftp_server", sanitize(request.forms.get("enable_ftp_server")) == "on" + ) + SETTINGS_HANDLER.set_setting( + "enable_remote_launch", + sanitize(request.forms.get("enable_remote_launch")) == "on", + ) + SETTINGS_HANDLER.set_setting( + "enable_content_sharing", + sanitize(request.forms.get("enable_content_sharing")) == "on", + ) + new_api_key = sanitize(request.forms.get("steamgriddb_api_key")) + if new_api_key: + SETTINGS_HANDLER.set_setting("steamgriddb_api_key", new_api_key) + + SETTINGS_HANDLER.set_setting( + "user_steamgriddb_key_set", + (new_api_key != "423ef7be0f4b9f8cfa1a471149c5b72c"), + ) + STEAMGRID_HANDLER.set_api_key(new_api_key) # Make sure the login password is long enough - login_password = sanitize(request.forms.get('login_password')) + login_password = sanitize(request.forms.get("login_password")) if len(login_password) > 7: - password = bcrypt.hashpw(login_password.encode('utf-8'), bcrypt.gensalt()) - SETTINGS_HANDLER.set_setting("password", password.decode('utf-8')) + password = bcrypt.hashpw(login_password.encode("utf-8"), bcrypt.gensalt()) + SETTINGS_HANDLER.set_setting("password", password.decode("utf-8")) # Only allow enabling keep password if a password is set - keep_password = sanitize(request.forms.get('generate_password')) != 'on' - if keep_password and SETTINGS_HANDLER.get_setting('password') or not keep_password: + keep_password = sanitize(request.forms.get("generate_password")) != "on" + if keep_password and SETTINGS_HANDLER.get_setting("password") or not keep_password: SETTINGS_HANDLER.set_setting("keep_password", keep_password) # Make sure the FTP username is not set to empty - ftp_username = sanitize(request.forms.get('ftp_username')) + ftp_username = sanitize(request.forms.get("ftp_username")) if ftp_username: SETTINGS_HANDLER.set_setting("ftp_username", ftp_username) # Make sure the FTP password is long enough - ftp_password = sanitize(request.forms.get('ftp_password')) + ftp_password = sanitize(request.forms.get("ftp_password")) if len(ftp_password) > 7: SETTINGS_HANDLER.set_setting("ftp_password", ftp_password) # port number for FTP server - ftp_port = int(sanitize(request.forms.get('ftp_port'))) + ftp_port = int(sanitize(request.forms.get("ftp_port"))) if ftp_port and 1024 < ftp_port < 65536 and ftp_port != 8844: SETTINGS_HANDLER.set_setting("ftp_port", ftp_port) @@ -707,92 +813,91 @@ def settings_update(): # Delete SSH keys if asked ssh_key_ids = SSH_KEY_HANDLER.get_key_ids() for key_id in ssh_key_ids: - if sanitize(request.forms.get(html.escape(key_id)) == 'on'): + if sanitize(request.forms.get(html.escape(key_id)) == "on"): SSH_KEY_HANDLER.remove_key(key_id) # After we are done deleting the selected ssh keys, add a new key if specified # The add_key function makes sanitization not needed - SSH_KEY_HANDLER.add_key(request.forms.get('ssh_key')) + SSH_KEY_HANDLER.add_key(request.forms.get("ssh_key")) if FTP_SERVER: FTP_SERVER.reload() - redirect('/system') - + redirect("/system") -@route('/actions/steam/restart') +@route("/actions/steam/restart") @authenticate def steam_restart(): try: subprocess.call(["steam", "-shutdown"]) finally: - redirect('/actions') + redirect("/actions") -@route('/actions/mangohud') +@route("/actions/mangohud") @authenticate def mangohud(): try: subprocess.call(["mangohudctl", "toggle", "no_display"]) finally: - redirect('/actions') + redirect("/actions") def retroarch_cmd(msg): sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - sock.sendto(bytes(msg, "utf-8"), ('127.0.0.1', 55355)) + sock.sendto(bytes(msg, "utf-8"), ("127.0.0.1", 55355)) -@route('/actions/retroarch/load_state') +@route("/actions/retroarch/load_state") @authenticate def retro_load_state(): try: - retroarch_cmd('LOAD_STATE') + retroarch_cmd("LOAD_STATE") finally: - redirect('/actions') + redirect("/actions") -@route('/actions/retroarch/save_state') +@route("/actions/retroarch/save_state") @authenticate def retro_save_state(): try: - retroarch_cmd('SAVE_STATE') + retroarch_cmd("SAVE_STATE") finally: - redirect('/actions') + redirect("/actions") -@route('/actions/reboot') +@route("/actions/reboot") @authenticate def reboot_system(): try: - os.system('reboot') + os.system("reboot") finally: - redirect('/actions') + redirect("/actions") -@route('/actions/poweroff') +@route("/actions/poweroff") @authenticate def poweroff_system(): try: - os.system('poweroff') + os.system("poweroff") finally: - redirect('/actions') + redirect("/actions") -@route('/actions/suspend') +@route("/actions/suspend") @authenticate def suspend_system(): try: - os.system('systemctl suspend') + os.system("systemctl suspend") finally: - redirect('/actions') + redirect("/actions") -@route('/system/storage', method='GET') +@route("/system/storage", method="GET") @authenticate def storage_page(): - return template('storage.tpl', content_share_only=CONTENT_SHARE_ONLY) + return template("storage.tpl", content_share_only=CONTENT_SHARE_ONLY) def operation_status(): @@ -802,14 +907,13 @@ def operation_status(): global storage_operation_log return { - 'type' : storage_operation_type, - 'options' : { - 'device' : storage_operation_device - }, - 'status' : storage_operation_status, - 'log' : storage_operation_log + "type": storage_operation_type, + "options": {"device": storage_operation_device}, + "status": storage_operation_status, + "log": storage_operation_log, } + # { # devices : [ # { @@ -830,15 +934,13 @@ def operation_status(): # log : '...' # }, # } -@route('/api/storage', method='GET') +@route("/api/storage", method="GET") @authenticate def storage_display(): devices = STORAGE_HANDLER.get_disks() - response.content_type = 'application/json' - return { - 'devices' : devices, - 'operation' : operation_status() - } + response.content_type = "application/json" + return {"devices": devices, "operation": operation_status()} + # { # operation : 'format', @@ -846,7 +948,7 @@ def storage_display(): # device : '/dev/sda' # } # } -@route('/api/storage', method='POST') +@route("/api/storage", method="POST") @authenticate def storage_operation(): global storage_operation_type @@ -854,42 +956,40 @@ def storage_operation(): global storage_operation_device global storage_operation_log - if storage_operation_status == 'in-progress': + if storage_operation_status == "in-progress": return data = request.json - operation = data['operation'] + operation = data["operation"] - if operation == 'reset': - storage_operation_type = None + if operation == "reset": + storage_operation_type = None storage_operation_status = None storage_operation_device = None - storage_operation_log = None + storage_operation_log = None return - if operation == 'format': + if operation == "format": func = STORAGE_HANDLER.format_disk - elif operation == 'add': + elif operation == "add": func = STORAGE_HANDLER.add_disk else: return storage_operation_type = operation - device = data['options']['device'] - thread = threading.Thread(target=storage_task, - args=[device, func]) + device = data["options"]["device"] + thread = threading.Thread(target=storage_task, args=[device, func]) - storage_operation_status = 'in-progress' + storage_operation_status = "in-progress" storage_operation_device = device - storage_operation_log = None + storage_operation_log = None thread.start() - response.content_type = 'application/json' - return { - 'operation' : operation_status() - } + response.content_type = "application/json" + return {"operation": operation_status()} + def storage_task(device, func): global storage_operation_status @@ -899,56 +999,59 @@ def storage_task(device, func): proc = func(device) if proc.returncode == 0: storage_operation_log = proc.stdout - storage_operation_status = 'success' + storage_operation_status = "success" else: storage_operation_log = proc.stderr - storage_operation_status = 'fail' + storage_operation_status = "fail" def get_audio(): - if not shutil.which('wpctl'): + if not shutil.which("wpctl"): return None try: - volume_raw = subprocess.check_output(["wpctl", "get-volume", "@DEFAULT_AUDIO_SINK@"]).decode('utf8') - volume = int(float(volume_raw.split(':')[1].split('[')[0].strip()) * 100) + volume_raw = subprocess.check_output( + ["wpctl", "get-volume", "@DEFAULT_AUDIO_SINK@"] + ).decode("utf8") + volume = int(float(volume_raw.split(":")[1].split("[")[0].strip()) * 100) - return { - 'volume': volume, - 'muted': '[MUTED]' in volume_raw - } + return {"volume": volume, "muted": "[MUTED]" in volume_raw} except: return None -@route('/actions/audio/toggle_mute') +@route("/actions/audio/toggle_mute") @authenticate def toggle_mute(): try: subprocess.call(["wpctl", "set-mute", "@DEFAULT_AUDIO_SINK@", "toggle"]) finally: - redirect('/actions') + redirect("/actions") -@route('/actions/audio/volume_up') +@route("/actions/audio/volume_up") @authenticate def volume_up(): try: - subprocess.call(["wpctl", "set-volume", "--limit", "1.0", "@DEFAULT_AUDIO_SINK@", "10%+"]) + subprocess.call( + ["wpctl", "set-volume", "--limit", "1.0", "@DEFAULT_AUDIO_SINK@", "10%+"] + ) finally: - redirect('/actions') + redirect("/actions") -@route('/actions/audio/volume_down') +@route("/actions/audio/volume_down") @authenticate def volume_down(): try: - subprocess.call(["wpctl", "set-volume", "--limit", "1.0", "@DEFAULT_AUDIO_SINK@", "10%-"]) + subprocess.call( + ["wpctl", "set-volume", "--limit", "1.0", "@DEFAULT_AUDIO_SINK@", "10%-"] + ) finally: - redirect('/actions') + redirect("/actions") -@route('/actions/power/tdp_down') +@route("/actions/power/tdp_down") @authenticate def tdp_down(): try: @@ -956,9 +1059,10 @@ def tdp_down(): if tdp: power.set_tdp(tdp - 1) finally: - redirect('/actions') + redirect("/actions") -@route('/actions/power/tdp_up') + +@route("/actions/power/tdp_up") @authenticate def tdp_up(): try: @@ -966,92 +1070,126 @@ def tdp_up(): if tdp: power.set_tdp(tdp + 1) finally: - redirect('/actions') + redirect("/actions") + -@route('/login') +@route("/login") def login(): - keep_password = SETTINGS_HANDLER.get_setting('keep_password') + keep_password = SETTINGS_HANDLER.get_setting("keep_password") if not keep_password and not CONTENT_SHARE_ONLY: AUTHENTICATOR.reset_password() AUTHENTICATOR.launch() - return template('login', keep_password=keep_password or CONTENT_SHARE_ONLY, failed=False, content_share_only=CONTENT_SHARE_ONLY) + return template( + "login", + keep_password=keep_password or CONTENT_SHARE_ONLY, + failed=False, + content_share_only=CONTENT_SHARE_ONLY, + ) -@route('/logout') +@route("/logout") def logout(): - session = request.environ.get('beaker.session') + session = request.environ.get("beaker.session") session.delete() - return template('logout', content_share_only=CONTENT_SHARE_ONLY) + return template("logout", content_share_only=CONTENT_SHARE_ONLY) -@route('/authenticate', method='GET') + +@route("/authenticate", method="GET") def authenticate_get(): return authenticate_route_handler() -@route('/authenticate', method='POST') + +@route("/authenticate", method="POST") def authenticate_post(): return authenticate_route_handler() + def authenticate_route_handler(): global LOCAL_PASSWORD if not CONTENT_SHARE_ONLY: AUTHENTICATOR.kill() - password = request.forms.get('password') or request.query.get('password') - session = request.environ.get('beaker.session') - keep_password = SETTINGS_HANDLER.get_setting('keep_password') or False - stored_hash = SETTINGS_HANDLER.get_setting('password') + password = request.forms.get("password") or request.query.get("password") + session = request.environ.get("beaker.session") + keep_password = SETTINGS_HANDLER.get_setting("keep_password") or False + stored_hash = SETTINGS_HANDLER.get_setting("password") local_password = LOCAL_PASSWORD - LOCAL_PASSWORD=refresh_local_password() - DEFAULT_PASSWORD='gamer' - if password == local_password or \ - AUTHENTICATOR.matches_password(password.upper()) or \ - (CONTENT_SHARE_ONLY and not stored_hash and password == DEFAULT_PASSWORD) or \ - (keep_password and stored_hash and bcrypt.checkpw(password.encode('utf-8'), stored_hash.encode('utf-8'))): - session['User-Agent'] = request.headers.get('User-Agent') - session['Logged-In'] = True + LOCAL_PASSWORD = refresh_local_password() + DEFAULT_PASSWORD = "gamer" + if ( + password == local_password + or AUTHENTICATOR.matches_password(password.upper()) + or (CONTENT_SHARE_ONLY and not stored_hash and password == DEFAULT_PASSWORD) + or ( + keep_password + and stored_hash + and bcrypt.checkpw(password.encode("utf-8"), stored_hash.encode("utf-8")) + ) + ): + session["User-Agent"] = request.headers.get("User-Agent") + session["Logged-In"] = True session.save() - redirect('/') + redirect("/") else: - if session.get('Logged-In', True): - session['Logged-In'] = False + if session.get("Logged-In", True): + session["Logged-In"] = False session.save() if not keep_password and not CONTENT_SHARE_ONLY: AUTHENTICATOR.reset_password() AUTHENTICATOR.launch() - return template('login', keep_password=keep_password or CONTENT_SHARE_ONLY, failed=True, content_share_only=CONTENT_SHARE_ONLY) + return template( + "login", + keep_password=keep_password or CONTENT_SHARE_ONLY, + failed=True, + content_share_only=CONTENT_SHARE_ONLY, + ) -@route('/forgotpassword') +@route("/forgotpassword") def forgot_password(): - SETTINGS_HANDLER.set_setting('keep_password', False) - redirect('/login') + SETTINGS_HANDLER.set_setting("keep_password", False) + redirect("/login") -@route('/steamgrid/search/') +@route("/steamgrid/search/") def steamgrid_search(search_string): return STEAMGRID_HANDLER.search_games(search_string) -@route('/steamgrid/images/') +@route("/steamgrid/findbestmatch/") +def steamgrid_findbestmatch(search_string) -> dict: + """Find the best match (Based on levenstein edit distance) for a game based on the based on any string + + Args: + search_string (_type_): file_name + + Returns: + _type_: dict + """ + + return STEAMGRID_HANDLER.find_best_match(search_string) + + +@route("/steamgrid/images/") def steamgrid_get_images(game_id): return STEAMGRID_HANDLER.get_images(game_id, request.query.type) -@route('/launch/') +@route("/launch/") def launch_game(id): - enabled = SETTINGS_HANDLER.get_setting('enable_remote_launch') + enabled = SETTINGS_HANDLER.get_setting("enable_remote_launch") if not enabled or not id.isnumeric() or type(id) != str: - redirect('/') + redirect("/") return subprocess.call(["steam", "steam://rungameid/{}".format(id)]) - return 'Launched {}...'.format(id) - + return "Launched {}...".format(id) ########## Content sharing feature + def find_remote_chimera(): import socket @@ -1061,12 +1199,13 @@ def find_remote_chimera(): client.bind(("", 48844)) while True: data, addr = client.recvfrom(1024) - if data != b'chimera service v1': + if data != b"chimera service v1": continue for platform in PLATFORMS: REMOTE_HANDLERS[platform] = ChimeraRemote(platform, addr[0]) break + def broadcast_service(): import socket import time @@ -1077,12 +1216,12 @@ def broadcast_service(): server.settimeout(0.2) message = b"chimera service v1" while True: - server.sendto(message, ('', 48844)) + server.sendto(message, ("", 48844)) time.sleep(10) -if 'pytest' not in sys.modules: # the threads interfere with tests - contentSharingEnabled = SETTINGS_HANDLER.get_setting('enable_content_sharing') +if "pytest" not in sys.modules: # the threads interfere with tests + contentSharingEnabled = SETTINGS_HANDLER.get_setting("enable_content_sharing") if contentSharingEnabled: broadcast_thread = threading.Thread(target=broadcast_service) broadcast_thread.start() @@ -1090,11 +1229,12 @@ def broadcast_service(): scan_thread = threading.Thread(target=find_remote_chimera) scan_thread.start() -@route('/share/images///') + +@route("/share/images///") def download_images(image_type, platform, filename): if not contentSharingEnabled: abort(404) - if image_type not in [ 'banner', 'poster', 'background', 'logo', 'icon' ]: + if image_type not in ["banner", "poster", "background", "logo", "icon"]: abort(404) if platform not in PLATFORMS: abort(404) @@ -1102,53 +1242,130 @@ def download_images(image_type, platform, filename): return static_file(unquote(filename), root) -@route('/share/content//') +@route("/share/content//") def download_content(platform, filename): if not contentSharingEnabled: abort(404) if platform not in PLATFORMS: abort(404) root = os.path.join(CONTENT_DIR, platform) - if is_direct(platform, 'content'): - root = os.path.join(root, f'.{platform}') + if is_direct(platform, "content"): + root = os.path.join(root, f".{platform}") return static_file(unquote(filename), root) -@route('/share/platforms/', method='GET') +@route("/share/platforms/", method="GET") def api_get_platform_content(platform): if not contentSharingEnabled: abort(404) shortcut_file = PlatformShortcutsFile(platform) - shortcuts = sorted(shortcut_file.get_shortcuts_data(), - key=lambda s: s['name']) + shortcuts = sorted(shortcut_file.get_shortcuts_data(), key=lambda s: s["name"]) data = [] for shortcut in shortcuts: - if 'hidden' in shortcut and shortcut['hidden']: + if "hidden" in shortcut and shortcut["hidden"]: continue - if 'deleted' in shortcut and shortcut['deleted']: + if "deleted" in shortcut and shortcut["deleted"]: continue - if not 'params' in shortcut or not shortcut['params']: + if not "params" in shortcut or not shortcut["params"]: continue - content_pretty_filename = os.path.basename(shortcut['params'].strip('"')) + content_pretty_filename = os.path.basename(shortcut["params"].strip('"')) content_original_filename = content_pretty_filename content_path = os.path.join(CONTENT_DIR, platform, content_pretty_filename) if os.path.islink(content_path): content_original_filename = os.path.basename(os.path.realpath(content_path)) entry = { - 'name': shortcut['name'], - 'content_filename': content_original_filename, - 'content_download_url': f'/share/content/{platform}/{quote(content_pretty_filename)}' + "name": shortcut["name"], + "content_filename": content_original_filename, + "content_download_url": f"/share/content/{platform}/{quote(content_pretty_filename)}", } - for image_type in [ 'banner', 'poster', 'background', 'logo', 'icon' ]: + for image_type in ["banner", "poster", "background", "logo", "icon"]: if image_type in shortcut: filename = os.path.basename(shortcut[image_type]) - entry[image_type] = f'/share/images/{image_type}/{platform}/{quote(filename)}' + entry[image_type] = ( + f"/share/images/{image_type}/{platform}/{quote(filename)}" + ) data.append(entry) - response.content_type = 'application/json' + response.content_type = "application/json" return json.dumps(data) + + +@route("/library//bulk-upload") +@authenticate +def new(platform): + handler = None + showAll = False + remote = False + if platform in PLATFORM_HANDLERS: + handler = PLATFORM_HANDLERS[platform] + showAll = request.query.showAll + elif request.query.remote == "true": + handler = REMOTE_HANDLERS[platform] + showAll = True + remote = True + + if handler: + if not authenticate_platform(platform): + return + + try: + app_list = handler.get_available_content(showAll) + except Exception as err: + print(err) + if remote: + return "

Remote server was disconnected and cannot be found

" + else: + return "

Failed to get list of available content

" + + return template( + "custom", + app_list=app_list, + showAll=showAll, + isInstalledOverview=False, + isNew=True, + platform=platform, + platformName=PLATFORMS[platform]["name"], + remote=remote, + content_share_only=CONTENT_SHARE_ONLY, + ) + uploaded_files = [ + {"key": key, "filename": filename, "path": path} + for key, (path, filename) in tmpfiles.items() + if os.path.exists(path) + ] + return template( + "bulk_upload.tpl", + isNew=True, + isEditing=False, + isEditingArtwork=False, + platform=platform, + platformName=PLATFORMS[platform]["name"], + name="", + content_id="", + hidden="", + steamShortcutID=None, + content_share_only=CONTENT_SHARE_ONLY, + uploaded_files=uploaded_files, + user_steamgriddb_key_set=SETTINGS_HANDLER.get_setting( + "user_steamgriddb_key_set" + ), + ) + + +@route("/api/uploaded-roms", METHOD="GET") +@authenticate +def api_uploaded_roms(): + """ + Returns a list of uploaded ROMs for the current session. + """ + uploaded_files = [] + for key, (path, filename) in tmpfiles.items(): + if os.path.exists(path): + uploaded_files.append({"key": key, "filename": filename, "path": path}) + response.content_type = "application/json" + return json.dumps(uploaded_files) diff --git a/chimera_app/shortcuts.py b/chimera_app/shortcuts.py index be1b1a91..3b33b037 100644 --- a/chimera_app/shortcuts.py +++ b/chimera_app/shortcuts.py @@ -13,6 +13,8 @@ from chimera_app.config import GameDbEntry from chimera_app.steam_collections import SteamCollections from chimera_app.file_utils import ensure_directory_for_file +import fcntl +import threading STATUS_TAGS = [ 'ChimeraOS Verified', 'ChimeraOS Playable', 'ChimeraOS Unsupported' ] @@ -186,7 +188,7 @@ def tag(self, tag, appid): def add_shortcut(self, entry: dict): """Creates a new shortcut with given dictionary. Will try to match - with an existing shortcut in this file. If no existing shortcut can + with an existing shortcut in this file. If no existing shortcut can` be found, then create a new entry. """ if 'name' not in entry: @@ -295,13 +297,32 @@ class ShortcutsFile(): path: str shortcuts_data: List[dict] + file: None + thread_lock = threading.Lock() def __init__(self, path: str, auto_load: bool = True): + self.path = path self.shortcuts_data = [] if auto_load: self.load_data() + def __enter__(self): + self.thread_lock.acquire() + self.file = open(self.path, 'a+') + fcntl.flock(self.file.fileno(), fcntl.LOCK_EX) + self.load_data(file_handle=self.file) + return self + + def __exit__(self, exc_type, exc_val, exc_tab): + try: + self.save(file_handle=self.file) + finally: + fcntl.flock(self.file.fileno(), fcntl.LOCK_UN) + self.file.close() + self.file = None + self.thread_lock.release() + def exists(self) -> bool: """Returns true if this file exists. False otherwise""" return os.path.exists(self.path) @@ -310,17 +331,23 @@ def get_shortcuts_data(self) -> List[dict]: """Returns this file shortcuts in a list of dictionaries""" return self.shortcuts_data - def load_data(self) -> None: + def load_data(self, file_handle=None) -> None: + """Load shortcuts from this file""" if not self.exists(): self.shortcuts_data = [] return - with open(self.path) as yaml_file: - data = yaml.load(yaml_file, Loader=yaml.FullLoader) - if isinstance(data, dict): - data = [data] - self.shortcuts_data = data + if file_handle: + file_handle.seek(0) + data = yaml.load(file_handle, Loader=yaml.FullLoader) + else: + with open(self.path) as yaml_file: + data = yaml.load(yaml_file, Loader=yaml.FullLoader) + + if isinstance(data, dict): + data = [data] + self.shortcuts_data = data or [] def add_shortcut(self, shortcut: dict) -> None: """Add a shortcut to the end of the shortcuts data list""" @@ -359,7 +386,7 @@ def prune_deleted(self) -> None: if 'deleted' in shortcut and shortcut['deleted'] == True: self.shortcuts_data.remove(shortcut) - def save(self) -> None: + def save(self, file_handle=None) -> None: """Save this file with current shortcuts data""" ensure_directory_for_file(self.path) with open(self.path, 'w') as file: diff --git a/chimera_app/steamgrid/steamgrid.py b/chimera_app/steamgrid/steamgrid.py index 2a4c027c..bdbf1f05 100644 --- a/chimera_app/steamgrid/steamgrid.py +++ b/chimera_app/steamgrid/steamgrid.py @@ -1,5 +1,8 @@ import requests import html +import difflib +import json +import re class Steamgrid: @@ -7,21 +10,30 @@ class Steamgrid: def __init__(self, api_key): self.__api_key = api_key + def set_api_key(self, api_key): + self.__api_key = api_key + def search_games(self, search_string): - url = "https://www.steamgriddb.com/api/v2/search/autocomplete/{}".format(html.escape(search_string)) + url = "https://www.steamgriddb.com/api/v2/search/autocomplete/{}".format( + html.escape(search_string) + ) response = self.__request(url) return response.text def get_images(self, game_id, imgtype=None): - if imgtype == 'banner': - url = "https://www.steamgriddb.com/api/v2/grids/game/{}?dimensions=460x215,920x430".format(game_id) - elif imgtype == 'poster': - url = "https://www.steamgriddb.com/api/v2/grids/game/{}?dimensions=600x900".format(game_id) - elif imgtype == 'background': + if imgtype == "banner": + url = "https://www.steamgriddb.com/api/v2/grids/game/{}?dimensions=460x215,920x430".format( + game_id + ) + elif imgtype == "poster": + url = "https://www.steamgriddb.com/api/v2/grids/game/{}?dimensions=600x900".format( + game_id + ) + elif imgtype == "background": url = "https://www.steamgriddb.com/api/v2/heroes/game/{}".format(game_id) - elif imgtype == 'logo': + elif imgtype == "logo": url = "https://www.steamgriddb.com/api/v2/logos/game/{}".format(game_id) - elif imgtype == 'icon': + elif imgtype == "icon": url = "https://www.steamgriddb.com/api/v2/icons/game/{}".format(game_id) else: return @@ -30,7 +42,29 @@ def get_images(self, game_id, imgtype=None): return response.text def __request(self, url): - headers = { - 'Authorization': "Bearer {}".format(self.__api_key) - } + headers = {"Authorization": "Bearer {}".format(self.__api_key)} return requests.get(url, headers=headers, timeout=20) + + def find_best_match(self, game_name: str) -> dict: + # strip non alpha numerica + game_name = re.sub(r"[^a-zA-Z0-9 ]", "", game_name) + + game_name_tokens = sorted(game_name.lower().split()) + steamgridlist: dict = json.loads(self.search_games(game_name))["data"] + best_match: dict = steamgridlist[0] + best_similarity: float = 0.0 + + for item in steamgridlist: + item_name = re.sub(r"[^a-zA-Z0-9 ]", "", item["name"]) + steam_response_tokens = sorted(item_name.lower().split()) + + similarity = difflib.SequenceMatcher( + None, game_name_tokens, steam_response_tokens + ).ratio() + + if similarity > best_similarity: + best_similarity = similarity + best_match = item + item["similarity_score"] = float(similarity) + + return best_match if best_match else None diff --git a/images/bulk-upload.png b/images/bulk-upload.png new file mode 100644 index 00000000..cd0a31a9 Binary files /dev/null and b/images/bulk-upload.png differ diff --git a/views/bulk_upload.tpl b/views/bulk_upload.tpl new file mode 100644 index 00000000..060ec79f --- /dev/null +++ b/views/bulk_upload.tpl @@ -0,0 +1,66 @@ +% rebase('base.tpl', content_share_only=get('content_share_only')) + + + + + +
+

Bulk Upload

+
+ +
+
+

Or click to select files for bulk upload.

+ +
+ + Back to Library + + +
+ + diff --git a/views/new.tpl b/views/new.tpl index 6e18e759..5203c2fe 100644 --- a/views/new.tpl +++ b/views/new.tpl @@ -6,6 +6,7 @@ const IMAGE_TYPES = ['banner', 'poster', 'background', 'logo', 'icon']; + function capitalize(word) { return word.charAt(0).toUpperCase() + word.slice(1); } @@ -161,7 +162,9 @@ function images() {
- +
File Name
+

{{name}}

+ % if isEditing and not isEditingArtwork:
Artwork
@@ -175,8 +178,11 @@ function images() {
Hidden
+ + +
Content
- + % end % if isNew or isEditingArtwork: @@ -208,6 +214,12 @@ FilePond.setOptions({ chunkUploads : true, chunkSize : 1000000 // 1 MB }); + +const pond =FilePond.find(document.querySelector('.filepond')); +pond.on('processfile', (error, file) => { + console.log('event triggered', file) + +}) % if steamShortcutID and not isEditingArtwork: diff --git a/views/platform.tpl b/views/platform.tpl index 1b5bbc60..9c8cc6ca 100644 --- a/views/platform.tpl +++ b/views/platform.tpl @@ -1,10 +1,20 @@ % rebase('base.tpl', content_share_only=get('content_share_only')) +
Add new shortcut
+ +
+ + Add new shortcut + +
+ + + % if remoteConnected:
@@ -13,6 +23,8 @@
% end + + % from urllib.parse import quote % for s in shortcuts: @@ -24,4 +36,5 @@ % end + % end diff --git a/views/settings.tpl b/views/settings.tpl index 43cf1421..e1a91c6f 100644 --- a/views/settings.tpl +++ b/views/settings.tpl @@ -69,6 +69,15 @@ % end +

SteamgridDB Api Token

+
+ SteamgridDB is a an open source 3rd party plugin for browsing and managing library art for non steam games. + In order to utilize the bulk upload functionality you must retreive an api key. + A key can be retreived easily by visiting SteamGridDb and clicking "Login via Steam". Then clicking "Create Api Key". + The generated key can then be inputted below. +
Api Token
+ +

Other

@@ -80,7 +89,6 @@ Allows other Chimera instances on the same network to download content from this instance. Only a single instance on your network should have this setting enabled. The server must be restarted for changes to this setting to take effect.

- diff --git a/views/uploaded_files.tpl b/views/uploaded_files.tpl new file mode 100644 index 00000000..15d7f8ae --- /dev/null +++ b/views/uploaded_files.tpl @@ -0,0 +1,12 @@ +% rebase('base.tpl', content_share_only=get('content_share_only')) + +

Uploaded Files

+
    +% for key, (path, filename) in tmpfiles.items(): +
  • + {{filename}} (Temp name: {{key}}) +
    + Path: {{path}} +
    • +% end +