SaraSort/Service.py

"""SaraSortd as a background service: status, start/stop/restart, autostart,
and its logs. SaraSortd's own log files are read from its LogDir.

Linux with systemd: its systemd user unit (systemctl / journalctl).
Windows, and Linux without systemd (runit, OpenRC, s6, ...): SaraSuite runs
it itself (its Services.py), and this does the same:
- it runs as `python ServiceRun.py SaraSortd SaraSortd.py <ConfigDir>`
  (pythonw.exe on Windows); ServiceRun.py restarts it if it fails, keeps its
  output in SaraSuite's Services folder as SaraSortd.log (the "Service log"
  here) and records its PID in SaraSortd.pid (that's how "running" is told);
- autostart is a shortcut to that command in the Startup folder (Windows)
  or an XDG autostart entry, ~/.config/autostart/SaraSortd.desktop (Linux).
Unit is the systemd unit name ("SaraSortd.service"); run by SaraSuite, the
service is called the same without ".service".
"""

import os
import re
import signal
import subprocess
import time
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path

from Core import AutostartDir, ChildEnvironment, ConfigHome, DataHome, IsWindows, LoadConfig, RunBySaraSuite


@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")


_NoWindow = getattr(subprocess, "CREATE_NO_WINDOW", 0)


def _Run(Arguments: list[str], Timeout: int = 15) -> subprocess.CompletedProcess:
    return subprocess.run(Arguments, capture_output=True, text=True, timeout=Timeout, env=ChildEnvironment(),
                          creationflags=_NoWindow)


# --- Run by SaraSuite (Windows; Linux without systemd) ------------------------------------------


def _Name(Unit: str) -> str:
    return Unit.removesuffix(".service")


def _ServicesFile(Name: str, Suffix: str) -> Path:
    return DataHome / "SaraSuite" / "Services" / f"{Name}{Suffix}"


def _AutostartFile(Name: str) -> Path:
    if IsWindows:
        return ConfigHome / "Microsoft" / "Windows" / "Start Menu" / "Programs" / "Startup" / f"{Name}.lnk"
    return AutostartDir / f"{Name}.desktop"


def _Command(Name: str) -> tuple[Path, list[str], Path]:
    """(python, its arguments, working folder), from SaraSuite's install layout."""
    Config = LoadConfig()
    ProgramDir = Path(Config.SortdAppDir).expanduser().parent
    Folder = Path(Config.SortdConfigDir).expanduser()
    if IsWindows:
        Python = ProgramDir / ".venv" / "Scripts" / "pythonw.exe"
        Arguments = ["-X", "utf8"]
    else:
        Python = ProgramDir / ".venv" / "bin" / "python"
        Arguments = []
    Arguments += [str(ProgramDir / "ServiceRun.py"), Name, str(Config.SortdScriptPath), str(Folder)]
    return Python, Arguments, Folder


def _Pid(Name: str) -> str:
    try:
        Pid = int(_ServicesFile(Name, ".pid").read_text().strip())
    except (OSError, ValueError):
        return ""
    if not IsWindows:
        try:
            return str(Pid) if b"ServiceRun.py" in Path(f"/proc/{Pid}/cmdline").read_bytes() else ""
        except OSError:
            return ""
    try:
        Output = _Run(["tasklist", "/FI", f"PID eq {Pid}", "/NH", "/FO", "CSV"]).stdout
    except (OSError, subprocess.SubprocessError):
        return ""
    return str(Pid) if re.search(rf'"{Pid}"', Output) else ""


def _OwnStatus(Name: str) -> ServiceStatus:
    Python, Arguments, _Folder = _Command(Name)
    Pid = _Pid(Name)
    Since = ""
    if Pid:
        Since = datetime.fromtimestamp(_ServicesFile(Name, ".pid").stat().st_mtime).strftime("%Y-%m-%d %H:%M:%S")
    Runner = Path(Arguments[-4])
    return ServiceStatus(
        Active="active" if Pid else "inactive",
        Sub="running" if Pid else "dead",
        Enabled="enabled" if _AutostartFile(Name).exists() else "disabled",
        Pid=Pid,
        Since=Since,
        Found=Python.is_file() and Runner.is_file(),
    )


def _AutostartEntry(Python: Path, Arguments: list[str], Name: str) -> str:
    """The same entry SaraSuite writes (its Services.AutostartEntry)."""
    def Quote(Text: str) -> str:
        Text = Text.replace("%", "%%")
        return Text if re.fullmatch(r"[\w@%+=:,./-]+", Text) else '"' + re.sub(r'(["`$\\])', r"\\\1", Text) + '"'

    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={' '.join(Quote(Part) for Part in [str(Python), *Arguments])}\n"
        "Terminal=false\n"
        "NoDisplay=true\n"
        "X-GNOME-Autostart-enabled=true\n"
    )


def _OwnControl(Name: str, Action: str) -> str:
    Python, Arguments, Folder = _Command(Name)
    try:
        if Action in ("stop", "restart"):
            Pid = _Pid(Name)
            if Pid and IsWindows:
                _Run(["taskkill", "/PID", Pid, "/T", "/F"])
            elif Pid:
                os.kill(int(Pid), signal.SIGTERM)  # ServiceRun.py finishes cleanly
                for _ in range(50):
                    if not _Pid(Name):
                        break
                    time.sleep(0.2)
                else:
                    os.kill(int(Pid), signal.SIGKILL)
            _ServicesFile(Name, ".pid").unlink(missing_ok=True)
        if Action in ("start", "restart"):
            if _Pid(Name):
                return ""
            Folder.mkdir(parents=True, exist_ok=True)
            Extra = ({"creationflags": getattr(subprocess, "DETACHED_PROCESS", 0) | _NoWindow} if IsWindows
                     else {"start_new_session": True})
            subprocess.Popen(
                [str(Python), *Arguments], cwd=str(Folder), env=ChildEnvironment(), close_fds=True,
                stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, **Extra,
            )
        elif Action == "enable":
            Entry = _AutostartFile(Name)
            Entry.parent.mkdir(parents=True, exist_ok=True)
            if IsWindows:
                import pylnk3

                Quoted = " ".join(f'"{Item}"' if " " in Item or "\\" in Item else Item for Item in Arguments)
                pylnk3.for_file(str(Python), str(Entry), arguments=Quoted, work_dir=str(Folder),
                                description=f"{Name} (background service)")
            else:
                Entry.write_text(_AutostartEntry(Python, Arguments, Name))
        elif Action == "disable":
            _AutostartFile(Name).unlink(missing_ok=True)
    except (OSError, subprocess.SubprocessError) as Error:
        return str(Error)
    return ""


# --- Both ------------------------------------------------------------------------------------


def Status(Unit: str) -> ServiceStatus:
    if RunBySaraSuite:
        return _OwnStatus(_Name(Unit))
    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."""
    if RunBySaraSuite:
        return _OwnControl(_Name(Unit), Action)
    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]:
    """Its systemd journal, or (run by SaraSuite) ServiceRun.py's log of its output."""
    if RunBySaraSuite:
        LogPath = _ServicesFile(_Name(Unit), ".log")
        if not LogPath.is_file():
            return [f"No service log yet ({LogPath}): it's written once the service has run."]
        return ReadTail(LogPath, Lines)
    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}"]