initial commit

core/__init__.py

@@ -0,0 +1,4 @@
+import requests
+from requests.packages.urllib3.exceptions import InsecureRequestWarning
+
+requests.packages.urllib3.disable_warnings(InsecureRequestWarning)  # disable ssl warning

core/argument_parser.py

@@ -0,0 +1,55 @@
+from argparse import ArgumentParser
+
+GITLAB_URL = 'https://git.do.x5.ru'
+
+USAGE = '''github_mirror [-h] [-g GROUP] (-u URLS [URLS ...] | -f FILE) -t TOKEN
+--------------------------------------------------
+
+python3 github_mirror -u "https://github.com/s3rius/FastAPI-template.git" -g 2059
+
+python3 github_mirror -u "https://github.com/s3rius/FastAPI-template.git" "https://github.com/sqlalchemy/sqlalchemy.git"
+
+python3 github_mirror -f github_mirrors.txt -g 59563
+
+python3 github_mirror -f github_mirrors.txt -u "https://github.com/s3rius/FastAPI-template.git"
+
+python3 gitlab_mirror.py --gitlab "https://gitlab.company.ru" -t "git-QwertY1245kde" -g 2059
+
+
+--------------------------------------------------
+'''
+
+
+def create_parser() -> ArgumentParser:
+    """
+    Create argparse parser
+
+    :return: command parser
+    """
+    parser = ArgumentParser(
+        prog='github_mirror',
+        description='''Script to add mirror repo into gitlab''',
+        epilog='''the developer is not responsible for the operation of the script :)''',
+        add_help=True,
+        usage=USAGE
+    )
+
+    parser.add_argument('-g', '--group', required=False, type=int,
+                        help='Add group id it can be found under group name. Id must be integer')
+
+    parser.add_argument('-u', '--urls', nargs='+', help='Provide url or urls to mirror with it in format: '
+                                                        'https://github.com/s3rius/FastAPI-template.git. '
+                                                        'You can provide multiple urls separated by space. '
+                                                        'Names will generate automatically from links')
+
+    parser.add_argument('-f', '--file', help='Add file with urls. Each url on new line. Can be combined with '
+                                             '--url option. Names will generate automatically from links')
+
+    parser.add_argument('-t', '--token', required=True,
+                        help='Access token to gitlab API. More information: https://docs.gitlab.com/ee/user/profile/'
+                             'personal_access_tokens.html#create-a-personal-access-token')
+
+    parser.add_argument('-l', '--gitlab', required=False, default=GITLAB_URL,
+                        help=f'Provide gitlab url. Default link {GITLAB_URL}')
+
+    return parser

core/repo_creator.py

