mirror of
https://github.com/safishamsi/graphify.git
synced 2026-07-12 10:27:11 +00:00
refactor(extract): migrate verilog and markdown extractors to extractors/
Continues the extract.py -> extractors/ split. Both are self-contained bespoke extractors (verified: closure-private, byte-identical verbatim move, facade identity + _DISPATCH resolution intact): - extractors/verilog.py: extract_verilog + SystemVerilog helpers (_sv_*, _augment_systemverilog_semantics) and _SV_* constants. - extractors/markdown.py: extract_markdown + _resolve_markdown_link and _MD_* regexes. extract.py 13,121 -> 12,632 LOC. Full suite unchanged: 3036 passed, 29 skipped. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
tpateeq
+2
-491
@@ -41,6 +41,7 @@ from graphify.extractors.elixir import extract_elixir # noqa: F401
|
||||
from graphify.extractors.fortran import _cpp_preprocess, extract_fortran # noqa: F401
|
||||
from graphify.extractors.go import extract_go # noqa: F401
|
||||
from graphify.extractors.json_config import extract_json # noqa: F401
|
||||
from graphify.extractors.markdown import extract_markdown # noqa: F401
|
||||
from graphify.extractors.pascal_forms import extract_delphi_form, extract_lazarus_form # noqa: F401
|
||||
from graphify.extractors.powershell import extract_powershell, extract_powershell_manifest # noqa: F401
|
||||
from graphify.extractors.razor import extract_razor # noqa: F401
|
||||
@@ -48,6 +49,7 @@ from graphify.extractors.rust import extract_rust # noqa: F401
|
||||
from graphify.extractors.sln import extract_sln # noqa: F401
|
||||
from graphify.extractors.sql import extract_sql # noqa: F401
|
||||
from graphify.extractors.terraform import extract_terraform # noqa: F401
|
||||
from graphify.extractors.verilog import extract_verilog # noqa: F401
|
||||
from graphify.extractors.zig import extract_zig # noqa: F401
|
||||
from graphify.security import sanitize_metadata
|
||||
from graphify.paths import disambiguate_ambiguous_candidates
|
||||
@@ -6406,334 +6408,8 @@ def extract_php(path: Path) -> dict:
|
||||
return _extract_generic(path, _PHP_CONFIG)
|
||||
|
||||
|
||||
def _sv_first_identifier(node, source: bytes) -> str | None:
|
||||
"""First `simple_identifier` under node in pre-order, or None.
|
||||
|
||||
tree-sitter-verilog 1.0.3 nests declaration names a few levels deep instead
|
||||
of exposing a `name` field. Scope the search to the right child node (e.g.
|
||||
`function_identifier`) or this returns the return-type instead of the name.
|
||||
"""
|
||||
if node is None:
|
||||
return None
|
||||
for child in node.children:
|
||||
if child.type == "simple_identifier":
|
||||
return _read_text(child, source)
|
||||
found = _sv_first_identifier(child, source)
|
||||
if found:
|
||||
return found
|
||||
return None
|
||||
|
||||
|
||||
def _sv_child(node, type_name: str) -> object | None:
|
||||
if node is None:
|
||||
return None
|
||||
for child in node.children:
|
||||
if child.type == type_name:
|
||||
return child
|
||||
return None
|
||||
|
||||
|
||||
_SV_BUILTIN_TYPES = frozenset({
|
||||
"bit", "logic", "reg", "wire", "int", "integer", "shortint", "longint",
|
||||
"byte", "time", "real", "shortreal", "void", "string", "type", "event",
|
||||
"mailbox", "semaphore", "process", "chandle",
|
||||
})
|
||||
|
||||
_SV_NON_TYPE_WORDS = frozenset({
|
||||
"return", "if", "else", "for", "foreach", "while", "case", "begin", "end",
|
||||
"function", "task", "class", "endclass", "endfunction", "endtask",
|
||||
})
|
||||
|
||||
# One level of balanced parens (e.g. `Foo #(Bar #(int))`) — bounded so malformed
|
||||
# input cannot trigger pathological backtracking.
|
||||
_SV_PARENS_INNER = r"(?:[^()]|\([^()]*\))*"
|
||||
_SV_PARENS = r"\(" + _SV_PARENS_INNER + r"\)"
|
||||
|
||||
_SV_FUNC_RE = re.compile(
|
||||
r"\bfunction\s+([A-Za-z_]\w*(?:\s*#\s*" + _SV_PARENS + r")?)\s+(\w+)\s*"
|
||||
r"\((" + _SV_PARENS_INNER + r")\)\s*;",
|
||||
re.MULTILINE,
|
||||
)
|
||||
|
||||
_SV_PARAM_RE = re.compile(
|
||||
r"\s*(?:input|output|inout|ref|const\s+ref)?\s*"
|
||||
r"([A-Za-z_]\w*(?:\s*#\s*" + _SV_PARENS + r")?)\s+\w+"
|
||||
)
|
||||
|
||||
|
||||
def _sv_strip_comments(text: str) -> str:
|
||||
text = re.sub(r"/\*.*?\*/", "", text, flags=re.DOTALL)
|
||||
return re.sub(r"//.*", "", text)
|
||||
|
||||
|
||||
def _sv_split_type_list(text: str) -> list[str]:
|
||||
parts: list[str] = []
|
||||
depth = 0
|
||||
start = 0
|
||||
for idx, ch in enumerate(text):
|
||||
if ch == "(":
|
||||
depth += 1
|
||||
elif ch == ")":
|
||||
depth = max(0, depth - 1)
|
||||
elif ch == "," and depth == 0:
|
||||
item = text[start:idx].strip()
|
||||
if item:
|
||||
parts.append(item)
|
||||
start = idx + 1
|
||||
item = text[start:].strip()
|
||||
if item:
|
||||
parts.append(item)
|
||||
return parts
|
||||
|
||||
|
||||
def _sv_collect_type_refs(type_text: str, generic: bool = False,
|
||||
skip: frozenset[str] = frozenset()) -> list[tuple[str, str]]:
|
||||
refs: list[tuple[str, str]] = []
|
||||
text = type_text.strip()
|
||||
if not text:
|
||||
return refs
|
||||
head = re.match(r"([A-Za-z_]\w*)", text)
|
||||
if head:
|
||||
name = head.group(1)
|
||||
# `skip` carries the enclosing class's `#(type T = ...)` parameters so
|
||||
# they are not mistaken for referenced types.
|
||||
if name not in _SV_BUILTIN_TYPES and name not in _SV_NON_TYPE_WORDS and name not in skip:
|
||||
refs.append((name, "generic_arg" if generic else "type"))
|
||||
params = re.search(r"#\s*\((" + _SV_PARENS_INNER + r")\)", text)
|
||||
if params:
|
||||
for arg in _sv_split_type_list(params.group(1)):
|
||||
refs.extend(_sv_collect_type_refs(arg, generic=True, skip=skip))
|
||||
return refs
|
||||
|
||||
|
||||
def _augment_systemverilog_semantics(
|
||||
raw: str,
|
||||
stem: str,
|
||||
str_path: str,
|
||||
file_nid: str,
|
||||
nodes: list[dict],
|
||||
edges: list[dict],
|
||||
seen_ids: set[str],
|
||||
) -> None:
|
||||
label_to_nid = {node["label"]: node["id"] for node in nodes}
|
||||
|
||||
def line_for(offset: int) -> int:
|
||||
return raw.count("\n", 0, offset) + 1
|
||||
|
||||
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}",
|
||||
"confidence_score": 1.0})
|
||||
label_to_nid[label] = nid
|
||||
|
||||
def ensure_type(label: str, line: int) -> str:
|
||||
if label in label_to_nid:
|
||||
return label_to_nid[label]
|
||||
nid = _make_id(stem, label)
|
||||
add_node(nid, label, line)
|
||||
return nid
|
||||
|
||||
def add_edge(src: str, target_label: str, relation: str, line: int, context: str | None = None) -> None:
|
||||
tgt = ensure_type(target_label, line)
|
||||
edge = {"source": src, "target": tgt, "relation": relation,
|
||||
"confidence": "EXTRACTED", "confidence_score": 1.0,
|
||||
"source_file": str_path, "source_location": f"L{line}", "weight": 1.0}
|
||||
if context:
|
||||
edge["context"] = context
|
||||
edges.append(edge)
|
||||
|
||||
text = _sv_strip_comments(raw)
|
||||
# Consuming `endclass` (rather than a lookahead) makes each match own its
|
||||
# terminator, so back-to-back or malformed classes cannot bleed bodies.
|
||||
class_re = re.compile(
|
||||
r"\b(?:(interface)\s+)?class\s+(\w+)([^;{]*)\s*;(.*?)\bendclass\b",
|
||||
re.DOTALL,
|
||||
)
|
||||
for match in class_re.finditer(text):
|
||||
class_name = match.group(2)
|
||||
header = match.group(3) or ""
|
||||
body = match.group(4) or ""
|
||||
line = line_for(match.start())
|
||||
# `#(type T = Payload)` declares `T` as a class type parameter, not a
|
||||
# referenced type — collect these to skip below.
|
||||
type_params = frozenset(re.findall(r"\btype\s+(\w+)", header))
|
||||
class_nid = _make_id(stem, class_name)
|
||||
add_node(class_nid, class_name, line)
|
||||
edges.append({"source": file_nid, "target": class_nid, "relation": "defines",
|
||||
"confidence": "EXTRACTED", "confidence_score": 1.0,
|
||||
"source_file": str_path, "source_location": f"L{line}", "weight": 1.0})
|
||||
|
||||
ext = re.search(r"\bextends\s+(\w+)", header)
|
||||
if ext:
|
||||
add_edge(class_nid, ext.group(1), "inherits", line)
|
||||
impl = re.search(r"\bimplements\s+([^;{]+)", header)
|
||||
if impl:
|
||||
for iface_name in _sv_split_type_list(impl.group(1)):
|
||||
add_edge(class_nid, iface_name.split("#", 1)[0].strip(), "implements", line)
|
||||
|
||||
body_without_functions = re.sub(
|
||||
r"\bfunction\b.*?\bendfunction\b",
|
||||
lambda m: "\n" * m.group(0).count("\n"),
|
||||
body,
|
||||
flags=re.DOTALL,
|
||||
)
|
||||
# Optional leading class-property qualifiers (rand/local/protected/etc.)
|
||||
# must be consumed: otherwise a qualified field like `rand Config x;`
|
||||
# (three tokens) fails the `<type> <name>;` shape and its type reference
|
||||
# is silently dropped.
|
||||
for field in re.finditer(r"^\s*(?:(?:rand|randc|local|protected|static|const|automatic|var)\s+)*([A-Za-z_]\w*(?:\s*#\s*\([^;]+?\))?)\s+\w+\s*;", body_without_functions, re.MULTILINE):
|
||||
# Count to the start of the type token (group 1), not the match
|
||||
# start: `^\s*` consumes the leading newline(s), so field.start()
|
||||
# would resolve to the class's line instead of the field's.
|
||||
field_line = line + body_without_functions.count("\n", 0, field.start(1))
|
||||
for ref_name, role in _sv_collect_type_refs(field.group(1), skip=type_params):
|
||||
add_edge(class_nid, ref_name, "references", field_line, "generic_arg" if role == "generic_arg" else "field")
|
||||
|
||||
for fm in _SV_FUNC_RE.finditer(body):
|
||||
return_type, func_name, params = fm.group(1), fm.group(2), fm.group(3)
|
||||
func_line = line + body.count("\n", 0, fm.start())
|
||||
func_nid = _make_id(class_nid, func_name)
|
||||
add_node(func_nid, func_name, func_line)
|
||||
edges.append({"source": class_nid, "target": func_nid, "relation": "method",
|
||||
"confidence": "EXTRACTED", "confidence_score": 1.0,
|
||||
"source_file": str_path, "source_location": f"L{func_line}", "weight": 1.0})
|
||||
for ref_name, role in _sv_collect_type_refs(return_type, skip=type_params):
|
||||
add_edge(func_nid, ref_name, "references", func_line, "generic_arg" if role == "generic_arg" else "return_type")
|
||||
for param in _sv_split_type_list(params):
|
||||
pm = _SV_PARAM_RE.match(param)
|
||||
if not pm:
|
||||
continue
|
||||
for ref_name, role in _sv_collect_type_refs(pm.group(1), skip=type_params):
|
||||
add_edge(func_nid, ref_name, "references", func_line, "generic_arg" if role == "generic_arg" else "parameter_type")
|
||||
|
||||
|
||||
def extract_verilog(path: Path) -> dict:
|
||||
"""Extract modules, functions, tasks, package imports, instantiations, and
|
||||
SystemVerilog class semantics (inherits/implements edges, field/parameter/
|
||||
return-type references) from .v/.sv files."""
|
||||
try:
|
||||
import tree_sitter_verilog as tsverilog
|
||||
from tree_sitter import Language, Parser
|
||||
except ImportError:
|
||||
return {"nodes": [], "edges": [], "error": "tree_sitter_verilog not installed"}
|
||||
|
||||
try:
|
||||
language = Language(tsverilog.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 = _file_stem(path)
|
||||
str_path = str(path)
|
||||
nodes: list[dict] = []
|
||||
edges: list[dict] = []
|
||||
seen_ids: set[str] = set()
|
||||
|
||||
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}",
|
||||
"confidence_score": 1.0})
|
||||
|
||||
def add_edge(src: str, tgt: str, relation: str, line: int,
|
||||
confidence: str = "EXTRACTED", score: float = 1.0) -> None:
|
||||
edges.append({"source": src, "target": tgt, "relation": relation,
|
||||
"confidence": confidence, "confidence_score": score,
|
||||
"source_file": str_path, "source_location": f"L{line}", "weight": 1.0})
|
||||
|
||||
file_nid = _make_id(str(path))
|
||||
add_node(file_nid, path.name, 1)
|
||||
|
||||
def walk(node, module_nid: str | None = None) -> None:
|
||||
t = node.type
|
||||
|
||||
# SystemVerilog class bodies are handled by _augment_systemverilog_semantics
|
||||
# (regex over source text). Skip their subtrees so in-class methods are not
|
||||
# double-emitted here — and with the wrong, return-type-derived name.
|
||||
if t in ("class_declaration", "interface_class_declaration"):
|
||||
return
|
||||
|
||||
if t == "module_declaration":
|
||||
mod_name = _sv_first_identifier(_sv_child(node, "module_header"), source)
|
||||
if mod_name:
|
||||
line = node.start_point[0] + 1
|
||||
nid = _make_id(stem, mod_name)
|
||||
add_node(nid, mod_name, line)
|
||||
add_edge(file_nid, nid, "defines", line)
|
||||
for child in node.children:
|
||||
walk(child, nid)
|
||||
return
|
||||
|
||||
# `function_prototype` only appears inside class/interface-class bodies
|
||||
# (skipped above) and nests its name differently; it is intentionally not
|
||||
# handled here.
|
||||
elif t == "function_declaration":
|
||||
fn_body = _sv_child(node, "function_body_declaration")
|
||||
func_name = _sv_first_identifier(_sv_child(fn_body, "function_identifier"), source)
|
||||
if func_name:
|
||||
line = node.start_point[0] + 1
|
||||
parent = module_nid or file_nid
|
||||
nid = _make_id(parent, func_name)
|
||||
add_node(nid, f"{func_name}()", line)
|
||||
add_edge(parent, nid, "contains", line)
|
||||
|
||||
elif t == "task_declaration":
|
||||
tk_body = _sv_child(node, "task_body_declaration")
|
||||
task_name = _sv_first_identifier(_sv_child(tk_body, "task_identifier"), source)
|
||||
if task_name:
|
||||
line = node.start_point[0] + 1
|
||||
parent = module_nid or file_nid
|
||||
nid = _make_id(parent, task_name)
|
||||
add_node(nid, task_name, line)
|
||||
add_edge(parent, nid, "contains", line)
|
||||
|
||||
elif t == "package_import_declaration":
|
||||
for child in node.children:
|
||||
if child.type == "package_import_item":
|
||||
pkg_text = _read_text(child, source)
|
||||
pkg_name = pkg_text.split("::")[0].strip()
|
||||
if pkg_name:
|
||||
line = node.start_point[0] + 1
|
||||
tgt_nid = _make_id(pkg_name)
|
||||
add_node(tgt_nid, pkg_name, line)
|
||||
src_nid = module_nid or file_nid
|
||||
add_edge(src_nid, tgt_nid, "imports_from", line)
|
||||
|
||||
elif t in ("module_instantiation", "checker_instantiation"):
|
||||
# `leaf u_leaf();` parses as checker_instantiation in 1.0.3;
|
||||
# module_instantiation (when it occurs) exposes a `module_type` field.
|
||||
# Both reduce to the first identifier under the node — the instantiated
|
||||
# type, not the instance name (which appears later).
|
||||
if module_nid:
|
||||
type_node = node.child_by_field_name("module_type")
|
||||
inst_type = (_read_text(type_node, source).strip() if type_node
|
||||
else _sv_first_identifier(node, source))
|
||||
if inst_type:
|
||||
line = node.start_point[0] + 1
|
||||
tgt_nid = _make_id(inst_type)
|
||||
add_node(tgt_nid, inst_type, line)
|
||||
add_edge(module_nid, tgt_nid, "instantiates", line)
|
||||
|
||||
for child in node.children:
|
||||
walk(child, module_nid)
|
||||
|
||||
walk(root)
|
||||
_augment_systemverilog_semantics(
|
||||
source.decode("utf-8", errors="replace"),
|
||||
stem,
|
||||
str_path,
|
||||
file_nid,
|
||||
nodes,
|
||||
edges,
|
||||
seen_ids,
|
||||
)
|
||||
return {"nodes": nodes, "edges": edges}
|
||||
|
||||
|
||||
def extract_lua(path: Path) -> dict:
|
||||
@@ -10261,177 +9937,12 @@ def extract_objc(path: Path) -> dict:
|
||||
# Inline markdown link: [text](target "optional title"). The negative lookbehind
|
||||
# excludes images (). The target stops at whitespace/closing paren so
|
||||
# an optional "title" after the URL is dropped; an optional <...> wrapper is too.
|
||||
_MD_INLINE_LINK_RE = re.compile(r'(?<!\!)\[[^\]]*\]\(\s*<?([^)\s>]+)>?(?:\s+[^)]*)?\)')
|
||||
# Reference-style link definition line: [label]: target "optional title"
|
||||
_MD_REF_DEF_RE = re.compile(r'^\s{0,3}\[[^\]]+\]:\s*<?([^\s>]+)>?')
|
||||
# Obsidian-style wikilink: [[target]] / [[target|alias]] / [[target#anchor]].
|
||||
_MD_WIKILINK_RE = re.compile(r'(?<!\!)\[\[([^\]|#]+)(?:[#|][^\]]*)?\]\]')
|
||||
|
||||
# Extensions graphify creates document file nodes for. A link to one of these
|
||||
# resolves to that file's node; links to code/assets are skipped (left to the
|
||||
# language extractors).
|
||||
_MD_LINKABLE_EXTS = {".md", ".mdx", ".qmd", ".markdown", ".rst", ".txt"}
|
||||
|
||||
|
||||
def _resolve_markdown_link(raw: str, source_dir: Path) -> "Path | None":
|
||||
"""Resolve a markdown link target to the absolute path of a sibling document.
|
||||
|
||||
Returns the resolved (normalized, not necessarily existing) path when the
|
||||
target is a *local* relative/absolute file-path link to a document, or None
|
||||
when it should be skipped: external URLs (http/https/mailto/protocol-
|
||||
relative/data), pure in-page anchors (``#section``), and links to non-doc
|
||||
file types (code/assets are handled by their own extractors).
|
||||
|
||||
The anchor fragment (``#section``) and query (``?x=1``) are stripped before
|
||||
resolution so ``./repo.md#setup`` resolves to the same node as ``./repo.md``.
|
||||
Extension-less targets (typical of wikilinks) are treated as sibling ``.md``.
|
||||
"""
|
||||
target = raw.strip()
|
||||
if not target:
|
||||
return None
|
||||
# Drop anchor / query so #section links still resolve to the target doc.
|
||||
target = target.split("#", 1)[0].split("?", 1)[0].strip()
|
||||
if not target:
|
||||
return None
|
||||
low = target.lower()
|
||||
if "://" in target or low.startswith(("mailto:", "tel:", "//", "data:")):
|
||||
return None
|
||||
suffix = Path(target).suffix.lower()
|
||||
if suffix == "":
|
||||
target = target + ".md"
|
||||
suffix = ".md"
|
||||
if suffix not in _MD_LINKABLE_EXTS:
|
||||
return None
|
||||
candidate = Path(target)
|
||||
if not candidate.is_absolute():
|
||||
candidate = source_dir / candidate
|
||||
return Path(os.path.normpath(str(candidate)))
|
||||
|
||||
|
||||
def extract_markdown(path: Path) -> dict:
|
||||
"""Extract structural nodes and edges from a Markdown file.
|
||||
|
||||
Produces nodes for:
|
||||
- The file itself
|
||||
- Each heading (# / ## / ### etc.)
|
||||
|
||||
Produces edges for:
|
||||
- file --contains--> heading
|
||||
- parent heading --contains--> child heading (nesting by level)
|
||||
- heading --references--> other node (when backtick `Name` matches a known pattern)
|
||||
- file --references--> linked document, for inline ``[text](./other.md)``,
|
||||
reference-style ``[label]: ./other.md`` and ``[[wikilink]]`` links, so a
|
||||
hub doc (``index.md`` / ``table-of-contents.md``) becomes a real hub node
|
||||
instead of an under-connected orphan (#1376). The target node ID is built
|
||||
from the resolved target path with the same recipe as the target file's
|
||||
own node, so the edge merges into that node (no ghost node). External
|
||||
URLs, in-page anchors, images and non-document targets are skipped.
|
||||
|
||||
Fenced code blocks (``` ... ```) are skipped during parsing so their
|
||||
contents don't get treated as headings, but no node is emitted for
|
||||
them — they were always orphans (only a single contains edge to the
|
||||
parent doc) and inflated the disconnected-component count (#1077).
|
||||
|
||||
No tree-sitter dependency — pure line-by-line parsing.
|
||||
"""
|
||||
try:
|
||||
source = path.read_text(encoding="utf-8", errors="replace")
|
||||
except Exception as e:
|
||||
return {"nodes": [], "edges": [], "error": str(e)}
|
||||
|
||||
stem = _file_stem(path)
|
||||
str_path = str(path)
|
||||
nodes: list[dict] = []
|
||||
edges: list[dict] = []
|
||||
seen_ids: set[str] = set()
|
||||
|
||||
def add_node(nid: str, label: str, line: int, file_type: str = "document") -> None:
|
||||
if nid not in seen_ids:
|
||||
seen_ids.add(nid)
|
||||
nodes.append({"id": nid, "label": label, "file_type": file_type,
|
||||
"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(str(path))
|
||||
add_node(file_nid, path.name, 1)
|
||||
|
||||
source_dir = path.parent
|
||||
# Dedup link edges by resolved target node so a hub doc that links to the
|
||||
# same sibling many times yields one edge, not N (keeps weights meaningful).
|
||||
linked_targets: set[str] = set()
|
||||
|
||||
def add_link(raw: str, line: int) -> None:
|
||||
resolved = _resolve_markdown_link(raw, source_dir)
|
||||
if resolved is None:
|
||||
return
|
||||
# Build the target ID with the SAME recipe as the target file's own
|
||||
# node (_make_id(str(path)) at extract time, canonicalized to
|
||||
# _file_node_id(rel) by the extract() post-pass). Using the absolute
|
||||
# resolved path means both endpoints get remapped identically, so the
|
||||
# edge merges into the existing doc node instead of spawning a ghost.
|
||||
tgt_nid = _make_id(str(resolved))
|
||||
if tgt_nid == file_nid or tgt_nid in linked_targets:
|
||||
return
|
||||
linked_targets.add(tgt_nid)
|
||||
add_edge(file_nid, tgt_nid, "references", line)
|
||||
|
||||
# Track heading stack for nesting: [(level, nid), ...]
|
||||
heading_stack: list[tuple[int, str]] = []
|
||||
in_code_block = False
|
||||
|
||||
lines = source.splitlines()
|
||||
for line_num_0, line_text in enumerate(lines):
|
||||
line_num = line_num_0 + 1
|
||||
|
||||
# Skip over fenced code blocks so their contents are not parsed as
|
||||
# headings, but do not emit nodes/edges for them (#1077).
|
||||
stripped = line_text.strip()
|
||||
if stripped.startswith("```"):
|
||||
in_code_block = not in_code_block
|
||||
continue
|
||||
|
||||
if in_code_block:
|
||||
continue
|
||||
|
||||
# Markdown links -> document references (#1376). Scanned on every
|
||||
# non-fenced line (including heading lines, which the heading branch
|
||||
# below `continue`s past) so links anywhere in the doc are captured.
|
||||
for m in _MD_INLINE_LINK_RE.finditer(line_text):
|
||||
add_link(m.group(1), line_num)
|
||||
for m in _MD_WIKILINK_RE.finditer(line_text):
|
||||
add_link(m.group(1), line_num)
|
||||
ref_def = _MD_REF_DEF_RE.match(line_text)
|
||||
if ref_def:
|
||||
add_link(ref_def.group(1), line_num)
|
||||
|
||||
# Detect headings: # Heading, ## Heading, etc.
|
||||
heading_match = re.match(r'^(#{1,6})\s+(.+)', line_text)
|
||||
if heading_match:
|
||||
level = len(heading_match.group(1))
|
||||
title = heading_match.group(2).strip()
|
||||
h_nid = _make_id(stem, title)
|
||||
# Avoid duplicate heading IDs by appending line number
|
||||
if h_nid in seen_ids:
|
||||
h_nid = _make_id(stem, title, str(line_num))
|
||||
add_node(h_nid, title, line_num)
|
||||
|
||||
# Pop headings at same or deeper level
|
||||
while heading_stack and heading_stack[-1][0] >= level:
|
||||
heading_stack.pop()
|
||||
|
||||
# Connect to parent heading or file
|
||||
parent = heading_stack[-1][1] if heading_stack else file_nid
|
||||
add_edge(parent, h_nid, "contains", line_num)
|
||||
|
||||
heading_stack.append((level, h_nid))
|
||||
continue
|
||||
|
||||
return {"nodes": nodes, "edges": edges, "input_tokens": 0, "output_tokens": 0}
|
||||
|
||||
|
||||
# ── Pascal / Delphi extractor ─────────────────────────────────────────────────
|
||||
|
||||
@@ -19,6 +19,7 @@ from graphify.extractors.elixir import extract_elixir
|
||||
from graphify.extractors.fortran import extract_fortran
|
||||
from graphify.extractors.go import extract_go
|
||||
from graphify.extractors.json_config import extract_json
|
||||
from graphify.extractors.markdown import extract_markdown
|
||||
from graphify.extractors.pascal_forms import extract_delphi_form, extract_lazarus_form
|
||||
from graphify.extractors.powershell import extract_powershell, extract_powershell_manifest
|
||||
from graphify.extractors.razor import extract_razor
|
||||
@@ -26,6 +27,7 @@ from graphify.extractors.rust import extract_rust
|
||||
from graphify.extractors.sln import extract_sln
|
||||
from graphify.extractors.sql import extract_sql
|
||||
from graphify.extractors.terraform import extract_terraform
|
||||
from graphify.extractors.verilog import extract_verilog
|
||||
from graphify.extractors.zig import extract_zig
|
||||
|
||||
LANGUAGE_EXTRACTORS: dict[str, Callable[[Path], dict]] = {
|
||||
@@ -43,6 +45,7 @@ LANGUAGE_EXTRACTORS: dict[str, Callable[[Path], dict]] = {
|
||||
"go": extract_go,
|
||||
"json": extract_json,
|
||||
"lazarus_form": extract_lazarus_form,
|
||||
"markdown": extract_markdown,
|
||||
"powershell": extract_powershell,
|
||||
"powershell_manifest": extract_powershell_manifest,
|
||||
"razor": extract_razor,
|
||||
@@ -50,5 +53,6 @@ LANGUAGE_EXTRACTORS: dict[str, Callable[[Path], dict]] = {
|
||||
"sln": extract_sln,
|
||||
"sql": extract_sql,
|
||||
"terraform": extract_terraform,
|
||||
"verilog": extract_verilog,
|
||||
"zig": extract_zig,
|
||||
}
|
||||
|
||||
@@ -0,0 +1,176 @@
|
||||
"""Markdown extractor. Moved verbatim from graphify/extract.py."""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import os
|
||||
|
||||
from pathlib import Path
|
||||
from graphify.extractors.base import _file_stem, _make_id
|
||||
|
||||
|
||||
_MD_INLINE_LINK_RE = re.compile(r'(?<!\!)\[[^\]]*\]\(\s*<?([^)\s>]+)>?(?:\s+[^)]*)?\)')
|
||||
|
||||
_MD_REF_DEF_RE = re.compile(r'^\s{0,3}\[[^\]]+\]:\s*<?([^\s>]+)>?')
|
||||
|
||||
_MD_WIKILINK_RE = re.compile(r'(?<!\!)\[\[([^\]|#]+)(?:[#|][^\]]*)?\]\]')
|
||||
|
||||
_MD_LINKABLE_EXTS = {".md", ".mdx", ".qmd", ".markdown", ".rst", ".txt"}
|
||||
|
||||
def _resolve_markdown_link(raw: str, source_dir: Path) -> "Path | None":
|
||||
"""Resolve a markdown link target to the absolute path of a sibling document.
|
||||
|
||||
Returns the resolved (normalized, not necessarily existing) path when the
|
||||
target is a *local* relative/absolute file-path link to a document, or None
|
||||
when it should be skipped: external URLs (http/https/mailto/protocol-
|
||||
relative/data), pure in-page anchors (``#section``), and links to non-doc
|
||||
file types (code/assets are handled by their own extractors).
|
||||
|
||||
The anchor fragment (``#section``) and query (``?x=1``) are stripped before
|
||||
resolution so ``./repo.md#setup`` resolves to the same node as ``./repo.md``.
|
||||
Extension-less targets (typical of wikilinks) are treated as sibling ``.md``.
|
||||
"""
|
||||
target = raw.strip()
|
||||
if not target:
|
||||
return None
|
||||
# Drop anchor / query so #section links still resolve to the target doc.
|
||||
target = target.split("#", 1)[0].split("?", 1)[0].strip()
|
||||
if not target:
|
||||
return None
|
||||
low = target.lower()
|
||||
if "://" in target or low.startswith(("mailto:", "tel:", "//", "data:")):
|
||||
return None
|
||||
suffix = Path(target).suffix.lower()
|
||||
if suffix == "":
|
||||
target = target + ".md"
|
||||
suffix = ".md"
|
||||
if suffix not in _MD_LINKABLE_EXTS:
|
||||
return None
|
||||
candidate = Path(target)
|
||||
if not candidate.is_absolute():
|
||||
candidate = source_dir / candidate
|
||||
return Path(os.path.normpath(str(candidate)))
|
||||
|
||||
def extract_markdown(path: Path) -> dict:
|
||||
"""Extract structural nodes and edges from a Markdown file.
|
||||
|
||||
Produces nodes for:
|
||||
- The file itself
|
||||
- Each heading (# / ## / ### etc.)
|
||||
|
||||
Produces edges for:
|
||||
- file --contains--> heading
|
||||
- parent heading --contains--> child heading (nesting by level)
|
||||
- heading --references--> other node (when backtick `Name` matches a known pattern)
|
||||
- file --references--> linked document, for inline ``[text](./other.md)``,
|
||||
reference-style ``[label]: ./other.md`` and ``[[wikilink]]`` links, so a
|
||||
hub doc (``index.md`` / ``table-of-contents.md``) becomes a real hub node
|
||||
instead of an under-connected orphan (#1376). The target node ID is built
|
||||
from the resolved target path with the same recipe as the target file's
|
||||
own node, so the edge merges into that node (no ghost node). External
|
||||
URLs, in-page anchors, images and non-document targets are skipped.
|
||||
|
||||
Fenced code blocks (``` ... ```) are skipped during parsing so their
|
||||
contents don't get treated as headings, but no node is emitted for
|
||||
them — they were always orphans (only a single contains edge to the
|
||||
parent doc) and inflated the disconnected-component count (#1077).
|
||||
|
||||
No tree-sitter dependency — pure line-by-line parsing.
|
||||
"""
|
||||
try:
|
||||
source = path.read_text(encoding="utf-8", errors="replace")
|
||||
except Exception as e:
|
||||
return {"nodes": [], "edges": [], "error": str(e)}
|
||||
|
||||
stem = _file_stem(path)
|
||||
str_path = str(path)
|
||||
nodes: list[dict] = []
|
||||
edges: list[dict] = []
|
||||
seen_ids: set[str] = set()
|
||||
|
||||
def add_node(nid: str, label: str, line: int, file_type: str = "document") -> None:
|
||||
if nid not in seen_ids:
|
||||
seen_ids.add(nid)
|
||||
nodes.append({"id": nid, "label": label, "file_type": file_type,
|
||||
"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(str(path))
|
||||
add_node(file_nid, path.name, 1)
|
||||
|
||||
source_dir = path.parent
|
||||
# Dedup link edges by resolved target node so a hub doc that links to the
|
||||
# same sibling many times yields one edge, not N (keeps weights meaningful).
|
||||
linked_targets: set[str] = set()
|
||||
|
||||
def add_link(raw: str, line: int) -> None:
|
||||
resolved = _resolve_markdown_link(raw, source_dir)
|
||||
if resolved is None:
|
||||
return
|
||||
# Build the target ID with the SAME recipe as the target file's own
|
||||
# node (_make_id(str(path)) at extract time, canonicalized to
|
||||
# _file_node_id(rel) by the extract() post-pass). Using the absolute
|
||||
# resolved path means both endpoints get remapped identically, so the
|
||||
# edge merges into the existing doc node instead of spawning a ghost.
|
||||
tgt_nid = _make_id(str(resolved))
|
||||
if tgt_nid == file_nid or tgt_nid in linked_targets:
|
||||
return
|
||||
linked_targets.add(tgt_nid)
|
||||
add_edge(file_nid, tgt_nid, "references", line)
|
||||
|
||||
# Track heading stack for nesting: [(level, nid), ...]
|
||||
heading_stack: list[tuple[int, str]] = []
|
||||
in_code_block = False
|
||||
|
||||
lines = source.splitlines()
|
||||
for line_num_0, line_text in enumerate(lines):
|
||||
line_num = line_num_0 + 1
|
||||
|
||||
# Skip over fenced code blocks so their contents are not parsed as
|
||||
# headings, but do not emit nodes/edges for them (#1077).
|
||||
stripped = line_text.strip()
|
||||
if stripped.startswith("```"):
|
||||
in_code_block = not in_code_block
|
||||
continue
|
||||
|
||||
if in_code_block:
|
||||
continue
|
||||
|
||||
# Markdown links -> document references (#1376). Scanned on every
|
||||
# non-fenced line (including heading lines, which the heading branch
|
||||
# below `continue`s past) so links anywhere in the doc are captured.
|
||||
for m in _MD_INLINE_LINK_RE.finditer(line_text):
|
||||
add_link(m.group(1), line_num)
|
||||
for m in _MD_WIKILINK_RE.finditer(line_text):
|
||||
add_link(m.group(1), line_num)
|
||||
ref_def = _MD_REF_DEF_RE.match(line_text)
|
||||
if ref_def:
|
||||
add_link(ref_def.group(1), line_num)
|
||||
|
||||
# Detect headings: # Heading, ## Heading, etc.
|
||||
heading_match = re.match(r'^(#{1,6})\s+(.+)', line_text)
|
||||
if heading_match:
|
||||
level = len(heading_match.group(1))
|
||||
title = heading_match.group(2).strip()
|
||||
h_nid = _make_id(stem, title)
|
||||
# Avoid duplicate heading IDs by appending line number
|
||||
if h_nid in seen_ids:
|
||||
h_nid = _make_id(stem, title, str(line_num))
|
||||
add_node(h_nid, title, line_num)
|
||||
|
||||
# Pop headings at same or deeper level
|
||||
while heading_stack and heading_stack[-1][0] >= level:
|
||||
heading_stack.pop()
|
||||
|
||||
# Connect to parent heading or file
|
||||
parent = heading_stack[-1][1] if heading_stack else file_nid
|
||||
add_edge(parent, h_nid, "contains", line_num)
|
||||
|
||||
heading_stack.append((level, h_nid))
|
||||
continue
|
||||
|
||||
return {"nodes": nodes, "edges": edges, "input_tokens": 0, "output_tokens": 0}
|
||||
@@ -0,0 +1,329 @@
|
||||
"""Verilog extractor. Moved verbatim from graphify/extract.py."""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
from pathlib import Path
|
||||
from graphify.extractors.base import _file_stem, _make_id, _read_text
|
||||
|
||||
|
||||
def _sv_first_identifier(node, source: bytes) -> str | None:
|
||||
"""First `simple_identifier` under node in pre-order, or None.
|
||||
|
||||
tree-sitter-verilog 1.0.3 nests declaration names a few levels deep instead
|
||||
of exposing a `name` field. Scope the search to the right child node (e.g.
|
||||
`function_identifier`) or this returns the return-type instead of the name.
|
||||
"""
|
||||
if node is None:
|
||||
return None
|
||||
for child in node.children:
|
||||
if child.type == "simple_identifier":
|
||||
return _read_text(child, source)
|
||||
found = _sv_first_identifier(child, source)
|
||||
if found:
|
||||
return found
|
||||
return None
|
||||
|
||||
def _sv_child(node, type_name: str) -> object | None:
|
||||
if node is None:
|
||||
return None
|
||||
for child in node.children:
|
||||
if child.type == type_name:
|
||||
return child
|
||||
return None
|
||||
|
||||
_SV_BUILTIN_TYPES = frozenset({
|
||||
"bit", "logic", "reg", "wire", "int", "integer", "shortint", "longint",
|
||||
"byte", "time", "real", "shortreal", "void", "string", "type", "event",
|
||||
"mailbox", "semaphore", "process", "chandle",
|
||||
})
|
||||
|
||||
_SV_NON_TYPE_WORDS = frozenset({
|
||||
"return", "if", "else", "for", "foreach", "while", "case", "begin", "end",
|
||||
"function", "task", "class", "endclass", "endfunction", "endtask",
|
||||
})
|
||||
|
||||
_SV_PARENS_INNER = r"(?:[^()]|\([^()]*\))*"
|
||||
|
||||
_SV_PARENS = r"\(" + _SV_PARENS_INNER + r"\)"
|
||||
|
||||
_SV_FUNC_RE = re.compile(
|
||||
r"\bfunction\s+([A-Za-z_]\w*(?:\s*#\s*" + _SV_PARENS + r")?)\s+(\w+)\s*"
|
||||
r"\((" + _SV_PARENS_INNER + r")\)\s*;",
|
||||
re.MULTILINE,
|
||||
)
|
||||
|
||||
_SV_PARAM_RE = re.compile(
|
||||
r"\s*(?:input|output|inout|ref|const\s+ref)?\s*"
|
||||
r"([A-Za-z_]\w*(?:\s*#\s*" + _SV_PARENS + r")?)\s+\w+"
|
||||
)
|
||||
|
||||
def _sv_strip_comments(text: str) -> str:
|
||||
text = re.sub(r"/\*.*?\*/", "", text, flags=re.DOTALL)
|
||||
return re.sub(r"//.*", "", text)
|
||||
|
||||
def _sv_split_type_list(text: str) -> list[str]:
|
||||
parts: list[str] = []
|
||||
depth = 0
|
||||
start = 0
|
||||
for idx, ch in enumerate(text):
|
||||
if ch == "(":
|
||||
depth += 1
|
||||
elif ch == ")":
|
||||
depth = max(0, depth - 1)
|
||||
elif ch == "," and depth == 0:
|
||||
item = text[start:idx].strip()
|
||||
if item:
|
||||
parts.append(item)
|
||||
start = idx + 1
|
||||
item = text[start:].strip()
|
||||
if item:
|
||||
parts.append(item)
|
||||
return parts
|
||||
|
||||
def _sv_collect_type_refs(type_text: str, generic: bool = False,
|
||||
skip: frozenset[str] = frozenset()) -> list[tuple[str, str]]:
|
||||
refs: list[tuple[str, str]] = []
|
||||
text = type_text.strip()
|
||||
if not text:
|
||||
return refs
|
||||
head = re.match(r"([A-Za-z_]\w*)", text)
|
||||
if head:
|
||||
name = head.group(1)
|
||||
# `skip` carries the enclosing class's `#(type T = ...)` parameters so
|
||||
# they are not mistaken for referenced types.
|
||||
if name not in _SV_BUILTIN_TYPES and name not in _SV_NON_TYPE_WORDS and name not in skip:
|
||||
refs.append((name, "generic_arg" if generic else "type"))
|
||||
params = re.search(r"#\s*\((" + _SV_PARENS_INNER + r")\)", text)
|
||||
if params:
|
||||
for arg in _sv_split_type_list(params.group(1)):
|
||||
refs.extend(_sv_collect_type_refs(arg, generic=True, skip=skip))
|
||||
return refs
|
||||
|
||||
def _augment_systemverilog_semantics(
|
||||
raw: str,
|
||||
stem: str,
|
||||
str_path: str,
|
||||
file_nid: str,
|
||||
nodes: list[dict],
|
||||
edges: list[dict],
|
||||
seen_ids: set[str],
|
||||
) -> None:
|
||||
label_to_nid = {node["label"]: node["id"] for node in nodes}
|
||||
|
||||
def line_for(offset: int) -> int:
|
||||
return raw.count("\n", 0, offset) + 1
|
||||
|
||||
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}",
|
||||
"confidence_score": 1.0})
|
||||
label_to_nid[label] = nid
|
||||
|
||||
def ensure_type(label: str, line: int) -> str:
|
||||
if label in label_to_nid:
|
||||
return label_to_nid[label]
|
||||
nid = _make_id(stem, label)
|
||||
add_node(nid, label, line)
|
||||
return nid
|
||||
|
||||
def add_edge(src: str, target_label: str, relation: str, line: int, context: str | None = None) -> None:
|
||||
tgt = ensure_type(target_label, line)
|
||||
edge = {"source": src, "target": tgt, "relation": relation,
|
||||
"confidence": "EXTRACTED", "confidence_score": 1.0,
|
||||
"source_file": str_path, "source_location": f"L{line}", "weight": 1.0}
|
||||
if context:
|
||||
edge["context"] = context
|
||||
edges.append(edge)
|
||||
|
||||
text = _sv_strip_comments(raw)
|
||||
# Consuming `endclass` (rather than a lookahead) makes each match own its
|
||||
# terminator, so back-to-back or malformed classes cannot bleed bodies.
|
||||
class_re = re.compile(
|
||||
r"\b(?:(interface)\s+)?class\s+(\w+)([^;{]*)\s*;(.*?)\bendclass\b",
|
||||
re.DOTALL,
|
||||
)
|
||||
for match in class_re.finditer(text):
|
||||
class_name = match.group(2)
|
||||
header = match.group(3) or ""
|
||||
body = match.group(4) or ""
|
||||
line = line_for(match.start())
|
||||
# `#(type T = Payload)` declares `T` as a class type parameter, not a
|
||||
# referenced type — collect these to skip below.
|
||||
type_params = frozenset(re.findall(r"\btype\s+(\w+)", header))
|
||||
class_nid = _make_id(stem, class_name)
|
||||
add_node(class_nid, class_name, line)
|
||||
edges.append({"source": file_nid, "target": class_nid, "relation": "defines",
|
||||
"confidence": "EXTRACTED", "confidence_score": 1.0,
|
||||
"source_file": str_path, "source_location": f"L{line}", "weight": 1.0})
|
||||
|
||||
ext = re.search(r"\bextends\s+(\w+)", header)
|
||||
if ext:
|
||||
add_edge(class_nid, ext.group(1), "inherits", line)
|
||||
impl = re.search(r"\bimplements\s+([^;{]+)", header)
|
||||
if impl:
|
||||
for iface_name in _sv_split_type_list(impl.group(1)):
|
||||
add_edge(class_nid, iface_name.split("#", 1)[0].strip(), "implements", line)
|
||||
|
||||
body_without_functions = re.sub(
|
||||
r"\bfunction\b.*?\bendfunction\b",
|
||||
lambda m: "\n" * m.group(0).count("\n"),
|
||||
body,
|
||||
flags=re.DOTALL,
|
||||
)
|
||||
# Optional leading class-property qualifiers (rand/local/protected/etc.)
|
||||
# must be consumed: otherwise a qualified field like `rand Config x;`
|
||||
# (three tokens) fails the `<type> <name>;` shape and its type reference
|
||||
# is silently dropped.
|
||||
for field in re.finditer(r"^\s*(?:(?:rand|randc|local|protected|static|const|automatic|var)\s+)*([A-Za-z_]\w*(?:\s*#\s*\([^;]+?\))?)\s+\w+\s*;", body_without_functions, re.MULTILINE):
|
||||
# Count to the start of the type token (group 1), not the match
|
||||
# start: `^\s*` consumes the leading newline(s), so field.start()
|
||||
# would resolve to the class's line instead of the field's.
|
||||
field_line = line + body_without_functions.count("\n", 0, field.start(1))
|
||||
for ref_name, role in _sv_collect_type_refs(field.group(1), skip=type_params):
|
||||
add_edge(class_nid, ref_name, "references", field_line, "generic_arg" if role == "generic_arg" else "field")
|
||||
|
||||
for fm in _SV_FUNC_RE.finditer(body):
|
||||
return_type, func_name, params = fm.group(1), fm.group(2), fm.group(3)
|
||||
func_line = line + body.count("\n", 0, fm.start())
|
||||
func_nid = _make_id(class_nid, func_name)
|
||||
add_node(func_nid, func_name, func_line)
|
||||
edges.append({"source": class_nid, "target": func_nid, "relation": "method",
|
||||
"confidence": "EXTRACTED", "confidence_score": 1.0,
|
||||
"source_file": str_path, "source_location": f"L{func_line}", "weight": 1.0})
|
||||
for ref_name, role in _sv_collect_type_refs(return_type, skip=type_params):
|
||||
add_edge(func_nid, ref_name, "references", func_line, "generic_arg" if role == "generic_arg" else "return_type")
|
||||
for param in _sv_split_type_list(params):
|
||||
pm = _SV_PARAM_RE.match(param)
|
||||
if not pm:
|
||||
continue
|
||||
for ref_name, role in _sv_collect_type_refs(pm.group(1), skip=type_params):
|
||||
add_edge(func_nid, ref_name, "references", func_line, "generic_arg" if role == "generic_arg" else "parameter_type")
|
||||
|
||||
def extract_verilog(path: Path) -> dict:
|
||||
"""Extract modules, functions, tasks, package imports, instantiations, and
|
||||
SystemVerilog class semantics (inherits/implements edges, field/parameter/
|
||||
return-type references) from .v/.sv files."""
|
||||
try:
|
||||
import tree_sitter_verilog as tsverilog
|
||||
from tree_sitter import Language, Parser
|
||||
except ImportError:
|
||||
return {"nodes": [], "edges": [], "error": "tree_sitter_verilog not installed"}
|
||||
|
||||
try:
|
||||
language = Language(tsverilog.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 = _file_stem(path)
|
||||
str_path = str(path)
|
||||
nodes: list[dict] = []
|
||||
edges: list[dict] = []
|
||||
seen_ids: set[str] = set()
|
||||
|
||||
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}",
|
||||
"confidence_score": 1.0})
|
||||
|
||||
def add_edge(src: str, tgt: str, relation: str, line: int,
|
||||
confidence: str = "EXTRACTED", score: float = 1.0) -> None:
|
||||
edges.append({"source": src, "target": tgt, "relation": relation,
|
||||
"confidence": confidence, "confidence_score": score,
|
||||
"source_file": str_path, "source_location": f"L{line}", "weight": 1.0})
|
||||
|
||||
file_nid = _make_id(str(path))
|
||||
add_node(file_nid, path.name, 1)
|
||||
|
||||
def walk(node, module_nid: str | None = None) -> None:
|
||||
t = node.type
|
||||
|
||||
# SystemVerilog class bodies are handled by _augment_systemverilog_semantics
|
||||
# (regex over source text). Skip their subtrees so in-class methods are not
|
||||
# double-emitted here — and with the wrong, return-type-derived name.
|
||||
if t in ("class_declaration", "interface_class_declaration"):
|
||||
return
|
||||
|
||||
if t == "module_declaration":
|
||||
mod_name = _sv_first_identifier(_sv_child(node, "module_header"), source)
|
||||
if mod_name:
|
||||
line = node.start_point[0] + 1
|
||||
nid = _make_id(stem, mod_name)
|
||||
add_node(nid, mod_name, line)
|
||||
add_edge(file_nid, nid, "defines", line)
|
||||
for child in node.children:
|
||||
walk(child, nid)
|
||||
return
|
||||
|
||||
# `function_prototype` only appears inside class/interface-class bodies
|
||||
# (skipped above) and nests its name differently; it is intentionally not
|
||||
# handled here.
|
||||
elif t == "function_declaration":
|
||||
fn_body = _sv_child(node, "function_body_declaration")
|
||||
func_name = _sv_first_identifier(_sv_child(fn_body, "function_identifier"), source)
|
||||
if func_name:
|
||||
line = node.start_point[0] + 1
|
||||
parent = module_nid or file_nid
|
||||
nid = _make_id(parent, func_name)
|
||||
add_node(nid, f"{func_name}()", line)
|
||||
add_edge(parent, nid, "contains", line)
|
||||
|
||||
elif t == "task_declaration":
|
||||
tk_body = _sv_child(node, "task_body_declaration")
|
||||
task_name = _sv_first_identifier(_sv_child(tk_body, "task_identifier"), source)
|
||||
if task_name:
|
||||
line = node.start_point[0] + 1
|
||||
parent = module_nid or file_nid
|
||||
nid = _make_id(parent, task_name)
|
||||
add_node(nid, task_name, line)
|
||||
add_edge(parent, nid, "contains", line)
|
||||
|
||||
elif t == "package_import_declaration":
|
||||
for child in node.children:
|
||||
if child.type == "package_import_item":
|
||||
pkg_text = _read_text(child, source)
|
||||
pkg_name = pkg_text.split("::")[0].strip()
|
||||
if pkg_name:
|
||||
line = node.start_point[0] + 1
|
||||
tgt_nid = _make_id(pkg_name)
|
||||
add_node(tgt_nid, pkg_name, line)
|
||||
src_nid = module_nid or file_nid
|
||||
add_edge(src_nid, tgt_nid, "imports_from", line)
|
||||
|
||||
elif t in ("module_instantiation", "checker_instantiation"):
|
||||
# `leaf u_leaf();` parses as checker_instantiation in 1.0.3;
|
||||
# module_instantiation (when it occurs) exposes a `module_type` field.
|
||||
# Both reduce to the first identifier under the node — the instantiated
|
||||
# type, not the instance name (which appears later).
|
||||
if module_nid:
|
||||
type_node = node.child_by_field_name("module_type")
|
||||
inst_type = (_read_text(type_node, source).strip() if type_node
|
||||
else _sv_first_identifier(node, source))
|
||||
if inst_type:
|
||||
line = node.start_point[0] + 1
|
||||
tgt_nid = _make_id(inst_type)
|
||||
add_node(tgt_nid, inst_type, line)
|
||||
add_edge(module_nid, tgt_nid, "instantiates", line)
|
||||
|
||||
for child in node.children:
|
||||
walk(child, module_nid)
|
||||
|
||||
walk(root)
|
||||
_augment_systemverilog_semantics(
|
||||
source.decode("utf-8", errors="replace"),
|
||||
stem,
|
||||
str_path,
|
||||
file_nid,
|
||||
nodes,
|
||||
edges,
|
||||
seen_ids,
|
||||
)
|
||||
return {"nodes": nodes, "edges": edges}
|
||||
Reference in New Issue
Block a user