add new system base in python for services

services/services.py

@@ -0,0 +1,168 @@
+import logging
+import os
+import subprocess
+
+import utils
+from main import PATH, PYTHON_EXEC, logger, LEVEL
+
+
+class Stack:
+    def __init__(self, name: str, path: str, port: int, url: str):
+        self.name = name
+        self.path = os.path.join(PATH, path)
+        self.url = url
+        self.port = port
+
+        self.process = None
+
+        if self.is_installed():
+            self.update()
+        else:
+            self.check_for_broken_install()
+            self.create_venv()
+            self.install()
+
+    def install(self):
+        self.create_file('.installed', 'true')
+        logger.info(f"Installed {self.name}")
+
+    def is_installed(self):
+        return self.file_exists('.installed')
+
+    def check_for_broken_install(self):
+        if not self.is_installed() and len(os.listdir(self.path)) > 0:
+            logger.warning("Found files from a previous/borked/crashed installation, cleaning up...")
+            self.bash(f"rm -rf {self.path}")
+            self.create_dir('')
+
+    def update(self, folder: str = 'webui'):
+        if self.dir_exists(folder):
+            logger.info(f"Updating {self.name}")
+            self.git_pull(folder)
+        else:
+            logger.warning(f"Could not update {self.name} as {folder} does not exist")
+
+    def uninstall(self):
+        self.bash(f"rm -rf {self.path}")
+
+    def start(self):
+        pass
+
+    def stop(self):
+        pass
+
+    def restart(self):
+        self.stop()
+        self.start()
+
+    def status(self):
+        pass
+
+    # Python/Bash utils
+    def create_venv(self):
+        venv_path = os.path.join(self.path, 'venv')
+        if not self.has_venv():
+            logger.debug(f"Creating venv for {self.name}")
+            self.bash(f"{PYTHON_EXEC} -m venv {venv_path} --system-site-packages")
+            self.pip("install --upgrade pip")
+        else:
+            logger.debug(f"Venv already exists for {self.name}")
+
+    def has_venv(self) -> bool:
+        return self.dir_exists('venv')
+
+    def pip_install(self, package: str | list, no_deps: bool = False):
+        if isinstance(package, list):
+            for p in package:
+                self.pip(f"install -U {p} {'--no-deps' if no_deps else ''}")
+        else:
+            self.pip(f"install -U {package} {'--no-deps' if no_deps else ''}")
+
+    def install_requirements(self, filename: str = 'requirements.txt'):
+        self.pip(f"install -r {filename}")
+
+    def pip(self, cmd: str, env=[], current_dir: str = None):
+        self.bash(f"{' '.join(env)} {self.path}/venv/bin/pip {cmd}", current_dir)
+
+    def python(self, cmd: str, env=[], current_dir: str = None):
+        self.bash(f"{' '.join(env)} {self.path}/venv/bin/python {cmd}", current_dir)
+
+    def bash(self, cmd: str, current_dir: str = None):
+        cmd = f"cd {self.path if current_dir is None else os.path.join(self.path, current_dir)} && {cmd}"
+
+        logger.debug(f"Running command: {cmd}")
+
+        if LEVEL == logging.DEBUG:
+            process = subprocess.Popen(cmd, shell=True)
+            process.wait()
+            if process.returncode != 0:
+                raise Exception(f"Failed to run command: {cmd}")
+
+        else:
+            process = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+            out, err = process.communicate()
+
+            if process.returncode != 0:
+                logger.fatal(f"Failed to run command: {cmd}")
+                logger.fatal(f"Error: {err.decode('utf-8')}")
+                logger.fatal(f"Output: {out.decode('utf-8')}")
+                raise Exception(f"Failed to run command: {cmd}")
+
+    # Git utils
+    def git_clone(self, url: str, branch: str = None, dest: str = None):
+        self.bash(f"git clone {f"-b {branch}" if branch is not None else ''} {url} {'' if dest is None else dest}")
+
+    def git_pull(self, repo_folder: str, force: bool = False):
+        self.bash(f"git reset --hard HEAD {'&& git clean -f -d' if force else ''} && git pull", repo_folder)
+
+    # Prebuilt utils
+    def install_from_prebuilt(self, name):
+        for prebuilt in utils.get_prebuilts():
+            if prebuilt['name'].split("-")[0] == name:
+                self.pip(f"install {prebuilt['browser_download_url']}")
+                return
+
+    # File utils
+    def create_file(self, name, content):
+        with open(os.path.join(self.path, name), 'w') as f:
+            f.write(content)
+
+    def create_dir(self, name):
+        logger.debug(f"Creating directory {name}")
+        os.makedirs(os.path.join(self.path, name), exist_ok=True)
+
+    def remove_file(self, name):
+        logger.debug(f"Removing file {name}")
+        os.remove(os.path.join(self.path, name))
+
+    def remove_dir(self, name):
+        logger.debug(f"Removing directory {name}")
+        os.rmdir(os.path.join(self.path, name))
+
+    def move_file_or_dir(self, src, dest):
+        logger.debug(f"Moving file/dir {src} to {dest}")
+        os.rename(os.path.join(self.path, src), os.path.join(self.path, dest))
+
+    def move_all_files_in_dir(self, src, dest):
+        logger.debug(f"Moving all files in directory {src} to {dest}")
+        for file in os.listdir(os.path.join(self.path, src)):
+            os.rename(os.path.join(self.path, src, file), os.path.join(self.path, dest, file))
+
+    def file_exists(self, name):
+        return os.path.exists(os.path.join(self.path, name))
+
+    def dir_exists(self, name):
+        return os.path.exists(os.path.join(self.path, name))
+
+    def remove_line_in_file(self, contains: str | list, file: str):
+        logger.debug(f"Removing lines containing {contains} in {file}")
+
+        if isinstance(contains, list):
+            for c in contains:
+                self.bash(f"sed -i '/{c}/d' {file}")
+        else:
+            self.bash(f"sed -i '/{contains}/d' {file}")
+
+    def replace_line_in_file(self, match: str, replace: str, file: str):
+        logger.debug(f"Replacing lines containing {match} with {replace} in {file}")
+        self.bash(f"sed -i 's/{match}/{replace}/g' {file}")