sonnet

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

from gzip import GzipFile
from pathlib import Path
from typing import Optional

from cleo.application import Application as BaseApplication
from cleo.commands.command import Command
from cleo.formatters.style import Style
from cleo.helpers import option
from cleo.io.inputs.input import Input
from cleo.io.io import IO
from cleo.io.outputs.output import Output


class Application(BaseApplication):
    def create_io(
        self,
        input: Optional[Input] = None,
        output: Optional[Output] = None,
        error_output: Optional[Output] = None,
    ) -> IO:
        io = super(Application, self).create_io(input, output, error_output)

        io.output.formatter.set_style("debug", Style("default", options=["dark"]))
        io.error_output.formatter.set_style("debug", Style("default", options=["dark"]))

        return io


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


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

    name = "make release"

    options = [option("--python", "-P", flag=False, multiple=True)]

    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.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
                )

                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


app = Application("sonnet")
app.add(MakeReleaseCommand())

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