@@ -0,0 +1,104 @@
+from typing import Union
+
+import requests
+from requests import Response
+
+from core.utils import logger
+
+
+class RepositoryCreator:
+
+    def __init__(self, gitlab_url: str, headers: dict):
+        self.gitlab_url = gitlab_url
+        self.headers = headers
+        self.HTTP_201_CREATED = 201
+        self.HTTP_200_OK = 200
+
+    def __gitlab_request(self, method: str, url: str, data: dict = None) -> Union[Response, None]:
+        """
+        Create request to gitlab
+
+        :param method: Request method can be changed
+        :param url: Url to request
+        :param data: Provide request data
+        :return: Response object on None
+        """
+        try:
+            request = requests.request(method, url, headers=self.headers, json=data, verify=False)
+            return request
+        except Exception as err:
+            logger.error(f'Connection not established. Check vpn is connected! \n{err}')
+
+    def __create_new_project(self, url: str, group_id: int = None) -> Union[str, None]:
+        """
+        Create new project in gitlab with name based on provided url
+
+        :param url: github url to mirror with:
+        :param group_id: namespace in gitlab to combine repos
+        :return: repo_id as string or None if any error
+        """
+
+        # name of repository will generate automatically from link
+        name = url.split('/')[-1].replace('.git', '')
+        git_data = {'name': name}
+        if group_id:
+            git_data['namespace_id'] = group_id
+
+        request = self.__gitlab_request('POST', f'{self.gitlab_url}/api/v4/projects', git_data)
+        try:
+            if request.status_code == self.HTTP_201_CREATED:
+                repo_data = request.json()
+                name_with_namespace = repo_data.get('name_with_namespace', None)
+                if name_with_namespace:
+                    logger.info(f'Repository {name_with_namespace} has been created')
+                else:
+                    logger.info(f'Repository {repo_data["name"]} has been created')
+                return repo_data['id']
+            else:
+                logger.error(f'Cant create new project. Status code: {request.status_code}. Reason: {request.text}')
+        except AttributeError:
+            pass
+
+    def __add_pull_mirror(self, url: str, repo_id: str) -> Union[str, None]:
+        """
+        Add pull mirror to Settings -> Repository -> Mirroring repositories
+
+        :param url: github url to mirror with
+        :param repo_id: id of repository which will be updated
+        :return: github url which will be mirrored
+        """
+
+        if repo_id:
+            git_data = {"mirror": True, "import_url": url}
+            request = self.__gitlab_request('PUT', f'{self.gitlab_url}/api/v4/projects/{repo_id}', git_data)
+            if request and request.status_code == self.HTTP_200_OK:
+                return url
+            elif request.status_code != self.HTTP_200_OK:
+                logger.error(f'Cant add mirror url to project. Status code: {request.status_code}. '
+                             f'Reason: {request.text}')
+
+    def __pull_github_repo(self, url: str, repo_id: str):
+        """
+        Initiate pull request for gitlab repository
+
+        :param url: github url to mirror with
+        :param repo_id: id of repository which will be updated
+        """
+
+        if repo_id:
+            request = self.__gitlab_request('POST', f'{self.gitlab_url}/api/v4/projects/{repo_id}/mirror/pull')
+            if request and request.status_code == self.HTTP_200_OK:
+                logger.info(f'Repository: {url} has been pulled')
+            elif request.status_code != self.HTTP_200_OK:
+                logger.error(f'Error pull repository. Status code: {request.status_code}. Reason: {request.text}')
+
+    def create_repository_mirror(self, **kwargs):
+        """
+        Base action for one thread. Creates repository, add mirror url and triggers pull at te end
+
+        :param kwargs: Can contain github url to mirror of, gitlab group ID
+        """
+        repo_id = self.__create_new_project(kwargs['url'], kwargs['group_id'])
+        url = self.__add_pull_mirror(kwargs['url'], repo_id)
+        if url:
+            self.__pull_github_repo(url, repo_id)

core/utils.py

@@ -0,0 +1,46 @@
+import importlib.util
+import logging
+import sys
+import time
+from collections import Counter
+from threading import Thread
+from typing import List
+
+# use loguru if it is possible for color output
+if importlib.util.find_spec('loguru') is not None:
+    from loguru import logger
+    logger.remove()
+    logger.add(sink=sys.stdout, colorize=True, level='DEBUG',
+               format="<cyan>{time:DD.MM.YYYY HH:mm:ss}</cyan> | <level>{level}</level> | "
+                      "<magenta>{message}</magenta>")
+
+# use standard logging
+else:
+    logger = logging.getLogger()
+    logger.setLevel(logging.INFO)
+
+    console_handler = logging.StreamHandler()
+    console_handler.setLevel(logging.INFO)
+    log_formatter = logging.Formatter("%(asctime)s | %(levelname)s | %(message)s")
+    console_handler.setFormatter(log_formatter)
+
+    logger.addHandler(console_handler)
+
+
+def threads_ready_statistic(threads: List[Thread]):
+    """
+    Getting information how many threads are running right now
+
+    :param threads: List of active threads
+    """
+    while True:
+        threads_statistic = [thread.is_alive() for thread in threads]
+        statistic = Counter(threads_statistic)
+        ready_count = statistic.get(False, 0)
+        percent = int(ready_count / len(threads) * 100)
+        time.sleep(1)
+        if 0 < percent < 100:
+            logger.info(f'Ready: {percent}%')
+        if not any(threads_statistic):
+            logger.info(f'Ready: 100%')
+            break