Initial commit

main.py

@@ -0,0 +1,92 @@
+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()