DJ-Goanna-Coding/oppo-node

1"""2Automation Script for HuggingFace Space Management3 4This script handles automated synchronization between GitHub and HuggingFace Spaces.5Can be run as a standalone script or scheduled via cron/GitHub Actions.6"""7 8import os9import sys10import argparse11import yaml12from pathlib import Path13from datetime import datetime14from dotenv import load_dotenv15from hf_space_sync import HFSpaceSync16from genesis_boiler import GenesisBoiler17import json18 19load_dotenv()20 21 22class SpaceAutomation:23    """Automate HuggingFace Space management tasks."""24 25    def __init__(self, config_path: str = "config.yaml", verbose: bool = True):26        """27        Initialize automation system.28 29        Args:30            config_path: Path to configuration file31            verbose: Enable verbose output32        """33        self.config_path = config_path34        self.verbose = verbose35        self.config = self._load_config()36        self.log_file = f"automation_log_{datetime.now().strftime('%Y%m%d')}.txt"37 38    def _load_config(self):39        """Load configuration."""40        try:41            with open(self.config_path, 'r') as f:42                return yaml.safe_load(f)43        except FileNotFoundError:44            self.log(f"ERROR: Config file {self.config_path} not found")45            sys.exit(1)46 47    def log(self, message: str):48        """Log a message."""49        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")50        log_msg = f"[{timestamp}] {message}"51 52        if self.verbose:53            print(log_msg)54 55        with open(self.log_file, 'a') as f:56            f.write(log_msg + "\n")57 58    def run_audit(self) -> dict:59        """60        Run file system audit.61 62        Returns:63            Audit results64        """65        self.log("Starting file audit...")66 67        try:68            boiler = GenesisBoiler(self.config_path)69            results = boiler.run_full_audit()70 71            self.log(f"Audit complete: {results['file_count']} files processed")72            self.log(f"Inventory: {results['inventory']}")73            self.log(f"Archive: {results['archive']}")74 75            return results76        except Exception as e:77            self.log(f"ERROR: Audit failed - {e}")78            raise79 80    def sync_all_spaces(self) -> dict:81        """82        Synchronize all configured spaces.83 84        Returns:85            Dictionary of sync results per space86        """87        self.log("Starting space synchronization...")88 89        try:90            hf_sync = HFSpaceSync(self.config_path)91            results = {}92 93            spaces = self.config.get('spaces', {})94            for space_key, space_config in spaces.items():95                if not space_config.get('auto_sync', False):96                    self.log(f"Skipping {space_key} (auto_sync disabled)")97                    continue98 99                space_name = space_config['name']100                self.log(f"Syncing {space_key} ({space_name})...")101 102                try:103                    sync_result = hf_sync.sync_directory(space_name, ".")104                    results[space_key] = sync_result105 106                    self.log(f"✓ {space_key}: {sync_result['uploaded']} uploaded, "107                           f"{sync_result['skipped']} skipped")108                except Exception as e:109                    self.log(f"ERROR: Failed to sync {space_key} - {e}")110                    results[space_key] = {"error": str(e)}111 112            return results113        except Exception as e:114            self.log(f"ERROR: Space sync failed - {e}")115            raise116 117    def sync_specific_space(self, space_name: str, local_dir: str = ".") -> dict:118        """119        Synchronize a specific space.120 121        Args:122            space_name: Name of the space to sync123            local_dir: Local directory to sync124 125        Returns:126            Sync results127        """128        self.log(f"Syncing {space_name} from {local_dir}...")129 130        try:131            hf_sync = HFSpaceSync(self.config_path)132            result = hf_sync.sync_directory(space_name, local_dir)133 134            self.log(f"✓ Sync complete: {result['uploaded']} uploaded, "135                   f"{result['skipped']} skipped")136 137            return result138        except Exception as e:139            self.log(f"ERROR: Sync failed - {e}")140            raise141 142    def backup_to_mapping_inventory(self) -> dict:143        """144        Backup current repository to Mapping-and-Inventory space.145 146        Returns:147            Backup results148        """149        self.log("Creating backup to Mapping-and-Inventory...")150 151        try:152            # First run audit to get current state153            audit_results = self.run_audit()154 155            # Get mapping-inventory space config156            mapping_space = self.config.get('spaces', {}).get('mapping_inventory', {})157            space_name = mapping_space.get('name', 'mapping-and-inventory')158 159            # Sync to mapping-inventory space160            hf_sync = HFSpaceSync(self.config_path)161            sync_result = hf_sync.sync_directory(space_name, ".")162 163            # Also upload the inventory and archive164            if audit_results.get('inventory'):165                hf_sync.upload_files(166                    space_name,167                    audit_results['inventory'],168                    commit_message="Automated inventory backup"169                )170 171            if audit_results.get('archive'):172                hf_sync.upload_files(173                    space_name,174                    audit_results['archive'],175                    commit_message="Automated archive backup"176                )177 178            self.log(f"✓ Backup complete to {space_name}")179 180            return {181                "audit": audit_results,182                "sync": sync_result183            }184        except Exception as e:185            self.log(f"ERROR: Backup failed - {e}")186            raise187 188    def create_summary_report(self, results: dict) -> str:189        """190        Create a summary report of automation results.191 192        Args:193            results: Dictionary of results from automation tasks194 195        Returns:196            Path to summary report file197        """198        report_path = f"automation_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"199 200        report = {201            "timestamp": datetime.now().isoformat(),202            "config_file": self.config_path,203            "results": results,204            "log_file": self.log_file205        }206 207        with open(report_path, 'w') as f:208            json.dump(report, f, indent=2)209 210        self.log(f"Summary report created: {report_path}")211        return report_path212 213 214def main():215    """Main entry point for automation script."""216    parser = argparse.ArgumentParser(217        description="VAMGUARD TITAN - HuggingFace Space Automation"218    )219 220    parser.add_argument(221        'command',222        choices=['audit', 'sync', 'sync-space', 'backup', 'full'],223        help='Command to execute'224    )225 226    parser.add_argument(227        '--space',228        help='Space name (for sync-space command)'229    )230 231    parser.add_argument(232        '--dir',233        default='.',234        help='Local directory to sync (default: current directory)'235    )236 237    parser.add_argument(238        '--config',239        default='config.yaml',240        help='Path to config file (default: config.yaml)'241    )242 243    parser.add_argument(244        '--quiet',245        action='store_true',246        help='Suppress verbose output'247    )248 249    args = parser.parse_args()250 251    # Initialize automation252    automation = SpaceAutomation(args.config, verbose=not args.quiet)253 254    try:255        results = {}256 257        if args.command == 'audit':258            results['audit'] = automation.run_audit()259 260        elif args.command == 'sync':261            results['sync'] = automation.sync_all_spaces()262 263        elif args.command == 'sync-space':264            if not args.space:265                automation.log("ERROR: --space required for sync-space command")266                sys.exit(1)267            results['sync_space'] = automation.sync_specific_space(args.space, args.dir)268 269        elif args.command == 'backup':270            results['backup'] = automation.backup_to_mapping_inventory()271 272        elif args.command == 'full':273            automation.log("Running full automation workflow...")274            results['audit'] = automation.run_audit()275            results['sync'] = automation.sync_all_spaces()276            results['backup'] = automation.backup_to_mapping_inventory()277 278        # Create summary report279        report_path = automation.create_summary_report(results)280        automation.log(f"Automation complete! Report: {report_path}")281 282    except Exception as e:283        automation.log(f"FATAL ERROR: {e}")284        sys.exit(1)285 286 287if __name__ == "__main__":288    main()289