GitSwitch/main.py

import os
import glob
import json

import argparse
from typing import List

from github import Github
import pygit2

from git import Repo

import requests

import logging
logger = logging.getLogger('log')
logger.setLevel(logging.INFO)

fh = logging.FileHandler('repos_log.log')
fh.setLevel(logging.INFO)
logger.addHandler(fh)


def parse_args():
    """
    Parses command line arguments.

    :return:
    """
    parser = argparse.ArgumentParser()

    parser.add_argument('--token', type=str,
                        help='GitHub access token')

    parser.add_argument('--gitflic_token', type=str,
                        help='GitFlic access token')

    parser.add_argument('--dst_folder', type=str, default='./cloned-repos',
                        help='A folder to clone repositories into.')

    parser.add_argument('--is_private', type=str,
                        help='Sets the mode for copying repositories.'
                             'True - copies only private repositories'
                             'False - copies only public repositories')
    return parser.parse_args()


def insert_token(clone_url: str,
                 token: str,
                 proto='https://'):
    """
    Inserts a token into a repo address.

    :param clone_url: url to insert token into
    :param token: GitHub access token
    :param proto: protocol
    :return: clone_url with token inserted
    """
    assert clone_url.startswith(proto)
    # "https://{token}@github.com/{username}/{repo}.git"
    authed_clone_url = clone_url.replace(proto, proto + token + '@')
    logger.info(f'Modified repo url: {authed_clone_url}')
    return authed_clone_url


def get_org_name(repo):
    """
    Parses organization name from a repo object.

    :param repo: GitHub repository
    :return: repo organization name if org is filled or 'default'
    """
    if not repo.organization:
        return 'default'
    return repo.organization.login

def get_description(repo):
    """
    Parses description from a repo object.

    :param repo: GitHub repository
    :return: repo empty description if description is empty
    """
    if not repo.organization:
        return ''
    return repo.organization.description

def clone_repos(repos,
                token: str,
                dst_dir: str,
                gitflic_token: str,
                is_private: bool):
    """
    Clone all repos(private and public) of a token holder.
    :param repos: GitHub repositories
    :param token: GitHub access token
    :param dst_dir: directory to clone repos into
    :return:
    """

    callbacks = pygit2.RemoteCallbacks(pygit2.UserPass(token, 'x-oauth-basic'))
    github_clonned = ''

    if os.path.exists("github_clonned.txt"):
        file = open("github_clonned.txt", "r")
        github_clonned = file.read()
        file.close()

    file = open("github_clonned.txt", "a+")



    for i, repo in enumerate(repos):
        id = repo.id
        name = repo.name
        private = repo.private
        org = get_org_name(repo)
        description = get_description(repo)
        language = repo.language


        if str(id) in github_clonned:
            logger.info(f'Repository {org}/{name} already copied')
            continue

        if is_private is not None:
            if is_private.lower() == 'true' and not private:
                continue

            if is_private.lower() == 'false' and private:
                continue

        json_data = {
                   "title": f"{org}-{name}",
                   "description": f"{description}",
                   "alias": f"{org}-{name}",
                   "language": f"{language}",
                   "private": "true"
               }

        # Создаем репозиторий на гитфлике
        gitflic_repo = requests.post(
            'http://localhost:8047/project',
            headers = {
                "Authorization": f"token {gitflic_token}",
                "Content-Type": "application/json"
            },
            data = json.dumps(json_data)
        )

        gitflic_url = gitflic_repo.json().get("sshTransportUrl")

        local_path = os.path.join(dst_dir, org, name)
        os.makedirs(local_path)
        authed_clone_url = insert_token(repo.clone_url, token)

        logger.info(f'Cloning: {org}/{name} into {local_path}')
        github_repo = Repo.clone_from(authed_clone_url, local_path)
        remote = github_repo.create_remote("gitflic", url = gitflic_url)
        remote.push(refspec='--all')

        file.write(str(id) + '\n')


def upload_repos(local_repos: List[str]):
    raise NotImplementedError


if __name__ == '__main__':
    args = parse_args()

    token = args.token
    gitflic_token = args.gitflic_token
    dst_folder = args.dst_folder
    is_private = args.is_private

    g = Github(token)
    user = g.get_user()
    repos = user.get_repos()

    for i, repo in enumerate(repos):
        logger.info('Existing repos:')
        logger.info(f'ORG: {get_org_name(repo)} - REPO: {repo.name}')

    clone_repos(repos, token, dst_folder, gitflic_token, is_private)

    local_repos = glob.glob(dst_folder + '/*/*')
    # all repos are cloned
    assert i + 1 == len(local_repos)

    upload_repos(local_repos)