# Description: Script to migrate repositories from GitLab to Gitea # Author: Enrico Ludwig # Version: 1.0 # Date: 2024-07-18 # License: MIT (see https://git.zion-networks.de/ZionNetworks/linux-bash-scripts/src/branch/main/LICENSE) # # Precedence of settings: arguments > .env file > environment variables > default values # # Usage: # python3 gitlab2gitea.py [options] # # --help Show this help message and exit # # --gitlab-token GitLab access token # --gitlab-url GitLab URL # --gitlab-api-version GitLab API version (default: v4) # --gitea-token Gitea access token # --gitea-url Gitea URL # --gitea-api-version Gitea API version (default: v1) # # --no-create-missing-groups Do not create missing groups on Gitea (default: False) # --no-create-missing-users Do not create missing users on Gitea (default: False) # --no-create-missing-projects Do not create missing projects on Gitea (default: False) # --no-update-existing-groups Do not update existing groups on Gitea (default: False) # --no-update-existing-users Do not update existing users on Gitea (default: False) # --no-update-existing-projects Do not update existing projects on Gitea (default: False) # # --include-wiki Include wiki repositories (default: False) # --include-issues Include issues repositories (default: False) # --include-merge-requests Include merge requests repositories (default: False) # # --override-groups Override existing groups on Gitea (default: False) # --override-users Override existing users on Gitea (default: False) # --override-projects Override existing projects on Gitea (default: False) # # --skip-empty-groups Skip empty groups (default: False) - not implemented yet # --skip-empty-projects Skip empty projects (default: False) - not implemented yet # --skip-users [user1,user2,...] Skip specific users (default: None) - not implemented yet # --skip-groups [group1,group2,...] Skip specific groups (default: None) - not implemented yet # --skip-projects [project1,project2,...] Skip specific projects (default: None) - not implemented yet # # --only-groups Migrate only groups (default: False) # --only-users Migrate only users (default: False) # --only-projects Migrate only projects (default: False) # # --quiet Enable quiet mode (default: False) # --debug Enable debug mode (default: False) # --trace Enable trace mode (default: False) # --dry-run Enable dry-run mode (default: False) # --log-file Log file (default: gitlab2gitea.log) # If not set, logs will be only printed to stdout # --append-log Append log file (default: False) # # Example: # python3 gitlab2gitea.py \ # --gitlab-url \ # --gitlab-token \ # --gitea-url \ # --gitea-token # Settings - GitLab GITLAB_URL = "" GITLAB_TOKEN = "" GITLAB_API_VERSION = "v4" # Settings - Gitea GITEA_URL = "" GITEA_TOKEN = "" GITEA_API_VERSION = "v1" # Settings - General Repository NO_CREATE_MISSING_GROUPS = False NO_CREATE_MISSING_USERS = False NO_CREATE_MISSING_PROJECTS = False NO_UPDATE_EXISTING_GROUPS = False NO_UPDATE_EXISTING_USERS = False NO_UPDATE_EXISTING_PROJECTS = False INCLUDE_WIKI = False INCLUDE_ISSUES = False INCLUDE_MERGE_REQUESTS = False OVERRIDE_EXISTING_GROUPS = False OVERRIDE_EXISTING_USERS = False OVERRIDE_EXISTING_PROJECTS = False SKIP_EMPTY_GROUPS = False SKIP_EMPTY_PROJECTS = False SKIP_USERS = [] SKIP_GROUPS = [] SKIP_PROJECTS = [] ONLY_GROUPS = False ONLY_USERS = False ONLY_PROJECTS = False # Settings - General DEBUG = False TRACE = False DRY_RUN = False LOG_FILE = "gitlab2gitea.log" APPEND_LOG = False QUIET = False ### DEBUG - REMOVE LATER ONLY_ONE_PROJECT = False ### END DEBUG # Internal variables - Do not change APP_NAME = "GitLab2Gitea" APP_VERSION = "1.0" APP_AUTHOR = "Zion Networks" EXIT_REQUESTED = False GITEA_RESERVED_USERNAMES = ["ghost", "notifications"] GITEA_RESERVED_ORGANAMES = [ "api", "assets", "attachments", "avatar", "commit", "commits", "debug", "error", "explore", "faq", "issues", "mail", "milestone", "new", "notifications", "org", "organizations", "plugins", "pull", "pulls", "repo", "repositories", "script", "user", "users", ] GITEA_RESERVED_REPONAMES = [ "api", "assets", "issues", "labels", "milestones", "notifications", "projects", "pr", "pulls", "repo", "repos", "settings", "wiki", ] # Runtime variables # fmt: off STATS = { "users": { "deleted": [], "created": [], "skipped": [], "updated": [], "errors": [] }, "groups": { "deleted": [], "created": [], "skipped": [], "updated": [], "errors": [] }, "projects": { "deleted": [], "created": [], "skipped": [], "updated": [], "errors": [] }, } # fmt: on # Imports import os import sys import argparse import requests import traceback import json import signal # Set cwd to script directory os.chdir(os.path.dirname(os.path.realpath(__file__))) # Read environment variables if "GITLAB_URL" in os.environ: GITLAB_URL = os.environ["GITLAB_URL"] if "GITLAB_TOKEN" in os.environ: GITLAB_TOKEN = os.environ["GITLAB_TOKEN"] if "GITLAB_API_VERSION" in os.environ: GITLAB_API_VERSION = os.environ["GITLAB_API_VERSION"] if "GITEA_URL" in os.environ: GITEA_URL = os.environ["GITEA_URL"] if "GITEA_TOKEN" in os.environ: GITEA_TOKEN = os.environ["GITEA_TOKEN"] if "GITEA_API_VERSION" in os.environ: GITEA_API_VERSION = os.environ["GITEA_API_VERSION"] if "NO_CREATE_MISSING_GROUPS" in os.environ: NO_CREATE_MISSING_GROUPS = bool(os.environ["NO_CREATE_MISSING_GROUPS"]) if "NO_CREATE_MISSING_USERS" in os.environ: NO_CREATE_MISSING_USERS = bool(os.environ["NO_CREATE_MISSING_USERS"]) if "NO_CREATE_MISSING_PROJECTS" in os.environ: NO_CREATE_MISSING_PROJECTS = bool(os.environ["NO_CREATE_MISSING_PROJECTS"]) if "NO_UPDATE_EXISTING_GROUPS" in os.environ: NO_UPDATE_EXISTING_GROUPS = bool(os.environ["NO_UPDATE_EXISTING_GROUPS"]) if "NO_UPDATE_EXISTING_USERS" in os.environ: NO_UPDATE_EXISTING_USERS = bool(os.environ["NO_UPDATE_EXISTING_USERS"]) if "NO_UPDATE_EXISTING_PROJECTS" in os.environ: NO_UPDATE_EXISTING_PROJECTS = bool(os.environ["NO_UPDATE_EXISTING_PROJECTS"]) if "INCLUDE_WIKI" in os.environ: INCLUDE_WIKI = bool(os.environ["INCLUDE_WIKI"]) if "INCLUDE_ISSUES" in os.environ: INCLUDE_ISSUES = bool(os.environ["INCLUDE_ISSUES"]) if "INCLUDE_MERGE_REQUESTS" in os.environ: INCLUDE_MERGE_REQUESTS = bool(os.environ["INCLUDE_MERGE_REQUESTS"]) if "OVERWRITE_EXISTING_GROUPS" in os.environ: OVERRIDE_EXISTING_GROUPS = bool(os.environ["OVERRIDE_EXISTING_GROUPS"]) if "OVERWRITE_EXISTING_USERS" in os.environ: OVERRIDE_EXISTING_USERS = bool(os.environ["OVERRIDE_EXISTING_USERS"]) if "OVERWRITE_EXISTING_PROJECTS" in os.environ: OVERRIDE_EXISTING_PROJECTS = bool(os.environ["OVERRIDE_EXISTING_PROJECTS"]) if "ONLY_GROUPS" in os.environ: ONLY_GROUPS = bool(os.environ["ONLY_GROUPS"]) if "ONLY_USERS" in os.environ: ONLY_USERS = bool(os.environ["ONLY_USERS"]) if "ONLY_PROJECTS" in os.environ: ONLY_PROJECTS = bool(os.environ["ONLY_PROJECTS"]) if "QUIET" in os.environ: QUIET = bool(os.environ["QUIET"]) if "DEBUG" in os.environ: DEBUG = bool(os.environ["DEBUG"]) if "TRACE" in os.environ: TRACE = bool(os.environ["TRACE"]) if "DRY_RUN" in os.environ: DRY_RUN = bool(os.environ["DRY_RUN"]) if "LOG_FILE" in os.environ: LOG_FILE = os.environ["LOG_FILE"] if "APPEND_LOG" in os.environ: APPEND_LOG = bool(os.environ["APPEND_LOG"]) # Read .env file if exists and override environment variables if os.path.exists(".env"): with open(".env", "r") as env_file: for line in env_file: # skip comments, empty lines and lines without '=' if line.startswith("#") or not "=" in line: continue key, value = line.strip().split("=") if key == "GITLAB_API_VERSION": GITLAB_API_VERSION = value if key == "GITLAB_URL": GITLAB_URL = value if key == "GITLAB_TOKEN": GITLAB_TOKEN = value if key == "GITEA_URL": GITEA_URL = value if key == "GITEA_TOKEN": GITEA_TOKEN = value if key == "GITEA_API_VERSION": GITEA_API_VERSION = value if key == "NO_CREATE_MISSING_GROUPS": if value.lower() == "true" or value == "1": NO_CREATE_MISSING_GROUPS = True if key == "NO_CREATE_MISSING_USERS": if value.lower() == "true" or value == "1": NO_CREATE_MISSING_USERS = True if key == "NO_CREATE_MISSING_PROJECTS": if value.lower() == "true" or value == "1": NO_CREATE_MISSING_PROJECTS = True if key == "NO_UPDATE_EXISTING_GROUPS": if value.lower() == "true" or value == "1": NO_UPDATE_EXISTING_GROUPS = True if key == "NO_UPDATE_EXISTING_USERS": if value.lower() == "true" or value == "1": NO_UPDATE_EXISTING_USERS = True if key == "NO_UPDATE_EXISTING_PROJECTS": if value.lower() == "true" or value == "1": NO_UPDATE_EXISTING_PROJECTS = True if key == "INCLUDE_WIKI": if value.lower() == "true" or value == "1": INCLUDE_WIKI = True if key == "INCLUDE_ISSUES": if value.lower() == "true" or value == "1": INCLUDE_ISSUES = True if key == "INCLUDE_MERGE_REQUESTS": if value.lower() == "true" or value == "1": INCLUDE_MERGE_REQUESTS = True if key == "OVERRIDE_EXISTING_GROUPS": if value.lower() == "true" or value == "1": OVERRIDE_EXISTING_GROUPS = True if key == "OVERRIDE_EXISTING_USERS": if value.lower() == "true" or value == "1": OVERRIDE_EXISTING_USERS = True if key == "OVERRIDE_EXISTING_PROJECTS": if value.lower() == "true" or value == "1": OVERRIDE_EXISTING_PROJECTS = True if key == "ONLY_GROUPS": if value.lower() == "true" or value == "1": ONLY_GROUPS = True if key == "ONLY_USERS": if value.lower() == "true" or value == "1": ONLY_USERS = True if key == "ONLY_PROJECTS": if value.lower() == "true" or value == "1": ONLY_PROJECTS = True if key == "QUIET": QUIET = bool(value) if key == "DEBUG": if value.lower() == "true" or value == "1": DEBUG = True if key == "TRACE": if value.lower() == "true" or value == "1": TRACE = True if key == "DRY_RUN": if value.lower() == "true" or value == "1": DRY_RUN = True if key == "LOG_FILE": LOG_FILE = value if key == "APPEND_LOG": APPEND_LOG = bool(value) # Read arguments and override environment variables parser = argparse.ArgumentParser( description="Script to migrate repositories from GitLab to Gitea" ) # fmt: off parser.add_argument("--gitlab-api-version", help="GitLab API version", default="v4") parser.add_argument("--gitlab-token", help="GitLab access token") parser.add_argument("--gitea-token", help="Gitea access token") parser.add_argument("--gitlab-url", help="GitLab URL") parser.add_argument("--gitea-url", help="Gitea URL") parser.add_argument("--gitea-api-version", help="Gitea API version", default="v1") parser.add_argument("--no-create-missing-groups", help="Do not create missing groups on Gitea", action="store_true") parser.add_argument("--no-create-missing-users", help="Do not create missing users on Gitea", action="store_true") parser.add_argument("--no-create-missing-projects", help="Do not create missing projects on Gitea", action="store_true") parser.add_argument("--no-update-existing-groups", help="Do not update existing groups on Gitea", action="store_true") parser.add_argument("--no-update-existing-users", help="Do not update existing users on Gitea", action="store_true") parser.add_argument("--no-update-existing-projects", help="Do not update existing projects on Gitea", action="store_true") parser.add_argument("--include-wiki", help="Include wiki repositories", action="store_true") parser.add_argument("--include-issues", help="Include issues repositories", action="store_true") parser.add_argument("--include-merge-requests", help="Include merge requests repositories", action="store_true") parser.add_argument("--override-groups", help="Override existing groups on Gitea", action="store_true") parser.add_argument("--override-users", help="Override existing users on Gitea", action="store_true") parser.add_argument("--override-projects", help="Override existing projects on Gitea", action="store_true") parser.add_argument("--only-groups", help="Migrate only groups", action="store_true") parser.add_argument("--only-users", help="Migrate only users", action="store_true") parser.add_argument("--only-projects", help="Migrate only projects", action="store_true") parser.add_argument("--quiet", help="Enable quiet mode", action="store_true") parser.add_argument("--debug", help="Enable debug mode", action="store_true") parser.add_argument("--trace", help="Enable trace mode", action="store_true") parser.add_argument("--dry-run", help="Enable dry-run mode", action="store_true") parser.add_argument("--log-file", help="Log file", default="gitlab2gitea.log") parser.add_argument("--append-log", help="Append log file", action="store_true") # fmt: on args = parser.parse_args() if args.gitlab_api_version: GITLAB_API_VERSION = args.gitlab_api_version if args.gitlab_token: GITLAB_TOKEN = args.gitlab_token if args.gitlab_url: GITLAB_URL = args.gitlab_url if args.gitea_token: GITEA_TOKEN = args.gitea_token if args.gitea_url: GITEA_URL = args.gitea_url if args.gitea_api_version: GITEA_API_VERSION = args.gitea_api_version if args.no_create_missing_groups: NO_CREATE_MISSING_GROUPS = True if args.no_create_missing_users: NO_CREATE_MISSING_USERS = True if args.no_create_missing_projects: NO_CREATE_MISSING_PROJECTS = True if args.no_update_existing_groups: NO_UPDATE_EXISTING_GROUPS = True if args.no_update_existing_users: NO_UPDATE_EXISTING_USERS = True if args.no_update_existing_projects: NO_UPDATE_EXISTING_PROJECTS = True if args.include_wiki: INCLUDE_WIKI = True if args.include_issues: INCLUDE_ISSUES = True if args.include_merge_requests: INCLUDE_MERGE_REQUESTS = True if args.override_groups: OVERRIDE_EXISTING_GROUPS = True if args.override_users: OVERRIDE_EXISTING_USERS = True if args.override_projects: OVERRIDE_EXISTING_PROJECTS = True if args.only_groups: ONLY_GROUPS = True if args.only_users: ONLY_USERS = True if args.only_projects: ONLY_PROJECTS = True if args.quiet: QUIET = True if args.debug: DEBUG = True if args.trace: TRACE = True if args.dry_run: DRY_RUN = True if args.log_file: LOG_FILE = args.log_file if args.append_log: APPEND_LOG = True # check for mutually exclusive options if ONLY_GROUPS and ONLY_USERS: _error("Options --only-groups and --only-users are mutually exclusive!") sys.exit(1) if ONLY_GROUPS and ONLY_PROJECTS: _error("Options --only-groups and --only-projects are mutually exclusive!") sys.exit(1) if ONLY_USERS and ONLY_PROJECTS: _error("Options --only-users and --only-projects are mutually exclusive!") sys.exit(1) if NO_CREATE_MISSING_GROUPS and OVERRIDE_EXISTING_GROUPS: _error( "Options --no-create-missing-groups and --override-groups are mutually exclusive!" ) sys.exit(1) if NO_CREATE_MISSING_USERS and OVERRIDE_EXISTING_USERS: _error( "Options --no-create-missing-users and --override-users are mutually exclusive!" ) sys.exit(1) if NO_CREATE_MISSING_PROJECTS and OVERRIDE_EXISTING_PROJECTS: _error( "Options --no-create-missing-projects and --override-projects are mutually exclusive!" ) sys.exit(1) if NO_UPDATE_EXISTING_GROUPS and OVERRIDE_EXISTING_GROUPS: _error( "Options --no-update-existing-groups and --override-groups are mutually exclusive!" ) sys.exit(1) if NO_UPDATE_EXISTING_USERS and OVERRIDE_EXISTING_USERS: _error( "Options --no-update-existing-users and --override-users are mutually exclusive!" ) sys.exit(1) if NO_UPDATE_EXISTING_PROJECTS and OVERRIDE_EXISTING_PROJECTS: _error( "Options --no-update-existing-projects and --override-projects are mutually exclusive!" ) sys.exit(1) # Remove trailing slashes from URLs GITLAB_URL = GITLAB_URL.rstrip("/") GITEA_URL = GITEA_URL.rstrip("/") # Internal functions def _trace(message): if TRACE: print("\033[1m\033[36m[TRC]\033[0m", message) if LOG_FILE: with open(LOG_FILE, "a") as log_file: log_file.write(f"[TRC] {message}\n") def _debug(message): if TRACE or DEBUG: print("\033[1m\033[34m[DBG]\033[0m", message) if LOG_FILE: with open(LOG_FILE, "a") as log_file: log_file.write(f"[DBG] {message}\n") def _info(message): print("\033[1m\033[32m[INF]\033[0m", message) if LOG_FILE: with open(LOG_FILE, "a") as log_file: log_file.write(f"[INF] {message}\n") def _warn(message): print("\033[1m\033[33m[WRN]\033[0m", message) if LOG_FILE: with open(LOG_FILE, "a") as log_file: log_file.write(f"[WRN] {message}\n") def _error(message): print("\033[1m\033[31m[ERR]\033[0m", message) if LOG_FILE: with open(LOG_FILE, "a") as log_file: log_file.write(f"[ERR] {message}\n") def _exception(exception, custom_message=None): exc_type, exc_obj, exc_tb = sys.exc_info() filename = exc_tb.tb_frame.f_code.co_filename lineno = exc_tb.tb_lineno formatted_traceback = traceback.format_exc() # Prepare formatted and clear exception messages formatted_exception_message = ( (f"{custom_message}\n" if custom_message else "") + f"\033[1m\033[31m[EXC]\033[0m {exception} " f"(file: {filename}, line: {lineno})\n" f"{formatted_traceback}\n" ) clear_exception_message = ( (f"{custom_message}\n" if custom_message else "") + f"[EXC] {exception} (file: {filename}, line: {lineno})\n" + formatted_traceback ) # Print the formatted exception message to the console print(formatted_exception_message) # Write the clear exception message to the log file if defined if LOG_FILE: with open(LOG_FILE, "a") as log_file: log_file.write(clear_exception_message + "\n") # PROGRAM def is_gitea_reserved_username(username: str) -> bool: return username.lower() in [name.lower() for name in GITEA_RESERVED_USERNAMES] def is_gitea_reserved_organame(organame: str) -> bool: return organame.lower() in [name.lower() for name in GITEA_RESERVED_ORGANAMES] def is_gitea_reserved_reponame(reponame: str) -> bool: return reponame.lower() in [name.lower() for name in GITEA_RESERVED_REPONAMES] def is_gitlab_project_in_subgroup(project: dict) -> bool: return ( project["namespace"]["kind"] == "group" and project["namespace"]["parent_id"] is not None ) def gitlab2gitea_visibility(visibility: str) -> str: if visibility == "private": return "private" elif visibility == "internal": return "limited" elif visibility == "public": return "public" else: return "private" # Endpoint: GET /api/{GITLAB_API_VERSION}/version def check_gitlab(): _debug(f"REQUEST: GET {GITLAB_URL}/api/{GITLAB_API_VERSION}/version") try: response = requests.get( f"{GITLAB_URL}/api/{GITLAB_API_VERSION}/version", headers={ "Content-Type": "application/json", "Accept": "application/json", "Authorization": f"Bearer {GITLAB_TOKEN}", }, ) _trace(f"RESPONSE: {response.json()}") except Exception as e: _exception(e, f"Failed to get GitLab version: {e}") EXIT_REQUESTED = True if response.status_code != 200: response_message = ( response.json()["message"] if "message" in response.json() else "Unknown error" ) raise Exception(f"GitLab endpoint test failed: {response_message}") else: _info(f'GitLab endpoint version: {response.json()["version"]}') # Endpoint: GET /api/{GITEA_API_VERSION}/version def check_gitea(): _debug(f"REQUEST: GET {GITEA_URL}/api/{GITEA_API_VERSION}/version") try: response = requests.get( f"{GITEA_URL}/api/{GITEA_API_VERSION}/version", headers={ "Content-Type": "application/json", "Accept": "application/json", "Authorization": f"token {GITEA_TOKEN}", }, ) _trace(f"RESPONSE: {response.json()}") except Exception as e: _exception(e, f"Failed to get Gitea version: {e}") EXIT_REQUESTED = True if response.status_code != 200: response_message = ( response.json()["message"] if "message" in response.json() else "Unknown error" ) raise Exception(f"Gitea endpoint test failed: {response_message}") else: _info(f'Gitea endpoint version: {response.json()["version"]}') # Endpoint: GET /api/{GITLAB_API_VERSION}/groups def get_gitlab_groups() -> list: groups = [] _debug(f"REQUEST: GET {GITLAB_URL}/api/{GITLAB_API_VERSION}/groups") response = requests.get( f"{GITLAB_URL}/api/{GITLAB_API_VERSION}/groups", params={"all_available": 1, "per_page": 100, "page": 1, "top_level_only": 1}, headers={ "Content-Type": "application/json", "Accept": "application/json", "Authorization": f"Bearer {GITLAB_TOKEN}", }, ) if response.status_code != 200: response_message = ( response.json()["message"] if "message" in response.json() else "Unknown error" ) raise Exception(f"Failed to get GitLab groups: {response_message}") else: groups = response.json() return groups # Endpoint: GET /api/{GITLAB_API_VERSION}/groups/{group_id} def get_gitlab_group(group_id: int) -> dict: _debug(f"REQUEST: GET {GITLAB_URL}/api/{GITLAB_API_VERSION}/groups/{group_id}") response = requests.get( f"{GITLAB_URL}/api/{GITLAB_API_VERSION}/groups/{group_id}", headers={ "Content-Type": "application/json", "Accept": "application/json", "Authorization": f"Bearer {GITLAB_TOKEN}", }, ) if response.status_code != 200: response_message = ( response.json()["message"] if "message" in response.json() else "Unknown error" ) raise Exception(f"Failed to get GitLab group: {response_message}") else: group = response.json() return group # Endpoint: GET /api/{GITLAB_API_VERSION}/users def get_gitlab_users() -> list: next_page_link = f"{GITLAB_URL}/api/{GITLAB_API_VERSION}/users" users = [] while next_page_link is not None: _debug(f'REQUEST: GET {next_page_link.split("?")[0]}') response = requests.get( next_page_link, headers={ "Content-Type": "application/json", "Accept": "application/json", "Authorization": f"Bearer {GITLAB_TOKEN}", }, ) next_page_link = ( response.links["next"]["url"] if "next" in response.links else None ) _trace(f"RESPONSE: {response.json()}") if response.status_code != 200: response_message = ( response.json()["message"] if "message" in response.json() else "Unknown error" ) raise Exception(f"Failed to get GitLab users: {response_message}") else: users += response.json() return users # Endpoint: GET /api/{GITLAB_API_VERSION}/projects def get_gitlab_projects() -> list: next_page_link = f"{GITLAB_URL}/api/{GITLAB_API_VERSION}/projects" projects = [] while next_page_link is not None: if EXIT_REQUESTED: return _debug(f'REQUEST: GET {next_page_link.split("?")[0]}') response = requests.get( next_page_link, params={"all_available": 1, "per_page": 100}, headers={ "Content-Type": "application/json", "Accept": "application/json", "Authorization": f"Bearer {GITLAB_TOKEN}", }, ) next_page_link = ( response.links["next"]["url"] if "next" in response.links else None ) _trace(f"RESPONSE: {response.json()}") if response.status_code != 200: response_message = ( response.json()["message"] if "message" in response.json() else "Unknown error" ) raise Exception(f"Failed to get GitLab projects: {response_message}") else: projects += response.json() return projects # Endpoint: POST /api/{GITLAB_API_VERSION}/repos/migrate def migrate_gitlab_project_to_gitea(gitlab_project: dict): if not gitlab_project: raise Exception("GitLab project is missing!") # Create Gitea project _debug(f"REQUEST: POST {GITEA_URL}/api/{GITEA_API_VERSION}/repos/migrate") response = requests.post( f"{GITEA_URL}/api/{GITEA_API_VERSION}/repos/migrate", json={ "auth_token": GITLAB_TOKEN, "repo_name": gitlab_project["path"], "repo_owner": ( gitlab_project["namespace"]["path"] if gitlab_project["namespace"]["kind"] == "group" else gitlab_project["owner"]["username"] ), "description": gitlab_project["description"], "clone_addr": gitlab_project["http_url_to_repo"], "service": "git", "issues": INCLUDE_ISSUES, "pull_requests": INCLUDE_MERGE_REQUESTS, "wiki": INCLUDE_WIKI, "releases": True, "labels": True, "lfs": True, "milestones": True, "mirror": False, # fmt: off "private": gitlab2gitea_visibility(gitlab_project["visibility"]) == "private", # fmt: on }, headers={ "Content-Type": "application/json", "Accept": "application/json", "Authorization": f"token {GITEA_TOKEN}", }, ) _trace(f"RESPONSE: {response.json()}") if response.status_code != 201: response_message = ( response.json()["message"] if "message" in response.json() else "Unknown error" ) STATS["projects"]["errors"].append( { "group": ( gitlab_project["namespace"]["path"] if gitlab_project["namespace"]["kind"] == "group" else gitlab_project["owner"]["username"] ), "name": gitlab_project["path"], "error": response_message if response_message else "Unknown error", } ) raise Exception(f"Failed to create Gitea project: {response_message}") else: project = response.json() STATS["projects"]["created"].append( { "group": ( gitlab_project["namespace"]["path"] if gitlab_project["namespace"]["kind"] == "group" else gitlab_project["owner"]["username"] ), "name": gitlab_project["path"], } ) return project # Endpoint: POST /api/{GITEA_API_VERSION}/admin/users def migrate_gitlab_user_to_gitea(user: dict): if not user: raise Exception("User is missing!") # Create Gitea user _debug(f"REQUEST: POST {GITEA_URL}/api/{GITEA_API_VERSION}/admin/users") response = requests.post( f"{GITEA_URL}/api/{GITEA_API_VERSION}/admin/users", json={ "login_name": user["username"], "username": user["username"], "email": user["email"], "full_name": user["name"], "password": "12345678", # TODO: Change to random password which will be sent to the user "send_notify": False, # TODO: Change to True as soon as the password is sent to the user "must_change_password": True, "admin": user["is_admin"], }, headers={ "Content-Type": "application/json", "Accept": "application/json", "Authorization": f"token {GITEA_TOKEN}", }, ) _trace(f"RESPONSE: {response.json()}") if response.status_code != 201: response_message = ( response.json()["message"] if "message" in response.json() else "Unknown error" ) STATS["users"]["errors"].append( { "username": user["username"], "error": response_message if response_message else "Unknown error", "admin": user["is_admin"], } ) raise Exception(f"Failed to create Gitea user: {response_message}") else: user = response.json() if user["is_admin"]: _info(f'Admin user "{user["username"]}" created on Gitea') else: _info(f'User "{user["username"]}" created on Gitea') STATS["users"]["created"].append( { "username": user["username"], "email": user["email"], "admin": user["is_admin"], } ) return user # Endpoint: GET /api/{GITEA_API_VERSION}/orgs def get_gitea_groups() -> list: groups = [] _debug( f"REQUEST: GET {GITEA_URL}/api/{GITEA_API_VERSION}/orgs?all_available=1&per_page=100" ) response = requests.get( f"{GITEA_URL}/api/{GITEA_API_VERSION}/orgs", headers={ "Content-Type": "application/json", "Accept": "application/json", "Authorization": f"token {GITEA_TOKEN}", }, ) if response.status_code != 200: response_message = ( response.json()["message"] if "message" in response.json() else "Unknown error" ) raise Exception(f"Failed to get Gitea groups: {response_message}") else: groups = response.json() return groups # Endpoint: GET /api/{GITEA_API_VERSION}/users def get_gitea_users() -> list: users = [] _debug(f"REQUEST: GET {GITEA_URL}/api/{GITEA_API_VERSION}/users/search") response = requests.get( f"{GITEA_URL}/api/{GITEA_API_VERSION}/users/search", headers={ "Content-Type": "application/json", "Accept": "application/json", "Authorization": f"token {GITEA_TOKEN}", }, ) _trace(f"RESPONSE: {response.json()}") if response.status_code != 200: response_message = ( response.json()["message"] if "message" in response.json() else "Unknown error" ) raise Exception(f"Failed to get Gitea users: {response_message}") else: users = response.json() return users # Endpoint: GET /api/{GITEA_API_VERSION}/repos def get_gitea_projects() -> list: projects = [] _debug(f"REQUEST: GET {GITEA_URL}/api/{GITEA_API_VERSION}/repos/search") response = requests.get( f"{GITEA_URL}/api/{GITEA_API_VERSION}/repos/search", headers={ "Content-Type": "application/json", "Accept": "application/json", "Authorization": f"token {GITEA_TOKEN}", }, ) _trace(f"RESPONSE: {response.json()}") if response.status_code != 200: response_message = ( response.json()["message"] if "message" in response.json() else "Unknown error" ) raise Exception(f"Failed to get Gitea projects: {response_message}") else: projects = response.json() return projects # Endpoint: POST /api/{GITEA_API_VERSION}/orgs def migrate_gitlab_group_to_gitea(gitlab_group: dict): if not gitlab_group: raise Exception("GitLab group is missing!") # Create Gitea group _debug(f"REQUEST: POST {GITEA_URL}/api/{GITEA_API_VERSION}/orgs") response = requests.post( f"{GITEA_URL}/api/{GITEA_API_VERSION}/orgs", json={ "username": gitlab_group["path"], "full_name": gitlab_group["full_name"], "description": gitlab_group["description"], "website": gitlab_group["web_url"], "visibility": gitlab2gitea_visibility(gitlab_group["visibility"]), }, headers={ "Content-Type": "application/json", "Accept": "application/json", "Authorization": f"token {GITEA_TOKEN}", }, ) _trace(f"RESPONSE: {response.json()}") if response.status_code != 201: response_message = ( response.json()["message"] if "message" in response.json() else "Unknown error" ) STATS["groups"]["errors"].append( { "name": gitlab_group["path"], "error": response_message if response_message else "Unknown error", } ) raise Exception(f"Failed to create Gitea group: {response_message}") else: group = response.json() STATS["groups"]["created"].append( {"name": gitlab_group["path"], "full_name": gitlab_group["full_name"]} ) return group # Endpoint: PATCH /api/{GITEA_API_VERSION}/orgs/{org} def update_gitea_org(data: dict) -> dict: if not data: raise Exception("Data is missing!") name = data["path"] _debug(f"REQUEST: PATCH {GITEA_URL}/api/{GITEA_API_VERSION}/orgs/{name}") response = requests.patch( f"{GITEA_URL}/api/{GITEA_API_VERSION}/orgs/{name}", json=data, headers={ "Content-Type": "application/json", "Accept": "application/json", "Authorization": f"token {GITEA_TOKEN}", }, ) _trace(f"RESPONSE: {response.json()}") if response.status_code != 200: response_message = ( response.json()["message"] if "message" in response.json() else "Unknown error" ) raise Exception(f"Failed to update Gitea group: {response_message}") else: group = response.json() return group # Endpoint: PATCH /api/{GITEA_API_VERSION}/admin/users/{username} def update_gitea_user(data: dict) -> dict: if not data: raise Exception("Data is missing!") username = data["username"] current_user = get_gitea_user(username) updated_user = convert_gitlab_user_to_gitea(data) to_compare = { "active": True, "login": True, "login_name": True, "username": True, "email": True, "full_name": True, "admin": True, "prohibit_login": True, "restricted": True, "website": True, "visibility": True, } mapping = { "is_admin": "admin", } ignore = {"allow_create_organization": True} is_different, results = cmp_dicts( current_user, updated_user, to_compare, mapping, ignore ) _trace(f"COMPARISON: {results}") if not is_different: _info(f'User "{username}" is already up-to-date on Gitea') return current_user if is_gitea_reserved_username(username): _warn(f'User "{username}" is a reserved username on Gitea!') return None if not current_user: raise Exception(f'User "{username}" not found on Gitea!') _debug(f"REQUEST: PATCH {GITEA_URL}/api/{GITEA_API_VERSION}/admin/users/{username}") # TODO seems to have issues, since users do not get updated properly response = requests.patch( f"{GITEA_URL}/api/{GITEA_API_VERSION}/admin/users/{username}", json=updated_user, headers={ "Content-Type": "application/json", "Accept": "application/json", "Authorization": f"token {GITEA_TOKEN}", }, ) _trace(f"RESPONSE: {response.json()}") if response.status_code != 200: response_message = ( response.json()["message"] if "message" in response.json() else "Unknown error" ) raise Exception(f"Failed to update Gitea user: {response_message}") else: user = response.json() return user def update_gitea_project(data: dict) -> dict: _warn("Function update_gitea_project is not implemented yet!") pass # Endpoint: GET /api/{GITEA_API_VERSION}/users/{username} def get_gitea_user(username: str) -> dict: _debug(f"REQUEST: GET {GITEA_URL}/api/{GITEA_API_VERSION}/users/{username}") response = requests.get( f"{GITEA_URL}/api/{GITEA_API_VERSION}/users/{username}", headers={ "Content-Type": "application/json", "Accept": "application/json", "Authorization": f"token {GITEA_TOKEN}", }, ) _trace(f"RESPONSE: {response.json()}") if response.status_code != 200: response_message = ( response.json()["message"] if "message" in response.json() else "Unknown error" ) raise Exception(f"Failed to get Gitea user: {response_message}") else: user = response.json() return user def remap_keys(d, mapping): def nested_get(d, keys): for key in keys.split("/"): if EXIT_REQUESTED: return d = d[key] return d def nested_set(d, keys, value): keys = keys.split("/") for key in keys[:-1]: if EXIT_REQUESTED: return d = d.setdefault(key, {}) d[keys[-1]] = value remapped = {} for k, v in d.items(): if EXIT_REQUESTED: return if k in mapping: nested_set(remapped, mapping[k], v) elif isinstance(v, dict): remapped[k] = remap_keys( v, { f"{k}/{subkey}": subvalue for subkey, subvalue in mapping.items() if subkey.startswith(f"{k}/") }, ) else: remapped[k] = v return remapped def is_ignored(path_to_check, ignore_dict): def nested_get(d, keys): for key in keys: if EXIT_REQUESTED: return if key in d: d = d[key] else: return False return True if isinstance(d, dict) and d else False return nested_get(ignore_dict, path_to_check) def cmp_dicts( dict_a: dict, dict_b: dict, check: dict = None, mapping: dict = None, ignore: dict = None, ) -> bool: result = {} if mapping: dict_a = remap_keys(dict_a, mapping) def compare_dicts(a, b, to_check, path=[]): all_keys = set(a.keys()).union(b.keys()) if to_check is not None: all_keys = all_keys.intersection(to_check.keys()) for key in all_keys: if EXIT_REQUESTED: return current_path = path + [key] if ignore and is_ignored(current_path, ignore): continue if key in a and key in b: if ( isinstance(a[key], dict) and isinstance(b[key], dict) and to_check and key in to_check ): compare_dicts(a[key], b[key], to_check[key], current_path) else: result[tuple(current_path)] = a[key] != b[key] else: result[tuple(current_path)] = True compare_dicts(dict_a, dict_b, check) has_changes = any(result.values()) return has_changes, result def convert_gitlab_user_to_gitea(user: dict, extra_data: dict = None) -> dict: gitea_user = { "active": user["state"] == "active", "login": user["username"], "login_name": user["username"], "username": user["username"], "email": user["email"], "full_name": user["name"], "admin": user["is_admin"], "prohibit_login": user["state"] == "blocked" or user["locked"], "restricted": user["state"] == "blocked" or user["locked"], "website": user["website_url"], "visibility": "private" if user["private_profile"] else "public", "allow_create_organization": user["can_create_group"], } if extra_data: gitea_user.update(extra_data) return gitea_user def cmp_gitlab_gitea_groups(gitlab_groups: list, gitea_groups: list) -> dict: compare_result = {} missing_matches = 0 for gitlab_group in gitlab_groups: if EXIT_REQUESTED: return name = gitlab_group["path"] exists = False for gitea_group in gitea_groups: if EXIT_REQUESTED: return if name == gitea_group["name"]: exists = True break if exists: _info(f'GITLAB: Group "{name}" exists on both GitLab and Gitea') else: _warn(f'GITLAB: Group "{name}" exists on GitLab only') missing_matches += 1 compare_result[name] = 0 if exists else 1 for gitea_group in gitea_groups: if EXIT_REQUESTED: return name = gitea_group["name"] exists = False for gitlab_group in gitlab_groups: if EXIT_REQUESTED: return if name == gitlab_group["path"]: exists = True break if not exists: _warn(f'GITEA: Group "{name}" exists on Gitea only') compare_result[name] = 2 return compare_result, missing_matches def cmp_gitlab_gitea_users(gitlab_users: list, gitea_users: list) -> dict: compare_result = {} missing_matches = 0 for gitlab_user in gitlab_users: if EXIT_REQUESTED: return name = gitlab_user["username"] exists = False for gitea_user in gitea_users: if EXIT_REQUESTED: return if name == gitea_user["login"]: exists = True break if exists: _info(f'GITLAB: User "{name}" exists on both GitLab and Gitea') else: _warn(f'GITLAB: User "{name}" exists on GitLab only') missing_matches += 1 compare_result[name] = 0 if exists else 1 for gitea_user in gitea_users: if EXIT_REQUESTED: return name = gitea_user["login"] exists = False for gitlab_user in gitlab_users: if EXIT_REQUESTED: return if name == gitlab_user["username"]: exists = True break if not exists: _warn(f'GITEA: User "{name}" exists on Gitea only') compare_result[name] = 2 return compare_result, missing_matches def cmp_gitlab_gitea_projects(gitlab_projects: list, gitea_projects: list) -> dict: compare_result = {} missing_matches = 0 for gitlab_project in gitlab_projects: if EXIT_REQUESTED: return exists = False name = gitlab_project["path"] owner = gitlab_project["namespace"]["path"] for gitea_project in gitea_projects["data"]: if EXIT_REQUESTED: return # fmt: off if (name == gitea_project["name"] and owner == gitea_project["owner"]["username"]): exists = True break if exists: _info(f'GITLAB: Project "{owner}/{name}" exists on both GitLab and Gitea') else: _warn(f'GITLAB: Project "{owner}/{name}" exists on GitLab only') missing_matches += 1 compare_result[name] = 0 if exists else 1 for gitea_project in gitea_projects["data"]: if EXIT_REQUESTED: return exists = False name = gitea_project["name"] owner = gitea_project["owner"]["username"] for gitlab_project in gitlab_projects: if EXIT_REQUESTED: return # fmt: off if name == gitlab_project["path"] and owner == gitlab_project["namespace"]["path"]: exists = True break # fmt: on if not exists: _warn(f'GITEA: Project "{owner}/{name}" exists on Gitea only') compare_result[name] = 2 return compare_result, missing_matches def create_missing_groups(gitlab_groups: list, gitea_groups: list): for gitlab_group in gitlab_groups: if EXIT_REQUESTED: return name = gitlab_group["path"] exists = False for gitea_group in gitea_groups: if EXIT_REQUESTED: return if name == gitea_group["name"]: exists = True break if is_gitea_reserved_organame(name): _warn(f'Skipping group "{name}": Group name is reserved on Gitea!') STATS["groups"]["skipped"].append({"name": name, "reason": "Reserved"}) continue if not exists: _info(f'Creating missing group "{name}" on Gitea...') try: _info(f'Migrating Gitlab group "{name}" to Gitea...') migrate_gitlab_group_to_gitea(gitlab_group) _info(f'Group "{name}" created on Gitea') except Exception as e: _exception(f'Failed to create Gitea group "{name}": {e}', e) else: if OVERRIDE_EXISTING_GROUPS: try: _info( f'Group "{name}" already exists on Gitea and will be overridden...' ) delete_gitea_group(name) _info(f'Group "{name}" deleted on Gitea') _info(f'Creating missing group "{name}" on Gitea...') migrate_gitlab_group_to_gitea(gitlab_group) _info(f'Group "{name}" created on Gitea') except Exception as e: _exception(f'Failed to override Gitea group "{name}": {e}', e) def create_missing_users(gitlab_users: list, gitea_users: list): for gitlab_user in gitlab_users: if EXIT_REQUESTED: return name = gitlab_user["username"] exists = False for gitea_user in gitea_users["data"]: if EXIT_REQUESTED: return if name == gitea_user["login"]: exists = True break if gitea_user["email"] == None or gitea_user["email"] == "": _warn( f'User "{name}" does not have an email address and will not be created!' ) STATS["users"]["skipped"].append( {"username": name, "reason": "No email address"} ) continue if gitea_user["username"] == None or gitea_user["username"] == "": _warn(f'User "{name}" does not have a username and will not be created!') STATS["users"]["skipped"].append( {"username": name, "reason": "No username"} ) continue if is_gitea_reserved_username(name): _warn(f'Skipping user "{name}": Username is reserved on Gitea!') STATS["users"]["skipped"].append({"username": name, "reason": "Reserved"}) continue if not exists: _info(f'Creating missing user "{name}" on Gitea...') try: _info(f'Migrating Gitlab user "{name}" to Gitea...') migrate_gitlab_user_to_gitea(gitlab_user) _info(f'User "{name}" created on Gitea') except Exception as e: _exception(f'Failed to create Gitea user "{name}": {e}', e) else: if OVERRIDE_EXISTING_USERS: try: _info( f'User "{name}" already exists on Gitea and will be overridden...' ) delete_gitea_user(name) _info(f'User "{name}" deleted on Gitea') _info(f'Creating missing user "{name}" on Gitea...') migrate_gitlab_user_to_gitea(gitlab_user) _info(f'User "{name}" created on Gitea') except Exception as e: _exception(f'Failed to override Gitea user "{name}": {e}', e) def create_missing_projects(gitlab_projects: list, gitea_projects: list): for gitlab_project in gitlab_projects: if EXIT_REQUESTED: return exists = False name = gitlab_project["path"] group = gitlab_project["namespace"]["path"] for gitea_project in gitea_projects["data"]: if EXIT_REQUESTED: return if ( name == gitea_project["name"] and group == gitea_project["owner"]["username"] ): exists = True break if is_gitea_reserved_reponame(name): _warn( f'Skipping project "{group}/{name}": Project name is reserved on Gitea!' ) STATS["projects"]["skipped"].append( {"group": group, "name": name, "reason": "Reserved"} ) continue if is_gitlab_project_in_subgroup(gitlab_project): _warn( f'Skipping project "{group}/{name}": Project is in a subgroup and not supported by Gitea!' ) STATS["projects"]["skipped"].append( {"group": group, "name": name, "reason": "Unsupported Subgroup"} ) continue if not exists: _info(f'Creating missing project "{group}/{name}" on Gitea...') try: _info(f'Migrating Gitlab project "{group}/{name}" to Gitea...') migrate_gitlab_project_to_gitea(gitlab_project) _info(f'Project "{group}/{name}" created on Gitea') if ONLY_ONE_PROJECT: _warn("DEBUG MODE ENABLED - BREAKING AFTER FIRST PROJECT") break except Exception as e: _exception(f'Failed to create Gitea project "{group}/{name}": {e}', e) if ONLY_ONE_PROJECT: _warn("DEBUG MODE ENABLED - BREAKING AFTER FIRST PROJECT") break else: if OVERRIDE_EXISTING_PROJECTS: try: _info( f'Project "{group}/{name}" already exists on Gitea and will be overridden...' ) delete_gitea_project(gitea_project) _info(f'Project "{group}/{name}" deleted on Gitea') _info(f'Creating missing project "{group}/{name}" on Gitea...') migrate_gitlab_project_to_gitea(gitlab_project) _info(f'Project "{group}/{name}" created on Gitea') if ONLY_ONE_PROJECT: _warn("DEBUG MODE ENABLED - BREAKING AFTER FIRST PROJECT") break except Exception as e: _exception( f'Failed to override Gitea project "{group}/{name}": {e}', e ) if ONLY_ONE_PROJECT: _warn("DEBUG MODE ENABLED - BREAKING AFTER FIRST PROJECT") break def update_existing_groups(gitlab_groups: list, gitea_groups: list): for gitlab_group in gitlab_groups: if EXIT_REQUESTED: return name = gitlab_group["path"] exists = False for gitea_group in gitea_groups: if EXIT_REQUESTED: return if name == gitea_group["name"]: exists = True break if is_gitea_reserved_organame(name): _warn(f'Skipping group "{name}": Group name is reserved on Gitea!') continue if exists: try: _info(f'Updating existing group "{name}" on Gitea...') update_gitea_org( { "path": gitlab_group["path"], "description": gitlab_group["description"], "website": gitlab_group["web_url"], "visibility": gitlab2gitea_visibility( gitlab_group["visibility"] ), "full_name": gitlab_group["full_name"], }, ) _info(f'Group "{name}" updated on Gitea') except Exception as e: _exception(f'Failed to update Gitea group "{name}": {e}', e) def update_existing_users(gitlab_users: list, gitea_users: list): for gitlab_user in gitlab_users: if EXIT_REQUESTED: return name = gitlab_user["username"] exists = False for gitea_user in gitea_users["data"]: if EXIT_REQUESTED: return if name == gitea_user["login"]: exists = True break if gitea_user["email"] == None or gitea_user["email"] == "": _warn( f'User "{name}" does not have an email address and will not be updated!' ) continue if gitea_user["username"] == None or gitea_user["username"] == "": _warn(f'User "{name}" does not have a username and will not be updated!') continue if is_gitea_reserved_username(name): _warn(f'Skipping user "{name}": Username is reserved on Gitea!') continue if exists: try: _info(f'Updating existing user "{name}" on Gitea...') update_gitea_user(gitlab_user) _info(f'User "{name}" updated on Gitea') except Exception as e: _exception(f'Failed to update Gitea user "{name}": {e}', e) else: _warn(f'User "{name}" does not exist on Gitea!') def update_existing_projects(gitlab_projects: list, gitea_projects: list): # update existing projects for gitlab_project in gitlab_projects: if EXIT_REQUESTED: return exists = False name = gitlab_project["path"] group = gitlab_project["namespace"]["path"] for gitea_project in gitea_projects["data"]: if EXIT_REQUESTED: return if ( name == gitea_project["name"] and group == gitea_project["owner"]["username"] ): exists = True break if is_gitea_reserved_reponame(name): _warn( f'Skipping project "{group}/{name}": Project name is reserved on Gitea!' ) continue if is_gitlab_project_in_subgroup(gitlab_project): _warn( f'Skipping project "{group}/{name}": Project is in a subgroup and not supported by Gitea!' ) continue if exists: try: _info(f'Updating existing project "{group}/{name}" on Gitea...') update_gitea_project(gitlab_project) _info(f'Project "{group}/{name}" updated on Gitea') if ONLY_ONE_PROJECT: _warn("DEBUG MODE ENABLED - BREAKING AFTER FIRST PROJECT") break except Exception as e: _exception(f'Failed to update Gitea project "{group}/{name}": {e}', e) if ONLY_ONE_PROJECT: _warn("DEBUG MODE ENABLED - BREAKING AFTER FIRST PROJECT") break else: _warn(f'Project "{group}/{name}" does not exist on Gitea!') # ENDPOINT: DELETE /api/{GITEA_API_VERSION}/orgs/{org} def delete_gitea_group(name: str): _debug(f"REQUEST: DELETE {GITEA_URL}/api/{GITEA_API_VERSION}/orgs/{name}") response = requests.delete( f"{GITEA_URL}/api/{GITEA_API_VERSION}/orgs/{name}", headers={ "Content-Type": "application/json", "Accept": "application/json", "Authorization": f"token {GITEA_TOKEN}", }, ) if response.status_code != 204: _trace(f"RESPONSE: {response.json()}") response_message = ( response.json()["message"] if "message" in response.json() else "Unknown error" ) STATS["groups"]["errors"].append( { "name": name, "error": response_message if response_message else "Unknown error", } ) raise Exception(f"Failed to delete Gitea group: {response_message}") else: _info(f'Group "{name}" deleted on Gitea') STATS["groups"]["deleted"].append({"name": name}) # ENDPOINT: DELETE /api/{GITEA_API_VERSION}/admin/users/{username} def delete_gitea_user(username: str): _debug( f"REQUEST: DELETE {GITEA_URL}/api/{GITEA_API_VERSION}/admin/users/{username}" ) response = requests.delete( f"{GITEA_URL}/api/{GITEA_API_VERSION}/admin/users/{username}", headers={ "Content-Type": "application/json", "Accept": "application/json", "Authorization": f"token {GITEA_TOKEN}", }, ) if response.status_code != 204: _trace(f"RESPONSE: {response.json()}") response_message = ( response.json()["message"] if "message" in response.json() else "Unknown error" ) STATS["users"]["errors"].append( { "username": username, "error": response_message if response_message else "Unknown error", } ) raise Exception(f"Failed to delete Gitea user: {response_message}") else: _info(f'User "{username}" deleted on Gitea') STATS["users"]["deleted"].append({"username": username}) # ENDPOINT: DELETE /api/{GITEA_API_VERSION}/repos/{owner}/{repo} def delete_gitea_project(project: dict): owner = project["owner"]["username"] repo = project["name"] _debug(f"REQUEST: DELETE {GITEA_URL}/api/{GITEA_API_VERSION}/repos/{owner}/{repo}") response = requests.delete( f"{GITEA_URL}/api/{GITEA_API_VERSION}/repos/{owner}/{repo}", headers={ "Content-Type": "application/json", "Accept": "application/json", "Authorization": f"token {GITEA_TOKEN}", }, ) if response.status_code != 204: _trace(f"RESPONSE: {response.json()}") response_message = ( response.json()["message"] if "message" in response.json() else "Unknown error" ) STATS["projects"]["errors"].append( { "group": owner, "name": repo, "error": response_message if response_message else "Unknown error", } ) raise Exception(f"Failed to delete Gitea project: {response_message}") else: _info(f'Project "{owner}/{repo}" deleted on Gitea') STATS["projects"]["deleted"].append({"group": owner, "name": repo}) def migrate_groups(): if OVERRIDE_EXISTING_GROUPS: _warn("EXISTING GROUPS WILL BE OVERRIDDEN!") gitlab_groups = get_gitlab_groups() gitea_groups = get_gitea_groups() _info(f"Groups on GitLab: {len(gitlab_groups)}") _trace(f"Groups on GitLab: {gitlab_groups}") _info(f"Groups on Gitea: {len(gitea_groups)}") _trace(f"Groups on Gitea: {gitea_groups}") try: group_result, missing_matches = cmp_gitlab_gitea_groups( gitlab_groups, gitea_groups ) except Exception as e: _exception(f"Failed to compare GitLab and Gitea groups: {e}", e) return if missing_matches > 0: _warn(f"{missing_matches} groups are missing on Gitea!") if OVERRIDE_EXISTING_GROUPS or ( missing_matches > 0 and not NO_CREATE_MISSING_GROUPS ): _info("Creating missing groups on Gitea...") if not DRY_RUN: try: create_missing_groups(gitlab_groups, gitea_groups) if EXIT_REQUESTED: return except Exception as e: _exception(f"Failed to create missing groups: {e}", e) else: _warn( "Dry-run mode enabled, skipping creation of missing groups on Gitea..." ) if not NO_UPDATE_EXISTING_GROUPS: _info("Updating existing groups on Gitea...") try: if not DRY_RUN: update_existing_groups(gitlab_groups, gitea_groups) if EXIT_REQUESTED: return else: _warn( "Dry-run mode enabled, skipping update of existing groups on Gitea..." ) except Exception as e: _exception(f"Failed to update existing groups: {e}", e) def migrate_users(): if OVERRIDE_EXISTING_USERS: _warn("EXISTING USERS WILL BE OVERRIDDEN!") gitlab_users = get_gitlab_users() gitea_users = get_gitea_users() _info(f"Users on GitLab: {len(gitlab_users)}") _trace(f"Users on GitLab: {gitlab_users}") _info(f"Users on Gitea: {len(gitea_users)}") _trace(f"Users on Gitea: {gitea_users}") try: user_result, missing_matches = cmp_gitlab_gitea_users( gitlab_users, gitea_users["data"] ) except Exception as e: _exception(f"Failed to compare GitLab and Gitea users: {e}", e) return if missing_matches > 0: _warn(f"{missing_matches} users are missing on Gitea!") if OVERRIDE_EXISTING_USERS or (missing_matches > 0 and not NO_CREATE_MISSING_USERS): _info("Creating missing users on Gitea...") if not DRY_RUN: create_missing_users(gitlab_users, gitea_users) if EXIT_REQUESTED: return else: _warn( "Dry-run mode enabled, skipping creation of missing users on Gitea..." ) if not NO_UPDATE_EXISTING_USERS: _info("Updating existing users on Gitea...") try: if not DRY_RUN: update_existing_users(gitlab_users, gitea_users) if EXIT_REQUESTED: return else: _warn( "Dry-run mode enabled, skipping update of existing users on Gitea..." ) except Exception as e: _exception(f"Failed to update existing users: {e}", e) def migrate_projects(): if OVERRIDE_EXISTING_PROJECTS: _warn("EXISTING PROJECTS WILL BE OVERRIDDEN!") gitlab_projects = get_gitlab_projects() if EXIT_REQUESTED: return gitea_projects = get_gitea_projects() if EXIT_REQUESTED: return # dump the projects to json files with open("gitlab_projects.json", "w") as f: json.dump(gitlab_projects, f, indent=4) if EXIT_REQUESTED: return with open("gitea_projects.json", "w") as f: json.dump(gitea_projects, f, indent=4) if EXIT_REQUESTED: return _info(f"Projects on GitLab: {len(gitlab_projects)}") _trace(f"Projects on GitLab: {gitlab_projects}") _info(f"Projects on Gitea: {len(gitea_projects)}") _trace(f"Projects on Gitea: {gitea_projects}") try: project_result, missing_matches = cmp_gitlab_gitea_projects( gitlab_projects, gitea_projects ) if EXIT_REQUESTED: return except Exception as e: _exception(f"Failed to compare GitLab and Gitea projects: {e}", e) return if missing_matches > 0: _warn(f"{missing_matches} projects are missing on Gitea!") if OVERRIDE_EXISTING_PROJECTS or ( missing_matches > 0 and not NO_CREATE_MISSING_PROJECTS ): _info("Creating missing projects on Gitea...") if not DRY_RUN: create_missing_projects(gitlab_projects, gitea_projects) if EXIT_REQUESTED: return else: _warn( "Dry-run mode enabled, skipping creation of missing projects on Gitea..." ) if not NO_UPDATE_EXISTING_PROJECTS: _info("Updating existing projects on Gitea...") try: if not DRY_RUN: update_existing_projects(gitlab_projects, gitea_projects) if EXIT_REQUESTED: return else: _warn( "Dry-run mode enabled, skipping update of existing projects on Gitea..." ) except Exception as e: _exception(f"Failed to update existing projects: {e}", e) def run_migration(): if ONLY_GROUPS: _warn("Skipping users!") _warn("Skipping projects!") _info("Migrating GitLab groups...") migrate_groups() _info("Group migration completed!") return elif ONLY_USERS: _warn("Skipping groups!") _warn("Skipping projects!") _info("Migrating GitLab users...") migrate_users() _info("User migration completed!") return elif ONLY_PROJECTS: _warn("Skipping groups!") _warn("Skipping users!") _info("Migrating GitLab projects...") migrate_projects() _info("Project migration completed!") return else: _info("Migrating GitLab groups...") migrate_groups() _info("Group migration completed!") if EXIT_REQUESTED: return _info("Migrating GitLab users...") migrate_users() _info("User migration completed!") if EXIT_REQUESTED: return _info("Migrating GitLab projects...") migrate_projects() _info("Project migration completed!") def print_stats(): _info("Migration Statistics:") _info("") _info("Groups:") _info(f" - Created: {len(STATS['groups']['created'])}") _info(f" - Updated: {len(STATS['groups']['updated'])}") _info(f" - Deleted: {len(STATS['groups']['deleted'])}") _info(f" - Skipped: {len(STATS['groups']['skipped'])}") _info(f" - Errors: {len(STATS['groups']['errors'])}") _info("") _info("Errors:") for error in STATS["groups"]["errors"]: _error(f' - Group "{error["name"]}": {error["error"]}') _info("Users:") _info(f" - Created: {len(STATS['users']['created'])}") _info(f" - Updated: {len(STATS['users']['updated'])}") _info(f" - Deleted: {len(STATS['users']['deleted'])}") _info(f" - Skipped: {len(STATS['users']['skipped'])}") _info(f" - Errors: {len(STATS['users']['errors'])}") _info("") _info("Errors:") for error in STATS["users"]["errors"]: _error(f' - User "{error["username"]}": {error["error"]}') _info("Projects:") _info(f" - Created: {len(STATS['projects']['created'])}") _info(f" - Updated: {len(STATS['projects']['updated'])}") _info(f" - Deleted: {len(STATS['projects']['deleted'])}") _info(f" - Skipped: {len(STATS['projects']['skipped'])}") _info(f" - Errors: {len(STATS['projects']['errors'])}") _info("") _info("Errors:") for error in STATS["projects"]["errors"]: _error(f' - Project "{error["group"]}/{error["name"]}": {error["error"]}') def main(): signal.signal(signal.SIGINT, signal_handler) header_info = f"{APP_NAME} v{APP_VERSION} - by {APP_AUTHOR}" _info(f"{header_info}") _info("-" * len(header_info)) _info("") if sys.version_info < (3, 6): _error("Python 3.6 or higher is required!") return else: _debug(f"Python version: {sys.version_info.major}.{sys.version_info.minor}") if not GITLAB_TOKEN: _error("GitLab access token is missing!") return if not GITEA_TOKEN: _error("Gitea access token is missing!") return if LOG_FILE and not APPEND_LOG: with open(LOG_FILE, "w") as log_file: log_file.write("") _debug(f"GitLab URL: {GITLAB_URL}") _debug(f"Gitea URL: {GITEA_URL}") _debug(f"Logging to: {LOG_FILE}") if DRY_RUN: _warn("Running in dry-run mode!") _info("Testing endpoints...") try: check_gitlab() check_gitea() except Exception as e: _exception(f"An error occurred: {e}", e) return _info("Starting migration...") try: run_migration() _info("Migration completed!") _info("") except Exception as e: _exception(f"An error occurred: {e}", e) finally: print_stats() def signal_handler(sig, frame): global EXIT_REQUESTED if EXIT_REQUESTED: _error("EXITING IMMEDIATELY!") try: sys.exit(1) except SystemExit: os._exit(1) else: _warn("EXIT REQUESTED! Please wait...") EXIT_REQUESTED = True if __name__ == "__main__": main()