sh_cpl/src/cpl_cli/command/install_service.py

import json
import os
import subprocess
import textwrap
import time

from packaging import version

from cpl_cli.cli_settings import CLISettings
from cpl_cli.command_abc import CommandABC
from cpl_cli.configuration.build_settings import BuildSettings
from cpl_cli.configuration.project_settings import ProjectSettings
from cpl_cli.configuration.settings_helper import SettingsHelper
from cpl_cli.configuration.venv_helper_service import VenvHelper
from cpl_cli.error import Error
from cpl_core.configuration.configuration_abc import ConfigurationABC
from cpl_core.console.console import Console
from cpl_core.console.foreground_color_enum import ForegroundColorEnum
from cpl_core.environment.application_environment_abc import ApplicationEnvironmentABC
from cpl_core.utils.pip import Pip


class InstallService(CommandABC):
    def __init__(
        self,
        config: ConfigurationABC,
        env: ApplicationEnvironmentABC,
        build_settings: BuildSettings,
        project_settings: ProjectSettings,
        cli_settings: CLISettings,
    ):
        """
        Service for the CLI command install
        :param config:
        :param env:
        :param build_settings:
        :param project_settings:
        :param cli_settings:
        """
        CommandABC.__init__(self)

        self._config = config
        self._env = env
        self._build_settings = build_settings
        self._project_settings = project_settings
        self._cli_settings = cli_settings

        self._is_simulation = False
        self._is_virtual = False
        self._is_dev = False

        self._project_file = f"{self._project_settings.name}.json"

    @property
    def help_message(self) -> str:
        return textwrap.dedent(
            """\
        Installs given package via pip
        Usage: cpl install <package>
        
        Arguments:
            package    The package to install 
        """
        )

    def _wait(self, t: int, *args, source: str = None, stdout=None, stderr=None):
        time.sleep(t)

    def _install_project(self):
        """
        Installs dependencies of CPl project
        :return:
        """
        if self._project_settings is None or self._build_settings is None:
            Error.error("The command requires to be run in an CPL project, but a project could not be found.")
            return

        if self._project_settings.dependencies is None:
            Error.error(f"Found invalid dependencies in {self._project_file}.")
            return

        for dependency in self._project_settings.dependencies:
            Console.spinner(
                f"Installing: {dependency}",
                Pip.install if not self._is_virtual else self._wait,
                dependency if not self._is_virtual else 2,
                "--upgrade",
                source=self._cli_settings.pip_path,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                text_foreground_color=ForegroundColorEnum.green,
                spinner_foreground_color=ForegroundColorEnum.cyan,
            )
            local_package = Pip.get_package(dependency)
            if local_package is None:
                Error.warn(f"Installation of package {dependency} failed!")
                return

        for dependency in self._project_settings.dev_dependencies:
            Console.spinner(
                f"Installing dev: {dependency}",
                Pip.install if not self._is_virtual else self._wait,
                dependency if not self._is_virtual else 2,
                "--upgrade",
                source=self._cli_settings.pip_path,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                text_foreground_color=ForegroundColorEnum.green,
                spinner_foreground_color=ForegroundColorEnum.cyan,
            )
            local_package = Pip.get_package(dependency)
            if local_package is None:
                Error.warn(f"Installation of package {dependency} failed!")
                return

        if not self._is_virtual:
            Pip.reset_executable()

    def _install_package(self, package: str):
        """
        Installs given package
        :param package:
        :return:
        """
        is_already_in_project = False
        if self._project_settings is None or self._build_settings is None:
            Error.error("The command requires to be run in an CPL project, but a project could not be found.")
            return

        if self._project_settings.dependencies is None:
            Error.error(f"Found invalid dependencies in {self._project_file}.")
            return

        package_version = ""
        name = package
        if "==" in package:
            name = package.split("==")[0]
            package_version = package.split("==")[1]
        elif ">=" in package:
            name = package.split(">=")[0]
            package_version = package.split(">=")[1]
        elif "<=" in package:
            name = package.split("<=")[0]
            package_version = package.split("<=")[1]

        to_remove_list = []
        deps = self._project_settings.dependencies
        if self._is_dev:
            deps = self._project_settings.dev_dependencies

        for dependency in deps:
            dependency_version = ""

            if "==" in dependency:
                dependency_version = dependency.split("==")[1]
            elif "<=" in dependency:
                dependency_version = dependency.split("<=")[1]
            elif ">=" in dependency:
                dependency_version = dependency.split(">=")[1]

            if name in dependency:
                if package_version != "" and version.parse(package_version) != version.parse(dependency_version):
                    to_remove_list.append(dependency)
                    break
                else:
                    is_already_in_project = True

        for to_remove in to_remove_list:
            if self._is_dev:
                self._project_settings.dev_dependencies.remove(to_remove)
            else:
                self._project_settings.dependencies.remove(to_remove)

        local_package = Pip.get_package(package)
        if local_package is not None and local_package in self._project_settings.dependencies:
            Error.warn(f"Package {local_package} is already installed.")
            return

        elif is_already_in_project:
            Error.warn(f"Package {package} is already installed.")
            return

        Console.spinner(
            f"Installing: {package}" if not self._is_dev else f"Installing dev: {package}",
            Pip.install if not self._is_virtual else self._wait,
            package if not self._is_virtual else 2,
            source=self._cli_settings.pip_path,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            text_foreground_color=ForegroundColorEnum.green,
            spinner_foreground_color=ForegroundColorEnum.cyan,
        )

        if self._is_virtual:
            new_package = name
        else:
            new_package = Pip.get_package(name)

        if (
            new_package is None
            or "==" in package
            and version.parse(package.split("==")[1]) != version.parse(new_package.split("==")[1])
            or "<=" in package
            and version.parse(package.split("<=")[1]) != version.parse(new_package.split("<=")[1])
            or ">=" in package
            and version.parse(package.split(">=")[1]) != version.parse(new_package.split(">=")[1])
        ):
            Console.error(f"Installation of package {package} failed")
            return

        if not is_already_in_project:
            new_name = package
            if "==" in new_package or ">=" in new_package or "<=" in new_package:
                new_name = new_package
            elif "==" in name or ">=" in name or "<=" in name:
                new_name = name

            if "/" in new_name:
                new_name = new_name.split("/")[0]

            if "\r" in new_name:
                new_name = new_name.replace("\r", "")

            if self._is_dev:
                self._project_settings.dev_dependencies.append(new_name)
            else:
                self._project_settings.dependencies.append(new_name)

            if not self._is_simulation:
                config = {
                    ProjectSettings.__name__: SettingsHelper.get_project_settings_dict(self._project_settings),
                    BuildSettings.__name__: SettingsHelper.get_build_settings_dict(self._build_settings),
                }

                with open(os.path.join(self._env.working_directory, self._project_file), "w") as project_file:
                    project_file.write(json.dumps(config, indent=2))
                    project_file.close()

        Pip.reset_executable()

    def execute(self, args: list[str]):
        """
        Entry point of command
        :param args:
        :return:
        """
        if "dev" in args:
            self._is_dev = True
            args.remove("dev")

        if "virtual" in args:
            self._is_virtual = True
            args.remove("virtual")
            Console.write_line("Running in virtual mode:")

        if "simulate" in args:
            self._is_simulation = True
            args.remove("simulate")
            Console.write_line("Running in simulation mode:")

        if "cpl-prod" in args:
            args.remove("cpl-prod")
            self._cli_settings.from_dict({"PipPath": "https://pip.sh-edraft.de"})

        if "cpl-exp" in args:
            args.remove("cpl-exp")
            self._cli_settings.from_dict({"PipPath": "https://pip-exp.sh-edraft.de"})

        if "cpl-dev" in args:
            args.remove("cpl-dev")
            self._cli_settings.from_dict({"PipPath": "https://pip-dev.sh-edraft.de"})

        VenvHelper.init_venv(self._is_virtual, self._env, self._project_settings)

        if len(args) == 0:
            self._install_project()
        else:
            self._install_package(args[0])

        if not self._is_virtual:
            Pip.reset_executable()