aman/packaging/portable/portable_installer.py
Thales Maciel 721248ca26
Decouple non-UI CLI startup from config_ui
Stop aman.py from importing the GTK settings module at module load so version, init, bench, diagnostics, and top-level help can start without pulling in the UI stack.\n\nPromote PyGObject and python-xlib into main project dependencies, switch the documented source install surface to plain uv/pip commands, and teach the portable, deb, and Arch packaging flows to install filtered runtime requirements before the Aman wheel so they still rely on distro-provided GTK/X11 packages.\n\nAdd regression coverage for importing aman with config_ui blocked and for the portable bundle's new requirements payload, then rerun the focused CLI/diagnostics/portable tests plus py_compile.
2026-03-14 13:38:15 -03:00

598 lines
20 KiB
Python
Executable file

#!/usr/bin/env python3
from __future__ import annotations
import argparse
import json
import os
import shutil
import subprocess
import sys
import tempfile
import textwrap
import time
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from pathlib import Path
APP_NAME = "aman"
INSTALL_KIND = "portable"
SERVICE_NAME = "aman"
MANAGED_MARKER = "# managed by aman portable installer"
SUPPORTED_PYTHON_TAGS = ("cp310", "cp311", "cp312")
DEFAULT_ARCHITECTURE = "x86_64"
DEFAULT_SMOKE_CHECK_CODE = textwrap.dedent(
"""
import gi
gi.require_version("Gtk", "3.0")
gi.require_version("AppIndicator3", "0.1")
from gi.repository import AppIndicator3, Gtk
import Xlib
import sounddevice
"""
).strip()
DEFAULT_RUNTIME_DEPENDENCY_HINT = (
"Install the documented GTK, AppIndicator, PyGObject, python-xlib, and "
"PortAudio runtime dependencies for your distro, then rerun install.sh."
)
class PortableInstallError(RuntimeError):
pass
@dataclass
class InstallPaths:
home: Path
share_root: Path
current_link: Path
state_path: Path
bin_dir: Path
shim_path: Path
systemd_dir: Path
service_path: Path
config_dir: Path
cache_dir: Path
@classmethod
def detect(cls) -> "InstallPaths":
home = Path.home()
share_root = home / ".local" / "share" / APP_NAME
return cls(
home=home,
share_root=share_root,
current_link=share_root / "current",
state_path=share_root / "install-state.json",
bin_dir=home / ".local" / "bin",
shim_path=home / ".local" / "bin" / APP_NAME,
systemd_dir=home / ".config" / "systemd" / "user",
service_path=home / ".config" / "systemd" / "user" / f"{SERVICE_NAME}.service",
config_dir=home / ".config" / APP_NAME,
cache_dir=home / ".cache" / APP_NAME,
)
def as_serializable(self) -> dict[str, str]:
return {
"share_root": str(self.share_root),
"current_link": str(self.current_link),
"state_path": str(self.state_path),
"shim_path": str(self.shim_path),
"service_path": str(self.service_path),
"config_dir": str(self.config_dir),
"cache_dir": str(self.cache_dir),
}
@dataclass
class Manifest:
app_name: str
version: str
architecture: str
supported_python_tags: list[str]
wheelhouse_dirs: list[str]
managed_paths: dict[str, str]
smoke_check_code: str
runtime_dependency_hint: str
bundle_format_version: int = 1
@classmethod
def default(cls, version: str) -> "Manifest":
return cls(
app_name=APP_NAME,
version=version,
architecture=DEFAULT_ARCHITECTURE,
supported_python_tags=list(SUPPORTED_PYTHON_TAGS),
wheelhouse_dirs=[
"wheelhouse/common",
"wheelhouse/cp310",
"wheelhouse/cp311",
"wheelhouse/cp312",
],
managed_paths={
"install_root": "~/.local/share/aman",
"current_link": "~/.local/share/aman/current",
"shim": "~/.local/bin/aman",
"service": "~/.config/systemd/user/aman.service",
"state": "~/.local/share/aman/install-state.json",
},
smoke_check_code=DEFAULT_SMOKE_CHECK_CODE,
runtime_dependency_hint=DEFAULT_RUNTIME_DEPENDENCY_HINT,
)
@dataclass
class InstallState:
app_name: str
install_kind: str
version: str
installed_at: str
service_mode: str
architecture: str
supported_python_tags: list[str]
paths: dict[str, str]
def _portable_tag() -> str:
test_override = os.environ.get("AMAN_PORTABLE_TEST_PYTHON_TAG", "").strip()
if test_override:
return test_override
return f"cp{sys.version_info.major}{sys.version_info.minor}"
def _load_manifest(bundle_dir: Path) -> Manifest:
manifest_path = bundle_dir / "manifest.json"
try:
payload = json.loads(manifest_path.read_text(encoding="utf-8"))
except FileNotFoundError as exc:
raise PortableInstallError(f"missing manifest: {manifest_path}") from exc
except json.JSONDecodeError as exc:
raise PortableInstallError(f"invalid manifest JSON: {manifest_path}") from exc
try:
return Manifest(**payload)
except TypeError as exc:
raise PortableInstallError(f"invalid manifest shape: {manifest_path}") from exc
def _load_state(state_path: Path) -> InstallState | None:
if not state_path.exists():
return None
try:
payload = json.loads(state_path.read_text(encoding="utf-8"))
except json.JSONDecodeError as exc:
raise PortableInstallError(f"invalid install state JSON: {state_path}") from exc
try:
return InstallState(**payload)
except TypeError as exc:
raise PortableInstallError(f"invalid install state shape: {state_path}") from exc
def _atomic_write(path: Path, content: str, *, mode: int = 0o644) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with tempfile.NamedTemporaryFile(
"w",
encoding="utf-8",
dir=path.parent,
prefix=f".{path.name}.tmp-",
delete=False,
) as handle:
handle.write(content)
tmp_path = Path(handle.name)
os.chmod(tmp_path, mode)
os.replace(tmp_path, path)
def _atomic_symlink(target: Path, link_path: Path) -> None:
link_path.parent.mkdir(parents=True, exist_ok=True)
tmp_link = link_path.parent / f".{link_path.name}.tmp-{os.getpid()}"
try:
if tmp_link.exists() or tmp_link.is_symlink():
tmp_link.unlink()
os.symlink(str(target), tmp_link)
os.replace(tmp_link, link_path)
finally:
if tmp_link.exists() or tmp_link.is_symlink():
tmp_link.unlink()
def _read_text_if_exists(path: Path) -> str | None:
if not path.exists():
return None
return path.read_text(encoding="utf-8")
def _current_target(current_link: Path) -> Path | None:
if current_link.is_symlink():
target = os.readlink(current_link)
target_path = Path(target)
if not target_path.is_absolute():
target_path = current_link.parent / target_path
return target_path
if current_link.exists():
return current_link
return None
def _is_managed_text(content: str | None) -> bool:
return bool(content and MANAGED_MARKER in content)
def _run(
args: list[str],
*,
check: bool = True,
capture_output: bool = False,
) -> subprocess.CompletedProcess[str]:
try:
return subprocess.run(
args,
check=check,
text=True,
capture_output=capture_output,
)
except subprocess.CalledProcessError as exc:
details = exc.stderr.strip() or exc.stdout.strip() or str(exc)
raise PortableInstallError(details) from exc
def _run_systemctl(args: list[str], *, check: bool = True) -> subprocess.CompletedProcess[str]:
return _run(["systemctl", "--user", *args], check=check, capture_output=True)
def _supported_tag_or_raise(manifest: Manifest) -> str:
if sys.implementation.name != "cpython":
raise PortableInstallError("portable installer requires CPython 3.10, 3.11, or 3.12")
tag = _portable_tag()
if tag not in manifest.supported_python_tags:
version = f"{sys.version_info.major}.{sys.version_info.minor}"
raise PortableInstallError(
f"unsupported python3 version {version}; supported versions are CPython 3.10, 3.11, and 3.12"
)
return tag
def _check_preflight(manifest: Manifest, paths: InstallPaths) -> InstallState | None:
_supported_tag_or_raise(manifest)
if shutil.which("systemctl") is None:
raise PortableInstallError("systemctl is required for the supported user service lifecycle")
try:
import venv as _venv # noqa: F401
except Exception as exc: # pragma: no cover - import failure is environment dependent
raise PortableInstallError("python3 venv support is required for the portable installer") from exc
state = _load_state(paths.state_path)
if state is not None:
if state.app_name != APP_NAME or state.install_kind != INSTALL_KIND:
raise PortableInstallError(f"unexpected install state in {paths.state_path}")
shim_text = _read_text_if_exists(paths.shim_path)
if shim_text is not None and (state is None or not _is_managed_text(shim_text)):
raise PortableInstallError(
f"refusing to overwrite unmanaged shim at {paths.shim_path}; remove it first"
)
service_text = _read_text_if_exists(paths.service_path)
if service_text is not None and (state is None or not _is_managed_text(service_text)):
raise PortableInstallError(
f"refusing to overwrite unmanaged service file at {paths.service_path}; remove it first"
)
detected_aman = shutil.which(APP_NAME)
if detected_aman:
expected_paths = {str(paths.shim_path)}
current_target = _current_target(paths.current_link)
if current_target is not None:
expected_paths.add(str(current_target / "venv" / "bin" / APP_NAME))
if detected_aman not in expected_paths:
raise PortableInstallError(
"detected another Aman install in PATH at "
f"{detected_aman}; remove that install before using the portable bundle"
)
return state
def _require_bundle_file(path: Path, description: str) -> Path:
if not path.exists():
raise PortableInstallError(f"missing {description}: {path}")
return path
def _aman_wheel(common_wheelhouse: Path) -> Path:
wheels = sorted(common_wheelhouse.glob(f"{APP_NAME}-*.whl"))
if not wheels:
raise PortableInstallError(f"no Aman wheel found in {common_wheelhouse}")
return wheels[-1]
def _render_wrapper(paths: InstallPaths) -> str:
exec_path = paths.current_link / "venv" / "bin" / APP_NAME
return textwrap.dedent(
f"""\
#!/usr/bin/env bash
set -euo pipefail
{MANAGED_MARKER}
exec "{exec_path}" "$@"
"""
)
def _render_service(template_text: str, paths: InstallPaths) -> str:
exec_start = (
f"{paths.current_link / 'venv' / 'bin' / APP_NAME} "
f"run --config {paths.home / '.config' / APP_NAME / 'config.json'}"
)
return template_text.replace("__EXEC_START__", exec_start)
def _write_state(paths: InstallPaths, manifest: Manifest, version_dir: Path) -> None:
state = InstallState(
app_name=APP_NAME,
install_kind=INSTALL_KIND,
version=manifest.version,
installed_at=datetime.now(timezone.utc).isoformat(),
service_mode="systemd-user",
architecture=manifest.architecture,
supported_python_tags=list(manifest.supported_python_tags),
paths={
**paths.as_serializable(),
"version_dir": str(version_dir),
},
)
_atomic_write(paths.state_path, json.dumps(asdict(state), indent=2, sort_keys=True) + "\n")
def _copy_bundle_support_files(bundle_dir: Path, stage_dir: Path) -> None:
for name in ("manifest.json", "install.sh", "uninstall.sh", "portable_installer.py"):
src = _require_bundle_file(bundle_dir / name, name)
dst = stage_dir / name
shutil.copy2(src, dst)
if dst.suffix in {".sh", ".py"}:
os.chmod(dst, 0o755)
src_service_dir = _require_bundle_file(bundle_dir / "systemd", "systemd directory")
dst_service_dir = stage_dir / "systemd"
if dst_service_dir.exists():
shutil.rmtree(dst_service_dir)
shutil.copytree(src_service_dir, dst_service_dir)
def _run_pip_install(bundle_dir: Path, stage_dir: Path, python_tag: str) -> None:
common_dir = _require_bundle_file(bundle_dir / "wheelhouse" / "common", "common wheelhouse")
version_dir = _require_bundle_file(bundle_dir / "wheelhouse" / python_tag, f"{python_tag} wheelhouse")
requirements_path = _require_bundle_file(
bundle_dir / "requirements" / f"{python_tag}.txt",
f"{python_tag} runtime requirements",
)
aman_wheel = _aman_wheel(common_dir)
venv_dir = stage_dir / "venv"
_run([sys.executable, "-m", "venv", "--system-site-packages", str(venv_dir)])
_run(
[
str(venv_dir / "bin" / "python"),
"-m",
"pip",
"install",
"--no-index",
"--find-links",
str(common_dir),
"--find-links",
str(version_dir),
"--requirement",
str(requirements_path),
]
)
_run(
[
str(venv_dir / "bin" / "python"),
"-m",
"pip",
"install",
"--no-index",
"--find-links",
str(common_dir),
"--find-links",
str(version_dir),
"--no-deps",
str(aman_wheel),
]
)
def _run_smoke_check(stage_dir: Path, manifest: Manifest) -> None:
venv_python = stage_dir / "venv" / "bin" / "python"
try:
_run([str(venv_python), "-c", manifest.smoke_check_code], capture_output=True)
except PortableInstallError as exc:
raise PortableInstallError(
f"runtime dependency smoke check failed: {exc}\n{manifest.runtime_dependency_hint}"
) from exc
def _remove_path(path: Path) -> None:
if path.is_symlink() or path.is_file():
path.unlink(missing_ok=True)
return
if path.is_dir():
shutil.rmtree(path, ignore_errors=True)
def _rollback_install(
*,
paths: InstallPaths,
manifest: Manifest,
old_state_text: str | None,
old_service_text: str | None,
old_shim_text: str | None,
old_current_target: Path | None,
new_version_dir: Path,
backup_dir: Path | None,
) -> None:
_remove_path(new_version_dir)
if backup_dir is not None and backup_dir.exists():
os.replace(backup_dir, new_version_dir)
if old_current_target is not None:
_atomic_symlink(old_current_target, paths.current_link)
else:
_remove_path(paths.current_link)
if old_shim_text is not None:
_atomic_write(paths.shim_path, old_shim_text, mode=0o755)
else:
_remove_path(paths.shim_path)
if old_service_text is not None:
_atomic_write(paths.service_path, old_service_text)
else:
_remove_path(paths.service_path)
if old_state_text is not None:
_atomic_write(paths.state_path, old_state_text)
else:
_remove_path(paths.state_path)
_run_systemctl(["daemon-reload"], check=False)
if old_current_target is not None and old_service_text is not None:
_run_systemctl(["enable", "--now", SERVICE_NAME], check=False)
def _prune_versions(paths: InstallPaths, keep_version: str) -> None:
for entry in paths.share_root.iterdir():
if entry.name in {"current", "install-state.json"}:
continue
if entry.is_dir() and entry.name != keep_version:
shutil.rmtree(entry, ignore_errors=True)
def install_bundle(bundle_dir: Path) -> int:
manifest = _load_manifest(bundle_dir)
paths = InstallPaths.detect()
previous_state = _check_preflight(manifest, paths)
python_tag = _supported_tag_or_raise(manifest)
paths.share_root.mkdir(parents=True, exist_ok=True)
stage_dir = paths.share_root / f".staging-{manifest.version}-{os.getpid()}"
version_dir = paths.share_root / manifest.version
backup_dir: Path | None = None
old_state_text = _read_text_if_exists(paths.state_path)
old_service_text = _read_text_if_exists(paths.service_path)
old_shim_text = _read_text_if_exists(paths.shim_path)
old_current_target = _current_target(paths.current_link)
service_template_path = _require_bundle_file(
bundle_dir / "systemd" / f"{SERVICE_NAME}.service.in",
"service template",
)
service_template = service_template_path.read_text(encoding="utf-8")
cutover_done = False
if previous_state is not None:
_run_systemctl(["stop", SERVICE_NAME], check=False)
_remove_path(stage_dir)
stage_dir.mkdir(parents=True, exist_ok=True)
try:
_run_pip_install(bundle_dir, stage_dir, python_tag)
_copy_bundle_support_files(bundle_dir, stage_dir)
_run_smoke_check(stage_dir, manifest)
if version_dir.exists():
backup_dir = paths.share_root / f".rollback-{manifest.version}-{int(time.time())}"
_remove_path(backup_dir)
os.replace(version_dir, backup_dir)
os.replace(stage_dir, version_dir)
_atomic_symlink(version_dir, paths.current_link)
_atomic_write(paths.shim_path, _render_wrapper(paths), mode=0o755)
_atomic_write(paths.service_path, _render_service(service_template, paths))
_write_state(paths, manifest, version_dir)
cutover_done = True
_run_systemctl(["daemon-reload"])
_run_systemctl(["enable", "--now", SERVICE_NAME])
except Exception:
_remove_path(stage_dir)
if cutover_done or backup_dir is not None:
_rollback_install(
paths=paths,
manifest=manifest,
old_state_text=old_state_text,
old_service_text=old_service_text,
old_shim_text=old_shim_text,
old_current_target=old_current_target,
new_version_dir=version_dir,
backup_dir=backup_dir,
)
else:
_remove_path(stage_dir)
raise
if backup_dir is not None:
_remove_path(backup_dir)
_prune_versions(paths, manifest.version)
print(f"installed {APP_NAME} {manifest.version} in {version_dir}")
return 0
def uninstall_bundle(bundle_dir: Path, *, purge: bool) -> int:
_ = bundle_dir
paths = InstallPaths.detect()
state = _load_state(paths.state_path)
if state is None:
raise PortableInstallError(f"no portable install state found at {paths.state_path}")
if state.app_name != APP_NAME or state.install_kind != INSTALL_KIND:
raise PortableInstallError(f"unexpected install state in {paths.state_path}")
shim_text = _read_text_if_exists(paths.shim_path)
if shim_text is not None and not _is_managed_text(shim_text):
raise PortableInstallError(f"refusing to remove unmanaged shim at {paths.shim_path}")
service_text = _read_text_if_exists(paths.service_path)
if service_text is not None and not _is_managed_text(service_text):
raise PortableInstallError(f"refusing to remove unmanaged service at {paths.service_path}")
_run_systemctl(["disable", "--now", SERVICE_NAME], check=False)
_remove_path(paths.service_path)
_run_systemctl(["daemon-reload"], check=False)
_remove_path(paths.shim_path)
_remove_path(paths.share_root)
if purge:
_remove_path(paths.config_dir)
_remove_path(paths.cache_dir)
print(f"uninstalled {APP_NAME} portable bundle")
return 0
def write_manifest(version: str, output_path: Path) -> int:
manifest = Manifest.default(version)
_atomic_write(output_path, json.dumps(asdict(manifest), indent=2, sort_keys=True) + "\n")
return 0
def _parse_args(argv: list[str]) -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Aman portable bundle helper")
subparsers = parser.add_subparsers(dest="command", required=True)
install_parser = subparsers.add_parser("install", help="Install or upgrade the portable bundle")
install_parser.add_argument("--bundle-dir", default=str(Path.cwd()))
uninstall_parser = subparsers.add_parser("uninstall", help="Uninstall the portable bundle")
uninstall_parser.add_argument("--bundle-dir", default=str(Path.cwd()))
uninstall_parser.add_argument("--purge", action="store_true", help="Remove config and cache too")
manifest_parser = subparsers.add_parser("write-manifest", help="Write the portable bundle manifest")
manifest_parser.add_argument("--version", required=True)
manifest_parser.add_argument("--output", required=True)
return parser.parse_args(argv)
def main(argv: list[str] | None = None) -> int:
args = _parse_args(argv or sys.argv[1:])
try:
if args.command == "install":
return install_bundle(Path(args.bundle_dir).resolve())
if args.command == "uninstall":
return uninstall_bundle(Path(args.bundle_dir).resolve(), purge=args.purge)
if args.command == "write-manifest":
return write_manifest(args.version, Path(args.output).resolve())
except PortableInstallError as exc:
print(str(exc), file=sys.stderr)
return 1
return 1
if __name__ == "__main__":
raise SystemExit(main())