104 lines
3.0 KiB
Python
104 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from app.core.command_runner import CommandRunner
|
|
|
|
|
|
APT_DPKG_OPTIONS = [
|
|
"-o",
|
|
"Dpkg::Use-Pty=0",
|
|
"-o",
|
|
"Dpkg::Options::=--force-confdef",
|
|
"-o",
|
|
"Dpkg::Options::=--force-confold",
|
|
]
|
|
|
|
APT_NONINTERACTIVE_ENV = {
|
|
"DEBIAN_FRONTEND": "noninteractive",
|
|
"DEBCONF_NONINTERACTIVE_SEEN": "true",
|
|
"APT_LISTCHANGES_FRONTEND": "none",
|
|
}
|
|
|
|
|
|
def _parse_deb_control_output(output: str) -> dict[str, str]:
|
|
metadata: dict[str, str] = {}
|
|
|
|
for line in output.splitlines():
|
|
key, separator, value = line.partition(":")
|
|
if not separator:
|
|
continue
|
|
|
|
normalized_key = key.strip().lower()
|
|
if normalized_key in {"package", "version", "architecture"}:
|
|
metadata[normalized_key] = value.strip()
|
|
|
|
return metadata
|
|
|
|
|
|
class DebInstaller:
|
|
def __init__(self, command_runner: CommandRunner) -> None:
|
|
self.command_runner = command_runner
|
|
|
|
def get_deb_metadata(self, file_path: Path) -> dict[str, str]:
|
|
result = self.command_runner.run([
|
|
"dpkg-deb",
|
|
"-f",
|
|
str(file_path),
|
|
"Package",
|
|
"Version",
|
|
"Architecture",
|
|
])
|
|
metadata = _parse_deb_control_output(result.stdout)
|
|
missing_fields = [
|
|
field
|
|
for field in ("package", "version", "architecture")
|
|
if not metadata.get(field)
|
|
]
|
|
|
|
if missing_fields:
|
|
raise ValueError(
|
|
"Downloaded .deb is missing metadata fields: "
|
|
f"{', '.join(missing_fields)}"
|
|
)
|
|
|
|
return metadata
|
|
|
|
def install_deb(self, file_path: Path) -> None:
|
|
self.command_runner.run(
|
|
[
|
|
"apt-get",
|
|
*APT_DPKG_OPTIONS,
|
|
"install",
|
|
"--yes",
|
|
str(file_path),
|
|
],
|
|
env=APT_NONINTERACTIVE_ENV,
|
|
)
|
|
|
|
def remove_package(self, package_name: str, purge: bool = False) -> None:
|
|
action = "purge" if purge else "remove"
|
|
self.command_runner.run(
|
|
["apt-get", *APT_DPKG_OPTIONS, action, "--yes", package_name],
|
|
env=APT_NONINTERACTIVE_ENV,
|
|
)
|
|
|
|
def cleanup_after_remove(self, purge: bool = False) -> None:
|
|
autoremove_command = ["apt-get", *APT_DPKG_OPTIONS, "autoremove", "--yes"]
|
|
if purge:
|
|
autoremove_command.append("--purge")
|
|
self.command_runner.run(autoremove_command, env=APT_NONINTERACTIVE_ENV)
|
|
self.command_runner.run(["apt-get", "clean"], env=APT_NONINTERACTIVE_ENV)
|
|
|
|
def get_package_version(self, package_name: str) -> str | None:
|
|
result = self.command_runner.run(["dpkg-query", "-W", "-f=${Version}", package_name])
|
|
version = result.stdout.strip()
|
|
return version or None
|
|
|
|
def check_package_installed(self, package_name: str) -> bool:
|
|
try:
|
|
self.get_package_version(package_name)
|
|
return True
|
|
except Exception:
|
|
return False
|