This commit is contained in:
JiriUhlir
2026-06-17 12:47:53 +02:00
parent 223f59d750
commit bd256d565c
8 changed files with 550 additions and 34 deletions
+67 -20
View File
@@ -8,6 +8,9 @@ from urllib.request import urlopen
from app.config import get_gitea_admin_token, get_gitea_server_url
from app.db.users import update_gitea_sync_state
from app.logging_config import get_logger
logger = get_logger(__name__)
GITEA_ACCOUNT_ROLES = {"admin", "developer"}
@@ -22,41 +25,84 @@ def desired_gitea_username(user: dict) -> str:
def apply_gitea_credentials(user: dict, new_username: str = "", new_password: str = "") -> dict:
"""Admin nastaví Gitea uživatelské jméno a/nebo heslo přes admin API.
Vrací {"status": "synced"/"error", "applied": [...], "gitea_username", "gitea_user_id"}.
Heslo a přejmenování se aplikují NEZÁVISLE když jedno selže, druhé se i tak provede a do
výsledku se vrátí přesný důvod selhání (chybí endpoint na staré Gitea, jméno obsazené apod.).
Vrací: {"status": synced|partial|error, "applied": [...], "errors": [...],
"gitea_username", "gitea_user_id", "renamed": bool}.
Heslo se nikam neukládá jen se předá do Gitea."""
new_username = (new_username or "").strip()
new_password = new_password or ""
if not new_username and not new_password:
return {"status": "error", "error": "Nebyla zadána žádná změna."}
return {"status": "error", "error": "Nebyla zadána žádná změna.", "errors": [], "applied": [], "renamed": False}
if not _requires_gitea_account(user):
return {
"status": "error",
"error": "Uživatel nemá mít Gitea účet (potřebuje roli admin/developer a aktivní účet).",
"errors": [],
"applied": [],
"renamed": False,
}
user_id = int(user["id"])
try:
current_username = desired_gitea_username(user)
gitea_user = _ensure_gitea_user(user, current_username)
current_login = _gitea_login(gitea_user) or current_username
applied: list[str] = []
if new_username and new_username != current_login:
_rename_gitea_user(current_login, new_username)
current_login = new_username
applied.append("username")
if new_password:
_set_gitea_password(current_login, user, new_password, gitea_user=gitea_user)
applied.append("password")
return {
"status": "synced",
"applied": applied,
"gitea_username": current_login,
"gitea_user_id": int(gitea_user["id"]),
}
except Exception as exc:
return {"status": "error", "error": str(exc)[:500]}
message = f"Gitea účet se nepodařilo načíst: {str(exc)[:400]}"
logger.error("apply_gitea_credentials: %s (user id=%s)", message, user_id)
return {
"status": "error",
"error": message,
"errors": [],
"applied": [],
"renamed": False,
}
current_login = _gitea_login(gitea_user) or current_username
final_username = current_login
applied: list[str] = []
errors: list[str] = []
# 1) Heslo nezávisle na přejmenování (vždy se zkusí, i kdyby rename neexistoval).
if new_password:
try:
_set_gitea_password(current_login, user, new_password, gitea_user=gitea_user)
applied.append("heslo")
logger.info("Gitea heslo nastaveno pro %s (user id=%s)", current_login, user_id)
except Exception as exc:
detail = str(exc)[:300]
errors.append(f"heslo: {detail}")
logger.error("Gitea nastavení hesla selhalo pro %s (user id=%s): %s", current_login, user_id, detail)
# 2) Přejmenování (Gitea >= 1.21). Na starší Gitea vrátí HTTP 404 → zobrazí se srozumitelně.
if new_username and new_username != current_login:
try:
_rename_gitea_user(current_login, new_username)
final_username = new_username
applied.append("jméno")
logger.info("Gitea účet přejmenován %s -> %s (user id=%s)", current_login, new_username, user_id)
except Exception as exc:
detail = str(exc)[:300]
errors.append(f"jméno: {detail}")
logger.error("Gitea přejmenování %s -> %s selhalo (user id=%s): %s", current_login, new_username, user_id, detail)
if applied and not errors:
status = "synced"
elif applied:
status = "partial"
else:
status = "error"
return {
"status": status,
"applied": applied,
"errors": errors,
"gitea_username": final_username,
"gitea_user_id": int(gitea_user["id"]),
"renamed": "jméno" in applied,
}
def _rename_gitea_user(current_username: str, new_username: str) -> dict:
@@ -121,6 +167,7 @@ def sync_gitea_user(user: dict) -> dict:
return {"status": "not_required", "action": "none"}
except Exception as exc:
error = str(exc)[:500]
logger.error("Gitea sync selhal pro uživatele id=%s (%s): %s", user_id, username, error)
update_gitea_sync_state(user_id, "error", error=error)
return {"status": "error", "error": error}