refactor(extract): move extract_terraform to extractors/terraform.py (verbatim)

graphify/extract.py

@@ -35,6 +35,7 @@
 )
 from graphify.extractors.elixir import extract_elixir  # noqa: F401
 from graphify.extractors.razor import extract_razor  # noqa: F401
+from graphify.extractors.terraform import extract_terraform  # noqa: F401
 from graphify.extractors.zig import extract_zig  # noqa: F401
 from graphify.security import sanitize_metadata
 from graphify.paths import disambiguate_ambiguous_candidates
@@ -15721,184 +15722,6 @@ def extract_dmf(path: Path) -> dict:
     return {"nodes": nodes, "edges": edges}
 
 
-# Head tokens in an HCL traversal that are meta/builtins, not references to a
-# block defined in the corpus (count.index, each.key, self.*, path.module, ...).
-_TF_META_HEADS = frozenset({"count", "each", "self", "path", "terraform"})
-
-
-def extract_terraform(path: Path) -> dict:
-    """Extract Terraform/HCL blocks and the references between them via tree-sitter.
-
-    Nodes: resources, data sources, modules, variables, outputs, providers, and
-    locals. Edges: `contains` (file -> block), `references` (block -> the blocks
-    it interpolates, e.g. `aws_instance.web` -> `var.region`), and `depends_on`
-    (explicit dependency edges).
-
-    Node IDs are scoped by the parent directory, not the file stem, because
-    Terraform resources are module(directory)-scoped: a resource defined in
-    main.tf is referenced from other .tf files in the same directory. Directory
-    scoping lets those cross-file references resolve when per-file extractions
-    are merged (stem scoping would split a definition from its references).
-    """
-    try:
-        import tree_sitter_hcl as tshcl
-        from tree_sitter import Language, Parser
-    except ImportError:
-        return {"nodes": [], "edges": [], "error": "tree_sitter_hcl not installed. Run: pip install tree-sitter-hcl"}
-
-    try:
-        language = Language(tshcl.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)}
-
-    str_path = str(path)
-    file_nid = _make_id(str_path)
-    scope = path.parent.name or "tf"
-
-    nodes: list[dict] = [{"id": file_nid, "label": path.name, "file_type": "code",
-                          "source_file": str_path, "source_location": None}]
-    edges: list[dict] = []
-    seen_ids: set[str] = {file_nid}
-    seen_edges: set[tuple[str, str, str]] = set()
-
-    def _read(n) -> str:
-        return source[n.start_byte:n.end_byte].decode("utf-8", errors="replace")
-
-    def _label_text(n) -> str:
-        return _read(n).strip().strip('"')
-
-    def _add_node(address: str, label: str, line: int) -> str:
-        nid = _make_id(scope, address)
-        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}"})
-            edges.append({"source": file_nid, "target": nid, "relation": "contains",
-                          "confidence": "EXTRACTED", "source_file": str_path,
-                          "source_location": f"L{line}", "weight": 1.0})
-        return nid
-
-    def _add_edge(src: str, address: str, relation: str, line: int) -> None:
-        tgt = _make_id(scope, address)
-        if src == tgt:
-            return
-        key = (src, tgt, relation)
-        if key in seen_edges:
-            return
-        seen_edges.add(key)
-        edges.append({"source": src, "target": tgt, "relation": relation,
-                      "confidence": "EXTRACTED", "source_file": str_path,
-                      "source_location": f"L{line}", "weight": 1.0})
-
-    def _block_parts(block) -> tuple:
-        btype = None
-        labels: list[str] = []
-        for c in block.children:
-            if c.type in ("block_start", "body", "block_end"):
-                break
-            if c.type == "identifier" and btype is None:
-                btype = _read(c)
-            elif c.type in ("string_lit", "identifier"):
-                labels.append(_label_text(c))
-        return btype, labels
-
-    def _ref_address(expr):
-        head = _read(expr)
-        parent = expr.parent
-        attrs: list[str] = []
-        if parent is not None:
-            seen_self = False
-            for c in parent.children:
-                if c.id == expr.id:
-                    seen_self = True
-                    continue
-                if seen_self and c.type == "get_attr":
-                    name = None
-                    for gc in c.children:
-                        if gc.type == "identifier":
-                            name = _read(gc)
-                            break
-                    if name is None:
-                        break
-                    attrs.append(name)
-                elif seen_self and c.type not in ("get_attr",):
-                    break
-        if head in _TF_META_HEADS or not head:
-            return None
-        if head == "var":
-            return f"var.{attrs[0]}" if attrs else None
-        if head == "local":
-            return f"local.{attrs[0]}" if attrs else None
-        if head == "module":
-            return f"module.{attrs[0]}" if attrs else None
-        if head == "data":
-            return f"data.{attrs[0]}.{attrs[1]}" if len(attrs) >= 2 else None
-        return f"{head}.{attrs[0]}" if attrs else None
-
-    def _collect_refs(node, owner_nid: str, relation: str) -> None:
-        rel = relation
-        if node.type == "attribute":
-            key_node = node.child_by_field_name("key") or (
-                node.children[0] if node.children else None
-            )
-            if key_node is not None and _read(key_node) == "depends_on":
-                rel = "depends_on"
-        if node.type == "variable_expr":
-            addr = _ref_address(node)
-            if addr:
-                _add_edge(owner_nid, addr, rel, node.start_point[0] + 1)
-        for c in node.children:
-            if c.is_named:
-                _collect_refs(c, owner_nid, rel)
-
-    def _body_of(block):
-        for c in block.children:
-            if c.type == "body":
-                return c
-        return None
-
-    body = next((c for c in root.children if c.type == "body"), root)
-    for block in body.children:
-        if block.type != "block":
-            continue
-        btype, labels = _block_parts(block)
-        line = block.start_point[0] + 1
-        blk_body = _body_of(block)
-        if btype == "resource" and len(labels) >= 2:
-            owner = _add_node(f"{labels[0]}.{labels[1]}", f"{labels[0]}.{labels[1]}", line)
-        elif btype == "data" and len(labels) >= 2:
-            owner = _add_node(f"data.{labels[0]}.{labels[1]}", f"data.{labels[0]}.{labels[1]}", line)
-        elif btype == "module" and labels:
-            owner = _add_node(f"module.{labels[0]}", f"module.{labels[0]}", line)
-        elif btype == "variable" and labels:
-            owner = _add_node(f"var.{labels[0]}", f"var.{labels[0]}", line)
-        elif btype == "output" and labels:
-            owner = _add_node(f"output.{labels[0]}", f"output.{labels[0]}", line)
-        elif btype == "provider" and labels:
-            owner = _add_node(f"provider.{labels[0]}", f"provider.{labels[0]}", line)
-        elif btype == "locals" and blk_body is not None:
-            for attr in blk_body.children:
-                if attr.type != "attribute":
-                    continue
-                key_node = attr.children[0] if attr.children else None
-                if key_node is None:
-                    continue
-                key = _read(key_node)
-                lnid = _add_node(f"local.{key}", f"local.{key}", attr.start_point[0] + 1)
-                _collect_refs(attr, lnid, "references")
-            continue
-        else:
-            continue
-        if blk_body is not None:
-            _collect_refs(blk_body, owner, "references")
-
-    return {"nodes": nodes, "edges": edges}
-
-
 _DISPATCH: dict[str, Any] = {
     ".py": extract_python,
     ".js": extract_js,

graphify/extractors/MIGRATION.md

@@ -12,7 +12,8 @@ written so an AI agent can execute it in a single session.
 | zig | yes |
 | elixir | yes |
 | razor | yes |
-| (40 more in extract.py) | no |
+| terraform | yes |
+| (39 more in extract.py) | no |
 
 Note: config-driven extractors (python, js, java, c, cpp, ruby, csharp,
 kotlin, scala, php, lua, swift, groovy) depend on the shared

graphify/extractors/__init__.py

@@ -13,11 +13,13 @@
 from graphify.extractors.blade import extract_blade
 from graphify.extractors.elixir import extract_elixir
 from graphify.extractors.razor import extract_razor
+from graphify.extractors.terraform import extract_terraform
 from graphify.extractors.zig import extract_zig
 
 LANGUAGE_EXTRACTORS: dict[str, Callable[[Path], dict]] = {
     "blade": extract_blade,
     "elixir": extract_elixir,
     "razor": extract_razor,
+    "terraform": extract_terraform,
     "zig": extract_zig,
 }