git-hub

#!/usr/bin/env python
import subprocess
import github
import yaml
import click
import sys
import textwrap
import toml
import os
from datetime import datetime
from os.path import join as pjoin
import ansi


def login():
    """Retrieves the user's remote.
    For example consider running 'git remote -v' and getting back:
        origin  git@github.com:test_user/test_repo.git (fetch)
        origin git@github.com:test_user/test_repo.git (push)

    Returns
    -------
    username : string
        username of the remote, i.e. test_user
    repo : string
        origin repository of the github user, i.e. test_repo
    remotes : string
        Git remotes: the output of 'git remote -v'

    """
    process = subprocess.Popen(["git", "remote", "-v"], stdout=subprocess.PIPE)
    remotes = str(process.stdout.read())
    url = remotes.split(" ", 1)[0]  # gets the fetch url
    arguments = url.split(".com")[1]  # gets just the username/repo.git
    arguments = arguments[1:]
    arguments = arguments.split(".git")[0]  # takes out ".git"
    username, repo = arguments.split("/")
    return (username, repo, remotes)


def get_token():
    """Retrieve token from configuration file.

    Returns
    -------
    token : string
        a number corresponding to user's authentication
    """
    token = ""
    if os.path.isfile(os.path.expanduser("~/.config/git-hub.yaml")):
        with open(os.path.expanduser("~/.config/git-hub.yaml")) as stream:
            yaml_file = str(yaml.load(stream))
            token = yaml_file.split("=")[1].strip()
    else:
        token = os.environ['GITHUB_TOKEN']

    if (token is not ""):
        return token
    else:
        print(textwrap.dedent("""\
            No authentication token specified in: ~/.config/git-hub.yaml

            Please see

              https://help.github.com/articles/creating-a-personal-access-token-for-the-command-line/

            for instruction on obtaining a token. Then update the configuration
            file as follows:

              token = abc123
        """))
        sys.exit(1)


def pr(num):
    """Pulls down and checkout the branch of the pr.

    This will run the following commands in the terminal:
    "git remote add user git@github.com:user/repo",
    "git fetch user",
    "git checkout -b pr/num user/branch"

    Parameters
    ----------
    num : integer
        The number of the pull request.
    """
    username, repo, remotes = login()
    token = get_token()

    # gets pr and runs command.
    try:
        g = github.Github(token)
        pr = g.get_user(username).get_repo(repo).get_pull(num)
        label = pr.head.label
        other_user, branch = label.split(":")
        remote_name = "\\n" + other_user + "\\t"
        if remote_name not in remotes:
            p = subprocess.Popen(["git", "remote", "add", other_user,
                                  f'git@github.com:{other_user}/{repo}'])
            p.communicate()
        p = subprocess.Popen(["git", "fetch", other_user],
                             stdout=subprocess.PIPE)
        p.stdout.read()
        subprocess.Popen(["git", "checkout", "-b",
                         f'pr/{num}', f'{other_user}/{branch}'])

    except github.BadCredentialsException as e:
        print(e)
        print("The authentification token is not valid.")
        sys.exit(1)

    except github.UnknownObjectException as e:
        print(e)
        print(f'There is no pr with number {num}.')
        sys.exit(1)


def push():
    """Pushes changes back to a branch.

    This will run the following command:
    "git push user pr/num:branch"

    """
    p = subprocess.Popen(["git", "branch"], stdout=subprocess.PIPE)
    pr = str(p.stdout.read())
    pr = pr.split("*", 1)[1].split()[0]
    pr = pr.replace("\\n", "").replace(" ", "").replace("'", "").split("/")[1]
    try:
        path_prs = path_to_toml()
        f = open(path_prs, "r")
        pr_dict = toml.load(f)
    except (OSError, IOError) as e:
        # if fails, then sync and retry
        sync()
        push()

    user = pr_dict['open pull requests'][pr]['user']
    branch = pr_dict['open pull requests'][pr]["branch"]
    p = subprocess.Popen(["git", "push", user,
                         f'pr/{pr}:{branch}'], stdout=subprocess.PIPE)
    p.stdout.read()


def path_to_git():
    """Finds path to .git folder
    """
    path_repo = os.path.abspath('.')
    # if not in directory with .git, keep going back to find file
    while (os.path.abspath(path_repo) != '/'
           and not os.path.isdir(pjoin(path_repo, '.git'))):
        path_repo = pjoin(path_repo, '..')
    path_git = pjoin(path_repo, ".git")
    return path_git


def path_to_toml():
    """Finds path to pull-requests.toml
    """
    path_git = path_to_git()
    path_github = pjoin(path_git, 'git-hub')
    path_prs = pjoin(path_github, 'pull-requests.toml')
    return path_prs


