forge

#!/usr/bin/env python3

from github import Github
import argh
from argh import arg
import os
import requests
import shutil
import subprocess

_organization = 'organization'
_user = 'user'

_public_subpath = 'public'
_private_subpath = 'private'
_forked_subpath = 'forked'
_starred_subpath = 'starred'

_repos_subpath = 'repos'
_gists_subpath = 'gists'

def _clone_repo(ssh_url, clone_path):
    'Clone a repo to a path and return True, or skip and return False if it already exists there.'
    if not os.path.exists(clone_path):
        print(f'Cloning {ssh_url}...')
        subprocess.check_call(['git', 'clone', ssh_url, clone_path])
        subprocess.check_call(['git', 'submodule', 'update', '--init', '--recursive'], cwd=clone_path)
        return True
    else:
        print(f'{ssh_url} already cloned')
        return False

def _remote_repo_exists(repo_ssh_url):
    p = subprocess.Popen(['git', 'ls-remote', '-h', repo_ssh_url], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    p.communicate()
    return p.returncode == 0

def _clone_wiki(repo, clone_path):
    'Clone the wiki associated with a repo, or skip if it already exists locally.'
    if repo.has_wiki:
        wiki_url = f'git@github.com:{repo.full_name}.wiki.git'
        if _remote_repo_exists(wiki_url):
            wiki_path = f'{clone_path}.wiki'
            if not os.path.exists(wiki_path):
                subprocess.check_call(['git', 'clone', wiki_url, wiki_path])
            else:
                print(f'{wiki_url} already cloned')

def _tag_repo(repo, clone_path, clear_first=False):
    'Set tags on the local repo directory with the language and topics from the GitHub repo, if present.'
    topics = repo.get_topics()
    if (topics is None or len(topics) == 0) and repo.language is None:
        return
    if topics is not None and len(topics) > 0:
        taglist = [topic.lower() for topic in topics]
    else:
        taglist = []
    if repo.language is not None:
        taglist.append(repo.language.lower())
    if clear_first:
        current_tags = subprocess.check_output(['tag', '--no-name', clone_path], encoding='utf-8').strip()
        subprocess.check_call(['tag', '-r', f'"{current_tags}"'])
    new_tags = ','.join({tag for tag in taglist})
    subprocess.check_call(['tag', '-a', f'"{new_tags}"', clone_path])

def _clone_nonforked_repo(repo, repo_type_path, no_wikis):
    'Clone a normal repo and any associated wiki.'
    clone_path = f'{repo_type_path}/{repo.name}'
    if _clone_repo(repo.ssh_url, clone_path):
        _tag_repo(repo, clone_path)
    if not no_wikis:
        _clone_wiki(repo, clone_path)

def _gist_ssh_url(gist):
    'Because gist entries from the GitHub API don\'t include an SSH url for cloning (but that does work in practice) this function takes the HTTPS address, pulls apart the necessary info, and reformulates it into an SSH Git URL.'
    gist_path = gist.git_pull_url.replace('https://gist.github.com/', '')
    return f'git@gist.github.com:{gist_path}'

def _is_kind_of_repo_to_sync(x, gist, repo, public, private, forked, starred):
    'Determine if the repo passed in as `x` should be synced based on its properties and the provided options.'
    if gist:
        return (
            x.name == repo.description
            and (
                (public and x.public)
                or (private and not x.public)
                or (forked and x.fork_of is not None)
            )
        )
    else:
        return (
            x.name == repo.name
            and (
                starred
                or (public and not x.private)
                or (private and x.private)
                or (forked and x.fork)
            )
        )

def _update_local_repos_under(path, remote_repo_list, push_to_fork_remotes, prune, pull_with_rebase, push_after_rebase, rebase_submodules, public=False, private=False, forked=False, gist=False, starred=False):
    'Given a path to a directory containing a collection of any type of locally cloned repos, and list of remote repositories, fetch and fast-forward commits from upstream. If it\'s a fork, pull (--ff-only or --rebase) from `fork` remote and then always pull --rebase from `upstream`, then optionally push to the remote downstream fork. If the local repo is no longer in the list of remote repos, optionally prune it by deleting the local clone.'
    if not os.path.exists(path):
        return
    for repo in os.scandir(path):
        repos_to_sync = [x for x in remote_repo_list if _is_kind_of_repo_to_sync(x, gist, repo, public, private, forked, starred)]
        repo_to_sync = next(iter(repos_to_sync), None)
        if repo_to_sync is not None:
            if forked:
                subprocess.check_call(f'git fetch fork && git pull {git_pull_option} fork', shell=True, cwd=repo.path)
                remote_name = 'upstream'
            else:
                remote_name = 'origin'
            print(f'syncing {repo.path}')
            git_pull_option = '--ff-only'
            if pull_with_rebase or forked:
                git_pull_option = '--rebase'
            subprocess.check_call(f'git fetch {remote_name} && git pull {git_pull_option} {remote_name}', shell=True, cwd=repo.path)
            submodule_update_args = ['git', 'submodule','--init', '--recursive']
            if rebase_submodules:
                submodule_update_args.append('--rebase')
            subprocess.check_call(submodule_update_args, cwd=repo.path)
            if (forked and push_to_fork_remotes) or (not forked and pull_with_rebase and push_after_rebase):
                subprocess.check_call(['git', 'push', remote_name], cwd=repo.path)
            _tag_repo(repo_to_sync, repo.path, clear_first=True)
        elif prune:
            print(f'pruning {repo.path}')
            shutil.rmtree(repo.path)

def _report_statuses_under(path):
    'Given a path to a directory containing a collection of any type of locally cloned repos, descend into each, fetch and report `git status`. Forks get a status for both the `fork` and `upstream` origins.'
    if not os.path.exists(path):
        return
    for repo in os.scandir(path):
        if forked:
            subprocess.check_call(f'git fetch fork && git fetch upstream && git status', shell=True, cwd=repo.path)
        else:
            subprocess.check_call(f'git fetch origin && git status', shell=True, cwd=repo.path)

@arg('base_path', help='Location of the repos for which to report statuses.')
@arg('--access_token', help='The GitHub access token of the GitHub user whose repos private repos should be reported in addition to public repos.')
def status(base_path, access_token=None):
    'Report on the statuses of all cloned repositories at `base_path`. By default only reports on public repositories; in order to also report on private repos, use the `--access_token` option.'
    g = Github(access_token)
    for account_type in os.scandir(base_path):
        if account_type.name == '.DS_Store':
            continue
        if account_type.name == _user:
            user_dir = f'{base_path}/{_user}'
            for user in os.scandir(user_dir):
                if user.name == '.DS_Store':
                    continue
                user_repos_path = f'{user_dir}/{user.name}/{_repos_subpath}'
                user_gists_path = f'{user_dir}/{user.name}/{_gists_subpath}'
                _report_statuses_under(f'{user_repos_path}/{_public_subpath}')
                _report_statuses_under(f'{user_gists_path}/{_public_subpath}')
                for forked_repo in os.scandir(f'{user_repos_path}/{_forked_subpath}'):
                    _report_statuses_under(forked_repo.path)
                for forked_gist in os.scandir(f'{user_gists_path}/{_forked_subpath}'):
                    _report_statuses_under(forked_gist.path)
                if access_token is not None:
                    authenticated_user = g.get_user()
                if authenticated_user is not None and authenticated_user == user.name:
                    account = g.get_user(user.name)
                    _report_statuses_under(f'{user_repos_path}/{_private_subpath}')
                    _report_statuses_under(f'{user_gists_path}/{_private_subpath}')
                    for starred_repo in os.scandir(f'{user_repos_path}/{_starred_subpath}'):
                        _report_statuses_under(starred_repo.path)
                    for starred_gist in os.scandir(f'{user_gists_path}/{_starred_subpath}'):
                        _report_statuses_under(starred_gist.path)
        elif account_type.name == _organization:
            for org in os.scandir(f'{base_path}/{_organization}'):
                if org.name == '.DS_Store':
                    continue
                org_repos_path = f'{org.path}/{_repos_subpath}'
                org_gists_path = f'{org.path}/{_gists_subpath}'
                _report_statuses_under(f'{org_repos_path}/{_public_subpath}')
                _report_statuses_under(f'{org_gists_path}/{_public_subpath}')
                for forked_repo in os.scandir(f'{org_repos_path}/{_forked_subpath}'):
                    _report_statuses_under(forked_repo.path)
                _report_statuses_under(f'{org_repos_path}/{_private_subpath}')
                _report_statuses_under(f'{org_gists_path}/{_private_subpath}')
        else:
            print(f'unexpected account type directory: {account_type}; expected either `{_user}` or `{_organization}`')

@arg('access_token', help='The GitHub access token of the GitHub user whose repos should be synced.')
@arg('base_path', help='Location of the repos to sync.')
@arg('--push_to_fork_remotes', help='After fast-forwarding any new commits from forks\' remote upstreams, push the new commits to fork remotes.', default=False)
@arg('--prune', help='If a local repository is no longer listed from the server, remove its local clone.')
@arg('--pull_with_rebase', help='Run `git pull --rebase` to rebase any local commits on top of the remote HEAD. (By default, `sync` runs `git pull --ff-only`.)', default=False)
@arg('--push_to_remotes', help='If --pull_with_rebase is provided, push HEAD to remote after rebasing any local commits on top of pulled remote commits.', default=False)
@arg('--rebase_submodules', help='If development has occurred in a submodule, the changes are rebased onto any updated submodule commit hash that is pulled down as part of updating the superproject. See `man git-submodule` -> update -> rebase.)')
def sync(access_token, base_path, push_to_fork_remotes=False, prune=False, pull_with_rebase=False, push_to_remotes=False, rebase_submodules=False):
    'Sync repos that have been previously cloned for all local users and organizations, by fetching and fast-forwarding to pick up any new commits. Forks pull from `fork` remote first, then try to rebase HEAD on top of any new `upstream` remote commits. Update the tags associated with cloned repos, as those may change over time. Will only be able to work with private repos for the user authenticated with the provided access token; other users\' will only have public repos updated, so this will need to be run once for each user whose private and starred repos/gists should be synced, using their access tokens in turn.'
    g = Github(access_token)
    authenticated_user = g.get_user()
    for account_type in os.scandir(base_path):
        if account_type.name == '.DS_Store':
            continue
        if account_type.name == _user:
            user_dir = f'{base_path}/{_user}'
            for user in os.scandir(user_dir):
                if user.name == '.DS_Store':
                    continue
                account = g.get_user(user.name)
                remote_repos = account.get_repos()
                remote_gists = account.get_gists()
                user_repos_path = f'{user_dir}/{user.name}/{_repos_subpath}'
                user_gists_path = f'{user_dir}/{user.name}/{_gists_subpath}'
                _report_statuses_under(f'{user_repos_path}/{_public_subpath}', remote_repos, push_to_fork_remotes, pull_with_rebase, push_after_rebase, prune, rebase_submodules, public=True)
                _update_local_repos_under(f'{user_gists_path}/{_public_subpath}', remote_gists, push_to_fork_remotes, pull_with_rebase, push_after_rebase, prune, rebase_submodules, public=True, gist=True)
                for forked_repo in os.scandir(f'{user_repos_path}/{_forked_subpath}'):
                    _update_local_repos_under(forked_repo.path, remote_repos, push_to_fork_remotes, pull_with_rebase, push_after_rebase, prune, rebase_submodules, forked=True)
                for forked_gist in os.scandir(f'{user_gists_path}/{_forked_subpath}'):
                    _update_local_repos_under(forked_gist.path, remote_gists, push_to_fork_remotes, pull_with_rebase, push_after_rebase, prune, rebase_submodules, forked=True, gist=True)
                if authenticated_user == user.name:
                    remote_starred_repos = account.get_starred()
                    remote_starred_gists = account.get_starred_gists()
                    _update_local_repos_under(f'{user_repos_path}/{_private_subpath}', remote_repos, push_to_fork_remotes, pull_with_rebase, push_after_rebase, prune, rebase_submodules, private=True)
                    _update_local_repos_under(f'{user_gists_path}/{_private_subpath}', remote_gists, push_to_fork_remotes, pull_with_rebase, push_after_rebase, prune, rebase_submodules, private=True, gist=True)
                    for starred_repo in os.scandir(f'{user_repos_path}/{_starred_subpath}'):
                        _update_local_repos_under(starred_repo.path, remote_starred_repos, push_to_fork_remotes, pull_with_rebase, push_after_rebase, prune, rebase_submodules, starred=True)
                    for starred_gist in os.scandir(f'{user_gists_path}/{_starred_subpath}'):
                        _update_local_repos_under(starred_gist.path, remote_starred_gists, push_to_fork_remotes, pull_with_rebase, push_after_rebase, prune, rebase_submodules, gist=True, starred=True)
        elif account_type.name == _organization:
            for org in os.scandir(f'{base_path}/{_organization}'):
                if org.name == '.DS_Store':
                    continue
                account = g.get_organization(org.name)
                remote_repos = account.get_repos()
                remote_public_gists = account.public_gists
                remote_private_gists = account.private_gists
                org_repos_path = f'{org.path}/{_repos_subpath}'
                org_gists_path = f'{org.path}/{_gists_subpath}'
                _update_local_repos_under(f'{org_repos_path}/{_public_subpath}', remote_repos, push_to_fork_remotes, pull_with_rebase, push_after_rebase, prune, rebase_submodules, public=True)
                _update_local_repos_under(f'{org_gists_path}/{_public_subpath}', remote_public_gists, push_to_fork_remotes, pull_with_rebase, push_after_rebase, prune, rebase_submodules, public=True, gist=True)
                for forked_repo in os.scandir(f'{org_repos_path}/{_forked_subpath}'):
                    _update_local_repos_under(forked_repo.path, remote_repos, push_to_fork_remotes, pull_with_rebase, push_after_rebase, prune, rebase_submodules, forked=True)
                _update_local_repos_under(f'{org_repos_path}/{_private_subpath}', remote_repos, push_to_fork_remotes, pull_with_rebase, push_after_rebase, prune, rebase_submodules, private=True)
                _update_local_repos_under(f'{org_gists_path}/{_private_subpath}', remote_private_gists, push_to_fork_remotes, pull_with_rebase, push_after_rebase, prune, rebase_submodules, private=True, gist=True)
        else:
            print(f'unexpected account type directory: {account_type}; expected either `{_user}` or `{_organization}`')

@arg('access_token', help='The GitHub access token of the GitHub user whose repos should be synced.')
@arg('base_path', help='Location of the repos to sync.')
@arg('--no_public_repos', help='Do not clone the authenticated user\'s or organization\'s public repos.', default=False)
@arg('--no_private_repos', help='Do not clone the authenticated user\'s or organization\'s private repos.', default=False)
@arg('--no_starred_repos', help='Do not clone the authenticated user\'s or organization\'s starred repos (does not apply to organizations).', default=False)
@arg('--no_forked_repos', help='Do not clone the authenticated user\'s or organization\'s forked repos.', default=False)
@arg('--no_public_gists', help='Do not clone the authenticated user\'s or organization\'s public gists.', default=False)
@arg('--no_private_gists', help='Do not clone the authenticated user\'s or organization\'s private gists.', default=False)
@arg('--no_starred_gists', help='Do not clone the authenticated user\'s starred gists (does not apply to organizations).', default=False)
@arg('--no_forked_gists', help='Do not clone the authenticated user\'s forked gists (does not apply to organiations).', default=False)
@arg('--no_wikis', help='Do not clone wikis associated with any repos that are cloned.', default=False)
@arg('--no_repos', help='Do not clone any repos.', default=False)
@arg('--no_gists', help='Do not clone any gists.', default=False)
@arg('--organization', help='Instead of fetching the list of the authenticated user\'s repos, fetch the specified organzation\'s. This means there will be no starred repos/gists.', default=None, type=str)
@arg('--dedupe_org_repos_owned_by_user', help='If a user created a repo also owned by an organization, then running `clone` for both the user and org would result in two copies of that repo. This option avoids cloning any repos owned by an organization from being cloned for a user.', default=False)
def clone(access_token, base_path, no_public_repos=False, no_private_repos=False, no_starred_repos=False, no_forked_repos=False, no_public_gists=False, no_private_gists=False, no_starred_gists=False, no_forked_gists=False, no_wikis=False, no_repos=False, no_gists=False, organization=None, dedupe_org_repos_owned_by_user=False):
    'Clone the public, private, starred and forked repos and gists, as well as any associated wikis, unless disabled with options, from the specified account managed by the provided access token, or optionally from a specified organization (which won\'t have starred repos or any gists).'
    g = Github(access_token)
    if organization is not None:
        account = g.get_organization(organization)
        account_type = _organization
        repos = account.get_repos()
    else:
        account = g.get_user()
        account_type = _user
        repos = account.get_repos(affiliation='owner')
    if not no_repos:
        repos_path = f'{base_path}/{account_type}/{account.login}/{_repos_subpath}'
        fork_path = f'{repos_path}/{_forked_subpath}'
        star_path = f'{repos_path}/{_starred_subpath}'
        public_path = f'{repos_path}/{_public_subpath}'
        private_path = f'{repos_path}/{_private_subpath}'
        subprocess.check_call(['mkdir', '-p', fork_path, public_path, private_path])
        if organization is None and not no_starred_repos:
            subprocess.check_call(['mkdir', '-p', star_path])
        for repo in repos:
            if repo.organization is not None and organization is None and dedupe_org_repos_owned_by_user:
                continue
            if repo.fork:
                if no_forked_repos:
                    continue
                r = g.get_repo(repo.full_name)
                clone_path = f'{fork_path}/{r.parent.owner.login}/{repo.name}'
                if _clone_repo(repo.ssh_url, clone_path):
                    subprocess.check_call(['git', 'remote', 'rename', 'origin', 'fork'], cwd=clone_path)
                    if not _remote_repo_exists(r.parent.ssh_url):
                        continue
                    subprocess.check_call(['git', 'remote', 'add', 'upstream', r.parent.ssh_url], cwd=clone_path)
                    default_branch = subprocess.check_output('git rev-parse --abbrev-ref fork/HEAD | cut -c6-', shell=True, encoding='utf-8', cwd=clone_path).strip()
                    subprocess.check_call(['git', 'config', '--unset', f'branch.{default_branch}.remote'], cwd=clone_path)
                    subprocess.check_call(['git', 'config', '--add', f'branch.{default_branch}.remote', 'upstream'], cwd=clone_path)
                    subprocess.check_call(['git', 'config', '--add', f'branch.{default_branch}.pushRemote', 'fork'], cwd=clone_path)
                    _tag_repo(r.parent, clone_path)
                    if not no_wikis:
                        _clone_wiki(r.parent, clone_path)
            elif repo.private:
                if no_private_repos:
                    continue
                _clone_nonforked_repo(repo, private_path, no_wikis)
            else:
                if no_public_repos:
                    continue
                _clone_nonforked_repo(repo, public_path, no_wikis)
        if organization is None and not no_starred_repos:
            for repo in account.get_starred():
                _clone_nonforked_repo(repo, f'{star_path}/{repo.owner.login}', no_wikis)
    if organization is not None: # orgs don't have gists, so we can skip all the logic to clone them
        return
    if no_gists:
        return
    gists_path = f'{base_path}/{account_type}/{account.login}/{_gists_subpath}'
    public_gists_path = f'{gists_path}/{_public_subpath}'
    private_gists_path = f'{gists_path}/{_private_subpath}'
    starred_gists_path = f'{gists_path}/{_starred_subpath}'
    forked_gists_path = f'{gists_path}/{_forked_subpath}'
    subprocess.check_call(['mkdir', '-p', public_gists_path, private_gists_path, starred_gists_path])
    for gist in account.get_gists():
        if not no_forked_gists and gist.fork_of is not None:
            upstream = g.get_gist(gist.fork_of.id)
            _clone_repo(_gist_ssh_url(upstream), f'{forked_gists_path}/{upstream.owner.login}/{gist.description}')
        else:
            gist_ssh_url = _gist_ssh_url(gist)
            if not no_public_gists and gist.public:
                _clone_repo(gist_ssh_url, f'{public_gists_path}/{gist.description}')
            elif not no_private_gists:
                _clone_repo(gist_ssh_url, f'{private_gists_path}/{gist.description}')
    if not no_starred_gists:
        for gist in account.get_starred_gists():
            gist_ssh_url = _gist_ssh_url(gist)
            _clone_repo(gist_ssh_url, f'{starred_gists_path}/{gist.owner.login}/{gist.description}')

parser = argh.ArghParser()
parser.add_commands([clone, status, sync])

if __name__ == '__main__':
    parser.dispatch()