sh_cpl/src/cpl_cli/command/update_service.py

import json
import os
import subprocess
import sys

from cpl.application import ApplicationRuntimeABC
from cpl.console import ForegroundColorEnum
from cpl.console.console import Console
from cpl_cli.command_abc import CommandABC
from cpl_cli.configuration.project_settings import ProjectSettings


class UpdateService(CommandABC):

    def __init__(self, runtime: ApplicationRuntimeABC, project_settings: ProjectSettings):
        CommandABC.__init__(self)

        self._runtime = runtime
        self._project_settings = project_settings

    @staticmethod
    def _install(package: str, *args, source: str = None, stdout=None, stderr=None):
        pip_args = [sys.executable, "-m", "pip", "install"]

        for arg in args:
            pip_args.append(arg)

        if source is not None:
            pip_args.append(f'--extra-index-url')
            pip_args.append(source)

        pip_args.append(package)
        subprocess.run(pip_args, stdout=stdout, stderr=stderr)

    @staticmethod
    def _get_outdated() -> bytes:
        return subprocess.check_output([sys.executable, "-m", "pip", "list", "--outdated"])

    @staticmethod
    def _get_package(package: str) -> str:
        result = subprocess.check_output([sys.executable, "-m", "pip", "show", package])

        new_package: list[str] = str(result, 'utf-8').lower().split('\n')
        new_version = ''

        for atr in new_package:
            if 'version' in atr:
                new_version = atr.split(': ')[1]

        return f'{package}=={new_version}'

    def _check_outdated(self):
        table_str: bytes = Console.spinner(
            'Analyzing for available package updates', self._get_outdated,
            text_foreground_color=ForegroundColorEnum.green,
            spinner_foreground_color=ForegroundColorEnum.cyan
        )

        Console.write_line('\tAvailable updates for packages:')
        table = str(table_str, 'utf-8').split('\n')
        for row in table:
            Console.write_line(f'\t{row}')

        Console.write_line(f'\tUpdate with {sys.executable} -m pip install --upgrade <package>')

    def _project_json_update_dependency(self, old_package: str, new_package: str):
        content = ''
        with open(os.path.join(self._runtime.working_directory, 'cpl.json'), 'r') as project:
            content = project.read()
            project.close()

        if content == '' or content == '{}':
            return

        if old_package in content:
            content = content.replace(old_package, new_package)

        with open(os.path.join(self._runtime.working_directory, 'cpl.json'), 'w') as project:
            project.write(json.dumps(json.loads(content), indent=2))
            project.close()

    def _update_project_dependencies(self):
        for package in self._project_settings.dependencies:
            name = package
            if '==' in package:
                name = package.split('==')[0]
            self._install(
                name,
                '--upgrade',
                '--upgrade-strategy',
                'eager',
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL
            )

            self._project_json_update_dependency(package, self._get_package(name))

    def _check_project_dependencies(self):
        Console.spinner(
            'Collecting installed dependencies', self._update_project_dependencies,
            text_foreground_color=ForegroundColorEnum.green,
            spinner_foreground_color=ForegroundColorEnum.cyan
        )
        Console.write_line(f'Found {len(self._project_settings.dependencies)} dependencies.')

    def run(self, args: list[str]):
        # target update discord 1.5.1 to discord 1.6.0
        self._check_project_dependencies()

        Console.spinner(
            'Checking update for sh_cpl',
            self._install, 'sh_cpl', source='https://pip.sh-edraft.de', stdout=subprocess.DEVNULL,
            text_foreground_color=ForegroundColorEnum.green,
            spinner_foreground_color=ForegroundColorEnum.cyan
        )

        self._check_outdated()

        Console.write('\n')