92 lines
3.3 KiB
Python
92 lines
3.3 KiB
Python
from collections import namedtuple
|
|
import os
|
|
import requests
|
|
import sys
|
|
|
|
|
|
GITLAB_PAT = os.environ.get("GITLAB_PAT")
|
|
GITLAB_BASE_URL = os.environ.get("GITLAB_BASE_URL")
|
|
FORGEJO_PAT = os.environ.get("FORGEJO_PAT")
|
|
FORGEJO_BASE_URL = os.environ.get("FORGEJO_BASE_URL")
|
|
|
|
|
|
def main():
|
|
if not all([GITLAB_PAT, GITLAB_BASE_URL, FORGEJO_PAT, FORGEJO_BASE_URL]):
|
|
print("Missing one of GITLAB_PAT, GITLAB_BASE_URL, FORGEJO_PAT, FORGEJO_BASE_URL environment variables.", file=sys.stderr)
|
|
|
|
gitlab_projects = get_gitlab_projects()
|
|
print(f"[*] Discovered {len(gitlab_projects)} GitLab projects")
|
|
|
|
created_orgs = set()
|
|
|
|
for project in gitlab_projects:
|
|
if project.owner != "codemicro" and project.owner not in created_orgs:
|
|
try:
|
|
create_forgejo_organisation(project.owner)
|
|
except requests.exceptions.HTTPError as e:
|
|
if e.response.status_code != 422:
|
|
raise e
|
|
created_orgs.add(project.owner)
|
|
|
|
try:
|
|
do_forgejo_migration(project)
|
|
except requests.exceptions.HTTPError as e:
|
|
if e.response.status_code != 409:
|
|
raise e
|
|
print(f"[S] {project.owner}/{project.name} already exists - skipping")
|
|
|
|
if project.is_archived:
|
|
archive_forgejo_project(project.owner, project.name)
|
|
|
|
|
|
def gitlab_request(url: str, method: str="GET", **kwargs) -> any:
|
|
resp = requests.request(method, GITLAB_BASE_URL.rstrip("/") + "/api/v4" + url, headers={"Authorization": "Bearer " + GITLAB_PAT}, **kwargs)
|
|
resp.raise_for_status()
|
|
print(f"[#] {resp.url} status {resp.status_code}")
|
|
return resp.json()
|
|
|
|
|
|
def forgejo_request(url: str, method: str="GET", **kwargs) -> any:
|
|
resp = requests.request(method, FORGEJO_BASE_URL.rstrip("/") + "/api/v1" + url, headers={"Authorization": "token " + FORGEJO_PAT}, **kwargs)
|
|
resp.raise_for_status()
|
|
print(f"[#] {resp.url} status {resp.status_code}")
|
|
return resp.json()
|
|
|
|
|
|
GitlabProject = namedtuple("GitlabProject", ["name", "owner", "is_private", "is_archived", "clone_url", "description"])
|
|
|
|
|
|
def get_gitlab_projects() -> list[GitlabProject]:
|
|
data = gitlab_request("/projects?per_page=100")
|
|
for item in data:
|
|
assert item["visibility"] == "public" or item["visibility"] == "private", item["visibility"]
|
|
return list(map(
|
|
lambda x: GitlabProject(x["path"], x["namespace"]["path"], x["visibility"] == "private", x["archived"], x["http_url_to_repo"], x["description"]),
|
|
data,
|
|
))
|
|
|
|
|
|
def create_forgejo_organisation(name: str):
|
|
resp = forgejo_request("/orgs", method="POST", data={"user_name": name})
|
|
print(f"[C] Created new organisation {resp}")
|
|
|
|
|
|
def do_forgejo_migration(project: GitlabProject):
|
|
resp = forgejo_request("/repos/migrate", method="POST", data={
|
|
"clone_addr": project.clone_url,
|
|
"auth_token": GITLAB_PAT,
|
|
"repo_owner": project.owner,
|
|
"repo_name": project.name,
|
|
"description": project.description,
|
|
"private": project.is_private
|
|
})
|
|
print(f"[C] Created new repository {resp}")
|
|
|
|
|
|
def archive_forgejo_project(owner: str, repo: str):
|
|
forgejo_request(f"/repos/{owner}/{repo}", method="PATCH", data={"archived": True})
|
|
print(f"[P] Archived repository {owner}/{repo}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|