# -*- coding: utf-8 -*- # noinspection GrazieInspection """ Программа для клонирования репозиториев с https://github.com/ на https://gitflic.ru Программа была написана: @dikola Программа была отрефакторена: @SantaSpeen """ # Удобная и встроенная в Python библиотека логирования. Подробнее: https://docs.python.org/3.10/library/logging.html import logging # Встроенные библиотеки. Нужны для некоторых функций программы. import os import sys import time # Типы. Используется для <того, что бы тебя любило IDE>. from typing import NoReturn, Union # Библиотека для работы с аргументами. Подробнее: https://click.palletsprojects.com/en/8.0.x import click # import requests # Библиотека для работы с git. Подробнее про систему git: https://habr.com/ru/post/588801/ from git import Repo # Библиотека для работы с Gitflic API. Подробнее: https://github.com/SantaSpeen/gitflic from gitflic import GitflicAuth, Gitflic # Библиотека для работы c GitHub API. Подробнее: https://github.com/PyGithub/PyGithub from github import Github, AuthenticatedUser, PaginatedList # Инициализируем логирование log_format = "%(asctime)s - %(name)-5s - %(levelname)-5s - %(message)s" logging.basicConfig(level=logging.INFO, format=log_format) log = logging.getLogger(name="GitSwitch") # Настройка логирование в файл. fh = logging.FileHandler('git-switch.log') fh.setLevel(logging.INFO) fh.setFormatter(logging.Formatter(log_format)) log.addHandler(fh) class GitSwitch: # noinspection PyTypeChecker def __init__(self, gf_token: str, gh_token: str, clone_folder: str, apply_private: bool, apply_organisations: bool, use_ssh: bool): self.gf_token = gf_token self.gh_token = gh_token self.clone_folder = clone_folder self.apply_private = apply_private self.apply_organisations = apply_organisations self.use_ssh = use_ssh self.gf: Gitflic = None self.gh: Github = None self.gh_user: AuthenticatedUser = None self.github_repos: PaginatedList = None self.gh_user_name = None self.get_login = None self.session: requests.Session = None def authorization(self) -> NoReturn: """ Авторизуемся и получаем список репозиториев из GitHub """ gf_session = GitflicAuth(self.gf_token) self.gf = Gitflic(gf_session) self.session = gf_session.session log.info(f"Logged into GitFlic as {self.gf.call('/user/me')['username']}") self.gh = Github(self.gh_token) self.gh_user = self.gh.get_user() log.info(f"Logged into Github as {self.gh_user.login}") self.github_repos = self.gh_user.get_repos() self.gh_user_name = self.gh_user.login self.get_login = lambda repo_info: repo_info.organization.login if repo_info.organization else self.gh_user_name def get_github_repo(self, repo_info) -> Union[Repo, None]: """ Получаем репозиторий из GitHub """ login = self.get_login(repo_info) name = repo_info.name path = os.path.join(self.clone_folder, login, name) proto = "https://" if not os.path.exists(path): clone_url = repo_info.clone_url.replace(proto, proto + self.gh_token + '@') log.info(f'Clone {login} : {name}; Clone path: {path};') return Repo.clone_from(clone_url, path) log.info(f"Path: {path} already exists.") return None def get_gitflic_repo(self, repo_info) -> Union[dict, None]: """ Создаём репозиторий на гитфлике """ login = self.get_login(repo_info) title = f"{login}-{repo_info.name}" if login != self.gh_user_name else repo_info.name config = { "title": title, "description": f"{repo_info.description}", "alias": title, "language": f"{repo_info.language}", "private": "true" } repo_object = self.session.post("https://api.gitflic.ru/project", json=config) code = repo_object.status_code if code != 200: log.error(f"GitFlic api send {repo_object.status_code} HTTP Error.") if code == 429: log.info("Waiting 10 seconds and try again.") time.sleep(10) return self.get_gitflic_repo(repo_info) log.warning("Skip repository.") return jsn = repo_object.json() log.info(f"Successfully created new empty repo: {jsn['httpTransportUrl'][:-3]}") return jsn @staticmethod def push_into_gitflic(repo, url) -> bool: """ Пушим репозиторий на GitFlic """ try: remote = repo.create_remote("gitflic", url=url) log.info(f"Pushing repository.") remote.push(refspec='--all') return True except Exception as e: log.error(f"Exception while pushing: {e}") return False def is_skip(self, repo_info) -> bool: if repo_info.private and not self.apply_private: return True if self.get_login(repo_info) != self.gh_user_name and not self.apply_organisations: return True return False def run(self) -> NoReturn: """ Запуск основной части """ for repo_info in self.github_repos: if self.is_skip(repo_info): continue github_repo = self.get_github_repo(repo_info) if not github_repo: continue gitflic_repo = self.get_gitflic_repo(repo_info) if not gitflic_repo: continue if self.push_into_gitflic( github_repo, gitflic_repo['sshTransportUrl' if self.use_ssh else 'httpTransportUrl'] ): log.info(f"Repository {self.get_login(repo_info)}/{repo_info.name} successfully cloned.") def start(self) -> NoReturn: self.authorization() i = j = 0 log.info("GitHub repositories:") for repo in self.github_repos: skip = self.is_skip(repo) if skip: j += 1 log.info(f'[SKIP] GitGub => {self.get_login(repo)} : {repo.name};') continue i += 1 log.info(f'GitGub => {self.get_login(repo)} : {repo.name}') log.info(f"Repositories found: {i + j}. Repositories to copy: {i}. Ignored repositories: {j}.") if input("Do you agree to copying these repositories to GitFlic? (y/n) ").lower() != "y": log.info("Stopped by the user.") exit(0) self.run() # Инициализируем наши аргументы @click.command() @click.option("--gf_token", help="Your GitFlic token.", type=str, required=True) @click.option("--gh_token", help="Your GitHub token.", type=str, required=True) @click.option("--clone_folder", help="Directory where to download repositories.", default="./cloned-repos", required=False, type=click.Path()) @click.option("--apply_private", help="Need to copy private repositories?", default=False, required=False, is_flag=True) @click.option("--apply_organisations", help="Need to copy organisations repositories?", default=False, required=False, is_flag=True) @click.option("--use_ssh", help="Use SSH mode to upload repositories.", default=False, required=False, is_flag=True) def main(**kwargs): log.info("New log start.") log.info(f"Local time: {time.asctime()}") log.info(f"GitSwitch start with: {sys.argv} arguments.") try: gs = GitSwitch(**kwargs) gs.start() except Exception as e: log.exception(f"GitSwitch send error: {e}") finally: log.info(f"Exiting at {time.asctime()}\n{'-----' * 20}") if __name__ == '__main__': main()