sonnet

#!/usr/bin/env python
import hashlib
import os
import shutil
import subprocess
import sys
import tarfile

from gzip import GzipFile

from cleo import Application
from cleo import Command
from clikit.api.formatter import Style


WINDOWS = sys.platform.startswith("win") or (sys.platform == "cli" and os.name == "nt")


class MakeReleaseCommand(Command):
    """
    Makes a self-contained package of Poetry.

    release
        {--P|python=?* : Python version to use}
    """

    PYTHON = {
        "3.6": "python3.6",
        "3.7": "python3.7",
        "3.8": "python3.8",
        "3.9": "python3.9",
    }

    def handle(self):
        pythons = self.PYTHON
        if self.option("python"):
            pythons = {}
            for python in self.option("python"):
                parts = python.split(":", 1)
                if len(parts) == 1:
                    python = self.PYTHON[parts[0]]
                    version = parts[0]
                else:
                    version, python = parts
                pythons[version] = python

        self.check_system(pythons)

        from poetry.__version__ import __version__
        from poetry.core.vcs import get_vcs
        from poetry.factory import Factory
        from poetry.puzzle import Solver
        from poetry.repositories.pool import Pool
        from poetry.repositories.repository import Repository
        from poetry.utils._compat import Path
        from poetry.utils.env import EnvManager
        from poetry.utils.env import VirtualEnv
        from poetry.utils.helpers import temporary_directory

        project = Factory().create_poetry(Path.cwd())
        package = project.package
        del package.dev_requires[:]

        # We only use the lock file to resolve the dependencies
        pool = Pool()
        pool.add_repository(project.locker.locked_repository(with_dev_reqs=True))

        vcs = get_vcs(Path(__file__).parent)
        if vcs:
            vcs_excluded = [str(f) for f in vcs.get_ignored_files()]
        else:
            vcs_excluded = []

        created_files = []
        with temporary_directory() as tmp_dir:
            # Copy poetry to tmp dir
            poetry_dir = os.path.join(tmp_dir, "poetry")
            shutil.copytree(
                os.path.join(os.path.dirname(__file__), "poetry"),
                poetry_dir,
                ignore=lambda dir_, names: set(vcs_excluded).intersection(
                    set([os.path.join(dir_, name) for name in names])
                ),
            )
            created_files += [
                p.relative_to(Path(tmp_dir))
                for p in Path(poetry_dir).glob("**/*")
                if p.is_file()
                and p.suffix != ".pyc"
                and str(p.relative_to(Path(tmp_dir))) not in vcs_excluded
            ]
            for version, python in sorted(pythons.items()):
                self.line(
                    "<info>Preparing files for Python <comment>{}</comment></info>".format(
                        version
                    )
                )

                with temporary_directory() as tmp_venv_dir:
                    venv_dir = Path(tmp_venv_dir) / ".venv"
                    EnvManager.build_venv(venv_dir.as_posix(), executable=python)

                    env = VirtualEnv(venv_dir, venv_dir)
                    solver = Solver(package, pool, Repository(), Repository(), self.io)
                    with solver.use_environment(env):
                        ops = solver.solve()
                        for op in ops:
                            if not env.is_valid_for_marker(op.package.marker):
                                op.skip("Not needed for the current environment")

                    vendor_dir = Path(
                        self.vendorize_for_python(
                            env,
                            [op.package for op in ops if not op.skipped],
                            poetry_dir,
                            version,
                        )
                    )
                    created_files += [
                        p.relative_to(Path(tmp_dir))
                        for p in vendor_dir.glob("**/*")
                        if p.is_file()
                        and p.suffix != ".pyc"
                        and str(p.relative_to(Path(tmp_dir))) not in vcs_excluded
                    ]

                self.line("")

            self.line("<info>Packaging files</info>")
            with temporary_directory() as tmp_dir2:
                base_name = "poetry-{}-{}".format(__version__, sys.platform)
                name = "{}.tar.gz".format(base_name)
                gz = GzipFile(os.path.join(tmp_dir2, name), mode="wb")
                try:
                    with tarfile.TarFile(
                        os.path.join(tmp_dir2, name),
                        mode="w",
                        fileobj=gz,
                        format=tarfile.PAX_FORMAT,
                    ) as tar:
                        for root, dirs, files in os.walk(tmp_dir):
                            for f in files:
                                if f.endswith(".pyc"):
                                    continue

                                path = os.path.join(os.path.realpath(root), f)

                                relpath = os.path.relpath(
                                    path, os.path.realpath(tmp_dir)
                                )

                                if relpath in vcs_excluded:
                                    continue

                                tar_info = tar.gettarinfo(str(path), arcname=relpath)

                                if tar_info.isreg():
                                    with open(path, "rb") as f:
                                        tar.addfile(tar_info, f)
                                else:
                                    tar.addfile(tar_info)  # Symlinks & ?
                finally:
                    gz.close()

                self.line("<info>Checking release file</info>")
                missing_files = []
                with tarfile.open(os.path.join(tmp_dir2, name), "r") as tar:
                    names = tar.getnames()

                    for created_file in created_files:
                        if created_file.as_posix() not in names:
                            missing_files.append(created_file.as_posix())

                if missing_files:
                    self.line("<error>Some files are missing:</error>")
                    for missing_file in missing_files:
                        self.line("<error>  - {}</error>".format(missing_file))

                    return 1

                releases_dir = os.path.join(os.path.dirname(__file__), "releases")
                if not os.path.exists(releases_dir):
                    os.mkdir(releases_dir)

                shutil.copyfile(
                    os.path.join(tmp_dir2, name), os.path.join(releases_dir, name)
                )

                # Compute hash
                sha = hashlib.sha256()
                with open(os.path.join(releases_dir, name), "rb") as f:
                    while True:
                        buffer = f.read(8192)
                        if not buffer:
                            break

                        sha.update(buffer)

                with open(
                    os.path.join(releases_dir, "{}.sha256sum".format(base_name)), "w"
                ) as f:
                    f.write(sha.hexdigest())

                self.line("<info>Built <comment>{}</comment></info>".format(name))

    def check_system(self, pythons):
        for version, python in sorted(pythons.items()):
            try:
                subprocess.check_output(
                    [python, "-V"], stderr=subprocess.STDOUT, shell=WINDOWS
                )
                if version == "3.4" and WINDOWS:
                    continue

                subprocess.check_output([python, "-m", "pip", "install", "pip", "-U"])
            except subprocess.CalledProcessError:
                raise RuntimeError("Python {} is not available".format(version))

    def vendorize_for_python(self, env, packages, dest, python_version):
        vendor_dir = os.path.join(dest, "_vendor", "py{}".format(python_version))

        bar = self.progress_bar(max=len(packages))
        bar.set_format("%message% %current%/%max%")
        bar.set_message(
            "<info>Vendorizing dependencies for Python <comment>{}</comment></info>".format(
                python_version
            )
        )
        bar.start()
        for package in packages:
            env.run_pip(
                "install",
                "{}=={}".format(package.name, package.version),
                "--no-deps",
                "--target",
                vendor_dir,
            )
            bar.advance()

        bar.finish()

        self.line("")

        return vendor_dir


class MakeCommand(Command):
    """
    Build poetry releases.

    make
    """

    commands = [MakeReleaseCommand()]

    def handle(self):
        return self.call("help", self.config.name)


app = Application("sonnet")
app.config.add_style(Style("debug").fg("default").dark())

app.add(MakeCommand())

if __name__ == "__main__":
    app.run()