Files
SaraSuite/Installer.py
T

671 lines
28 KiB
Python

"""Install, update and remove programs described by a manifest.
Install and Update are the same operation (_Apply): make everything match the
manifest, recording each placed piece in the Receipt as it goes. Remove walks
the Receipt plus the program's Leftovers and takes all of it back out.
"""
import configparser
import datetime
import hashlib
import os
import re
import shutil
import subprocess
import tempfile
from dataclasses import dataclass, field
from glob import glob
from pathlib import Path, PurePosixPath
from typing import Callable, Optional
from ConfigMerge import ApplySet, MergeConfig, MergeResult
from Core import BackupsDir, ChildEnvironment, DownloadsDir, LoadConfig, PristineDir
from Manifests import Leftover, Program, Render
from Receipts import IniRecord, LoadReceipt, PlacedFile, Receipt
from Sources import ClearDownloads, Download, LatestCommit, SourceError
LogFn = Callable[[str], None]
class InstallError(Exception):
pass
# --- Small helpers ------------------------------------------------------------
def _Now() -> str:
return datetime.datetime.now().isoformat(timespec="seconds")
def _Run(Command: list[str], Log: LogFn, Check: bool = True, Cwd: Optional[Path] = None) -> int:
"""Run a command, streaming its output into the log."""
Log(f"$ {' '.join(Command)}")
Process = subprocess.Popen(
Command, cwd=Cwd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, env=ChildEnvironment(),
)
for Line in Process.stdout:
Log(f" {Line.rstrip()}")
Code = Process.wait()
if Check and Code != 0:
raise InstallError(f"{Command[0]} failed (exit code {Code})")
return Code
def _Systemctl(Log: LogFn, *Args: str, Check: bool = False) -> int:
return _Run(["systemctl", "--user", *Args], Log, Check=Check)
def _RemovePath(Target: Path) -> None:
if Target.is_symlink() or Target.is_file():
Target.unlink()
elif Target.is_dir():
shutil.rmtree(Target)
def _BackupName(Target: str) -> str:
"""A flat, unique file name for backing up an absolute path."""
return Target.strip("/").replace("/", "__")
def _WriteRoot(Target: str, Text: str, Log: LogFn) -> None:
with tempfile.NamedTemporaryFile("w", delete=False, suffix=".sarasuite") as Temporary:
Temporary.write(Text)
os.chmod(Temporary.name, 0o644)
try:
_Run(["pkexec", "install", "-D", "-m", "644", Temporary.name, Target], Log)
finally:
os.unlink(Temporary.name)
def _RemoveRoot(Target: str, Log: LogFn) -> None:
_Run(["pkexec", "rm", "-f", Target], Log)
def _ReadIni(FilePath: Path) -> configparser.ConfigParser:
Parser = configparser.ConfigParser(interpolation=None, strict=False, delimiters=("=",))
Parser.optionxform = str # keep key case (MIME types, D-Bus names)
if FilePath.exists():
Parser.read(FilePath)
return Parser
def _WriteIni(FilePath: Path, Parser: configparser.ConfigParser) -> None:
FilePath.parent.mkdir(parents=True, exist_ok=True)
with FilePath.open("w") as File:
Parser.write(File, space_around_delimiters=False)
# --- Sources ------------------------------------------------------------------
@dataclass
class Fetched:
"""One downloaded commit of a program, ready to install from."""
Commit: str
Dir: Path
def Fetch(Program: Program, Log: LogFn) -> Fetched:
Commit = LatestCommit(Program.GitUrl, Program.Branch)
return Fetched(Commit, Download(Program.Name, Program.GitUrl, Commit, Log))
def FetchCommit(Program: Program, Commit: str, Log: LogFn) -> Fetched:
"""A specific commit, e.g. the installed one when only options change."""
return Fetched(Commit, Download(Program.Name, Program.GitUrl, Commit, Log))
def DefaultOptions(Program: Program) -> dict[str, bool]:
"""The manifest's option defaults, overridden by the user's own
(Settings → the program's tab)."""
Overrides = LoadConfig().OptionDefaults.get(Program.Name, {})
return {Option.Id: bool(Overrides.get(Option.Id, Option.Default)) for Option in Program.Options}
def DefaultAnswers(Program: Program) -> dict[str, str]:
Overrides = LoadConfig().AnswerDefaults.get(Program.Name, {})
return {Question.Id: str(Overrides.get(Question.Id, Question.Default)) for Question in Program.Questions}
def SourceFiles(Program: Program, SourceDir: Path) -> list[PurePosixPath]:
"""Relative paths of the files to copy, per the manifest's Include/Exclude."""
Found = []
for DirPath, DirNames, FileNames in os.walk(SourceDir):
RelDir = PurePosixPath(Path(DirPath).relative_to(SourceDir).as_posix())
DirNames[:] = sorted(
Name for Name in DirNames
if not any((RelDir / Name).full_match(Pattern) or Name == Pattern for Pattern in Program.Exclude)
)
for Name in sorted(FileNames):
Rel = RelDir / Name if str(RelDir) != "." else PurePosixPath(Name)
if any(Rel.full_match(Pattern) or Name == Pattern for Pattern in Program.Exclude):
continue
if any(Rel.full_match(Pattern) for Pattern in Program.Include):
Found.append(Rel)
return Found
def ManifestHash(Program: Program) -> str:
"""Changes when the bundled manifest/templates change (a new SaraSuite
AppImage), so that shows up as an update even with no new commit."""
Hasher = hashlib.sha256()
for ManifestFile in sorted(Program.Folder.rglob("*")):
if ManifestFile.is_file() and "__pycache__" not in ManifestFile.parts:
Hasher.update(str(ManifestFile.relative_to(Program.Folder)).encode())
Hasher.update(ManifestFile.read_bytes())
return Hasher.hexdigest()
def SourceVersion(Program: Program, SourceDir: Path) -> str:
if not Program.VersionFile:
return ""
VersionPath = SourceDir / Program.VersionFile
if not VersionPath.exists():
return ""
Match = re.search(Program.VersionPattern, VersionPath.read_text(errors="replace"))
return Match[1] if Match else ""
# --- Status -------------------------------------------------------------------
@dataclass
class Status:
Receipt: Optional[Receipt]
LatestCommit: str = "" # empty when the git host couldn't be reached
ManifestHash: str = ""
Error: str = ""
@property
def Installed(self) -> bool:
return self.Receipt is not None
@property
def Broken(self) -> bool:
return self.Receipt is not None and not self.Receipt.Complete
@property
def UpdateAvailable(self) -> bool:
if self.Receipt is None:
return False
NewCommit = bool(self.LatestCommit) and self.LatestCommit != self.Receipt.Commit
return self.Broken or NewCommit or self.ManifestHash != self.Receipt.ManifestHash
def GetStatus(Program: Program) -> Status:
"""Installed state from the receipt, plus the latest commit from the git
host (one small HTTPS request, no download)."""
Current = Status(Receipt=LoadReceipt(Program.Name), ManifestHash=ManifestHash(Program))
try:
Current.LatestCommit = LatestCommit(Program.GitUrl, Program.Branch)
except SourceError as Error:
Current.Error = str(Error)
return Current
# --- Configs --------------------------------------------------------------------
@dataclass
class ConfigPlan:
Target: str
Result: MergeResult
Pristine: str # the new default, saved as the next merge's Base
def PlanConfigs(Program: Program, SourceDir: Path, Answers: dict[str, str]) -> list[ConfigPlan]:
"""What an install/update would do to each config (also shown in the
Update dialog before anything is touched)."""
Context = Program.Context(Answers)
Plans = []
for Config in Program.Configs:
Target = Render(Config.Target, Context)
ConfigSource = SourceDir / Config.Source
if not ConfigSource.exists():
raise InstallError(f"Config {Config.Source} is missing from {SourceDir}")
Theirs = ConfigSource.read_text()
Theirs = ApplySet(Theirs, {Key: Render(Value, Context) for Key, Value in Config.Set.items()})
PristinePath = PristineDir / Program.Name / Path(Target).name
Base = PristinePath.read_text() if PristinePath.exists() else None
Ours = Path(Target).read_text() if Path(Target).exists() else None
Plans.append(ConfigPlan(Target, MergeConfig(Base, Ours, Theirs, Target.endswith(".toml")), Theirs))
return Plans
def _ApplyConfigs(Program: Program, Plans: list[ConfigPlan], NewReceipt: Receipt, Log: LogFn) -> None:
MergeBackups = BackupsDir / Program.Name / "Configs"
Stamp = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
for Plan in Plans:
Target = Path(Plan.Target)
Result = Plan.Result
Log(f"Config {Target.name}: {Result.Describe()}")
for Change in Result.Changes:
Log(f" - {Change}")
if Result.Text is not None:
Target.parent.mkdir(parents=True, exist_ok=True)
if Result.Action == "Merge" and Target.exists():
MergeBackups.mkdir(parents=True, exist_ok=True)
BackupPath = MergeBackups / f"{Target.name}.{Stamp}"
shutil.copy2(Target, BackupPath)
Log(f" (your previous version is backed up at {BackupPath})")
Target.write_text(Result.Text)
NewFile = Target.with_name(Target.name + ".new")
if Result.Action == "KeepAndNew":
NewFile.write_text(Plan.Pristine)
NewReceipt.Files.append(PlacedFile(str(NewFile), "ConfigNew"))
PristinePath = PristineDir / Program.Name / Target.name
PristinePath.parent.mkdir(parents=True, exist_ok=True)
PristinePath.write_text(Plan.Pristine)
NewReceipt.Files.append(PlacedFile(str(Target), "Config"))
# --- Install / Update -------------------------------------------------------------
def _CopyProgramFiles(Program: Program, SourceDir: Path, Log: LogFn) -> None:
Files = SourceFiles(Program, SourceDir)
if not Files:
raise InstallError(f"No files matched Include {Program.Include} in {SourceDir}")
Staging = Program.ProgramDir / "App.new"
_RemovePath(Staging)
for Rel in Files:
Destination = Staging / Rel
Destination.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(SourceDir / Rel, Destination)
_RemovePath(Program.AppDir)
Staging.rename(Program.AppDir)
Log(f"Copied {len(Files)} program files to {Program.AppDir}")
def _RequirementsText(Program: Program) -> str:
if Program.RequirementsFile:
return (Program.AppDir / Program.RequirementsFile).read_text()
return "\n".join(Program.Requirements)
def _SetUpVenv(Program: Program, OldHash: str, Log: LogFn) -> str:
"""(Re)install requirements only when they changed. Returns the new hash."""
if not Program.UsesPython:
return ""
Venv = Program.ProgramDir / ".venv"
NewHash = hashlib.sha256(_RequirementsText(Program).encode()).hexdigest()
if (Venv / "bin" / "python").exists() and NewHash == OldHash:
Log("Python requirements unchanged, skipping pip")
return NewHash
if not (Venv / "bin" / "python").exists():
# Always the system python: inside the AppImage sys.executable is the
# AppImage itself, and the venv has to outlive SaraSuite anyway.
Python = shutil.which("python3")
if Python is None:
raise InstallError(f"{Program.Name} needs python3, and it isn't installed (sudo dnf install python3)")
_Run([Python, "-m", "venv", str(Venv)], Log)
Pip = [str(Venv / "bin" / "python"), "-m", "pip", "install", "--disable-pip-version-check"]
if Program.RequirementsFile:
_Run([*Pip, "-r", str(Program.AppDir / Program.RequirementsFile)], Log)
else:
_Run([*Pip, *Program.Requirements], Log)
return NewHash
def _PutBackDisplaced(Placed: PlacedFile, Log: LogFn) -> None:
if not Placed.Displaced or not Path(Placed.Displaced).exists():
return
Current = Path(Placed.Path)
if Current.exists() and Current.read_bytes() == Path(Placed.Displaced).read_bytes():
pass # already identical (typical for a root file that was never removed)
elif Placed.Root:
_WriteRoot(Placed.Path, Path(Placed.Displaced).read_text(), Log)
else:
shutil.copy2(Placed.Displaced, Placed.Path)
Path(Placed.Displaced).unlink()
Log(f"Restored your previous {Placed.Path}")
def _RemovePlacedTemplate(Placed: PlacedFile, Log: LogFn) -> None:
if Placed.Root and Placed.Displaced:
pass # overwritten in place by _PutBackDisplaced: one password prompt, not two
elif Placed.Root:
if Path(Placed.Path).exists():
_RemoveRoot(Placed.Path, Log)
Log(f"Removed {Placed.Path}")
else:
Path(Placed.Path).unlink(missing_ok=True)
Log(f"Removed {Placed.Path}")
_PutBackDisplaced(Placed, Log)
def _ApplyTemplates(Program: Program, Context: dict, OldReceipt: Optional[Receipt], NewReceipt: Receipt, Log: LogFn) -> None:
Wanted = [Template for Template in Program.Templates if Program.OptionEnabled(Template.Option, NewReceipt.Options)]
WantedTargets = {Render(Template.Target, Context) for Template in Wanted}
# Anything the old install placed that this one no longer wants (an option
# was unticked, or the manifest dropped it).
if OldReceipt is not None:
for Placed in OldReceipt.Files:
if Placed.Kind == "Template" and Placed.Path not in WantedTargets:
_RemovePlacedTemplate(Placed, Log)
for Template in Wanted:
Target = Render(Template.Target, Context)
Text = Render((Program.Folder / "Templates" / Template.Source).read_text(), Context)
Previous = OldReceipt.FindFile(Target) if OldReceipt is not None else None
Placed = PlacedFile(Target, "Template", Root=Template.Root, Displaced=Previous.Displaced if Previous else "")
# A file that's already there but wasn't placed by SaraSuite (e.g. a
# hand-made desktop entry from development) is moved aside, not lost.
if Previous is None and Path(Target).exists():
Displaced = BackupsDir / Program.Name / "Displaced" / _BackupName(Target)
Displaced.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(Target, Displaced)
Placed.Displaced = str(Displaced)
Log(f"Moved your existing {Target} aside (it comes back if {Program.Name} is removed)")
NewReceipt.Files.append(Placed)
if Template.Root:
if Path(Target).exists() and Path(Target).read_text() == Text:
continue
_WriteRoot(Target, Text, Log)
else:
Path(Target).parent.mkdir(parents=True, exist_ok=True)
Path(Target).write_text(Text)
if Template.Executable:
Path(Target).chmod(0o755)
Log(f"Wrote {Target}")
def _RevertIni(Record: IniRecord, Log: LogFn) -> None:
FilePath = Path(Record.File)
Parser = _ReadIni(FilePath)
if not Parser.has_section(Record.Section):
return
Current = Parser.get(Record.Section, Record.Key, fallback=None)
if Current is not None and Current != Record.Value:
Log(f"Left {Record.Key} in {FilePath} alone (it was changed after SaraSuite set it)")
return
if Record.HadPrevious:
Parser.set(Record.Section, Record.Key, Record.Previous)
Log(f"Restored {Record.Key}={Record.Previous} in {FilePath}")
else:
Parser.remove_option(Record.Section, Record.Key)
if not Parser.options(Record.Section):
Parser.remove_section(Record.Section)
Log(f"Removed {Record.Key} from {FilePath}")
if Parser.sections():
_WriteIni(FilePath, Parser)
else:
FilePath.unlink(missing_ok=True)
def _ApplyIniEdits(Program: Program, Context: dict, OldReceipt: Optional[Receipt], NewReceipt: Receipt, Log: LogFn) -> None:
Wanted = []
for Edit in Program.IniEdits:
if Program.OptionEnabled(Edit.Option, NewReceipt.Options):
Wanted.append((Render(Edit.File, Context), Edit.Section, Edit.Key, Render(Edit.Value, Context)))
if OldReceipt is not None:
WantedKeys = {(File, Section, Key) for File, Section, Key, _Value in Wanted}
for Record in OldReceipt.IniEdits:
if (Record.File, Record.Section, Record.Key) not in WantedKeys:
_RevertIni(Record, Log)
for File, Section, Key, Value in Wanted:
Parser = _ReadIni(Path(File))
Previous = OldReceipt.FindIni(File, Section, Key) if OldReceipt is not None else None
if Previous is not None:
Record = IniRecord(File, Section, Key, Value, Previous.HadPrevious, Previous.Previous)
else:
Existing = Parser.get(Section, Key, fallback=None) if Parser.has_section(Section) else None
Record = IniRecord(File, Section, Key, Value, Existing is not None, Existing or "")
NewReceipt.IniEdits.append(Record)
if Parser.has_section(Section) and Parser.get(Section, Key, fallback=None) == Value:
continue
if not Parser.has_section(Section):
Parser.add_section(Section)
Parser.set(Section, Key, Value)
_WriteIni(Path(File), Parser)
Log(f"Set {Key}={Value} in {File}")
def _ApplyServices(Program: Program, OldReceipt: Optional[Receipt], NewReceipt: Receipt, Log: LogFn) -> None:
if not Program.Services and not (OldReceipt and OldReceipt.EnabledServices):
return
_Systemctl(Log, "daemon-reload")
WasEnabled = set(OldReceipt.EnabledServices) if OldReceipt else set()
for Service in Program.Services:
if Program.OptionEnabled(Service.Option, NewReceipt.Options):
NewReceipt.EnabledServices.append(Service.Unit)
if Service.Unit in WasEnabled:
_Systemctl(Log, "try-restart", Service.Unit)
else:
_Systemctl(Log, "enable", "--now", Service.Unit, Check=True)
elif Service.Unit in WasEnabled:
_Systemctl(Log, "disable", "--now", Service.Unit)
def _AfterIntegrationChanges(Program: Program, Log: LogFn) -> None:
Applications = Path(Program.Context()["ApplicationsDir"])
if any(Template.Target.endswith(".desktop") for Template in Program.Templates) and Applications.exists():
_Run(["update-desktop-database", str(Applications)], Log, Check=False)
for Unit in Program.RestartUnits:
_Systemctl(Log, "try-restart", Unit)
def Apply(
Program: Program,
Log: LogFn,
Options: Optional[dict[str, bool]] = None,
Answers: Optional[dict[str, str]] = None,
Source: Optional[Fetched] = None,
) -> Receipt:
"""Install (no receipt yet) or update (receipt exists) a program to the
latest commit, or to Source if it was already fetched (the Update dialog
fetches first to preview the config merges; changing options passes the
installed commit, so nothing else changes).
Answers are only used on install. Options are used on install, and on an
installed program they replace the recorded ones (e.g. turning the file
picker off); otherwise the recorded options are kept."""
OldReceipt = LoadReceipt(Program.Name)
if OldReceipt is None:
Action = "Installing"
elif Options is not None and Source is not None and Source.Commit == OldReceipt.Commit:
Action = "Changing options for"
else:
Action = "Updating"
Log(f"== {Action} {Program.Name}")
Source = Source or Fetch(Program, Log)
SourceDir = Source.Dir
NewReceipt = Receipt(
Name=Program.Name,
Version=SourceVersion(Program, SourceDir),
Commit=Source.Commit,
ManifestHash=ManifestHash(Program),
InstalledAt=OldReceipt.InstalledAt if OldReceipt else _Now(),
UpdatedAt=_Now(),
Options=dict(OldReceipt.Options) if OldReceipt else {},
Answers=dict(OldReceipt.Answers) if OldReceipt else dict(Answers or {}),
)
NewReceipt.Options.update(Options or {})
# Anything not given or recorded yet (e.g. an option added to the manifest
# since this program was installed) starts at the user's/manifest default.
for OptionId, Default in DefaultOptions(Program).items():
NewReceipt.Options.setdefault(OptionId, Default)
for QuestionId, Default in DefaultAnswers(Program).items():
NewReceipt.Answers.setdefault(QuestionId, Default)
for Question in Program.Questions:
if Question.Kind == "Path":
NewReceipt.Answers[Question.Id] = str(Path(NewReceipt.Answers[Question.Id]).expanduser().absolute())
Context = Program.Context(NewReceipt.Answers)
ConfigPlans = PlanConfigs(Program, SourceDir, NewReceipt.Answers)
try:
Program.ProgramDir.mkdir(parents=True, exist_ok=True)
_CopyProgramFiles(Program, SourceDir, Log)
NewReceipt.RequirementsHash = _SetUpVenv(Program, OldReceipt.RequirementsHash if OldReceipt else "", Log)
_ApplyTemplates(Program, Context, OldReceipt, NewReceipt, Log)
_ApplyConfigs(Program, ConfigPlans, NewReceipt, Log)
# .new files from an earlier update that are still around stay tracked.
if OldReceipt is not None:
for Placed in OldReceipt.Files:
if Placed.Kind == "ConfigNew" and Path(Placed.Path).exists() and NewReceipt.FindFile(Placed.Path) is None:
NewReceipt.Files.append(Placed)
_ApplyIniEdits(Program, Context, OldReceipt, NewReceipt, Log)
_ApplyServices(Program, OldReceipt, NewReceipt, Log)
_AfterIntegrationChanges(Program, Log)
NewReceipt.Complete = True
ClearDownloads(Program.Name)
finally:
if OldReceipt is not None and not NewReceipt.Complete:
# A failure partway: keep everything the old receipt knew about too,
# so Remove still finds every file either install placed.
for Placed in OldReceipt.Files:
if NewReceipt.FindFile(Placed.Path) is None:
NewReceipt.Files.append(Placed)
for Record in OldReceipt.IniEdits:
if NewReceipt.FindIni(Record.File, Record.Section, Record.Key) is None:
NewReceipt.IniEdits.append(Record)
NewReceipt.EnabledServices = sorted(set(NewReceipt.EnabledServices) | set(OldReceipt.EnabledServices))
NewReceipt.Save()
Label = f"{Program.Name} {NewReceipt.Version} ({NewReceipt.Commit[:7]})".replace(" ", " ")
Done = {"Installing": "installed", "Updating": "updated", "Changing options for": "options changed"}[Action]
Log(f"== {Label} {Done}")
return NewReceipt
# --- Remove -------------------------------------------------------------------------
@dataclass
class RemovalItem:
Path: str
Group: str # "Program", "Integration", "Config" or "Runtime"
Note: str = ""
Mode: str = "Remove" # "Remove" or "IfEmpty"
@dataclass
class RemovalPlan:
Items: list[RemovalItem] = field(default_factory=list)
def Group(self, Name: str) -> list[RemovalItem]:
return [Item for Item in self.Items if Item.Group == Name]
def _ExpandLeftover(Item: Leftover, Context: dict) -> list[RemovalItem]:
Pattern = str(Path(Render(Item.Path, Context)).expanduser())
Paths = sorted(glob(Pattern, include_hidden=True)) if any(Char in Pattern for Char in "*?[") else [Pattern]
return [
RemovalItem(Found, Item.Kind, Item.Note, Item.Mode)
for Found in Paths
if Path(Found).exists() or Path(Found).is_symlink()
]
def PlanRemoval(Program: Program) -> RemovalPlan:
Plan = RemovalPlan()
Current = LoadReceipt(Program.Name)
Context = Program.Context(Current.Answers if Current else None)
if Program.ProgramDir.exists():
Plan.Items.append(RemovalItem(str(Program.ProgramDir), "Program", "program files and Python venv"))
for Downloaded in sorted(DownloadsDir.glob(f"{Program.Name}-*")):
Plan.Items.append(RemovalItem(str(Downloaded), "Program", "downloaded source"))
if Current is not None:
for Unit in Current.EnabledServices:
Plan.Items.append(RemovalItem(Unit, "Integration", "systemd user service, stopped and disabled"))
for Placed in Current.Files:
if Placed.Kind == "Template":
Note = "put back to your previous version" if Placed.Displaced else ""
if Placed.Root:
Note = ", ".join(Part for Part in ("needs your password", Note) if Part)
Plan.Items.append(RemovalItem(Placed.Path, "Integration", Note))
elif Path(Placed.Path).exists():
Plan.Items.append(RemovalItem(Placed.Path, "Config"))
for Record in Current.IniEdits:
Note = f"restore {Record.Key}={Record.Previous}" if Record.HadPrevious else f"remove {Record.Key}"
Plan.Items.append(RemovalItem(Record.File, "Integration", Note))
if Program.ConfigDir.exists():
Plan.Items.append(RemovalItem(str(Program.ConfigDir), "Config", "the whole config folder"))
MergeBackups = BackupsDir / Program.Name / "Configs"
if MergeBackups.exists():
Plan.Items.append(RemovalItem(str(MergeBackups), "Config", "config backups made during updates"))
Leftovers = list(Program.Leftovers)
if Program.Hooks is not None and hasattr(Program.Hooks, "Leftovers"):
Leftovers += Program.Hooks.Leftovers(Context)
for Item in Leftovers:
Plan.Items.extend(_ExpandLeftover(Item, Context))
# The same path can come from both the receipt and a Leftover.
Seen: set[str] = set()
Unique = []
for Item in Plan.Items:
if Item.Path in Seen:
continue
Seen.add(Item.Path)
Unique.append(Item)
Plan.Items = Unique
return Plan
def Remove(Program: Program, RemoveConfigs: bool, RemoveRuntime: bool, Log: LogFn) -> None:
Log(f"== Removing {Program.Name}")
Plan = PlanRemoval(Program)
Current = LoadReceipt(Program.Name)
if Current is not None:
for Unit in Current.EnabledServices:
_Systemctl(Log, "disable", "--now", Unit)
for Placed in Current.Files:
if Placed.Kind == "Template":
_RemovePlacedTemplate(Placed, Log)
for Record in Current.IniEdits:
_RevertIni(Record, Log)
if Current.EnabledServices or any("/systemd/" in Placed.Path for Placed in Current.Files):
_Systemctl(Log, "daemon-reload")
Groups = {"Program"}
if RemoveConfigs:
Groups.add("Config")
if RemoveRuntime:
Groups.add("Runtime")
# Files first, then "IfEmpty" folders deepest-first, so a folder that only
# held things removed above is empty by the time it's checked.
Chosen = [Item for Item in Plan.Items if Item.Group in Groups]
for Item in [Item for Item in Chosen if Item.Mode == "Remove"]:
Target = Path(Item.Path)
if Target.exists() or Target.is_symlink():
_RemovePath(Target)
Log(f"Removed {Target}")
for Item in sorted((Item for Item in Chosen if Item.Mode == "IfEmpty"), key=lambda Item: -len(Path(Item.Path).parts)):
Target = Path(Item.Path)
if Target.is_dir() and not any(Target.iterdir()):
Target.rmdir()
Log(f"Removed empty {Target}")
elif Target.exists():
Log(f"Kept {Target} (not empty)")
# SaraSuite's own bookkeeping. The Pristine configs are kept along with the
# user's configs, so a later reinstall can still merge against them.
if RemoveConfigs:
_RemovePath(PristineDir / Program.Name)
_RemovePath(BackupsDir / Program.Name / "Displaced")
ProgramBackups = BackupsDir / Program.Name
if ProgramBackups.is_dir() and not any(ProgramBackups.iterdir()):
ProgramBackups.rmdir()
if Current is not None:
Current.FilePath.unlink(missing_ok=True)
_AfterIntegrationChanges(Program, Log)
Log(f"== {Program.Name} removed" + ("" if RemoveConfigs else " (configs kept)"))