zaz-pip/source/pythonpath/install_flatpak.py

from __future__ import annotations
from typing import cast
import os
import sys
import tempfile
import subprocess
from pathlib import Path
from typing import Any, List, Dict
import easymacro as app
from contextlib import contextmanager
import uno


@contextmanager
def change_dir(directory):
    """
    A context manager that changes the current working directory to the specified directory
    temporarily and then changes it back when the context is exited.
    """
    current_dir = os.getcwd()
    os.chdir(directory)
    try:
        yield
    finally:
        os.chdir(current_dir)


class InstallPipFromWheel:
    """Download and install PIP from wheel url"""

    def __init__(self, pip_wheel_url: str, lo_identifier: str) -> None:
        self._pip_url = pip_wheel_url
        self._lo_identifier = lo_identifier

    def install(self, dst: str | Path = "") -> None:
        """
        Install pip from wheel file.

        Downloads the pip wheel file from the url provided in the config file and unzips it to the destination directory.

        Args:
            dst (str | Path, Optional): The destination directory where the pip wheel file will be installed. If not provided, the ``pythonpath`` location will be used.

        Returns:
            None:

        Raises:
            None:
        """
        if not self._pip_url:
            app.error("PIP installation has failed - No wheel url")
            return

        if not dst:
            root_pth = Path(app.Paths.from_id(self._lo_identifier))
            dst = root_pth / "pythonpath"

        with tempfile.TemporaryDirectory() as temp_dir:
            # temp_dir = tempfile.gettempdir()
            path_pip = Path(temp_dir)

            filename = path_pip / "pip-wheel.whl"

            data, _, err = app.url_open(self._pip_url, verify=False)
            if err:
                app.error("Unable to download PIP installation wheel file")
                return
            filename.write_bytes(data)

            if filename.exists():
                app.info("PIP wheel file has been saved")
            else:
                app.error("PIP wheel file has not been saved")
                return

            try:
                self._unzip_wheel(filename=filename, dst=dst)
            except Exception:
                return
            # now that pip has been installed from wheel force a reinstall to ensure it is the latest version
            self._force_install_pip()

    def _unzip_wheel(self, filename: Path, dst: str | Path) -> None:
        """Unzip the downloaded wheel file"""
        if isinstance(dst, str):
            dst = Path(dst)
        try:
            # app.zip.unzip(source=str(filename), target=str(dst))
            self.unzip_file(zip_file=filename, dest_dir=dst)
            if dst.exists():
                app.debug(f"PIP wheel file has been unzipped to {dst}")
            else:
                app.error("PIP wheel file has not been unzipped")
                raise Exception("PIP wheel file has not been unzipped")
        except Exception as err:
            app.error(f"Unable to unzip wheel file: {err}", exc_info=True)
            raise

    def _force_install_pip(self) -> None:
        """Now that pip has been installed, force reinstall it to ensure it is the latest version"""
        cmd = [app.paths.python, "-m", "pip", "install", "--upgrade", "pip"]
        app.popen(command=" ".join(cmd))

    def unzip_file(self, zip_file: str | Path, dest_dir: str | Path = "") -> None:
        """
        Unzip the given zip file to the specified destination directory.

        Args:
            zip_file (str | Path): The zip file to unzip.
            dest_dir (str | Path, optional): The destination directory to unzip to.

        Returns:
            None:
        """
        from zipfile import ZipFile

        zip_file_path = Path(zip_file) if isinstance(zip_file, str) else zip_file
        if not zip_file_path.is_file():
            raise ValueError(f"Expected file, got '{zip_file_path}'")
        if not zip_file_path.is_absolute():
            zip_file_path = zip_file_path.absolute()
        if not zip_file_path.exists():
            raise FileNotFoundError(f"File '{zip_file_path}' not found")

        if isinstance(dest_dir, str):
            dest_dir = zip_file_path.parent if dest_dir == "" else Path(dest_dir)
        else:
            dest_dir = dest_dir.absolute()

        if not dest_dir.is_dir():
            raise ValueError(f"Expected folder, got '{dest_dir}'")
        if not dest_dir.exists():
            try:
                dest_dir.mkdir(parents=True)
            except Exception as e:
                raise FileNotFoundError(
                    f"Folder '{dest_dir}' not found, unable to create folder."
                ) from e
            if not dest_dir.exists():
                raise FileNotFoundError(f"Folder '{dest_dir}' not found")

        with change_dir(dest_dir):
            with ZipFile(zip_file_path) as f:
                f.extractall(dest_dir)
        # with change_dir(dest_dir):
        #     shutil.unpack_archive(zip_file_path, dest_dir)


class FlatpakInstaller:
    """class for the PIP install."""

    def __init__(self, pip_wheel_url: str, lo_identifier: str) -> None:
        self.path_python = app.Paths.python
        app.debug(f"Python path: {self.path_python}")
        self._pip_url = pip_wheel_url
        self._lo_identifier = lo_identifier
        self._site_packages = cast(str, app.get_flatpak_site_packages_dir())

    def install_pip(self) -> None:
        if self.is_pip_installed():
            app.info("PIP is already installed")
            return
        if self._install_wheel():
            if self.is_pip_installed():
                app.info("PIP was installed successfully")
        else:
            app.error("PIP installation has failed")

        return

    def _get_pip_cmd(self, filename: Path) -> List[str]:
        return [str(self.path_python), f"{filename}", "--user"]

    def _get_env(self) -> Dict[str, str]:
        """
        Gets Environment used for subprocess.
        """
        my_env = os.environ.copy()
        py_path = ""
        p_sep = ";" if app.IS_WIN else ":"
        for d in sys.path:
            py_path = py_path + d + p_sep
        my_env["PYTHONPATH"] = py_path
        return my_env

    def _cmd_pip(self, *args: str) -> List[str]:
        cmd: List[str] = [str(self.path_python), "-m", "pip", *args]
        return cmd

    def _install_wheel(self) -> bool:
        result = False

        installer = InstallPipFromWheel(
            pip_wheel_url=self._pip_url, lo_identifier=self._lo_identifier
        )
        try:
            installer.install(self._site_packages)
            if self._site_packages not in sys.path:
                sys.path.append(self._site_packages)
            result = True
        except Exception as err:
            app.error(err)
            return result
        return result

    def is_pip_installed(self) -> bool:
        """Check if PIP is installed."""
        # cmd = self._cmd_pip("--version")
        # cmd = '"{}" -m pip -V'.format(self.path_python)
        cmd = [str(self.path_python), "-m", "pip", "-V"]
        result = subprocess.run(
            cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=self._get_env()
        )
        return result.returncode == 0