Current section
Files
Jump to
Current section
Files
priv/toolchain/tsq.py
#!/usr/bin/env python3
"""tsq.py — structural search for the judged leg (tree-sitter).
The judge's investigation instrument: query source STRUCTURE (call
nodes, not strings), get back matches whose spans ARE verbatim quotes —
so every citation is discovered, not composed. grep matches comments and
near-misses; this matches code.
Queries (v1):
tsq.py call <name> every call whose target is <name>
(bare `plug`, or dotted `System.get_env`,
or rightmost match `get_env`)
tsq.py --scm <query.scm> raw tree-sitter query (escape hatch)
Options:
--repo DIR repository root (default .)
--in PREFIX limit to paths under PREFIX (repeatable)
--lang elixir grammar (only elixir today)
--json records with byte spans instead of grep-style lines
Output (default): file:line: first-line-of-match
Output (--json): {"file","line","start_byte","end_byte","text"} per match
— "text" is the exact source span: paste it as a quote
and the instrument's verification will pass by
construction.
Deps: pip install tree-sitter tree-sitter-elixir
"""
import argparse
import json
import os
import re
import sys
import tree_sitter
import tree_sitter_elixir
LANGS = {"elixir": tree_sitter.Language(tree_sitter_elixir.language())}
def code_files(repo, prefixes):
out = []
for base, dirs, names in os.walk(repo):
rel_base = os.path.relpath(base, repo)
top = rel_base.split(os.sep)[0]
if top in {".git", "deps", "_build", "node_modules", "assets", "docs"}:
dirs[:] = []
continue
for n in names:
if not (n.endswith(".ex") or n.endswith(".exs")):
continue
rel = os.path.normpath(os.path.join(rel_base, n)).replace(os.sep, "/")
if prefixes and not any(rel.startswith(p) for p in prefixes):
continue
out.append(rel)
return sorted(out)
def call_target(src, node):
"""The textual target of a call node: `plug`, `System.get_env`, …"""
if not node.children:
return None
t = node.children[0]
if t.type == "identifier":
return src[t.start_byte : t.end_byte].decode(errors="replace")
if t.type == "dot":
return src[t.start_byte : t.end_byte].decode(errors="replace")
return None
def match_call(target, name):
if target is None:
return False
if target == name:
return True
# rightmost-segment match: `get_env` finds `System.get_env`
return "." in target and target.rsplit(".", 1)[-1] == name
def find_calls(src, root, name):
out = []
stack = [root]
while stack:
n = stack.pop()
if n.type == "call" and match_call(call_target(src, n), name):
out.append(n)
stack.extend(reversed(n.children))
return out
def run_scm(lang, src, root, scm):
"""One node per MATCH (the widest captured span), not per capture —
a query with helper captures (@f inside @m) still counts each
logical match once, and the cited span is the enclosing one."""
query = tree_sitter.Query(lang, scm)
cursor = tree_sitter.QueryCursor(query)
out = []
for _match_id, captures in cursor.matches(root):
nodes = [n for ns in captures.values() for n in ns]
if nodes:
out.append(max(nodes, key=lambda n: n.end_byte - n.start_byte))
return out
def textual_hits(repo, prefixes, name):
"""Occurrences of `name` as TEXT in the scoped files — the cross-check
behind absence claims (0 structural + textual hits = not citable)."""
frag = name.rsplit(".", 1)[-1].encode()
n = 0
for rel in code_files(repo, prefixes):
try:
n += open(os.path.join(repo, rel), "rb").read().count(frag)
except OSError:
continue
return n
def main():
ap = argparse.ArgumentParser()
ap.add_argument("kind", nargs="?", choices=["call"])
ap.add_argument("name", nargs="?")
ap.add_argument("--scm")
ap.add_argument("--repo", default=".")
ap.add_argument("--in", dest="prefixes", action="append", default=[])
ap.add_argument("--lang", default="elixir")
ap.add_argument("--json", action="store_true")
args = ap.parse_args()
if not args.scm and not (args.kind and args.name):
ap.error("either `call <name>` or --scm <file>")
lang = LANGS[args.lang]
parser = tree_sitter.Parser(lang)
scm = open(args.scm).read() if args.scm else None
records = []
for rel in code_files(args.repo, args.prefixes):
try:
src = open(os.path.join(args.repo, rel), "rb").read()
except OSError:
continue
root = parser.parse(src).root_node
nodes = run_scm(lang, src, root, scm) if scm else find_calls(src, root, args.name)
for n in nodes:
text = src[n.start_byte : n.end_byte].decode(errors="replace")
records.append(
{
"file": rel,
"line": n.start_point[0] + 1,
"start_byte": n.start_byte,
"end_byte": n.end_byte,
"text": text,
}
)
if args.json:
json.dump(records, sys.stdout, indent=1)
print()
else:
for r in records:
print(f"{r['file']}:{r['line']}: {r['text'].splitlines()[0][:160]}")
print(f"-- {len(records)} match(es)", file=sys.stderr)
# the absence guard: a zero-structural result with textual hits is NOT
# a citable absence — the query may be wrong, or the code macro-generated.
if not records and args.kind == "call" and args.name:
hits = textual_hits(args.repo, args.prefixes, args.name)
if hits:
print(
f"!! 0 structural matches but {hits} textual hit(s) for "
f"'{args.name}' — inspect before citing absence",
file=sys.stderr,
)
if __name__ == "__main__":
main()