def find_in_dictionary(pr_id, pr_data):
    """Fetches all the information on a certain PR

    Parameters
    ----------
    pr_id : integer
        the pull request's number
    pr_data : dictionary
        dictionary of open or closed PRs to find pr_id in
    """
    if(pr_data is not None):
        for pr_nr in pr_data:
            if(pr_nr == str(pr_id)):
                return pr_nr, pr_data[pr_nr]
    return None, None


def get_info(pr_data):
    """Fetches all the information on a certain PR

    Parameters
    ----------
    pr_data : integer
        number of PR to fetch information from
    """

    try:
        path_prs = path_to_toml()
        f = open(path_prs, "r")
        pr_dict = toml.load(f)  # fetches toml file and creates a dictionary
        open_dict = pr_dict['open pull requests']
        closed_dict = pr_dict['closed pull requests']
    except (OSError, IOError) as e:
        sync()
        get_info(pr_data)

    open_PR = "O"
    key, pr_dictionary = find_in_dictionary(pr_data, open_dict)
    if(pr_dictionary is None):
        open_PR = "C"
        key, pr_dictionary = find_in_dictionary(pr_data, closed_dict)
    if(pr_dictionary is None):
        click.echo(f"Could not find PR #{pr_data}."
                   "Run 'git hub sync' and try again.")
    else:
        reviewers = pr_dictionary.get('reviewers', 'None')
        assignee = pr_dictionary.get('assignee', 'None')
        milestone = pr_dictionary.get('milestone', 'None')
        labels = pr_dictionary.get('labels', 'None')
        branch = pr_dictionary.get('branch', 'None')
        commits = pr_dictionary.get('commits', 'None')
        most_recent_comment = pr_dictionary.get('most_recent_comment', 'None')
        comment_count = pr_dictionary.get('comment_count', 'None')
        created_at = pr_dictionary.get('created_at', 'None')
        updated_at = pr_dictionary.get('updated_at', 'None')
        click.echo(f'{key} {ansi.GREEN} {open_PR} {ansi.CLOSE}'
                   f'{pr_dictionary["user"]}/{pr_dictionary["branch"]}'
                   f' {pr_dictionary["comment"]}')
        click.echo(f'-Created at: {created_at}  -Updated at: {updated_at}')
        click.echo(f'-Commits: {commits}  -Comment count: {comment_count}'
                   f'-Most recent comment: {most_recent_comment}')
        click.echo(f'-Labels: {labels}  -Reviewers: {reviewers}'
                   f'-Branch: {branch}  -Assignees: {assignee}'
                   f'-Milestones: {milestone}')


def parse_time(time):
    """
    Converts time into python datetime object

    Parameters
    ----------
    time : String
        Time that git-hub gives
    """
    date = time.split('T')[0]
    time = time.split('T')[1][:-1]
    year, date = date.split('-', 1)
    month, date = date.split('-', 1)
    day = date
    hour, time = time.split(':', 1)
    minute, time = time.split(':', 1)
    second = time
    d = datetime(int(year), int(month),
                 int(day), int(hour), int(minute), int(second))
    return d


def print_in_order(dict, increasing=False):
    """
    Prints items in q sorted by order given by sort

    Parameters
    ----------
    Q : PriorityQueue
        items to be sorted, currently in increasing order
    inreasing : boolean
        tells if the user wants dates in order of most recent first
    """
    if(increasing):
        sorted_list = sorted(dict)
        for x in sorted_list:
            click.echo(f"{dict[x]} : {x[:10]}")
    else:
        sorted_list = sorted(dict, reverse=True)
        for x in sorted_list:
            click.echo(f"{dict[x]} : {x[:10]}")


