stanza

#!/usr/bin/env python
import os
import subprocess
import tarfile
import zipfile

from pathlib import Path
from typing import Dict
from typing import List

from cleo import Application
from cleo import Command
from cleo import argument
from vendoring.configuration import Configuration
from vendoring.configuration import load_configuration
from vendoring.tasks.cleanup import cleanup_existing_vendored
from vendoring.tasks.license import find_and_extract_license
from vendoring.tasks.license import license_fallback
from vendoring.tasks.vendor import apply_patches
from vendoring.tasks.vendor import detect_vendored_libs
from vendoring.tasks.vendor import download_libraries
from vendoring.tasks.vendor import remove_unnecessary_items
from vendoring.utils import remove_all
from vendoring.utils import run


def extract_license(
    destination: Path,
    sdist: Path,
    license_directories: Dict[str, str],
    license_fallback_urls: Dict[str, str],
) -> None:
    def extract_from_source_tarfile(sdist: Path) -> bool:
        ext = sdist.suffixes[-1][1:]
        with tarfile.open(sdist, mode="r:{}".format(ext)) as tar:
            return find_and_extract_license(
                destination,
                tar,
                tar.getmembers(),
                license_directories,
            )

    def extract_from_source_zipfile(sdist: Path) -> bool:
        with zipfile.ZipFile(sdist) as zip:
            return find_and_extract_license(
                destination,
                zip,
                zip.infolist(),
                license_directories,
            )

    if sdist.suffixes[-2] == ".tar":
        found = extract_from_source_tarfile(sdist)
    elif sdist.suffixes[-1] == ".zip":
        found = extract_from_source_zipfile(sdist)
    elif sdist.suffixes[-1] == ".whl":
        found = extract_from_source_zipfile(sdist)
    else:
        raise NotImplementedError("new sdist type!")

    if found:
        return

    license_fallback(
        destination, sdist.name, license_directories, license_fallback_urls
    )


def fetch_licenses(config: Configuration) -> None:
    destination = config.destination
    license_directories = config.license_directories
    license_fallback_urls = config.license_fallback_urls
    requirements = config.requirements

    tmp_dir = destination / "__tmp__"
    download_sources(tmp_dir, requirements)

    for sdist in tmp_dir.iterdir():
        extract_license(destination, sdist, license_directories, license_fallback_urls)

    remove_all([tmp_dir])


def vendor_libraries(config: Configuration) -> List[str]:
    destination = config.destination

    # Download the relevant libraries.
    download_libraries(config.requirements, destination)

    # Cleanup unnecessary directories/files created.
    remove_unnecessary_items(destination, config.drop_paths)

    # Detect what got downloaded.
    vendored_libs = detect_vendored_libs(destination, config.protected_files)

    # Apply user provided patches.
    apply_patches(config.patches_dir, working_directory=config.base_directory)

    return vendored_libs


def download_sources(location: Path, requirements: Path) -> None:
    cmd = [
        "pip",
        "download",
        "-r",
        str(requirements),
        "--no-deps",
        "--dest",
        str(location),
    ]

    run(cmd, working_directory=None)


class VendorUpdateCommand(Command):

    name = "update"

    description = "Update one or more vendor packages"

    arguments = [
        argument("packages", "The packages to vendor.", optional=True, multiple=True)
    ]

    def handle(self):
        packages = self.argument("packages")
        current_dir = os.getcwd()
        base = os.path.dirname(__file__)
        try:
            os.chdir(base.join(["vendors"]))
            if not packages:
                subprocess.run(["poetry", "lock"])
            else:
                subprocess.run(["poetry", "update", "--lock"])

            subprocess.run(["poetry", "show", "--all", "--tree"])
            subprocess.run(
                [
                    "poetry",
                    "export",
                    "-f",
                    "requirements.txt",
                    "-o",
                    "../poetry/core/_vendor/vendor.txt",
                    "--without-hashes",
                ]
            )
        finally:
            os.chdir(current_dir)

        lines = []
        with open("poetry/core/_vendor/vendor.txt") as f:
            for line in f.readlines():
                if ";" in line:
                    line, _ = line.split(";", maxsplit=1)

                if line.startswith("wheels/"):
                    line = "vendors/" + line

                if line.startswith(
                    ("enum34", "functools32", "pathlib2", "typing", "scandir", "typing")
                ):
                    continue

                lines.append(line.strip())

        with open("poetry/core/_vendor/vendor.txt", "w") as f:
            f.write("\n".join(lines))

        config = load_configuration(Path(base))
        cleanup_existing_vendored(config)
        vendor_libraries(config)
        fetch_licenses(config)


class VendorCommand(Command):

    name = "vendor"

    description = "Vendor related commands."

    commands = [VendorUpdateCommand()]

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


app = Application("stanza")
app.add(VendorCommand())


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