view env/lib/python3.9/site-packages/planemo/github_util.py @ 0:4f3585e2f14b draft default tip

"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"

"""Utilities for interacting with Github."""
from __future__ import absolute_import

import io
import os
import stat
import tarfile
import tempfile
from distutils.dir_util import copy_tree
from pathlib import Path

import requests

from planemo import git
from planemo.io import (
    communicate,
    IS_OS_X,
)

try:
    import github
    has_github_lib = True
except ImportError:
    github = None
    has_github_lib = False

GH_VERSION = "1.5.0"

NO_GITHUB_DEP_ERROR = ("Cannot use github functionality - "
                       "PyGithub library not available.")
FAILED_TO_DOWNLOAD_GH = "No gh executable available and it could not be installed."
DEFAULT_REMOTE_NAME = 'planemo-remote'


def get_github_config(ctx, allow_anonymous=False):
    """Return a :class:`planemo.github_util.GithubConfig` for given configuration."""
    global_github_config = _get_raw_github_config(ctx)
    return GithubConfig(global_github_config, allow_anonymous=allow_anonymous)


def clone_fork_branch(ctx, target, path, remote_name=DEFAULT_REMOTE_NAME, **kwds):
    """Clone, fork, and branch a repository ahead of building a pull request."""
    git.checkout(
        ctx,
        target,
        path,
        branch=kwds.get("branch", None),
        remote="origin",
        from_branch="master"
    )
    if kwds.get("fork"):
        try:
            fork(ctx, path, remote_name=remote_name, **kwds)
        except Exception:
            pass
    if 'GITHUB_USER' in os.environ:
        # On CI systems fork doesn't add a local remote under circumstances I don't quite understand,
        # but that's probably linked to https://github.com/cli/cli/issues/2722
        cmd = ['git', 'remote', 'add', remote_name, f"https://github.com/{os.environ['GITHUB_USER']}/{os.path.basename(target)}"]
        try:
            communicate(cmd, cwd=path)
        except RuntimeError:
            # Can add the remote only once
            pass
    return remote_name


def fork(ctx, path, remote_name=DEFAULT_REMOTE_NAME, **kwds):
    """Fork the target repository using ``gh``."""
    gh_path = ensure_gh(ctx, **kwds)
    gh_env = get_gh_env(ctx, path, **kwds)
    cmd = [gh_path, "repo", "fork", '--remote=true', '--remote-name', remote_name]
    communicate(cmd, cwd=path, env=gh_env)
    return remote_name


def get_or_create_repository(ctx, owner, repo, dry_run=True, **kwds):
    """Clones or creates a repository and returns path on disk"""
    target = os.path.realpath(tempfile.mkdtemp())
    remote_repo = "https://github.com/{owner}/{repo}".format(owner=owner, repo=repo)
    try:
        ctx.log('Cloning {}'.format(remote_repo))
        git.clone(ctx, src=remote_repo, dest=target)
    except Exception:
        ctx.log('Creating repository {}'.format(remote_repo))
        target = create_repository(ctx, owner=owner, repo=repo, dest=target, dry_run=dry_run)
    return target


def create_repository(ctx, owner, repo, dest, dry_run, **kwds):
    gh_path = ensure_gh(ctx, **kwds)
    gh_env = get_gh_env(ctx, dry_run=dry_run, **kwds)
    cmd = [gh_path, 'repo', 'create', '-y', '--public', "{owner}/{repo}".format(owner=owner, repo=repo)]
    if dry_run:
        "Would run command '{}'".format(" ".join(cmd))
        git.init(ctx, dest)
        return dest
    communicate(cmd, env=gh_env, cwd=dest)
    return os.path.join(dest, repo)


def rm_dir_contents(directory, ignore_dirs=(".git")):
    directory = Path(directory)
    for item in directory.iterdir():
        if item.name not in ignore_dirs:
            if item.is_dir():
                rm_dir_contents(item)
            else:
                item.unlink()


def add_dir_contents_to_repo(ctx, from_dir, target_dir, target_repository_path, version, dry_run, notes=""):
    ctx.log("From {} to {}".format(from_dir, target_repository_path))
    rm_dir_contents(target_repository_path)
    copy_tree(from_dir, target_repository_path)
    git.add(ctx, target_repository_path, target_repository_path)
    message = "Update for version {version}".format(version=version)
    if notes:
        message += "\n{notes}".format(notes=notes)
    git.commit(ctx, repo_path=target_repository_path, message=message)
    if not dry_run:
        git.push(ctx, target_repository_path)


def assert_new_version(ctx, version, owner, repo):
    remote_repo = "https://github.com/{owner}/{repo}".format(owner=owner, repo=repo)
    try:
        tags_and_versions = git.ls_remote(ctx, remote_repo=remote_repo)
        if "refs/tags/v{}".format(version) in tags_and_versions or "refs/tags/{}".format(version) in tags_and_versions:
            raise Exception("Version '{}' for {}/{} exists already. Please change the version.".format(version, owner, repo))
    except RuntimeError:
        # repo doesn't exist
        pass


def changelog_in_repo(target_repository_path):
    changelog = []
    for path in os.listdir(target_repository_path):
        if 'changelog.md' in path.lower():
            header_seen = False
            header_chars = ('---', '===', '~~~')
            with(open(os.path.join(target_repository_path, path))) as changelog_fh:
                for line in changelog_fh:
                    if line.startswith(header_chars):
                        if header_seen:
                            return "\n".join(changelog[:-1])
                        else:
                            header_seen = True
    return "\n".join(changelog)


