|
| 1 | +# @WHO: scripts/commands/source.py |
| 2 | +# @WHAT: `context --check source` — return a function's source by name |
| 3 | +# @PART: command (sub-check of `context`) |
| 4 | +# @ENTRY: execute() |
| 5 | +"""`context --check source --name X` — a function's source, by name. |
| 6 | +
|
| 7 | +The most direct replacement for "Read the whole file to see one function": |
| 8 | +resolve X to its file and start line, bound it by the next declaration, and |
| 9 | +return just those lines. Read-only. Boundaries are heuristic — the next |
| 10 | +declaration in the file, or EOF — which is exact for the common case and |
| 11 | +never guesses beyond a file's own structure. |
| 12 | +""" |
| 13 | + |
| 14 | +import os |
| 15 | +from typing import Any, Dict, List |
| 16 | + |
| 17 | +from outline_engine import get_file_outline |
| 18 | + |
| 19 | + |
| 20 | +def add_args(parser): |
| 21 | + """Register CLI arguments (workspace/name/file carried by the umbrella).""" |
| 22 | + parser.add_argument( |
| 23 | + "workspace", nargs="?", default=None, |
| 24 | + help="Path to workspace root (auto-detected if omitted)", |
| 25 | + ) |
| 26 | + |
| 27 | + |
| 28 | +def _symbol_lines(outline: Dict) -> List[int]: |
| 29 | + """Every declaration line in a file, sorted — the boundary candidates.""" |
| 30 | + lines = [] |
| 31 | + for section in ("functions", "classes"): |
| 32 | + for entry in outline.get(section, []): |
| 33 | + if isinstance(entry.get("line"), int): |
| 34 | + lines.append(entry["line"]) |
| 35 | + return sorted(set(lines)) |
| 36 | + |
| 37 | + |
| 38 | +def _extract(abs_file: str, rel_file: str, workspace: str, |
| 39 | + start: int, name: str) -> Dict[str, Any]: |
| 40 | + """Slice one function's source from its start line to the next declaration.""" |
| 41 | + res = get_file_outline(abs_file, workspace, "normal") |
| 42 | + outline = res.get("outline") or {} |
| 43 | + line_count = outline.get("line_count", 0) |
| 44 | + |
| 45 | + decl_lines = _symbol_lines(outline) |
| 46 | + end = line_count |
| 47 | + for ln in decl_lines: |
| 48 | + if ln > start: |
| 49 | + end = ln - 1 |
| 50 | + break |
| 51 | + |
| 52 | + try: |
| 53 | + with open(abs_file, "r", encoding="utf-8", errors="replace") as f: |
| 54 | + file_lines = f.read().splitlines() |
| 55 | + except OSError as e: |
| 56 | + return {"symbol": name, "file": rel_file, "error": str(e)} |
| 57 | + |
| 58 | + body = file_lines[start - 1:end] |
| 59 | + # Drop blank lines between this function and the next declaration. |
| 60 | + while body and not body[-1].strip(): |
| 61 | + body.pop() |
| 62 | + end -= 1 |
| 63 | + return { |
| 64 | + "symbol": name, |
| 65 | + "file": rel_file, |
| 66 | + "start_line": start, |
| 67 | + "end_line": end, |
| 68 | + "source": "\n".join(body), |
| 69 | + } |
| 70 | + |
| 71 | + |
| 72 | +def execute(args, workspace) -> Dict[str, Any]: |
| 73 | + """Return the source of function(s) named ``--name``. |
| 74 | +
|
| 75 | + @FLOW: SOURCE_VIEW |
| 76 | + @CALLS: graph_model.find_nodes_by_name(), outline_engine.get_file_outline() |
| 77 | + @MUTATES: nothing (read-only) |
| 78 | + """ |
| 79 | + name = getattr(args, "name", None) |
| 80 | + if not name: |
| 81 | + return { |
| 82 | + "status": "error", |
| 83 | + "error": "source needs --name X (the function to show)", |
| 84 | + "error_type": "missing_argument", |
| 85 | + } |
| 86 | + |
| 87 | + workspace = os.path.abspath(workspace) if workspace else os.getcwd() |
| 88 | + only_file = getattr(args, "file", None) |
| 89 | + |
| 90 | + # Resolve where X is defined: an explicit --file needs no graph; otherwise |
| 91 | + # ask the call-graph (populated by a prior scan). |
| 92 | + locations = [] # (abs_file, rel_file, start_line) |
| 93 | + if only_file: |
| 94 | + abs_file = only_file if os.path.isabs(only_file) else os.path.join(workspace, only_file) |
| 95 | + res = get_file_outline(abs_file, workspace, "normal") |
| 96 | + outline = res.get("outline") or {} |
| 97 | + for fn in outline.get("functions", []): |
| 98 | + if fn.get("name") == name and isinstance(fn.get("line"), int): |
| 99 | + locations.append((abs_file, os.path.relpath(abs_file, workspace), fn["line"])) |
| 100 | + else: |
| 101 | + try: |
| 102 | + from utils import default_db_path |
| 103 | + import graph_model as gm |
| 104 | + except Exception: |
| 105 | + return {"status": "error", |
| 106 | + "error": "graph unavailable; pass --file to locate the function", |
| 107 | + "error_type": "no_graph"} |
| 108 | + db = getattr(args, "db_path", None) or default_db_path(workspace) |
| 109 | + if not db or not os.path.exists(db): |
| 110 | + return {"status": "error", |
| 111 | + "error": "no graph DB — scan the workspace first, or pass --file", |
| 112 | + "error_type": "no_graph"} |
| 113 | + for node in gm.find_nodes_by_name(name, db): |
| 114 | + rel = node.get("file", "") |
| 115 | + if not rel: |
| 116 | + continue |
| 117 | + abs_file = rel if os.path.isabs(rel) else os.path.join(workspace, rel) |
| 118 | + line = node.get("line") |
| 119 | + if isinstance(line, int) and line > 0 and os.path.exists(abs_file): |
| 120 | + locations.append((abs_file, rel.replace("\\", "/"), line)) |
| 121 | + |
| 122 | + if not locations: |
| 123 | + where = f" in {only_file}" if only_file else "" |
| 124 | + return {"status": "ok", "symbol": name, "found": False, |
| 125 | + "message": f"No function named '{name}' found{where}. " |
| 126 | + "If the workspace was never scanned, run scan first or pass --file."} |
| 127 | + |
| 128 | + matches = [_extract(a, r, workspace, ln, name) for a, r, ln in locations] |
| 129 | + return { |
| 130 | + "status": "ok", |
| 131 | + "symbol": name, |
| 132 | + "found": True, |
| 133 | + "count": len(matches), |
| 134 | + "matches": matches, |
| 135 | + } |
0 commit comments