Perplexity Bot
65 lines
2.5 KiB
Python
65 lines
2.5 KiB
Python
from datetime import datetime, timezone
|
|
from typing import List
|
|
import dbus
|
|
|
|
from app.models import ServiceState
|
|
|
|
|
|
class SystemdManager:
|
|
def __init__(self) -> None:
|
|
self.bus = dbus.SystemBus()
|
|
systemd = self.bus.get_object('org.freedesktop.systemd1', '/org/freedesktop/systemd1')
|
|
self.manager = dbus.Interface(systemd, 'org.freedesktop.systemd1.Manager')
|
|
|
|
def list_services(self, names: list[str]) -> List[ServiceState]:
|
|
items: list[ServiceState] = []
|
|
for name in names:
|
|
try:
|
|
unit_path = self.manager.GetUnit(name)
|
|
unit = self.bus.get_object('org.freedesktop.systemd1', unit_path)
|
|
props = dbus.Interface(unit, 'org.freedesktop.DBus.Properties')
|
|
active_state = str(props.Get('org.freedesktop.systemd1.Unit', 'ActiveState'))
|
|
sub_state = str(props.Get('org.freedesktop.systemd1.Unit', 'SubState'))
|
|
loaded = str(props.Get('org.freedesktop.systemd1.Unit', 'LoadState'))
|
|
description = str(props.Get('org.freedesktop.systemd1.Unit', 'Description'))
|
|
try:
|
|
pid = int(props.Get('org.freedesktop.systemd1.Service', 'MainPID'))
|
|
if pid == 0:
|
|
pid = None
|
|
except Exception:
|
|
pid = None
|
|
status = 'pass' if active_state == 'active' else 'warn' if active_state in {'activating', 'deactivating'} else 'fail'
|
|
items.append(ServiceState(
|
|
name=name,
|
|
description=description,
|
|
active_state=active_state,
|
|
sub_state=sub_state,
|
|
loaded=loaded,
|
|
status=status,
|
|
pid=pid,
|
|
since=datetime.now(timezone.utc),
|
|
))
|
|
except Exception as exc:
|
|
items.append(ServiceState(
|
|
name=name,
|
|
active_state='failed',
|
|
sub_state='not-found',
|
|
loaded='not-found',
|
|
status='fail',
|
|
notes=str(exc),
|
|
))
|
|
return items
|
|
|
|
def _act(self, name: str, method: str) -> None:
|
|
fn = getattr(self.manager, method)
|
|
fn(name, 'replace')
|
|
|
|
def start(self, name: str) -> None:
|
|
self._act(name, 'StartUnit')
|
|
|
|
def stop(self, name: str) -> None:
|
|
self._act(name, 'StopUnit')
|
|
|
|
def restart(self, name: str) -> None:
|
|
self._act(name, 'RestartUnit')
|