154 lines
5.1 KiB
Python
154 lines
5.1 KiB
Python
import httpx
|
|
import os
|
|
from pathlib import Path
|
|
from fastapi import APIRouter, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
class SubscribeRequest(BaseModel):
|
|
"""Schema für Subscription Request"""
|
|
repo_url: str
|
|
target_name: str = "imported-repo"
|
|
token_provided: str | None = None
|
|
|
|
|
|
class UnsubscribeRequest(BaseModel):
|
|
"""Schema fuer Deabonination"""
|
|
project_name: str
|
|
|
|
|
|
@router.get("/projects") # Listet abonierte Projekte!
|
|
async def list_projects():
|
|
"""Listet lokale (abonierte) Projekte an."""
|
|
from .main import PROJECTS_DIR
|
|
|
|
projects = []
|
|
|
|
if PROJECTS_DIR.exists() and str(PROJECTS_DIR).strip():
|
|
# Sortiere nach dem Status der Changes zuerst
|
|
all_projects = []
|
|
for project_dir in sorted(PROJECTS_DIR.iterdir()):
|
|
if project_dir.is_dir() and not project_dir.name.startswith('.'):
|
|
has_git = (project_dir / '.git').exists()
|
|
|
|
# Pruefe ob dieses Projekt bearbeitet werden darf (chmod-check)
|
|
can_write = os.access(str(project_dir), os.W_OK) if project_dir.exists() else False
|
|
|
|
all_projects.append({
|
|
"name": project_dir.name,
|
|
"path": str(project_dir),
|
|
"subscribed": True,
|
|
"has_git": bool(has_git),
|
|
"can_write": can_write
|
|
})
|
|
|
|
# Sortiere: Projects mit Changes zuerst (kan signale anzeigen)
|
|
projects = sorted(all_projects, key=lambda x: ("." in x["name"]), reverse=True)
|
|
|
|
return {"projects": projects}
|
|
|
|
|
|
@router.post("/unsubscribe") # Deabonination Endpoint!
|
|
async def unsubscribe_repo(req: UnsubscribeRequest):
|
|
"""Loeschen eines abonnierten Projekts."""
|
|
from .main import PROJECTS_DIR, get_token
|
|
|
|
target_dir = PROJECTS_DIR / req.project_name
|
|
|
|
if not target_dir.exists():
|
|
return {"success": False, "error": f"Projekt '{req.project_name}' nicht gefunden"}
|
|
|
|
try:
|
|
import shutil
|
|
|
|
# ✅ SCHRITT 1: Alle Berechtigungen normalisieren damit wir es loeschen können!
|
|
set_permissions(str(target_dir))
|
|
|
|
# ✅ SCHRITT 2: Mit FORCE-LöSCHUNG entfernen (wegen root-Besitzern)
|
|
import stat
|
|
|
|
def force_remove_dir(dir_path):
|
|
"""Erzwingt das Loeschen von Verzeichnissen."""
|
|
for root, dirs, files in os.walk(str(dir_path), topdown=False):
|
|
for d in dirs:
|
|
dir_full = Path(root) / d
|
|
try:
|
|
dir_full.chmod(0o755 | stat.S_IWUSR | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
|
|
except Exception:
|
|
pass
|
|
for f in files:
|
|
file_full = Path(root) / f
|
|
try:
|
|
# Schreibrechte wiederherstellen
|
|
os.chmod(str(file_full), 0o666)
|
|
file_full.unlink()
|
|
except Exception:
|
|
pass
|
|
|
|
force_remove_dir(target_dir)
|
|
shutil.rmtree(str(target_dir))
|
|
|
|
return {"success": True, "message": f"{req.project_name} erfolgreich geloescht"}
|
|
|
|
except Exception as e:
|
|
# Fallback mit rm -rf (Unix-Force)
|
|
try:
|
|
import subprocess
|
|
subprocess.run(["rm", "-rf", str(target_dir)], check=True)
|
|
return {"success": True, "message": f"{req.project_name} geloescht (force-mode)"}
|
|
except Exception as e2:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
|
|
@router.get("/repos") # Listet Gitea-Repos!
|
|
async def list_repos():
|
|
"""Listet Repositories von Gitea auf."""
|
|
from .main import GITEE_URL, ACCESS_TOKEN
|
|
|
|
token = ACCESS_TOKEN or ""
|
|
|
|
if not token:
|
|
return {"error": "Kein Token vorhanden!", "hint": "Setze ACCESS_TOKEN Umgebungsvariable"}
|
|
|
|
headers = {"Authorization": f"token {token}"}
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.get(f"{GITEE_URL.rstrip('/')}/api/v1/user/repos", headers=headers)
|
|
return resp.json()
|
|
|
|
|
|
@router.post("/subscribe") # Klont Repo!
|
|
async def subscribe_repo(req: SubscribeRequest):
|
|
"""Abonniert und klont ein Repository."""
|
|
from .main import ACCESS_TOKEN
|
|
|
|
effective_token = req.token_provided or ACCESS_TOKEN or ""
|
|
|
|
if not effective_token:
|
|
raise HTTPException(status_code=401, detail="Kein TOKEN vorhanden!")
|
|
|
|
target_dir = Path("/projects") / req.target_name
|
|
|
|
result = clone_repo(req.repo_url, target_dir, token=effective_token)
|
|
|
|
if not result["success"]:
|
|
return {"error": result.get("error"), "success": False}
|
|
|
|
# ✅ BERECHTIGUNGEN SETZEN NACH DEM KLOONEN!
|
|
set_permissions(str(target_dir))
|
|
|
|
generate_changelog(target_dir)
|
|
|
|
return {
|
|
"message": f"✅ {req.target_name} geklont!",
|
|
"path": str(target_dir),
|
|
"success": True,
|
|
"can_write": os.access(str(target_dir), os.W_OK) if target_dir.exists() else False
|
|
}
|
|
|
|
|
|
# Importiere die Funktionen am Ende damit sie verfügbar sind!
|
|
from .git_manager import clone_repo, set_permissions
|
|
from .changelog_gen import generate_changelog |