Gitlab to Gitea migration script.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

11 ay önce
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614
  1. import base64
  2. import os
  3. import time
  4. import random
  5. import string
  6. import requests
  7. import json
  8. import dateutil.parser
  9. import datetime
  10. import gitlab # pip install python-gitlab
  11. import gitlab.v4.objects
  12. import pygitea # pip install pygitea (https://github.com/h44z/pygitea)
  13. SCRIPT_VERSION = "1.0"
  14. GLOBAL_ERROR_COUNT = 0
  15. #######################
  16. # CONFIG SECTION START
  17. #######################
  18. GITLAB_URL = 'https://gitlab.source.com'
  19. GITLAB_TOKEN = 'gitlab token'
  20. # needed to clone the repositories, keep empty to try publickey (untested)
  21. GITLAB_ADMIN_USER = 'admin username'
  22. GITLAB_ADMIN_PASS = 'admin password'
  23. GITEA_URL = 'https://gitea.dest.com'
  24. GITEA_TOKEN = 'gitea token'
  25. #######################
  26. # CONFIG SECTION END
  27. #######################
  28. def main():
  29. print_color(bcolors.HEADER, "---=== Gitlab to Gitea migration ===---")
  30. print("Version: " + SCRIPT_VERSION)
  31. print()
  32. # private token or personal token authentication
  33. gl = gitlab.Gitlab(GITLAB_URL, private_token=GITLAB_TOKEN)
  34. gl.auth()
  35. assert(isinstance(gl.user, gitlab.v4.objects.CurrentUser))
  36. gt = pygitea.API(GITEA_URL, token=GITEA_TOKEN)
  37. gt_version = gt.get('/version').json()
  38. print_info("Connected to Gitea, version: " + str(gt_version['version']))
  39. # IMPORT USERS AND GROUPS
  40. import_users_groups(gl, gt)
  41. # IMPORT PROJECTS
  42. import_projects(gl, gt)
  43. print()
  44. if GLOBAL_ERROR_COUNT == 0:
  45. print_success("Migration finished with no errors!")
  46. else:
  47. print_error("Migration finished with " + str(GLOBAL_ERROR_COUNT) + " errors!")
  48. #
  49. # Data loading helpers for Gitea
  50. #
  51. def get_labels(gitea_api: pygitea, owner: string, repo: string) -> []:
  52. existing_labels = []
  53. label_response: requests.Response = gitea_api.get("/repos/" + owner + "/" + repo + "/labels")
  54. if label_response.ok:
  55. existing_labels = label_response.json()
  56. else:
  57. print_error("Failed to load existing milestones for project " + repo + "! " + label_response.text)
  58. return existing_labels
  59. def get_milestones(gitea_api: pygitea, owner: string, repo: string) -> []:
  60. existing_milestones = []
  61. milestone_response: requests.Response = gitea_api.get("/repos/" + owner + "/" + repo + "/milestones")
  62. if milestone_response.ok:
  63. existing_milestones = milestone_response.json()
  64. else:
  65. print_error("Failed to load existing milestones for project " + repo + "! " + milestone_response.text)
  66. return existing_milestones
  67. def get_issues(gitea_api: pygitea, owner: string, repo: string) -> []:
  68. existing_issues = []
  69. issue_response: requests.Response = gitea_api.get("/repos/" + owner + "/" + repo + "/issues", params={
  70. "state": "all",
  71. "page": -1
  72. })
  73. if issue_response.ok:
  74. existing_issues = issue_response.json()
  75. else:
  76. print_error("Failed to load existing issues for project " + repo + "! " + issue_response.text)
  77. return existing_issues
  78. def get_teams(gitea_api: pygitea, orgname: string) -> []:
  79. existing_teams = []
  80. team_response: requests.Response = gitea_api.get("/orgs/" + orgname + "/teams")
  81. if team_response.ok:
  82. existing_teams = team_response.json()
  83. else:
  84. print_error("Failed to load existing teams for organization " + orgname + "! " + team_response.text)
  85. return existing_teams
  86. def get_team_members(gitea_api: pygitea, teamid: int) -> []:
  87. existing_members = []
  88. member_response: requests.Response = gitea_api.get("/teams/" + str(teamid) + "/members")
  89. if member_response.ok:
  90. existing_members = member_response.json()
  91. else:
  92. print_error("Failed to load existing members for team " + str(teamid) + "! " + member_response.text)
  93. return existing_members
  94. def get_collaborators(gitea_api: pygitea, owner: string, repo: string) -> []:
  95. existing_collaborators = []
  96. collaborator_response: requests.Response = gitea_api.get("/repos/" + owner+ "/" + repo + "/collaborators")
  97. if collaborator_response.ok:
  98. existing_collaborators = collaborator_response.json()
  99. else:
  100. print_error("Failed to load existing collaborators for project " + repo + "! " + collaborator_response.text)
  101. return existing_collaborators
  102. def get_user_or_group(gitea_api: pygitea, name: string) -> {}:
  103. result = None
  104. response: requests.Response = gitea_api.get("/users/" + name)
  105. if response.ok:
  106. result = response.json()
  107. else:
  108. print_error("Failed to load user or group " + name + "! " + response.text)
  109. return result
  110. def get_user_keys(gitea_api: pygitea, username: string) -> {}:
  111. result = []
  112. key_response: requests.Response = gitea_api.get("/users/" + username + "/keys")
  113. if key_response.ok:
  114. result = key_response.json()
  115. else:
  116. print_error("Failed to load user keys for user " + username + "! " + key_response.text)
  117. return result
  118. def user_exists(gitea_api: pygitea, username: string) -> bool:
  119. user_response: requests.Response = gitea_api.get("/users/" + username)
  120. if user_response.ok:
  121. print_warning("User " + username + " does already exist in Gitea, skipping!")
  122. else:
  123. print("User " + username + " not found in Gitea, importing!")
  124. return user_response.ok
  125. def user_key_exists(gitea_api: pygitea, username: string, keyname: string) -> bool:
  126. existing_keys = get_user_keys(gitea_api, username)
  127. if existing_keys:
  128. existing_key = next((item for item in existing_keys if item["title"] == keyname), None)
  129. if existing_key is not None:
  130. print_warning("Public key " + keyname + " already exists for user " + username + ", skipping!")
  131. return True
  132. else:
  133. print("Public key " + keyname + " does not exists for user " + username + ", importing!")
  134. return False
  135. else:
  136. print("No public keys for user " + username + ", importing!")
  137. return False
  138. def organization_exists(gitea_api: pygitea, orgname: string) -> bool:
  139. group_response: requests.Response = gitea_api.get("/orgs/" + orgname)
  140. if group_response.ok:
  141. print_warning("Group " + orgname + " does already exist in Gitea, skipping!")
  142. else:
  143. print("Group " + orgname + " not found in Gitea, importing!")
  144. return group_response.ok
  145. def member_exists(gitea_api: pygitea, username: string, teamid: int) -> bool:
  146. existing_members = get_team_members(gitea_api, teamid)
  147. if existing_members:
  148. existing_member = next((item for item in existing_members if item["username"] == username), None)
  149. if existing_member:
  150. print_warning("Member " + username + " is already in team " + str(teamid) + ", skipping!")
  151. return True
  152. else:
  153. print("Member " + username + " is not in team " + str(teamid) + ", importing!")
  154. return False
  155. else:
  156. print("No members in team " + str(teamid) + ", importing!")
  157. return False
  158. def collaborator_exists(gitea_api: pygitea, owner: string, repo: string, username: string) -> bool:
  159. collaborator_response: requests.Response = gitea_api.get("/repos/" + owner + "/" + repo + "/collaborators/" + username)
  160. if collaborator_response.ok:
  161. print_warning("Collaborator " + username + " does already exist in Gitea, skipping!")
  162. else:
  163. print("Collaborator " + username + " not found in Gitea, importing!")
  164. return collaborator_response.ok
  165. def repo_exists(gitea_api: pygitea, owner: string, repo: string) -> bool:
  166. repo_response: requests.Response = gitea_api.get("/repos/" + owner + "/" + repo)
  167. if repo_response.ok:
  168. print_warning("Project " + repo + " does already exist in Gitea, skipping!")
  169. else:
  170. print("Project " + repo + " not found in Gitea, importing!")
  171. return repo_response.ok
  172. def label_exists(gitea_api: pygitea, owner: string, repo: string, labelname: string) -> bool:
  173. existing_labels = get_labels(gitea_api, owner, repo)
  174. if existing_labels:
  175. existing_label = next((item for item in existing_labels if item["name"] == labelname), None)
  176. if existing_label is not None:
  177. print_warning("Label " + labelname + " already exists in project " + repo + ", skipping!")
  178. return True
  179. else:
  180. print("Label " + labelname + " does not exists in project " + repo + ", importing!")
  181. return False
  182. else:
  183. print("No labels in project " + repo + ", importing!")
  184. return False
  185. def milestone_exists(gitea_api: pygitea, owner: string, repo: string, milestone: string) -> bool:
  186. existing_milestones = get_milestones(gitea_api, owner, repo)
  187. if existing_milestones:
  188. existing_milestone = next((item for item in existing_milestones if item["title"] == milestone), None)
  189. if existing_milestone is not None:
  190. print_warning("Milestone " + milestone + " already exists in project " + repo + ", skipping!")
  191. return True
  192. else:
  193. print("Milestone " + milestone + " does not exists in project " + repo + ", importing!")
  194. return False
  195. else:
  196. print("No milestones in project " + repo + ", importing!")
  197. return False
  198. def issue_exists(gitea_api: pygitea, owner: string, repo: string, issue: string) -> bool:
  199. existing_issues = get_issues(gitea_api, owner, repo)
  200. if existing_issues:
  201. existing_issue = next((item for item in existing_issues if item["title"] == issue), None)
  202. if existing_issue is not None:
  203. print_warning("Issue " + issue + " already exists in project " + repo + ", skipping!")
  204. return True
  205. else:
  206. print("Issue " + issue + " does not exists in project " + repo + ", importing!")
  207. return False
  208. else:
  209. print("No issues in project " + repo + ", importing!")
  210. return False
  211. #
  212. # Import helper functions
  213. #
  214. def _import_project_labels(gitea_api: pygitea, labels: [gitlab.v4.objects.ProjectLabel], owner: string, repo: string):
  215. for label in labels:
  216. if not label_exists(gitea_api, owner, repo, label.name):
  217. import_response: requests.Response = gitea_api.post("/repos/" + owner + "/" + repo + "/labels", json={
  218. "name": label.name,
  219. "color": label.color,
  220. "description": label.description # currently not supported
  221. })
  222. if import_response.ok:
  223. print_info("Label " + label.name + " imported!")
  224. else:
  225. print_error("Label " + label.name + " import failed: " + import_response.text)
  226. def _import_project_milestones(gitea_api: pygitea, milestones: [gitlab.v4.objects.ProjectMilestone], owner: string, repo: string):
  227. for milestone in milestones:
  228. if not milestone_exists(gitea_api, owner, repo, milestone.title):
  229. due_date = None
  230. if milestone.due_date is not None and milestone.due_date != '':
  231. due_date = dateutil.parser.parse(milestone.due_date).strftime('%Y-%m-%dT%H:%M:%SZ')
  232. import_response: requests.Response = gitea_api.post("/repos/" + owner + "/" + repo + "/milestones", json={
  233. "description": milestone.description,
  234. "due_on": due_date,
  235. "title": milestone.title,
  236. })
  237. if import_response.ok:
  238. print_info("Milestone " + milestone.title + " imported!")
  239. existing_milestone = import_response.json()
  240. if existing_milestone:
  241. # update milestone state, this cannot be done in the initial import :(
  242. # TODO: gitea api ignores the closed state...
  243. update_response: requests.Response = gitea_api.patch("/repos/" + owner + "/" + repo + "/milestones/" + str(existing_milestone['id']), json={
  244. "description": milestone.description,
  245. "due_on": due_date,
  246. "title": milestone.title,
  247. "state": milestone.state
  248. })
  249. if update_response.ok:
  250. print_info("Milestone " + milestone.title + " updated!")
  251. else:
  252. print_error("Milestone " + milestone.title + " update failed: " + update_response.text)
  253. else:
  254. print_error("Milestone " + milestone.title + " import failed: " + import_response.text)
  255. def _import_project_issues(gitea_api: pygitea, issues: [gitlab.v4.objects.ProjectIssue], owner: string, repo: string):
  256. # reload all existing milestones and labels, needed for assignment in issues
  257. existing_milestones = get_milestones(gitea_api, owner, repo)
  258. existing_labels = get_labels(gitea_api, owner, repo)
  259. for issue in issues:
  260. if not issue_exists(gitea_api, owner, repo, issue.title):
  261. due_date = ''
  262. if issue.due_date is not None:
  263. due_date = dateutil.parser.parse(issue.due_date).strftime('%Y-%m-%dT%H:%M:%SZ')
  264. assignee = None
  265. if issue.assignee is not None:
  266. assignee = issue.assignee['username']
  267. assignees = []
  268. for tmp_assignee in issue.assignees:
  269. assignees.append(tmp_assignee['username'])
  270. milestone = None
  271. if issue.milestone is not None:
  272. existing_milestone = next((item for item in existing_milestones if item["title"] == issue.milestone['title']), None)
  273. if existing_milestone:
  274. milestone = existing_milestone['id']
  275. labels = []
  276. for label in issue.labels:
  277. existing_label = next((item for item in existing_labels if item["name"] == label), None)
  278. if existing_label:
  279. labels.append(existing_label['id'])
  280. import_response: requests.Response = gitea_api.post("/repos/" + owner + "/" + repo + "/issues", json={
  281. "assignee": assignee,
  282. "assignees": assignees,
  283. "body": issue.description,
  284. "closed": issue.state == 'closed',
  285. "due_on": due_date,
  286. "labels": labels,
  287. "milestone": milestone,
  288. "title": issue.title,
  289. })
  290. if import_response.ok:
  291. print_info("Issue " + issue.title + " imported!")
  292. else:
  293. print_error("Issue " + issue.title + " import failed: " + import_response.text)
  294. def _import_project_repo(gitea_api: pygitea, project: gitlab.v4.objects.Project):
  295. if not repo_exists(gitea_api, project.namespace['name'], project.name):
  296. clone_url = project.http_url_to_repo
  297. if GITLAB_ADMIN_PASS is '' and GITLAB_ADMIN_USER is '':
  298. clone_url = project.ssh_url_to_repo
  299. private = project.visibility == 'private' or project.visibility == 'internal'
  300. # Load the owner (users and groups can both be fetched using the /users/ endpoint)
  301. owner = get_user_or_group(gitea_api, project.namespace['name'])
  302. if owner:
  303. import_response: requests.Response = gitea_api.post("/repos/migrate", json={
  304. "auth_password": GITLAB_ADMIN_PASS,
  305. "auth_username": GITLAB_ADMIN_USER,
  306. "clone_addr": clone_url,
  307. "description": project.description,
  308. "mirror": False,
  309. "private": private,
  310. "repo_name": project.name,
  311. "uid": owner['id']
  312. })
  313. if import_response.ok:
  314. print_info("Project " + project.name + " imported!")
  315. else:
  316. print_error("Project " + project.name + " import failed: " + import_response.text)
  317. else:
  318. print_error("Failed to load project owner for project " + project.name)
  319. def _import_project_repo_collaborators(gitea_api: pygitea, collaborators: [gitlab.v4.objects.ProjectMember], project: gitlab.v4.objects.Project):
  320. for collaborator in collaborators:
  321. if not collaborator_exists(gitea_api, project.namespace['name'], project.name, collaborator.username):
  322. permission = "read"
  323. if collaborator.access_level == 10: # guest access
  324. permission = "read"
  325. elif collaborator.access_level == 20: # reporter access
  326. permission = "read"
  327. elif collaborator.access_level == 30: # developer access
  328. permission = "write"
  329. elif collaborator.access_level == 40: # maintainer access
  330. permission = "admin"
  331. elif collaborator.access_level == 50: # owner access (only for groups)
  332. print_error("Groupmembers are currently not supported!")
  333. continue # groups are not supported
  334. else:
  335. print_warning("Unsupported access level " + str(collaborator.access_level) + ", setting permissions to 'read'!")
  336. import_response: requests.Response = gitea_api.put("/repos/" + project.namespace['name'] +"/" + project.name + "/collaborators/" + collaborator.username, json={
  337. "permission": permission
  338. })
  339. if import_response.ok:
  340. print_info("Collaborator " + collaborator.username + " imported!")
  341. else:
  342. print_error("Collaborator " + collaborator.username + " import failed: " + import_response.text)
  343. def _import_users(gitea_api: pygitea, users: [gitlab.v4.objects.User], notify: bool = False):
  344. for user in users:
  345. keys: [gitlab.v4.objects.UserKey] = user.keys.list(all=True)
  346. print("Importing user " + user.username + "...")
  347. print("Found " + str(len(keys)) + " public keys for user " + user.username)
  348. if not user_exists(gitea_api, user.username):
  349. tmp_password = ''.join(random.choices(string.ascii_uppercase + string.digits, k=10))
  350. import_response: requests.Response = gitea_api.post("/admin/users", json={
  351. "email": user.email,
  352. "full_name": user.name,
  353. "login_name": user.username,
  354. "password": tmp_password,
  355. "send_notify": notify,
  356. "source_id": 0, # local user
  357. "username": user.username
  358. })
  359. if import_response.ok:
  360. print_info("User " + user.username + " imported, temporary password: " + tmp_password)
  361. else:
  362. print_error("User " + user.username + " import failed: " + import_response.text)
  363. # import public keys
  364. _import_user_keys(gitea_api, keys, user)
  365. def _import_user_keys(gitea_api: pygitea, keys: [gitlab.v4.objects.UserKey], user: gitlab.v4.objects.User):
  366. for key in keys:
  367. if not user_key_exists(gitea_api, user.username, key.title):
  368. import_response: requests.Response = gitea_api.post("/admin/users/" + user.username + "/keys", json={
  369. "key": key.key,
  370. "read_only": True,
  371. "title": key.title,
  372. })
  373. if import_response.ok:
  374. print_info("Public key " + key.title + " imported!")
  375. else:
  376. print_error("Public key " + key.title + " import failed: " + import_response.text)
  377. def _import_groups(gitea_api: pygitea, groups: [gitlab.v4.objects.Group]):
  378. for group in groups:
  379. members: [gitlab.v4.objects.GroupMember] = group.members.list(all=True)
  380. print("Importing group " + group.name + "...")
  381. print("Found " + str(len(members)) + " gitlab members for group " + group.name)
  382. if not organization_exists(gitea_api, group.name):
  383. import_response: requests.Response = gitea_api.post("/orgs", json={
  384. "description": group.description,
  385. "full_name": group.full_name,
  386. "location": "",
  387. "username": group.name,
  388. "website": ""
  389. })
  390. if import_response.ok:
  391. print_info("Group " + group.name + " imported!")
  392. else:
  393. print_error("Group " + group.name + " import failed: " + import_response.text)
  394. # import group members
  395. _import_group_members(gitea_api, members, group)
  396. def _import_group_members(gitea_api: pygitea, members: [gitlab.v4.objects.GroupMember], group: gitlab.v4.objects.Group):
  397. # TODO: create teams based on gitlab permissions (access_level of group member)
  398. existing_teams = get_teams(gitea_api, group.name)
  399. if existing_teams:
  400. first_team = existing_teams[0]
  401. print("Organization teams fetched, importing users to first team: " + first_team['name'])
  402. # add members to teams
  403. for member in members:
  404. if not member_exists(gitea_api, member.username, first_team['id']):
  405. import_response: requests.Response = gitea_api.put("/teams/" + str(first_team['id']) + "/members/" + member.username)
  406. if import_response.ok:
  407. print_info("Member " + member.username + " added to group " + group.name + "!")
  408. else:
  409. print_error("Failed to add member " + member.username + " to group " + group.name + "!")
  410. else:
  411. print_error("Failed to import members to group " + group.name + ": no teams found!")
  412. #
  413. # Import functions
  414. #
  415. def import_users_groups(gitlab_api: gitlab.Gitlab, gitea_api: pygitea, notify=False):
  416. # read all users
  417. users: [gitlab.v4.objects.User] = gitlab_api.users.list(all=True)
  418. groups: [gitlab.v4.objects.Group] = gitlab_api.groups.list(all=True)
  419. print("Found " + str(len(users)) + " gitlab users as user " + gitlab_api.user.username)
  420. print("Found " + str(len(groups)) + " gitlab groups as user " + gitlab_api.user.username)
  421. # import all non existing users
  422. _import_users(gitea_api, users, notify)
  423. # import all non existing groups
  424. _import_groups(gitea_api, groups)
  425. def import_projects(gitlab_api: gitlab.Gitlab, gitea_api: pygitea):
  426. # read all projects and their issues
  427. projects: gitlab.v4.objects.Project = gitlab_api.projects.list(all=True)
  428. print("Found " + str(len(projects)) + " gitlab projects as user " + gitlab_api.user.username)
  429. for project in projects:
  430. collaborators: [gitlab.v4.objects.ProjectMember] = project.members.list(all=True)
  431. labels: [gitlab.v4.objects.ProjectLabel] = project.labels.list(all=True)
  432. milestones: [gitlab.v4.objects.ProjectMilestone] = project.milestones.list(all=True)
  433. issues: [gitlab.v4.objects.ProjectIssue] = project.issues.list(all=True)
  434. print("Importing project " + project.name + " from owner " + project.namespace['name'])
  435. print("Found " + str(len(collaborators)) + " collaborators for project " + project.name)
  436. print("Found " + str(len(labels)) + " labels for project " + project.name)
  437. print("Found " + str(len(milestones)) + " milestones for project " + project.name)
  438. print("Found " + str(len(issues)) + " issues for project " + project.name)
  439. # import project repo
  440. _import_project_repo(gitea_api, project)
  441. # import collaborators
  442. _import_project_repo_collaborators(gitea_api, collaborators, project)
  443. # import labels
  444. _import_project_labels(gitea_api, labels, project.namespace['name'], project.name)
  445. # import milestones
  446. _import_project_milestones(gitea_api, milestones, project.namespace['name'], project.name)
  447. # import issues
  448. _import_project_issues(gitea_api, issues, project.namespace['name'], project.name)
  449. #
  450. # Helper functions
  451. #
  452. class bcolors:
  453. HEADER = '\033[95m'
  454. OKBLUE = '\033[94m'
  455. OKGREEN = '\033[92m'
  456. WARNING = '\033[93m'
  457. FAIL = '\033[91m'
  458. ENDC = '\033[0m'
  459. BOLD = '\033[1m'
  460. UNDERLINE = '\033[4m'
  461. def color_message(color, message, colorend=bcolors.ENDC, bold=False):
  462. if bold:
  463. return bcolors.BOLD + color_message(color, message, colorend, False)
  464. return color + message + colorend
  465. def print_color(color, message, colorend=bcolors.ENDC, bold=False):
  466. print(color_message(color, message, colorend))
  467. def print_info(message):
  468. print_color(bcolors.OKBLUE, message)
  469. def print_success(message):
  470. print_color(bcolors.OKGREEN, message)
  471. def print_warning(message):
  472. print_color(bcolors.WARNING, message)
  473. def print_error(message):
  474. global GLOBAL_ERROR_COUNT
  475. GLOBAL_ERROR_COUNT += 1
  476. print_color(bcolors.FAIL, message)
  477. if __name__ == "__main__":
  478. main()