diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..3d78e47 --- /dev/null +++ b/.gitignore @@ -0,0 +1,75 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg + +# Virtual environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ + +# Jupyter +.ipynb_checkpoints + +# pyenv +.python-version + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre +.pyre/ + +# UV +uv.lock + +# Backup/config artifacts +*.bak +inventory.toml.last +inventory.toml.rollback diff --git a/README.md b/README.md index d757aea..a3459b1 100644 --- a/README.md +++ b/README.md @@ -1,62 +1,49 @@ -# curator - A Home-Manager Like Tool for Fedora +# curator - A Home-Manager Like Tool for Universal Blue builds -A lightweight Python CLI (managed with `uv`) for Fedora that provides COPR repository management, dotfile management and package installation functionality through a centralized TOML configuration. +A lightweight CLI that provides repository and dotfile management and package installation functionality through a centralized TOML configuration. Ported and expanded from [forge](https://github.com/ijadux2/forge) and inspired by [home-manager](https://github.com/nix-community/home-manager/). ## Overview curator is a Python CLI that helps you manage your Fedora system configuration by: - Managing COPR repositories (enable/disable automatically) -- Installing and managing system packages via dnf +- Installing and managing system packages via `dnf` or `rpm-ostree` +- Installing and managing userspace packages via `brew`, `flatpak`, or `nix` - Managing dotfiles through symlinks to actual files - Centralized configuration via a single `inventory.toml` file -- User-level configuration similar to home-manager/nixos +- Declarative user-level configuration similar to `home-manager`/`nixos` - Automatic backup of existing files before replacement ## Installation -1. Clone or download this repository. -2. Install dependencies with `uv` (none beyond the standard library, but this sets up the venv): - ```bash - uv sync - ``` -3. Run with `uv`: - ```bash - uv run curator --help - ``` -4. Or use the local shim directly: - ```bash - ./curator --help - ``` -5. Optionally install globally via `uv`: - ```bash - uv tool install . - ``` +Install the CLI directly from GitHub with [uv](https://github.com/astral-sh/uv): + +```bash +uv tool install --from git+https://codeberg.org/randogoth/curator/ curator +``` ## Quick Start ```bash # Initialize the configuration structure -uv run curator init +curator init # Edit the configuration file to add packages and dotfiles nano ~/.config/curator/inventory.toml # Apply configuration -uv run curator switch +curator switch # Check status -uv run curator status +curator status ``` -You can swap `uv run curator ...` for `./curator ...` if you prefer the local shim. - ## Commands ### `init` Initialize the configuration structure and create `inventory.toml`. ```bash -uv run curator init +curator init # or ./curator init ``` @@ -68,10 +55,10 @@ Creates: Apply the current configuration (enable COPR, install packages, deploy dotfiles). Use `--rollback` to restore the previous `inventory.toml` snapshot before applying. ```bash -uv run curator switch +curator switch # or ./curator switch # rollback to the previous inventory.toml and apply it -uv run curator switch --rollback +curator switch --rollback ``` This command: @@ -86,7 +73,7 @@ This command: Show current configuration status and information. ```bash -uv run curator status +curator status # or ./curator status ``` @@ -99,10 +86,37 @@ Displays: Show help message with all available commands. ```bash -uv run curator help +curator help # or ./curator help ``` +### `from` +Import currently installed packages/repos for a manager into `inventory.toml` so curator can manage them. + +```bash +curator from dnf +curator from brew +curator from flatpak +curator from rpm-ostree +curator from nix +curator from copr +``` + +### `add` / `remove` +Add or remove entries directly in `inventory.toml` using `section:value` pairs. Supports `dnf`, `brew`, `flatpak`, `rpm-ostree`, `nix`, and `copr`. + +```bash +curator add dnf:uv nix:micro +curator remove dnf:curl flatpak:org.mozilla.firefox +``` + +### `reset` +Remove the rollback snapshot (`inventory.toml.rollback`). + +```bash +curator reset +``` + ## Configuration ### inventory.toml @@ -197,6 +211,12 @@ List of packages to install via Homebrew. **One package per line**—presence me #### `[flatpak]` List of Flatpak refs to install. **One ref per line**—presence means install. +#### `[rpm-ostree]` +List of rpm-ostree layered packages. **One package per line**—presence means install. + +#### `[nix]` +List of nix packages. **One package per line**—presence means install. You can specify plain names (e.g., `neovim`) or `nixpkgs#name`; curator will prefix `nixpkgs#` for installs and use the base name for removals. + #### `[dotfiles]` Dotfile mappings using symlinks: - **Key**: Target path where symlink should be created (relative to home directory) @@ -222,209 +242,6 @@ Additional configuration options: └── vimrc.20231121_143022.bak ``` -## How It Works - -### COPR Repository Management -COPR repositories are managed through the `[copr]` section in `inventory.toml`: -- **Add COPR repo**: Add the repository name on its own line -- **Remove COPR repo**: Remove the entry -- **Automatic cleanup**: curator automatically disables COPR repos that are removed from configuration -- **No flags needed**: Just presence/absence of the repository name matters -- curator stores the previous configuration at `~/.config/curator/inventory.toml.prev` and compares it to the current file during `switch`; newly added repos are enabled, removed repos are disabled. - -### Package Management -Packages are managed through the `[dnf]` section in `inventory.toml`: -- **Add package**: Add the package name on its own line -- **Remove package**: Remove the entry -- **No flags needed**: Just presence/absence of the package name matters -- The previous configuration snapshot (`inventory.toml.prev`) is used to detect additions/removals on each `switch`; added packages are installed and removed packages are uninstalled. - -### Brew Management -Homebrew packages are managed through the `[brew]` section: -- **Add package**: Add the package name on its own line -- **Remove package**: Remove the entry -- Additions/removals are detected against `inventory.toml.prev` on `switch`; removed packages are uninstalled. - -### Flatpak Management -Flatpaks are managed through the `[flatpak]` section: -- **Add ref**: Add the ref on its own line -- **Remove ref**: Remove the entry -- Additions/removals are detected against `inventory.toml.prev` on `switch`; removed refs are uninstalled. - -### rpm-ostree Management -rpm-ostree packages are managed through the `[rpm-ostree]` section: -- **Add package**: Add the package name on its own line -- **Remove package**: Remove the entry -- Additions/removals are detected against `inventory.toml.prev` on `switch`; removed packages are uninstalled. - -### Nix Management -Nix packages are managed through the `[nix]` section (if `nix` is available on the system): -- **Add package**: Add the package name (e.g., `nixpkgs#git` or just `git`) on its own line -- **Remove package**: Remove the entry -- Additions/removals are detected against `inventory.toml.prev` on `switch`; installs/removals use `nix profile` and will prefix `nixpkgs#` if missing. - -### Dotfile Management -curator uses symlinks to manage dotfiles: -1. Your actual dotfiles are stored in `~/.config/curator/dotfiles/` -2. Symlinks are created from your home directory to these files using relative paths -3. This allows you to version control your dotfiles in one place -4. Changes to the source files are immediately reflected in your home directory -5. Source paths are automatically prefixed with "dotfiles/" for convenience - -### Backup System -Before creating symlinks, curator: -1. Checks if the target file exists and is not a symlink -2. Creates a timestamped backup in the backup directory -3. Removes the original file -4. Creates the symlink to your managed dotfile - -`inventory.toml` is also backed up before the last switch timestamp is updated. - -The previously applied configuration is stored separately as `~/.config/curator/inventory.toml.prev` to compute diffs for COPR and package changes. - -## Examples - -### Basic Setup -```bash -# Initialize curator -uv run curator init - -# Edit inventory.toml to add COPR repos and packages -nano ~/.config/curator/inventory.toml - -# Add to the sections: -# [copr] -# copr.fedorainfracloud.org/username/cool-repo -# [dnf] -# git -# vim -# curl -# [brew] -# wget -# [flatpak] -# org.mozilla.firefox -# [rpm-ostree] -# podman -# [nix] -# nixpkgs#git (or just "git") - -# Create your dotfiles directory and add files -mkdir -p ~/.config/curator/dotfiles -echo "export EDITOR=vim" > ~/.config/curator/dotfiles/.bashrc - -# Add to [dotfiles] section: -".bashrc" = ".bashrc" - -# Apply configuration -uv run curator switch -``` - -### Managing Application Configurations -```bash -# Add alacritty configuration -mkdir -p ~/.config/curator/dotfiles -cp ~/.config/alacritty/alacritty.yml ~/.config/curator/dotfiles/ - -# Edit inventory.toml -nano ~/.config/curator/inventory.toml - -# Add to [dotfiles] section: -".config/alacritty/alacritty.yml" = "alacritty.yml" - -# Apply changes -uv run curator switch -``` - -### COPR Repository Management Examples -```toml -[copr] -# Development tools COPR -copr.fedorainfracloud.org/development/tools -copr.fedorainfracloud.org/user/neovim-nightly -# To remove a COPR repo, just delete the entry -# curator will automatically disable it during the next switch -``` - -### Package Management Examples -```toml -[dnf] -# Development tools -git -vim -nodejs -npm - -# System utilities -curl -wget -tree -htop -# To remove a package, just delete the entry - -[brew] -# Utilities and tools -wget -coreutils - -[flatpak] -org.mozilla.firefox -com.spotify.Client - -[rpm-ostree] -podman -htop -``` - -### Version Control Your Configuration -```bash -# Initialize git repository in curator directory -cd ~/.config/curator -git init -git add . -git commit -m "Initial configuration" - -# Now you can version control your entire system configuration -git add inventory.toml dotfiles/ -git commit -m "Updated vim configuration" -``` - ## Environment Variables -- `CURATOR_DIR`: Override the default configuration directory (default: `~/.config/curator`) - -## Dependencies - -- Python 3.11+ -- `uv` for environment and script management -- `dnf` - Fedora package manager -- `dnf-plugins-core` - For COPR repository management - -## Migration from Previous Version - -If you were using the old version of curator with `git = true` or bare keys under `[packages]`/`[copr]`: - -1. Your existing configuration will not be automatically migrated, but the CLI will still read the legacy format. -2. Run `uv run curator init` (or `./curator init`) to create the new `inventory.toml` structure. -3. Convert to section-per-line format: - ```toml - # Old formats - [dnf] - git = true - vim = true - # or - [dnf] - git - vim - - # New format - [dnf] - git - vim - [copr] - copr.fedorainfracloud.org/username/repository - ``` -4. Move your existing dotfiles from the old directory to `~/.config/curator/dotfiles/` - -## License - -This project is open source. Feel free to contribute or report issues. +- `CURATOR_DIR`: Override the default configuration directory (default: `~/.config/curator`) \ No newline at end of file diff --git a/src/curator/cli.py b/src/curator/cli.py index c495e1c..5f46e85 100644 --- a/src/curator/cli.py +++ b/src/curator/cli.py @@ -3,6 +3,7 @@ from __future__ import annotations import argparse import datetime as dt +import json import os import re import shutil @@ -10,7 +11,7 @@ import subprocess import sys from dataclasses import dataclass from pathlib import Path -from typing import Any, Optional +from typing import Any, Optional, Callable BLUE = "\033[0;34m" GREEN = "\033[0;32m" @@ -117,6 +118,27 @@ def parse_value(value: str) -> Any: return strip_quotes(value) +def normalize_nix_ref(ref: str) -> str: + ref = ref.strip() + if not ref: + return ref + return ref if "#" in ref else f"nixpkgs#{ref}" + + +def nix_base_name(val: str) -> str: + val = val.strip() + if not val: + return "" + base = val.split("#", 1)[1] if "#" in val else val + base = re.sub(r"-\d[^\s]*$", "", base) + return base + + +def nix_install_ref(val: str) -> str: + base = nix_base_name(val) + return f"nixpkgs#{base}" if base else "" + + def parse_config(config_path: Path) -> CuratorConfig: copr: list[str] = [] dnf_packages: list[str] = [] @@ -237,15 +259,16 @@ def diff_items(current: list[str], previous: list[str]) -> tuple[list[str], list return added, removed -def get_paths() -> tuple[Path, Path, Path]: +def get_paths() -> tuple[Path, Path, Path, Path]: curator_dir = Path(os.environ.get("CURATOR_DIR", Path.home() / ".config" / "curator")).expanduser() curator_toml = curator_dir / "inventory.toml" - previous_toml = curator_dir / "inventory.toml.prev" - return curator_dir, curator_toml, previous_toml + last_toml = curator_dir / "inventory.toml.last" + rollback_toml = curator_dir / "inventory.toml.rollback" + return curator_dir, curator_toml, last_toml, rollback_toml def init_command() -> None: - curator_dir, curator_toml, _ = get_paths() + curator_dir, curator_toml, _, _ = get_paths() log_info("Initializing curator...") curator_dir.mkdir(parents=True, exist_ok=True) @@ -269,6 +292,60 @@ def run_command(command: list[str]) -> bool: return result.returncode == 0 +def run_capture(command: list[str]) -> tuple[bool, str]: + try: + result = subprocess.run(command, check=False, capture_output=True, text=True) + except FileNotFoundError: + log_error(f"Command not found: {' '.join(command)}") + return False, "" + if result.returncode != 0: + return False, result.stdout + return True, result.stdout + + +def install_entries( + entries: list[str], + installed: set[str], + label: str, + command_builder: Callable[[str], list[str]], +) -> None: + log_info(f"Installing {label}...") + if not entries: + log_info(f"No {label} to install") + return + for entry in entries: + if entry in installed: + log_info(f"Skipping already installed {entry}") + continue + log_info(f"Installing {entry}") + if run_command(command_builder(entry)): + log_success(f"Installed {entry}") + else: + log_error(f"Failed to install {entry}") + + +def remove_entries( + entries: list[str], + label: str, + command_builder: Callable[[str], list[str]], + installed: set[str] | None = None, +) -> None: + installed_set = installed if installed is not None else None + log_info(f"Removing {label}...") + if not entries: + log_info(f"No {label} to remove") + return + for entry in entries: + if installed_set is not None and entry not in installed_set: + log_info(f"Skipping removal (not installed according to inventory): {entry}") + continue + log_info(f"Removing {entry}") + if run_command(command_builder(entry)): + log_success(f"Removed {entry}") + else: + log_error(f"Failed to remove {entry}") + + def enable_copr_repos(repos: list[str]) -> None: log_info("Enabling COPR repositories...") if not repos: @@ -283,7 +360,7 @@ def enable_copr_repos(repos: list[str]) -> None: log_error(f"Failed to enable COPR: {copr_repo}") -def list_enabled_copr() -> set[str]: +def gather_copr_repos() -> list[str]: try: result = subprocess.run( ["sudo", "dnf", "copr", "list", "--enabled"], @@ -293,31 +370,26 @@ def list_enabled_copr() -> set[str]: ) except FileNotFoundError: log_error("dnf not found while listing enabled COPR repositories") - return set() + return [] if result.returncode != 0: log_warning("Unable to list enabled COPR repositories") - return set() + return [] enabled: set[str] = set() for line in result.stdout.splitlines(): if "copr.fedorainfracloud.org" in line: repo = line.split()[0] enabled.add(repo) - return enabled + return sorted(enabled) -def disable_copr_repos(configured: set[str], explicit_removed: set[str] | None = None) -> None: - log_info("Checking for COPR repositories to disable...") - enabled = list_enabled_copr() - to_disable: set[str] = {repo for repo in enabled if repo not in configured} - if explicit_removed: - to_disable.update(explicit_removed) - +def disable_copr_repos(to_disable: set[str]) -> None: if not to_disable: log_info("No COPR repositories to disable") return + log_info("Disabling COPR repositories...") for repo in sorted(to_disable): log_info(f"Disabling COPR repository: {repo}") if run_command(["sudo", "dnf", "copr", "disable", "-y", repo]): @@ -326,160 +398,215 @@ def disable_copr_repos(configured: set[str], explicit_removed: set[str] | None = log_error(f"Failed to disable COPR: {repo}") -def install_packages(packages: list[str]) -> None: - log_info("Installing packages...") - if not packages: - log_info("No packages to install") - return - - for package in packages: - log_info(f"Installing package: {package}") - if run_command(["sudo", "dnf", "install", "-y", package]): - log_success(f"Installed {package}") - else: - log_error(f"Failed to install {package}") +def install_packages(packages: list[str], installed: set[str] | None = None) -> None: + install_entries(packages, installed or set(), "dnf packages", lambda p: ["sudo", "dnf", "install", "-y", p]) -def install_brew_packages(packages: list[str]) -> None: - log_info("Installing brew packages...") - if not packages: - log_info("No brew packages to install") - return - - for package in packages: - log_info(f"Installing brew package: {package}") - if run_command(["brew", "install", package]): - log_success(f"Installed {package}") - else: - log_error(f"Failed to install {package}") +def install_brew_packages(packages: list[str], installed: set[str] | None = None) -> None: + install_entries(packages, installed or set(), "brew packages", lambda p: ["brew", "install", p]) -def install_flatpaks(refs: list[str]) -> None: - log_info("Installing flatpak refs...") - if not refs: - log_info("No flatpaks to install") - return - - for ref in refs: - log_info(f"Installing flatpak: {ref}") - if run_command(["flatpak", "install", "-y", ref]): - log_success(f"Installed {ref}") - else: - log_error(f"Failed to install {ref}") +def install_flatpaks(refs: list[str], installed: set[str] | None = None) -> None: + install_entries(refs, installed or set(), "flatpaks", lambda r: ["flatpak", "install", "-y", r]) -def install_rpm_ostree_packages(packages: list[str]) -> None: - log_info("Installing rpm-ostree packages...") - if not packages: - log_info("No rpm-ostree packages to install") - return - - for package in packages: - log_info(f"Installing rpm-ostree package: {package}") - if run_command(["rpm-ostree", "install", "-y", package]): - log_success(f"Installed {package}") - else: - log_error(f"Failed to install {package}") +def install_rpm_ostree_packages(packages: list[str], installed: set[str] | None = None) -> None: + install_entries( + packages, + installed or set(), + "rpm-ostree packages", + lambda p: ["rpm-ostree", "install", "-y", p], + ) def nix_available() -> bool: return shutil.which("nix") is not None -def normalize_nix_ref(ref: str) -> str: - return ref if "#" in ref else f"nixpkgs#{ref}" - - -def install_nix_packages(packages: list[str]) -> None: - log_info("Installing nix packages...") +def install_nix_packages(packages: list[str], installed: set[str] | None = None) -> None: if not packages: log_info("No nix packages to install") return if not nix_available(): log_warning("nix not found on PATH; skipping nix package installs") return - - for package in packages: - ref = normalize_nix_ref(package) - log_info(f"Installing nix package: {ref}") - if run_command(["nix", "profile", "install", ref]): - log_success(f"Installed {ref}") - else: - log_error(f"Failed to install {ref}") + normalized = [nix_install_ref(p) for p in packages if nix_install_ref(p)] + install_entries(normalized, installed or set(), "nix packages", lambda r: ["nix", "profile", "install", r]) -def remove_packages(packages: list[str]) -> None: - log_info("Removing packages...") - if not packages: - log_info("No packages to remove") - return - - for package in packages: - log_info(f"Removing package: {package}") - if run_command(["sudo", "dnf", "remove", "-y", package]): - log_success(f"Removed {package}") - else: - log_error(f"Failed to remove {package}") +def remove_packages(packages: list[str], installed: set[str] | None = None) -> None: + remove_entries(packages, "dnf packages", lambda p: ["sudo", "dnf", "remove", "-y", p], installed) -def remove_brew_packages(packages: list[str]) -> None: - log_info("Removing brew packages...") - if not packages: - log_info("No brew packages to remove") - return - - for package in packages: - log_info(f"Removing brew package: {package}") - if run_command(["brew", "uninstall", package]): - log_success(f"Removed {package}") - else: - log_error(f"Failed to remove {package}") +def remove_brew_packages(packages: list[str], installed: set[str] | None = None) -> None: + remove_entries(packages, "brew packages", lambda p: ["brew", "uninstall", p], installed) -def remove_flatpaks(refs: list[str]) -> None: - log_info("Removing flatpaks...") - if not refs: - log_info("No flatpaks to remove") - return - - for ref in refs: - log_info(f"Removing flatpak: {ref}") - if run_command(["flatpak", "uninstall", "-y", ref]): - log_success(f"Removed {ref}") - else: - log_error(f"Failed to remove {ref}") +def remove_flatpaks(refs: list[str], installed: set[str] | None = None) -> None: + remove_entries(refs, "flatpaks", lambda r: ["flatpak", "uninstall", "-y", r], installed) -def remove_rpm_ostree_packages(packages: list[str]) -> None: - log_info("Removing rpm-ostree packages...") - if not packages: - log_info("No rpm-ostree packages to remove") - return - - for package in packages: - log_info(f"Removing rpm-ostree package: {package}") - if run_command(["rpm-ostree", "uninstall", "-y", package]): - log_success(f"Removed {package}") - else: - log_error(f"Failed to remove {package}") +def remove_rpm_ostree_packages(packages: list[str], installed: set[str] | None = None) -> None: + remove_entries(packages, "rpm-ostree packages", lambda p: ["rpm-ostree", "uninstall", "-y", p], installed) -def remove_nix_packages(packages: list[str]) -> None: - log_info("Removing nix packages...") +def remove_nix_packages(packages: list[str], installed: set[str] | None = None) -> None: if not packages: log_info("No nix packages to remove") return if not nix_available(): log_warning("nix not found on PATH; skipping nix package removals") return + base_names = [nix_base_name(p) for p in packages if nix_base_name(p)] + remove_entries(base_names, "nix packages", lambda r: ["nix", "profile", "remove", r], installed) - for package in packages: - ref = normalize_nix_ref(package) - log_info(f"Removing nix package: {ref}") - if run_command(["nix", "profile", "remove", ref]): - log_success(f"Removed {ref}") + +def gather_dnf_packages() -> list[str]: + ok, output = run_capture(["rpm", "-qa", "--qf", "%{NAME}\n"]) + if not ok: + log_warning("Failed to list dnf packages") + return [] + return sorted({line.strip() for line in output.splitlines() if line.strip()}) + + +def gather_brew_packages() -> list[str]: + ok, output = run_capture(["brew", "list", "--formula"]) + if not ok: + log_warning("Failed to list brew packages") + return [] + return sorted({line.strip() for line in output.splitlines() if line.strip()}) + + +def gather_flatpak_refs() -> list[str]: + ok, output = run_capture(["flatpak", "list", "--app", "--columns=ref"]) + if not ok: + log_warning("Failed to list flatpak refs") + return [] + return sorted({line.strip() for line in output.splitlines() if line.strip() and "/" in line}) + + +def gather_rpm_ostree_packages() -> list[str]: + ok, output = run_capture(["rpm-ostree", "status", "--json"]) + pkgs: set[str] = set() + + if ok: + try: + data = json.loads(output) + deployments = data.get("deployments", []) + if deployments: + deployment = deployments[0] + + def collect(entries: Any) -> None: + if isinstance(entries, list): + for entry in entries: + if isinstance(entry, str): + if entry: + pkgs.add(entry) + elif isinstance(entry, dict) and entry.get("name"): + pkgs.add(str(entry["name"])) + + for key in ( + "requested-packages", + "packages", + "layered-packages", + "layered", + "requested-local-packages", + "local-packages", + ): + collect(deployment.get(key)) + except json.JSONDecodeError: + log_warning("Failed to parse rpm-ostree status json") + + # Fallback: parse `rpm-ostree override list` to catch layered packages + if not pkgs: + ok_override, override_out = run_capture(["rpm-ostree", "override", "list"]) + if ok_override: + collecting = False + for line in override_out.splitlines(): + if line.strip().startswith("Packages:"): + collecting = True + continue + if collecting: + if not line.strip(): + break + parts = line.split() + if parts: + pkgs.add(parts[0]) else: - log_error(f"Failed to remove {ref}") + log_warning("Failed to list rpm-ostree overrides") + + if not pkgs and not ok: + log_warning("Failed to list rpm-ostree packages") + + return sorted(pkgs) + + +def gather_nix_packages() -> list[str]: + if not nix_available(): + log_warning("nix not found on PATH; skipping nix package import") + return [] + collected: set[str] = set() + + def collect_profile() -> None: + ok, output = run_capture(["nix", "profile", "list", "--json"]) + if not ok: + log_warning("Failed to list nix profile entries") + return + try: + data = json.loads(output) + elements = data.get("elements") + if isinstance(elements, dict) and elements: + for key, entry in elements.items(): + attr_name = entry.get("attrPath").split(".")[-1] if entry.get("attrPath") else None + candidates = [ + attr_name, + key, + entry.get("name"), + entry.get("originalInput"), + entry.get("source"), + entry.get("originalUrl"), + entry.get("url"), + ] + for val in candidates: + if not val: + continue + norm = nix_base_name(str(val)) + if norm: + collected.add(norm) + break + else: + entries = data.get("entries", []) + for entry in entries: + val = entry.get("originalInput") or entry.get("source") or entry.get("name") + if not val: + continue + norm = nix_base_name(str(val)) + if norm: + collected.add(norm) + except json.JSONDecodeError: + log_warning("Failed to parse nix profile list json") + + def collect_nix_env() -> None: + ok, output = run_capture(["nix-env", "--query", "--installed", "--json"]) + if not ok: + return + try: + data = json.loads(output) + for item in data: + name = item.get("name") + if not name: + continue + norm = nix_base_name(str(name)) + if norm: + collected.add(norm) + except json.JSONDecodeError: + pass + + collect_profile() + if not collected: + collect_nix_env() + + return sorted(collected) def backup_target(target_path: Path, backup_dir: Path) -> None: @@ -585,6 +712,83 @@ def save_previous_config(current_path: Path, previous_path: Path) -> None: log_warning(f"Failed to write previous configuration snapshot: {exc}") +def reset_previous_config() -> None: + _, _, _, rollback_toml = get_paths() + if rollback_toml.exists(): + try: + rollback_toml.unlink() + log_success(f"Removed previous inventory snapshot: {rollback_toml}") + except OSError as exc: + log_error(f"Failed to remove previous inventory snapshot: {exc}") + else: + log_info("No previous inventory snapshot to remove.") + + +def update_section_entries(config_path: Path, section: str, entries: list[str]) -> None: + entries = sorted({entry.strip() for entry in entries if entry.strip()}) + header = f"[{section}]" + lines = config_path.read_text().splitlines() + output: list[str] = [] + i = 0 + found = False + + while i < len(lines): + line = lines[i] + section_match = re.match(r"\s*\[(.+)]\s*$", line) + if section_match: + current_section = section_match.group(1).strip() + if current_section == section: + found = True + output.append(header) + i += 1 + while i < len(lines) and not re.match(r"\s*\[.+]\s*$", lines[i]): + i += 1 + output.extend(entries) + continue + output.append(line) + i += 1 + + if not found: + if output and output[-1].strip(): + output.append("") + output.append(header) + output.extend(entries) + + config_path.write_text("\n".join(output) + "\n") + + +def merge_section_entries(config_path: Path, section: str, add: set[str], remove: set[str]) -> None: + add_clean = {entry.strip() for entry in add if entry.strip()} + remove_clean = {entry.strip() for entry in remove if entry.strip()} + # Load current entries + current_config = parse_config(config_path) + current_entries = set() + if section == "dnf": + current_entries.update(current_config.packages) + elif section == "brew": + current_entries.update(current_config.brew_packages) + elif section == "flatpak": + current_entries.update(current_config.flatpak_refs) + elif section == "rpm-ostree": + current_entries.update(current_config.rpm_ostree_packages) + elif section == "nix": + current_entries.update(current_config.nix_packages) + elif section == "copr": + current_entries.update(current_config.copr) + elif section == "dotfiles": + current_entries.update(f"{k} = {v}" for k, v in current_config.dotfiles.items()) + else: + log_error(f"Unknown section: {section}") + return + + if section == "nix": + add_clean = {normalize_nix_ref(item) for item in add_clean} + remove_clean = {normalize_nix_ref(item) for item in remove_clean} + + new_entries = (current_entries | add_clean) - remove_clean + update_section_entries(config_path, section, sorted(new_entries)) + + def switch_command() -> None: _switch(rollback=False) @@ -594,16 +798,16 @@ def switch_command_with_args(rollback: bool = False) -> None: def _switch(rollback: bool) -> None: - curator_dir, curator_toml, previous_toml = get_paths() + curator_dir, curator_toml, last_toml, rollback_toml = get_paths() if not curator_toml.exists() and not rollback: log_error("inventory.toml not found. Run 'curator init' first.") sys.exit(1) if rollback: - if not previous_toml.exists(): - log_error("No previous inventory.toml to roll back to.") + if not rollback_toml.exists() or not last_toml.exists(): + log_error("No rollback snapshots found. Run a normal switch first.") sys.exit(1) - # Backup current config if present + current_before = load_config(curator_toml) if curator_toml.exists() else empty_config() current_options = current_before.options or {"backup": True, "backup_dir": "backup"} backup_dir_name = str(current_options.get("backup_dir", "backup") or "backup") @@ -614,13 +818,12 @@ def _switch(rollback: bool) -> None: except OSError as exc: log_warning(f"Failed to back up current inventory.toml before rollback: {exc}") - shutil.copy2(previous_toml, curator_toml) - log_info(f"Restored inventory.toml from {previous_toml}") - current_config = parse_config(curator_toml) - previous_config = current_before + target_config = parse_config(last_toml) + baseline_config = parse_config(rollback_toml) + current_config, previous_config = target_config, baseline_config else: current_config = parse_config(curator_toml) - previous_config = load_config(previous_toml) + previous_config = load_config(last_toml) copr_added, copr_removed = diff_items(current_config.copr, previous_config.copr) packages_added, packages_removed = diff_items(current_config.packages, previous_config.packages) @@ -629,28 +832,108 @@ def _switch(rollback: bool) -> None: rpm_ostree_added, rpm_ostree_removed = diff_items( current_config.rpm_ostree_packages, previous_config.rpm_ostree_packages ) - nix_added, nix_removed = diff_items(current_config.nix_packages, previous_config.nix_packages) + current_nix_base = [nix_base_name(p) for p in current_config.nix_packages if nix_base_name(p)] + previous_nix_base = [nix_base_name(p) for p in previous_config.nix_packages if nix_base_name(p)] + nix_added_base, nix_removed_base = diff_items(current_nix_base, previous_nix_base) + current_nix_install = [nix_install_ref(p) for p in current_nix_base if nix_install_ref(p)] + nix_added_install = [nix_install_ref(p) for p in nix_added_base if nix_install_ref(p)] + + need_dnf = bool(current_config.packages or packages_removed) + need_brew = bool(current_config.brew_packages or brew_removed) + need_flatpak = bool(current_config.flatpak_refs or flatpak_removed) + need_rpm_ostree = bool(current_config.rpm_ostree_packages or rpm_ostree_removed) + need_nix = bool(current_nix_install or nix_removed_base) + + dnf_installed = set(gather_dnf_packages()) if need_dnf else set() + brew_installed = set(gather_brew_packages()) if need_brew else set() + flatpak_installed = set(gather_flatpak_refs()) if need_flatpak else set() + rpm_ostree_installed = set(gather_rpm_ostree_packages()) if need_rpm_ostree else set() + nix_installed = set(nix_install_ref(p) for p in gather_nix_packages()) if need_nix else None enable_copr_repos(copr_added if copr_added else current_config.copr) - disable_copr_repos(set(current_config.copr), set(copr_removed)) - install_packages(packages_added if packages_added else current_config.packages) - install_brew_packages(brew_added if brew_added else current_config.brew_packages) - install_flatpaks(flatpak_added if flatpak_added else current_config.flatpak_refs) - install_rpm_ostree_packages(rpm_ostree_added if rpm_ostree_added else current_config.rpm_ostree_packages) - install_nix_packages(nix_added if nix_added else current_config.nix_packages) - remove_packages(packages_removed) - remove_brew_packages(brew_removed) - remove_flatpaks(flatpak_removed) - remove_rpm_ostree_packages(rpm_ostree_removed) - remove_nix_packages(nix_removed) + disable_copr_repos(set(copr_removed)) + install_packages(packages_added if packages_added else current_config.packages, dnf_installed) + install_brew_packages(brew_added if brew_added else current_config.brew_packages, brew_installed) + install_flatpaks(flatpak_added if flatpak_added else current_config.flatpak_refs, flatpak_installed) + install_rpm_ostree_packages( + rpm_ostree_added if rpm_ostree_added else current_config.rpm_ostree_packages, rpm_ostree_installed + ) + install_nix_packages(nix_added_install if nix_added_install else current_nix_install, nix_installed) + remove_packages(packages_removed, dnf_installed) + remove_brew_packages(brew_removed, brew_installed) + remove_flatpaks(flatpak_removed, flatpak_installed) + remove_rpm_ostree_packages(rpm_ostree_removed, rpm_ostree_installed) + remove_nix_packages(nix_removed_base, set(nix_base_name(p) for p in gather_nix_packages()) if need_nix else None) deploy_dotfiles(current_config, curator_dir) update_last_switch(curator_toml, current_config.options) - save_previous_config(curator_toml, previous_toml) + if rollback: + # Restore files to rollback state + shutil.copy2(rollback_toml, curator_toml) + shutil.copy2(rollback_toml, last_toml) + else: + if last_toml.exists(): + save_previous_config(last_toml, rollback_toml) + if curator_toml.exists(): + save_previous_config(curator_toml, last_toml) log_success("Switch completed successfully!") +def import_from_manager(manager: str) -> None: + curator_dir, curator_toml, _, _ = get_paths() + if not curator_toml.exists(): + log_error("inventory.toml not found. Run 'curator init' first.") + sys.exit(1) + + gatherers = { + "dnf": (gather_dnf_packages, "dnf"), + "brew": (gather_brew_packages, "brew"), + "flatpak": (gather_flatpak_refs, "flatpak"), + "rpm-ostree": (gather_rpm_ostree_packages, "rpm-ostree"), + "nix": (gather_nix_packages, "nix"), + "copr": (gather_copr_repos, "copr"), + } + + gather_fn, section = gatherers[manager] + entries = gather_fn() + if not entries: + log_info(f"No entries found for {manager}; inventory not updated.") + return + + update_section_entries(curator_toml, section, entries) + log_success(f"Imported {len(entries)} {manager} entries into {curator_toml}") + + +def apply_inline_updates(entry_args: list[str], add: bool) -> None: + _, curator_toml, _, _ = get_paths() + if not curator_toml.exists(): + log_error("inventory.toml not found. Run 'curator init' first.") + sys.exit(1) + + changes: dict[str, dict[str, set[str]]] = {} + for arg in entry_args: + if ":" not in arg: + log_warning(f"Skipping invalid entry (expected section:value): {arg}") + continue + section, value = arg.split(":", 1) + section = section.strip() + value = value.strip() + if not section or not value: + log_warning(f"Skipping invalid entry (empty section or value): {arg}") + continue + if section not in {"dnf", "brew", "flatpak", "rpm-ostree", "nix", "copr"}: + log_warning(f"Unknown section '{section}' in entry: {arg}") + continue + bucket = changes.setdefault(section, {"add": set(), "remove": set()}) + target = "add" if add else "remove" + bucket[target].add(value) + + for section, change in changes.items(): + merge_section_entries(curator_toml, section, change["add"], change["remove"]) + log_success(f"Updated section [{section}] in {curator_toml}") + + def status_command() -> None: - curator_dir, curator_toml, _ = get_paths() + curator_dir, curator_toml, _, _ = get_paths() log_info("curator Status") print(f" Config directory: {curator_dir}") print(f" Config file: {curator_toml}") @@ -746,6 +1029,17 @@ def build_parser() -> argparse.ArgumentParser: action="store_true", help="Restore the previous inventory.toml snapshot and apply it", ) + from_parser = subparsers.add_parser("from", help="Import currently installed packages into inventory.toml") + from_parser.add_argument( + "manager", + choices=["dnf", "brew", "flatpak", "rpm-ostree", "nix", "copr"], + help="Package manager to import from", + ) + subparsers.add_parser("reset", help="Remove the previous inventory.toml snapshot") + add_parser = subparsers.add_parser("add", help="Add entries to inventory.toml (e.g. dnf:uv nix:micro)") + add_parser.add_argument("entries", nargs="+", help="Entries in the form section:value") + remove_parser = subparsers.add_parser("remove", help="Remove entries from inventory.toml (e.g. dnf:uv nix:micro)") + remove_parser.add_argument("entries", nargs="+", help="Entries in the form section:value") subparsers.add_parser("status", help="Show current configuration status") subparsers.add_parser("help", help="Show help message") return parser @@ -759,6 +1053,14 @@ def main(argv: list[str] | None = None) -> None: init_command() elif args.command == "switch": switch_command_with_args(getattr(args, "rollback", False)) + elif args.command == "from": + import_from_manager(args.manager) + elif args.command == "reset": + reset_previous_config() + elif args.command == "add": + apply_inline_updates(args.entries, add=True) + elif args.command == "remove": + apply_inline_updates(args.entries, add=False) elif args.command == "status": status_command() elif args.command == "help": diff --git a/tests/test_config.py b/tests/test_config.py index 9313f3d..a5b2a62 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -122,3 +122,43 @@ last_switch = "" backups = list(backup_dir.iterdir()) assert backups, "expected a backup of inventory.toml to be created" assert any(path.name.startswith("inventory.toml.") for path in backups) + +def test_reset_previous_config(tmp_path: Path, monkeypatch) -> None: + rollback = tmp_path / "inventory.toml.rollback" + rollback.write_text("prev") + + def fake_get_paths(): + return tmp_path, tmp_path / "inventory.toml", tmp_path / "inventory.toml.last", rollback + + monkeypatch.setattr(cli, "get_paths", fake_get_paths) + cli.reset_previous_config() + assert not rollback.exists() + + +def test_update_section_entries_rewrites_section(tmp_path: Path) -> None: + config_path = tmp_path / "inventory.toml" + config_path.write_text( + """ +[dnf] +git + +[copr] +copr.fedorainfracloud.org/user/repo + +[brew] +wget + """.strip() + ) + + cli.update_section_entries(config_path, "dnf", ["vim", "curl", "vim"]) + cli.update_section_entries(config_path, "copr", ["copr.fedorainfracloud.org/user/repo", "copr.fedorainfracloud.org/another/repo"]) + + content = config_path.read_text().strip().splitlines() + dnf_index = content.index("[dnf]") + assert content[dnf_index + 1 : dnf_index + 3] == ["curl", "vim"] + assert "[copr]" in content + copr_index = content.index("[copr]") + assert content[copr_index + 1 : copr_index + 3] == [ + "copr.fedorainfracloud.org/another/repo", + "copr.fedorainfracloud.org/user/repo", + ]