JiriUhlir
100 lines
4.0 KiB
Python
100 lines
4.0 KiB
Python
"""Sdílená git logika pro zápis do pracovního klonu appfactory-tools.
|
|
|
|
Portál zapisuje .sh soubory přímo do podsložek tohoto repa (maintenance/, alerts/).
|
|
Aby to nedělalo nepořádek v gitea, každý zápis/smazání rovnou commitne + pushne.
|
|
"""
|
|
|
|
import html
|
|
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from app.config import (
|
|
DEFAULT_GITEA_ORG,
|
|
get_gitea_admin_token,
|
|
get_gitea_server_url,
|
|
read_env_value,
|
|
)
|
|
|
|
TOOLS_REPO_DIR = Path("/opt/appfactory/workspace/appfactory-tools")
|
|
TOOLS_REPO_NAME = "appfactory-tools"
|
|
GIT_AUTHOR_NAME = "AppFactory Portal"
|
|
GIT_AUTHOR_EMAIL = "portal@appfactory.local"
|
|
|
|
|
|
def _git(args: list[str]):
|
|
env = {**os.environ, "GIT_TERMINAL_PROMPT": "0"}
|
|
try:
|
|
return subprocess.run(
|
|
["git", "-C", str(TOOLS_REPO_DIR), *args],
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
timeout=30,
|
|
)
|
|
except (subprocess.TimeoutExpired, OSError):
|
|
raise HTTPException(status_code=500, detail="Git příkaz selhal nebo vypršel limit")
|
|
|
|
|
|
def _git_actor(user: dict) -> str:
|
|
return (user.get("email") or user.get("display_name") or user.get("username") or "neznámý").strip()
|
|
|
|
|
|
def _gitea_push_remote() -> tuple[str, str | None]:
|
|
"""Vrátí (remote, token). Remote je autentizovaná gitea URL z tokenů v proměnných,
|
|
při absenci tokenu fallback na origin. Token vracíme zvlášť, aby šel zamaskovat v chybách."""
|
|
token = get_gitea_admin_token()
|
|
server = get_gitea_server_url()
|
|
if not token or "://" not in server:
|
|
return "origin", None
|
|
scheme, rest = server.split("://", 1)
|
|
org = read_env_value("GITEA_ORG", DEFAULT_GITEA_ORG)
|
|
return f"{scheme}://{token}@{rest}/{org}/{TOOLS_REPO_NAME}.git", token
|
|
|
|
|
|
def _git_error_detail(result, token: str | None = None) -> str:
|
|
"""Zkombinuje git stderr/stdout do čitelného důvodu chyby (s maskováním tokenu)."""
|
|
detail = (result.stderr or "").strip() or (result.stdout or "").strip()
|
|
if token and detail:
|
|
detail = detail.replace(token, "***")
|
|
return html.escape(detail) if detail else "git nevrátil žádný výstup"
|
|
|
|
|
|
def commit_and_push(rel_path: str, message: str, user: dict) -> None:
|
|
"""Zacommituje a pushne změnu jednoho souboru (cesta relativní k repu) do gitea."""
|
|
# Repozitář v nedořešeném merge konfliktu blokuje jakýkoli commit (i jen jednoho souboru).
|
|
# Soubor je už uložený na disku; commit projde po ručním vyřešení konfliktu na serveru.
|
|
if (_git(["ls-files", "--unmerged"]).stdout or "").strip():
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail=(
|
|
f"Repozitář {TOOLS_REPO_DIR} je v nedořešeném merge konfliktu (unmerged soubory), "
|
|
"proto nelze commitnout. Soubor je uložený na disku. Vyřešte konflikt na serveru "
|
|
"(např. `git merge --abort` nebo ručně `git add` + commit) a akci zopakujte."
|
|
),
|
|
)
|
|
|
|
add = _git(["add", "--", rel_path])
|
|
if add.returncode != 0:
|
|
raise HTTPException(status_code=500, detail=f"Git add selhal (v {TOOLS_REPO_DIR}): {_git_error_detail(add)}")
|
|
|
|
# Žádná změna oproti HEAD -> přeskočíme, ať nevznikají prázdné commity.
|
|
if _git(["diff", "--cached", "--quiet", "--", rel_path]).returncode == 0:
|
|
return
|
|
|
|
commit = _git([
|
|
"-c", f"user.name={GIT_AUTHOR_NAME}",
|
|
"-c", f"user.email={GIT_AUTHOR_EMAIL}",
|
|
"commit", "-m", f"{message} (portál: {_git_actor(user)})",
|
|
])
|
|
if commit.returncode != 0:
|
|
raise HTTPException(status_code=500, detail=f"Git commit selhal (v {TOOLS_REPO_DIR}): {_git_error_detail(commit)}")
|
|
|
|
branch = (_git(["rev-parse", "--abbrev-ref", "HEAD"]).stdout or "").strip() or "main"
|
|
remote, token = _gitea_push_remote()
|
|
push = _git(["push", remote, f"HEAD:{branch}"])
|
|
if push.returncode != 0:
|
|
raise HTTPException(status_code=500, detail=f"Git push selhal: {_git_error_detail(push, token)}")
|