Spaces:
Running
Running
| """ | |
| core/intent.py — read what a codebase is *for* out of the codebase. | |
| Every repository already declares its own callable surface, precisely, in a form | |
| meant to be authoritative: | |
| argparse every flag, its type, whether it is required, its help | |
| click / typer the same, as decorators | |
| FastAPI / Flask every route, its method, its path | |
| console_scripts what the package installs as a command | |
| exported functions what a module offers to its importers | |
| A person reading that repo derives its intent from exactly these. So does this. | |
| The alternative — someone writing an adapter per project, naming coordinates by | |
| hand — does not scale past the projects that person has time for, and encodes | |
| their reading rather than the author's declaration. | |
| The rule that keeps it honest: a parameter's name, requiredness and help text | |
| come from the source or are absent. Nothing is invented to make a coordinate | |
| look complete. A CLI that documents nothing produces a coordinate that says it | |
| documents nothing, which is a fact worth having. | |
| """ | |
| from __future__ import annotations | |
| import ast | |
| import json | |
| import re | |
| from pathlib import Path | |
| from typing import Any, Dict, List, Optional, Sequence | |
| SCHEMA = "signalmesh.intent/v1" | |
| SKIP_DIRS = {".git", "node_modules", "__pycache__", ".venv", "venv", "build", | |
| "dist", ".next", "target", "vendor", ".tox", "site-packages"} | |
| MAX_FILES = 3000 | |
| def _walk(root: Path, suffix: str) -> List[Path]: | |
| out: List[Path] = [] | |
| for p in root.rglob(f"*{suffix}"): | |
| if any(part in SKIP_DIRS for part in p.parts): | |
| continue | |
| out.append(p) | |
| if len(out) >= MAX_FILES: | |
| break | |
| return sorted(out) | |
| def _const(node: ast.AST) -> Any: | |
| try: | |
| return ast.literal_eval(node) | |
| except Exception: | |
| return None | |
| def _kw(call: ast.Call, name: str) -> Any: | |
| for k in call.keywords: | |
| if k.arg == name: | |
| return _const(k.value) | |
| return None | |
| def cli_from_python(path: Path) -> Optional[Dict[str, Any]]: | |
| """An argparse/click CLI, read from the parser the author actually built.""" | |
| try: | |
| tree = ast.parse(path.read_text(encoding="utf-8", errors="ignore")) | |
| except (OSError, SyntaxError): | |
| return None | |
| args: List[Dict[str, Any]] = [] | |
| description = "" | |
| kind = "" | |
| for node in ast.walk(tree): | |
| if not isinstance(node, ast.Call): | |
| continue | |
| fn = node.func | |
| # argparse: ap.add_argument("x", "--y", help=..., required=..., action=...) | |
| if isinstance(fn, ast.Attribute) and fn.attr == "add_argument": | |
| kind = kind or "argparse" | |
| flags = [v for v in (_const(a) for a in node.args) if isinstance(v, str)] | |
| if not flags: | |
| continue | |
| positional = not any(f.startswith("-") for f in flags) | |
| name = next((f.lstrip("-") for f in flags if f.startswith("--")), | |
| flags[0].lstrip("-")) | |
| action = _kw(node, "action") or "" | |
| args.append({ | |
| "name": name, "flags": flags, "positional": positional, | |
| "help": _kw(node, "help") or "", | |
| "required": bool(_kw(node, "required")) or positional, | |
| "type": "bool" if action in ("store_true", "store_false") else | |
| ("list" if _kw(node, "nargs") in ("*", "+") else "str"), | |
| "default": _kw(node, "default"), | |
| }) | |
| elif isinstance(fn, ast.Attribute) and fn.attr == "ArgumentParser": | |
| description = _kw(node, "description") or description | |
| # click / typer: @click.option("--x"), @click.argument("y") | |
| elif isinstance(fn, ast.Attribute) and fn.attr in ("option", "argument"): | |
| mod = getattr(fn.value, "id", "") | |
| if mod not in ("click", "typer"): | |
| continue | |
| kind = kind or mod | |
| flags = [v for v in (_const(a) for a in node.args) if isinstance(v, str)] | |
| if not flags: | |
| continue | |
| positional = fn.attr == "argument" | |
| name = next((f.lstrip("-") for f in flags if f.startswith("--")), | |
| flags[0].lstrip("-")) | |
| args.append({ | |
| "name": name, "flags": flags, "positional": positional, | |
| "help": _kw(node, "help") or "", | |
| "required": bool(_kw(node, "required")) or positional, | |
| "type": "bool" if _kw(node, "is_flag") else "str", | |
| "default": _kw(node, "default"), | |
| }) | |
| if not args: | |
| return None | |
| if not description: | |
| description = (ast.get_docstring(tree) or "").strip().split("\n\n")[0] | |
| # dedupe, keeping the first declaration of each flag | |
| seen, uniq = set(), [] | |
| for a in args: | |
| if a["name"] in seen: | |
| continue | |
| seen.add(a["name"]) | |
| uniq.append(a) | |
| return {"kind": kind or "argparse", "module": path.stem, "path": str(path), | |
| "summary": description[:400], "args": uniq} | |
| def routes_from_python(path: Path) -> List[Dict[str, Any]]: | |
| """HTTP routes, read from the decorators that define them.""" | |
| try: | |
| tree = ast.parse(path.read_text(encoding="utf-8", errors="ignore")) | |
| except (OSError, SyntaxError): | |
| return [] | |
| out = [] | |
| for node in ast.walk(tree): | |
| if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): | |
| continue | |
| for dec in node.decorator_list: | |
| if not isinstance(dec, ast.Call) or not isinstance(dec.func, ast.Attribute): | |
| continue | |
| method = dec.func.attr.upper() | |
| if method not in ("GET", "POST", "PUT", "PATCH", "DELETE", "ROUTE"): | |
| continue | |
| route = next((v for v in (_const(a) for a in dec.args) | |
| if isinstance(v, str)), None) | |
| if not route: | |
| continue | |
| out.append({"method": "GET" if method == "ROUTE" else method, | |
| "path": route, "handler": node.name, | |
| "summary": (ast.get_docstring(node) or "").strip() | |
| .split("\n\n")[0][:200], | |
| "params": [a.arg for a in node.args.args | |
| if a.arg not in ("self", "request")]}) | |
| return out | |
| def scripts_from_manifest(root: Path) -> List[Dict[str, Any]]: | |
| """Commands the package says it installs — pyproject, setup.cfg, package.json.""" | |
| out = [] | |
| py = root / "pyproject.toml" | |
| if py.exists(): | |
| txt = py.read_text(encoding="utf-8", errors="ignore") | |
| block = re.search(r"\[project\.scripts\](.*?)(\n\[|\Z)", txt, re.S) | |
| if block: | |
| for line in block.group(1).splitlines(): | |
| m = re.match(r'\s*([\w.-]+)\s*=\s*["\']([^"\']+)', line) | |
| if m: | |
| out.append({"command": m.group(1), "entry": m.group(2), | |
| "from": "pyproject.toml"}) | |
| pkg = root / "package.json" | |
| if pkg.exists(): | |
| try: | |
| data = json.loads(pkg.read_text(encoding="utf-8", errors="ignore")) | |
| except json.JSONDecodeError: | |
| data = {} | |
| bins = data.get("bin") | |
| if isinstance(bins, str): | |
| out.append({"command": data.get("name", "cli"), "entry": bins, | |
| "from": "package.json"}) | |
| elif isinstance(bins, dict): | |
| out += [{"command": k, "entry": v, "from": "package.json"} | |
| for k, v in bins.items()] | |
| for name, script in (data.get("scripts") or {}).items(): | |
| if name in ("start", "serve", "dev", "build"): | |
| out.append({"command": f"npm run {name}", "entry": script, | |
| "from": "package.json scripts"}) | |
| return out | |
| def _slug(text: str) -> str: | |
| return re.sub(r"[^a-z0-9]+", "_", str(text).lower()).strip("_") or "op" | |
| def read(root: Path, *, name: str = "") -> Dict[str, Any]: | |
| """Everything a repository says about how it wants to be called.""" | |
| root = Path(root).expanduser().resolve() | |
| slug = _slug(name or root.name) | |
| clis, routes = [], [] | |
| for f in _walk(root, ".py"): | |
| got = cli_from_python(f) | |
| if got: | |
| clis.append(got) | |
| routes += routes_from_python(f) | |
| scripts = scripts_from_manifest(root) | |
| coords: List[Dict[str, Any]] = [] | |
| for c in clis: | |
| coords.append({ | |
| "coord": f"{slug}.{_slug(c['module'])}", | |
| "protocol": "cli", "kind": c["kind"], | |
| "summary": c["summary"], "run": c["path"], | |
| "params": [a["name"] for a in c["args"]], | |
| "required": [a["name"] for a in c["args"] if a["required"]], | |
| "args": c["args"], | |
| }) | |
| for r in routes: | |
| coords.append({ | |
| "coord": f"{slug}.{_slug(r['handler'])}", | |
| "protocol": "http", "method": r["method"], "path": r["path"], | |
| "summary": r["summary"], "params": r["params"], "required": [], | |
| }) | |
| for s in scripts: | |
| coords.append({ | |
| "coord": f"{slug}.{_slug(s['command'])}", | |
| "protocol": "command", "summary": f"{s['entry']} ({s['from']})", | |
| "run": s["command"], "params": [], "required": [], | |
| }) | |
| return { | |
| "schema": SCHEMA, "repo": str(root), "slug": slug, | |
| "coordinates": coords, | |
| "totals": {"coordinates": len(coords), "clis": len(clis), | |
| "routes": len(routes), "commands": len(scripts)}, | |
| "note": ("read from the repository's own declarations — argparse, click, " | |
| "route decorators, packaged entry points. Nothing inferred, and " | |
| "nothing named by hand."), | |
| } | |
| def invocation(coord: Dict[str, Any], values: Dict[str, Any]) -> List[str]: | |
| """Build the command this coordinate's own parser accepts. | |
| Refuses when a required argument is missing, rather than emitting something | |
| that will fail after the work of getting there. | |
| """ | |
| if coord.get("protocol") != "cli": | |
| raise ValueError(f"{coord['coord']} is {coord.get('protocol')}, not a CLI") | |
| missing = [r for r in coord.get("required", []) if values.get(r) in (None, "", False)] | |
| if missing: | |
| raise ValueError(f"{coord['coord']} requires {missing} — its own parser " | |
| f"will reject the call without them") | |
| cmd = ["python", coord["run"]] | |
| for a in coord.get("args", []): | |
| v = values.get(a["name"]) | |
| if v in (None, "", False): | |
| continue | |
| if a["positional"]: | |
| cmd += [str(x) for x in v] if isinstance(v, list) else [str(v)] | |
| elif a["type"] == "bool": | |
| cmd.append(a["flags"][-1]) | |
| else: | |
| cmd += [a["flags"][-1], str(v)] | |
| return cmd | |
| if __name__ == "__main__": | |
| import tempfile | |
| src = ''' | |
| """Do a specific job. | |
| Longer prose nobody needs in a summary. | |
| """ | |
| import argparse | |
| def main(): | |
| ap = argparse.ArgumentParser(description="Compile and annotate templates.") | |
| ap.add_argument("template", nargs="*", help="template directories") | |
| ap.add_argument("--all", action="store_true", help="process every template") | |
| ap.add_argument("--profile", help="force a profile") | |
| ap.add_argument("--no-overlay", action="store_true") | |
| ''' | |
| api = ''' | |
| from fastapi import FastAPI | |
| app = FastAPI() | |
| @app.get("/health") | |
| def health(): | |
| """Say whether it is up.""" | |
| return {} | |
| @app.post("/convert/{kind}") | |
| async def convert(kind: str, body: dict): | |
| """Turn one thing into another.""" | |
| return {} | |
| ''' | |
| with tempfile.TemporaryDirectory() as tmp: | |
| root = Path(tmp) | |
| (root / "tool.py").write_text(src) | |
| (root / "server.py").write_text(api) | |
| (root / "pyproject.toml").write_text( | |
| '[project]\nname="x"\n\n[project.scripts]\nmytool = "x.cli:main"\n') | |
| (root / "node_modules").mkdir() | |
| (root / "node_modules" / "junk.py").write_text(src) # must be skipped | |
| got = read(root, name="demo") | |
| by = {c["coord"]: c for c in got["coordinates"]} | |
| cli = by["demo.tool"] | |
| assert cli["protocol"] == "cli" and cli["kind"] == "argparse" | |
| # the author's own description, not the module docstring's first line | |
| assert cli["summary"] == "Compile and annotate templates." | |
| assert cli["params"] == ["template", "all", "profile", "no-overlay"], cli["params"] | |
| # a positional is required; a flag is not — read from the declaration | |
| assert cli["required"] == ["template"], cli["required"] | |
| assert next(a for a in cli["args"] if a["name"] == "all")["type"] == "bool" | |
| assert next(a for a in cli["args"] if a["name"] == "template")["type"] == "list" | |
| assert by["demo.health"]["method"] == "GET" | |
| assert by["demo.convert"]["method"] == "POST" | |
| assert by["demo.convert"]["path"] == "/convert/{kind}" | |
| assert by["demo.convert"]["summary"] == "Turn one thing into another." | |
| assert by["demo.mytool"]["protocol"] == "command" | |
| # vendored code is not this repository's intent | |
| assert got["totals"]["clis"] == 1, "node_modules must not be read" | |
| # the built command matches the parser that will receive it | |
| cmd = invocation(cli, {"template": ["a", "b"], "all": False, "profile": "news"}) | |
| assert cmd[1].endswith("tool.py") | |
| assert cmd[2:] == ["a", "b", "--profile", "news"], cmd | |
| assert "--all" in invocation(cli, {"template": ["a"], "all": True}) | |
| assert "--no-overlay" in invocation(cli, {"template": ["a"], "no-overlay": True}) | |
| try: | |
| invocation(cli, {}) | |
| raise AssertionError("a missing required arg must be refused here") | |
| except ValueError as e: | |
| assert "requires ['template']" in str(e) | |
| assert read(root)["slug"] == _slug(root.name) | |
| # ── against the real checkouts, if they are here ───────────────────────── | |
| for repo, expect in ( | |
| ("layout-detection-synthetic-data-main", "annotate"), | |
| ("ConvertX-main", None), | |
| ): | |
| p = Path.home() / "Downloads" / "fixMesh" / repo | |
| if not p.is_dir(): | |
| continue | |
| got = read(p) | |
| print(f"{repo}: {got['totals']}") | |
| if expect: | |
| hit = next((c for c in got["coordinates"] | |
| if c["coord"].endswith(expect)), None) | |
| assert hit, f"did not discover {expect} in {repo}" | |
| print(f" discovered: {hit['coord']} params={hit['params']} " | |
| f"required={hit['required']}") | |
| print("intent self-check OK") | |