Manages zip-based snapshots of a code directory.
| 25 | # --------------------------------------------------------------------------- |
| 26 | |
| 27 | class CodeSnapshotManager: |
| 28 | """Manages zip-based snapshots of a code directory.""" |
| 29 | |
| 30 | def __init__(self, code_dir: Path) -> None: |
| 31 | self.code_dir = code_dir.resolve() |
| 32 | self.backup_dir = self.code_dir / ".code_backups" |
| 33 | self.backup_dir.mkdir(parents=True, exist_ok=True) |
| 34 | |
| 35 | # -- create -------------------------------------------------------- |
| 36 | |
| 37 | def create_snapshot(self, label: str = "") -> str: |
| 38 | """Create a zip snapshot. Returns the snapshot id.""" |
| 39 | ts = datetime.now().strftime("%Y%m%d_%H%M%S") |
| 40 | safe_label = "".join(c if c.isalnum() or c in "_-" else "_" for c in label)[:40] |
| 41 | snap_id = f"{ts}_{safe_label}" if safe_label else ts |
| 42 | zip_path = self.backup_dir / f"{snap_id}.zip" |
| 43 | |
| 44 | with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf: |
| 45 | for fp in sorted(self.code_dir.rglob("*")): |
| 46 | if fp.is_file() and ".code_backups" not in fp.parts: |
| 47 | arcname = fp.relative_to(self.code_dir) |
| 48 | zf.write(fp, arcname) |
| 49 | |
| 50 | return snap_id |
| 51 | |
| 52 | # -- list ---------------------------------------------------------- |
| 53 | |
| 54 | def list_snapshots(self) -> list[dict[str, str]]: |
| 55 | """Return list of ``{"id": ..., "time": ..., "path": ...}``.""" |
| 56 | snaps: list[dict[str, str]] = [] |
| 57 | for zp in sorted(self.backup_dir.glob("*.zip")): |
| 58 | snap_id = zp.stem |
| 59 | parts = snap_id.split("_", 2) |
| 60 | if len(parts) >= 2: |
| 61 | time_str = f"{parts[0][:4]}-{parts[0][4:6]}-{parts[0][6:8]} {parts[1][:2]}:{parts[1][2:4]}:{parts[1][4:6]}" |
| 62 | else: |
| 63 | time_str = snap_id |
| 64 | snaps.append({"id": snap_id, "time": time_str, "path": str(zp)}) |
| 65 | return snaps |
| 66 | |
| 67 | # -- rollback ------------------------------------------------------ |
| 68 | |
| 69 | def rollback(self, snapshot_id: str) -> bool: |
| 70 | """Restore code from a snapshot. Returns True on success.""" |
| 71 | zip_path = self.backup_dir / f"{snapshot_id}.zip" |
| 72 | if not zip_path.exists(): |
| 73 | return False |
| 74 | |
| 75 | # Remove current files (except .code_backups) |
| 76 | for fp in list(self.code_dir.rglob("*")): |
| 77 | if fp.is_file() and ".code_backups" not in fp.parts: |
| 78 | fp.unlink() |
| 79 | |
| 80 | # Extract |
| 81 | with zipfile.ZipFile(zip_path, "r") as zf: |
| 82 | zf.extractall(self.code_dir) |
| 83 | return True |
| 84 |