Gogs projects cloner

README.md

Gogs project cloner

Usage

1. Create a backup user on your Gogs server
2. Add this backup user as read-only collaborator to the projects that you want to backup
3. Generate a token for this user (here)
4. Update configuration of the script (especially BACKUP_DIRECTORY, BACKUP_USERNAME, BACKUP_TOKEN, API_BASE_URL and GIT_HOST)
5. Run the script
6. (Optionally) add this script to your crontab or rsnapshot.conf

clone.py

#!/usr/bin/env python3
# -*- coding:Utf-8 -*-

"""Gogs projects cloner."""

import json
import os
import shutil
import subprocess
import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning

__author__ = 'Baptiste MOINE <[email protected]>'
__version__ = '0.1-dev'
__date__ = '30 October 2017'

PROXY = False
BACKUP_DIRECTORY = '/home/backup'
BACKUP_USERNAME = 'backup'
BACKUP_TOKEN = '722ec3d5a20fce12e655ff5cdff37f642148f3c9'
API_BASE_URL = 'https://git.cyber.lan/api/v1'
GIT_HOST = 'git.cyber.lan'

requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

def http_get_request(url, proxies):
    ##
    # HTTP Request Content
    #
    method = 'GET'  # HTTP method

    headers = {      # HTTP headers
        'User-Agent':       'Mozilla/5.0 (X11; Linux x86_64; rv:45.0) Gecko/20100101 Firefox/45.0',
        'Accept':           'text/html',
        'Accept-Language':  'en-US,en;q=0.5',
        'Accept-Encoding':  'gzip, deflate',
        'Connection':       'close'
    }

    try:
        ##
        # HTTP Request
        #
        req = requests.Request(  # Create the HTTP request with data and headers
            method=method,
            url=url,
            headers=headers
        )
        res = requests.Session()               # Create an HTTP session
        res.trust_env = False
        req = res.prepare_request(req)  # Prepare the request for send

        ##
        # HTTP Response
        #
        res = res.send(  # Send the prepared HTTP request
            req,
            verify=False,
            proxies=proxies,
            timeout=5,
            allow_redirects=False
        )

        headers = dict()

        for key, value in res.headers.items():
            headers[key] = value

        return {'headers': headers, 'text': res.text}
    except TypeError as e:
        pass

def git_clone(repo, dest=''):
    if dest == '': dest = '{repo}/'.format(repo=repo.split('/')[-1])
    # cwd = os.getcwd()
    with open(os.devnull, 'w') as nullfd:
        subprocess.call(['git',
                         'clone',
                         '--mirror',
                         'https://{token}@{git}/{repo}'.format(token=BACKUP_TOKEN,
                                                               git=GIT_HOST,
                                                               repo=repo),
                         dest], stdout=nullfd, stderr=nullfd)
    # os.chdir(dest)
    # subprocess.call(['git', 'config', '--bool', 'core.bare', 'false'])
    # os.chdir(cwd)
    # git_reset_checkout(repo)

def git_reset_checkout(repo):
    dest = '{repo}/'.format(repo=repo.split('/')[-1])
    cwd = os.getcwd()
    os.chdir(dest)
    with open(os.devnull, 'w') as nullfd:
        subprocess.call(['git', 'reset', '.'], stdout=nullfd, stderr=nullfd)
    with open(os.devnull, 'w') as nullfd:
        subprocess.call(['git', 'checkout', '.'], stdout=nullfd, stderr=nullfd)
    os.chdir(cwd)

def git_pull(repo, dest=''):
    if dest == '': dest = '{repo}/'.format(repo=repo.split('/')[-1])
    cwd = os.getcwd()
    os.chdir(dest)
    with open(os.devnull, 'w') as nullfd:
        subprocess.call(['git', 'remote', 'update', '--prune'], stdout=nullfd, stderr=nullfd)
    os.chdir(cwd)

def check_user(users, search):
    found = False
    i = 0
    while not found and i < len(users):
        if users[i]['username'] == search:
            found = True
        i += 1
    return found

def call_api(api_url, proxies):
    json_ = json.JSONDecoder()
    data = json_.decode(http_get_request('{host}{api_url}?token={token}'.format(host=API_BASE_URL,
                                                                                api_url=api_url,
                                                                                token=BACKUP_TOKEN),
                                         proxies)['text'])
    return data

def enum(proxies):
    repositories = []
    endpoint = '/user/repos'
    try:
        repos = call_api(endpoint, proxies)
    except (ValueError, requests.exceptions.ReadTimeout) as e:
        raise ValueError('An error occurred while trying to fetch repos')
    else:
        for repo in repos:
            endpoint = '/repos/{}/collaborators/'.format(repo['full_name'])
            try:
                collabs = call_api(endpoint, proxies)
            except (ValueError, requests.exceptions.ReadTimeout) as e:
                raise ValueError('An error occurred while trying to fetch collaborators')
            else:
                if check_user(collabs, BACKUP_USERNAME):
                    repositories += [repo['full_name']]
    return repositories

def main():
    """Main process."""
    try:
        if PROXY:
            proxies = {
                'http':  'http://127.0.0.1:8080',  # Proxy for HTTP streams
                'https': 'http://127.0.0.1:8080',  # Proxy for HTTPs streams
            }
        else:
            proxies = None

        cwd = os.getcwd()
        os.chdir(BACKUP_DIRECTORY)

        projects = enum(proxies)
        for project in projects:
            dest = '{repo}/'.format(repo=project.split('/')[-1])
            if not os.path.isdir(dest):
                print('Cloning {}...'.format(project))
                git_clone(project, dest)
            else:
                print('Updating {}...'.format(project))
                git_pull(project, dest)

        os.chdir(cwd)
    except (KeyError, TypeError, ValueError) as exception_:
        os.chdir(cwd)
        print('Error: {}'.format(exception_))

# Runtime processor
if __name__ == '__main__':
    main()