def create_release(ctx, from_dir, target_dir, owner, repo, version, dry_run, notes="", **kwds):
    assert_new_version(ctx, version, owner=owner, repo=repo)
    target_repository_path = get_or_create_repository(ctx, owner=owner, repo=repo, dry_run=dry_run)
    add_dir_contents_to_repo(ctx, from_dir, target_dir, target_repository_path, version=version, dry_run=dry_run, notes=notes)
    gh_path = ensure_gh(ctx, **kwds)
    gh_env = get_gh_env(ctx, dry_run=dry_run, **kwds)
    cmd = [
        gh_path,
        'release',
        '-R',
        "{}/{}".format(owner, repo),
        'create',
        "v{version}".format(version=version),
        '--title',
        str(version),
    ]
    cmd.extend(['--notes', notes or changelog_in_repo(target_repository_path)])
    if not dry_run:
        communicate(cmd, env=gh_env)
    else:
        ctx.log("Would run command '{}'".format(" ".join(cmd)))


def pull_request(ctx, path, message=None, repo=None, **kwds):
    """Create a pull request against the origin of the path using ``gh``."""
    gh_path = ensure_gh(ctx, **kwds)
    gh_env = get_gh_env(ctx, path, **kwds)
    cmd = [gh_path, "pr", "create"]
    if message is None:
        cmd.append('--fill')
    else:
        lines = message.splitlines()
        cmd.extend(['--title', lines[0]])
        if len(lines) > 1:
            cmd.extend(["--body", "\n".join(lines[1:])])
    if repo:
        cmd.extend(['--repo', repo])
    communicate(cmd, env=gh_env)


def get_gh_env(ctx, path=None, dry_run=False, **kwds):
    """Return a environment dictionary to run gh with given user and repository target."""
    if path is None:
        env = {}
    else:
        env = git.git_env_for(path).copy()
    if not dry_run:
        github_config = _get_raw_github_config(ctx)
        if github_config is not None:
            if "access_token" in github_config:
                env["GITHUB_TOKEN"] = github_config["access_token"]

    return env


def ensure_gh(ctx, **kwds):
    """Ensure gh is available for planemo

    This method will ensure ``gh`` is installed at the correct version.

    For more information on ``gh`` checkout https://cli.github.com/
    """
    planemo_gh_path = os.path.join(ctx.workspace, f"gh-{GH_VERSION}")
    if not os.path.exists(planemo_gh_path):
        _try_download_gh(planemo_gh_path)

    if not os.path.exists(planemo_gh_path):
        raise Exception(FAILED_TO_DOWNLOAD_GH)

    return planemo_gh_path


def _try_download_gh(planemo_gh_path):
    link = _gh_link()
    path = Path(planemo_gh_path)
    resp = requests.get(link)
    with tarfile.open(fileobj=io.BytesIO(resp.content)) as tf, path.open('wb') as outfile:
        for member in tf.getmembers():
            if member.name.endswith('bin/gh'):
                outfile.write(tf.extractfile(member).read())
    path.chmod(path.stat().st_mode | stat.S_IEXEC)


def _get_raw_github_config(ctx):
    """Return a :class:`planemo.github_util.GithubConfig` for given configuration."""
    if "github" not in ctx.global_config:
        if "GITHUB_TOKEN" in os.environ:
            return {
                "access_token": os.environ["GITHUB_TOKEN"],
            }
    if "github" not in ctx.global_config:
        raise Exception("github account not found in planemo config and GITHUB_TOKEN environment variables unset")
    return ctx.global_config["github"]


class GithubConfig(object):
    """Abstraction around a Github account.

    Required to use ``github`` module methods that require authorization.
    """

    def __init__(self, config, allow_anonymous=False):
        if not has_github_lib:
            raise Exception(NO_GITHUB_DEP_ERROR)
        if "access_token" not in config:
            if not allow_anonymous:
                raise Exception("github authentication unavailable")
            github_object = github.Github()
        else:
            github_object = github.Github(config["access_token"])
        self._github = github_object


def _gh_link():
    if IS_OS_X:
        template_link = "https://github.com/cli/cli/releases/download/v%s/gh_%s_macOS_amd64.tar.gz"
    else:
        template_link = "https://github.com/cli/cli/releases/download/v%s/gh_%s_linux_amd64.tar.gz"
    return template_link % (GH_VERSION, GH_VERSION)


def publish_as_gist_file(ctx, path, name="index"):
    """Publish a gist.

    More information on gists at http://gist.github.com/.
    """
    github_config = get_github_config(ctx, allow_anonymous=False)
    user = github_config._github.get_user()
    with open(path, "r") as fh:
        content = fh.read()
    content_file = github.InputFileContent(content)
    gist = user.create_gist(False, {name: content_file})
    return gist.files[name].raw_url


def get_repository_object(ctx, name):
    github_object = get_github_config(ctx, allow_anonymous=True)
    return github_object._github.get_repo(name)


__all__ = (
    "add_dir_contents_to_repo",
    "clone_fork_branch",
    "create_release",
    "ensure_gh",
    "fork",
    "get_github_config",
    "get_gh_env",
    "get_or_create_repository",
    "publish_as_gist_file",
)