server-up/server-up/core/git.py

"""Git operations — clone, pull, scan for stacks and modules."""
import json, os, re, subprocess, shutil
from pathlib import Path
from core.docker import COMPOSE_NAMES
from core import boilerplates as bp


def guess_logo_url(name: str) -> str:
    """Leid een dashboard-icons logo-URL af uit een app-naam.
    Bestaat het icoon niet, dan faalt de <img> en valt de UI terug op de emoji."""
    slug = re.sub(r"[^a-z0-9]+", "-", (name or "").lower()).strip("-")
    if not slug:
        return ""
    return f"https://cdn.jsdelivr.net/gh/walkxcode/dashboard-icons/png/{slug}.png"

CACHE = Path(os.environ.get("SU_GIT_CACHE", "/data/git"))

# Fix git "dubious ownership" — container draait als root, cache kan door andere uid zijn
_git_safe_set = False


def _ensure_git_safe():
    global _git_safe_set
    if not _git_safe_set:
        os.environ["GIT_CONFIG_COUNT"] = "1"
        os.environ["GIT_CONFIG_KEY_0"] = "safe.directory"
        os.environ["GIT_CONFIG_VALUE_0"] = "*"
        _git_safe_set = True


# Run immediately at import
_ensure_git_safe()


def cache_dir(repo_id: str) -> Path:
    return CACHE / repo_id


def clone_or_pull(repo: dict, log_fn=None) -> tuple[bool, str]:
    _ensure_git_safe()
    url = repo.get("url", "").strip()
    if not url:
        return False, "Geen URL"
    # Vangnet: ook repo's die al in een oude config.json staan mogen geen
    # `ext::`- of `file://`-transport gebruiken (ext:: voert een shell uit).
    from core import valid_repo_url
    if not valid_repo_url(url):
        return False, f"Ongeldige of niet-toegestane repo-URL: {url[:80]}"
    branch = repo.get("branch", "main")
    token = repo.get("token", "").strip()
    rid = repo.get("id", "repo")
    dest = cache_dir(rid)

    clone_url = url
    if token and url.startswith("https://"):
        host = url.split("://", 1)[1]
        clone_url = f"https://x-access-token:{token}@{host}"

    env = os.environ.copy()
    env["GIT_TERMINAL_PROMPT"] = "0"

    try:
        if dest.exists() and (dest / ".git").exists():
            if log_fn:
                log_fn(f"Pull {rid}…")
            r = subprocess.run(["git", "pull", "--rebase", "--autostash"],
                               cwd=str(dest), capture_output=True, text=True,
                               env=env, timeout=60)
        else:
            if log_fn:
                log_fn(f"Clone {url}…")
            dest.parent.mkdir(parents=True, exist_ok=True)
            if dest.exists():
                shutil.rmtree(dest)
            r = subprocess.run(["git", "clone", "--branch", branch, "--depth", "1",
                                clone_url, str(dest)],
                               capture_output=True, text=True, env=env, timeout=120)

        output = (r.stdout + "\n" + r.stderr).strip()
        # Verwijder token uit output voordat het gelogd of teruggegeven wordt
        safe_output = output.replace(clone_url, url) if clone_url != url else output
        for line in safe_output.splitlines():
            if line.strip() and log_fn:
                log_fn(line)

        if r.returncode != 0:
            return False, safe_output[:200]

        # Optioneel vastzetten op een commit. Zo bepaalt de beheerder wanneer
        # er nieuwe templates/modulecode binnenkomt, in plaats van de upstream.
        pin = (repo.get("commit") or "").strip()
        if pin:
            if not re.fullmatch(r"[0-9a-fA-F]{7,40}", pin):
                return False, f"Ongeldige commit-pin: {pin[:40]}"
            fetch = subprocess.run(["git", "fetch", "--depth", "1", "origin", pin],
                                   cwd=str(dest), capture_output=True, text=True,
                                   env=env, timeout=60)
            if fetch.returncode != 0 and log_fn:
                log_fn(f"fetch {pin}: {fetch.stderr.strip()[:120]}")
            co = subprocess.run(["git", "checkout", "--force", pin],
                                cwd=str(dest), capture_output=True, text=True,
                                env=env, timeout=60)
            if co.returncode != 0:
                return False, f"Commit {pin} niet gevonden"
            if log_fn:
                log_fn(f"Vastgezet op commit {pin}")
        return True, "OK"
    except subprocess.TimeoutExpired:
        return False, "Timeout"
    except Exception as e:
        return False, str(e)


