#!/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("/api/docker/stats") async def docker_stats(): """Get Docker container stats from CT 100 via Proxmox SSH.""" import subprocess try: # Run docker stats on CT 100 via proxmox (use raw output to avoid Go template escaping) result = subprocess.run( ["sshpass", "-p", _PVE_PASS, "ssh", "-o", "StrictHostKeyChecking=no", "root@192.168.0.144", "pct", "exec", "100", "--", "docker", "stats", "--no-stream"], capture_output=True, text=True, timeout=15 ) if result.returncode != 0: return {"error": result.stderr.strip(), "containers": []} containers = [] for line in result.stdout.strip().split("\n"): if not line.strip() or line.startswith("CONTAINER"): continue # Parse default docker stats output: # CONTAINER ID NAME CPU % MEM USAGE / LIMIT MEM % NET I/O BLOCK I/O PIDS parts = line.split() if len(parts) < 8: continue # Account for container ID being first column cid = parts[0] name = parts[1] cpu = parts[2].replace("%", "") mem_usage = parts[3] # e.g., "311.2MiB" mem_limit = parts[5] # e.g., "2GiB" (parts[4] is "/") mem_pct = parts[6].replace("%", "") # NET I/O: parts[7] and parts[9] with "/" at parts[8] # BLOCK I/O: parts[10] and parts[12] with "/" at parts[11] # PIDS: last element (could be at different positions depending on name length) # More reliable: find from the right pids = parts[-1] # Block I/O is tricky because of spaces in column names; # use a simpler approach: rest of line after NET I/O position # Actually, docker stats columns are fixed-width, let's just grab what we can net_in = parts[7] if len(parts) > 7 else "0" net_out = parts[9] if len(parts) > 9 else "0" blk_in = parts[10] if len(parts) > 10 else "0" blk_out = parts[12] if len(parts) > 12 else "0" containers.append({ "name": name, "cpu": cpu, "mem_pct": mem_pct, "mem_used": mem_usage, "mem_limit": mem_limit, "net_in": net_in, "net_out": net_out, "blk_in": blk_in, "blk_out": blk_out, "pids": pids, }) return {"containers": containers, "host": "CT 100 (docker)", "count": len(containers)} except Exception as e: return {"error": str(e)[:200], "containers": []} @app.get("/api/windows/stats") async def windows_stats(): """Get live stats from Windows Gaming PC via PowerShell over SSH.""" import subprocess, base64, json as _json # PowerShell script: backslash counter paths need single \, Python raw string preserves them ps_script = r''' $cpu=(Get-Counter '\Processor(_Total)\% Processor Time').CounterSamples[0].CookedValue $os=Get-CimInstance Win32_OperatingSystem $disk=Get-CimInstance Win32_LogicalDisk -Filter "DeviceID='C:'" $uptime=(Get-Date)-$os.LastBootUpTime @{ 'cpu'=[math]::Round($cpu,1) 'mem_total_gb'=[math]::Round($os.TotalVisibleMemorySize/1MB,1) 'mem_free_gb'=[math]::Round($os.FreePhysicalMemory/1MB,1) 'disk_c_total_gb'=[math]::Round($disk.Size/1GB,1) 'disk_c_free_gb'=[math]::Round($disk.FreeSpace/1GB,1) 'uptime_days'=[math]::Round($uptime.TotalDays,1) } | ConvertTo-Json '''.strip() try: encoded = base64.b64encode(ps_script.encode('utf-16le')).decode() result = subprocess.run( ["ssh", "gamingpc", "powershell", "-NoProfile", "-EncodedCommand", encoded], capture_output=True, text=True, timeout=30 ) if result.returncode != 0: return {"error": result.stderr.strip()[:300], "online": False} # Filter CLIXML noise from stdout clean = "\n".join(l for l in result.stdout.split("\n") if l.strip() and not l.startswith("#<") and not l.startswith("<")) data = _json.loads(clean) data["online"] = True data["mem_used_gb"] = round(data["mem_total_gb"] - data["mem_free_gb"], 1) data["disk_c_used_gb"] = round(data["disk_c_total_gb"] - data["disk_c_free_gb"], 1) return data except Exception as e: return {"error": str(e)[:300], "online": False} @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

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