mirror of
https://gitflic.ru/project/dbi471/git-switch.git
synced 2025-07-02 02:05:25 +00:00
Optimize imports
This commit is contained in:
SantaSpeen
parent
b7f16c3669
commit
3f5d58ef52
2
main.py
2
main.py
@ -8,7 +8,7 @@
|
||||
"""
|
||||
|
||||
# Типы. Используется для <того, что бы тебя любило IDE>.
|
||||
from typing import NoReturn, Union, Tuple
|
||||
from typing import NoReturn, Union
|
||||
|
||||
# Удобная и встроенная в Python библиотека логирования. Подробнее: https://docs.python.org/3.10/library/logging.html
|
||||
import logging
|
||||
|
||||
220
main_old.py
220
main_old.py
@ -1,220 +0,0 @@
|
||||
import os
|
||||
import glob
|
||||
import json
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
|
||||
from github import Github
|
||||
|
||||
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)
|
||||
|
||||
console = logging.StreamHandler(sys.stdout)
|
||||
logging.getLogger().addHandler(console)
|
||||
|
||||
|
||||
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: str):
|
||||
"""
|
||||
Clone all repos(private and public) of a token holder.
|
||||
:param is_private:
|
||||
:param repos: GitHub repositories
|
||||
:param token: GitHub access token
|
||||
:param dst_dir: directory to clone repos into
|
||||
:return:
|
||||
"""
|
||||
|
||||
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
|
||||
|
||||
logger.info(f'\nCopying {name}:')
|
||||
|
||||
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:
|
||||
logger.info('Repository is public, skipping...')
|
||||
continue
|
||||
|
||||
if is_private.lower() == 'false' and private:
|
||||
logger.info('Repository is private, skipping...')
|
||||
continue
|
||||
|
||||
try:
|
||||
logger.info(f'Creating directory to copy.')
|
||||
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)
|
||||
except Exception:
|
||||
logger.info(f'Error copying repository from github. Probably repository has already been copied locally.')
|
||||
continue
|
||||
|
||||
json_data = {
|
||||
"title": f"{org}-{name}",
|
||||
"description": f"{description}",
|
||||
"alias": f"{org}-{name}",
|
||||
"language": f"{language}",
|
||||
"private": "true"
|
||||
}
|
||||
|
||||
try:
|
||||
logger.info(f'Creating repository {org}-{name} on gitflic.')
|
||||
# Создаём репозиторий на гитфлике
|
||||
gitflic_repo = requests.post(
|
||||
'http://localhost:8047/project', # Это кто ??
|
||||
headers={
|
||||
"Authorization": f"token {gitflic_token}",
|
||||
"Content-Type": "application/json"
|
||||
},
|
||||
data=json.dumps(json_data)
|
||||
)
|
||||
|
||||
if gitflic_repo.status_code == 429:
|
||||
logger.info(f'Request limit exceeded.')
|
||||
sys.exit()
|
||||
|
||||
except Exception:
|
||||
logger.info(f'Error while creating repository on gitflic.')
|
||||
continue
|
||||
|
||||
try:
|
||||
logger.info(f'Geting ssh link to repository...')
|
||||
gitflic_url = gitflic_repo.json().get("sshTransportUrl")
|
||||
|
||||
logger.info(f'Establishing a connection to the repository...')
|
||||
remote = github_repo.create_remote("gitflic", url=gitflic_url)
|
||||
except Exception:
|
||||
logger.info(f'Error creating connection to gitflic repository.')
|
||||
continue
|
||||
|
||||
try:
|
||||
logger.info(f'Pushing files into repository...')
|
||||
remote.push(refspec='--all')
|
||||
|
||||
file.write(str(id) + '\n')
|
||||
logger.info(f'Repository {org}-{name} successfully cloned.')
|
||||
except Exception:
|
||||
logger.info(f'Error while pushing files into gitflic repository.')
|
||||
continue
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
args = parse_args()
|
||||
|
||||
github_token = args.token
|
||||
gitflic_token = args.gitflic_token
|
||||
dst_folder = args.dst_folder
|
||||
is_private = args.is_private
|
||||
|
||||
g = Github(github_token)
|
||||
user = g.get_user()
|
||||
github_repos = user.get_repos()
|
||||
|
||||
logger.info('Existing repos:')
|
||||
for repo in github_repos:
|
||||
logger.info(f'ORG: {get_org_name(repo)} - REPO: {repo.name}')
|
||||
|
||||
clone_repos(github_repos, github_token, dst_folder, gitflic_token, is_private)
|
||||
|
||||
local_repos = glob.glob(dst_folder + '/*/*')
|
||||
# all repos are cloned
|
||||
assert github_repos.totalCount == len(local_repos)
|
||||
@ -1,6 +1,4 @@
|
||||
GitPython==3.1.27
|
||||
pygit2==1.9.1
|
||||
PyGithub==1.55
|
||||
gitflic==0.11
|
||||
requests==2.27.1
|
||||
click~=8.0.4
|
||||
requests~=2.27.1
|
||||
PyGithub~=1.55
|
||||
gitflic~=0.11
|
||||
Loading…
x
Reference in New Issue
Block a user