diff.py
python
| 1 | """muse diff — compare working tree against HEAD, or compare two commits.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import json |
| 6 | import logging |
| 7 | import pathlib |
| 8 | |
| 9 | import typer |
| 10 | |
| 11 | from muse.core.errors import ExitCode |
| 12 | from muse.core.repo import require_repo |
| 13 | from muse.core.store import get_commit_snapshot_manifest, get_head_snapshot_manifest, resolve_commit_ref |
| 14 | from muse.domain import DomainOp, SnapshotManifest |
| 15 | from muse.plugins.registry import read_domain, resolve_plugin |
| 16 | |
| 17 | logger = logging.getLogger(__name__) |
| 18 | |
| 19 | app = typer.Typer() |
| 20 | |
| 21 | |
| 22 | def _read_branch(root: pathlib.Path) -> str: |
| 23 | head_ref = (root / ".muse" / "HEAD").read_text().strip() |
| 24 | return head_ref.removeprefix("refs/heads/").strip() |
| 25 | |
| 26 | |
| 27 | def _read_repo_id(root: pathlib.Path) -> str: |
| 28 | return str(json.loads((root / ".muse" / "repo.json").read_text())["repo_id"]) |
| 29 | |
| 30 | |
| 31 | def _print_structured_delta(ops: list[DomainOp]) -> int: |
| 32 | """Print a structured delta op-by-op. Returns the number of ops printed. |
| 33 | |
| 34 | Each branch checks ``op["op"]`` directly so mypy can narrow the |
| 35 | TypedDict union to the specific subtype before accessing its fields. |
| 36 | """ |
| 37 | for op in ops: |
| 38 | if op["op"] == "insert": |
| 39 | typer.echo(f"A {op['address']}") |
| 40 | elif op["op"] == "delete": |
| 41 | typer.echo(f"D {op['address']}") |
| 42 | elif op["op"] == "replace": |
| 43 | typer.echo(f"M {op['address']}") |
| 44 | elif op["op"] == "move": |
| 45 | typer.echo( |
| 46 | f"R {op['address']} ({op['from_position']} → {op['to_position']})" |
| 47 | ) |
| 48 | elif op["op"] == "patch": |
| 49 | typer.echo(f"M {op['address']}") |
| 50 | if op["child_summary"]: |
| 51 | typer.echo(f" └─ {op['child_summary']}") |
| 52 | return len(ops) |
| 53 | |
| 54 | |
| 55 | @app.callback(invoke_without_command=True) |
| 56 | def diff( |
| 57 | ctx: typer.Context, |
| 58 | commit_a: str | None = typer.Argument(None, help="Base commit ID (default: HEAD)."), |
| 59 | commit_b: str | None = typer.Argument(None, help="Target commit ID (default: working tree)."), |
| 60 | stat: bool = typer.Option(False, "--stat", help="Show summary statistics only."), |
| 61 | ) -> None: |
| 62 | """Compare working tree against HEAD, or compare two commits.""" |
| 63 | root = require_repo() |
| 64 | repo_id = _read_repo_id(root) |
| 65 | branch = _read_branch(root) |
| 66 | domain = read_domain(root) |
| 67 | plugin = resolve_plugin(root) |
| 68 | |
| 69 | if commit_a is None: |
| 70 | base_snap = SnapshotManifest( |
| 71 | files=get_head_snapshot_manifest(root, repo_id, branch) or {}, |
| 72 | domain=domain, |
| 73 | ) |
| 74 | target_snap = plugin.snapshot(root / "muse-work") |
| 75 | elif commit_b is None: |
| 76 | base_snap = SnapshotManifest( |
| 77 | files=get_head_snapshot_manifest(root, repo_id, branch) or {}, |
| 78 | domain=domain, |
| 79 | ) |
| 80 | target_snap = SnapshotManifest( |
| 81 | files=get_commit_snapshot_manifest(root, commit_a) or {}, |
| 82 | domain=domain, |
| 83 | ) |
| 84 | else: |
| 85 | base_snap = SnapshotManifest( |
| 86 | files=get_commit_snapshot_manifest(root, commit_a) or {}, |
| 87 | domain=domain, |
| 88 | ) |
| 89 | target_snap = SnapshotManifest( |
| 90 | files=get_commit_snapshot_manifest(root, commit_b) or {}, |
| 91 | domain=domain, |
| 92 | ) |
| 93 | |
| 94 | delta = plugin.diff(base_snap, target_snap, repo_root=root) |
| 95 | |
| 96 | if stat: |
| 97 | typer.echo(delta["summary"] if delta["ops"] else "No differences.") |
| 98 | return |
| 99 | |
| 100 | changed = _print_structured_delta(delta["ops"]) |
| 101 | |
| 102 | if changed == 0: |
| 103 | typer.echo("No differences.") |
| 104 | else: |
| 105 | typer.echo(f"\n{delta['summary']}") |