#!/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")))