gitea-migrator/migrator/tasks.py

from workers import task
from .models import *
from time import sleep
from random import randint
from tempfile import TemporaryDirectory
from . import settings
from .api import *
from os import path
import re
import subprocess
import traceback

class MigrationError(Exception):
    pass

def handle_migration(repository):
    gitea = GiteaAPI(repository.migration.gitea_token)
    gitlab = GitlabAPI(repository.migration.gitlab_token)

    match = re.search('^([^/]+)/',repository.path_with_namespace)
    if match is None:
        raise MigrationError("Could not get organization")
    else:
        organization = match.group(1)

    match = re.search('([^/]+)$', repository.path_with_namespace)
    if match is None:
        raise MigrationError("Could not get repo name")
    else:
        reponame = match.group(1)

    # Create organization if necessary
    if organization == repository.migration.username:
        organization = None

    if organization is None:
        response = gitea.get("/orgs/%s" % organization)

        if response.status_code != 200:
            response = gitea.post("/orgs", json={
                "full_name" : organization,
                "username" : organization,
            })

            if response.status_code != 201:
                raise MigrationError("Could not create organization: %s" % (response.text) )

    owner = organization or repository.migration.username

    # Create repo if necessary
    if organization is None:
        response = gitea.post('/user/repos', data={
            'auto_init': False,
            'default_branch': 'master',
            'description': "Automatically created",
            'name': reponame,
        })
    else:
        response = gitea.post('/org/%s/repos' % organization, data={
            'name': reponame,
            'private': True,
        })

    if response.status_code == 409:
        pass # exists
    elif response.status_code != 201:
        raise MigrationError("Could not create repo: %s" % (response.text))

    gitlab_remote = settings.GITLAB_REPO_URL % (
        repository.migration.username,
        repository.migration.gitlab_token,
        repository.path_with_namespace
    )

    gitea_remote = settings.GITEA_REPO_URL % (
        repository.migration.username,
        repository.migration.gitea_token,
        repository.path_with_namespace
    )

    # Migrate
    with TemporaryDirectory() as tempdir:
        gitdir = path.join(tempdir,'git')

        try:
            subprocess.check_call([
                'git','clone','--mirror',
                gitlab_remote,
                gitdir
            ])
        except subprocess.CalledProcessError:
            raise MigrationError("Could not clone")

        try:
            subprocess.check_call([
                'git', '-C', gitdir,
                'remote','add',
                'gitea', gitea_remote
            ])
        except subprocess.CalledProcessError:
            raise MigrationError("Could not set remote")

        try:
            subprocess.check_call([
                'git', '-C', gitdir,
                'push', '--all', 'gitea'
            ])
        except subprocess.CalledProcessError:
            raise MigrationError("Could not push")

    return Status.SUCCESS

@task()
def migrate_repository(repository_id):
    try:
        repository = Repository.objects.select_related('migration').get(pk=repository_id)
    except Repository.DoesNotExist:
        return

    print(">>> Migrating repository %s..." % (repository.path_with_namespace), end="\n\n")

    repository.result = Status.IN_PROGRESS
    repository.save()

    # Might take a while
    try:
        result = handle_migration(repository)
        repository.result = result
    except Exception as e:
        result = Status.ERROR
        repository.result = result
        repository.error_message = traceback.format_exc()
    finally:
        repository.save()
        print(end="\n\n")