| 1 | import zipfile |
| 2 | import json |
| 3 | import os |
| 4 | import tempfile |
| 5 | import datetime |
| 6 | import platform |
| 7 | from typing import List, Dict, Any, Optional |
| 8 | |
| 9 | from pathspec import PathSpec |
| 10 | |
| 11 | from helpers import files, runtime, git, dotenv |
| 12 | from helpers.localization import Localization |
| 13 | from helpers.print_style import PrintStyle |
| 14 | |
| 15 | |
| 16 | class BackupService: |
| 17 | """ |
| 18 | Core backup and restore service for Agent Zero. |
| 19 | |
| 20 | Features: |
| 21 | - JSON-based metadata with user-editable path specifications |
| 22 | - Comprehensive system information collection |
| 23 | - Checksum validation for integrity |
| 24 | - RFC compatibility through existing file helpers |
| 25 | - Git version integration consistent with main application |
| 26 | """ |
| 27 | |
| 28 | def __init__(self): |
| 29 | self.agent_zero_version = self._get_agent_zero_version() |
| 30 | self.agent_zero_root = files.get_abs_path("") # Resolved Agent Zero root |
| 31 | |
| 32 | # Build base paths map for pattern resolution |
| 33 | self.base_paths = { |
| 34 | self.agent_zero_root: self.agent_zero_root, |
| 35 | } |
| 36 | |
| 37 | def get_default_backup_metadata(self) -> Dict[str, Any]: |
| 38 | """Get default backup patterns and metadata""" |
| 39 | timestamp = Localization.get().now_iso() |
| 40 | |
| 41 | default_patterns = self._get_default_patterns() |
| 42 | include_patterns, exclude_patterns = self._parse_patterns(default_patterns) |
| 43 | |
| 44 | return { |
| 45 | "backup_name": f"agent-zero-backup-{timestamp[:10]}", |
| 46 | "include_hidden": True, |
| 47 | "include_patterns": include_patterns, |
| 48 | "exclude_patterns": exclude_patterns, |
| 49 | "backup_config": { |
| 50 | "compression_level": 6, |
| 51 | "integrity_check": True |
| 52 | } |
| 53 | } |
| 54 | |
| 55 | def _get_default_patterns(self) -> str: |
| 56 | """Get default backup patterns with resolved absolute paths. |
| 57 | |
| 58 | Only includes Agent Zero project directory patterns. |
| 59 | """ |
| 60 | # Ensure paths don't have double slashes |
| 61 | agent_root = self.agent_zero_root.rstrip('/') |
| 62 | |
| 63 | return f"""# User data |
| 64 | # All persistent user data is now centralized in /usr for easier backup and restore |
| 65 | {agent_root}/usr/** |
| 66 | !{agent_root}/usr/.time_travel/** |
| 67 | """ |
| 68 | |
| 69 | def _get_agent_zero_version(self) -> str: |
| 70 | """Get current Agent Zero version""" |
| 71 | try: |
| 72 | # Get version from git info (same as run_ui.py) |
| 73 | gitinfo = git.get_git_info() |
| 74 | return gitinfo.get("version", "development") |
| 75 | except Exception: |
| 76 | return "unknown" |
| 77 | |
| 78 | def _resolve_path(self, pattern_path: str) -> str: |
| 79 | """Resolve pattern path to absolute system path (now patterns are already absolute)""" |
| 80 | return pattern_path |
| 81 | |
| 82 | def _unresolve_path(self, abs_path: str) -> str: |
| 83 | """Convert absolute path back to pattern path (now patterns are already absolute)""" |
| 84 | return abs_path |
| 85 | |
| 86 | def _parse_patterns(self, patterns: str) -> tuple[list[str], list[str]]: |
| 87 | """Parse patterns string into include and exclude pattern arrays""" |
| 88 | include_patterns = [] |
| 89 | exclude_patterns = [] |
| 90 | |
| 91 | for line in patterns.split('\n'): |
| 92 | line = line.strip() |
| 93 | if not line or line.startswith('#'): |
| 94 | continue |
| 95 | |
| 96 | if line.startswith('!'): |
| 97 | # Exclude pattern |
| 98 | exclude_patterns.append(line[1:]) # Remove the '!' prefix |
| 99 | else: |
| 100 | # Include pattern |
| 101 | include_patterns.append(line) |
| 102 | |
| 103 | return include_patterns, exclude_patterns |
| 104 | |
| 105 | def _patterns_to_string(self, include_patterns: list[str], exclude_patterns: list[str]) -> str: |
| 106 | """Convert pattern arrays back to patterns string for pathspec processing""" |
| 107 | patterns = [] |
| 108 | |
| 109 | # Add include patterns |
| 110 | for pattern in include_patterns: |
| 111 | patterns.append(pattern) |
| 112 | |
| 113 | # Add exclude patterns with '!' prefix |
| 114 | for pattern in exclude_patterns: |
| 115 | patterns.append(f"!{pattern}") |
| 116 | |
| 117 | return '\n'.join(patterns) |
| 118 | |
| 119 | async def _get_system_info(self) -> Dict[str, Any]: |
| 120 | """Collect system information for metadata""" |
| 121 | import psutil |
| 122 | |
| 123 | try: |
| 124 | return { |
| 125 | "platform": platform.platform(), |
| 126 | "system": platform.system(), |
| 127 | "release": platform.release(), |
| 128 | "version": platform.version(), |
| 129 | "machine": platform.machine(), |
| 130 | "processor": platform.processor(), |
| 131 | "architecture": platform.architecture()[0], |
| 132 | "hostname": platform.node(), |
| 133 | "python_version": platform.python_version(), |
| 134 | "cpu_count": str(psutil.cpu_count()), |
| 135 | "memory_total": str(psutil.virtual_memory().total), |
| 136 | "disk_usage": str(psutil.disk_usage('/').total if os.path.exists('/') else 0) |
| 137 | } |
| 138 | except Exception as e: |
| 139 | return {"error": f"Failed to collect system info: {str(e)}"} |
| 140 | |
| 141 | async def _get_environment_info(self) -> Dict[str, Any]: |
| 142 | """Collect environment information for metadata""" |
| 143 | try: |
| 144 | return { |
| 145 | "user": os.environ.get("USER", "unknown"), |
| 146 | "home": os.environ.get("HOME", "unknown"), |
| 147 | "shell": os.environ.get("SHELL", "unknown"), |
| 148 | "path": os.environ.get("PATH", "")[:200] + "..." if len(os.environ.get("PATH", "")) > 200 else os.environ.get("PATH", ""), |
| 149 | "timezone": Localization.get().get_timezone(), |
| 150 | "working_directory": os.getcwd(), |
| 151 | "agent_zero_root": files.get_abs_path(""), |
| 152 | "runtime_mode": "development" if runtime.is_development() else "production" |
| 153 | } |
| 154 | except Exception as e: |
| 155 | return {"error": f"Failed to collect environment info: {str(e)}"} |
| 156 | |
| 157 | async def _get_backup_author(self) -> str: |
| 158 | """Get backup author/system identifier""" |
| 159 | try: |
| 160 | import getpass |
| 161 | username = getpass.getuser() |
| 162 | hostname = platform.node() |
| 163 | return f"{username}@{hostname}" |
| 164 | except Exception: |
| 165 | return "unknown" |
| 166 | |
| 167 | def _count_directories(self, matched_files: List[Dict[str, Any]]) -> int: |
| 168 | """Count unique directories in file list""" |
| 169 | directories = set() |
| 170 | for file_info in matched_files: |
| 171 | dir_path = os.path.dirname(file_info["path"]) |
| 172 | if dir_path: |
| 173 | directories.add(dir_path) |
| 174 | return len(directories) |
| 175 | |
| 176 | def _get_explicit_patterns(self, include_patterns: List[str]) -> set[str]: |
| 177 | """Extract explicit (non-wildcard) patterns that should always be included""" |
| 178 | explicit_patterns = set() |
| 179 | |
| 180 | for pattern in include_patterns: |
| 181 | # If pattern doesn't contain wildcards, it's explicit |
| 182 | if '*' not in pattern and '?' not in pattern: |
| 183 | # Remove leading slash for comparison |
| 184 | explicit_patterns.add(pattern.lstrip('/')) |
| 185 | |
| 186 | # Also add parent directories as explicit (so hidden dirs can be traversed) |
| 187 | path_parts = pattern.lstrip('/').split('/') |
| 188 | for i in range(1, len(path_parts)): |
| 189 | parent_path = '/'.join(path_parts[:i]) |
| 190 | explicit_patterns.add(parent_path) |
| 191 | |
| 192 | return explicit_patterns |
| 193 | |
| 194 | def _is_explicitly_included(self, file_path: str, explicit_patterns: set[str]) -> bool: |
| 195 | """Check if a file/directory is explicitly included in patterns""" |
| 196 | relative_path = file_path.lstrip('/') |
| 197 | return relative_path in explicit_patterns |
| 198 | |
| 199 | def _translate_patterns(self, patterns: List[str], backup_metadata: Dict[str, Any]) -> List[str]: |
| 200 | """Translate patterns from backed up system to current system. |
| 201 | |
| 202 | Replaces the backed up Agent Zero root path with the current Agent Zero root path |
| 203 | in all patterns if there's an exact match at the beginning of the pattern. |
| 204 | |
| 205 | Args: |
| 206 | patterns: List of patterns from the backed up system |
| 207 | backup_metadata: Backup metadata containing the original agent_zero_root |
| 208 | |
| 209 | Returns: |
| 210 | List of translated patterns for the current system |
| 211 | """ |
| 212 | # Get the backed up agent zero root path from metadata |
| 213 | environment_info = backup_metadata.get("environment_info", {}) |
| 214 | backed_up_agent_root = environment_info.get("agent_zero_root", "") |
| 215 | |
| 216 | # Get current agent zero root path |
| 217 | current_agent_root = self.agent_zero_root |
| 218 | |
| 219 | # If we don't have the backed up root path, return patterns as-is |
| 220 | if not backed_up_agent_root: |
| 221 | return patterns |
| 222 | |
| 223 | # Ensure paths have consistent trailing slash handling |
| 224 | backed_up_agent_root = backed_up_agent_root.rstrip('/') |
| 225 | current_agent_root = current_agent_root.rstrip('/') |
| 226 | |
| 227 | translated_patterns = [] |
| 228 | for pattern in patterns: |
| 229 | # Check if the pattern starts with the backed up agent zero root |
| 230 | if pattern.startswith(backed_up_agent_root + '/') or pattern == backed_up_agent_root: |
| 231 | # Replace the backed up root with the current root |
| 232 | relative_pattern = pattern[len(backed_up_agent_root):].lstrip('/') |
| 233 | if relative_pattern: |
| 234 | translated_pattern = current_agent_root + '/' + relative_pattern |
| 235 | else: |
| 236 | translated_pattern = current_agent_root |
| 237 | translated_patterns.append(translated_pattern) |
| 238 | else: |
| 239 | # Pattern doesn't start with backed up agent root, keep as-is |
| 240 | translated_patterns.append(pattern) |
| 241 | |
| 242 | return translated_patterns |
| 243 | |
| 244 | async def test_patterns(self, metadata: Dict[str, Any], max_files: Optional[int] = 1000) -> List[Dict[str, Any]]: |
| 245 | """Test backup patterns and return list of matched files. |
| 246 | |
| 247 | Pass max_files=None for internal flows that must process the complete |
| 248 | match set, such as backup creation and restore cleanup. |
| 249 | """ |
| 250 | include_patterns = metadata.get("include_patterns", []) |
| 251 | exclude_patterns = metadata.get("exclude_patterns", []) |
| 252 | include_hidden = metadata.get("include_hidden", True) |
| 253 | |
| 254 | # Convert to patterns string for pathspec |
| 255 | patterns_string = self._patterns_to_string(include_patterns, exclude_patterns) |
| 256 | |
| 257 | # Parse patterns using pathspec |
| 258 | pattern_lines = [line.strip() for line in patterns_string.split('\n') if line.strip() and not line.strip().startswith('#')] |
| 259 | |
| 260 | if not pattern_lines: |
| 261 | return [] |
| 262 | |
| 263 | # Get explicit patterns for hidden file handling |
| 264 | explicit_patterns = self._get_explicit_patterns(include_patterns) |
| 265 | |
| 266 | has_limit = max_files is not None |
| 267 | matched_files = [] |
| 268 | processed_count = 0 |
| 269 | |
| 270 | try: |
| 271 | spec = PathSpec.from_lines("gitignore", pattern_lines) |
| 272 | |
| 273 | # Walk through base directories |
| 274 | for base_pattern_path, base_real_path in self.base_paths.items(): |
| 275 | if not os.path.exists(base_real_path): |
| 276 | continue |
| 277 | |
| 278 | for root, dirs, files_list in os.walk(base_real_path): |
| 279 | # Filter hidden directories if not included, BUT allow explicit ones |
| 280 | if not include_hidden: |
| 281 | dirs_to_keep = [] |
| 282 | for d in dirs: |
| 283 | if not d.startswith('.'): |
| 284 | dirs_to_keep.append(d) |
| 285 | else: |
| 286 | # Check if this hidden directory is explicitly included |
| 287 | dir_path = os.path.join(root, d) |
| 288 | pattern_path = self._unresolve_path(dir_path) |
| 289 | if self._is_explicitly_included(pattern_path, explicit_patterns): |
| 290 | dirs_to_keep.append(d) |
| 291 | dirs[:] = dirs_to_keep |
| 292 | |
| 293 | for file in files_list: |
| 294 | if has_limit and processed_count >= max_files: |
| 295 | break |
| 296 | |
| 297 | file_path = os.path.join(root, file) |
| 298 | pattern_path = self._unresolve_path(file_path) |
| 299 | |
| 300 | # Skip hidden files if not included, BUT allow explicit ones |
| 301 | if not include_hidden and file.startswith('.'): |
| 302 | if not self._is_explicitly_included(pattern_path, explicit_patterns): |
| 303 | continue |
| 304 | |
| 305 | # Remove leading slash for pathspec matching |
| 306 | relative_path = pattern_path.lstrip('/') |
| 307 | |
| 308 | if spec.match_file(relative_path): |
| 309 | try: |
| 310 | stat = os.stat(file_path) |
| 311 | matched_files.append({ |
| 312 | "path": pattern_path, |
| 313 | "real_path": file_path, |
| 314 | "size": stat.st_size, |
| 315 | "modified": datetime.datetime.fromtimestamp( |
| 316 | stat.st_mtime, |
| 317 | tz=Localization.get().get_tzinfo(), |
| 318 | ).isoformat(), |
| 319 | "type": "file" |
| 320 | }) |
| 321 | processed_count += 1 |
| 322 | except (OSError, IOError): |
| 323 | # Skip files we can't access |
| 324 | continue |
| 325 | |
| 326 | if has_limit and processed_count >= max_files: |
| 327 | break |
| 328 | |
| 329 | if has_limit and processed_count >= max_files: |
| 330 | break |
| 331 | |
| 332 | except Exception as e: |
| 333 | raise Exception(f"Error processing patterns: {str(e)}") |
| 334 | |
| 335 | return matched_files |
| 336 | |
| 337 | async def create_backup( |
| 338 | self, |
| 339 | include_patterns: List[str], |
| 340 | exclude_patterns: List[str], |
| 341 | include_hidden: bool = True, |
| 342 | backup_name: str = "agent-zero-backup" |
| 343 | ) -> str: |
| 344 | """Create backup archive and return path to created file""" |
| 345 | |
| 346 | # Create metadata for test_patterns |
| 347 | metadata = { |
| 348 | "include_patterns": include_patterns, |
| 349 | "exclude_patterns": exclude_patterns, |
| 350 | "include_hidden": include_hidden |
| 351 | } |
| 352 | |
| 353 | # Get the complete matched file set. Preview and dry-run callers may |
| 354 | # cap their scans for UI responsiveness, but the archive itself must be |
| 355 | # complete. |
| 356 | matched_files = await self.test_patterns(metadata, max_files=None) |
| 357 | |
| 358 | if not matched_files: |
| 359 | raise Exception("No files matched the backup patterns") |
| 360 | |
| 361 | # Create temporary zip file |
| 362 | temp_dir = tempfile.mkdtemp() |
| 363 | zip_path = os.path.join(temp_dir, f"{backup_name}.zip") |
| 364 | |
| 365 | try: |
| 366 | with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf: |
| 367 | # Add comprehensive metadata |
| 368 | metadata = { |
| 369 | # Basic backup information |
| 370 | "agent_zero_version": self.agent_zero_version, |
| 371 | "timestamp": Localization.get().now_iso(), |
| 372 | "backup_name": backup_name, |
| 373 | "include_hidden": include_hidden, |
| 374 | |
| 375 | # Pattern arrays for granular control during restore |
| 376 | "include_patterns": include_patterns, |
| 377 | "exclude_patterns": exclude_patterns, |
| 378 | |
| 379 | # System and environment information |
| 380 | "system_info": await self._get_system_info(), |
| 381 | "environment_info": await self._get_environment_info(), |
| 382 | "backup_author": await self._get_backup_author(), |
| 383 | |
| 384 | # Backup configuration |
| 385 | "backup_config": { |
| 386 | "include_patterns": include_patterns, |
| 387 | "exclude_patterns": exclude_patterns, |
| 388 | "include_hidden": include_hidden, |
| 389 | "compression_level": 6, |
| 390 | "integrity_check": True |
| 391 | }, |
| 392 | |
| 393 | # File information |
| 394 | "files": [ |
| 395 | { |
| 396 | "path": f["path"], |
| 397 | "size": f["size"], |
| 398 | "modified": f["modified"], |
| 399 | "type": "file" |
| 400 | } |
| 401 | for f in matched_files |
| 402 | ], |
| 403 | |
| 404 | # Statistics |
| 405 | "total_files": len(matched_files), |
| 406 | "backup_size": sum(f["size"] for f in matched_files), |
| 407 | "directory_count": self._count_directories(matched_files), |
| 408 | } |
| 409 | |
| 410 | zipf.writestr("metadata.json", json.dumps(metadata, indent=2)) |
| 411 | |
| 412 | # Add files |
| 413 | for file_info in matched_files: |
| 414 | real_path = file_info["real_path"] |
| 415 | archive_path = file_info["path"].lstrip('/') |
| 416 | |
| 417 | try: |
| 418 | if os.path.exists(real_path) and os.path.isfile(real_path): |
| 419 | zipf.write(real_path, archive_path) |
| 420 | except (OSError, IOError) as e: |
| 421 | # Log error but continue with other files |
| 422 | PrintStyle().warning(f"Warning: Could not backup file {real_path}: {e}") |
| 423 | continue |
| 424 | |
| 425 | return zip_path |
| 426 | |
| 427 | except Exception as e: |
| 428 | # Cleanup on error |
| 429 | if os.path.exists(zip_path): |
| 430 | os.remove(zip_path) |
| 431 | raise Exception(f"Error creating backup: {str(e)}") |
| 432 | |
| 433 | async def inspect_backup(self, backup_file) -> Dict[str, Any]: |
| 434 | """Inspect backup archive and return metadata""" |
| 435 | |
| 436 | # Save uploaded file temporarily |
| 437 | temp_dir = tempfile.mkdtemp() |
| 438 | temp_file = os.path.join(temp_dir, "backup.zip") |
| 439 | |
| 440 | try: |
| 441 | backup_file.save(temp_file) |
| 442 | |
| 443 | with zipfile.ZipFile(temp_file, 'r') as zipf: |
| 444 | # Read metadata |
| 445 | if "metadata.json" not in zipf.namelist(): |
| 446 | raise Exception("Invalid backup file: missing metadata.json") |
| 447 | |
| 448 | metadata_content = zipf.read("metadata.json").decode('utf-8') |
| 449 | metadata = json.loads(metadata_content) |
| 450 | |
| 451 | # Add file list from archive |
| 452 | files_in_archive = [name for name in zipf.namelist() if name != "metadata.json"] |
| 453 | metadata["files_in_archive"] = files_in_archive |
| 454 | |
| 455 | return metadata |
| 456 | |
| 457 | except zipfile.BadZipFile: |
| 458 | raise Exception("Invalid backup file: not a valid zip archive") |
| 459 | except json.JSONDecodeError: |
| 460 | raise Exception("Invalid backup file: corrupted metadata") |
| 461 | finally: |
| 462 | # Cleanup |
| 463 | if os.path.exists(temp_file): |
| 464 | os.remove(temp_file) |
| 465 | if os.path.exists(temp_dir): |
| 466 | os.rmdir(temp_dir) |
| 467 | |
| 468 | async def preview_restore( |
| 469 | self, |
| 470 | backup_file, |
| 471 | restore_include_patterns: Optional[List[str]] = None, |
| 472 | restore_exclude_patterns: Optional[List[str]] = None, |
| 473 | overwrite_policy: str = "overwrite", |
| 474 | clean_before_restore: bool = False, |
| 475 | user_edited_metadata: Optional[Dict[str, Any]] = None |
| 476 | ) -> Dict[str, Any]: |
| 477 | """Preview which files would be restored based on patterns""" |
| 478 | |
| 479 | # Save uploaded file temporarily |
| 480 | temp_dir = tempfile.mkdtemp() |
| 481 | temp_file = os.path.join(temp_dir, "backup.zip") |
| 482 | |
| 483 | files_to_restore = [] |
| 484 | skipped_files = [] |
| 485 | |
| 486 | try: |
| 487 | backup_file.save(temp_file) |
| 488 | |
| 489 | with zipfile.ZipFile(temp_file, 'r') as zipf: |
| 490 | # Read backup metadata from archive |
| 491 | original_backup_metadata = {} |
| 492 | if "metadata.json" in zipf.namelist(): |
| 493 | metadata_content = zipf.read("metadata.json").decode('utf-8') |
| 494 | original_backup_metadata = json.loads(metadata_content) |
| 495 | |
| 496 | # Use user-edited metadata if provided, otherwise fall back to original |
| 497 | backup_metadata = user_edited_metadata if user_edited_metadata else original_backup_metadata |
| 498 | |
| 499 | # Get files from archive (excluding metadata files) |
| 500 | archive_files = [name for name in zipf.namelist() |
| 501 | if name not in ["metadata.json", "checksums.json"]] |
| 502 | |
| 503 | # Create pathspec for restore patterns if provided |
| 504 | restore_spec = None |
| 505 | if restore_include_patterns or restore_exclude_patterns: |
| 506 | pattern_lines = [] |
| 507 | if restore_include_patterns: |
| 508 | # Translate patterns from backed up system to current system |
| 509 | translated_include_patterns = self._translate_patterns(restore_include_patterns, original_backup_metadata) |
| 510 | for pattern in translated_include_patterns: |
| 511 | # Remove leading slash for pathspec matching |
| 512 | pattern_lines.append(pattern.lstrip('/')) |
| 513 | if restore_exclude_patterns: |
| 514 | # Translate patterns from backed up system to current system |
| 515 | translated_exclude_patterns = self._translate_patterns(restore_exclude_patterns, original_backup_metadata) |
| 516 | for pattern in translated_exclude_patterns: |
| 517 | # Remove leading slash for pathspec matching |
| 518 | pattern_lines.append(f"!{pattern.lstrip('/')}") |
| 519 | |
| 520 | if pattern_lines: |
| 521 | from pathspec import PathSpec |
| 522 | restore_spec = PathSpec.from_lines("gitignore", pattern_lines) |
| 523 | |
| 524 | # Process each file in archive |
| 525 | for archive_path in archive_files: |
| 526 | # Archive path is already the correct relative path (e.g., "a0/tmp/settings.json") |
| 527 | original_path = archive_path |
| 528 | |
| 529 | # Translate path from backed up system to current system |
| 530 | # Use original metadata for path translation (environment_info needed for this) |
| 531 | target_path = self._translate_restore_path(archive_path, original_backup_metadata) |
| 532 | |
| 533 | # For pattern matching, we need to use the translated path (current system) |
| 534 | # so that patterns like "/home/rafael/a0/data/**" can match files correctly |
| 535 | translated_path_for_matching = target_path.lstrip('/') |
| 536 | |
| 537 | # Check if file matches restore patterns |
| 538 | if restore_spec and not restore_spec.match_file(translated_path_for_matching): |
| 539 | skipped_files.append({ |
| 540 | "archive_path": archive_path, |
| 541 | "original_path": original_path, |
| 542 | "reason": "not_matched_by_pattern" |
| 543 | }) |
| 544 | continue |
| 545 | |
| 546 | # Check file conflict policy for existing files |
| 547 | if os.path.exists(target_path): |
| 548 | if overwrite_policy == "skip": |
| 549 | skipped_files.append({ |
| 550 | "archive_path": archive_path, |
| 551 | "original_path": original_path, |
| 552 | "reason": "file_exists_skip_policy" |
| 553 | }) |
| 554 | continue |
| 555 | |
| 556 | # File will be restored |
| 557 | files_to_restore.append({ |
| 558 | "archive_path": archive_path, |
| 559 | "original_path": original_path, |
| 560 | "target_path": target_path, |
| 561 | "action": "restore" |
| 562 | }) |
| 563 | |
| 564 | # Handle clean before restore if requested |
| 565 | files_to_delete = [] |
| 566 | if clean_before_restore: |
| 567 | # Use user-edited metadata for clean operations so patterns from ACE editor are used |
| 568 | files_to_delete = await self._find_files_to_clean_with_user_metadata(backup_metadata, original_backup_metadata) |
| 569 | |
| 570 | # Combine delete and restore operations for preview |
| 571 | all_operations = files_to_delete + files_to_restore |
| 572 | |
| 573 | return { |
| 574 | "files": all_operations, |
| 575 | "files_to_delete": files_to_delete, |
| 576 | "files_to_restore": files_to_restore, |
| 577 | "skipped_files": skipped_files, |
| 578 | "total_count": len(all_operations), |
| 579 | "delete_count": len(files_to_delete), |
| 580 | "restore_count": len(files_to_restore), |
| 581 | "skipped_count": len(skipped_files), |
| 582 | "backup_metadata": backup_metadata, # Return user-edited metadata |
| 583 | "overwrite_policy": overwrite_policy, |
| 584 | "clean_before_restore": clean_before_restore |
| 585 | } |
| 586 | |
| 587 | except zipfile.BadZipFile: |
| 588 | raise Exception("Invalid backup file: not a valid zip archive") |
| 589 | except json.JSONDecodeError: |
| 590 | raise Exception("Invalid backup file: corrupted metadata") |
| 591 | except Exception as e: |
| 592 | raise Exception(f"Error previewing restore: {str(e)}") |
| 593 | finally: |
| 594 | # Cleanup |
| 595 | if os.path.exists(temp_file): |
| 596 | os.remove(temp_file) |
| 597 | if os.path.exists(temp_dir): |
| 598 | os.rmdir(temp_dir) |
| 599 | |
| 600 | async def restore_backup( |
| 601 | self, |
| 602 | backup_file, |
| 603 | restore_include_patterns: Optional[List[str]] = None, |
| 604 | restore_exclude_patterns: Optional[List[str]] = None, |
| 605 | overwrite_policy: str = "overwrite", |
| 606 | clean_before_restore: bool = False, |
| 607 | user_edited_metadata: Optional[Dict[str, Any]] = None |
| 608 | ) -> Dict[str, Any]: |
| 609 | """Restore files from backup archive""" |
| 610 | |
| 611 | allowed_origins = dotenv.get_dotenv_value("ALLOWED_ORIGINS", "") |
| 612 | dotenv_path = os.path.abspath(dotenv.get_dotenv_file_path()) |
| 613 | |
| 614 | # Save uploaded file temporarily |
| 615 | temp_dir = tempfile.mkdtemp() |
| 616 | temp_file = os.path.join(temp_dir, "backup.zip") |
| 617 | |
| 618 | restored_files = [] |
| 619 | skipped_files = [] |
| 620 | errors = [] |
| 621 | deleted_files = [] |
| 622 | |
| 623 | try: |
| 624 | backup_file.save(temp_file) |
| 625 | |
| 626 | with zipfile.ZipFile(temp_file, 'r') as zipf: |
| 627 | # Read backup metadata from archive |
| 628 | original_backup_metadata = {} |
| 629 | if "metadata.json" in zipf.namelist(): |
| 630 | metadata_content = zipf.read("metadata.json").decode('utf-8') |
| 631 | original_backup_metadata = json.loads(metadata_content) |
| 632 | |
| 633 | # Use user-edited metadata if provided, otherwise fall back to original |
| 634 | backup_metadata = user_edited_metadata if user_edited_metadata else original_backup_metadata |
| 635 | |
| 636 | # Perform clean before restore if requested |
| 637 | if clean_before_restore: |
| 638 | # Use user-edited metadata for clean operations so patterns from ACE editor are used |
| 639 | files_to_delete = await self._find_files_to_clean_with_user_metadata(backup_metadata, original_backup_metadata) |
| 640 | for delete_info in files_to_delete: |
| 641 | try: |
| 642 | real_path = delete_info["real_path"] |
| 643 | if os.path.exists(real_path) and os.path.isfile(real_path): |
| 644 | os.remove(real_path) |
| 645 | deleted_files.append({ |
| 646 | "path": delete_info["path"], |
| 647 | "real_path": real_path, |
| 648 | "action": "deleted", |
| 649 | "reason": "clean_before_restore" |
| 650 | }) |
| 651 | except Exception as e: |
| 652 | errors.append({ |
| 653 | "path": delete_info["path"], |
| 654 | "real_path": delete_info.get("real_path", "unknown"), |
| 655 | "error": f"Failed to delete: {str(e)}" |
| 656 | }) |
| 657 | |
| 658 | # Get files from archive (excluding metadata files) |
| 659 | archive_files = [name for name in zipf.namelist() |
| 660 | if name not in ["metadata.json", "checksums.json"]] |
| 661 | |
| 662 | # Create pathspec for restore patterns if provided |
| 663 | restore_spec = None |
| 664 | if restore_include_patterns or restore_exclude_patterns: |
| 665 | pattern_lines = [] |
| 666 | if restore_include_patterns: |
| 667 | # Translate patterns from backed up system to current system |
| 668 | translated_include_patterns = self._translate_patterns(restore_include_patterns, original_backup_metadata) |
| 669 | for pattern in translated_include_patterns: |
| 670 | # Remove leading slash for pathspec matching |
| 671 | pattern_lines.append(pattern.lstrip('/')) |
| 672 | if restore_exclude_patterns: |
| 673 | # Translate patterns from backed up system to current system |
| 674 | translated_exclude_patterns = self._translate_patterns(restore_exclude_patterns, original_backup_metadata) |
| 675 | for pattern in translated_exclude_patterns: |
| 676 | # Remove leading slash for pathspec matching |
| 677 | pattern_lines.append(f"!{pattern.lstrip('/')}") |
| 678 | |
| 679 | if pattern_lines: |
| 680 | from pathspec import PathSpec |
| 681 | restore_spec = PathSpec.from_lines("gitignore", pattern_lines) |
| 682 | |
| 683 | # Process each file in archive |
| 684 | for archive_path in archive_files: |
| 685 | # Archive path is already the correct relative path (e.g., "a0/tmp/settings.json") |
| 686 | original_path = archive_path |
| 687 | |
| 688 | # Translate path from backed up system to current system |
| 689 | # Use original metadata for path translation (environment_info needed for this) |
| 690 | target_path = self._translate_restore_path(archive_path, original_backup_metadata) |
| 691 | |
| 692 | # For pattern matching, we need to use the translated path (current system) |
| 693 | # so that patterns like "/home/rafael/a0/data/**" can match files correctly |
| 694 | translated_path_for_matching = target_path.lstrip('/') |
| 695 | |
| 696 | # Check if file matches restore patterns |
| 697 | if restore_spec and not restore_spec.match_file(translated_path_for_matching): |
| 698 | skipped_files.append({ |
| 699 | "archive_path": archive_path, |
| 700 | "original_path": original_path, |
| 701 | "reason": "not_matched_by_pattern" |
| 702 | }) |
| 703 | continue |
| 704 | |
| 705 | try: |
| 706 | # Handle overwrite policy |
| 707 | if os.path.exists(target_path): |
| 708 | if overwrite_policy == "skip": |
| 709 | skipped_files.append({ |
| 710 | "archive_path": archive_path, |
| 711 | "original_path": original_path, |
| 712 | "reason": "file_exists_skip_policy" |
| 713 | }) |
| 714 | continue |
| 715 | elif overwrite_policy == "backup": |
| 716 | timestamp = Localization.get().now().strftime('%Y%m%d_%H%M%S') |
| 717 | backup_path = f"{target_path}.backup.{timestamp}" |
| 718 | import shutil |
| 719 | shutil.move(target_path, backup_path) |
| 720 | |
| 721 | # Create target directory if needed |
| 722 | target_dir = os.path.dirname(target_path) |
| 723 | if target_dir: |
| 724 | os.makedirs(target_dir, exist_ok=True) |
| 725 | |
| 726 | # Extract file |
| 727 | import shutil |
| 728 | with zipf.open(archive_path) as source, open(target_path, 'wb') as target: |
| 729 | shutil.copyfileobj(source, target) |
| 730 | |
| 731 | if os.path.abspath(target_path) == dotenv_path: |
| 732 | dotenv.save_dotenv_value( |
| 733 | "ALLOWED_ORIGINS", allowed_origins, reload_env=False |
| 734 | ) |
| 735 | |
| 736 | restored_files.append({ |
| 737 | "archive_path": archive_path, |
| 738 | "original_path": original_path, |
| 739 | "target_path": target_path, |
| 740 | "status": "restored" |
| 741 | }) |
| 742 | |
| 743 | except Exception as e: |
| 744 | errors.append({ |
| 745 | "path": archive_path, |
| 746 | "original_path": original_path, |
| 747 | "error": str(e) |
| 748 | }) |
| 749 | |
| 750 | return { |
| 751 | "restored_files": restored_files, |
| 752 | "deleted_files": deleted_files, |
| 753 | "skipped_files": skipped_files, |
| 754 | "errors": errors, |
| 755 | "backup_metadata": backup_metadata, # Return user-edited metadata |
| 756 | "clean_before_restore": clean_before_restore |
| 757 | } |
| 758 | |
| 759 | except zipfile.BadZipFile: |
| 760 | raise Exception("Invalid backup file: not a valid zip archive") |
| 761 | except json.JSONDecodeError: |
| 762 | raise Exception("Invalid backup file: corrupted metadata") |
| 763 | except Exception as e: |
| 764 | raise Exception(f"Error restoring backup: {str(e)}") |
| 765 | finally: |
| 766 | # Cleanup |
| 767 | if os.path.exists(temp_file): |
| 768 | os.remove(temp_file) |
| 769 | if os.path.exists(temp_dir): |
| 770 | os.rmdir(temp_dir) |
| 771 | |
| 772 | def _translate_restore_path(self, archive_path: str, backup_metadata: Dict[str, Any]) -> str: |
| 773 | """Translate file path from backed up system to current system. |
| 774 | |
| 775 | Replaces the backed up Agent Zero root path with the current Agent Zero root path |
| 776 | if there's an exact match at the beginning of the path. |
| 777 | |
| 778 | Args: |
| 779 | archive_path: Original file path from the archive |
| 780 | backup_metadata: Backup metadata containing the original agent_zero_root |
| 781 | |
| 782 | Returns: |
| 783 | Translated path for the current system |
| 784 | """ |
| 785 | # Get the backed up agent zero root path from metadata |
| 786 | environment_info = backup_metadata.get("environment_info", {}) |
| 787 | backed_up_agent_root = environment_info.get("agent_zero_root", "") |
| 788 | |
| 789 | # Get current agent zero root path |
| 790 | current_agent_root = self.agent_zero_root |
| 791 | |
| 792 | # If we don't have the backed up root path, use original path with leading slash |
| 793 | if not backed_up_agent_root: |
| 794 | return "/" + archive_path.lstrip('/') |
| 795 | |
| 796 | # Ensure paths have consistent trailing slash handling |
| 797 | backed_up_agent_root = backed_up_agent_root.rstrip('/') |
| 798 | current_agent_root = current_agent_root.rstrip('/') |
| 799 | |
| 800 | # Convert archive path to absolute path (add leading slash if missing) |
| 801 | if not archive_path.startswith('/'): |
| 802 | absolute_archive_path = "/" + archive_path |
| 803 | else: |
| 804 | absolute_archive_path = archive_path |
| 805 | |
| 806 | # Check if the archive path starts with the backed up agent zero root |
| 807 | if absolute_archive_path.startswith(backed_up_agent_root + '/') or absolute_archive_path == backed_up_agent_root: |
| 808 | # Replace the backed up root with the current root |
| 809 | relative_path = absolute_archive_path[len(backed_up_agent_root):].lstrip('/') |
| 810 | if relative_path: |
| 811 | translated_path = current_agent_root + '/' + relative_path |
| 812 | else: |
| 813 | translated_path = current_agent_root |
| 814 | return translated_path |
| 815 | else: |
| 816 | # Path doesn't start with backed up agent root, return as-is |
| 817 | return absolute_archive_path |
| 818 | |
| 819 | async def _find_files_to_clean_with_user_metadata(self, user_metadata: Dict[str, Any], original_metadata: Dict[str, Any]) -> List[Dict[str, Any]]: |
| 820 | """Find existing files that match patterns from user-edited metadata for clean operations""" |
| 821 | # Use user-edited patterns for what to clean |
| 822 | user_include_patterns = user_metadata.get("include_patterns", []) |
| 823 | user_exclude_patterns = user_metadata.get("exclude_patterns", []) |
| 824 | include_hidden = user_metadata.get("include_hidden", True) |
| 825 | |
| 826 | if not user_include_patterns: |
| 827 | return [] |
| 828 | |
| 829 | # Translate user-edited patterns from backed up system to current system |
| 830 | # Use original metadata for path translation (environment_info) |
| 831 | translated_include_patterns = self._translate_patterns(user_include_patterns, original_metadata) |
| 832 | translated_exclude_patterns = self._translate_patterns(user_exclude_patterns, original_metadata) |
| 833 | |
| 834 | # Create metadata object for testing translated patterns |
| 835 | metadata = { |
| 836 | "include_patterns": translated_include_patterns, |
| 837 | "exclude_patterns": translated_exclude_patterns, |
| 838 | "include_hidden": include_hidden |
| 839 | } |
| 840 | |
| 841 | # Find existing files that match the translated user-edited patterns |
| 842 | try: |
| 843 | existing_files = await self.test_patterns(metadata, max_files=None) |
| 844 | |
| 845 | # Convert to delete operations format |
| 846 | files_to_delete = [] |
| 847 | for file_info in existing_files: |
| 848 | if os.path.exists(file_info["real_path"]): |
| 849 | files_to_delete.append({ |
| 850 | "path": file_info["path"], |
| 851 | "real_path": file_info["real_path"], |
| 852 | "action": "delete", |
| 853 | "reason": "clean_before_restore" |
| 854 | }) |
| 855 | |
| 856 | return files_to_delete |
| 857 | except Exception: |
| 858 | # If pattern testing fails, return empty list to avoid breaking restore |
| 859 | return [] |