homelab-configs/apps/homepage/admin/server.py

#!/usr/bin/env python3
"""Homepage Admin UI — sites toevoegen via browser."""
import os
import re
import subprocess
import threading
from pathlib import Path

from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse, HTMLResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel

APP_ROOT = Path(os.environ.get("HOMEPAGE_DIR", Path(__file__).resolve().parent.parent))
WEBSITES_ADD = APP_ROOT / "websites-add.txt"
HOMELAB_APPS = APP_ROOT.parent
GENERATE = APP_ROOT / "generate-config.py"
DEPLOY = APP_ROOT / "deploy-to-pve.sh"

GROUPS = [
    "Infrastructure",
    "Media & TV",
    "Smart Home",
    "Productivity",
    "Tools & Utils",
    "AI Assistants",
    "Dev & Docs",
    "Web Design",
]

TAB_FOR_GROUP = {
    "Infrastructure": "Ops",
    "Media & TV": "Media",
    "Smart Home": "Home",
    "Productivity": "Work",
    "Tools & Utils": "Ops",
    "AI Assistants": "AI",
    "Dev & Docs": "AI",
    "Web Design": "AI",
}

_apply_lock = threading.Lock()
_apply_status = {"running": False, "ok": None, "log": ""}


class SiteIn(BaseModel):
    group: str
    name: str
    url: str


class SiteOut(BaseModel):
    id: int
    group: str
    name: str
    url: str
    tab: str


def read_sites() -> list[dict]:
    if not WEBSITES_ADD.exists():
        WEBSITES_ADD.write_text(
            "# groep|naam|url\n# Of gebruik de UI op poort 3010\n\n",
            encoding="utf-8",
        )
    rows = []
    for i, raw in enumerate(WEBSITES_ADD.read_text(encoding="utf-8").splitlines()):
        line = raw.strip()
        if not line or line.startswith("#"):
            continue
        parts = [p.strip() for p in line.split("|", 2)]
        if len(parts) != 3:
            continue
        g, n, u = parts
        rows.append(
            {
                "id": len(rows),
                "group": g,
                "name": n,
                "url": u,
                "tab": TAB_FOR_GROUP.get(g, "?"),
            }
        )
    return rows


def write_sites(rows: list[tuple[str, str, str]]) -> None:
    lines = [
        "# Extra websites — beheerd via Homepage Admin UI",
        "# groep|naam|url",
        "",
    ]
    for g, n, u in rows:
        n = n.replace("|", "-")
        lines.append(f"{g}|{n}|{u}")
    lines.append("")
    WEBSITES_ADD.write_text("\n".join(lines), encoding="utf-8")


def run_apply():
    global _apply_status
    with _apply_lock:
        _apply_status = {"running": True, "ok": None, "log": "Start…\n"}
        try:
            gen = subprocess.run(
                ["python3", str(GENERATE)],
                cwd=str(APP_ROOT),
                capture_output=True,
                text=True,
                timeout=120,
            )
            _apply_status["log"] += gen.stdout or ""
            if gen.stderr:
                _apply_status["log"] += gen.stderr
            if gen.returncode != 0:
                _apply_status["ok"] = False
                _apply_status["log"] += "\nFout bij generate-config.py"
                return

            _apply_status["log"] += "\nDeploy naar Proxmox…\n"
            dep = subprocess.run(
                ["/bin/sh", str(DEPLOY)],
                cwd=str(APP_ROOT),
                capture_output=True,
                text=True,
                timeout=300,
                env={**os.environ},
            )
            _apply_status["log"] += dep.stdout or ""
            if dep.stderr:
                _apply_status["log"] += dep.stderr
            _apply_status["ok"] = dep.returncode == 0
            if _apply_status["ok"]:
                _apply_status["log"] += "\n✓ Klaar — refresh Homepage (Ctrl+Shift+R)"
            else:
                _apply_status["log"] += "\n✗ Deploy mislukt"
        except Exception as e:
            _apply_status["ok"] = False
            _apply_status["log"] += f"\nFout: {e}"
        finally:
            _apply_status["running"] = False


app = FastAPI(title="Homepage Admin")
static_dir = Path(__file__).parent / "static"
app.mount("/static", StaticFiles(directory=str(static_dir)), name="static")


@app.get("/", response_class=HTMLResponse)
def index():
    return FileResponse(static_dir / "index.html")


@app.get("/api/groups")
def api_groups():
    return [{"name": g, "tab": TAB_FOR_GROUP.get(g, "")} for g in GROUPS]


@app.get("/api/sites")
def api_sites():
    return read_sites()


@app.post("/api/sites")
def api_add(site: SiteIn):
    if site.group not in GROUPS:
        raise HTTPException(400, f"Onbekende groep: {site.group}")
    name = site.name.strip()
    url = site.url.strip()
    if not name or not url:
        raise HTTPException(400, "Naam en URL zijn verplicht")
    if not re.match(r"^https?://", url, re.I):
        raise HTTPException(400, "URL moet beginnen met http:// of https://")

    rows = [(s["group"], s["name"], s["url"]) for s in read_sites()]
    rows.append((site.group, name, url))
    write_sites(rows)
    return {"ok": True}


@app.delete("/api/sites/{site_id}")
def api_delete(site_id: int):
    sites = read_sites()
    if site_id < 0 or site_id >= len(sites):
        raise HTTPException(404, "Site niet gevonden")
    rows = [(s["group"], s["name"], s["url"]) for i, s in enumerate(sites) if i != site_id]
    write_sites(rows)
    return {"ok": True}


@app.post("/api/apply")
def api_apply():
    if _apply_status.get("running"):
        raise HTTPException(409, "Bezig met toepassen…")
    t = threading.Thread(target=run_apply, daemon=True)
    t.start()
    return {"ok": True, "message": "Gestart"}


@app.get("/api/apply/status")
def api_apply_status():
    return _apply_status


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=3010)