From e8a22893c92afa2efd4cd46432c4103b69e9856b Mon Sep 17 00:00:00 2001 From: tpateeq Date: Thu, 9 Jul 2026 01:31:17 +0530 Subject: [PATCH] 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) --- graphify/extract.py | 493 +------------------------------- graphify/extractors/__init__.py | 4 + graphify/extractors/markdown.py | 176 ++++++++++++ graphify/extractors/verilog.py | 329 +++++++++++++++++++++ 4 files changed, 511 insertions(+), 491 deletions(-) create mode 100644 graphify/extractors/markdown.py create mode 100644 graphify/extractors/verilog.py diff --git a/graphify/extract.py b/graphify/extract.py index 332da7e1..8380e0de 100644 --- a/graphify/extract.py +++ b/graphify/extract.py @@ -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 ` ;` 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 (![alt](src)). 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+[^)]*)?\)') # Reference-style link definition line: [label]: target "optional title" -_MD_REF_DEF_RE = re.compile(r'^\s{0,3}\[[^\]]+\]:\s*]+)>?') # Obsidian-style wikilink: [[target]] / [[target|alias]] / [[target#anchor]]. -_MD_WIKILINK_RE = re.compile(r'(? "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 ───────────────────────────────────────────────── diff --git a/graphify/extractors/__init__.py b/graphify/extractors/__init__.py index 46b484b3..883e1181 100644 --- a/graphify/extractors/__init__.py +++ b/graphify/extractors/__init__.py @@ -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, } diff --git a/graphify/extractors/markdown.py b/graphify/extractors/markdown.py new file mode 100644 index 00000000..d6d45ac8 --- /dev/null +++ b/graphify/extractors/markdown.py @@ -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+[^)]*)?\)') + +_MD_REF_DEF_RE = re.compile(r'^\s{0,3}\[[^\]]+\]:\s*]+)>?') + +_MD_WIKILINK_RE = re.compile(r'(? "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} diff --git a/graphify/extractors/verilog.py b/graphify/extractors/verilog.py new file mode 100644 index 00000000..2f5fea49 --- /dev/null +++ b/graphify/extractors/verilog.py @@ -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 ` ;` 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}