509 lines
20 KiB
Python
509 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import argparse
|
|
import os
|
|
import platform
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import yaml
|
|
from pathlib import Path
|
|
from typing import Dict, Any, List, Optional
|
|
import re
|
|
import urllib.request
|
|
import json
|
|
|
|
|
|
class DotfilesManager:
|
|
def __init__(self, manifest_path: str = "manifest.yaml"):
|
|
self.manifest_path = Path(manifest_path)
|
|
self.dotfiles_dir = Path.home() / ".dotfiles"
|
|
self.config_dir = self.dotfiles_dir / "config"
|
|
self.variables = {}
|
|
self.manifest = {}
|
|
|
|
# Detect system info
|
|
self.system_os = "macos" if platform.system() == "Darwin" else "linux"
|
|
self.system_arch = self._get_system_arch()
|
|
self.system_platform = f"{self.system_os}-{self.system_arch}"
|
|
|
|
# Load manifest
|
|
self._load_manifest()
|
|
|
|
def _get_system_arch(self) -> str:
|
|
arch = platform.machine().lower()
|
|
if arch in ["x86_64", "amd64"]:
|
|
return "amd64"
|
|
elif arch in ["aarch64", "arm64"]:
|
|
return "arm64"
|
|
else:
|
|
return arch
|
|
|
|
def _load_manifest(self):
|
|
"""Load the YAML manifest file"""
|
|
try:
|
|
with open(self.manifest_path, 'r') as f:
|
|
self.manifest = yaml.safe_load(f)
|
|
except FileNotFoundError:
|
|
self.error(f"Manifest file not found: {self.manifest_path}")
|
|
except yaml.YAMLError as e:
|
|
self.error(f"Error parsing manifest: {e}")
|
|
|
|
def _substitute_variables(self, text: str) -> str:
|
|
"""Substitute variables in text"""
|
|
if not isinstance(text, str):
|
|
return text
|
|
|
|
# Substitute environment variables and custom variables
|
|
for var, value in self.variables.items():
|
|
text = text.replace(f"${var}", str(value))
|
|
text = text.replace(f"${{{var}}}", str(value))
|
|
|
|
# Substitute common environment variables
|
|
text = text.replace("$USER", os.getenv("USER", ""))
|
|
text = text.replace("$HOME", str(Path.home()))
|
|
|
|
return text
|
|
|
|
def info(self, message: str):
|
|
"""Print info message"""
|
|
print(f"[INFO] {message}")
|
|
|
|
def warn(self, message: str):
|
|
"""Print warning message"""
|
|
print(f"[WARN] {message}")
|
|
|
|
def error(self, message: str):
|
|
"""Print error message and exit"""
|
|
print(f"[ERROR] {message}")
|
|
sys.exit(1)
|
|
|
|
def run_command(self, command: str, check: bool = True, shell: bool = True) -> subprocess.CompletedProcess:
|
|
"""Run a shell command"""
|
|
self.info(f"Running: {command}")
|
|
try:
|
|
result = subprocess.run(command, shell=shell, check=check,
|
|
capture_output=True, text=True)
|
|
if result.stdout:
|
|
print(result.stdout)
|
|
return result
|
|
except subprocess.CalledProcessError as e:
|
|
self.error(f"Command failed: {command}\nError: {e.stderr}")
|
|
|
|
def _detect_package_manager(self, os_type: str) -> str:
|
|
"""Detect available package manager"""
|
|
if os_type == "macos":
|
|
if shutil.which("brew"):
|
|
return "brew"
|
|
else:
|
|
self.info("Installing Homebrew...")
|
|
self.run_command('/bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/HEAD/install.sh)"')
|
|
return "brew"
|
|
else: # linux
|
|
if shutil.which("apt"):
|
|
return "apt"
|
|
elif shutil.which("yum"):
|
|
return "yum"
|
|
elif shutil.which("pacman"):
|
|
return "pacman"
|
|
else:
|
|
self.error("No supported package manager found")
|
|
|
|
def _install_package(self, package_manager: str, package_type: str, package_name: str):
|
|
"""Install a single package"""
|
|
if package_manager == "brew":
|
|
if package_type == "cask":
|
|
self.run_command(f"brew install --cask {package_name}")
|
|
else:
|
|
self.run_command(f"brew install {package_name}")
|
|
elif package_manager == "apt":
|
|
self.run_command(f"sudo apt-get update && sudo apt-get install -y {package_name}")
|
|
elif package_manager == "yum":
|
|
self.run_command(f"sudo yum install -y {package_name}")
|
|
elif package_manager == "pacman":
|
|
self.run_command(f"sudo pacman -S --noconfirm {package_name}")
|
|
|
|
def _get_github_release_url(self, repo: str, version: str, asset_pattern: str, platform_map: Dict) -> str:
|
|
"""Get GitHub release download URL"""
|
|
if self.system_platform not in platform_map:
|
|
self.error(f"Platform {self.system_platform} not supported for {repo}")
|
|
|
|
platform_info = platform_map[self.system_platform]
|
|
asset_name = asset_pattern.replace("{{os}}", platform_info["os"]).replace("{{arch}}", platform_info["arch"])
|
|
|
|
return f"https://github.com/{repo}/releases/download/v{version}/{asset_name}"
|
|
|
|
def _install_binary(self, binary_name: str, binary_config: Dict):
|
|
"""Install a binary from the binaries section"""
|
|
self.info(f"Installing binary: {binary_name}")
|
|
|
|
# Install dependencies first
|
|
if "dependencies" in binary_config:
|
|
pm = self._detect_package_manager(self.system_os)
|
|
for dep in binary_config["dependencies"]:
|
|
self._install_package(pm, "package", dep)
|
|
|
|
# Get download URL
|
|
if binary_config["source"].startswith("github:"):
|
|
repo = binary_config["source"].replace("github:", "")
|
|
download_url = self._get_github_release_url(
|
|
repo,
|
|
binary_config["version"],
|
|
binary_config["asset-pattern"],
|
|
binary_config["platform-map"]
|
|
)
|
|
else:
|
|
self.error(f"Unsupported binary source: {binary_config['source']}")
|
|
|
|
# Substitute variables in install script
|
|
install_script = binary_config["install-script"]
|
|
install_script = install_script.replace("{{downloadUrl}}", download_url)
|
|
install_script = install_script.replace("{{version}}", binary_config["version"])
|
|
if self.system_platform in binary_config["platform-map"]:
|
|
platform_info = binary_config["platform-map"][self.system_platform]
|
|
install_script = install_script.replace("{{os}}", platform_info["os"])
|
|
install_script = install_script.replace("{{arch}}", platform_info["arch"])
|
|
|
|
# Run install script
|
|
self.run_command(install_script)
|
|
|
|
def _symlink_config(self, source_path: Path, target_path: Path, copy: bool = False, force: bool = False):
|
|
"""Create symlink or copy for configuration file"""
|
|
if target_path.exists() and not force:
|
|
self.warn(f"Target already exists, skipping: {target_path}")
|
|
return
|
|
|
|
if target_path.exists() and force:
|
|
if target_path.is_dir():
|
|
shutil.rmtree(target_path)
|
|
else:
|
|
target_path.unlink()
|
|
|
|
# Create parent directories
|
|
target_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
if copy:
|
|
if source_path.is_dir():
|
|
shutil.copytree(source_path, target_path)
|
|
else:
|
|
shutil.copy2(source_path, target_path)
|
|
self.info(f"Copied: {source_path} -> {target_path}")
|
|
else:
|
|
target_path.symlink_to(source_path)
|
|
self.info(f"Linked: {source_path} -> {target_path}")
|
|
|
|
def _link_config_directory(self, config_name: str, environment: str, copy: bool = False, force: bool = False):
|
|
"""Link all files from a config directory"""
|
|
# Try environment-specific config first, then shared
|
|
env_config_dir = self.config_dir / environment / config_name
|
|
shared_config_dir = self.config_dir / "shared" / config_name
|
|
|
|
source_dir = env_config_dir if env_config_dir.exists() else shared_config_dir
|
|
|
|
if not source_dir.exists():
|
|
self.warn(f"Config directory not found: {config_name}")
|
|
return
|
|
|
|
self.info(f"Linking config: {config_name} from {source_dir}")
|
|
|
|
# Walk through all files and directories
|
|
for item in source_dir.rglob("*"):
|
|
if item.is_file():
|
|
# Calculate relative path from source_dir
|
|
rel_path = item.relative_to(source_dir)
|
|
target_path = Path.home() / rel_path
|
|
self._symlink_config(item, target_path, copy, force)
|
|
|
|
def _set_hostname(self, hostname: str):
|
|
"""Set system hostname"""
|
|
hostname = self._substitute_variables(hostname)
|
|
self.info(f"Setting hostname to: {hostname}")
|
|
|
|
if self.system_os == "macos":
|
|
self.run_command(f"sudo scutil --set ComputerName '{hostname}'")
|
|
self.run_command(f"sudo scutil --set HostName '{hostname}'")
|
|
self.run_command(f"sudo scutil --set LocalHostName '{hostname}'")
|
|
else:
|
|
self.run_command(f"sudo hostnamectl set-hostname '{hostname}'")
|
|
|
|
def _set_shell(self, shell: str):
|
|
"""Set default shell"""
|
|
shell_path = shutil.which(shell)
|
|
if not shell_path:
|
|
self.error(f"Shell not found: {shell}")
|
|
|
|
self.info(f"Setting shell to: {shell_path}")
|
|
|
|
# Add shell to /etc/shells if not present
|
|
try:
|
|
with open("/etc/shells", "r") as f:
|
|
shells = f.read()
|
|
if shell_path not in shells:
|
|
self.run_command(f"echo '{shell_path}' | sudo tee -a /etc/shells")
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
# Change user shell
|
|
self.run_command(f"chsh -s {shell_path}")
|
|
|
|
def _set_locale(self, locale: str):
|
|
"""Set system locale"""
|
|
if self.system_os == "linux":
|
|
self.info(f"Setting locale to: {locale}")
|
|
self.run_command(f"sudo locale-gen {locale}")
|
|
self.run_command(f"sudo update-locale LANG={locale}")
|
|
|
|
def _generate_ssh_keys(self, ssh_configs: List[Dict]):
|
|
"""Generate SSH keys"""
|
|
ssh_dir = Path.home() / ".ssh"
|
|
ssh_dir.mkdir(mode=0o700, exist_ok=True)
|
|
|
|
for config in ssh_configs:
|
|
key_type = config["type"]
|
|
comment = self._substitute_variables(config.get("comment", ""))
|
|
filename = config.get("filename", f"id_{key_type}")
|
|
|
|
key_path = ssh_dir / filename
|
|
|
|
if key_path.exists():
|
|
self.warn(f"SSH key already exists: {key_path}")
|
|
continue
|
|
|
|
self.info(f"Generating SSH key: {key_path}")
|
|
cmd = f'ssh-keygen -t {key_type} -f "{key_path}" -N "" -C "{comment}"'
|
|
self.run_command(cmd)
|
|
|
|
def setup_environment(self, environment_name: str):
|
|
"""Setup complete environment"""
|
|
if environment_name not in self.manifest.get("environments", {}):
|
|
self.error(f"Environment not found: {environment_name}")
|
|
|
|
env_config = self.manifest["environments"][environment_name]
|
|
|
|
# Check required variables
|
|
if "requires" in env_config:
|
|
for req_var in env_config["requires"]:
|
|
if req_var not in self.variables:
|
|
self.error(f"Required variable not set: {req_var}")
|
|
|
|
self.info(f"Setting up environment: {environment_name}")
|
|
|
|
# Set hostname
|
|
if "hostname" in env_config:
|
|
self._set_hostname(env_config["hostname"])
|
|
|
|
# Set shell
|
|
if "shell" in env_config:
|
|
self._set_shell(env_config["shell"])
|
|
|
|
# Set locale
|
|
if "locale" in env_config:
|
|
self._set_locale(env_config["locale"])
|
|
|
|
# Install packages
|
|
self.install_packages(environment_name)
|
|
|
|
# Link configurations
|
|
self.link_configs(environment_name)
|
|
|
|
# Generate SSH keys
|
|
if "ssh_keygen" in env_config:
|
|
self._generate_ssh_keys(env_config["ssh_keygen"])
|
|
|
|
# Run custom commands
|
|
if "runcmd" in env_config:
|
|
for command in env_config["runcmd"]:
|
|
command = self._substitute_variables(command)
|
|
self.run_command(command)
|
|
|
|
self.info(f"Environment setup complete: {environment_name}")
|
|
|
|
def install_packages(self, environment_name: str, package_name: str = None):
|
|
"""Install packages for environment"""
|
|
if environment_name not in self.manifest.get("environments", {}):
|
|
self.error(f"Environment not found: {environment_name}")
|
|
|
|
env_config = self.manifest["environments"][environment_name]
|
|
|
|
if "packages" not in env_config:
|
|
self.info("No packages to install")
|
|
return
|
|
|
|
# Detect package manager
|
|
pm = env_config.get("package-manager")
|
|
if not pm:
|
|
pm = self._detect_package_manager(env_config.get("os", self.system_os))
|
|
|
|
packages = env_config["packages"]
|
|
|
|
# Install specific package if requested
|
|
if package_name:
|
|
self._install_single_package(packages, package_name, pm, environment_name)
|
|
return
|
|
|
|
# Install all packages
|
|
for package_type, package_list in packages.items():
|
|
for package in package_list:
|
|
if isinstance(package, str):
|
|
package_spec = {"name": package}
|
|
else:
|
|
package_spec = package
|
|
|
|
self._install_package_spec(package_spec, package_type, pm, environment_name)
|
|
|
|
def _install_single_package(self, packages: Dict, package_name: str, pm: str, environment_name: str):
|
|
"""Install a single package by name"""
|
|
for package_type, package_list in packages.items():
|
|
for package in package_list:
|
|
if isinstance(package, str):
|
|
if package == package_name:
|
|
self._install_package_spec({"name": package}, package_type, pm, environment_name)
|
|
return
|
|
else:
|
|
if package.get("name") == package_name:
|
|
self._install_package_spec(package, package_type, pm, environment_name)
|
|
return
|
|
|
|
self.error(f"Package not found: {package_name}")
|
|
|
|
def _install_package_spec(self, package_spec: Dict, package_type: str, pm: str, environment_name: str):
|
|
"""Install a package specification"""
|
|
package_name = package_spec["name"]
|
|
|
|
if package_type == "binary":
|
|
# Install from binaries section
|
|
if package_name not in self.manifest.get("binaries", {}):
|
|
self.error(f"Binary not found in manifest: {package_name}")
|
|
self._install_binary(package_name, self.manifest["binaries"][package_name])
|
|
else:
|
|
# Install via package manager
|
|
self._install_package(pm, package_type, package_name)
|
|
|
|
# Run post-install script
|
|
if "post-install" in package_spec:
|
|
script = self._substitute_variables(package_spec["post-install"])
|
|
self.run_command(script)
|
|
|
|
# Show post-install comment
|
|
if "post-install-comment" in package_spec:
|
|
comment = self._substitute_variables(package_spec["post-install-comment"])
|
|
self.info(f"POST-INSTALL: {comment}")
|
|
|
|
# Link config if not disabled
|
|
if package_spec.get("link", True):
|
|
self._link_config_directory(package_name, environment_name)
|
|
|
|
# Run post-link script
|
|
if "post-link" in package_spec:
|
|
script = self._substitute_variables(package_spec["post-link"])
|
|
self.run_command(script)
|
|
|
|
# Show post-link comment
|
|
if "post-link-comment" in package_spec:
|
|
comment = self._substitute_variables(package_spec["post-link-comment"])
|
|
self.info(f"POST-LINK: {comment}")
|
|
|
|
def link_configs(self, environment_name: str, config_name: str = None, copy: bool = False, force: bool = False):
|
|
"""Link configuration files"""
|
|
if environment_name not in self.manifest.get("environments", {}):
|
|
self.error(f"Environment not found: {environment_name}")
|
|
|
|
env_config = self.manifest["environments"][environment_name]
|
|
|
|
if config_name:
|
|
# Link specific config
|
|
self._link_config_directory(config_name, environment_name, copy, force)
|
|
self._run_config_post_link(config_name, env_config)
|
|
else:
|
|
# Link all configs
|
|
configs = env_config.get("configs", [])
|
|
for config in configs:
|
|
if isinstance(config, str):
|
|
config_spec = {"name": config}
|
|
else:
|
|
config_spec = config
|
|
|
|
config_name = config_spec["name"] if "name" in config_spec else config
|
|
self._link_config_directory(config_name, environment_name, copy, force)
|
|
self._run_config_post_link(config_spec, env_config)
|
|
|
|
def _run_config_post_link(self, config_spec, env_config):
|
|
"""Run post-link actions for config"""
|
|
if isinstance(config_spec, str):
|
|
return
|
|
|
|
# Run post-link script
|
|
if "post-link" in config_spec:
|
|
script = self._substitute_variables(config_spec["post-link"])
|
|
self.run_command(script)
|
|
|
|
# Show post-link comment
|
|
if "post-link-comment" in config_spec:
|
|
comment = self._substitute_variables(config_spec["post-link-comment"])
|
|
self.info(f"POST-LINK: {comment}")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Dotfiles Manager")
|
|
subparsers = parser.add_subparsers(dest="command", help="Commands")
|
|
|
|
# Setup command
|
|
setup_parser = subparsers.add_parser("setup", help="Setup complete environment")
|
|
setup_parser.add_argument("environment", help="Environment name")
|
|
setup_parser.add_argument("--set", action="append", default=[],
|
|
help="Set variable (format: VAR=value)")
|
|
|
|
# Link command
|
|
link_parser = subparsers.add_parser("link", help="Link configurations")
|
|
link_parser.add_argument("environment", help="Environment name")
|
|
link_parser.add_argument("--copy", action="store_true", help="Copy instead of symlink")
|
|
link_parser.add_argument("-f", "--force", action="store_true", help="Force overwrite")
|
|
link_parser.add_argument("-p", "--package", help="Link specific package")
|
|
link_parser.add_argument("--set", action="append", default=[],
|
|
help="Set variable (format: VAR=value)")
|
|
|
|
# Install command
|
|
install_parser = subparsers.add_parser("install", help="Install packages")
|
|
install_parser.add_argument("environment", help="Environment name")
|
|
install_parser.add_argument("-p", "--package", help="Install specific package")
|
|
install_parser.add_argument("--set", action="append", default=[],
|
|
help="Set variable (format: VAR=value)")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if not args.command:
|
|
parser.print_help()
|
|
return
|
|
|
|
# Initialize manager
|
|
manager = DotfilesManager()
|
|
|
|
# Parse variables
|
|
for var_setting in args.set:
|
|
if "=" not in var_setting:
|
|
manager.error(f"Invalid variable format: {var_setting}")
|
|
key, value = var_setting.split("=", 1)
|
|
manager.variables[key] = value
|
|
|
|
# Add TARGET_HOSTNAME if not set
|
|
if "TARGET_HOSTNAME" not in manager.variables:
|
|
manager.variables["TARGET_HOSTNAME"] = manager.variables.get("TARGET_HOSTNAME", "localhost")
|
|
|
|
# Execute command
|
|
try:
|
|
if args.command == "setup":
|
|
manager.setup_environment(args.environment)
|
|
elif args.command == "link":
|
|
manager.link_configs(args.environment, args.package, args.copy, args.force)
|
|
elif args.command == "install":
|
|
manager.install_packages(args.environment, args.package)
|
|
except KeyboardInterrupt:
|
|
manager.info("Operation cancelled by user")
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
manager.error(f"Unexpected error: {e}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|