diff --git a/bex/tag_preprocessor/analyze.py b/bex/tag_preprocessor/analyze.py index cec0dc5..5036e85 100644 --- a/bex/tag_preprocessor/analyze.py +++ b/bex/tag_preprocessor/analyze.py @@ -7,14 +7,16 @@ Runs the full Phase 1.0 pipeline over a directory of source files. """ import argparse +import json import os +import re import sys -from pathlib import Path, PurePath +from pathlib import Path from collections import Counter import pathspec -from .code import preprocess_by_method, _extract_call_tokens +from .code import preprocess_by_method, _extract_call_tokens, extract_arg_info, _summarize_arg_info from bex.ensemble import infer_ensemble SUPPORTED_EXTENSIONS = { @@ -46,6 +48,77 @@ def _match_glob(filepath, pattern): return spec.match_file(filepath) +IMPORT_PATTERNS = [ + re.compile(r"^\s*import\s+"), + re.compile(r"^\s*from\s+"), + re.compile(r"^\s*require\s+"), + re.compile(r"^\s*require_relative\s+"), + re.compile(r"^\s*#\s*include\s+"), + re.compile(r"^\s*use\s+"), + re.compile(r"^\s*include\s+"), +] + + +def _file_to_package(fp, project_root): + """Infer the package/module namespace from a file path. + + Uses the directory of the file relative to the project root. + Zero per-language or per-convention assumptions — pure path math. + """ + rel = os.path.relpath(os.path.dirname(fp), project_root) + if rel == ".": + return "" + return rel + + +def _top_packages(file_paths, project_root, top_n=3): + """Return the most common packages among a set of files.""" + pkg_counts = Counter() + for fp in file_paths: + pkg = _file_to_package(fp, project_root) + pkg_counts[pkg] += 1 + return [pkg for pkg, _ in pkg_counts.most_common(top_n)] + + +def _extract_imports(file_paths): + """Extract unique import lines from source files. + + Scans top 200 lines of each file for common import patterns + across all supported languages. Deduplicates across files. + """ + seen = set() + result = [] + for fp in sorted(file_paths): + try: + with open(fp) as f: + for i, line in enumerate(f): + if i >= 200: + break + stripped = line.strip() + if any(p.match(stripped) for p in IMPORT_PATTERNS): + if stripped not in seen: + seen.add(stripped) + result.append(stripped) + except OSError: + continue + return result + + +def _build_arg_patterns(file_paths): + """Extract merged argument patterns across a set of files.""" + merged = {} + for fp in file_paths: + try: + with open(fp) as f: + code = f.read() + except OSError: + continue + info = extract_arg_info(fp, code) + for call_name, observations in info.items(): + merged.setdefault(call_name, []).extend(observations) + return _summarize_arg_info(merged) + + def scan_directory(dir_path, gitignore_spec=None): """Walk dir_path, return dict mapping extension → [file paths]. @@ -161,19 +234,22 @@ def cluster_methods(sequences, min_cluster_size=3, ngram_size=3): return clusters -def analyze_clusters(file_paths, extension, min_coverage=0.2, prefer=None, kmax=2, N=3): +def analyze_clusters(file_paths, extension, project_root="", min_coverage=0.2, prefer=None, kmax=2, N=3): """Run full pipeline with clustering: preprocess → cluster → per-cluster infer. Returns: - list of (label, ensemble_result_dict, method_count) tuples. + list of (label, ensemble_result_dict, method_count, meta) tuples. + meta = {"files": set(paths), "imports": [lines], "arg_patterns": {...}, "packages": [...]}. """ sequences = [] + seq_files = [] for fp in file_paths: with open(fp) as f: code = f.read() for method_seq in preprocess_by_method(fp, code): if method_seq: sequences.append(method_seq) + seq_files.append(fp) if not sequences: return [] @@ -183,9 +259,17 @@ def analyze_clusters(file_paths, extension, min_coverage=0.2, prefer=None, kmax= results = [] for label, cluster_seqs in clusters: + cluster_fps = set() + for seq in cluster_seqs: + idx = next(i for i, s in enumerate(sequences) if s is seq) + cluster_fps.add(seq_files[idx]) + imports = _extract_imports(cluster_fps) + arg_patterns = _build_arg_patterns(cluster_fps) symbol_seqs = [[text for _, text, _ in seq] for seq in cluster_seqs] result = infer_ensemble(symbol_seqs, kmax=kmax, N=N, prefer=prefer) - results.append((label, result, len(cluster_seqs))) + packages = _top_packages(cluster_fps, project_root) + meta = {"files": cluster_fps, "imports": imports, "arg_patterns": arg_patterns, "packages": packages} + results.append((label, result, len(cluster_seqs), meta)) return results @@ -238,7 +322,7 @@ def analyze_directory( cluster: if True, run method-level clustering (default). Returns: - dict mapping extension → list of (label, result_dict, count) tuples. + dict mapping extension → list of (label, result_dict, count, meta) tuples. """ groups = scan_directory(dir_path) results = {} @@ -252,16 +336,42 @@ def analyze_directory( if cluster: results[ext] = analyze_clusters( files, ext, + project_root=dir_path, min_coverage=min_coverage, prefer=prefer, kmax=kmax, ) else: r = infer(files, ext, min_coverage=min_coverage, prefer=prefer, kmax=kmax) - results[ext] = [("(all methods)", r, 0)] + imports = _extract_imports(files) + results[ext] = [("(all methods)", r, 0, {"files": set(files), "imports": imports, "arg_patterns": {}, "packages": _top_packages(files, dir_path)})] return results +def _build_json_output(results): + """Convert results dict to a compact JSON structure for prompt injection.""" + output = [] + for ext, clusters in results.items(): + lang = {"language": ext, "conventions": []} + total_methods = 0 + for label, result, count, meta in clusters: + total_methods += count + entry = { + "label": label, + "method_count": count, + } + if result and result.get("best"): + entry["algorithm"] = result["best"]["algorithm"] + entry["grammar"] = result["best"]["grammar"] + entry["mdl_score"] = round(result["best"]["mdl_score"], 1) + entry["imports"] = meta.get("imports", []) + entry["arg_patterns"] = meta.get("arg_patterns", {}) + lang["conventions"].append(entry) + lang["total_methods"] = total_methods + output.append(lang) + return json.dumps(output, indent=2) + + def _parse_args(argv=None): parser = argparse.ArgumentParser( description="Analyze a directory of source code for behavioral conventions.", @@ -292,6 +402,14 @@ def _parse_args(argv=None): "--ngram-size", type=int, default=3, help="N-gram length for clustering (default: 3)", ) + parser.add_argument( + "--format", choices=["text", "json"], default="text", + help="Output format (default: text)", + ) + parser.add_argument( + "--json", action="store_true", dest="json_flag", + help="Shortcut for --format json", + ) return parser.parse_args(argv) @@ -304,9 +422,14 @@ def main(): kmax=args.kmax, include=args.include, ) + + if args.json_flag or args.format == "json": + print(_build_json_output(results)) + return + for ext, clusters in results.items(): print(f"\n{ext}:") - for label, result, count in clusters: + for label, result, count, meta in clusters: if result and result.get("best"): best = result["best"] print(f" ╰─ {label} ({count} methods)") @@ -315,6 +438,21 @@ def main(): print(f" MDL: {best['mdl_score']}") else: print(f" ╰─ {label} ({count} methods) — no grammar") + imps = meta.get("imports", []) + if imps: + joined = " | ".join(imps[:6]) + print(f" Imports: {joined}") + if len(imps) > 6: + print(f" ... and {len(imps) - 6} more") + argp = meta.get("arg_patterns", {}) + if argp: + top_calls = sorted(argp.items(), key=lambda x: -x[1]["occurrences"])[:4] + for call_name, asum in top_calls: + ac = asum["arg_count"] + pats = asum["patterns"][:2] + pat_strs = [f"{p['args']}:{','.join(p['types'])}" for p in pats] + print(f" Args({call_name}): n={ac['common']} " + f"[{'; '.join(pat_strs)}]") if __name__ == "__main__": diff --git a/bex/tag_preprocessor/code.py b/bex/tag_preprocessor/code.py index 206b575..6d91985 100644 --- a/bex/tag_preprocessor/code.py +++ b/bex/tag_preprocessor/code.py @@ -69,6 +69,106 @@ def _extract_call_tokens(seq): return result +ARG_LITERAL_TYPES = { + "string", "string_literal", "number", "integer", "float", + "decimal", "true", "false", "null", "nil", "none", +} + +LAMBDA_TYPES = { + "lambda", "do_block", "block", +} + + +def _classify_arg_node(node): + t = node.type + if t in ARG_LITERAL_TYPES: + return "lit" + if t in LAMBDA_TYPES or "ambda" in t or "block" in t: + return "lambda" + if t == "keyword_argument" or t.endswith("named_argument"): + return "kwarg" + if t.endswith("call_expression") or t in ("call", "method_invocation"): + return "call" + if t.endswith("identifier") or t.endswith("name"): + return "var" + if "subscript" in t: + return "subscript" + if "binary" in t or "unary" in t or "ternary" in t or "operator" in t: + return "expr" + if t in ("interpolation", "template_string"): + return "template" + return "other" + + +def _find_arglist_node(parent): + args = parent.child_by_field_name("arguments") + if args: + return args + for child in parent.children: + if child.type in ("argument_list", "arguments"): + return child + if child.type == "template_string": + return child + return None + + +def _iterate_arg_nodes(arglist): + for child in arglist.children: + if child.is_named: + yield child + + +def extract_arg_info(file_path, code): + ext = os.path.splitext(file_path)[1].lower() + lang, query_name = _load_grammar(ext) + query_src = _load_query(query_name) + + parser = Parser(lang) + tree = parser.parse(code.encode()) + query = Query(lang, query_src) + + cursor = QueryCursor(query) + captures = cursor.captures(tree.root_node) + + info = {} + for capname, nodes in captures.items(): + if not capname.startswith(BEHAVIORAL_PREFIXES): + continue + for node in nodes: + text = code[node.start_byte:node.end_byte].strip() + parent = node.parent + if not parent: + continue + arglist = _find_arglist_node(parent) + if not arglist: + continue + types = [_classify_arg_node(c) for c in _iterate_arg_nodes(arglist)] + info.setdefault(text, []).append((len(types), tuple(types))) + return info + + +def _summarize_arg_info(info): + from collections import Counter + summary = {} + for call_name, observations in info.items(): + counts = [c for c, _ in observations] + pattern_counts = Counter(observations) + top_patterns = pattern_counts.most_common(5) + summary[call_name] = { + "occurrences": len(observations), + "arg_count": { + "min": min(counts), + "max": max(counts), + "common": max(set(counts), key=counts.count), + }, + "patterns": [ + {"count": c, "args": n, "types": list(t)} + for (n, t), c in top_patterns + ], + } + return summary + + _grammar_cache = {} _query_cache = {} diff --git a/tests/test_analyze.py b/tests/test_analyze.py index b562285..99135a5 100644 --- a/tests/test_analyze.py +++ b/tests/test_analyze.py @@ -138,7 +138,7 @@ def test_analyze_directory_include_glob(): results = analyze_directory(td, include="**/src/main/**") assert ".py" in results assert len(results[".py"]) >= 1 - for label, r, count in results[".py"]: + for label, r, count, meta in results[".py"]: if r and r.get("best"): assert r["best"]["grammar"] is not None print(" PASS test_analyze_directory_include_glob")