RPM_sync.py

import subprocess
import json
import os
import sys
import shutil # Import shutil to check for command existence
import argparse # Import argparse

def detect_system_type():
    """Detects if the system uses rpm-ostree or dnf."""
    if os.path.exists('/run/ostree-booted'):
        return 'ostree'
    elif shutil.which('dnf'):
        return 'dnf'
    else:
        return None

def get_installed_packages(system_type):
    """Get the set of installed package names based on system type."""
    packages = set()
    try:
        if system_type == 'ostree':
            result = subprocess.run(['rpm-ostree', 'status', '--json'], capture_output=True, text=True, check=True)
            status = json.loads(result.stdout)
            # Assuming the first deployment is the relevant one
            if status and status[0].get('packages'):
                 packages = set(status[0]['packages'])
        elif system_type == 'dnf':
            # Use rpm -qa to get package names, as dnf list installed is slower
            result = subprocess.run(['rpm', '-qa', '--qf', '%{NAME}\n'], capture_output=True, text=True, check=True)
            packages = set(result.stdout.strip().split('\n'))
    except FileNotFoundError:
        print(f"Error: Command not found for system type '{system_type}'.", file=sys.stderr)
        sys.exit(1)
    except subprocess.CalledProcessError as e:
        print(f"Error running package listing command: {e}", file=sys.stderr)
        sys.exit(1)
    except json.JSONDecodeError as e:
         print(f"Error parsing JSON from rpm-ostree status: {e}", file=sys.stderr)
         sys.exit(1)
    return packages

def sync_packages(remote_host, system_type):
    """Synchronize packages with a remote host using Unison."""
    print(f"Detected system type: {system_type}")
    local_packages = get_installed_packages(system_type)

    # Define temporary file names
    local_pkg_file = f'local_packages_{system_type}.txt'
    remote_pkg_file_path = f'/tmp/remote_packages_{system_type}.txt' # Use /tmp on remote

    print(f"Writing local packages to {local_pkg_file}...")
    try:
        with open(local_pkg_file, 'w') as f:
            for package in sorted(list(local_packages)):
                f.write(f"{package}\n")
    except IOError as e:
        print(f"Error writing local package file: {e}", file=sys.stderr)
        sys.exit(1)

    # Use Unison to synchronize the package list
    unison_cmd = ['unison', local_pkg_file, f'ssh://{remote_host}/{remote_pkg_file_path}']
    print(f"Running Unison: {' '.join(unison_cmd)}")
    try:
        subprocess.run(unison_cmd, check=True)
    except FileNotFoundError:
        print("Error: 'unison' command not found. Please install Unison.", file=sys.stderr)
        sys.exit(1)
    except subprocess.CalledProcessError as e:
        print(f"Error running Unison: {e}", file=sys.stderr)
        # Decide if we should exit or try to continue with local file?
        # For now, exiting might be safer.
        sys.exit(1)

    # Read the synchronized package list
    print(f"Reading synchronized packages from {local_pkg_file}...")
    try:
        with open(local_pkg_file, 'r') as f:
            synced_packages = set(f.read().strip().splitlines())
    except IOError as e:
        print(f"Error reading synchronized package file: {e}", file=sys.stderr)
        sys.exit(1)

    # Clean up local temp file immediately after reading
    try:
        os.remove(local_pkg_file)
        print(f"Removed temporary file: {local_pkg_file}")
    except OSError as e:
        print(f"Warning: Could not remove temporary file {local_pkg_file}: {e}", file=sys.stderr)

    # Compare and install missing packages
    # Get current packages *again* in case Unison sync took time or user changed things
    current_packages = get_installed_packages(system_type)
    packages_to_install = synced_packages - current_packages

    if packages_to_install:
        print(f"\nFound {len(packages_to_install)} packages to install: {', '.join(sorted(list(packages_to_install)))}")
        install_cmd = []
        if system_type == 'ostree':
            install_cmd = ['rpm-ostree', 'install'] + sorted(list(packages_to_install))
            print(f"Running: {' '.join(install_cmd)}")
        elif system_type == 'dnf':
            install_cmd = ['sudo', 'dnf', 'install', '-y'] + sorted(list(packages_to_install))
            print(f"Running: {' '.join(install_cmd)}")
            print("Note: This may require sudo privileges.")

        try:
            subprocess.run(install_cmd, check=True)
            print("\nPackage installation command finished.")
        except FileNotFoundError:
             print(f"Error: Install command ('{install_cmd[0]}') not found.", file=sys.stderr)
             sys.exit(1)
        except subprocess.CalledProcessError as e:
            print(f"Error during package installation: {e}", file=sys.stderr)
            # Don't exit here, maybe some packages installed?
            print("Installation may have partially failed.")

    else:
        print("\nSystem is already in sync. No new packages to install.")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Synchronize installed packages (rpm-ostree or dnf) between two hosts using Unison.")
    parser.add_argument("-r", "--remote-host", required=True, help="The remote host (running a compatible system) to synchronize with.")
    args = parser.parse_args()

    system_type = detect_system_type()

    if system_type is None:
        print("Error: Could not detect a supported package manager (rpm-ostree or dnf).", file=sys.stderr)
        print("This script requires either an ostree-based system or a system with DNF.", file=sys.stderr)
        sys.exit(1)

    # Check if Unison is installed before proceeding
    if not shutil.which('unison'):
        print("Error: 'unison' command not found. Please install Unison first.", file=sys.stderr)
        sys.exit(1)

    # Get remote_host from arguments instead of input
    remote_host = args.remote_host
    # remote_host = input(f"Enter the remote host (running a {system_type} system) to synchronize with: ") # Removed input prompt
    # if not remote_host:
    #     print("Error: Remote host cannot be empty.", file=sys.stderr)
    #     sys.exit(1)

    sync_packages(remote_host, system_type)
    print("\nPackage synchronization process complete.")