contrib/cargo2ebuild.py

#!/usr/bin/env python3
# pylint: disable=line-too-long, subprocess-run-check, unused-argument
# Copyright 2021 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Convert Cargo metadata to ebuilds using cros-rust.

Used to add new Rust projects and building up ebuilds for the dependency tree.
"""

import argparse
from datetime import datetime
import json
import os
import re
import shutil
import subprocess
import sys

SCRIPT_NAME = 'cargo2ebuild.py'
AUTOGEN_NOTICE = '\n# This file was automatically generated by {}'.format(
    SCRIPT_NAME)

CRATE_DL_URI = 'https://crates.io/api/v1/crates/{PN}/{PV}/download'

# Required parameters
#   copyright_year: Current year for copyright assignment.
#   description: Description of the crates.
#   homepage: Homepage of the crates.
#   license: Ebuild compatible license string.
#   dependencies: Ebuild compatible dependency string.
#   autogen_notice: Autogenerated notification string.
EBUILD_TEMPLATE = (
"""# Copyright {copyright_year} The Chromium OS Authors. All rights reserved.
# Distributed under the terms of the GNU General Public License v2

EAPI="7"

CROS_RUST_REMOVE_DEV_DEPS=1

inherit cros-rust

DESCRIPTION="{description}"
HOMEPAGE="{homepage}"
SRC_URI="https://crates.io/api/v1/crates/${{PN}}/${{PV}}/download -> ${{P}}.crate"

LICENSE="{license}"
SLOT="${{PV}}/${{PR}}"
KEYWORDS="*"

{dependencies}{autogen_notice}
""")

# Required parameters:
#   copyright_year: Current year for copyright assignment.
#   crate_features: Features to add to this empty crate.
#   autogen_notice: Autogenerated notification string.
EMPTY_CRATE = (
"""# Copyright {copyright_year} The Chromium OS Authors. All rights reserved.
# Distributed under the terms of the GNU General Public License v2

EAPI="7"

CROS_RUST_EMPTY_CRATE=1
{crate_features}
inherit cros-rust

DESCRIPTION="Empty crate"
HOMEPAGE=""

