Files
gitea-sync-ai/app/git_manager.py
T
2026-08-03 16:22:07 +02:00

127 lines
4.5 KiB
Python

import subprocess
import os
import shutil
import stat
from pathlib import Path
def set_permissions(path: str):
"""Setzt korrekte POSIX-Berechtigungen fuer ein Verzeichnis."""
try:
# Ordner = 755 (rwxr-xr-x)
os.chmod(str(path), 0o755)
# Rekursiv alle Teildateien bearbeiten
for root, dirs, files in os.walk(str(path)):
for d in dirs:
full_path = os.path.join(root, d)
try:
os.chmod(full_path, 0o755)
except Exception:
pass
for f in files:
full_path = os.path.join(root, f)
try:
# Setze WRITE permission damit wir es loeschen können!
os.chmod(full_path, stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH)
except Exception:
pass
# Nachher auch den Hauptordner beschreiben machen
os.chmod(str(path), 0o755 | stat.S_ISVTX) # Sticky-Bit hilft beim Loeschen
except Exception as e:
print(f"Warnung bei Berechtigungen: {e}")
def clone_repo(repo_url: str, target_dir: Path, token: str = None):
"""Klont ein Repository in den Zielordner."""
if not target_dir.exists():
target_dir.mkdir(parents=True)
# URL mit Token (falls noetig)
if token and "@" not in repo_url:
repo_url = repo_url.replace("https://", f"https://{token}@")
result = subprocess.run(
["git", "clone", repo_url, str(target_dir)],
capture_output=True,
text=True
)
# ✅ BERECHTIGUNGEN NACH DEM KLOONEN SETZEN!
if result.returncode == 0:
set_permissions(str(target_dir))
return {"success": result.returncode == 0, "output": result.stdout, "error": result.stderr}
def unsubscribe_repo(project_name: str, projects_dir: Path):
"""Loeschen eines abonnierten Projekts inklusive Berechtigungen."""
target_dir = projects_dir / project_name
if not target_dir.exists():
return {"success": False, "error": "Projekt nicht gefunden"}
try:
# ✅ SCHRITT 1: Alle Berechtigungen normalisieren damit wir es loeschen können!
set_permissions(str(target_dir))
# ✅ SCHRITT 2: Mit REMOVE-TREATMENTS einen Ordner mit root-Besitzern loeschen
# Dies schreibt neu und zwingt das Loeschen:
for root, dirs, files in os.walk(str(target_dir), topdown=False):
for d in dirs:
dir_path = Path(root) / d
try:
dir_path.chmod(0o755) # Schreibrechte wiederherstellen
except Exception:
pass
for f in files:
file_path = Path(root) / f
try:
file_path.unlink() # Datei loeschen
except Exception:
pass
shutil.rmtree(str(target_dir))
return {"success": True, "message": f"{project_name} erfolgreich geloescht"}
except PermissionError as e:
# Fallback: Versuche mit FERZWUNGSLÖSCHUNG!
try:
import subprocess as sp
sp.run(["rm", "-rf", str(target_dir)], check=True) # UNIX-force-loeschen
return {"success": True, "message": f"{project_name} geloescht (force-mode)"}
except Exception as e2:
return {"success": False, "error": f"Permission denied even after fix: {e}"}
except Exception as e:
# Letzter Versuch mit rm -rf
try:
import subprocess as sp
sp.run(["rm", "-rf", str(target_dir)], check=True)
return {"success": True, "message": f"{project_name} geloescht (force-mode)"}
except Exception as e2:
return {"success": False, "error": str(e)}
def push_changes(repo_path: Path, message: str = "Update 🧠"):
"""Pusht lokale Änderungen ins Repository."""
result = subprocess.run(
["git", "add", "."], cwd=repo_path, capture_output=True, text=True
)
if result.returncode != 0:
return {"success": False, "error": f"git add fehlgeschlagen: {result.stderr}"}
result = subprocess.run(
["git", "commit", "-m", message], cwd=repo_path, capture_output=True, text=True
)
if result.returncode != 0 and "nothing to commit" not in result.stderr.lower():
return {"success": False, "error": f"Commit fehlgeschlagen: {result.stderr}"}
result = subprocess.run(
["git", "push"], cwd=repo_path, capture_output=True, text=True
)
return {"success": result.returncode == 0, "output": result.stdout}