foodlinkk-command-center/scripts/hermes-webbuilder-api.py

#!/usr/bin/env python3
"""Hermes Web Builder API — agy builds voor Cockpit Web Builder agent."""
from __future__ import annotations

import asyncio
import json
import os
import re
import subprocess
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

app = FastAPI(title="Hermes Web Builder API", version="1.0.0")

PROJECTS_BASE = Path(os.getenv("WEBBUILDER_PROJECTS", "/home/hermes/projects"))
PREVIEW_PORT = int(os.getenv("WEBBUILDER_PREVIEW_PORT", "8080"))
PREVIEW_HOST = os.getenv("WEBBUILDER_PREVIEW_HOST", "10.4.7.27")
NAS_PREFIX = os.getenv("WEBBUILDER_NAS_PREFIX", "//10.4.7.11/share/Websites")
AGY_BIN = os.getenv("AGY_BIN", "/home/hermes/.local/bin/agy")
BUILD_TIMEOUT = os.getenv("WEBBUILDER_BUILD_TIMEOUT", "30m")

JOBS: dict[str, dict[str, Any]] = {}


def slugify(name: str) -> str:
    s = (name or "").strip().lower()
    s = re.sub(r"[^a-z0-9_-]+", "-", s)
    s = re.sub(r"-+", "-", s).strip("-")
    return s[:64] or "website"


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _list_files(project_dir: Path) -> list[str]:
    if not project_dir.is_dir():
        return []
    out: list[str] = []
    for p in sorted(project_dir.rglob("*")):
        if p.is_file():
            out.append(str(p.relative_to(project_dir)))
    return out[:50]


def _preview_url() -> str:
    return f"http://{PREVIEW_HOST}:{PREVIEW_PORT}/"


async def _run_shell(cmd: str, *, timeout: float = 120.0) -> tuple[int, str, str]:
    proc = await asyncio.create_subprocess_shell(
        cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    try:
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
    except asyncio.TimeoutError:
        proc.kill()
        return 124, "", "timeout"
    return proc.returncode or 0, stdout.decode(errors="replace"), stderr.decode(errors="replace")


async def _ensure_project(slug: str) -> Path:
    proj = PROJECTS_BASE / slug
    proj.mkdir(parents=True, exist_ok=True)
    await _run_shell(f"chown hermes:hermes '{proj}'")
    return proj


async def _start_preview(slug: str) -> None:
    cmd = (
        f"pkill -f 'http.server.*{PREVIEW_PORT}' 2>/dev/null || true; "
        f"su - hermes -c 'cd {PROJECTS_BASE}/{slug} && "
        f"nohup python3 -m http.server {PREVIEW_PORT} > /tmp/webbuilder-preview.log 2>&1 &'"
    )
    await _run_shell(cmd, timeout=15.0)


async def _run_build(job_id: str, slug: str, prompt: str) -> None:
    job = JOBS[job_id]
    job["status"] = "running"
    job["message"] = f"agy bouwt {slug}…"
    job["updated_at"] = _now()

    proj = await _ensure_project(slug)
    safe_prompt = prompt.replace('"', '\\"')[:8000]
    agy_cmd = (
        f'su - hermes -c \'export PATH="$HOME/.local/bin:$PATH"; '
        f"cd '{proj}' && "
        f"{AGY_BIN} --print --dangerously-skip-permissions --print-timeout {BUILD_TIMEOUT} "
        f'"{safe_prompt}"\''
    )

    job["message"] = "Antigravity CLI gestart…"
    rc, out, err = await _run_shell(agy_cmd, timeout=3600.0)
    log_tail = (out + "\n" + err).strip()[-4000:]
    job["log_tail"] = log_tail
    job["updated_at"] = _now()

    files = _list_files(proj)
    job["files"] = files

    if rc == 0 and files:
        await _start_preview(slug)
        job["status"] = "completed"
        job["message"] = f"Website klaar — {len(files)} bestand(en)"
        job["preview_url"] = _preview_url()
        job["completed_at"] = _now()
    elif rc == 0:
        job["status"] = "completed"
        job["message"] = "agy klaar (geen bestanden gevonden)"
        job["preview_url"] = _preview_url() if files else None
        job["completed_at"] = _now()
    else:
        job["status"] = "failed"
        job["message"] = f"agy exit code {rc}"
        job["error"] = log_tail[-800:] or f"exit {rc}"
        job["completed_at"] = _now()


class BuildRequest(BaseModel):
    project: str = Field(..., min_length=1, max_length=64)
    prompt: str = Field(..., min_length=1, max_length=8000)


@app.get("/health")
async def health() -> dict[str, Any]:
    return {"ok": True, "service": "hermes-webbuilder", "projects_base": str(PROJECTS_BASE)}


@app.get("/api/projects")
async def list_projects() -> dict[str, Any]:
    items: list[dict[str, Any]] = []
    if PROJECTS_BASE.is_dir():
        for p in sorted(PROJECTS_BASE.iterdir()):
            if p.is_dir() and not p.name.startswith("."):
                files = _list_files(p)
                items.append({"slug": p.name, "files": len(files), "has_index": (p / "index.html").is_file()})
    return {"ok": True, "items": items, "count": len(items)}


@app.post("/api/build")
async def start_build(body: BuildRequest) -> dict[str, Any]:
    slug = slugify(body.project)
    if not slug:
        raise HTTPException(400, "Ongeldige projectnaam")

    await _ensure_project(slug)
    job_id = str(uuid.uuid4())
    JOBS[job_id] = {
        "job_id": job_id,
        "project": slug,
        "status": "queued",
        "message": "Build in wachtrij",
        "prompt": body.prompt[:500],
        "nas_path": f"{NAS_PREFIX}/{slug}/",
        "preview_url": _preview_url(),
        "created_at": _now(),
        "updated_at": _now(),
    }
    JOBS[slug] = JOBS[job_id]

    asyncio.create_task(_run_build(job_id, slug, body.prompt))
    return {
        "ok": True,
        "job_id": job_id,
        "project": slug,
        "status": "queued",
        "nas_path": f"{NAS_PREFIX}/{slug}/",
        "preview_url": _preview_url(),
    }


@app.get("/api/build/{project}/status")
async def build_status(project: str) -> dict[str, Any]:
    slug = slugify(project)
    job = JOBS.get(slug) or JOBS.get(project)
    if not job:
        proj = PROJECTS_BASE / slug
        files = _list_files(proj)
        if files:
            return {
                "ok": True,
                "project": slug,
                "status": "completed",
                "message": "Project bestaat (geen actieve job)",
                "files": files,
                "preview_url": _preview_url(),
                "nas_path": f"{NAS_PREFIX}/{slug}/",
            }
        return {"ok": True, "project": slug, "status": "idle", "message": "Geen build actief", "files": []}

    return {
        "ok": True,
        "project": slug,
        "job_id": job.get("job_id"),
        "status": job.get("status"),
        "message": job.get("message"),
        "files": job.get("files") or [],
        "preview_url": job.get("preview_url"),
        "nas_path": f"{NAS_PREFIX}/{slug}/",
        "log_tail": job.get("log_tail", ""),
        "error": job.get("error"),
        "updated_at": job.get("updated_at"),
    }


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=int(os.getenv("PORT", "8798")))