#!/usr/bin/env python3 """Serve project-directory static HTML with a /api/status proxy.""" import os from pathlib import Path from fastapi import FastAPI, Request from fastapi.responses import HTMLResponse, Response from fastapi.staticfiles import StaticFiles import httpx SCRIPT_DIR = Path(__file__).resolve().parent STATIC_DIR = SCRIPT_DIR PORT = int(os.environ.get("SERVE_PORT", "18834")) HOST = os.environ.get("SERVE_HOST", "127.0.0.1") app = FastAPI(title="Project Directory") @app.get("/api/status") async def status_proxy(): """Proxy status from the internal network-status service.""" try: async with httpx.AsyncClient( timeout=httpx.Timeout(10.0), ) as client: r = await client.get("http://127.0.0.1:18898/api/status") return Response(content=r.text, media_type="application/json", status_code=r.status_code) except Exception: return {"error": "status service unreachable"} import base64 as _base64 import time as _time import asyncio as _asyncio _SYSDB_AUTH = "Basic " + _base64.b64encode(b"shawn:Brett85!@").decode() # Proxmox host configs _PVE_HOSTS = [ {"id": "primary", "name": "Primary Proxmox", "url": "https://192.168.0.144:8006", "node": "pve"}, {"id": "secondary", "name": "Secondary Proxmox", "url": "https://admin.ourpad.casa", "node": "ourpad"}, ] _PVE_USER = "root@pam" _PVE_PASS = "Brett85!@" _pve_cache = {"data": None, "ts": 0} _PVE_CACHE_TTL = 30.0 async def _pve_ticket(client: httpx.AsyncClient, base_url: str) -> str: """Get a Proxmox API ticket.""" r = await client.post(f"{base_url}/api2/json/access/ticket", data={"username": _PVE_USER, "password": _PVE_PASS}) return r.json()["data"]["ticket"] async def _pve_node_status(client: httpx.AsyncClient, base_url: str, node: str, ticket: str) -> dict: r = await client.get(f"{base_url}/api2/json/nodes/{node}/status", headers={"Cookie": f"PVEAuthCookie={ticket}"}) d = r.json()["data"] mem = d.get("memory", {}) rootfs = d.get("rootfs", {}) swap = d.get("swap", {}) return { "node": node, "status": d.get("status", "unknown"), "cpu": round(d.get("cpu", 0) * 100, 1), "cpu_model": (d.get("cpuinfo", {}).get("model", "")[:60] if d.get("cpuinfo") else ""), "memory": { "total_mb": round(mem.get("total", 0) / 1024 / 1024, 0), "used_mb": round(mem.get("used", 0) / 1024 / 1024, 0), "free_mb": round(mem.get("free", 0) / 1024 / 1024, 0), "pct": round(mem.get("used", 0) / max(mem.get("total", 1), 1) * 100, 1) if mem else 0, }, "disk": { "total_gb": round(rootfs.get("total", 0) / 1024 / 1024 / 1024, 1), "used_gb": round(rootfs.get("used", 0) / 1024 / 1024 / 1024, 1), "free_gb": round(rootfs.get("free", 0) / 1024 / 1024 / 1024, 1), "pct": round(rootfs.get("used", 0) / max(rootfs.get("total", 1), 1) * 100, 1) if rootfs else 0, }, "swap": { "total_mb": round(swap.get("total", 0) / 1024 / 1024, 0), "used_mb": round(swap.get("used", 0) / 1024 / 1024, 0), }, "uptime_hrs": round(d.get("uptime", 0) / 3600, 1), "pveversion": d.get("pveversion", ""), } async def _pve_lxc_list(client: httpx.AsyncClient, base_url: str, node: str, ticket: str) -> list: r = await client.get(f"{base_url}/api2/json/nodes/{node}/lxc", headers={"Cookie": f"PVEAuthCookie={ticket}"}) results = [] for ct in r.json()["data"]: disk_bytes = ct.get("disk", 0) maxdisk_bytes = ct.get("maxdisk", 0) mem_bytes = ct.get("mem", 0) maxmem_bytes = ct.get("maxmem", 0) results.append({ "vmid": ct["vmid"], "name": ct.get("name", "?"), "status": ct.get("status"), "cpu_pct": round(ct.get("cpu", 0) * 100, 1), "mem_mb": round(mem_bytes / 1024 / 1024, 0), "mem_max_mb": round(maxmem_bytes / 1024 / 1024, 0), "mem_pct": round(mem_bytes / max(maxmem_bytes, 1) * 100, 1), "disk_gb": round(disk_bytes / 1024 / 1024 / 1024, 1), "disk_max_gb": round(maxdisk_bytes / 1024 / 1024 / 1024, 1), "disk_pct": round(disk_bytes / max(maxdisk_bytes, 1) * 100, 1), }) return results @app.get("/api/proxmox/stats") async def proxmox_stats(): """Aggregate stats from both Proxmox hosts (cached 30s).""" global _pve_cache now = _time.time() if _pve_cache["data"] is not None and (now - _pve_cache["ts"]) < _PVE_CACHE_TTL: return _pve_cache["data"] hosts_data = [] async with httpx.AsyncClient(verify=False, timeout=httpx.Timeout(15.0)) as client: for host in _PVE_HOSTS: try: ticket = await _pve_ticket(client, host["url"]) node_data, lxcs = await _asyncio.gather( _pve_node_status(client, host["url"], host["node"], ticket), _pve_lxc_list(client, host["url"], host["node"], ticket), ) hosts_data.append({ "id": host["id"], "name": host["name"], "node": node_data, "lxcs": lxcs, }) except Exception as e: hosts_data.append({ "id": host["id"], "name": host["name"], "error": str(e)[:120], }) data = {"hosts": hosts_data, "timestamp": now} _pve_cache["data"] = data _pve_cache["ts"] = now return data @app.get("/api/system-stats") async def system_stats_proxy(): """Proxy system stats from the internal system-dashboard service.""" try: async with httpx.AsyncClient( timeout=httpx.Timeout(10.0), ) as client: r = await client.get("http://127.0.0.1:8899/api/stats", headers={"Authorization": _SYSDB_AUTH}) return Response(content=r.text, media_type="application/json", status_code=r.status_code) except Exception: return {"error": "system dashboard unreachable"} @app.get("/api/system-history") async def system_history_proxy(): """Proxy history from the internal system-dashboard service.""" try: async with httpx.AsyncClient( timeout=httpx.Timeout(10.0), ) as client: r = await client.get("http://127.0.0.1:8899/api/history", headers={"Authorization": _SYSDB_AUTH}) return Response(content=r.text, media_type="application/json", status_code=r.status_code) except Exception: return {"error": "system dashboard unreachable"} @app.get("/health") async def health(): return {"status": "ok"} @app.get("/{path:path}") async def index(path: str = ""): """Serve static files, fallback to index.html.""" file_path = STATIC_DIR / (path or "index.html") if file_path.is_file(): return HTMLResponse(file_path.read_text()) file_path = STATIC_DIR / "index.html" if file_path.is_file(): return HTMLResponse(file_path.read_text()) return HTMLResponse("

404

", 404) if __name__ == "__main__": import uvicorn uvicorn.run(app, host=HOST, port=PORT)