diff options
Diffstat (limited to 'backend/browser.py')
| -rw-r--r-- | backend/browser.py | 120 |
1 files changed, 84 insertions, 36 deletions
diff --git a/backend/browser.py b/backend/browser.py index dbdf2d5f..0a365e52 100644 --- a/backend/browser.py +++ b/backend/browser.py @@ -1,27 +1,33 @@ # Full imports import json + # import pprint # from pprint import pformat # Partial imports -from aiohttp import ClientSession, web -from asyncio import get_event_loop, sleep -from concurrent.futures import ProcessPoolExecutor +from aiohttp import ClientSession +from asyncio import sleep from hashlib import sha256 from io import BytesIO from logging import getLogger -from os import R_OK, W_OK, path, rename, listdir, access, mkdir +from os import R_OK, W_OK, path, listdir, access, mkdir from shutil import rmtree from subprocess import call from time import time from zipfile import ZipFile # Local modules -from helpers import get_ssl_context, get_user, get_user_group, download_remote_binary_to_path +from helpers import ( + get_ssl_context, + get_user, + get_user_group, + download_remote_binary_to_path, +) from injector import get_gamepadui_tab logger = getLogger("Browser") + class PluginInstallContext: def __init__(self, artifact, name, version, hash) -> None: self.artifact = artifact @@ -29,6 +35,7 @@ class PluginInstallContext: self.version = version self.hash = hash + class PluginBrowser: def __init__(self, plugin_path, plugins, loader) -> None: self.plugin_path = plugin_path @@ -43,32 +50,40 @@ class PluginBrowser: zip_file = ZipFile(zip) zip_file.extractall(self.plugin_path) plugin_dir = self.find_plugin_folder(name) - code_chown = call(["chown", "-R", get_user()+":"+get_user_group(), plugin_dir]) + code_chown = call( + ["chown", "-R", get_user() + ":" + get_user_group(), plugin_dir] + ) code_chmod = call(["chmod", "-R", "555", plugin_dir]) if code_chown != 0 or code_chmod != 0: - logger.error(f"chown/chmod exited with a non-zero exit code (chown: {code_chown}, chmod: {code_chmod})") + logger.error( + f"chown/chmod exited with a non-zero exit code (chown: {code_chown}," + f" chmod: {code_chmod})" + ) return False return True - + async def _download_remote_binaries_for_plugin_with_name(self, pluginBasePath): rv = False try: - packageJsonPath = path.join(pluginBasePath, 'package.json') - pluginBinPath = path.join(pluginBasePath, 'bin') + packageJsonPath = path.join(pluginBasePath, "package.json") + pluginBinPath = path.join(pluginBasePath, "bin") if access(packageJsonPath, R_OK): with open(packageJsonPath, "r", encoding="utf-8") as f: packageJson = json.load(f) - if "remote_binary" in packageJson and len(packageJson["remote_binary"]) > 0: + if ( + "remote_binary" in packageJson + and len(packageJson["remote_binary"]) > 0 + ): # create bin directory if needed. - rc=call(["chmod", "-R", "777", pluginBasePath]) + call(["chmod", "-R", "777", pluginBasePath]) if access(pluginBasePath, W_OK): - + if not path.exists(pluginBinPath): mkdir(pluginBinPath) - + if not access(pluginBinPath, W_OK): - rc=call(["chmod", "-R", "777", pluginBinPath]) + call(["chmod", "-R", "777", pluginBinPath]) rv = True for remoteBinary in packageJson["remote_binary"]: @@ -76,16 +91,29 @@ class PluginBrowser: binName = remoteBinary["name"] binURL = remoteBinary["url"] binHash = remoteBinary["sha256hash"] - if not await download_remote_binary_to_path(binURL, binHash, path.join(pluginBinPath, binName)): + if not await download_remote_binary_to_path( + binURL, binHash, path.join(pluginBinPath, binName) + ): rv = False - raise Exception(f"Error Downloading Remote Binary {binName}@{binURL} with hash {binHash} to {path.join(pluginBinPath, binName)}") - - code_chown = call(["chown", "-R", get_user()+":"+get_user_group(), self.plugin_path]) - rc=call(["chmod", "-R", "555", pluginBasePath]) + raise Exception( + "Error Downloading Remote Binary" + f" {binName}@{binURL} with hash {binHash} to" + f" {path.join(pluginBinPath, binName)}" + ) + + call( + [ + "chown", + "-R", + get_user() + ":" + get_user_group(), + self.plugin_path, + ] + ) + call(["chmod", "-R", "555", pluginBasePath]) else: rv = True - logger.debug(f"No Remote Binaries to Download") - + logger.debug("No Remote Binaries to Download") + except Exception as e: rv = False logger.debug(str(e)) @@ -95,12 +123,16 @@ class PluginBrowser: def find_plugin_folder(self, name): for folder in listdir(self.plugin_path): try: - with open(path.join(self.plugin_path, folder, 'plugin.json'), "r", encoding="utf-8") as f: + with open( + path.join(self.plugin_path, folder, "plugin.json"), + "r", + encoding="utf-8", + ) as f: plugin = json.load(f) - if plugin['name'] == name: + if plugin["name"] == name: return str(path.join(self.plugin_path, folder)) - except: + except Exception: logger.debug(f"skipping {folder}") async def uninstall_plugin(self, name): @@ -127,8 +159,10 @@ class PluginBrowser: except FileNotFoundError: logger.warning(f"Plugin {name} not installed, skipping uninstallation") except Exception as e: - logger.error(f"Plugin {name} in {self.find_plugin_folder(name)} was not uninstalled") - logger.error(f"Error at %s", exc_info=e) + logger.error( + f"Plugin {name} in {self.find_plugin_folder(name)} was not uninstalled" + ) + logger.error("Error at %s", exc_info=e) if self.loader.watcher: self.loader.watcher.disabled = False @@ -140,8 +174,11 @@ class PluginBrowser: pluginFolderPath = self.find_plugin_folder(name) if pluginFolderPath: isInstalled = True - except: - logger.error(f"Failed to determine if {name} is already installed, continuing anyway.") + except Exception: + logger.error( + f"Failed to determine if {name} is already installed, continuing" + " anyway." + ) logger.info(f"Installing {name} (Version: {version})") async with ClientSession() as client: logger.debug(f"Fetching {artifact}") @@ -155,22 +192,26 @@ class PluginBrowser: try: logger.debug("Uninstalling existing plugin...") await self.uninstall_plugin(name) - except: + except Exception: logger.error(f"Plugin {name} could not be uninstalled.") logger.debug("Unzipping...") ret = self._unzip_to_plugin_dir(res_zip, name, hash) if ret: plugin_dir = self.find_plugin_folder(name) - ret = await self._download_remote_binaries_for_plugin_with_name(plugin_dir) + ret = await self._download_remote_binaries_for_plugin_with_name( + plugin_dir + ) if ret: logger.info(f"Installed {name} (Version: {version})") if name in self.loader.plugins: self.loader.plugins[name].stop() self.loader.plugins.pop(name, None) await sleep(1) - self.loader.import_plugin(path.join(plugin_dir, "main.py"), plugin_dir) + self.loader.import_plugin( + path.join(plugin_dir, "main.py"), plugin_dir + ) else: - logger.fatal(f"Failed Downloading Remote Binaries") + logger.fatal("Failed Downloading Remote Binaries") else: self.log.fatal(f"SHA-256 Mismatch!!!! {name} (Version: {version})") if self.loader.watcher: @@ -180,14 +221,21 @@ class PluginBrowser: async def request_plugin_install(self, artifact, name, version, hash): request_id = str(time()) - self.install_requests[request_id] = PluginInstallContext(artifact, name, version, hash) + self.install_requests[request_id] = PluginInstallContext( + artifact, name, version, hash + ) tab = await get_gamepadui_tab() await tab.open_websocket() - await tab.evaluate_js(f"DeckyPluginLoader.addPluginInstallPrompt('{name}', '{version}', '{request_id}', '{hash}')") + await tab.evaluate_js( + f"DeckyPluginLoader.addPluginInstallPrompt('{name}', '{version}'," + f" '{request_id}', '{hash}')" + ) async def confirm_plugin_install(self, request_id): request = self.install_requests.pop(request_id) - await self._install(request.artifact, request.name, request.version, request.hash) + await self._install( + request.artifact, request.name, request.version, request.hash + ) def cancel_plugin_install(self, request_id): self.install_requests.pop(request_id) |