def find_match(open_or_closed, list_of_dictionaries,
               keyword, user, comment, number, branch, label):
    """Helper function that finds all matches in the
       given dictionary that fits the specified parameters.

    Parameters
    ----------
    open_or_closed : boolean
        Whether looking in closed PR or open PR
    list_of_dictionaries : dictionary
        pull-requests.toml information mapping PR number to info
    keyword : string
        Searches if any part of user, branch, comment, or number match
    user : string
        Search by PR username
    comment : string
        Search by PR comment
    number : string
        Search by PR number
    branch : string
        Search by PR branch

    Returns
    ----------
    appeared_before : boolean
        Whether the search match is the first match
    all_prs : list
        All the prs that match the criteria
    """
    appeared_before = False
    all_prs = {}
    for dictionary in list_of_dictionaries:
        for sub_keys in list(dictionary.keys()):
            sub_dict = dictionary[sub_keys]
            if keyword:         # checks keyword
                concate = " ".join(sub_dict.values())
                if keyword.upper() in concate.upper():
                    appeared_before = True
                    if(open_or_closed):
                        all_prs[sub_dict['modified']] = (
                            f'{sub_keys} '
                            f'{ansi.GREEN} "O" {ansi.CLOSE}'
                            f'{sub_dict["user"]}/{sub_dict["branch"]}'
                            f' {sub_dict["comment"]}')
                    else:
                        all_prs[sub_dict['modified']] = (
                            f'{sub_keys} {ansi.RED} "C"'
                            f'{ansi.CLOSE} {sub_dict["user"]}/'
                            f'{sub_dict["branch"]} {sub_dict["comment"]}')
            else:               # checks for all other parameters
                temp_user = sub_dict['user']
                temp_branch = sub_dict["branch"]
                temp_comment = sub_dict["comment"]
                temp_label = sub_dict["labels"]
                if (label.upper() in temp_label.upper() and comment.upper()
                        in temp_comment.upper() and user.upper()
                        in temp_user.upper() and branch.upper()
                        in temp_branch.upper()):
                    appeared_before = True
                    if(open_or_closed):
                        all_prs[sub_dict['modified']] = (
                            f'{sub_keys} {ansi.GREEN} "O"'
                            f'{ansi.CLOSE} {sub_dict["user"]}/'
                            f'{sub_dict["branch"]} {sub_dict["comment"]}')
                    else:
                        all_prs[sub_dict['modified']] = (
                            f'{sub_keys} {ansi.RED} "C" '
                            f'{ansi.CLOSE} {sub_dict["user"]}/'
                            f'{sub_dict["branch"]} {sub_dict["comment"]}')
    return appeared_before, all_prs


def search(keyword, user, comment, number,
           branch, opened_or_closed, label, sort):
    """Searches open and closed pull request comments for specified keyword.

    Opens pull-requests.toml file in .git folder to fetch pull requests.

    Parameters
    ----------
    keyword : string
        Searches if any part of user, branch, comment, or number match
    user : string
        Search by PR username
    comment : string
        Search by PR comment
    number : string
        Search by PR number
    branch : string
        Search by PR branch
    """
    try:
        open_prs = False
        closed_prs = False
        path_prs = path_to_toml()
        f = open(path_prs, "r")
        pr_dict = toml.load(f)  # fetches toml file and creates a dictionary
    except (OSError, IOError) as e:
        # if pull-requests.toml hasnt been created yet calls
        # sync and then retries to fetch
        sync()
        search(keyword, user, comment, number,
               branch,  opened_or_closed, label, sort)

    open_dict = pr_dict['open pull requests']
    closed_dict = pr_dict['closed pull requests']
    if(not opened_or_closed or opened_or_closed.lower() == 'open'):
        open_prs, q = find_match(True, open_dict, keyword,
                                 user, comment, number, branch, label)
        print_in_order(q, sort.startswith('i'))
    if(not opened_or_closed or opened_or_closed.lower() == 'closed'):
        closed_prs, q = find_match(False, closed_dict, keyword,
                                   user, comment, number, branch, label)
        print_in_order(q, sort.startswith('i'))
    # outputs if keyword was not contained in pull requests comments
    if not open_prs and not closed_prs:
        click.echo(f"Could not find in pull requests. Update your "
                   f"pull requests with 'git hub sync' and try again.")


def find_pr_info(pr, repo, token, open_or_closed):
    """Helper function that finds all the information we want to record from the
    pull requests from the API and converts it to toml format syntax

    Parameters
    ----------
    pr : object
        PR dictionary
    repo : object
        github object of the repo
    token : string
        authentication token for user
    open_or_closed : string
        if the PR is open or closed
    """
    to_write = ""
    comment_dict = pr.get_comments()
    labels_dict = repo.get_issue(pr.number).labels
    reviews = [reviewer.login for reviewer in
               list(pr.get_review_comments())]
    assignees = [assignee.login
                 for assignee in list(pr.assignees)]
    labels = [label.name for label in list(labels_dict)]
    comment_dates = [c.updated_at for c in list(comment_dict)]

    to_write = f"""
    ['{open_or_closed} pull requests'.{pr.number}]\n
    title = "{repr(pr.title)}" \n
    body = "{repr(pr.body)}" \n
    user = "{pr.user.login}" \n
    branch = "{pr.head.ref}" \n
    mergeable = "{pr.mergeable}" \n
    comment = "{repr(pr.title)}" \n
    commits = "{repr(pr.commits)}" \n
    modified = "{pr.updated_at}" \n
    created_at = "{pr.created_at}" \n
    updated_at = "{pr.updated_at}" \n
    closed_at = "{pr.closed_at}" \n
    merged_at = "{pr.merged_at}" \n
    reviews = {repr(reviews)} \n
    assignees = {assignees} \n
    milestone = "{pr.milestone.title if pr.milestone else ""}"\n
    labels = {labels}\n
    self_comment = "{any([c.user.login == pr.user.login
                    for c in list(comment_dict)])}"\n
    comment_dates = {comment_dates}\n
    most_recent = '{max(comment_dates) if comment_dates  else ""}'\n
    comment_content = {repr([c.body for c in list(comment_dict)])}\n
    comment_count = '{len(list(comment_dict))}'\n
    """
    return textwrap.dedent(to_write)


