2398 lines
70 KiB
Python
2398 lines
70 KiB
Python
# Description: Script to migrate repositories from GitLab to Gitea
|
|
# Author: Enrico Ludwig <enrico.ludwig@zion-networks.de>
|
|
# 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_TOKEN> GitLab access token
|
|
# --gitlab-url <GITLAB_URL> GitLab URL
|
|
# --gitlab-api-version <GITLAB_API_VERSION> GitLab API version (default: v4)
|
|
# --gitea-token <GITEA_TOKEN> Gitea access token
|
|
# --gitea-url <GITEA_URL> Gitea URL
|
|
# --gitea-api-version <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> 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_URL> \
|
|
# --gitlab-token <GITLAB_TOKEN> \
|
|
# --gitea-url <GITEA_URL> \
|
|
# --gitea-token <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()
|