aaronrene / gabriel-muse public
symbols.py python
487 lines 17.1 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 32 days ago
1 """muse code symbols -- list every semantic symbol in a snapshot.
2
3 Muse tracks the semantic interior of every source file -- the full symbol graph
4 the code plugin builds at commit time -- giving each function, class, method,
5 and variable a stable, content-addressed identity independent of line numbers
6 or formatting.
7
8 Output (default -- human-readable table)::
9
10 src/utils.py
11 fn calculate_total line 12
12 fn _validate_amount line 28
13 class Invoice line 45
14 method Invoice.to_dict line 52
15 method Invoice.from_dict line 61
16
17 src/models.py
18 class User line 8
19 method User.__init__ line 10
20 method User.save line 19
21
22 12 symbols across 2 files (Python: 12)
23
24 Flags::
25
26 --commit <ref>
27 Inspect a specific commit instead of the working tree.
28 Accepts a full or abbreviated commit SHA, a branch name, or HEAD~N.
29
30 --kind <kind>
31 Filter to a specific symbol kind:
32 function, async_function, class, method, async_method,
33 variable, import, section, rule.
34
35 --file <path>
36 Show symbols from a single file. Accepts an exact path or a
37 unique suffix (e.g. "billing.py" matches "src/billing.py").
38
39 --language <lang>
40 Show symbols from files of this language only (case-insensitive,
41 e.g. "python", "TypeScript", "go").
42
43 --count
44 Print only the total symbol count and per-language breakdown.
45
46 --hashes
47 Include content hashes alongside each symbol.
48
49 --json
50 Emit a structured JSON object for tooling integration::
51
52 {
53 "source_ref": "a1b2c3d4",
54 "working_tree": true,
55 "total_symbols": 12,
56 "results": [
57 {"address": "src/utils.py::calculate_total",
58 "file": "src/utils.py", "kind": "function", ...}
59 ]
60 }
61 """
62
63 from __future__ import annotations
64
65 import argparse
66 import json
67 import logging
68 import pathlib
69 import sys
70
71 from muse.core.errors import ExitCode
72 from muse.core.repo import read_repo_id, require_repo
73 from muse.core.store import (
74 Manifest,
75 get_commit_snapshot_manifest,
76 read_current_branch,
77 resolve_commit_ref,
78 )
79 from muse.plugins.code._query import language_of, symbols_for_snapshot
80 from muse.plugins.code.ast_parser import SymbolTree
81
82 type _SymbolTreeMap = dict[str, SymbolTree]
83 type _CounterMap = dict[str, int]
84 type _KindDisplay = dict[str, tuple[str, list[str]]]
85
86 logger = logging.getLogger(__name__)
87
88 # ---------------------------------------------------------------------------
89 # ANSI colour helpers — only emitted when stdout is a TTY.
90 # ---------------------------------------------------------------------------
91
92 _RESET = "\033[0m"
93 _BOLD = "\033[1m"
94 _DIM = "\033[2m"
95 _CYAN = "\033[36m"
96 _YELLOW = "\033[33m"
97 _BLUE = "\033[34m"
98 _GREEN = "\033[32m"
99 _MAGENTA = "\033[35m"
100
101
102 def _c(text: str, *codes: str, tty: bool) -> str:
103 """Wrap *text* in ANSI *codes* when *tty* is True."""
104 if not tty:
105 return text
106 return "".join(codes) + text + _RESET
107
108
109 # Maps symbol kind → (short icon, ANSI colour codes).
110 _KIND_DISPLAY: _KindDisplay = {
111 "function": ("fn", [_BLUE]),
112 "async_function": ("fn~", [_BLUE, _DIM]),
113 "class": ("class", [_YELLOW, _BOLD]),
114 "method": ("method", [_CYAN]),
115 "async_method": ("method~", [_CYAN, _DIM]),
116 "variable": ("var", [_DIM]),
117 "import": ("import", [_DIM]),
118 "section": ("section", [_GREEN]),
119 "rule": ("rule", [_MAGENTA]),
120 }
121
122 _VALID_KINDS: frozenset[str] = frozenset(_KIND_DISPLAY)
123
124 # ---------------------------------------------------------------------------
125 # Language helpers
126 # ---------------------------------------------------------------------------
127
128 # Canonical map: lowercase language name → display name.
129 # Built from _SUFFIX_LANG in _query.py to stay in sync.
130 from muse.plugins.code._query import _SUFFIX_LANG # noqa: E402 (module-level import)
131 from muse.core.validation import sanitize_display
132
133 _LANG_CANONICAL: Manifest = {lang.lower(): lang for lang in set(_SUFFIX_LANG.values())}
134
135
136 def _normalise_language(lang: str) -> str:
137 """Return the canonical capitalisation for *lang*, or *lang* unchanged."""
138 return _LANG_CANONICAL.get(lang.strip().lower(), lang.strip())
139
140
141 # ---------------------------------------------------------------------------
142 # File-filter helpers
143 # ---------------------------------------------------------------------------
144
145
146 def _file_matches(file_path: str, file_filter: str) -> bool:
147 """Return True if *file_path* equals or uniquely ends with *file_filter*.
148
149 Allows passing ``"billing.py"`` to match ``"src/billing.py"`` without
150 requiring callers to know the full directory prefix. Uses a separator
151 anchor (``/``) to prevent ``y.py`` matching ``billy.py``.
152 """
153 if file_path == file_filter:
154 return True
155 normalized = file_filter.replace("\\", "/")
156 return file_path.endswith("/" + normalized)
157
158
159 def _resolve_file_filter(
160 file_filter: str,
161 manifest: Manifest,
162 ) -> str | None:
163 """Resolve *file_filter* to the exact manifest path, or ``None`` on ambiguity/miss.
164
165 Prints a helpful message to stderr and raises ``SystemExit`` on ambiguity.
166 Returns ``None`` when there is no match (caller emits "no symbols found").
167 """
168 matching = [p for p in sorted(manifest) if _file_matches(p, file_filter)]
169 if len(matching) == 1:
170 return matching[0]
171 if len(matching) > 1:
172 print(
173 f"❌ '{file_filter}' is ambiguous — matches {len(matching)} files. "
174 "Use a more specific path:",
175 file=sys.stderr,
176 )
177 for m in matching[:10]:
178 print(f" {m}", file=sys.stderr)
179 if len(matching) > 10:
180 print(f" … and {len(matching) - 10} more", file=sys.stderr)
181 raise SystemExit(ExitCode.USER_ERROR)
182 return None # no match — caller handles the empty result
183
184
185 # ---------------------------------------------------------------------------
186 # Repository helpers
187 # ---------------------------------------------------------------------------
188
189
190
191 # ---------------------------------------------------------------------------
192 # Output helpers
193 # ---------------------------------------------------------------------------
194
195
196 def _lang_counts(symbol_map: _SymbolTreeMap) -> _CounterMap:
197 """Return a language-name → symbol-count mapping for *symbol_map*."""
198 counts: _CounterMap = {}
199 for file_path, tree in symbol_map.items():
200 lang = language_of(file_path)
201 counts[lang] = counts.get(lang, 0) + len(tree)
202 return counts
203
204
205 def _print_human(
206 symbol_map: _SymbolTreeMap,
207 show_hashes: bool,
208 tty: bool,
209 ) -> None:
210 """Render symbol_map as a human-readable, optionally coloured table."""
211 if not symbol_map:
212 print(" (no semantic symbols found)")
213 return
214
215 total = 0
216 for file_path, tree in symbol_map.items():
217 total += len(tree)
218 print(f"\n{_c(sanitize_display(file_path), _BOLD, tty=tty)}")
219 for _addr, rec in sorted(tree.items(), key=lambda kv: kv[1]["lineno"]):
220 kind = rec["kind"]
221 icon, colour_codes = _KIND_DISPLAY.get(kind, (kind, []))
222 name = rec["qualified_name"]
223 lineno = rec["lineno"]
224 icon_str = _c(f"{icon:<10}", *colour_codes, tty=tty)
225 name_str = f"{name:<40}"
226 line_str = _c(f"line {lineno:>4}", _DIM, tty=tty)
227 hash_suffix = (
228 _c(f" {rec['content_id'][:8]}..", _DIM, tty=tty)
229 if show_hashes
230 else ""
231 )
232 print(f" {icon_str} {name_str} {line_str}{hash_suffix}")
233
234 counts = _lang_counts(symbol_map)
235 lang_str = ", ".join(f"{lang}: {count:,}" for lang, count in sorted(counts.items()))
236 sym_word = "symbol" if total == 1 else "symbols"
237 file_word = "file" if len(symbol_map) == 1 else "files"
238 print(
239 f"\n{_c(f'{total:,}', _BOLD, tty=tty)} {sym_word} across "
240 f"{len(symbol_map):,} {file_word} ({lang_str})"
241 )
242
243
244 def _emit_json(
245 symbol_map: _SymbolTreeMap,
246 source_ref: str,
247 working_tree: bool,
248 ) -> None:
249 """Emit the symbol map as a structured JSON object.
250
251 Schema::
252
253 {
254 "source_ref": "<commit-sha8>", // or "working-tree"
255 "working_tree": true | false,
256 "total_symbols": <int>,
257 "results": [
258 {
259 "address": "<file>::<symbol>",
260 "kind": "function" | "class" | ...,
261 "name": "<bare name>",
262 "qualified_name": "<qualified.name>",
263 "file": "<file_path>",
264 "lineno": <int>,
265 "end_lineno": <int>,
266 "content_id": "<sha256>",
267 "body_hash": "<sha256>",
268 "signature_id": "<sha256>"
269 }
270 ]
271 }
272 """
273 results: list[dict[str, str | int]] = []
274 for file_path, tree in symbol_map.items():
275 for addr, rec in sorted(tree.items(), key=lambda kv: kv[1]["lineno"]):
276 results.append({
277 "address": addr,
278 "kind": rec["kind"],
279 "name": rec["name"],
280 "qualified_name": rec["qualified_name"],
281 "file": file_path,
282 "lineno": rec["lineno"],
283 "end_lineno": rec["end_lineno"],
284 "content_id": rec["content_id"],
285 "body_hash": rec["body_hash"],
286 "signature_id": rec["signature_id"],
287 })
288 print(json.dumps({
289 "source_ref": source_ref,
290 "working_tree": working_tree,
291 "total_symbols": len(results),
292 "results": results,
293 }, indent=2))
294
295
296 # ---------------------------------------------------------------------------
297 # Argument parser registration
298 # ---------------------------------------------------------------------------
299
300
301 def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None:
302 """Register the symbols subcommand."""
303 parser = subparsers.add_parser(
304 "symbols",
305 help="List every semantic symbol (function, class, method…) in a snapshot.",
306 description=__doc__,
307 formatter_class=argparse.RawDescriptionHelpFormatter,
308 )
309 parser.add_argument(
310 "--commit", "-c",
311 dest="ref",
312 default=None,
313 metavar="REF",
314 help="Commit ID or branch to inspect (default: working tree).",
315 )
316 parser.add_argument(
317 "--kind", "-k",
318 dest="kind_filter",
319 default=None,
320 metavar="KIND",
321 help=(
322 "Filter to symbols of a specific kind "
323 "(function, async_function, class, method, async_method, "
324 "variable, import, section, rule)."
325 ),
326 )
327 parser.add_argument(
328 "--file", "-f",
329 dest="file_filter",
330 default=None,
331 metavar="PATH",
332 help=(
333 "Show symbols from a single file. Accepts an exact path or a "
334 "unique path suffix (e.g. 'billing.py' matches 'src/billing.py')."
335 ),
336 )
337 parser.add_argument(
338 "--language", "-l",
339 dest="language_filter",
340 default=None,
341 metavar="LANG",
342 help="Show symbols from files of this language only (case-insensitive).",
343 )
344 parser.add_argument(
345 "--hashes",
346 dest="show_hashes",
347 action="store_true",
348 help="Include content hashes in the output.",
349 )
350
351 output_group = parser.add_mutually_exclusive_group()
352 output_group.add_argument(
353 "--count",
354 dest="count_only",
355 action="store_true",
356 help="Print only the total symbol count and language breakdown.",
357 )
358 output_group.add_argument(
359 "--json",
360 dest="as_json",
361 action="store_true",
362 help="Emit the full symbol table as JSON.",
363 )
364
365 parser.set_defaults(func=run)
366
367
368 # ---------------------------------------------------------------------------
369 # Command entry point
370 # ---------------------------------------------------------------------------
371
372
373 def run(args: argparse.Namespace) -> None:
374 """List every semantic symbol (function, class, method…) in a snapshot.
375
376 Unlike ``git grep`` or ``ctags``, ``muse code symbols`` reads the semantic
377 symbol graph produced by the domain plugin's AST analysis — stable,
378 content-addressed identities for every symbol, independent of line numbers
379 or formatting.
380
381 When ``--commit`` is omitted the command reads directly from the working
382 tree, reflecting edits that have not yet been committed. Pass
383 ``--commit HEAD`` (or any ref) to inspect a historical snapshot instead.
384
385 Use ``--kind`` and ``--file`` to narrow the output. Use ``--json`` for
386 tooling and agent integration.
387 """
388 ref: str | None = args.ref
389 kind_filter: str | None = args.kind_filter
390 file_filter: str | None = args.file_filter
391 language_filter: str | None = args.language_filter
392 count_only: bool = args.count_only
393 show_hashes: bool = args.show_hashes
394 as_json: bool = args.as_json
395 tty: bool = sys.stdout.isatty()
396
397 # ── Input validation ──────────────────────────────────────────────────────
398
399 if kind_filter is not None and kind_filter not in _VALID_KINDS:
400 valid = ", ".join(sorted(_VALID_KINDS))
401 print(f"❌ Unknown kind '{kind_filter}'. Valid kinds: {valid}", file=sys.stderr)
402 raise SystemExit(ExitCode.USER_ERROR)
403
404 if language_filter is not None:
405 language_filter = _normalise_language(language_filter)
406
407 # ── Repo / commit resolution ──────────────────────────────────────────────
408
409 root = require_repo()
410 repo_id = read_repo_id(root)
411 branch = read_current_branch(root)
412
413 commit = resolve_commit_ref(root, repo_id, branch, ref)
414 if commit is None:
415 label = ref or "HEAD"
416 print(f"❌ Commit '{label}' not found.", file=sys.stderr)
417 raise SystemExit(ExitCode.USER_ERROR)
418
419 manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {}
420 if not manifest:
421 print(
422 f"❌ Snapshot for commit {commit.commit_id[:8]} has no files.",
423 file=sys.stderr,
424 )
425 raise SystemExit(ExitCode.USER_ERROR)
426
427 # ── Working-tree vs object-store mode / file-filter resolution ───────────
428
429 working_tree = ref is None # True when no --commit was given
430 workdir = root if working_tree else None
431
432 resolved_file_filter = file_filter
433 if file_filter is not None:
434 found = _resolve_file_filter(file_filter, manifest)
435 if found is not None:
436 resolved_file_filter = found
437 elif working_tree:
438 # File not in HEAD manifest — may be a new uncommitted file.
439 # Inject a synthetic manifest entry so symbols_for_snapshot can
440 # parse it directly from the working directory.
441 candidate = root / file_filter
442 if candidate.is_file():
443 manifest = {file_filter: ""}
444 resolved_file_filter = file_filter
445
446 # ── Symbol extraction ─────────────────────────────────────────────────────
447
448 symbol_map = symbols_for_snapshot(
449 root,
450 manifest,
451 kind_filter=kind_filter,
452 file_filter=resolved_file_filter,
453 language_filter=language_filter,
454 workdir=workdir,
455 )
456
457 # ── Source reference label ────────────────────────────────────────────────
458
459 if working_tree:
460 source_ref = "working-tree"
461 else:
462 source_ref = commit.commit_id[:8]
463
464 # ── Output ────────────────────────────────────────────────────────────────
465
466 if count_only:
467 total = sum(len(t) for t in symbol_map.values())
468 counts = _lang_counts(symbol_map)
469 lang_str = ", ".join(f"{lang}: {count:,}" for lang, count in sorted(counts.items()))
470 sym_word = "symbol" if total == 1 else "symbols"
471 print(f"{total:,} {sym_word} ({lang_str})")
472 return
473
474 if as_json:
475 _emit_json(symbol_map, source_ref=source_ref, working_tree=working_tree)
476 return
477
478 if working_tree:
479 header = (
480 f'working tree '
481 f'(HEAD {commit.commit_id[:8]} "{sanitize_display(commit.message)}")'
482 )
483 else:
484 header = f'commit {commit.commit_id[:8]} "{sanitize_display(commit.message)}"'
485
486 print(_c(header, _DIM, tty=tty))
487 _print_human(symbol_map, show_hashes, tty)
File History 2 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 32 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 32 days ago