git-hub

#!/usr/bin/env python
import subprocess
import github
import yaml
import click
import sys
import textwrap
import toml
import app
import plot_pr
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 = 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(":")
        process = subprocess.Popen(["git", "remote", "-v"], stdout=subprocess.PIPE)
        remotes = str(process.stdout.read())
        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 = ""
    url = f'https://api.github.com/repos/{username}/{repo}/pulls/{pr.number}'
    request = Request(url)
    request.add_header('Authorization', 'token %s' % token)
    response = urlopen(request)
    response = json.loads(response.read().decode('utf-8'))
    issue_url = response.get('issue_url')
    labels = ""
    issue_request = Request(issue_url)
    issue_response = urlopen(issue_request)
    issue_response = json.loads(issue_response.read().decode('utf-8'))
    labels_dict = issue_response.get('labels')
    comment_url = response.get('comments_url')
    comment_request = Request(comment_url)
    comment_response = urlopen(comment_request)
    comment_dict = json.loads(comment_response.read().decode('utf-8'))

    reviews = [reviewer.get('login') for reviewer in
               list(response.get('requested_reviewers'))]
    assignees = [assignee.get('login')
                 for assignee in list(response.get('assignees'))]
    labels = [label.get('name') for label in list(labels_dict)]
    comment_dates = [c.get('updated_at') for c in list(comment_dict)]
    if(pr.number % 5 == 0):
        print("Syncing PR ", pr.number)
    to_write = f"""
    ['{open_or_closed} pull requests'.{pr.number}]\n
    title = "{repr(pr.title)}" \n
    url = "{pr.html_url}" \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)


def render():
    try:
        plot_pr.execute()
    except ImportError:
        print("Matplotlib not installed, cannot render image. Continuing without image.")
    app.main()


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


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


@cli.command()
@click.argument("command", default="")
@click.argument("args", nargs=-1)
@click.option('--open', '-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, open, 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)

    elif command == "render":
        render()

    elif command == "checkout":
        checkout(str(args[0]))

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


if __name__ == "__main__":
    hub()