path: root/share/extensions/other/inkman/inkman/target.py

#
# Copyright (C) 2019 Martin Owens
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
#
"""
Target a directory to install resources into.
"""

import os
import sys
import json
import logging
from shutil import which

from inkex.inx import InxFile
from inkex.command import call, CommandNotFound, ProgramRunError

from .archive import Archive, UnrecognizedArchiveFormat
from .remote import RemoteArchive, LocalFile
from .utils import INKSCAPE_PROFILE, CACHE_DIR, ExtensionInx
from .package import PackageItem, OrphanedItem, PythonItem, PythonPackage


class BasicTarget(object):
    """
    A location where to install something, plus how to search for it.
    """

    version_specific = False

    def __init__(self, category, label, path, is_search=False, filters=()):
        self.category = category
        self.label = label
        self.path = os.path.join(INKSCAPE_PROFILE, path)
        self.is_search = is_search
        self.archive = RemoteArchive(category)
        self.filters = filters
        self._installed = None

    def search(self, query, filtered=False):
        """Search the online archive for things"""
        for pkg in self.archive.search(query, filtered):
            pkg.set_installer(self._install)
            yield pkg

    def is_writable(self):
        """Can the folder be written to (for installation)"""
        try:
            return os.access(self.path, os.W_OK)
        except IOError:
            return False

    def _install(self, filename, info):
        if not info.get("id"):
            info["id"] = self.generate_id(filename)

        if not info.get("id"):
            raise ValueError("Id is a required field for packages")

        if filename.endswith(".zip"):
            location = info["id"]
            fname = "info.json"
            info["files"] = list(self.install_zip_files(filename, location))
        else:
            location = "pkg"
            fname = info["id"] + ".json"
            info["files"] = [self.write_file(filename, os.path.basename(filename))]

        self.write_file(json.dumps(info).encode("utf8"), fname, extra=location)

        return f"Package installed! Remember to restart inkscape to use it!"

    def _uninstall(self, info, json_file):
        self._installed = None
        for fname in info.get("files"):
            self.remove_file(fname)
        if json_file and os.path.isfile(json_file):
            self.remove_file(json_file)
        return True

    def install_zip_files(self, filename, location):
        """Install the files in the zip filename as non-pip files"""
        with Archive(filename) as archive:
            for filename in archive.filenames():
                yield self.write_file(archive.read(filename), filename, extra=location)

    def generate_id(self, filename):
        """User submitted zip file, generate an id as best we can"""
        return filename.replace(".zip", "")

    def list_installed(self, cached=True):
        """
        Loops through all the files in the target path and finds all the installed items.
        """
        if cached and self._installed:
            yield from self._installed
            return

        self._installed = []
        for item in self._list_installed():
            self._installed.append(item)
            yield item

    def _list_installed(self):
        orphans = OrphanedItem(self.path)
        for root, subs, files in os.walk(self.path):
            for fname in files:
                fpath = os.path.join(root, fname)
                name = self.unprefix_path(fpath)
                if fname.endswith(".json"):
                    if os.path.basename(fpath) == "package.json":
                        continue
                    with open(fpath, "rb") as fhl:
                        data = fhl.read()
                        info = json.loads(data)
                        if not PackageItem.is_pkg(info):
                            continue

                        item = PackageItem(
                            info, remote=self._remote_or_local_file(root)
                        )
                        item.set_uninstaller(self._uninstall, fpath)
                        yield item

                        for pkg_file in item.get_files(missing=True):
                            orphans.remove_file(pkg_file, item)
                            orphans.remove_file(os.path.join(root, pkg_file), item)
                else:
                    orphans.add_file(name)

        for fname, item in orphans.get_missing():
            if item is not None:
                item._missing.append(fname)

        if orphans.get_files(filters=self.filters):
            yield orphans

    def _remote_or_local_file(self, basedir):
        # If a json file specifies something that's local, it's "ALWAYS" local to the
        # json file as a basedir.
        def _inner(url):
            if "://" not in url:
                return LocalFile(basedir, url)
            return self.archive._remote_file(url)

        return _inner

    def write_file(self, source, source_name=None, extra=None):
        target = os.path.join(self.path, extra) if extra else self.path

        if isinstance(source, str):
            if not source_name:
                source_name = source
            with open(source, "rb") as fhl:
                source = fhl.read()

        path = os.path.join(target, source_name)
        filedir = os.path.dirname(path)
        if not os.path.isdir(filedir):
            os.makedirs(filedir)

        # Ignore paths
        if not os.path.isdir(path):
            with open(path, "wb") as whl:
                whl.write(source)

        return self.unprefix_path(path)

    def remove_file(self, filename):
        """
        Remove the given file and clean up
        """
        if not filename.startswith(self.path):
            filename = os.path.join(self.path, filename)
        if os.path.isfile(filename):
            os.unlink(filename)

        # Recursively clean up directories (if empty)
        path = os.path.dirname(filename)
        while path.lstrip("/") != self.path.lstrip("/"):
            if os.listdir(path):
                break
            os.rmdir(path)
            path = os.path.dirname(path)

    def unprefix_path(self, path):
        """
        Removes the prefix of the given path, if it's based in self.path
        """
        # Strip just the OS seperator, but what if the files were moved from another OS?
        return path.replace(self.path, "").lstrip("/").lstrip("\\")


