Perplexity Bot
113 lines
3.5 KiB
Python
113 lines
3.5 KiB
Python
import os
|
|
import systemd.dbus
|
|
import dbus
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from pydantic import BaseModel
|
|
from typing import List
|
|
|
|
app = FastAPI(title="NetMapper Agent", version="0.1.0")
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"], # da limitare in produzione
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
ALLOWED_SERVICES = [
|
|
"netmapper-frontend.service",
|
|
"netmapper-backend.service",
|
|
"netmapper-worker.service",
|
|
]
|
|
|
|
class ServiceInfo(BaseModel):
|
|
name: str
|
|
active_state: str
|
|
sub_state: str
|
|
loaded: str
|
|
pid: int | None = None
|
|
|
|
class ActionRequest(BaseModel):
|
|
mode: str = "sync"
|
|
|
|
def get_unit(name: str):
|
|
bus = dbus.SystemBus()
|
|
systemd = bus.get_object('org.freedesktop.systemd1', '/org/freedesktop/systemd1')
|
|
manager = dbus.Interface(systemd, 'org.freedesktop.systemd1.Manager')
|
|
unit_path = manager.GetUnit(name)
|
|
unit = bus.get_object('org.freedesktop.systemd1', str(unit_path))
|
|
props = dbus.Interface(unit, 'org.freedesktop.DBus.Properties')
|
|
return props.GetAll('org.freedesktop.systemd1.Unit')
|
|
|
|
def refresh_dbus():
|
|
bus = dbus.SystemBus()
|
|
systemd = bus.get_object('org.freedesktop.systemd1', '/org/freedesktop/systemd1')
|
|
manager = dbus.Interface(systemd, 'org.freedesktop.systemd1.Manager')
|
|
return manager
|
|
|
|
@app.get("/health")
|
|
def health():
|
|
return {"status": "ok"}
|
|
|
|
@app.get("/services", response_model=List[ServiceInfo])
|
|
def list_services():
|
|
bus = dbus.SystemBus()
|
|
systemd = bus.get_object('org.freedesktop.systemd1', '/org/freedesktop/systemd1')
|
|
manager = dbus.Interface(systemd, 'org.freedesktop.systemd1.Manager')
|
|
units = manager.ListUnits()
|
|
result = []
|
|
for u in units:
|
|
name = u[0]
|
|
if name not in ALLOWED_SERVICES:
|
|
continue
|
|
result.append(ServiceInfo(
|
|
name=name,
|
|
active_state=u[3],
|
|
sub_state=u[4],
|
|
loaded=u[2],
|
|
pid=None,
|
|
))
|
|
return result
|
|
|
|
@app.get("/services/{name}")
|
|
def get_service(name: str):
|
|
if name not in ALLOWED_SERVICES:
|
|
raise HTTPException(status_code=403, detail="Service not allowed")
|
|
props = get_unit(name)
|
|
pid = None
|
|
if "MainPID" in props:
|
|
pid = int(props["MainPID"]) or None
|
|
return ServiceInfo(
|
|
name=name,
|
|
active_state=str(props.get("ActiveState", "unknown")),
|
|
sub_state=str(props.get("SubState", "unknown")),
|
|
loaded=str(props.get("LoadState", "unknown")),
|
|
pid=pid,
|
|
)
|
|
|
|
@app.post("/services/{name}/start")
|
|
def start_service(name: str):
|
|
if name not in ALLOWED_SERVICES:
|
|
raise HTTPException(status_code=403, detail="Service not allowed")
|
|
manager = refresh_dbus()
|
|
job = manager.StartUnit(name, "replace")
|
|
return {"job": str(job), "service": name, "action": "start"}
|
|
|
|
@app.post("/services/{name}/stop")
|
|
def stop_service(name: str):
|
|
if name not in ALLOWED_SERVICES:
|
|
raise HTTPException(status_code=403, detail="Service not allowed")
|
|
manager = refresh_dbus()
|
|
job = manager.StopUnit(name, "replace")
|
|
return {"job": str(job), "service": name, "action": "stop"}
|
|
|
|
@app.post("/services/{name}/restart")
|
|
def restart_service(name: str):
|
|
if name not in ALLOWED_SERVICES:
|
|
raise HTTPException(status_code=403, detail="Service not allowed")
|
|
manager = refresh_dbus()
|
|
job = manager.RestartUnit(name, "replace")
|
|
return {"job": str(job), "service": name, "action": "restart"}
|