mirror of
https://github.com/safishamsi/graphify.git
synced 2026-09-23 05:55:54 +00:00
Add Elixir language support (.ex/.exs)
Extracts defmodule, def/defp, alias/import/require/use, and call graph. Follows same custom-walk pattern as Zig and PowerShell extractors.
This commit is contained in:
+1
-1
@@ -17,7 +17,7 @@ class FileType(str, Enum):
|
||||
|
||||
_MANIFEST_PATH = "graphify-out/manifest.json"
|
||||
|
||||
CODE_EXTENSIONS = {'.py', '.ts', '.js', '.tsx', '.go', '.rs', '.java', '.cpp', '.cc', '.cxx', '.c', '.h', '.hpp', '.rb', '.swift', '.kt', '.kts', '.cs', '.scala', '.php', '.lua', '.toc', '.zig', '.ps1'}
|
||||
CODE_EXTENSIONS = {'.py', '.ts', '.js', '.tsx', '.go', '.rs', '.java', '.cpp', '.cc', '.cxx', '.c', '.h', '.hpp', '.rb', '.swift', '.kt', '.kts', '.cs', '.scala', '.php', '.lua', '.toc', '.zig', '.ps1', '.ex', '.exs'}
|
||||
DOC_EXTENSIONS = {'.md', '.txt', '.rst'}
|
||||
PAPER_EXTENSIONS = {'.pdf'}
|
||||
IMAGE_EXTENSIONS = {'.png', '.jpg', '.jpeg', '.gif', '.webp', '.svg'}
|
||||
|
||||
@@ -1928,6 +1928,183 @@ def _resolve_cross_file_imports(
|
||||
return new_edges
|
||||
|
||||
|
||||
def extract_elixir(path: Path) -> dict:
|
||||
"""Extract modules, functions, imports, and calls from a .ex/.exs file."""
|
||||
try:
|
||||
import tree_sitter_elixir as tselixir
|
||||
from tree_sitter import Language, Parser
|
||||
except ImportError:
|
||||
return {"nodes": [], "edges": [], "error": "tree_sitter_elixir not installed"}
|
||||
|
||||
try:
|
||||
language = Language(tselixir.language())
|
||||
parser = Parser(language)
|
||||
source = path.read_bytes()
|
||||
tree = parser.parse(source)
|
||||
root = tree.root_node
|
||||
except Exception as e:
|
||||
return {"nodes": [], "edges": [], "error": str(e)}
|
||||
|
||||
stem = path.stem
|
||||
str_path = str(path)
|
||||
nodes: list[dict] = []
|
||||
edges: list[dict] = []
|
||||
seen_ids: set[str] = set()
|
||||
function_bodies: list[tuple[str, Any]] = []
|
||||
|
||||
def add_node(nid: str, label: str, line: int) -> None:
|
||||
if nid not in seen_ids:
|
||||
seen_ids.add(nid)
|
||||
nodes.append({"id": nid, "label": label, "file_type": "code",
|
||||
"source_file": str_path, "source_location": f"L{line}"})
|
||||
|
||||
def add_edge(src: str, tgt: str, relation: str, line: int,
|
||||
confidence: str = "EXTRACTED", weight: float = 1.0) -> None:
|
||||
edges.append({"source": src, "target": tgt, "relation": relation,
|
||||
"confidence": confidence, "source_file": str_path,
|
||||
"source_location": f"L{line}", "weight": weight})
|
||||
|
||||
file_nid = _make_id(stem)
|
||||
add_node(file_nid, path.name, 1)
|
||||
|
||||
_IMPORT_KEYWORDS = frozenset({"alias", "import", "require", "use"})
|
||||
|
||||
def _get_alias_text(node) -> str | None:
|
||||
for child in node.children:
|
||||
if child.type == "alias":
|
||||
return source[child.start_byte:child.end_byte].decode("utf-8", errors="replace")
|
||||
return None
|
||||
|
||||
def walk(node, parent_module_nid: str | None = None) -> None:
|
||||
if node.type != "call":
|
||||
for child in node.children:
|
||||
walk(child, parent_module_nid)
|
||||
return
|
||||
|
||||
identifier_node = None
|
||||
arguments_node = None
|
||||
do_block_node = None
|
||||
for child in node.children:
|
||||
if child.type == "identifier":
|
||||
identifier_node = child
|
||||
elif child.type == "arguments":
|
||||
arguments_node = child
|
||||
elif child.type == "do_block":
|
||||
do_block_node = child
|
||||
|
||||
if identifier_node is None:
|
||||
for child in node.children:
|
||||
walk(child, parent_module_nid)
|
||||
return
|
||||
|
||||
keyword = source[identifier_node.start_byte:identifier_node.end_byte].decode("utf-8", errors="replace")
|
||||
line = node.start_point[0] + 1
|
||||
|
||||
if keyword == "defmodule":
|
||||
module_name = _get_alias_text(arguments_node) if arguments_node else None
|
||||
if not module_name:
|
||||
return
|
||||
module_nid = _make_id(stem, module_name)
|
||||
add_node(module_nid, module_name, line)
|
||||
add_edge(file_nid, module_nid, "contains", line)
|
||||
if do_block_node:
|
||||
for child in do_block_node.children:
|
||||
walk(child, parent_module_nid=module_nid)
|
||||
return
|
||||
|
||||
if keyword in ("def", "defp"):
|
||||
func_name = None
|
||||
if arguments_node:
|
||||
for child in arguments_node.children:
|
||||
if child.type == "call":
|
||||
for sub in child.children:
|
||||
if sub.type == "identifier":
|
||||
func_name = source[sub.start_byte:sub.end_byte].decode("utf-8", errors="replace")
|
||||
break
|
||||
elif child.type == "identifier":
|
||||
func_name = source[child.start_byte:child.end_byte].decode("utf-8", errors="replace")
|
||||
break
|
||||
if not func_name:
|
||||
return
|
||||
container = parent_module_nid or file_nid
|
||||
func_nid = _make_id(container, func_name)
|
||||
add_node(func_nid, f"{func_name}()", line)
|
||||
if parent_module_nid:
|
||||
add_edge(parent_module_nid, func_nid, "method", line)
|
||||
else:
|
||||
add_edge(file_nid, func_nid, "contains", line)
|
||||
if do_block_node:
|
||||
function_bodies.append((func_nid, do_block_node))
|
||||
return
|
||||
|
||||
if keyword in _IMPORT_KEYWORDS and arguments_node:
|
||||
module_name = _get_alias_text(arguments_node)
|
||||
if module_name:
|
||||
tgt_nid = _make_id(module_name)
|
||||
add_edge(file_nid, tgt_nid, "imports", line)
|
||||
return
|
||||
|
||||
for child in node.children:
|
||||
walk(child, parent_module_nid)
|
||||
|
||||
walk(root)
|
||||
|
||||
label_to_nid: dict[str, str] = {}
|
||||
for n in nodes:
|
||||
normalised = n["label"].strip("()").lstrip(".")
|
||||
label_to_nid[normalised.lower()] = n["id"]
|
||||
|
||||
seen_call_pairs: set[tuple[str, str]] = set()
|
||||
_SKIP_KEYWORDS = frozenset({
|
||||
"def", "defp", "defmodule", "defmacro", "defmacrop",
|
||||
"defstruct", "defprotocol", "defimpl", "defguard",
|
||||
"alias", "import", "require", "use",
|
||||
"if", "unless", "case", "cond", "with", "for",
|
||||
})
|
||||
|
||||
def walk_calls(node, caller_nid: str) -> None:
|
||||
if node.type != "call":
|
||||
for child in node.children:
|
||||
walk_calls(child, caller_nid)
|
||||
return
|
||||
for child in node.children:
|
||||
if child.type == "identifier":
|
||||
kw = source[child.start_byte:child.end_byte].decode("utf-8", errors="replace")
|
||||
if kw in _SKIP_KEYWORDS:
|
||||
for c in node.children:
|
||||
walk_calls(c, caller_nid)
|
||||
return
|
||||
break
|
||||
callee_name: str | None = None
|
||||
for child in node.children:
|
||||
if child.type == "dot":
|
||||
dot_text = source[child.start_byte:child.end_byte].decode("utf-8", errors="replace")
|
||||
parts = dot_text.rstrip(".").split(".")
|
||||
if parts:
|
||||
callee_name = parts[-1]
|
||||
break
|
||||
if child.type == "identifier":
|
||||
callee_name = source[child.start_byte:child.end_byte].decode("utf-8", errors="replace")
|
||||
break
|
||||
if callee_name:
|
||||
tgt_nid = label_to_nid.get(callee_name.lower())
|
||||
if tgt_nid and tgt_nid != caller_nid:
|
||||
pair = (caller_nid, tgt_nid)
|
||||
if pair not in seen_call_pairs:
|
||||
seen_call_pairs.add(pair)
|
||||
add_edge(caller_nid, tgt_nid, "calls",
|
||||
node.start_point[0] + 1, confidence="INFERRED", weight=0.8)
|
||||
for child in node.children:
|
||||
walk_calls(child, caller_nid)
|
||||
|
||||
for caller_nid, body in function_bodies:
|
||||
walk_calls(body, caller_nid)
|
||||
|
||||
clean_edges = [e for e in edges if e["source"] in seen_ids and
|
||||
(e["target"] in seen_ids or e["relation"] == "imports")]
|
||||
return {"nodes": nodes, "edges": clean_edges, "input_tokens": 0, "output_tokens": 0}
|
||||
|
||||
|
||||
# ── Main extract and collect_files ────────────────────────────────────────────
|
||||
|
||||
def extract(paths: list[Path]) -> dict:
|
||||
@@ -1980,6 +2157,8 @@ def extract(paths: list[Path]) -> dict:
|
||||
".toc": extract_lua,
|
||||
".zig": extract_zig,
|
||||
".ps1": extract_powershell,
|
||||
".ex": extract_elixir,
|
||||
".exs": extract_elixir,
|
||||
}
|
||||
|
||||
for path in paths:
|
||||
|
||||
+2
-1
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "graphifyy"
|
||||
version = "0.3.5"
|
||||
version = "0.3.6"
|
||||
description = "AI coding assistant skill (Claude Code, Codex, OpenCode, OpenClaw) - turn any folder of code, docs, papers, or images into a queryable knowledge graph"
|
||||
readme = "README.md"
|
||||
license = { file = "LICENSE" }
|
||||
@@ -30,6 +30,7 @@ dependencies = [
|
||||
"tree-sitter-lua",
|
||||
"tree-sitter-zig",
|
||||
"tree-sitter-powershell",
|
||||
"tree-sitter-elixir",
|
||||
]
|
||||
|
||||
[project.urls]
|
||||
|
||||
Vendored
+28
@@ -0,0 +1,28 @@
|
||||
defmodule MyApp.Accounts.User do
|
||||
@moduledoc """
|
||||
Handles user accounts and authentication.
|
||||
"""
|
||||
|
||||
alias MyApp.Repo
|
||||
import Ecto.Query
|
||||
|
||||
defstruct [:id, :name, :email]
|
||||
|
||||
def create(attrs) do
|
||||
%__MODULE__{}
|
||||
|> validate(attrs)
|
||||
|> Repo.insert()
|
||||
end
|
||||
|
||||
def find(id) do
|
||||
Repo.get(__MODULE__, id)
|
||||
end
|
||||
|
||||
defp validate(user, attrs) do
|
||||
if Map.has_key?(attrs, :email) do
|
||||
user
|
||||
else
|
||||
{:error, :missing_email}
|
||||
end
|
||||
end
|
||||
end
|
||||
@@ -61,7 +61,7 @@ def test_collect_files_from_dir():
|
||||
supported = {".py", ".js", ".ts", ".tsx", ".go", ".rs",
|
||||
".java", ".c", ".cpp", ".cc", ".cxx", ".rb",
|
||||
".cs", ".kt", ".kts", ".scala", ".php", ".h", ".hpp",
|
||||
".swift", ".lua", ".toc", ".zig", ".ps1"}
|
||||
".swift", ".lua", ".toc", ".zig", ".ps1", ".ex", ".exs"}
|
||||
assert all(f.suffix in supported for f in files)
|
||||
assert len(files) > 0
|
||||
|
||||
|
||||
@@ -337,3 +337,37 @@ def test_swift_emits_calls():
|
||||
r = extract_swift(FIXTURES / "sample.swift")
|
||||
calls = _calls(r)
|
||||
assert any("process" in src and "validate" in tgt for src, tgt in calls)
|
||||
|
||||
|
||||
# ── Elixir ────────────────────────────────────────────────────────────────────
|
||||
|
||||
from graphify.extract import extract_elixir
|
||||
|
||||
def test_elixir_finds_module():
|
||||
r = extract_elixir(FIXTURES / "sample.ex")
|
||||
assert "error" not in r
|
||||
labels = [n["label"] for n in r["nodes"]]
|
||||
assert any("MyApp.Accounts.User" in l for l in labels)
|
||||
|
||||
def test_elixir_finds_functions():
|
||||
r = extract_elixir(FIXTURES / "sample.ex")
|
||||
labels = [n["label"] for n in r["nodes"]]
|
||||
assert any("create" in l for l in labels)
|
||||
assert any("find" in l for l in labels)
|
||||
assert any("validate" in l for l in labels)
|
||||
|
||||
def test_elixir_finds_imports():
|
||||
r = extract_elixir(FIXTURES / "sample.ex")
|
||||
import_edges = [e for e in r["edges"] if e["relation"] == "imports"]
|
||||
assert len(import_edges) >= 2
|
||||
|
||||
def test_elixir_finds_calls():
|
||||
r = extract_elixir(FIXTURES / "sample.ex")
|
||||
calls = {(e["source"], e["target"]) for e in r["edges"] if e["relation"] == "calls"}
|
||||
labels = {n["id"]: n["label"] for n in r["nodes"]}
|
||||
assert any("create" in labels.get(src, "") and "validate" in labels.get(tgt, "") for src, tgt in calls)
|
||||
|
||||
def test_elixir_method_edges():
|
||||
r = extract_elixir(FIXTURES / "sample.ex")
|
||||
methods = [e for e in r["edges"] if e["relation"] == "method"]
|
||||
assert len(methods) >= 3
|
||||
|
||||
Reference in New Issue
Block a user