def find_issue_info(issue, token):
    """Helper function that finds all the information
       we want to record from the issues
       from the API and converts it to toml format syntax.

    Parameters
    ----------
    issue : object
        github object that is an issue
    token : string
        authentication token for user
    """

    comment_dict = issue.get_comments()
    comment_dates = [c.updated_at for c in list(comment_dict)]
    comment_content = [c.body for c in list(comment_dict)]
    to_write = f"""
    ['issues'.{issue.number}]\n
    title = "{issue.title}"\n
    number = "{issue.number}"\n
    body = '''{repr(issue.body)}'''\n
    self_comment = "{any([c.user.login == issue.user.login
                    for c in list(comment_dict)])}"\n
    comment_dates = "{comment_dates}"\n
    most_recent = '{max(comment_dates) if comment_dates  else ""}'\n
    comment_content = {repr(comment_content)}\n
    comment_count = '{len(list(comment_dict))}'\n
    """
    return textwrap.dedent(to_write)


def sync():
    """Updates and saves pull-requests
       in pull-requests.toml in the .git folder.
    """
    username, repo, remotes = login()
    token = get_token()

    # saves pr into toml file
    try:
        g = github.Github(token)
        path_git = path_to_git()
        path_github = pjoin(path_git, 'git-hub')
        path_prs = path_to_toml()
        if not os.path.isdir(path_github):
            os.makedirs(path_github)
        f = open(path_prs, "w")
        to_write = "['open pull requests']\n"
    except github.BadCredentialsException as e:
        print(e)
        click.echo("The authentification token is not valid.")
        sys.exit(1)

    # creates dictionaries in toml format as we scan though pull requests
    github_repo = g.get_user(username).get_repo(repo)
    open_prs = g.get_user(username).get_repo(repo).get_pulls("open")
    for pr in open_prs:
        to_write = to_write + find_pr_info(pr, github_repo, token, "open")
    to_write = to_write[:-1]+'\n'
    to_write = to_write + "['closed pull requests'] \n"
    closed_prs = g.get_user(username).get_repo(repo).get_pulls("closed")
    for pr in closed_prs:
        to_write = to_write + find_pr_info(pr, github_repo, token, "closed")

    issues = g.get_user(username).get_repo(repo).get_issues()
    to_write = to_write + "['issues'] \n"
    for issue in issues:
        if f"pull requests'.{issue.number}" not in to_write:
            to_write = to_write + find_issue_info(issue, token)
    # converts it to toml and stores in file
    toml_string = toml.loads(to_write)
    toml.dump(toml_string, f)


@click.group()
def cli():
    pass


@cli.command()
@click.argument("command", default="")
@click.argument("args", nargs=-1)
@click.option('--opened', '-o', default="",
              help="search by open or closed PRs")
@click.option('--user', '-u', default="", help="search by user")
@click.option('--branch', '-b', default="", help="search by branch")
@click.option('--comment', '-c', default="", help="search by comment")
@click.option('--number', '-n', default="", help="search by PR number")
@click.option('--sort', '-s', default="",
              help="sort PRs by increasing or decreasing date")
@click.option('--label', '-l', default="", help="search PR by label")
def hub(command, args, user, comment, number, branch, sort, opened, label):
    if command == "pr":
        pr_num = int(args[0])
        pr(pr_num)

    elif command == "push":
        push()

    elif command == "search":
        if(not args):
            search(None, user, comment, number, branch, opened, label, sort)
        else:
            search(" ".join(args), user, comment,
                   number, branch, opened, label, sort)
    elif command == "sync":
        sync()

    elif command == "info":
        pr_num = int(args[0])
        get_info(pr_num)

    else:
        print("invalid command")
        sys.exit(1)


if __name__ == "__main__":
    hub()