hotspots.py
python
| 1 | """muse hotspots — symbol churn leaderboard. |
| 2 | |
| 3 | Walks the commit history and counts how many commits touched each symbol. |
| 4 | High churn = instability signal. The functions that change most are the |
| 5 | ones that need the most attention — refactoring targets, test coverage gaps, |
| 6 | or domain logic under active evolution. |
| 7 | |
| 8 | Usage:: |
| 9 | |
| 10 | muse hotspots |
| 11 | muse hotspots --top 20 |
| 12 | muse hotspots --kind function --language Python |
| 13 | muse hotspots --from HEAD~30 --to HEAD |
| 14 | |
| 15 | Output:: |
| 16 | |
| 17 | Symbol churn — top 10 most-changed symbols |
| 18 | Commits analysed: 47 |
| 19 | |
| 20 | 1 src/billing.py::compute_invoice_total 12 changes |
| 21 | 2 src/api.py::handle_request 9 changes |
| 22 | 3 src/auth.py::validate_token 7 changes |
| 23 | 4 src/models.py::User.save 5 changes |
| 24 | |
| 25 | High churn = instability signal. |
| 26 | """ |
| 27 | |
| 28 | import json |
| 29 | import logging |
| 30 | import pathlib |
| 31 | |
| 32 | import typer |
| 33 | |
| 34 | from muse.core.errors import ExitCode |
| 35 | from muse.core.repo import require_repo |
| 36 | from muse.core.store import resolve_commit_ref |
| 37 | from muse.plugins.code._query import flat_symbol_ops, language_of, walk_commits_range |
| 38 | |
| 39 | logger = logging.getLogger(__name__) |
| 40 | |
| 41 | app = typer.Typer() |
| 42 | |
| 43 | |
| 44 | def _read_repo_id(root: pathlib.Path) -> str: |
| 45 | return str(json.loads((root / ".muse" / "repo.json").read_text())["repo_id"]) |
| 46 | |
| 47 | |
| 48 | def _read_branch(root: pathlib.Path) -> str: |
| 49 | head_ref = (root / ".muse" / "HEAD").read_text().strip() |
| 50 | return head_ref.removeprefix("refs/heads/").strip() |
| 51 | |
| 52 | |
| 53 | def _collect_churn( |
| 54 | root: pathlib.Path, |
| 55 | to_commit_id: str, |
| 56 | from_commit_id: str | None, |
| 57 | kind_filter: str | None, |
| 58 | language_filter: str | None, |
| 59 | ) -> tuple[dict[str, int], int]: |
| 60 | """Return ``(churn_counts, commits_analysed)``.""" |
| 61 | commits = walk_commits_range(root, to_commit_id, from_commit_id) |
| 62 | counts: dict[str, int] = {} |
| 63 | for commit in commits: |
| 64 | if commit.structured_delta is None: |
| 65 | continue |
| 66 | for op in flat_symbol_ops(commit.structured_delta["ops"]): |
| 67 | addr = op["address"] |
| 68 | if "::" not in addr: |
| 69 | continue |
| 70 | file_path = addr.split("::")[0] |
| 71 | if language_filter and language_of(file_path) != language_filter: |
| 72 | continue |
| 73 | counts[addr] = counts.get(addr, 0) + 1 |
| 74 | return counts, len(commits) |
| 75 | |
| 76 | |
| 77 | @app.callback(invoke_without_command=True) |
| 78 | def hotspots( |
| 79 | ctx: typer.Context, |
| 80 | top: int = typer.Option(20, "--top", "-n", metavar="N", help="Number of symbols to show (default: 20)."), |
| 81 | kind_filter: str | None = typer.Option( |
| 82 | None, "--kind", "-k", metavar="KIND", |
| 83 | help="Restrict to symbols of this kind (function, class, method, …).", |
| 84 | ), |
| 85 | language_filter: str | None = typer.Option( |
| 86 | None, "--language", "-l", metavar="LANG", |
| 87 | help="Restrict to symbols from files of this language.", |
| 88 | ), |
| 89 | from_ref: str | None = typer.Option( |
| 90 | None, "--from", metavar="REF", |
| 91 | help="Exclusive start of the commit range (default: initial commit).", |
| 92 | ), |
| 93 | to_ref: str | None = typer.Option( |
| 94 | None, "--to", metavar="REF", |
| 95 | help="Inclusive end of the commit range (default: HEAD).", |
| 96 | ), |
| 97 | as_json: bool = typer.Option(False, "--json", help="Emit results as JSON."), |
| 98 | ) -> None: |
| 99 | """Show the symbols that change most often — the churn leaderboard. |
| 100 | |
| 101 | Walks the commit history and counts how many commits touched each symbol. |
| 102 | High churn at the function level reveals instability that file-level |
| 103 | metrics miss: a single file may be stable while one specific function |
| 104 | inside it burns. |
| 105 | |
| 106 | Use ``--from`` / ``--to`` to scope the analysis to a sprint, a release, |
| 107 | or any custom range. Use ``--kind function`` to focus on functions only. |
| 108 | """ |
| 109 | root = require_repo() |
| 110 | repo_id = _read_repo_id(root) |
| 111 | branch = _read_branch(root) |
| 112 | |
| 113 | to_commit = resolve_commit_ref(root, repo_id, branch, to_ref) |
| 114 | if to_commit is None: |
| 115 | typer.echo(f"❌ Commit '{to_ref or 'HEAD'}' not found.", err=True) |
| 116 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 117 | |
| 118 | from_commit_id: str | None = None |
| 119 | if from_ref is not None: |
| 120 | from_commit = resolve_commit_ref(root, repo_id, branch, from_ref) |
| 121 | if from_commit is None: |
| 122 | typer.echo(f"❌ Commit '{from_ref}' not found.", err=True) |
| 123 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 124 | from_commit_id = from_commit.commit_id |
| 125 | |
| 126 | counts, total_commits = _collect_churn( |
| 127 | root, to_commit.commit_id, from_commit_id, kind_filter, language_filter |
| 128 | ) |
| 129 | |
| 130 | if not counts: |
| 131 | typer.echo(" (no symbol-level changes found in this range)") |
| 132 | return |
| 133 | |
| 134 | ranked = sorted(counts.items(), key=lambda kv: kv[1], reverse=True)[:top] |
| 135 | |
| 136 | if as_json: |
| 137 | typer.echo(json.dumps( |
| 138 | {"commits_analysed": total_commits, "hotspots": [{"address": a, "changes": c} for a, c in ranked]}, |
| 139 | indent=2, |
| 140 | )) |
| 141 | return |
| 142 | |
| 143 | filters = "" |
| 144 | if kind_filter: |
| 145 | filters += f" kind={kind_filter}" |
| 146 | if language_filter: |
| 147 | filters += f" language={language_filter}" |
| 148 | typer.echo(f"\nSymbol churn — top {len(ranked)} most-changed symbols{filters}") |
| 149 | typer.echo(f"Commits analysed: {total_commits}") |
| 150 | typer.echo("") |
| 151 | |
| 152 | width = len(str(len(ranked))) |
| 153 | for rank, (addr, count) in enumerate(ranked, 1): |
| 154 | label = "change" if count == 1 else "changes" |
| 155 | typer.echo(f" {rank:>{width}} {addr:<60} {count:>4} {label}") |
| 156 | |
| 157 | typer.echo("") |
| 158 | typer.echo("High churn = instability signal. Consider refactoring or adding tests.") |