class ExtensionsTarget(BasicTarget):
    """
    Extra functional target for extensions (pip based)
    """

    version_specific = True

    def get_pip(self):
        path = os.path.abspath(os.path.join(self.path, "bin"))
        return which("pip", path=path + ":" + os.environ["PATH"])

    def _install(self, filename, info):
        if self.is_pip_package(filename):
            results = self.pip_install(filename)
            if results:
                info["pip"] = True
                info["id"] = results.strip().split()[-1]
                fname = info["id"] + ".json"
                self.write_file(json.dumps(info).encode("utf8"), fname, extra="lib")
                return (
                    f"Python Package installed! Remember to restart inkscape to use it!"
                )
            return f"Failed to install, something is wrong with your setup."

        return super()._install(filename, info)

    def _uninstall(self, info, json_file):
        self._installed = None
        if not info.get("pip"):
            return super()._uninstall(info, json_file)
        self.pip_uninstall(info["name"])
        if json_file and os.path.isfile(json_file):
            self.remove_file(json_file)
        return True

    def is_pip_package(self, filename):
        """Return true if this is a detectable pip package"""
        if filename.endswith(".whl"):
            return True
        try:
            with Archive(filename) as archive:
                for filename in archive.filenames():
                    if filename.endswith("setup.py"):
                        return True
        except UnrecognizedArchiveFormat:
            return False
        return False

    def pip_install(self, filename):
        """Install the filename as a pip package"""
        pip = self.get_pip()
        if pip is None:
            logging.error(
                "This package requires python VirtualEnv which is not available on your system."
            )
            return None
        try:
            results = call(
                pip,
                "install",
                ("isolated", True),
                ("disable-pip-version-check", True),
                ("cache-dir", CACHE_DIR),
                filename,
            ).decode("utf8")
        except ProgramRunError as err:
            raise
        return results

    def pip_uninstall(self, name):
        """Uninstall the given pip package name"""
        try:
            results = call(
                self.get_pip(), "uninstall", ("disable-pip-version-check", True), name
            ).decode("utf8")
        except ProgramRunError as err:
            raise
        return results

    def generate_id(self, filename):
        """Extensions have an id internally, try and use it"""
        try:
            with Archive(filename) as archive:
                inxes = [item for item in archive.filenames() if item.endswith(".inx")]
                if not inxes:
                    raise IOError("Refusing to install extension without inx file!")
                inx = ExtensionInx(archive.read(inxes[0]).decode("utf-8"))
            return inx.ident
        except UnrecognizedArchiveFormat:
            raise IOError(
                "Refusing the install extension without inx file (unknown archive)"
            )
        except:
            raise IOError("Refusing the install extension with bad inx file!")

    def _list_installed(self):
        """
        Add pip packages to file lists.
        """
        orphans = None
        packages = {}
        all_deps = set()
        all_files = set()

        # First collect a list of python packages installed
        for node in self.get_python_paths():
            if node.endswith(".dist-info") or node.endswith(".egg-info"):
                package = PythonPackage(node, self.path)
                packages[package.name] = package
                all_files |= set(package.package_files())
                for dep, _ in package.get_depedencies():
                    all_deps.add(dep)

        # Now return all non pip packaged extensions (from super)
        for item in super()._list_installed():
            if item.info.get("pip", False):
                if self.info.ident not in packages:
                    print(f"Can't find python package: {item.ient}")
                    continue

                pip_pkg = packages[item.ident]
                item.info["version"] = pip_pkg.version

            if isinstance(item, OrphanedItem):
                orphans = item
            else:
                yield item

        # Remove all orphaned files that were installed by pip packages
        if orphans is not None:
            for fn in all_files:
                orphans.remove_file(fn)
            if orphans.get_files(filters=self.filters):
                # Yield if we still have orphans
                yield orphans

        # Now what to do with all these remaining packages, pretend their installed?
        for name, package in packages.items():
            for inx in package.get_inx():
                item = PythonItem(package)
                item.set_uninstaller(self._uninstall, None)
                yield item
                break

        for dep in all_deps:
            if dep not in packages:
                # These packages are often just installed into the system, nothing to say.
                # XXX But, there is a future where pip could be interigated.
                # logging.error(f"Missing python depedency: {dep}")
                continue
            packages.pop(dep)

    def get_python_paths(self):
        """Returns paths related to the python packages"""
        pyver = "python" + sys.version[:3]
        for varient in [
            os.path.join(self.path, "lib", pyver, "site-packages"),
        ]:
            if os.path.isdir(varient):
                for subpath in os.listdir(varient):
                    yield os.path.join(varient, subpath)

    def get_package(self, name, version=None):
        """Test every package in this list if it matches the name and version"""
        for package in self.iter():
            found = package.is_package(name, version=version)
            if found:
                return package
        return None