def scan_stacks(repo_id: str, lib: Path, subdir: str = "") -> list[dict]:
    base = cache_dir(repo_id)
    if subdir:
        base = base / subdir
    if not base.exists():
        return []

    stacks = _scan_compose_dirs(base, lib)

    # Fallback: als geen stacks gevonden en geen subdir opgegeven,
    # zoek automatisch in bekende submappen
    if not stacks and not subdir:
        for fallback in ("apps", "stacks", "docker"):
            fb = cache_dir(repo_id) / fallback
            if fb.exists():
                stacks = _scan_compose_dirs(fb, lib)
                if stacks:
                    break

    return stacks


def scan_dir(base: Path, lib: Path) -> list[dict]:
    """Scan een lokale map voor stacks/boilerplates zonder git (voor ingebouwde apps)."""
    if not base.exists():
        return []
    return _scan_compose_dirs(base, lib)


def _scan_compose_dirs(base: Path, lib: Path) -> list[dict]:
    """Scan één directory voor compose stacks. Detecteert zowel het klassieke
    Server Up formaat (compose.yml + stack.json) als het ChristianLempa
    Boilerplates formaat (template.json + files/)."""
    stacks = []
    for d in sorted(base.iterdir()):
        if not d.is_dir() or d.name.startswith((".", "_")):
            continue

        is_legacy = any((d / n).exists() for n in COMPOSE_NAMES)
        is_bp = bp.is_boilerplate(d)
        if not (is_legacy or is_bp):
            continue

        if is_bp:
            meta = bp.metadata(d)
            # boilerplate met draft=true wordt overgeslagen tenzij expliciet
            if meta.get("draft"):
                continue
        else:
            meta = _meta(d)
            meta.setdefault("format", "compose")
            # Auto-logo voor legacy-apps: gebruik expliciete logo_url, anders raden
            # op basis van de naam (dashboard-icons). Emoji blijft fallback in de UI.
            if not meta.get("logo_url"):
                ic = meta.get("icon", "")
                if isinstance(ic, str) and (ic.startswith("http://") or ic.startswith("https://")):
                    meta["logo_url"] = ic
                else:
                    meta["logo_url"] = guess_logo_url(meta.get("name") or d.name)

        installed = []
        if lib.exists():
            for ld in lib.iterdir():
                if ld.is_dir() and (ld.name == d.name or ld.name.startswith(d.name + "-")):
                    installed.append(ld.name)
        entry = {
            "dir": d.name,
            "name": meta.pop("name", d.name),
            "path": str(d),
            "instances": sorted(installed),
        }
        entry.update(meta)
        stacks.append(entry)
    return stacks


def scan_modules(repo_id: str, subdir: str = "") -> list[dict]:
    base = cache_dir(repo_id)
    if subdir:
        base = base / subdir
    if not base.exists():
        return []

    # Fallback: als base geen subdirectories heeft, probeer bekende submappen
    if not any(d.is_dir() for d in base.iterdir()):
        if not subdir:
            for fb in ("modules",):
                fb_path = cache_dir(repo_id) / fb
                if fb_path.exists():
                    base = fb_path
                    break

    mods = []
    for d in sorted(base.iterdir()):
        if not d.is_dir() or d.name.startswith((".", "_")):
            continue
        # Vereist module.json
        mf = d / "module.json"
        if not mf.exists():
            continue
        init = d / "__init__.py"
        if not init.exists():
            continue
        # Valideer dat __init__.py daadwerkelijk een Module subclass bevat
        try:
            src = init.read_text()
            if "Module" not in src or "class " not in src:
                continue
        except Exception:
            continue
        meta = {}
        try:
            meta = json.loads(mf.read_text())
        except Exception:
            pass
        mid = meta.get("id", d.name)
        # Skip core features
        from core.modules import CORE
        if mid in CORE:
            continue
        mods.append({
            "id": mid,
            "name": meta.get("name", d.name),
            "icon": meta.get("icon", "🧩"),
            "desc": meta.get("description", ""),
            "version": meta.get("version", "1.0.0"),
            "tags": meta.get("tags", []),
            "author": meta.get("author", ""),
            "repo_id": repo_id,
            "_src": str(d),
        })
    return mods


def _meta(d: Path) -> dict:
    for n in ("app.json", "stack.json", "meta.json"):
        f = d / n
        if f.exists():
            try:
                return json.loads(f.read_text())
            except Exception:
                pass
    return {"description": "", "icon": "", "tags": [], "ports": []}