123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614 |
- 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()
|