mirror of
https://git.srhtnin.garden/sarah/SaraSuite.git
synced 2026-10-13 18:18:12 +02:00
124 lines
4.7 KiB
Python
124 lines
4.7 KiB
Python
"""Background services SaraSuite runs itself, where there's no systemd --user
|
|
to do it: Windows, and Linux with another init system (runit on Void, OpenRC,
|
|
s6, dinit, ...). Each runs as Programs/ServiceRun.py <Name> <Script>
|
|
<WorkingDir>, which keeps it going (restarts it if it fails), logs its output
|
|
and records its PID, all in ServicesDir. At login it's started by
|
|
- Windows: a shortcut in the Startup folder;
|
|
- Linux: an XDG autostart entry (~/.config/autostart/<Name>.desktop), which
|
|
desktop sessions start whatever the init system is.
|
|
The installer places those (Installer._ApplyStandInServices); this starts,
|
|
stops and checks the services. SaraSort's Service.py does the same for SaraSortd.
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
import signal
|
|
import subprocess
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Callable, Optional
|
|
|
|
from Core import ChildEnvironment, IsWindows, ServicesDir
|
|
|
|
LogFn = Callable[[str], None]
|
|
|
|
# Windows: child processes of a windowed program would each flash a console window.
|
|
NoWindow = getattr(subprocess, "CREATE_NO_WINDOW", 0)
|
|
_Detached = getattr(subprocess, "DETACHED_PROCESS", 0) | getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)
|
|
|
|
|
|
def _PidFile(Name: str) -> Path:
|
|
return ServicesDir / f"{Name}.pid"
|
|
|
|
|
|
def LogFile(Name: str) -> Path:
|
|
return ServicesDir / f"{Name}.log"
|
|
|
|
|
|
def ServicePid(Name: str) -> Optional[int]:
|
|
"""The PID ServiceRun.py recorded, if that process is still it."""
|
|
try:
|
|
Pid = int(_PidFile(Name).read_text().strip())
|
|
except (OSError, ValueError):
|
|
return None
|
|
if not IsWindows:
|
|
try:
|
|
return Pid if b"ServiceRun.py" in Path(f"/proc/{Pid}/cmdline").read_bytes() else None
|
|
except OSError:
|
|
return None
|
|
try:
|
|
Output = subprocess.run(
|
|
["tasklist", "/FI", f"PID eq {Pid}", "/NH", "/FO", "CSV"],
|
|
capture_output=True, text=True, timeout=15, creationflags=NoWindow,
|
|
).stdout
|
|
except (OSError, subprocess.SubprocessError):
|
|
return None
|
|
return Pid if re.search(rf'"{Pid}"', Output) else None
|
|
|
|
|
|
def Command(Python: Path, Runner: Path, Name: str, Script: Path, WorkingDir: Path) -> list[str]:
|
|
"""What runs the service (Python: pythonw.exe on Windows)."""
|
|
return [str(Python), str(Runner), Name, str(Script), str(WorkingDir)]
|
|
|
|
|
|
def Start(Name: str, Python: Path, Runner: Path, Script: Path, WorkingDir: Path, Log: LogFn) -> None:
|
|
"""Start a service now (it also starts at every login, the same way)."""
|
|
if ServicePid(Name) is not None:
|
|
Log(f"{Name} is already running")
|
|
return
|
|
WorkingDir.mkdir(parents=True, exist_ok=True)
|
|
Extra = {"creationflags": _Detached | NoWindow} if IsWindows else {"start_new_session": True}
|
|
subprocess.Popen(
|
|
Command(Python, Runner, Name, Script, WorkingDir), cwd=str(WorkingDir), env=ChildEnvironment(),
|
|
close_fds=True, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, **Extra,
|
|
)
|
|
Log(f"Started {Name}")
|
|
|
|
|
|
def Stop(Name: str, Log: LogFn) -> None:
|
|
Pid = ServicePid(Name)
|
|
if Pid is not None:
|
|
if IsWindows:
|
|
subprocess.run(["taskkill", "/PID", str(Pid), "/T", "/F"], capture_output=True, creationflags=NoWindow)
|
|
else:
|
|
# ServiceRun.py finishes cleanly on SIGTERM; one that hangs is killed.
|
|
os.kill(Pid, signal.SIGTERM)
|
|
for _ in range(50):
|
|
if ServicePid(Name) is None:
|
|
break
|
|
time.sleep(0.2)
|
|
else:
|
|
os.kill(Pid, signal.SIGKILL)
|
|
Log(f"Stopped {Name}")
|
|
_PidFile(Name).unlink(missing_ok=True)
|
|
|
|
|
|
def Restart(Name: str, Python: Path, Runner: Path, Script: Path, WorkingDir: Path, Log: LogFn) -> None:
|
|
Stop(Name, Log)
|
|
Start(Name, Python, Runner, Script, WorkingDir, Log)
|
|
|
|
|
|
def _ExecArgument(Text: str) -> str:
|
|
"""One argument for a desktop entry's Exec key (the spec's quoting)."""
|
|
Text = Text.replace("%", "%%")
|
|
if re.fullmatch(r"[\w@%+=:,./-]+", Text):
|
|
return Text
|
|
return '"' + re.sub(r'(["`$\\])', r"\\\1", Text) + '"'
|
|
|
|
|
|
def AutostartEntry(Name: str, Python: Path, Runner: Path, Script: Path, WorkingDir: Path) -> str:
|
|
"""Linux: the XDG autostart entry that starts the service at login."""
|
|
Exec = " ".join(_ExecArgument(Part) for Part in Command(Python, Runner, Name, Script, WorkingDir))
|
|
return (
|
|
"# Written by SaraSuite: starts this background service at login on systems\n"
|
|
"# without systemd (runit, OpenRC, s6, ...). Removed with the program.\n"
|
|
"[Desktop Entry]\n"
|
|
"Type=Application\n"
|
|
f"Name={Name}\n"
|
|
f"Comment={Name} (background service, run by SaraSuite)\n"
|
|
f"Exec={Exec}\n"
|
|
"Terminal=false\n"
|
|
"NoDisplay=true\n"
|
|
"X-GNOME-Autostart-enabled=true\n"
|
|
)
|