gitea sync update

app/gitea_provisioning.py

@@ -2,7 +2,7 @@ import json
 import secrets
 import string
 from urllib.error import HTTPError, URLError
-from urllib.parse import quote
+from urllib.parse import quote, urlencode
 from urllib.request import Request as UrlRequest
 from urllib.request import urlopen
 
@@ -21,7 +21,8 @@ def sync_gitea_user(user: dict) -> dict:
     try:
         if desired_active:
             gitea_user = _ensure_gitea_user(user, username)
-            _update_gitea_user(username, user, active=True)
+            gitea_username = _gitea_login(gitea_user) or username
+            _update_gitea_user(gitea_username, user, active=True, gitea_user=gitea_user)
             update_gitea_sync_state(
                 user_id,
                 "synced",
@@ -34,9 +35,10 @@ def sync_gitea_user(user: dict) -> dict:
             update_gitea_sync_state(user_id, "not_required", synced=True)
             return {"status": "not_required", "action": "none"}
 
-        gitea_user = _get_gitea_user(username, missing_ok=True)
+        gitea_user = _find_linked_gitea_user(user, username)
         if gitea_user:
-            _update_gitea_user(username, user, active=False)
+            gitea_username = _gitea_login(gitea_user) or username
+            _update_gitea_user(gitea_username, user, active=False, gitea_user=gitea_user)
             update_gitea_sync_state(
                 user_id,
                 "not_required",
@@ -67,6 +69,10 @@ def _ensure_gitea_user(user: dict, username: str) -> dict:
     if existing:
         return existing
 
+    existing = _find_gitea_user_by_email(_email(user))
+    if existing:
+        return existing
+
     return _create_gitea_user(user, username)
 
 
@@ -85,14 +91,14 @@ def _create_gitea_user(user: dict, username: str) -> dict:
     return _request_json("POST", "/api/v1/admin/users", payload, expected=(201,))
 
 
-def _update_gitea_user(username: str, user: dict, active: bool) -> dict:
+def _update_gitea_user(username: str, user: dict, active: bool, gitea_user: dict | None = None) -> dict:
     payload = {
         "email": _email(user),
         "full_name": _display_name(user),
-        "login_name": username,
+        "login_name": _gitea_login_name(gitea_user) or username,
-        "source_id": 0,
+        "source_id": _gitea_source_id(gitea_user),
         "active": active,
-        "admin": False,
+        "admin": bool((gitea_user or {}).get("is_admin", False)),
         "visibility": "private",
     }
     return _request_json("PATCH", f"/api/v1/admin/users/{quote(username)}", payload, expected=(200,))
@@ -107,6 +113,57 @@ def _get_gitea_user(username: str, missing_ok: bool = False) -> dict | None:
         raise
 
 
+def _find_linked_gitea_user(user: dict, username: str) -> dict | None:
+    existing = _get_gitea_user(username, missing_ok=True)
+    if existing:
+        return existing
+
+    gitea_user_id = user.get("gitea_user_id")
+    if not gitea_user_id:
+        return None
+
+    existing = _find_gitea_user_by_email(_email(user))
+    if existing and int(existing.get("id") or 0) == int(gitea_user_id):
+        return existing
+
+    return None
+
+
+def _find_gitea_user_by_email(email: str) -> dict | None:
+    email = (email or "").strip().lower()
+    if not email:
+        return None
+
+    response = _request_json("GET", f"/api/v1/admin/users?{urlencode({'q': email})}", expected=(200,))
+    users = response if isinstance(response, list) else response.get("data", [])
+    for gitea_user in users:
+        if (gitea_user.get("email") or "").strip().lower() == email:
+            return gitea_user
+
+    return None
+
+
+def _gitea_login(gitea_user: dict | None) -> str:
+    if not gitea_user:
+        return ""
+    return (gitea_user.get("login") or gitea_user.get("username") or "").strip()
+
+
+def _gitea_login_name(gitea_user: dict | None) -> str:
+    if not gitea_user:
+        return ""
+    return (gitea_user.get("login_name") or _gitea_login(gitea_user)).strip()
+
+
+def _gitea_source_id(gitea_user: dict | None) -> int:
+    if not gitea_user:
+        return 0
+    try:
+        return int(gitea_user.get("source_id") or 0)
+    except (TypeError, ValueError):
+        return 0
+
+
 def _request_json(method: str, path: str, payload: dict | None = None, expected: tuple[int, ...] = (200,)) -> dict:
     base_url = get_gitea_server_url().rstrip("/")
     token = get_gitea_admin_token()