From e870b993217bc735b1cc4bb44508e191b1eab5bb Mon Sep 17 00:00:00 2001 From: Enrico Ludwig Date: Mon, 15 Jul 2024 11:36:01 +0200 Subject: [PATCH] Added params for functions to be implemented --- gitlab2gitea/gitlab2gitea.py | 851 +++++++++++++++++++++-------------- 1 file changed, 504 insertions(+), 347 deletions(-) diff --git a/gitlab2gitea/gitlab2gitea.py b/gitlab2gitea/gitlab2gitea.py index d160547..0c2d796 100644 --- a/gitlab2gitea/gitlab2gitea.py +++ b/gitlab2gitea/gitlab2gitea.py @@ -25,6 +25,10 @@ # --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) # +# --override-groups Override existing groups on Gitea (default: False) - not implemented yet +# --override-users Override existing users on Gitea (default: False) - not implemented yet +# --override-projects Override existing projects on Gitea (default: False) - not implemented yet +# # --quiet Enable quiet mode (default: False) # --debug Enable debug mode (default: False) # --trace Enable trace mode (default: False) @@ -42,15 +46,15 @@ # Settings - GitLab -GITLAB_URL = '' -GITLAB_TOKEN = '' -GITLAB_API_VERSION = 'v4' +GITLAB_URL = "" +GITLAB_TOKEN = "" +GITLAB_API_VERSION = "v4" # Settings - Gitea -GITEA_URL = '' -GITEA_TOKEN = '' -GITEA_API_VERSION = 'v1' +GITEA_URL = "" +GITEA_TOKEN = "" +GITEA_API_VERSION = "v1" # Settings - General Repository @@ -67,7 +71,7 @@ NO_UPDATE_EXISTING_PROJECTS = False DEBUG = False TRACE = False DRY_RUN = False -LOG_FILE = 'gitlab2gitea.log' +LOG_FILE = "gitlab2gitea.log" APPEND_LOG = False QUIET = False @@ -84,156 +88,182 @@ import traceback 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_URL" in os.environ: + GITLAB_URL = os.environ["GITLAB_URL"] -if 'GITLAB_TOKEN' in os.environ: - GITLAB_TOKEN = os.environ['GITLAB_TOKEN'] +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 "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_URL" in os.environ: + GITEA_URL = os.environ["GITEA_URL"] -if 'GITEA_TOKEN' in os.environ: - GITEA_TOKEN = os.environ['GITEA_TOKEN'] +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 "GITEA_API_VERSION" in os.environ: + GITEA_API_VERSION = os.environ["GITEA_API_VERSION"] -if 'NO_CREATE_MISSING_GROUPS' in os.environ: +if "NO_CREATE_MISSING_GROUPS" in os.environ: NO_CREATE_MISSING_GROUPS = False -if 'NO_CREATE_MISSING_USERS' in os.environ: +if "NO_CREATE_MISSING_USERS" in os.environ: NO_CREATE_MISSING_USERS = False -if 'NO_CREATE_MISSING_PROJECTS' in os.environ: +if "NO_CREATE_MISSING_PROJECTS" in os.environ: NO_CREATE_MISSING_PROJECTS = False -if 'NO_UPDATE_EXISTING_GROUPS' in os.environ: +if "NO_UPDATE_EXISTING_GROUPS" in os.environ: NO_UPDATE_EXISTING_GROUPS = False -if 'NO_UPDATE_EXISTING_USERS' in os.environ: +if "NO_UPDATE_EXISTING_USERS" in os.environ: NO_UPDATE_EXISTING_USERS = False -if 'NO_UPDATE_EXISTING_PROJECTS' in os.environ: +if "NO_UPDATE_EXISTING_PROJECTS" in os.environ: NO_UPDATE_EXISTING_PROJECTS = False -if 'QUIET' in os.environ: - QUIET = bool(os.environ['QUIET']) +if "QUIET" in os.environ: + QUIET = bool(os.environ["QUIET"]) -if 'DEBUG' in os.environ: - DEBUG = bool(os.environ['DEBUG']) +if "DEBUG" in os.environ: + DEBUG = bool(os.environ["DEBUG"]) -if 'TRACE' in os.environ: - TRACE = bool(os.environ['TRACE']) +if "TRACE" in os.environ: + TRACE = bool(os.environ["TRACE"]) -if 'DRY_RUN' in os.environ: - DRY_RUN = bool(os.environ['DRY_RUN']) +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 "LOG_FILE" in os.environ: + LOG_FILE = os.environ["LOG_FILE"] -if 'APPEND_LOG' in os.environ: - APPEND_LOG = bool(os.environ['APPEND_LOG']) +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: +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: + if line.startswith("#") or not "=" in line: continue - key, value = line.strip().split('=') + key, value = line.strip().split("=") - if key == 'GITLAB_API_VERSION': + if key == "GITLAB_API_VERSION": GITLAB_API_VERSION = value - if key == 'GITLAB_URL': + if key == "GITLAB_URL": GITLAB_URL = value - if key == 'GITLAB_TOKEN': + if key == "GITLAB_TOKEN": GITLAB_TOKEN = value - if key == 'GITEA_URL': + if key == "GITEA_URL": GITEA_URL = value - if key == 'GITEA_TOKEN': + if key == "GITEA_TOKEN": GITEA_TOKEN = value - - if key == 'GITEA_API_VERSION': + + if key == "GITEA_API_VERSION": GITEA_API_VERSION = value - - if key == 'NO_CREATE_MISSING_GROUPS': - if value.lower() == 'true' or value == '1': + + 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': + + 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': + 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': + 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': + 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': + if key == "NO_UPDATE_EXISTING_PROJECTS": + if value.lower() == "true" or value == "1": NO_UPDATE_EXISTING_PROJECTS = True - if key == 'QUIET': + if key == "QUIET": QUIET = bool(value) - if key == 'DEBUG': - if value.lower() == 'true' or value == '1': + if key == "DEBUG": + if value.lower() == "true" or value == "1": DEBUG = True - - if key == 'TRACE': - if value.lower() == 'true' or value == '1': + + if key == "TRACE": + if value.lower() == "true" or value == "1": TRACE = True - if key == 'DRY_RUN': - if value.lower() == 'true' or value == '1': + if key == "DRY_RUN": + if value.lower() == "true" or value == "1": DRY_RUN = True - if key == 'LOG_FILE': + if key == "LOG_FILE": LOG_FILE = value - - if key == 'APPEND_LOG': + + 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') +parser = argparse.ArgumentParser( + description="Script to migrate repositories from GitLab to Gitea" +) -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('--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') +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("--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") args = parser.parse_args() @@ -293,47 +323,53 @@ if args.append_log: # Remove trailing slashes from URLs -GITLAB_URL = GITLAB_URL.rstrip('/') -GITEA_URL = GITEA_URL.rstrip('/') +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) + 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') + 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) + 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') + 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) + 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') + 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) + 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') + 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) + 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') + 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() @@ -342,303 +378,388 @@ def _exception(exception, custom_message=None): formatted_traceback = traceback.format_exc() # Prepare the exception message - 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' + 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" ) # Print the exception message to the console print(exception_message) - + # Write the exception message to the log file if defined if LOG_FILE: - with open(LOG_FILE, 'a') as log_file: - log_file.write(f'[EXC] {exception_message}\n') + with open(LOG_FILE, "a") as log_file: + log_file.write(f"[EXC] {exception_message}\n") + # PROGRAM + def gitlab2gitea_visibility(visibility: str) -> str: - if visibility == 'private': - return 'private' - elif visibility == 'internal': - return 'limited' - elif visibility == 'public': - return 'public' + if visibility == "private": + return "private" + elif visibility == "internal": + return "limited" + elif visibility == "public": + return "public" else: - return 'private' + return "private" + # Endpoint: GET /api/{GITLAB_API_VERSION}/version def check_gitlab(): - _debug(f'REQUEST: GET {GITLAB_URL}/api/{GITLAB_API_VERSION}/version') + _debug(f"REQUEST: GET {GITLAB_URL}/api/{GITLAB_API_VERSION}/version") - response = requests.get(f'{GITLAB_URL}/api/{GITLAB_API_VERSION}/version', headers={ - 'Content-Type': 'application/json', - 'Accept': 'application/json', - 'Authorization': f'Bearer {GITLAB_TOKEN}' - }) + 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()}') + _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'GitLab endpoint test failed: {response_message}') + 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') + _debug(f"REQUEST: GET {GITEA_URL}/api/{GITEA_API_VERSION}/version") - response = requests.get(f'{GITEA_URL}/api/{GITEA_API_VERSION}/version', headers={ - 'Content-Type': 'application/json', - 'Accept': 'application/json', - 'Authorization': f'token {GITEA_TOKEN}' - }) + 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()}") - _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'Gitea endpoint test failed: {response_message}') + 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') + _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 - }, + 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}' - }) + "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}') + 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}') + _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}', + 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}' - }) + "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}') + 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' + + 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, + 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 + "Content-Type": "application/json", + "Accept": "application/json", + "Authorization": f"Bearer {GITLAB_TOKEN}", + }, + ) - _trace(f'RESPONSE: {response.json()}') + 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}') + 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' + + next_page_link = f"{GITLAB_URL}/api/{GITLAB_API_VERSION}/projects" projects = [] while next_page_link is not None: _debug(f'REQUEST: GET {next_page_link.split("?")[0]}') - response = requests.get(next_page_link, - params={ - 'all_available': 1, - 'per_page': 100 - }, + 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()}') + "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}') + 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/{GITEA_API_VERSION}/admin/users def migrate_gitlab_user_to_gitea(user: dict): pass + # 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') + _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', + response = requests.get( + f"{GITEA_URL}/api/{GITEA_API_VERSION}/orgs", headers={ - 'Content-Type': 'application/json', - 'Accept': 'application/json', - 'Authorization': f'token {GITEA_TOKEN}' - }) + "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}') + 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') + _debug(f"REQUEST: GET {GITEA_URL}/api/{GITEA_API_VERSION}/users/search") - response = requests.get(f'{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()}') + "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}') + 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') + _debug(f"REQUEST: GET {GITEA_URL}/api/{GITEA_API_VERSION}/repos/search") - response = requests.get(f'{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()}') + "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}') + 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!') - + raise Exception("GitLab group is missing!") + # Create Gitea group - _debug(f'REQUEST: POST {GITEA_URL}/api/{GITEA_API_VERSION}/orgs') + _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}' - }) + 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()}') + _trace(f"RESPONSE: {response.json()}") if response.status_code != 201: - response_message = response.json()['message'] if 'message' in response.json() else 'Unknown error' - raise Exception(f'Failed to create Gitea group: {response_message}') + response_message = ( + response.json()["message"] + if "message" in response.json() + else "Unknown error" + ) + raise Exception(f"Failed to create Gitea group: {response_message}") else: group = response.json() 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'] + raise Exception("Data is missing!") - _debug(f'REQUEST: PATCH {GITEA_URL}/api/{GITEA_API_VERSION}/orgs/{name}') + name = data["path"] - 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}' - }) + _debug(f"REQUEST: PATCH {GITEA_URL}/api/{GITEA_API_VERSION}/orgs/{name}") - _trace(f'RESPONSE: {response.json()}') + 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}') + 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 + def cmp_gitlab_gitea_groups(gitlab_groups: list, gitea_groups: list) -> dict: # 0 = exists on both @@ -714,11 +835,11 @@ def cmp_gitlab_gitea_groups(gitlab_groups: list, gitea_groups: list) -> dict: missing_matches = 0 for gitlab_group in gitlab_groups: - name = gitlab_group['path'] + name = gitlab_group["path"] exists = False for gitea_group in gitea_groups: - if name == gitea_group['name']: + if name == gitea_group["name"]: exists = True break @@ -727,24 +848,25 @@ def cmp_gitlab_gitea_groups(gitlab_groups: list, gitea_groups: list) -> dict: 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: - name = gitea_group['name'] + name = gitea_group["name"] exists = False for gitlab_group in gitlab_groups: - if name == gitlab_group['path']: + 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: # 0 = exists on both @@ -851,11 +973,11 @@ def cmp_gitlab_gitea_users(gitlab_users: list, gitea_users: list) -> dict: missing_matches = 0 for gitlab_user in gitlab_users: - name = gitlab_user['username'] + name = gitlab_user["username"] exists = False for gitea_user in gitea_users: - if name == gitea_user['login']: + if name == gitea_user["login"]: exists = True break @@ -864,15 +986,15 @@ def cmp_gitlab_gitea_users(gitlab_users: list, gitea_users: list) -> dict: 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: - name = gitea_user['login'] + name = gitea_user["login"] exists = False for gitlab_user in gitlab_users: - if name == gitlab_user['username']: + if name == gitlab_user["username"]: exists = True break @@ -884,17 +1006,18 @@ def cmp_gitlab_gitea_users(gitlab_users: list, gitea_users: list) -> dict: def cmp_gitlab_gitea_projects(gitlab_projects: list, gitea_projects: list) -> dict: - + return {}, 0 + def create_missing_groups(gitlab_groups: list, gitea_groups: list): - + for gitlab_group in gitlab_groups: - name = gitlab_group['path'] + name = gitlab_group["path"] exists = False for gitea_group in gitea_groups: - if name == gitea_group['name']: + if name == gitea_group["name"]: exists = True break @@ -906,19 +1029,22 @@ def create_missing_groups(gitlab_groups: list, gitea_groups: list): except Exception as e: _exception(f'Failed to create Gitea group "{name}": {e}', e) + def create_missing_users(gitlab_users: list, gitea_users: list): pass + def create_missing_projects(gitlab_projects: list, gitea_projects: list): pass + def update_existing_groups(gitlab_groups: list, gitea_groups: list): for gitlab_group in gitlab_groups: - name = gitlab_group['path'] + name = gitlab_group["path"] exists = False for gitea_group in gitea_groups: - if name == gitea_group['name']: + if name == gitea_group["name"]: exists = True break @@ -926,194 +1052,225 @@ def update_existing_groups(gitlab_groups: list, gitea_groups: list): _info(f'Updating existing group "{name}" on Gitea...') try: - update_gitea_org(name, { - 'description': gitlab_group['description'], - 'website': gitlab_group['web_url'], - 'visibility': gitlab2gitea_visibility(gitlab_group['visibility']), - 'full_name': gitlab_group['full_name'] - }) + update_gitea_org( + name, + { + "description": gitlab_group["description"], + "website": gitlab_group["web_url"], + "visibility": gitlab2gitea_visibility( + gitlab_group["visibility"] + ), + "full_name": gitlab_group["full_name"], + }, + ) except Exception as e: _exception(f'Failed to update Gitea group "{name}": {e}', e) + def update_existing_users(gitlab_users: list, gitea_users: list): pass + def update_existing_projects(gitlab_projects: list, gitea_projects: list): pass + def migrate_groups(): 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}') + _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) + 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) + _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!') - + _warn(f"{missing_matches} groups are missing on Gitea!") + if missing_matches > 0 and not NO_CREATE_MISSING_GROUPS: - _info('Creating missing groups on Gitea...') + _info("Creating missing groups on Gitea...") if not DRY_RUN: try: create_missing_groups(gitlab_groups, gitea_groups) except Exception as e: - _exception(f'Failed to create missing groups: {e}', e) + _exception(f"Failed to create missing groups: {e}", e) else: - _warn('Dry-run mode enabled, skipping creation of missing groups on Gitea...') + _warn( + "Dry-run mode enabled, skipping creation of missing groups on Gitea..." + ) if not NO_UPDATE_EXISTING_GROUPS: - _info('Updating existing groups on Gitea...') + _info("Updating existing groups on Gitea...") try: if not DRY_RUN: update_existing_groups(gitlab_groups, gitea_groups) else: - _warn('Dry-run mode enabled, skipping update of existing groups on Gitea...') + _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) + _exception(f"Failed to update existing groups: {e}", e) + def migrate_users(): 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}') + _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']) + 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) + _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!') + _warn(f"{missing_matches} users are missing on Gitea!") if missing_matches > 0 and not NO_CREATE_MISSING_USERS: - _info('Creating missing users on Gitea...') + _info("Creating missing users on Gitea...") if not DRY_RUN: create_missing_users(gitlab_users, gitea_users) else: - _warn('Dry-run mode enabled, skipping creation of missing users on Gitea...') + _warn( + "Dry-run mode enabled, skipping creation of missing users on Gitea..." + ) if not NO_UPDATE_EXISTING_USERS: - _info('Updating existing users on Gitea...') + _info("Updating existing users on Gitea...") try: if not DRY_RUN: update_existing_users(gitlab_users, gitea_users) else: - _warn('Dry-run mode enabled, skipping update of existing users on Gitea...') + _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) + _exception(f"Failed to update existing users: {e}", e) + def migrate_projects(): gitlab_projects = get_gitlab_projects() gitea_projects = get_gitea_projects() - _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}') + _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) + project_result, missing_matches = cmp_gitlab_gitea_projects( + gitlab_projects, gitea_projects + ) except Exception as e: - _exception(f'Failed to compare GitLab and Gitea projects: {e}', 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!') + _warn(f"{missing_matches} projects are missing on Gitea!") if missing_matches > 0 and not NO_CREATE_MISSING_PROJECTS: - _info('Creating missing projects on Gitea...') + _info("Creating missing projects on Gitea...") if not DRY_RUN: create_missing_projects(gitlab_projects, gitea_projects) else: - _warn('Dry-run mode enabled, skipping creation of missing projects on Gitea...') - + _warn( + "Dry-run mode enabled, skipping creation of missing projects on Gitea..." + ) + if not NO_UPDATE_EXISTING_PROJECTS: - _info('Updating existing projects on Gitea...') + _info("Updating existing projects on Gitea...") try: if not DRY_RUN: update_existing_projects(gitlab_projects, gitea_projects) else: - _warn('Dry-run mode enabled, skipping update of existing projects on Gitea...') + _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) + _exception(f"Failed to update existing projects: {e}", e) + def run_migration(): - - _info('Migrating GitLab groups...') + + _info("Migrating GitLab groups...") migrate_groups() - _info('Group migration completed!') + _info("Group migration completed!") - _info('Migrating GitLab users...') + _info("Migrating GitLab users...") migrate_users() - _info('User migration completed!') + _info("User migration completed!") - _info('Migrating GitLab projects...') + _info("Migrating GitLab projects...") migrate_projects() - _info('Project migration completed!') + _info("Project migration completed!") + def main(): - _info('Gitlab2Gitea v1.0 - by Zion Networks') - _info('------------------------------------') - _info('') + _info("Gitlab2Gitea v1.0 - by Zion Networks") + _info("------------------------------------") + _info("") if sys.version_info < (3, 6): - _error('Python 3.6 or higher is required!') + _error("Python 3.6 or higher is required!") return else: - _debug(f'Python version: {sys.version_info.major}.{sys.version_info.minor}') + _debug(f"Python version: {sys.version_info.major}.{sys.version_info.minor}") if not GITLAB_TOKEN: - _error('GitLab access token is missing!') + _error("GitLab access token is missing!") return if not GITEA_TOKEN: - _error('Gitea access token is missing!') + _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}') + 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!') + _warn("Running in dry-run mode!") - _info('Testing endpoints...') + _info("Testing endpoints...") try: check_gitlab() check_gitea() except Exception as e: - _exception(f'An error occurred: {e}', e) + _exception(f"An error occurred: {e}", e) return - - _info('Starting migration...') + + _info("Starting migration...") try: run_migration() except Exception as e: - _exception(f'An error occurred: {e}', e) + _exception(f"An error occurred: {e}", e) -if __name__ == '__main__': + +if __name__ == "__main__": main()