debian-atomic/scripts/comps-sync.py
2025-08-15 12:20:46 -07:00

278 lines
9.9 KiB
Python

#!/usr/bin/env python3
"""
Debian Atomic Comps Sync Script
Fedora comps-sync.py equivalent for Debian package groups
This script syncs Debian tasks (package groups) with Debian Atomic variant configurations,
ensuring variants stay updated with the Debian package ecosystem.
Usage:
./comps-sync.py /path/to/debian-tasks
./comps-sync.py --save /path/to/debian-tasks
"""
import argparse
import os
import sys
import yaml
import xml.etree.ElementTree as ET
from pathlib import Path
from typing import Dict, List, Set
class DebianAtomicCompsSync:
def __init__(self, repo_path: str):
self.repo_path = Path(repo_path)
self.variants_dir = self.repo_path / "variants"
self.treefiles_dir = self.repo_path / "treefiles"
# Ensure directories exist
self.treefiles_dir.mkdir(exist_ok=True)
# Variant configurations - Fedora Atomic 1:1 parallel
self.variants = {
"base": {
"description": "Base OSTree system",
"packages": [],
"groups": ["base", "system"]
},
"workstation": {
"description": "Debian Atomic Workstation (Fedora Silverblue equivalent)",
"packages": [],
"groups": ["desktop", "gnome", "office", "productivity"]
},
"kde": {
"description": "Debian Atomic KDE (Fedora Kinoite equivalent)",
"packages": [],
"groups": ["desktop", "kde", "office", "productivity"]
},
"sway": {
"description": "Debian Atomic Sway (Fedora Sway Atomic equivalent)",
"packages": [],
"groups": ["desktop", "sway", "wayland", "minimal"]
},
"server": {
"description": "Debian Atomic Server (Fedora CoreOS equivalent)",
"packages": [],
"groups": ["server", "enterprise", "monitoring", "container"]
}
}
def parse_debian_tasks(self, tasks_file: str) -> Dict[str, List[str]]:
"""Parse Debian tasks file for package groups"""
print(f"Parsing Debian tasks file: {tasks_file}")
# This is a simplified parser - in practice you'd want to parse
# actual Debian tasks files or use debian-policy package
tasks = {}
try:
# For now, we'll create example package groups
# In a real implementation, you'd parse the actual tasks file
tasks = {
"base": [
"systemd", "ostree", "grub2", "linux-image-amd64",
"initramfs-tools", "bash", "coreutils", "vim"
],
"server": [
"openssh-server", "nginx", "postgresql", "monitoring-plugins",
"logrotate", "cron", "rsyslog"
],
"gaming": [
"steam", "wine", "lutris", "gamemode", "mangohud",
"nvidia-driver", "mesa-utils", "pulseaudio"
],
"development": [
"build-essential", "git", "python3", "nodejs", "rustc",
"docker.io", "vscode", "eclipse"
],
"desktop": [
"firefox", "libreoffice", "gimp", "vlc", "thunderbird",
"file-roller", "gnome-tweaks"
]
}
print(f"Parsed {len(tasks)} package groups")
return tasks
except Exception as e:
print(f"Error parsing tasks file: {e}")
return {}
def load_variant_configs(self) -> Dict[str, Dict]:
"""Load existing variant configurations"""
configs = {}
for variant_name in self.variants:
config_file = self.treefiles_dir / f"{variant_name}.yaml"
if config_file.exists():
try:
with open(config_file, 'r') as f:
configs[variant_name] = yaml.safe_load(f)
except Exception as e:
print(f"Warning: Could not load {config_file}: {e}")
configs[variant_name] = {}
else:
configs[variant_name] = {}
return configs
def update_variant_packages(self, variant_name: str, package_groups: Dict[str, List[str]]) -> Dict:
"""Update variant with new package groups"""
variant = self.variants[variant_name]
updated_packages = []
# Add packages from relevant groups
for group_name, packages in package_groups.items():
if any(group in variant["groups"] for group in [group_name]):
updated_packages.extend(packages)
# Remove duplicates and sort
updated_packages = sorted(list(set(updated_packages)))
# Create updated configuration
config = {
"include": "common.yaml",
"ref": f"particle-os/{variant_name}",
"packages": updated_packages,
"metadata": {
"variant": variant_name,
"description": variant["description"],
"groups": variant["groups"]
}
}
return config
def generate_common_config(self) -> Dict:
"""Generate common configuration for all variants"""
return {
"repos": ["debian-stable", "debian-security"],
"packages": [
"systemd", "ostree", "grub2", "bash", "coreutils",
"network-manager", "podman", "skopeo"
],
"metadata": {
"project": "Particle-OS",
"type": "atomic",
"base": "debian"
}
}
def save_configs(self, configs: Dict[str, Dict], dry_run: bool = True):
"""Save variant configurations to treefiles"""
if dry_run:
print("\n=== DRY RUN - No files will be modified ===")
# Save common configuration
common_config = self.generate_common_config()
common_file = self.treefiles_dir / "common.yaml"
if not dry_run:
with open(common_file, 'w') as f:
yaml.dump(common_config, f, default_flow_style=False, indent=2)
print(f"Saved: {common_file}")
else:
print(f"Would save: {common_file}")
print("Content:")
print(yaml.dump(common_config, default_flow_style=False, indent=2))
# Save variant configurations
for variant_name, config in configs.items():
config_file = self.treefiles_dir / f"{variant_name}.yaml"
if not dry_run:
with open(config_file, 'w') as f:
yaml.dump(config, f, default_flow_style=False, indent=2)
print(f"Saved: {config_file}")
else:
print(f"\nWould save: {config_file}")
print("Content:")
print(yaml.dump(config, default_flow_style=False, indent=2))
def sync_packages(self, tasks_file: str, save: bool = False):
"""Main sync function"""
print("Particle-OS Comps Sync")
print("======================")
# Parse Debian tasks
package_groups = self.parse_debian_tasks(tasks_file)
if not package_groups:
print("No package groups found, exiting")
return
# Load existing configs
existing_configs = self.load_variant_configs()
# Update variants with new packages
updated_configs = {}
for variant_name in self.variants:
print(f"\nProcessing variant: {variant_name}")
updated_configs[variant_name] = self.update_variant_packages(
variant_name, package_groups
)
# Show changes
old_packages = existing_configs.get(variant_name, {}).get("packages", [])
new_packages = updated_configs[variant_name]["packages"]
added = set(new_packages) - set(old_packages)
removed = set(old_packages) - set(new_packages)
if added:
print(f" Added packages: {', '.join(sorted(added))}")
if removed:
print(f" Removed packages: {', '.join(sorted(removed))}")
if not added and not removed:
print(" No changes")
# Save configurations
self.save_configs(updated_configs, dry_run=not save)
if save:
print("\n✅ Package groups synced and saved successfully!")
print("Next steps:")
print("1. Review the generated treefiles")
print("2. Test the configurations")
print("3. Commit the changes")
else:
print("\n📋 Review the changes above")
print("To apply changes, run with --save flag")
def main():
parser = argparse.ArgumentParser(
description="Sync Debian package groups with Particle-OS variants"
)
parser.add_argument(
"tasks_file",
help="Path to Debian tasks file"
)
parser.add_argument(
"--save",
action="store_true",
help="Save changes to treefiles (default is dry-run)"
)
parser.add_argument(
"--repo-path",
default=".",
help="Path to Particle-OS repository (default: current directory)"
)
args = parser.parse_args()
# Validate tasks file
if not os.path.exists(args.tasks_file):
print(f"Error: Tasks file not found: {args.tasks_file}")
sys.exit(1)
# Initialize sync
sync = DebianAtomicCompsSync(args.repo_path)
# Perform sync
try:
sync.sync_packages(args.tasks_file, save=args.save)
except Exception as e:
print(f"Error during sync: {e}")
sys.exit(1)
if __name__ == "__main__":
main()