randogoth/curator
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

finalized for publication

Browse source
This commit is contained in:
randogoth 2025-12-24 12:56:51 +02:00
parent df1f387795
commit 7d51de1052
4 changed files with 626 additions and 392 deletions
Download patch file Download diff file Expand all files Collapse all files

75
.gitignore vendored Normal file
View file

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

287
README.md
View file

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

616
src/curator/cli.py
View file

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

40
tests/test_config.py
View file

@ -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",
]