Inspect backup archive and return metadata
(self, backup_file)
| 430 | raise Exception(f"Error creating backup: {str(e)}") |
| 431 | |
| 432 | async def inspect_backup(self, backup_file) -> Dict[str, Any]: |
| 433 | """Inspect backup archive and return metadata""" |
| 434 | |
| 435 | # Save uploaded file temporarily |
| 436 | temp_dir = tempfile.mkdtemp() |
| 437 | temp_file = os.path.join(temp_dir, "backup.zip") |
| 438 | |
| 439 | try: |
| 440 | backup_file.save(temp_file) |
| 441 | |
| 442 | with zipfile.ZipFile(temp_file, 'r') as zipf: |
| 443 | # Read metadata |
| 444 | if "metadata.json" not in zipf.namelist(): |
| 445 | raise Exception("Invalid backup file: missing metadata.json") |
| 446 | |
| 447 | metadata_content = zipf.read("metadata.json").decode('utf-8') |
| 448 | metadata = json.loads(metadata_content) |
| 449 | |
| 450 | # Add file list from archive |
| 451 | files_in_archive = [name for name in zipf.namelist() if name != "metadata.json"] |
| 452 | metadata["files_in_archive"] = files_in_archive |
| 453 | |
| 454 | return metadata |
| 455 | |
| 456 | except zipfile.BadZipFile: |
| 457 | raise Exception("Invalid backup file: not a valid zip archive") |
| 458 | except json.JSONDecodeError: |
| 459 | raise Exception("Invalid backup file: corrupted metadata") |
| 460 | finally: |
| 461 | # Cleanup |
| 462 | if os.path.exists(temp_file): |
| 463 | os.remove(temp_file) |
| 464 | if os.path.exists(temp_dir): |
| 465 | os.rmdir(temp_dir) |
| 466 | |
| 467 | async def preview_restore( |
| 468 | self, |