Implemented user migration

gitlab2gitea/gitlab2gitea.py

@@ -75,6 +75,10 @@ LOG_FILE = "gitlab2gitea.log"
 APPEND_LOG = False
 QUIET = False
 
+# Internal variables - Do not change
+
+GITEA_RESERVED_USERNAMES = ["ghost", "notifications"]
+
 # Imports
 
 import os
@@ -396,6 +400,10 @@ def _exception(exception, custom_message=None):
 # PROGRAM
 
 
+def is_gitea_reserved_username(username: str) -> bool:
+    return username in GITEA_RESERVED_USERNAMES
+
+
 def gitlab2gitea_visibility(visibility: str) -> str:
     if visibility == "private":
         return "private"
@@ -589,7 +597,51 @@ def get_gitlab_projects() -> list:
 
 # Endpoint: POST /api/{GITEA_API_VERSION}/admin/users
 def migrate_gitlab_user_to_gitea(user: dict):
-    pass
+
+    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"
+        )
+        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')
+
+    return user
 
 
 # Endpoint: GET /api/{GITEA_API_VERSION}/orgs
@@ -760,6 +812,122 @@ def update_gitea_org(data: dict) -> dict:
     return group
 
 
+# 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
+
+
+# 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, 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}")
+
+    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 cmp_gitea_userdata(userdata_a: dict, userdata_b: dict) -> bool:
+
+    result = {}
+
+    for key in userdata_a:
+        if key in userdata_b:
+            if userdata_a[key] != userdata_b[key]:
+                result[key] = True
+            else:
+                result[key] = False
+        else:
+            result[key] = True
+
+    has_changes = False
+    for key in result:
+        if result[key]:
+            has_changes = True
+            break
+
+    return has_changes, result
+
+
+def convert_gitlab_user_to_gitea(user: dict, extra_data: dict = None) -> dict:
+
+    gitea_user = {
+        "active": user["state"] == "active",
+        "login_name": user["username"],
+        "avatar_url": user["avatar_url"],
+        "created": user["created_at"],
+        "description": user["bio"],
+        "email": user["email"],
+        "full_name": user["name"],
+        "is_admin": user["is_admin"],
+        "prohibit_login": user["state"] == "blocked" or user["locked"],
+        "website": user["website_url"],
+    }
+
+    if extra_data:
+        gitea_user.update(extra_data)
+
+    return gitea_user
+
+
 def cmp_gitlab_gitea_groups(gitlab_groups: list, gitea_groups: list) -> dict:
 
     # 0 = exists on both
@@ -1031,7 +1199,37 @@ def create_missing_groups(gitlab_groups: list, gitea_groups: list):
 
 
 def create_missing_users(gitlab_users: list, gitea_users: list):
-    pass
+
+    for gitlab_user in gitlab_users:
+        name = gitlab_user["username"]
+        exists = False
+
+        for gitea_user in gitea_users["data"]:
+            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!'
+            )
+            continue
+
+        if gitea_user["username"] == None or gitea_user["username"] == "":
+            _warn(f'User "{name}" does not have a username and will not be created!')
+            continue
+
+        if is_gitea_reserved_username(name):
+            _warn(f'User "{name}" is a reserved username on Gitea!')
+            continue
+
+        if not exists:
+            _info(f'Creating missing user "{name}" on Gitea...')
+
+            try:
+                migrate_gitlab_user_to_gitea(gitlab_user)
+            except Exception as e:
+                _exception(f'Failed to create Gitea user "{name}": {e}', e)
 
 
 def create_missing_projects(gitlab_projects: list, gitea_projects: list):
@@ -1053,8 +1251,8 @@ def update_existing_groups(gitlab_groups: list, gitea_groups: list):
 
             try:
                 update_gitea_org(
-                    name,
                     {
+                        "path": gitlab_group["path"],
                         "description": gitlab_group["description"],
                         "website": gitlab_group["web_url"],
                         "visibility": gitlab2gitea_visibility(
@@ -1068,7 +1266,34 @@ def update_existing_groups(gitlab_groups: list, gitea_groups: list):
 
 
 def update_existing_users(gitlab_users: list, gitea_users: list):
-    pass
+
+    for gitlab_user in gitlab_users:
+        name = gitlab_user["username"]
+        exists = False
+
+        for gitea_user in gitea_users["data"]:
+            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 exists:
+            _info(f'Updating existing user "{name}" on Gitea...')
+            try:
+                update_gitea_user(gitlab_user)
+            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):