Gogs migration script

_gogsmigrate.py

import json
import sys
from collections import namedtuple
from urllib import request, parse, error


HostConfig = namedtuple("HostConfig", ["endpoint", "token", "owner", "isorg"])


def headers(hostconfig):
    return {
        "Accept": "*/*",
        "authorization": "token {}".format(hostconfig.token)
    }


def user_url(hostconfig):
    user_or_org = user_or_org_string(hostconfig)
    url = '/'.join([hostconfig.endpoint, user_or_org, hostconfig.owner])
    return url


def user_or_org_string(hostconfig):
    return "orgs" if hostconfig.isorg else "users"


def my_login(hostconfig):
    url = hostconfig.endpoint + "/user"
    head = headers(hostconfig)
    req = request.Request(url, headers=head, method="GET")
    with request.urlopen(req) as resp:
        resp_text = resp.read()
        decoded = json.loads(resp_text)
        uid = decoded["login"]
    print("My login: {}".format(uid))
    return uid


def create_org(hostconfig):
    print("Creating organization {}.".format(hostconfig.owner))
    assert hostconfig.isorg
    owner = my_login(hostconfig)
    url = '/'.join([hostconfig.endpoint, "admin/users", owner, "orgs"])
    head = headers(hostconfig)
    body = {"username": hostconfig.owner}
    data = bytes(parse.urlencode(body).encode())
    req = request.Request(url, headers=head, data=data, method="POST")
    try:
        with request.urlopen(req) as resp:
            resp_text = resp.read()
            print(resp_text)
            print()
    except error.HTTPError as e:
        print("HTTPError {}: {}".format(e.code, e.reason))


def lookup_uid(hostconfig, create_org_if_missing=False):
    url = user_url(hostconfig)
    head = headers(hostconfig)
    req = request.Request(url, headers=head, method="GET")
    try:
        with request.urlopen(req) as resp:
            resp_text = resp.read()
            decoded = json.loads(resp_text)
            uid = decoded["id"]
        print("Got UID {} from {}".format(uid, url))
        return uid
    except error.HTTPError as e:
        if create_org_if_missing and hostconfig.isorg and e.code == 404:
            create_org(hostconfig)
            return lookup_uid(hostconfig, create_org_if_missing=False)
        else:
            print("Userid not found for {}: {}".format(user_or_org_string(hostconfig), hostconfig.owner))
            return None


def list_repos(hostconfig):
    url = user_url(hostconfig) + "/repos"
    head = headers(hostconfig)
    req = request.Request(url, headers=head, method="GET")
    with request.urlopen(req) as resp:
        resp_text = resp.read()
        decoded = json.loads(resp_text)
    repos = {repo["name"]: repo["clone_url"]
             for repo in decoded
             if "name" in repo and "clone_url" in repo}
    return repos


def migrate_repo(hostconfig, clone_url, new_name, uid):
    url = '/'.join([hostconfig.endpoint, "repos/migrate"])
    head = headers(hostconfig)
    body = {
        "clone_addr": clone_url,
        "description": "Mirrored from " + clone_url,
        "mirror": True,
        "private": False,
        "repo_name": new_name,
        "uid": uid
    }
    data = bytes(parse.urlencode(body).encode())
    req = request.Request(url, headers=head, data=data, method="POST")
    with request.urlopen(req) as resp:
        resp_text = resp.read()
        print(resp_text)
        # decoded = json.loads(resp_text)
        # pprint(decoded)
        print()


def main():
    try:
        config_file = sys.argv[1] if len(sys.argv) > 1 else "config.json"
        configs = json.load(open(config_file))
        configs = [{hk: HostConfig(**hv) for hk, hv in c.items()} for c in configs]
    except:
        sys.exit(2)

    error = False

    for config in configs:
        try:
            source_config = config["source"]
            dest_config = config["dest"]

            print()
            print("==============")
            print("Migrating from {} @ {}".format(source_config.owner, source_config.endpoint))
            print("Migrating to {} @ {}".format(dest_config.owner, dest_config.endpoint))

            source_repos = list_repos(source_config)
            print("Found {} source repos.".format(len(source_repos)))

            dest_uid = lookup_uid(dest_config, create_org_if_missing=True)
            if dest_uid is None:
                print("ERROR!!! Skipping this migration.")
                error = True
                continue

            existing_dest_repos = set(list_repos(dest_config))
            to_skip = source_repos.keys() & existing_dest_repos
            to_migrate = source_repos.keys() - existing_dest_repos

            for repo in to_skip:
                print("Skipping already present repo: {}".format(repo))

            for repo in to_migrate:
                print("Migrating repo: {}".format(repo))
                url = source_repos[repo]
                migrate_repo(dest_config, url, repo, dest_uid)
        except Exception as e:
            print(e)
            error = True


    sys.exit(1 if error else 0)


if __name__ == '__main__':
    main()

config.json

[
  {
    "source": {
      "endpoint": "https://git.door43.org/api/v1", "token": "SOURCE_SECRET_TOKEN",
      "owner": "rba", "isorg": false
    },
    "dest": {
      "endpoint": "http://git.bibletranslationtools.org/api/v1", "token": "DEST_SECRET_TOKEN",
      "owner": "russell", "isorg": false
    }
  }
]

crontab

45 23 * * * cd $HOME/gogsmigrate && python3 gogsmigrate.py >> gogsmigrate.py.`date +\%Y\%m\%d`.log 2>&1

The config file gives pairs of source/destination users (or orgs). All repos owned by the source user are migrated to the destination account.

Modify the .py script to change "mirror" to false if you don't want to set the migrated repos to keep sync automatically with the source.