""" Version Manager for RPA Vision V3 Handles: - Version tracking - Update checking - Package verification - Rollback management """ import os import json import hashlib import shutil from pathlib import Path from datetime import datetime from typing import Dict, List, Optional, Any from dataclasses import dataclass, asdict # Current version CURRENT_VERSION = "3.0.0" VERSION_DATE = "2026-01-19" @dataclass class VersionInfo: """Version information.""" version: str date: str build: str components: Dict[str, str] def to_dict(self) -> Dict[str, Any]: return asdict(self) @classmethod def from_dict(cls, data: Dict) -> 'VersionInfo': return cls(**data) @dataclass class UpdateManifest: """Update package manifest.""" version: str date: str changelog: List[str] min_version: str # Minimum version required to update package_hash: str package_size: int components_updated: List[str] requires_restart: bool = True def to_dict(self) -> Dict[str, Any]: return asdict(self) @classmethod def from_dict(cls, data: Dict) -> 'UpdateManifest': return cls(**data) class VersionManager: """ Manages version information, updates, and rollbacks. Features: - Track current version - Check for updates (from local manifest or remote) - Verify package integrity - Manage rollback versions """ def __init__(self, base_path: Optional[Path] = None): """ Initialize version manager. Args: base_path: Base path for version data """ if base_path is None: base_path = Path(__file__).parent.parent.parent self.base_path = Path(base_path) self.version_dir = self.base_path / "data" / "versions" self.backups_dir = self.version_dir / "backups" self.updates_dir = self.version_dir / "updates" # Create directories self.version_dir.mkdir(parents=True, exist_ok=True) self.backups_dir.mkdir(parents=True, exist_ok=True) self.updates_dir.mkdir(parents=True, exist_ok=True) # Version file self.version_file = self.version_dir / "current_version.json" # Initialize version if not exists if not self.version_file.exists(): self._initialize_version() def _initialize_version(self) -> None: """Initialize version file.""" version_info = VersionInfo( version=CURRENT_VERSION, date=VERSION_DATE, build=datetime.now().strftime("%Y%m%d%H%M%S"), components={ "core": CURRENT_VERSION, "dashboard": CURRENT_VERSION, "vwb_backend": CURRENT_VERSION, "vwb_frontend": CURRENT_VERSION, } ) self._save_version(version_info) def _save_version(self, version_info: VersionInfo) -> None: """Save version info to file.""" with open(self.version_file, 'w') as f: json.dump(version_info.to_dict(), f, indent=2) def get_current_version(self) -> VersionInfo: """Get current version information.""" if self.version_file.exists(): with open(self.version_file, 'r') as f: data = json.load(f) return VersionInfo.from_dict(data) else: self._initialize_version() return self.get_current_version() def get_version_string(self) -> str: """Get simple version string.""" return self.get_current_version().version def check_for_updates(self, manifest_path: Optional[Path] = None) -> Optional[UpdateManifest]: """ Check if updates are available. Args: manifest_path: Path to update manifest (local file) Returns: UpdateManifest if update available, None otherwise """ if manifest_path is None: manifest_path = self.updates_dir / "update_manifest.json" if not manifest_path.exists(): return None try: with open(manifest_path, 'r') as f: data = json.load(f) manifest = UpdateManifest.from_dict(data) current = self.get_current_version() # Compare versions if self._compare_versions(manifest.version, current.version) > 0: # Check minimum version requirement if self._compare_versions(current.version, manifest.min_version) >= 0: return manifest except Exception: pass return None def _compare_versions(self, v1: str, v2: str) -> int: """ Compare two version strings. Returns: 1 if v1 > v2, -1 if v1 < v2, 0 if equal """ parts1 = [int(x) for x in v1.split('.')] parts2 = [int(x) for x in v2.split('.')] # Pad shorter version while len(parts1) < len(parts2): parts1.append(0) while len(parts2) < len(parts1): parts2.append(0) for p1, p2 in zip(parts1, parts2): if p1 > p2: return 1 elif p1 < p2: return -1 return 0 def verify_package(self, package_path: Path, expected_hash: str) -> bool: """ Verify package integrity using SHA-256. Args: package_path: Path to package file expected_hash: Expected SHA-256 hash Returns: True if valid, False otherwise """ if not package_path.exists(): return False sha256 = hashlib.sha256() with open(package_path, 'rb') as f: for chunk in iter(lambda: f.read(8192), b''): sha256.update(chunk) return sha256.hexdigest() == expected_hash def create_backup(self, label: Optional[str] = None) -> Path: """ Create a backup of current version for rollback. Args: label: Optional label for the backup Returns: Path to backup directory """ current = self.get_current_version() timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") if label: backup_name = f"backup_{current.version}_{label}_{timestamp}" else: backup_name = f"backup_{current.version}_{timestamp}" backup_path = self.backups_dir / backup_name backup_path.mkdir(parents=True, exist_ok=True) # Save version info with open(backup_path / "version.json", 'w') as f: json.dump(current.to_dict(), f, indent=2) # Save timestamp with open(backup_path / "backup_info.json", 'w') as f: json.dump({ "created_at": datetime.now().isoformat(), "label": label, "version": current.version, }, f, indent=2) return backup_path def list_backups(self) -> List[Dict[str, Any]]: """List available backups for rollback.""" backups = [] for backup_dir in sorted(self.backups_dir.iterdir(), reverse=True): if backup_dir.is_dir(): info_file = backup_dir / "backup_info.json" if info_file.exists(): with open(info_file, 'r') as f: info = json.load(f) info["path"] = str(backup_dir) info["name"] = backup_dir.name backups.append(info) return backups def get_system_info(self) -> Dict[str, Any]: """Get comprehensive system information.""" current = self.get_current_version() return { "version": current.to_dict(), "system": { "base_path": str(self.base_path), "python_version": f"{os.sys.version_info.major}.{os.sys.version_info.minor}.{os.sys.version_info.micro}", }, "backups_available": len(self.list_backups()), "update_available": self.check_for_updates() is not None, } # Singleton instance _version_manager: Optional[VersionManager] = None def get_version_manager() -> VersionManager: """Get or create the global version manager.""" global _version_manager if _version_manager is None: _version_manager = VersionManager() return _version_manager