Browse Source

first commit

master
Christoph Haas 3 years ago
commit
7713b0eef2
  1. 123
      .gitignore
  2. 5
      LICENSE
  3. 17
      README.md
  4. 614
      migrate.py

123
.gitignore

@ -0,0 +1,123 @@
# ---> Python
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
.hypothesis/
.pytest_cache/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
.python-version
# celery beat schedule file
celerybeat-schedule
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# ---> VisualStudioCode
.vscode/*
!.vscode/settings.json
!.vscode/tasks.json
!.vscode/launch.json
!.vscode/extensions.json

5
LICENSE

@ -0,0 +1,5 @@
MIT License
Copyright (c) <year> <copyright holders>
Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

17
README.md

@ -0,0 +1,17 @@
# Gitlab to Gitea migration script.
This script uses the Gitlab and Gitea API's to migrate all data from
Gitlab to Gitea.
This script support migrating the following data:
- Repositories & Wiki
- Milestones
- Labels
- Issues
- Users
- Groups
## Usage
Change items in the config section of the script.
Install all dependencies and use python3 to execute the script.

614
migrate.py

@ -0,0 +1,614 @@
import base64
import os
import time
import random
import string
import requests
import json
import dateutil.parser
import datetime
import gitlab # pip install python-gitlab
import gitlab.v4.objects
import pygitea # pip install pygitea (https://github.com/h44z/pygitea)
SCRIPT_VERSION = "1.0"
GLOBAL_ERROR_COUNT = 0
#######################
# CONFIG SECTION START
#######################
GITLAB_URL = 'https://gitlab.source.com'
GITLAB_TOKEN = 'gitlab token'
# needed to clone the repositories, keep empty to try publickey (untested)
GITLAB_ADMIN_USER = 'admin username'
GITLAB_ADMIN_PASS = 'admin password'
GITEA_URL = 'https://gitea.dest.com'
GITEA_TOKEN = 'gitea token'
#######################
# CONFIG SECTION END
#######################
def main():
print_color(bcolors.HEADER, "---=== Gitlab to Gitea migration ===---")
print("Version: " + SCRIPT_VERSION)
print()
# private token or personal token authentication
gl = gitlab.Gitlab(GITLAB_URL, private_token=GITLAB_TOKEN)
gl.auth()
assert(isinstance(gl.user, gitlab.v4.objects.CurrentUser))
gt = pygitea.API(GITEA_URL, token=GITEA_TOKEN)
gt_version = gt.get('/version').json()
print_info("Connected to Gitea, version: " + str(gt_version['version']))
# IMPORT USERS AND GROUPS
import_users_groups(gl, gt)
# IMPORT PROJECTS
import_projects(gl, gt)
print()
if GLOBAL_ERROR_COUNT == 0:
print_success("Migration finished with no errors!")
else:
print_error("Migration finished with " + str(GLOBAL_ERROR_COUNT) + " errors!")
#
# Data loading helpers for Gitea
#
def get_labels(gitea_api: pygitea, owner: string, repo: string) -> []:
existing_labels = []
label_response: requests.Response = gitea_api.get("/repos/" + owner + "/" + repo + "/labels")
if label_response.ok:
existing_labels = label_response.json()
else:
print_error("Failed to load existing milestones for project " + repo + "! " + label_response.text)
return existing_labels
def get_milestones(gitea_api: pygitea, owner: string, repo: string) -> []:
existing_milestones = []
milestone_response: requests.Response = gitea_api.get("/repos/" + owner + "/" + repo + "/milestones")
if milestone_response.ok:
existing_milestones = milestone_response.json()
else:
print_error("Failed to load existing milestones for project " + repo + "! " + milestone_response.text)
return existing_milestones
def get_issues(gitea_api: pygitea, owner: string, repo: string) -> []:
existing_issues = []
issue_response: requests.Response = gitea_api.get("/repos/" + owner + "/" + repo + "/issues", params={
"state": "all",
"page": -1
})
if issue_response.ok:
existing_issues = issue_response.json()
else:
print_error("Failed to load existing issues for project " + repo + "! " + issue_response.text)
return existing_issues
def get_teams(gitea_api: pygitea, orgname: string) -> []:
existing_teams = []
team_response: requests.Response = gitea_api.get("/orgs/" + orgname + "/teams")
if team_response.ok:
existing_teams = team_response.json()
else:
print_error("Failed to load existing teams for organization " + orgname + "! " + team_response.text)
return existing_teams
def get_team_members(gitea_api: pygitea, teamid: int) -> []:
existing_members = []
member_response: requests.Response = gitea_api.get("/teams/" + str(teamid) + "/members")
if member_response.ok:
existing_members = member_response.json()
else:
print_error("Failed to load existing members for team " + str(teamid) + "! " + member_response.text)
return existing_members
def get_collaborators(gitea_api: pygitea, owner: string, repo: string) -> []:
existing_collaborators = []
collaborator_response: requests.Response = gitea_api.get("/repos/" + owner+ "/" + repo + "/collaborators")
if collaborator_response.ok:
existing_collaborators = collaborator_response.json()
else:
print_error("Failed to load existing collaborators for project " + repo + "! " + collaborator_response.text)
return existing_collaborators
def get_user_or_group(gitea_api: pygitea, name: string) -> {}:
result = None
response: requests.Response = gitea_api.get("/users/" + name)
if response.ok:
result = response.json()
else:
print_error("Failed to load user or group " + name + "! " + response.text)
return result
def get_user_keys(gitea_api: pygitea, username: string) -> {}:
result = []
key_response: requests.Response = gitea_api.get("/users/" + username + "/keys")
if key_response.ok:
result = key_response.json()
else:
print_error("Failed to load user keys for user " + username + "! " + key_response.text)
return result
def user_exists(gitea_api: pygitea, username: string) -> bool:
user_response: requests.Response = gitea_api.get("/users/" + username)
if user_response.ok:
print_warning("User " + username + " does already exist in Gitea, skipping!")
else:
print("User " + username + " not found in Gitea, importing!")
return user_response.ok
def user_key_exists(gitea_api: pygitea, username: string, keyname: string) -> bool:
existing_keys = get_user_keys(gitea_api, username)
if existing_keys:
existing_key = next((item for item in existing_keys if item["title"] == keyname), None)
if existing_key is not None:
print_warning("Public key " + keyname + " already exists for user " + username + ", skipping!")
return True
else:
print("Public key " + keyname + " does not exists for user " + username + ", importing!")
return False
else:
print("No public keys for user " + username + ", importing!")
return False
def organization_exists(gitea_api: pygitea, orgname: string) -> bool:
group_response: requests.Response = gitea_api.get("/orgs/" + orgname)
if group_response.ok:
print_warning("Group " + orgname + " does already exist in Gitea, skipping!")
else:
print("Group " + orgname + " not found in Gitea, importing!")
return group_response.ok
def member_exists(gitea_api: pygitea, username: string, teamid: int) -> bool:
existing_members = get_team_members(gitea_api, teamid)
if existing_members:
existing_member = next((item for item in existing_members if item["username"] == username), None)
if existing_member:
print_warning("Member " + username + " is already in team " + str(teamid) + ", skipping!")
return True
else:
print("Member " + username + " is not in team " + str(teamid) + ", importing!")
return False
else:
print("No members in team " + str(teamid) + ", importing!")
return False
def collaborator_exists(gitea_api: pygitea, owner: string, repo: string, username: string) -> bool:
collaborator_response: requests.Response = gitea_api.get("/repos/" + owner + "/" + repo + "/collaborators/" + username)
if collaborator_response.ok:
print_warning("Collaborator " + username + " does already exist in Gitea, skipping!")
else:
print("Collaborator " + username + " not found in Gitea, importing!")
return collaborator_response.ok
def repo_exists(gitea_api: pygitea, owner: string, repo: string) -> bool:
repo_response: requests.Response = gitea_api.get("/repos/" + owner + "/" + repo)
if repo_response.ok:
print_warning("Project " + repo + " does already exist in Gitea, skipping!")
else:
print("Project " + repo + " not found in Gitea, importing!")
return repo_response.ok
def label_exists(gitea_api: pygitea, owner: string, repo: string, labelname: string) -> bool:
existing_labels = get_labels(gitea_api, owner, repo)
if existing_labels:
existing_label = next((item for item in existing_labels if item["name"] == labelname), None)
if existing_label is not None:
print_warning("Label " + labelname + " already exists in project " + repo + ", skipping!")
return True
else:
print("Label " + labelname + " does not exists in project " + repo + ", importing!")
return False
else:
print("No labels in project " + repo + ", importing!")
return False
def milestone_exists(gitea_api: pygitea, owner: string, repo: string, milestone: string) -> bool:
existing_milestones = get_milestones(gitea_api, owner, repo)
if existing_milestones:
existing_milestone = next((item for item in existing_milestones if item["title"] == milestone), None)
if existing_milestone is not None:
print_warning("Milestone " + milestone + " already exists in project " + repo + ", skipping!")
return True
else:
print("Milestone " + milestone + " does not exists in project " + repo + ", importing!")
return False
else:
print("No milestones in project " + repo + ", importing!")
return False
def issue_exists(gitea_api: pygitea, owner: string, repo: string, issue: string) -> bool:
existing_issues = get_issues(gitea_api, owner, repo)
if existing_issues:
existing_issue = next((item for item in existing_issues if item["title"] == issue), None)
if existing_issue is not None:
print_warning("Issue " + issue + " already exists in project " + repo + ", skipping!")
return True
else:
print("Issue " + issue + " does not exists in project " + repo + ", importing!")
return False
else:
print("No issues in project " + repo + ", importing!")
return False
#
# Import helper functions
#
def _import_project_labels(gitea_api: pygitea, labels: [gitlab.v4.objects.ProjectLabel], owner: string, repo: string):
for label in labels:
if not label_exists(gitea_api, owner, repo, label.name):
import_response: requests.Response = gitea_api.post("/repos/" + owner + "/" + repo + "/labels", json={
"name": label.name,
"color": label.color,
"description": label.description # currently not supported
})
if import_response.ok:
print_info("Label " + label.name + " imported!")
else:
print_error("Label " + label.name + " import failed: " + import_response.text)
def _import_project_milestones(gitea_api: pygitea, milestones: [gitlab.v4.objects.ProjectMilestone], owner: string, repo: string):
for milestone in milestones:
if not milestone_exists(gitea_api, owner, repo, milestone.title):
due_date = None
if milestone.due_date is not None and milestone.due_date != '':
due_date = dateutil.parser.parse(milestone.due_date).strftime('%Y-%m-%dT%H:%M:%SZ')
import_response: requests.Response = gitea_api.post("/repos/" + owner + "/" + repo + "/milestones", json={
"description": milestone.description,
"due_on": due_date,
"title": milestone.title,
})
if import_response.ok:
print_info("Milestone " + milestone.title + " imported!")
existing_milestone = import_response.json()
if existing_milestone:
# update milestone state, this cannot be done in the initial import :(
# TODO: gitea api ignores the closed state...
update_response: requests.Response = gitea_api.patch("/repos/" + owner + "/" + repo + "/milestones/" + str(existing_milestone['id']), json={
"description": milestone.description,
"due_on": due_date,
"title": milestone.title,
"state": milestone.state
})
if update_response.ok:
print_info("Milestone " + milestone.title + " updated!")
else:
print_error("Milestone " + milestone.title + " update failed: " + update_response.text)
else:
print_error("Milestone " + milestone.title + " import failed: " + import_response.text)
def _import_project_issues(gitea_api: pygitea, issues: [gitlab.v4.objects.ProjectIssue], owner: string, repo: string):
# reload all existing milestones and labels, needed for assignment in issues
existing_milestones = get_milestones(gitea_api, owner, repo)
existing_labels = get_labels(gitea_api, owner, repo)
for issue in issues:
if not issue_exists(gitea_api, owner, repo, issue.title):
due_date = ''
if issue.due_date is not None:
due_date = dateutil.parser.parse(issue.due_date).strftime('%Y-%m-%dT%H:%M:%SZ')
assignee = None
if issue.assignee is not None:
assignee = issue.assignee['username']
assignees = []
for tmp_assignee in issue.assignees:
assignees.append(tmp_assignee['username'])
milestone = None
if issue.milestone is not None:
existing_milestone = next((item for item in existing_milestones if item["title"] == issue.milestone['title']), None)
if existing_milestone:
milestone = existing_milestone['id']
labels = []
for label in issue.labels:
existing_label = next((item for item in existing_labels if item["name"] == label), None)
if existing_label:
labels.append(existing_label['id'])
import_response: requests.Response = gitea_api.post("/repos/" + owner + "/" + repo + "/issues", json={
"assignee": assignee,
"assignees": assignees,
"body": issue.description,
"closed": issue.state == 'closed',
"due_on": due_date,
"labels": labels,
"milestone": milestone,
"title": issue.title,
})
if import_response.ok:
print_info("Issue " + issue.title + " imported!")
else:
print_error("Issue " + issue.title + " import failed: " + import_response.text)
def _import_project_repo(gitea_api: pygitea, project: gitlab.v4.objects.Project):
if not repo_exists(gitea_api, project.namespace['name'], project.name):
clone_url = project.http_url_to_repo
if GITLAB_ADMIN_PASS is '' and GITLAB_ADMIN_USER is '':
clone_url = project.ssh_url_to_repo
private = project.visibility == 'private' or project.visibility == 'internal'
# Load the owner (users and groups can both be fetched using the /users/ endpoint)
owner = get_user_or_group(gitea_api, project.namespace['name'])
if owner:
import_response: requests.Response = gitea_api.post("/repos/migrate", json={
"auth_password": GITLAB_ADMIN_PASS,
"auth_username": GITLAB_ADMIN_USER,
"clone_addr": clone_url,
"description": project.description,
"mirror": False,
"private": private,
"repo_name": project.name,
"uid": owner['id']
})
if import_response.ok:
print_info("Project " + project.name + " imported!")
else:
print_error("Project " + project.name + " import failed: " + import_response.text)
else:
print_error("Failed to load project owner for project " + project.name)
def _import_project_repo_collaborators(gitea_api: pygitea, collaborators: [gitlab.v4.objects.ProjectMember], project: gitlab.v4.objects.Project):
for collaborator in collaborators:
if not collaborator_exists(gitea_api, project.namespace['name'], project.name, collaborator.username):
permission = "read"
if collaborator.access_level == 10: # guest access
permission = "read"
elif collaborator.access_level == 20: # reporter access
permission = "read"
elif collaborator.access_level == 30: # developer access
permission = "write"
elif collaborator.access_level == 40: # maintainer access
permission = "admin"
elif collaborator.access_level == 50: # owner access (only for groups)
print_error("Groupmembers are currently not supported!")
continue # groups are not supported
else:
print_warning("Unsupported access level " + str(collaborator.access_level) + ", setting permissions to 'read'!")
import_response: requests.Response = gitea_api.put("/repos/" + project.namespace['name'] +"/" + project.name + "/collaborators/" + collaborator.username, json={
"permission": permission
})
if import_response.ok:
print_info("Collaborator " + collaborator.username + " imported!")
else:
print_error("Collaborator " + collaborator.username + " import failed: " + import_response.text)
def _import_users(gitea_api: pygitea, users: [gitlab.v4.objects.User], notify: bool = False):
for user in users:
keys: [gitlab.v4.objects.UserKey] = user.keys.list(all=True)
print("Importing user " + user.username + "...")
print("Found " + str(len(keys)) + " public keys for user " + user.username)
if not user_exists(gitea_api, user.username):
tmp_password = ''.join(random.choices(string.ascii_uppercase + string.digits, k=10))
import_response: requests.Response = gitea_api.post("/admin/users", json={
"email": user.email,
"full_name": user.name,
"login_name": user.username,
"password": tmp_password,
"send_notify": notify,
"source_id": 0, # local user
"username": user.username
})
if import_response.ok:
print_info("User " + user.username + " imported, temporary password: " + tmp_password)
else:
print_error("User " + user.username + " import failed: " + import_response.text)
# import public keys
_import_user_keys(gitea_api, keys, user)
def _import_user_keys(gitea_api: pygitea, keys: [gitlab.v4.objects.UserKey], user: gitlab.v4.objects.User):
for key in keys:
if not user_key_exists(gitea_api, user.username, key.title):
import_response: requests.Response = gitea_api.post("/admin/users/" + user.username + "/keys", json={
"key": key.key,
"read_only": True,
"title": key.title,
})
if import_response.ok:
print_info("Public key " + key.title + " imported!")
else:
print_error("Public key " + key.title + " import failed: " + import_response.text)
def _import_groups(gitea_api: pygitea, groups: [gitlab.v4.objects.Group]):
for group in groups:
members: [gitlab.v4.objects.GroupMember] = group.members.list(all=True)
print("Importing group " + group.name + "...")
print("Found " + str(len(members)) + " gitlab members for group " + group.name)
if not organization_exists(gitea_api, group.name):
import_response: requests.Response = gitea_api.post("/orgs", json={
"description": group.description,
"full_name": group.full_name,
"location": "",
"username": group.name,
"website": ""
})
if import_response.ok:
print_info("Group " + group.name + " imported!")
else:
print_error("Group " + group.name + " import failed: " + import_response.text)
# import group members
_import_group_members(gitea_api, members, group)
def _import_group_members(gitea_api: pygitea, members: [gitlab.v4.objects.GroupMember], group: gitlab.v4.objects.Group):
# TODO: create teams based on gitlab permissions (access_level of group member)
existing_teams = get_teams(gitea_api, group.name)
if existing_teams:
first_team = existing_teams[0]
print("Organization teams fetched, importing users to first team: " + first_team['name'])
# add members to teams
for member in members:
if not member_exists(gitea_api, member.username, first_team['id']):
import_response: requests.Response = gitea_api.put("/teams/" + str(first_team['id']) + "/members/" + member.username)
if import_response.ok:
print_info("Member " + member.username + " added to group " + group.name + "!")
else:
print_error("Failed to add member " + member.username + " to group " + group.name + "!")
else:
print_error("Failed to import members to group " + group.name + ": no teams found!")
#
# Import functions
#
def import_users_groups(gitlab_api: gitlab.Gitlab, gitea_api: pygitea, notify=False):
# read all users
users: [gitlab.v4.objects.User] = gitlab_api.users.list(all=True)
groups: [gitlab.v4.objects.Group] = gitlab_api.groups.list(all=True)
print("Found " + str(len(users)) + " gitlab users as user " + gitlab_api.user.username)
print("Found " + str(len(groups)) + " gitlab groups as user " + gitlab_api.user.username)
# import all non existing users
_import_users(gitea_api, users, notify)
# import all non existing groups
_import_groups(gitea_api, groups)
def import_projects(gitlab_api: gitlab.Gitlab, gitea_api: pygitea):
# read all projects and their issues
projects: gitlab.v4.objects.Project = gitlab_api.projects.list(all=True)
print("Found " + str(len(projects)) + " gitlab projects as user " + gitlab_api.user.username)
for project in projects:
collaborators: [gitlab.v4.objects.ProjectMember] = project.members.list(all=True)
labels: [gitlab.v4.objects.ProjectLabel] = project.labels.list(all=True)
milestones: [gitlab.v4.objects.ProjectMilestone] = project.milestones.list(all=True)
issues: [gitlab.v4.objects.ProjectIssue] = project.issues.list(all=True)
print("Importing project " + project.name + " from owner " + project.namespace['name'])
print("Found " + str(len(collaborators)) + " collaborators for project " + project.name)
print("Found " + str(len(labels)) + " labels for project " + project.name)
print("Found " + str(len(milestones)) + " milestones for project " + project.name)
print("Found " + str(len(issues)) + " issues for project " + project.name)
# import project repo
_import_project_repo(gitea_api, project)
# import collaborators
_import_project_repo_collaborators(gitea_api, collaborators, project)
# import labels
_import_project_labels(gitea_api, labels, project.namespace['name'], project.name)
# import milestones
_import_project_milestones(gitea_api, milestones, project.namespace['name'], project.name)
# import issues
_import_project_issues(gitea_api, issues, project.namespace['name'], project.name)
#
# Helper functions
#
class bcolors:
HEADER = '\033[95m'
OKBLUE = '\033[94m'
OKGREEN = '\033[92m'
WARNING = '\033[93m'
FAIL = '\033[91m'
ENDC = '\033[0m'
BOLD = '\033[1m'
UNDERLINE = '\033[4m'
def color_message(color, message, colorend=bcolors.ENDC, bold=False):
if bold:
return bcolors.BOLD + color_message(color, message, colorend, False)
return color + message + colorend
def print_color(color, message, colorend=bcolors.ENDC, bold=False):
print(color_message(color, message, colorend))
def print_info(message):
print_color(bcolors.OKBLUE, message)
def print_success(message):
print_color(bcolors.OKGREEN, message)
def print_warning(message):
print_color(bcolors.WARNING, message)
def print_error(message):
global GLOBAL_ERROR_COUNT
GLOBAL_ERROR_COUNT += 1
print_color(bcolors.FAIL, message)
if __name__ == "__main__":
main()