diff.py
python
| 1 | """muse diff — compare working tree against HEAD, or compare two commits.""" |
| 2 | from __future__ import annotations |
| 3 | |
| 4 | import json |
| 5 | import logging |
| 6 | import pathlib |
| 7 | |
| 8 | import typer |
| 9 | |
| 10 | from muse.core.errors import ExitCode |
| 11 | from muse.core.repo import require_repo |
| 12 | from muse.core.store import get_commit_snapshot_manifest, get_head_snapshot_manifest, resolve_commit_ref |
| 13 | from muse.domain import SnapshotManifest |
| 14 | from muse.plugins.registry import read_domain, resolve_plugin |
| 15 | |
| 16 | logger = logging.getLogger(__name__) |
| 17 | |
| 18 | app = typer.Typer() |
| 19 | |
| 20 | |
| 21 | def _read_branch(root: pathlib.Path) -> str: |
| 22 | head_ref = (root / ".muse" / "HEAD").read_text().strip() |
| 23 | return head_ref.removeprefix("refs/heads/").strip() |
| 24 | |
| 25 | |
| 26 | def _read_repo_id(root: pathlib.Path) -> str: |
| 27 | return str(json.loads((root / ".muse" / "repo.json").read_text())["repo_id"]) |
| 28 | |
| 29 | |
| 30 | def _print_delta(added: list[str], removed: list[str], modified: list[str]) -> int: |
| 31 | """Print a file-level delta. Returns number of changed files.""" |
| 32 | for p in sorted(added): |
| 33 | typer.echo(f"A {p}") |
| 34 | for p in sorted(removed): |
| 35 | typer.echo(f"D {p}") |
| 36 | for p in sorted(modified): |
| 37 | typer.echo(f"M {p}") |
| 38 | return len(added) + len(removed) + len(modified) |
| 39 | |
| 40 | |
| 41 | @app.callback(invoke_without_command=True) |
| 42 | def diff( |
| 43 | ctx: typer.Context, |
| 44 | commit_a: str | None = typer.Argument(None, help="Base commit ID (default: HEAD)."), |
| 45 | commit_b: str | None = typer.Argument(None, help="Target commit ID (default: working tree)."), |
| 46 | stat: bool = typer.Option(False, "--stat", help="Show summary statistics only."), |
| 47 | ) -> None: |
| 48 | """Compare working tree against HEAD, or compare two commits.""" |
| 49 | root = require_repo() |
| 50 | repo_id = _read_repo_id(root) |
| 51 | branch = _read_branch(root) |
| 52 | domain = read_domain(root) |
| 53 | plugin = resolve_plugin(root) |
| 54 | |
| 55 | if commit_a is None: |
| 56 | base_snap = SnapshotManifest( |
| 57 | files=get_head_snapshot_manifest(root, repo_id, branch) or {}, |
| 58 | domain=domain, |
| 59 | ) |
| 60 | target_snap = plugin.snapshot(root / "muse-work") |
| 61 | elif commit_b is None: |
| 62 | base_snap = SnapshotManifest( |
| 63 | files=get_head_snapshot_manifest(root, repo_id, branch) or {}, |
| 64 | domain=domain, |
| 65 | ) |
| 66 | target_snap = SnapshotManifest( |
| 67 | files=get_commit_snapshot_manifest(root, commit_a) or {}, |
| 68 | domain=domain, |
| 69 | ) |
| 70 | else: |
| 71 | base_snap = SnapshotManifest( |
| 72 | files=get_commit_snapshot_manifest(root, commit_a) or {}, |
| 73 | domain=domain, |
| 74 | ) |
| 75 | target_snap = SnapshotManifest( |
| 76 | files=get_commit_snapshot_manifest(root, commit_b) or {}, |
| 77 | domain=domain, |
| 78 | ) |
| 79 | |
| 80 | delta = plugin.diff(base_snap, target_snap) |
| 81 | changed = _print_delta(delta["added"], delta["removed"], delta["modified"]) |
| 82 | |
| 83 | if changed == 0: |
| 84 | typer.echo("No differences.") |
| 85 | elif stat: |
| 86 | typer.echo(f"\n{changed} file(s) changed") |