Last active
October 12, 2020 20:40
-
-
Save aunger/c239e41e9b99eed6eb1dd1ae3c34cb34 to your computer and use it in GitHub Desktop.
Gogs migration script
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
[ | |
{ | |
"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 | |
} | |
} | |
] |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
45 23 * * * cd $HOME/gogsmigrate && python3 gogsmigrate.py >> gogsmigrate.py.`date +\%Y\%m\%d`.log 2>&1 |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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.