archive.py
python
| 1 | """``muse archive`` — export a snapshot as a portable archive. |
| 2 | |
| 3 | Creates a ``tar.gz`` or ``zip`` archive from any historical snapshot — |
| 4 | HEAD by default. The archive contains only the tracked files (the contents |
| 5 | of ``state/`` at that point in time), making it the canonical way to share |
| 6 | a specific version without exposing the ``.muse/`` internals. |
| 7 | |
| 8 | Usage:: |
| 9 | |
| 10 | muse archive # HEAD snapshot → archive.tar.gz |
| 11 | muse archive --ref feat/audio # branch tip |
| 12 | muse archive --ref a1b2c3d4 # specific commit SHA prefix |
| 13 | muse archive --format zip # zip instead of tar.gz |
| 14 | muse archive --output release-v1.0.zip # custom output path |
| 15 | muse archive --prefix myproject/ # add a directory prefix inside the archive |
| 16 | |
| 17 | The archive is purely content — no Muse metadata (``.muse/``) is included. |
| 18 | This is intentional: archives are for *distribution*, not collaboration. |
| 19 | Use ``muse push`` / ``muse clone`` for distribution with full history. |
| 20 | """ |
| 21 | |
| 22 | from __future__ import annotations |
| 23 | |
| 24 | import io |
| 25 | import logging |
| 26 | import pathlib |
| 27 | import tarfile |
| 28 | import zipfile |
| 29 | from typing import Annotated, Literal |
| 30 | |
| 31 | import typer |
| 32 | |
| 33 | from muse.core.errors import ExitCode |
| 34 | from muse.core.object_store import object_path |
| 35 | from muse.core.repo import require_repo |
| 36 | from muse.core.store import get_head_commit_id, read_commit, read_snapshot, resolve_commit_ref |
| 37 | from muse.core.validation import contain_path, sanitize_display |
| 38 | |
| 39 | logger = logging.getLogger(__name__) |
| 40 | app = typer.Typer(help="Export a snapshot as a portable tar.gz or zip archive.") |
| 41 | |
| 42 | _FORMAT_CHOICES = {"tar.gz", "zip"} |
| 43 | |
| 44 | |
| 45 | def _read_repo_id(root: pathlib.Path) -> str: |
| 46 | import json |
| 47 | return str(json.loads((root / ".muse" / "repo.json").read_text())["repo_id"]) |
| 48 | |
| 49 | |
| 50 | def _read_branch(root: pathlib.Path) -> str: |
| 51 | head = (root / ".muse" / "HEAD").read_text().strip() |
| 52 | return head.removeprefix("refs/heads/").strip() |
| 53 | |
| 54 | |
| 55 | def _build_tar( |
| 56 | root: pathlib.Path, |
| 57 | manifest: dict[str, str], |
| 58 | output_path: pathlib.Path, |
| 59 | prefix: str, |
| 60 | ) -> int: |
| 61 | """Write a tar.gz archive; return file count.""" |
| 62 | count = 0 |
| 63 | with tarfile.open(output_path, "w:gz") as tar: |
| 64 | for rel_path, object_id in sorted(manifest.items()): |
| 65 | obj = object_path(root, object_id) |
| 66 | if not obj.exists(): |
| 67 | logger.warning("⚠️ Missing object %s for %s — skipping", object_id[:12], rel_path) |
| 68 | continue |
| 69 | arcname = (prefix.rstrip("/") + "/" + rel_path) if prefix else rel_path |
| 70 | tar.add(str(obj), arcname=arcname, recursive=False) |
| 71 | count += 1 |
| 72 | return count |
| 73 | |
| 74 | |
| 75 | def _build_zip( |
| 76 | root: pathlib.Path, |
| 77 | manifest: dict[str, str], |
| 78 | output_path: pathlib.Path, |
| 79 | prefix: str, |
| 80 | ) -> int: |
| 81 | """Write a zip archive; return file count.""" |
| 82 | count = 0 |
| 83 | with zipfile.ZipFile(output_path, "w", compression=zipfile.ZIP_DEFLATED) as zf: |
| 84 | for rel_path, object_id in sorted(manifest.items()): |
| 85 | obj = object_path(root, object_id) |
| 86 | if not obj.exists(): |
| 87 | logger.warning("⚠️ Missing object %s for %s — skipping", object_id[:12], rel_path) |
| 88 | continue |
| 89 | arcname = (prefix.rstrip("/") + "/" + rel_path) if prefix else rel_path |
| 90 | zf.write(str(obj), arcname=arcname) |
| 91 | count += 1 |
| 92 | return count |
| 93 | |
| 94 | |
| 95 | @app.callback(invoke_without_command=True) |
| 96 | def archive( |
| 97 | ref: Annotated[ |
| 98 | str | None, |
| 99 | typer.Option("--ref", "-r", help="Branch name or commit SHA (default: HEAD)."), |
| 100 | ] = None, |
| 101 | fmt: Annotated[ |
| 102 | str, |
| 103 | typer.Option("--format", "-f", help="Archive format: tar.gz or zip."), |
| 104 | ] = "tar.gz", |
| 105 | output: Annotated[ |
| 106 | str | None, |
| 107 | typer.Option("--output", "-o", help="Output file path (default: <commit12>.<format>)."), |
| 108 | ] = None, |
| 109 | prefix: Annotated[ |
| 110 | str, |
| 111 | typer.Option("--prefix", help="Add a directory prefix to all paths inside the archive."), |
| 112 | ] = "", |
| 113 | ) -> None: |
| 114 | """Export any historical snapshot as a portable archive. |
| 115 | |
| 116 | The archive contains only tracked files — no ``.muse/`` metadata. It is |
| 117 | the canonical distribution format for a specific version. |
| 118 | |
| 119 | Examples:: |
| 120 | |
| 121 | muse archive # HEAD → <sha12>.tar.gz |
| 122 | muse archive --ref v1.0.0 # tag → v1.0.0.tar.gz |
| 123 | muse archive --format zip --output dist/release.zip |
| 124 | muse archive --prefix myproject/ # all files under myproject/ |
| 125 | """ |
| 126 | if fmt not in _FORMAT_CHOICES: |
| 127 | typer.echo(f"❌ Unknown format '{sanitize_display(fmt)}'. Choose from: {', '.join(sorted(_FORMAT_CHOICES))}") |
| 128 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 129 | |
| 130 | root = require_repo() |
| 131 | repo_id = _read_repo_id(root) |
| 132 | branch = _read_branch(root) |
| 133 | |
| 134 | if ref is None: |
| 135 | commit_id = get_head_commit_id(root, branch) |
| 136 | if not commit_id: |
| 137 | typer.echo("❌ No commits yet on this branch.") |
| 138 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 139 | commit = read_commit(root, commit_id) |
| 140 | else: |
| 141 | commit = resolve_commit_ref(root, repo_id, branch, ref) |
| 142 | |
| 143 | if commit is None: |
| 144 | typer.echo(f"❌ Ref '{sanitize_display(ref or 'HEAD')}' not found.") |
| 145 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 146 | |
| 147 | snapshot = read_snapshot(root, commit.snapshot_id) |
| 148 | if snapshot is None: |
| 149 | typer.echo(f"❌ Snapshot {commit.snapshot_id[:8]} not found.") |
| 150 | raise typer.Exit(code=ExitCode.INTERNAL_ERROR) |
| 151 | |
| 152 | short = commit.commit_id[:12] |
| 153 | out_name = output or f"{short}.{fmt}" |
| 154 | out_path = pathlib.Path(out_name) |
| 155 | |
| 156 | if fmt == "tar.gz": |
| 157 | count = _build_tar(root, snapshot.manifest, out_path, prefix) |
| 158 | else: |
| 159 | count = _build_zip(root, snapshot.manifest, out_path, prefix) |
| 160 | |
| 161 | size_kb = out_path.stat().st_size / 1024 if out_path.exists() else 0 |
| 162 | typer.echo( |
| 163 | f"✅ Archive: {out_path} ({count} file(s), {size_kb:.1f} KiB)\n" |
| 164 | f" Commit: {short} {sanitize_display(commit.message)}" |
| 165 | ) |