"""SaraSortd's systemd user unit: status, start/stop/restart, autostart, and its journal. SaraSortd's own log files are read from its LogDir.""" import subprocess from dataclasses import dataclass from pathlib import Path from Core import ChildEnvironment @dataclass class ServiceStatus: Active: str = "unknown" # active, inactive, failed, activating, ... Sub: str = "" # running, dead, ... Enabled: str = "unknown" # enabled, disabled, ... Pid: str = "" Since: str = "" Found: bool = False @property def Running(self) -> bool: return self.Active in ("active", "activating", "reloading") def _Run(Arguments: list[str], Timeout: int = 15) -> subprocess.CompletedProcess: return subprocess.run(Arguments, capture_output=True, text=True, timeout=Timeout, env=ChildEnvironment()) def Status(Unit: str) -> ServiceStatus: try: Result = _Run(["systemctl", "--user", "show", Unit, "-p", "LoadState,ActiveState,SubState,UnitFileState,MainPID,ActiveEnterTimestamp"]) except (OSError, subprocess.SubprocessError): return ServiceStatus() Values = dict(Line.split("=", 1) for Line in Result.stdout.splitlines() if "=" in Line) return ServiceStatus( Active=Values.get("ActiveState", "unknown"), Sub=Values.get("SubState", ""), Enabled=Values.get("UnitFileState", "") or "unknown", Pid=Values.get("MainPID", "") if Values.get("MainPID") != "0" else "", Since=Values.get("ActiveEnterTimestamp", ""), Found=Values.get("LoadState") == "loaded", ) def Control(Unit: str, Action: str) -> str: """Action: start, stop, restart, enable, disable. Returns "" or an error.""" try: Result = _Run(["systemctl", "--user", Action, Unit]) except (OSError, subprocess.SubprocessError) as Error: return str(Error) return "" if Result.returncode == 0 else (Result.stderr.strip() or f"systemctl {Action} failed") def RestartIfRunning(Unit: str) -> str: """For after a GlobalConf.toml change: SaraSortd only reads it at start.""" if not Status(Unit).Running: return "" return Control(Unit, "restart") def Journal(Unit: str, Lines: int) -> list[str]: try: Result = _Run(["journalctl", "--user", "-u", Unit, "-n", str(Lines), "--no-pager", "-o", "short-iso"]) except (OSError, subprocess.SubprocessError) as Error: return [f"Couldn't read the journal: {Error}"] return Result.stdout.splitlines() or [Result.stderr.strip() or "(empty)"] def LogFiles(LogDir: Path) -> list[Path]: """Newest first.""" if not LogDir.is_dir(): return [] return sorted((Item for Item in LogDir.iterdir() if Item.is_file()), key=lambda Item: Item.stat().st_mtime, reverse=True) def ReadTail(FilePath: Path, Lines: int) -> list[str]: try: return FilePath.read_text(encoding="utf-8", errors="replace").splitlines()[-Lines:] except OSError as Error: return [f"Couldn't read {FilePath}: {Error}"]