LICENSE="BSD-Google"
SLOT="${{PV}}/${{PR}}"
KEYWORDS="*"
{autogen_notice}
""")

LICENSES = {
    'Apache-2.0': 'Apache-2.0',
    'MIT': 'MIT',
    'BSD-3-Clause': 'BSD',
    '0BSD': '0BSD',
    'ISC': 'ISC',
}

VERSION_RE = (
    '^(?P<dep>[\\^~=])?'  # Dependency type: ^, ~, =
    '(?P<major>[0-9]+|[*])'  # Major version (can be *)
    '(.(?P<minor>[0-9]+|[*]))?'  # Minor version
    '(.(?P<patch>[0-9]+|[*]))?'  # Patch version
    '([+\\-].*)?$'  # Any semver values beyond patch version
)


class VersionParseError(Exception):
    """Error that is returned when parsing a version fails."""


def prepare_staging(args):
    """Prepare staging directory."""
    sdir = args.staging_dir
    dirs = [
        os.path.join(sdir, 'ebuild', 'dev-rust'),
        os.path.join(sdir, 'crates')
    ]
    for d in dirs:
        os.makedirs(d, exist_ok=True)


def load_metadata(manifest_path):
    """Run cargo metadata and get metadata for build."""
    cwd = os.path.dirname(manifest_path)
    cmd = [
        'cargo', 'metadata', '--format-version', '1', '--manifest-path',
        manifest_path
    ]
    output = subprocess.check_output(cmd, cwd=cwd)

    return json.loads(output)


def get_crate_path(package, staging_dir):
    """Get path to crate in staging directory."""
    return os.path.join(
        staging_dir, 'crates', '{}-{}.crate'.format(package['name'],
                                                    package['version']))


def get_clean_crate_name(package):
    """Clean up crate name to {name}-{major}.{minor}.{patch}."""
    return '{}-{}.crate'.format(package['name'],
                                get_clean_package_version(package))


def version_to_tuple(name, version, missing=-1):
    """Extract dependency type and semver from a given version string."""
    def version_to_int(num):
        if not num or num == '*':
            return missing
        return int(num)

    m = re.match(VERSION_RE, version)
    if not m:
        print('{} failed to parse dep version: {}'.format(name, version))
        raise VersionParseError

    dep = m.group('dep')
    major = m.group('major')
    minor = m.group('minor')
    patch = m.group('patch')

    has_star = any([x == '*' for x in [major, minor, patch]])

    major = version_to_int(major)
    minor = version_to_int(minor)
    patch = version_to_int(patch)

    if has_star:
        dep = '~'
    elif not dep:
        dep = '^'

    return (dep, major, minor, patch)


def get_clean_package_version(package):
    """Get package version in the format {major}.{minor}.{patch}."""
    (_, major, minor, patch) = version_to_tuple(package['name'],
                                                package['version'],
                                                missing=0)
    return '{}.{}.{}'.format(major, minor, patch)


def get_ebuild_path(package, staging_dir, make_dir=False):
    """Get path to ebuild in given directory."""
    ebuild_path = os.path.join(
        staging_dir, 'dev-rust', package['name'],
        '{}-{}.ebuild'.format(package['name'],
                              get_clean_package_version(package)))

    if make_dir:
        os.makedirs(os.path.dirname(ebuild_path), exist_ok=True)

    return ebuild_path


def download_package(package, staging_dir):
    """Download the crate from crates.io."""
    dl_uri = CRATE_DL_URI.format(PN=package['name'], PV=package['version'])
    crate_path = get_crate_path(package, staging_dir)

    # Already downloaded previously
    if os.path.isfile(crate_path):
        return

    ret = subprocess.run(['curl', '-L', dl_uri, '-o', crate_path],
                   stdout=subprocess.DEVNULL,
                   stderr=subprocess.DEVNULL).returncode

    if ret:
        print('{} failed to download: {}'.format(dl_uri, ret))


def get_description(package):
    """Get a description of the crate from metadata."""
    # pylint: disable=invalid-string-quote
    if package.get('description', None):
        desc = package['description'].replace('`', '\'').replace('"', '\'')
        return desc.rstrip('\n')

    return ''


def get_homepage(package):
    """Get the homepage of the crate from metadata or use crates.io."""
    if package.get('homepage', None):
        return package['homepage']

    return 'https://crates.io/crates/{}'.format(package['name'])


def convert_license(cargo_license, package):
    """Convert licenses from cargo to a format usable in ebuilds."""
    cargo_license = '' if not cargo_license else cargo_license
    has_or = ' OR ' in cargo_license
    delim = ' OR ' if has_or else '/'

    found = cargo_license.split(delim)
    licenses_or = []
    for f in found:
        if f in LICENSES:
            licenses_or.append(LICENSES[f])

    if not licenses_or:
        print('{} is missing an appropriate license: {}'.format(
            package['name'], license))
        return "$(die 'Please replace with appropriate license')"

    if len(licenses_or) > 1:
        lstr = '|| ( {} )'.format(' '.join(licenses_or))
    else:
        lstr = licenses_or[0]

    return lstr


def convert_dependencies(dependencies, package, optional_packages):
    """Convert dependencies from metadata into the ebuild format."""
    def caret_to_ebuild(info):
        prefix = '>=dev-rust/{name}-{major}.{minor}.{patch}:='.format(**info)
        suffix = '<dev-rust/{name}-{major_1}'.format(**info)

        if info['minor'] == -1:
            prefix = '>=dev-rust/{name}-{major}:='.format(**info)
        elif info['patch'] == -1:
            prefix = '>=dev-rust/{name}-{major}.{minor}:='.format(**info)

        if info['major'] == 0:
            suffix = '<dev-rust/{name}-{major}.{minor_1}'.format(**info)

        return '{} {}'.format(prefix, suffix)

    def tilde_to_ebuild(info):
        prefix = '>=dev-rust/{name}-{major}.{minor}.{patch}:='.format(**info)
        suffix = '<dev-rust/{name}-{major}.{minor_1}'.format(**info)
        ebuild = '{} {}'.format(prefix, suffix)

        if info['minor'] == -1:
            ebuild = '=dev-rust/{name}-{major}*:='.format(**info)
        elif info['patch'] == -1:
            ebuild = '=dev-rust/{name}-{major}.{minor}*:='.format(**info)

        return ebuild

    def dep_to_ebuild(name, dep_type, major, minor, patch):
        info = {
            'name': name,
            'major': major,
            'minor': minor,
            'patch': patch,
            'major_1': major + 1,
            'minor_1': minor + 1,
            'patch_1': patch + 1
        }

        if dep_type == '^':
            return caret_to_ebuild(info)
        if dep_type == '~':
            return tilde_to_ebuild(info)

        # Remaining dep type is =
        return '=dev-rust/{name}-{major}.{minor}.{patch}:='.format(**info)

    deps = []
    for dep in dependencies:
        # Skip all dev dependencies. We will still handle normal and build deps.
        if dep.get('kind', None) == 'dev':
            continue

        # Optional dependencies get empty packages created
        if dep.get('optional', None):
            optional_packages[dep['name']] = {
                    'name': dep['name'],
                    'version': dep['req'],
                    'features': dep['features'],
            }

        # Convert version requirement to ebuild DEPEND.
        try:
            # Convert requirement to version tuple
            (deptype, major, minor, patch) = version_to_tuple(dep['name'], dep['req'])
            ebuild = dep_to_ebuild(dep['name'], deptype, major, minor, patch)
            deps.append('\t{}'.format(ebuild))
        except VersionParseError:
            # Rarely dependencies look something like ">=0.6, <0.8"
            deps.append("\t$(die 'Please replace with proper DEPEND: {} = {}')".format(dep['name'], dep['req']))

    if not deps:
        return ''

    # Add DEPEND= into template with all dependencies
    # RDEPEND="${DEPEND}" required for race in cros-rust
    fmtstring = 'DEPEND="\n{}\n"\nRDEPEND="${{DEPEND}}"\n'
    return fmtstring.format('\n'.join(deps))


def package_ebuild(package, ebuild_dir, optional_packages):
    """Create ebuild from metadata and write to ebuild directory."""
    ebuild_path = get_ebuild_path(package, ebuild_dir, make_dir=True)

    autogen_notice = AUTOGEN_NOTICE

    # Check if version matches clean version or modify the autogen notice
    if package['version'] != get_clean_package_version(package):
        autogen_notice = '\n'.join([
            autogen_notice,
            '# ${{PV}} was changed from the original {}'.format(
                package['version'])
        ])

    dependencies = convert_dependencies(package['dependencies'], package,
                                        optional_packages)
    template_info = {
        'copyright_year': datetime.now().year,
        'description': get_description(package),
        'homepage': get_homepage(package),
        'license': convert_license(package['license'], package),
        'dependencies': dependencies,
        'autogen_notice': autogen_notice,
    }

    with open(ebuild_path, 'w') as ebuild:
        ebuild.write(EBUILD_TEMPLATE.format(**template_info))


def upload_gsutil(package, staging_dir, no_upload=False):
    """Upload crate to distfiles."""
    if no_upload:
        return

    crate_path = get_crate_path(package, staging_dir)
    crate_name = get_clean_crate_name(package)
    ret = subprocess.run([
        'gsutil', 'cp', '-a', 'public-read', crate_path,
        'gs://chromeos-localmirror/distfiles/{}'.format(crate_name)
    ]).returncode

    if ret:
        print('{} failed to upload to chromeos-localmirror: {}'.format(
            crate_name, ret))


def update_ebuild(package, args, ebuild_dir, target_dir):
    """Update ebuild with generated one and generate MANIFEST."""
    ebuild_src = get_ebuild_path(package, ebuild_dir)
    ebuild_dest = get_ebuild_path(package, target_dir, make_dir=True)

    # Do not overwrite existing ebuilds unless explicity asked to.
    if args.overwrite_existing_ebuilds or not os.path.exists(ebuild_dest):
        shutil.copy(ebuild_src, ebuild_dest)
    else:
        print('ebuild {} already exists, skipping.'.format(ebuild_dest))

    # Generate manifest w/ ebuild digest
    ret = subprocess.run(['ebuild', ebuild_dest, 'digest']).returncode
    if ret:
        print('ebuild {} digest failed: {}'.format(ebuild_dest, ret))


def process_package(package, args, optional_packages):
    """Process each package listed in the metadata."""
    staging_dir = args.staging_dir
    ebuild_dir = os.path.join(staging_dir, 'ebuild')
    target_dir = args.target_dir

    download_package(package, staging_dir)
    package_ebuild(package, ebuild_dir, optional_packages)

    if not args.dry_run and package['name'] not in args.skip:
        upload_gsutil(package, staging_dir, no_upload=args.no_upload)
        update_ebuild(package, args, ebuild_dir, target_dir)

def process_empty_package(empty_package, args):
    """Process packages that should generate empty ebuilds."""
    staging_dir = args.staging_dir
    ebuild_dir = os.path.join(staging_dir, 'ebuild')
    target_dir = args.target_dir

    ebuild_src = get_ebuild_path(empty_package, ebuild_dir, make_dir=True)
    ebuild_dest = get_ebuild_path(empty_package, target_dir, make_dir=True)

    crate_features = ''
    if empty_package.get('features', None):
        crate_features = 'CROS_RUST_EMPTY_CRATE_FEATURES=( {} )'.format(
            ' '.join(['"{}"'.format(x) for x in empty_package['features']]))

    template_info = {
        'copyright_year': datetime.now().year,
        'crate_features': crate_features,
        'autogen_notice': AUTOGEN_NOTICE,
    }

    with open(ebuild_src, 'w') as ebuild:
        ebuild.write(EMPTY_CRATE.format(**template_info))

    if not args.dry_run and empty_package['name'] not in args.skip:
        shutil.copy(ebuild_src, ebuild_dest)

def check_if_package_is_required(empty_package, args):
    """Check if an empty package already has a non-empty ebuild."""
    ebuild_dest = get_ebuild_path(empty_package, args.target_dir, make_dir=False)
    ebuild_target_dir = os.path.dirname(ebuild_dest)

    # Find all ebuilds in ebuild_target dir and confirm they have
    # CROS_RUST_EMPTY_CRATE in them. If they do not, they need to be manually
    # handled.
    for root, _, files in os.walk(ebuild_target_dir):
        for efile in files:
            if not efile.endswith('.ebuild'):
                continue

            with open(os.path.join(root, efile), 'r') as ebuild:
                if 'CROS_RUST_EMPTY_CRATE' not in ebuild.read():
                    print('{} was not an empty crate.'.format(efile))
                    return True

    return False

def main(argv):
    """Convert dependencies from Cargo.toml into ebuilds."""
    args = parse_args(argv)

    prepare_staging(args)

    processed_packages = {}
    optional_packages = {}
    metadata = load_metadata(args.manifest_path)
    for p in metadata['packages']:
        process_package(p, args, optional_packages)
        processed_packages[p['name']] = True

    for key in optional_packages:
        try:
            if key not in processed_packages and not check_if_package_is_required(
                    optional_packages[key], args):
                process_empty_package(optional_packages[key], args)
        except VersionParseError:
            print('{} has a malformed version'.format(key))


def parse_args(argv):
    """Parse the command-line arguments."""
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        '-s',
        '--staging-dir',
        required=True,
        help='Staging directory for temporary and generated files')

    parser.add_argument(
        '-d',
        '--dry-run',
        action='store_true',
        help='Generate dependency tree but do not upload anything')

    parser.add_argument('-t',
                        '--target-dir',
                        help='Path to chromiumos-overlay')

    parser.add_argument('-k',
                        '--skip',
                        action='append',
                        help='Skip these packages when updating ebuilds')
    parser.add_argument('-n',
                        '--no-upload',
                        action='store_true',
                        help='Skip uploading crates to distfiles')
    parser.add_argument('-x',
                        '--overwrite-existing-ebuilds',
                        action='store_true',
                        help='Skip uploading crates to distfiles')

    parser.add_argument('manifest_path',
                        help='Cargo.toml used to generate ebuilds.')

    args = parser.parse_args(argv)

    # Require target directory if not dry run
    if not args.target_dir and not args.dry_run:
        raise Exception('Target directory must be set unless dry-run is True.')

    if not args.skip:
        args.skip = []

    return args


if __name__ == '__main__':
    sys.exit(main(sys.argv[1:]))