debian-atomic/scripts/comps-sync.py

#!/usr/bin/env python3
"""
Debian Atomic Package Group Synchronization
Simplified version for current treefile structure
"""

import yaml
import os
import sys
import argparse
from pathlib import Path

class DebianAtomicCompsSync:
    def __init__(self, repo_path: str = "."):
        self.repo_path = Path(repo_path)
        self.treefiles_dir = self.repo_path / "treefiles"
        self.variants = ["base", "workstation", "server", "testing", "debian-bootc-base"]

    def parse_debian_tasks(self, tasks_file: str) -> dict:
        """Parse Debian tasks file"""
        try:
            with open(tasks_file, 'r') as f:
                data = yaml.safe_load(f)
            
            # Extract package groups from tasks
            package_groups = {}
            if 'tasks' in data:
                for task in data['tasks']:
                    if 'name' in task and 'packages' in task:
                        package_groups[task['name']] = task['packages']
            
            print(f"Parsed {len(package_groups)} package groups")
            return package_groups
            
        except Exception as e:
            print(f"Error parsing tasks file: {e}")
            return {}

    def load_variant_configs(self) -> 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) -> dict:
        """Update variant with new package groups"""
        # Find matching package group for this variant
        matching_group = None
        for group_name, packages in package_groups.items():
            if variant_name in group_name:
                matching_group = packages
                break
        
        if not matching_group:
            print(f"  No matching package group found for {variant_name}")
            return {"packages": []}
        
        # Create updated configuration
        config = {
            "include": "common.yaml",
            "packages": matching_group,
            "metadata": {
                "variant": variant_name,
                "description": f"Debian Atomic {variant_name} variant"
            }
        }
        
        return config

    def save_configs(self, configs: dict, dry_run: bool = True):
        """Save variant configurations to treefiles"""
        if dry_run:
            print("\n=== DRY RUN - No files will be modified ===")
        
        # Save variant configurations
        for variant_name, config in configs.items():
            if config.get("packages"):  # Only save if we have packages
                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("Debian Atomic 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", []) or []
            new_packages = updated_configs[variant_name].get("packages", []) or []
            
            if new_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")
            else:
                print("  No packages to add")
        
        # 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 Debian Atomic 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 Debian Atomic 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()