40 lines
1.4 KiB
Python
40 lines
1.4 KiB
Python
from app.models.device import Device
|
|
from app.models.object import Object
|
|
from app.plugins.base import ActionResult, BaseAction, Emitter
|
|
from app.services import ssh as ssh_service
|
|
|
|
|
|
class GetTimedatectlAction(BaseAction):
|
|
name = "get_timedatectl"
|
|
display_name = "Get Time/NTP Info"
|
|
compatible_categories = ["raspberry", "server"]
|
|
params_schema = []
|
|
|
|
async def execute(self, device: Device, obj: Object, params: dict, emit: Emitter) -> ActionResult:
|
|
await emit(f"[{device.ip}] Running timedatectl...", "info")
|
|
result = await ssh_service.run_command(device, obj, "timedatectl show --no-pager 2>/dev/null || timedatectl")
|
|
|
|
if result.error:
|
|
await emit(f"[{device.ip}] Error: {result.error}", "error")
|
|
return ActionResult(success=False, error=result.error)
|
|
|
|
parsed = _parse_timedatectl(result.stdout)
|
|
await emit(f"[{device.ip}] timezone={parsed.get('Timezone', '?')} ntp={parsed.get('NTP', '?')}", "info")
|
|
|
|
return ActionResult(
|
|
success=result.success,
|
|
stdout=result.stdout,
|
|
data=parsed,
|
|
)
|
|
|
|
|
|
def _parse_timedatectl(output: str) -> dict:
|
|
data: dict = {}
|
|
for line in output.splitlines():
|
|
if "=" in line:
|
|
key, _, value = line.partition("=")
|
|
data[key.strip()] = value.strip()
|
|
elif ":" in line:
|
|
key, _, value = line.partition(":")
|
|
data[key.strip()] = value.strip()
|
|
return data
|