from __future__ import annotations import subprocess ALLOWED_SERVICES = { "biliup-next-worker.service", "biliup-next-api.service", } ALLOWED_ACTIONS = {"start", "stop", "restart"} class SystemdRuntime: def list_services(self) -> dict[str, object]: items = [] for service in sorted(ALLOWED_SERVICES): items.append(self._inspect_service(service)) return {"items": items} def act(self, service: str, action: str) -> dict[str, object]: if service not in ALLOWED_SERVICES: raise ValueError(f"unsupported service: {service}") if action not in ALLOWED_ACTIONS: raise ValueError(f"unsupported action: {action}") result = subprocess.run( ["sudo", "systemctl", action, service], capture_output=True, text=True, check=False, ) payload = self._inspect_service(service) payload["action"] = action payload["command_ok"] = result.returncode == 0 payload["stderr"] = (result.stderr or "").strip() payload["stdout"] = (result.stdout or "").strip() return payload def _inspect_service(self, service: str) -> dict[str, object]: show = subprocess.run( [ "systemctl", "show", service, "--property=Id,Description,LoadState,ActiveState,SubState,MainPID,ExecMainStatus,FragmentPath", ], capture_output=True, text=True, check=False, ) info = { "id": service, "description": "", "load_state": "unknown", "active_state": "unknown", "sub_state": "unknown", "main_pid": 0, "exec_main_status": None, "fragment_path": "", } for line in (show.stdout or "").splitlines(): if "=" not in line: continue key, value = line.split("=", 1) if key == "Id": info["id"] = value elif key == "Description": info["description"] = value elif key == "LoadState": info["load_state"] = value elif key == "ActiveState": info["active_state"] = value elif key == "SubState": info["sub_state"] = value elif key == "MainPID": try: info["main_pid"] = int(value) except ValueError: info["main_pid"] = 0 elif key == "ExecMainStatus": info["exec_main_status"] = value elif key == "FragmentPath": info["fragment_path"] = value return info