) or explicit aria-level attributes used on elements.
+ #This is a semantic property coming directly from the browser's accessibility layer. pw_depth: This is a structural
+ }
+
+
+ # ── 1. Clean & extract box details ────────────────────────────
+ if "box=" in attr_string:
+ box_match = re.search(r'box=\[?(-?\d+),(-?\d+),(-?\d+),(-?\d+)\]?', attr_string)
+ if box_match:
+ attrs["pw_box"] = [int(x) for x in box_match.groups()]
+ attr_string = re.sub(r',?\s*box=\[?[\d,,-]+\]?', '', attr_string).strip()
+
+ # ── 2. FIX: Clean & extract level details safely ──────────────
+ if "level=" in attr_string:
+ level_match = re.search(r'level=(\d+)', attr_string)
+ if level_match:
+ attrs["pw_level"] = int(level_match.group(1))
+ # Clean out the level substring so the comma loop doesn't get corrupted brackets
+ attr_string = re.sub(r',?\s*level=\d+\]?\s*\[?', '', attr_string).strip()
+
+ # ── 3. Parse remaining standard inline attributes ─────────────
+ if attr_string:
+ # Clean up residual broken brackets before splitting
+ attr_string = attr_string.replace('[', '').replace(']', '').strip()
+ for pair in attr_string.split(','):
+ pair = pair.strip()
+ if not pair:
+ continue
+ if '=' in pair:
+ k, v = pair.split('=', 1)
+ attrs[f"pw_{k.strip()}"] = v.strip().strip('"')
+ else:
+ attrs[f"pw_{pair}"] = True
+
+ # ── 4. Extract nested structural sub-nodes ────────────────────
+ if isinstance(node_value, dict):
+ if "box" in node_value:
+ attrs["pw_box"] = node_value["box"]
+
+ for k, v in node_value.items():
+ if k != "box" and not node_regex.match(k.strip()):
+ attrs[f"pw_{k}"] = v
+
+ result[key] = {k: v for k, v in attrs.items() if v is not None and v != ""}
+
+ # ── 5. Handle child hierarchies ───────────────────────────────────
+ if isinstance(node_value, list):
+ _process_node(node_value, depth + 1)
+ elif isinstance(node_value, dict):
+ for k, v in node_value.items():
+ if k != "box" and node_regex.match(k.strip()):
+ _parse_single_node(k, v, depth + 1)
+
+ _process_node(parsed_data)
+ return result
+
+def _unwrap_ax(ax_value: Optional[dict]) -> Any:
+ """
+ Unwrap a CDP AXValue dict { type, value, relatedNodes?, … }
+ to a plain Python scalar.
+ """
+ if ax_value is None:
+ return None
+ raw = ax_value.get("value")
+ typ = ax_value.get("type", "")
+ if typ == "boolean":
+ return bool(raw)
+ if typ in ("integer", "number"):
+ try:
+ return float(raw)
+ except (TypeError, ValueError):
+ return raw
+ if typ=="nodeList": #typical for labelledby properties
+ # Return a list of backendDOMNodeId for related nodes
+ #print(f"[*] Unwrapping nodeList: {ax_value}")
+ return [n.get("idref") for n in ax_value.get("relatedNodes", []) if n.get("backendDOMNodeId") is not None]
+ if isinstance(raw, dict):
+ return raw.get("value") or str(raw)
+ return raw
+
+
+def _ax_node_id(node_id: str) -> str:
+ return f"ax:{node_id}"
+
+
+def build_ax_layer(
+ cdp_nodes: list[dict],
+) -> tuple[nx.MultiDiGraph, dict[int, str]]:
+ """
+ Build AX sub-graph.
+ Returns (graph, backendDOMNodeId → graph_node_id mapping).
+ """
+ G: nx.MultiDiGraph = nx.MultiDiGraph()
+ backend_to_gid: dict[int, str] = {}
+ id_to_raw: dict[str, dict] = {}
+
+ # pass 1 – nodes
+ for raw in cdp_nodes:
+ nid = raw.get("nodeId", "")
+ if not nid:
+ continue
+ gid = _ax_node_id(nid)
+
+ # Flatten all ARIA properties (focusable, focused, disabled, level, …)
+ #NB aria-hidden property is missing*, because it is not returned directly from CDP nor playwright. Probably can be inferred with some heuristic
+ # is available from DOM layer
+ #* probably can be retrived with ax_properties = await cdp.send(
+ # "Accessibility.getAccessibilityPropertiesForNode",
+ # {"nodeId": node_id}
+ #)
+ # that seems return
+ #{
+ #"properties": {
+ # "focusable": true,
+ # "disabled": false,
+ # "hidden": false
+ #}
+ #}
+
+
+ props: dict[str, Any] = {}
+ # https://chromedevtools.github.io/devtools-protocol/tot/Accessibility/#type-AXPropertyName
+ for prop in raw.get("properties") or []:# manage properties attributes
+ pname = prop.get("name", "")
+ pval = _unwrap_ax(prop.get("value"))
+ if pname:
+ props[f"aria_{pname}"] = pval
+
+ ignored_reasons = "; ".join(
+ #str(_unwrap_ax(r.get("value")))
+ r.get("name")
+ for r in (raw.get("ignoredReasons") or [])
+ ) or None
+
+ backend_id = raw.get("backendDOMNodeId")
+ # https://chromedevtools.github.io/devtools-protocol/tot/Accessibility/#type-AXNode
+ node_attrs = _attrs(
+ layer = "ax",
+ node_id = nid,
+ backend_dom_id = backend_id,#The backend ID for the associated DOM node
+ ignored = raw.get("ignored", False),
+ ignored_reasons = ignored_reasons,
+ role = _unwrap_ax(raw.get("role")),
+ chrome_role = _unwrap_ax(raw.get("chromeRole")),
+ name = _unwrap_ax(raw.get("name")),
+ description = _unwrap_ax(raw.get("description")),
+ value = _unwrap_ax(raw.get("value")),
+ frame_id= raw.get("frameId"),
+ type= "ax_"+_unwrap_ax(raw.get("role")),# already use for role
+ **props,
+ )
+ G.add_node(gid, **node_attrs)
+ id_to_raw[nid] = raw
+ if backend_id is not None:
+ backend_to_gid[backend_id] = gid
+
+ # pass 2 – structural edges
+ edge_attributes_child = {
+ "relation": "hasChild",
+ "type": "hasChild_ax",
+ "layer": "ax"
+ }
+
+ edge_attributes_parent = {
+ "relation": "hasParent",
+ "type": "hasParent_ax",
+ "layer": "ax"
+ }
+ for nid, raw in id_to_raw.items():
+ gid = _ax_node_id(nid)
+ parent_id = raw.get("parentId")
+ if parent_id:
+ pgid = _ax_node_id(parent_id)
+ if pgid in G and not G.has_edge(gid,pgid):
+ G.add_edge(gid,pgid, **edge_attributes_parent)
+ for child_id in raw.get("childIds") or []:
+ cgid = _ax_node_id(child_id)
+ if cgid in G and not G.has_edge(gid, cgid):
+ G.add_edge(gid, cgid, **edge_attributes_child)
+
+ # pass 3 – ARIA relational edges (labelledby, controls, owns, …)
+ # these links are built from node properties setted before (from "node properties" attributes). In reality they are built from id_to_raw but they follow the same elaboration
+ # these link has superior importance respect to parent-child relationship because they represent a semantic relationship. ths is why they override he previous
+ # because they are both node attributes and edge attributes, mayybe are redundant. TBM
+ ARIA_RELS = { #aria-describedby has the name describedby and so on...
+
+ #Labeling and Describing. These attributes form a relationship between an element and the text that identifies or explains it.
+ 'describedby', #'aria-describedby' has the name 'describedby' in CDP properties
+ 'labelledby',
+ 'details',
+
+ #Controlling and Owning. These attributes define hierarchy, control, and structural relationships when elements aren't nested inside each other in the actual HTML code.
+ 'controls',
+ 'owns',
+ 'activedescendant',
+
+ #Navigating and Ordering. These attributes help assistive technologies understand how to navigate between distinct elements or content blocks.
+ 'flowto',
+ 'colindex',
+ 'rowindex'
+ }
+
+ for nid, raw in id_to_raw.items():
+ gid = _ax_node_id(nid)
+ for prop in raw.get("properties") or []:
+ pname = prop.get("name", "")
+ if pname not in ARIA_RELS:
+ continue
+ else:
+ print(f" [*] Processing ARIA relation property: {pname}")
+
+ for rel_node in (prop.get("value") or {}).get("relatedNodes") or []:
+ bid = rel_node.get("backendDOMNodeId")
+ target = backend_to_gid.get(bid)
+
+ if not target:
+ print(f" [!] Warning: No target node found for ARIA relation {pname} with backendDOMNodeId {bid}")
+ if G.has_edge(nid, target):
+ print(f" [!] Warning: Edge already exists for {nid} --{pname}--> {target}, updating/add link")
+ #if target and not G.has_edge(nid, target):
+ if target :# multigraph, add multiiple edge type #ovverride! this link is more important than the parent-child relationship
+ print(f" [*] Adding additional ARIA edge: {nid} --{pname}--> {target}")
+ edge_attributes = {
+ "relation": pname,
+ "type": pname+"_ax",
+ "layer": "ax"
+ }
+ G.add_edge(gid, target, **edge_attributes)
+
+ return G, backend_to_gid
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Layer C – Visual layout graph
+# ─────────────────────────────────────────────────────────────────────────────
+
+def _quad_to_bbox(quad: list[float]) -> tuple[float, float, float, float]:
+ """
+ CDP Quad is [x1,y1, x2,y2, x3,y3, x4,y4] (clockwise from top-left).
+ Returns (x, y, width, height).
+ """
+ xs = quad[0::2]
+ ys = quad[1::2]
+ x, y = min(xs), min(ys)
+ return x, y, max(xs) - x, max(ys) - y
+
+
+def _vis_node_id(backend_id: int) -> str:
+ return f"vis:{backend_id}"
+
+
+def should_keep_ui_node(node_info,ui_target_nodes=None,ui_target_roles=None):
+ # 1. Check tag name
+ node_name = node_info.get("node_name", "").lower()
+ if node_name in ui_target_nodes:
+ return True
+
+ # 2. Check for interactive ARIA roles in attributes
+ attributes_list = node_info.get("attributes", [])
+ # Parse the flattened CDP array ["key1", "val1", "key2", "val2"]
+ attrs = dict(zip(attributes_list[0::2], attributes_list[1::2]))
+
+ role = attrs.get("role", "").lower()
+ if role in ui_target_roles:
+ return True
+
+ return False
+
+async def build_visual_layer(
+ cdp: CDPClient,
+ dom_flat: dict,
+ viewport_w: float,
+ viewport_h: float,
+) -> tuple[nx.MultiDiGraph, dict[int, str]]:
+ """
+ Build visual layout sub-graph by calling DOM.getBoxModel for every
+ DOM element node (nodeType == 1).
+
+ Returns (graph, backendNodeId → graph_node_id mapping).
+ """
+ G: nx.MultiDiGraph = nx.MultiDiGraph()
+ backend_to_gid: dict[int, str] = {}
+ viewport_area = max(viewport_w * viewport_h, 1)
+
+ ui_target_nodes = {
+ "h1", "h2", "h3", "h4", "h5", "h6", "p", "span", "a", "img", "svg",
+ "video", "audio", "input", "textarea", "button", "select", "option",
+ "label", "form", #"ul", "ol", "li",
+ "table", #"tr", "td", "th",
+ "nav",
+ "header", "footer",
+ "details", "summary", "dialog", "iframe"
+}
+
+ # ARIA roles that transform generic elements (like
) into interactive ones
+ ui_target_roles = {
+ "button", "link", "checkbox", "radio", "switch",
+ "tab", "tabpanel", "combobox", "listbox", "menu", "dialog"
+ }
+
+
+
+ # Only Element nodes (nodeType=3) have a box model
+ element_nodes = [
+ (nid, info)
+ for nid, info in dom_flat.items()
+ if info.get("node_type") == 3 or should_keep_ui_node(info,ui_target_nodes=ui_target_nodes,ui_target_roles=ui_target_roles) #info.get("node_name") in ("h1","h2","h3","h4","h5","h6","img", "input", "textarea", "button", "select", "a", "label", "svg","video","audio","text")
+
+
+ ]
+
+ print(f" → querying box model for {len(element_nodes)} DOM elements …")
+
+ # 1. Enable the CSS domain (Only need to do this once)
+
+
+ for node_id, info in element_nodes:
+ backend_id = info.get("backend_id")
+ if backend_id is None:
+ continue
+
+ try:
+ #https://chromedevtools.github.io/devtools-protocol/tot/DOM/#method-getBoxModel
+ box_result = await cdp.send_raw(
+ "DOM.getBoxModel",
+ {"nodeId": node_id}
+ )
+ model = box_result.get("model", {})
+ except Exception:
+ # Non-rendered nodes (display:none, detached, …) and JS raise an error. NB: "hidden" is not coverred
+ #print(f" [!] Warning: Failed to get box model for DOM node {node_id} (backend {backend_id}). Probably not rendered or detached.")
+ continue
+
+ try: #extract visibility (if hidden should be skipped)
+ css_results = await cdp.send_raw(
+ "CSS.getComputedStyleForNode",
+ {"nodeId": node_id}
+ )
+
+ computed_properties = css_results.get("computedStyle", [])
+ # Extract the visibility property
+ visibility = next(
+ (p["value"] for p in computed_properties if p["name"] == "visibility"),
+ "unknown"
+ )
+ if (visibility=="hidden"): # skip unvisible elements
+ #print(f" Visibility is 'hidden' for DOM node {node_id} (backend {backend_id}).")
+ continue
+
+
+ except Exception as e:
+ print(f" [!] Warning: Failed to get computed style for DOM node {node_id} (backend {backend_id}). Error: {e}")
+ continue
+
+ content_quad = model.get("content")
+ if not content_quad:
+ continue
+
+ x, y, w, h = _quad_to_bbox(content_quad)
+ area = w * h
+
+ # Skip zero-size elements (invisible / collapsed)
+ if area == 0:
+ continue
+
+ gid = _vis_node_id(backend_id)
+
+ # Visual prominence = fraction of viewport occupied
+ prominence = round(area / viewport_area, 6)
+
+ # Estimate reading order rank within the viewport
+ # (simple top-to-bottom, left-to-right heuristic)
+ reading_rank = y * 10_000 + x # will be normalized later
+
+ # Fetch outerHTML for debugging / inspection
+ node_outerHtml=await cdp.send_raw(
+ "DOM.getOuterHTML",
+ {"nodeId": node_id}
+ )
+ #print(f"[*] Adding visual node {gid} for DOM node {node_id} ({info.get('node_name')}) at ({x:.1f},{y:.1f}) size {w:.1f}x{h:.1f}, prominence={prominence:.6f}, reading_rank={reading_rank}, outerHTML={node_outerHtml.get('outerHTML')[:30]}...")
+ node_attrs = _attrs(
+ layer = "visual",
+ backend_id = backend_id,
+ dom_node_id = node_id,
+ tag = info.get("tag"),
+ bbox_x = round(x, 2),
+ bbox_y = round(y, 2),
+ bbox_w = round(w, 2),
+ bbox_h = round(h, 2),
+ area = round(area, 2),
+ prominence = prominence,
+ reading_rank = reading_rank, # raw; normalized below
+ # padding / border / margin summaries
+ border_w = round(model.get("width", 0), 2),
+ border_h = round(model.get("height", 0), 2),
+ name= info.get("node_name").replace("#",""),#my setup replace because problem to push to neo4j
+ value = info.get("node_value").replace("#","") or None,
+ outerHtml=node_outerHtml.get("outerHTML")[:100].replace("#","") or None,
+ visibility=visibility,
+ type= "visual_"+info.get("node_name").replace("#","")
+ )
+ G.add_node(gid, **node_attrs)
+ backend_to_gid[backend_id] = gid
+
+ # Normalize reading_rank to 1..N (top-to-bottom order)
+ sorted_nodes = sorted(
+ G.nodes(data=True),
+ key=lambda x: x[1].get("reading_rank", 0),
+ )
+ for rank, (gid, _) in enumerate(sorted_nodes, start=1):
+ G.nodes[gid]["reading_rank"] = rank
+
+ # Spatial edges: "visual_overlap" and "visual_sibling"
+ # For efficiency, only connect nodes whose bounding boxes overlap
+ # or are within a small vertical proximity (same visual "row")
+ # some overlap are caused because of boundix box retrived, not real visual elements occupation. This will be improved using object detection and segmentation in the future
+
+ nodes_list = list(G.nodes(data=True))
+ for i, (gid_a, a) in enumerate(nodes_list):
+ for gid_b, b in nodes_list[i + 1:]:
+ ax, ay, aw, ah = a["bbox_x"], a["bbox_y"], a["bbox_w"], a["bbox_h"]
+ bx, by, bw, bh = b["bbox_x"], b["bbox_y"], b["bbox_w"], b["bbox_h"]
+
+ # 1. FIXED Overlap check: Use an epsilon threshold (e.g., 1 pixel)
+ # to prevent elements that are just touching or have subpixel rounding from overlapping.
+ EPSILON = 1.0
+ h_overlap = (ax + EPSILON < bx + bw) and (bx + EPSILON < ax + aw)
+ v_overlap = (ay + EPSILON < by + bh) and (by + EPSILON < ay + ah)
+
+ if h_overlap and v_overlap:
+ G.add_edge(gid_a, gid_b, relation="visual_overlap", layer="visual", type="visual_overlap")
+ continue # Only skip if it's a true substantial overlap
+
+ # 2. Same-row heuristic
+ cy_a = ay + ah / 2
+ cy_b = by + bh / 2
+ if abs(cy_a - cy_b) < min(ah, bh) / 2:
+ if abs(ax - bx) < max(aw, bw) * 3: # horizontally close
+ left, right = (gid_a, gid_b) if ax < bx else (gid_b, gid_a)
+ G.add_edge(left, right, relation="visual_sibling", layer="visual", type="visual_sibling")
+
+ return G, backend_to_gid
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Cross-edges (inter-layer)
+# ─────────────────────────────────────────────────────────────────────────────
+
+def add_cross_edges(
+ # the cross-edges are built using backendDOMNodeId / backendNodeId as the join key — it is the only identifier that means the same thing across the DOM, AX, and Visual layers
+ HAG: nx.MultiDiGraph,
+ dom_backend_to_gid: dict[int, str], # backendNodeId → dom:N
+ ax_backend_to_gid: dict[int, str], # backendDOMNodeId → ax:N
+ vis_backend_to_gid: dict[int, str], # backendNodeId → vis:N
+) -> None:
+ """
+ Connect the three layers with typed cross-edges:
+ dom_to_ax : dom:N → ax:M (same element, different view)
+ dom_to_visual : dom:N → vis:M (same element, different view)
+ ax_to_visual : ax:N → vis:M (semantic node ↔ rendered bbox)
+ """
+ cross_counts = {"dom_to_ax": 0, "dom_to_visual": 0, "ax_to_visual": 0}
+
+ # DOM ↔ AX (both keyed by backendNodeId / backendDOMNodeId)
+ common_ids_dom_ax = set(dom_backend_to_gid) & set(ax_backend_to_gid)
+ for bid in common_ids_dom_ax:
+ dom_gid = dom_backend_to_gid[bid]
+ ax_gid = ax_backend_to_gid[bid]
+ HAG.add_edge(dom_gid, ax_gid,
+ relation="dom_to_ax", layer="cross",
+ backend_id=bid,type="dom_to_ax")
+ cross_counts["dom_to_ax"] += 1
+
+ # DOM ↔ Visual
+ common_ids_dom_vis = set(dom_backend_to_gid) & set(vis_backend_to_gid)
+ for bid in common_ids_dom_vis:
+ dom_gid = dom_backend_to_gid[bid]
+ vis_gid = vis_backend_to_gid[bid]
+ HAG.add_edge(dom_gid, vis_gid,
+ relation="dom_to_visual", layer="cross",
+ backend_id=bid,type="dom_to_visual")
+ cross_counts["dom_to_visual"] += 1
+
+ # AX ↔ Visual
+ common_ids_ax_vis = set(ax_backend_to_gid) & set(vis_backend_to_gid)
+ for bid in common_ids_ax_vis:
+ ax_gid = ax_backend_to_gid[bid]
+ vis_gid = vis_backend_to_gid[bid]
+ HAG.add_edge(ax_gid, vis_gid,
+ relation="ax_to_visual", layer="cross",
+ backend_id=bid,type="ax_to_visual")
+ cross_counts["ax_to_visual"] += 1
+
+ print(f" → cross-edges: {cross_counts}")
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Main pipeline
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def build_heterogeneous_graph(url: str) -> nx.MultiDiGraph:
+ """
+ Full pipeline:
+ 1. Launch Chromium (Playwright) with CDP debug port.
+ 2. Navigate and wait for idle.
+ 3. Extract DOM tree → Layer A.
+ 4. Extract AX tree → Layer B.
+ 5. Extract box models → Layer C.
+ 6. Merge into a MultiDiGraph with cross-edges.
+ """
+ debug_port = _free_port()
+ print(f"[*] CDP debug port : {debug_port}")
+
+ async with async_playwright() as pw:
+ browser = await pw.chromium.launch(
+ headless=True,
+ args=[f"--remote-debugging-port={debug_port}"],
+ )
+ viewport_res = {"width": 1920, "height": 1080} #{"width": 1280, "height": 800}
+ context = await browser.new_context(viewport=viewport_res)
+ page = await context.new_page()
+
+ print(f"[*] Navigating → {url}")
+ await page.goto(url, wait_until="networkidle", timeout=30_000)
+ await page.wait_for_timeout(1000)
+
+ # ── Step B-B1: Playwright accessibility snapshot ─────────────────────
+ print("[*] Playwright: extracting accessibility snapshot …")
+
+ pw_snapshot_str: str = ""
+ try:
+ # Captures the new YAML-based ARIA snapshot including bounding box coordinates
+ # https://playwright.dev/python/docs/aria-snapshots
+ snapshot = await page.aria_snapshot(boxes=True,mode="ai")
+ if snapshot:
+ pw_snapshot_str = snapshot
+ # Since it's a YAML string, we can count lines or characters to log density
+ #print(f"yaml file: {pw_snapshot_str.strip().splitlines()}")
+ line_count = len(pw_snapshot_str.strip().splitlines())
+ print(f" → Playwright ARIA snapshot collected ({line_count} lines of YAML structure)")
+ pw_flat = _flatten_playwright_snapshot(snapshot)
+ else:
+ print(" [!] Snapshot returned None (page may be empty)")
+ except Exception as e:
+ print(f" [!] Playwright snapshot failed: {e}")
+ #---------------------
+
+ viewport = page.viewport_size or viewport_res
+ vw, vh = viewport["width"], viewport["height"]
+
+ # ── CDP target discovery ──────────────────────────────────────────
+ with urllib.request.urlopen(
+ f"http://localhost:{debug_port}/json", timeout=5
+ ) as resp:
+ targets = json.loads(resp.read())
+
+ page_targets = [
+ t for t in targets
+ if t.get("type") == "page" and "webSocketDebuggerUrl" in t
+ ]
+ if not page_targets:
+ raise RuntimeError("No CDP page target found")
+
+ ws_url = page_targets[0]["webSocketDebuggerUrl"]
+ print(f"[*] CDP WebSocket : {ws_url}")
+
+ async with CDPClient(ws_url) as cdp:
+
+ # ── Layer A: DOM ──────────────────────────────────────────────
+ print("\n[A] Extracting DOM tree …")
+ await cdp.send_raw("DOM.enable")
+ await cdp.send_raw("CSS.enable")
+ #https://chromedevtools.github.io/devtools-protocol/tot/DOM/#method-getDocument
+ dom_result = await cdp.send_raw(
+ "DOM.getDocument",
+ {"depth": -1, "pierce": True} # full tree, pierce shadow DOM
+ )
+ dom_root = dom_result.get("root", {})
+ dom_flat = _flatten_dom_tree(dom_root)
+ print(f" → {len(dom_flat)} DOM nodes")
+
+ G_dom, dom_backend_map = await build_dom_layer(dom_flat,cdp)
+ print(f" → DOM graph: {G_dom.number_of_nodes()} nodes, "
+ f"{G_dom.number_of_edges()} edges")
+
+ # ── Layer B: AX tree ──────────────────────────────────────────
+
+ # ── Step B2: CDP via cdp_use ───────────────────────────────────────
+ print("\n[B2] Extracting Accessibility tree …")
+ await cdp.send_raw("Accessibility.enable")
+ ax_result = await cdp.send_raw("Accessibility.getFullAXTree", {})
+ cdp_nodes = ax_result.get("nodes", [])
+ await cdp.send_raw("Accessibility.disable")
+
+ if cdp_nodes:
+ print("[*] Building AX graph from CDP nodes …")
+ print(f" → {len(cdp_nodes)} AX nodes (raw)")
+
+ G_ax, ax_backend_map = build_ax_layer(cdp_nodes)
+ print(f" → AX graph: {G_ax.number_of_nodes()} nodes, "
+ f"{G_ax.number_of_edges()} edges")
+
+ if pw_flat:
+ print("[*] Merging Playwright AX attributes …")
+ merge_playwright_data(G_ax, pw_flat)
+
+ elif pw_flat:
+ print("[!] CDP unavailable — building AX graph from Playwright snapshot only")
+ G_ax = _build_graph_from_pw(pw_flat)
+ else:
+ raise RuntimeError("No accessibility data could be extracted.")
+
+ # ── Layer C: Visual layout ────────────────────────────────────
+ print("\n[C] Extracting visual layout (box models) …")
+ G_vis, vis_backend_map = await build_visual_layer(# it is just extracted from the dom layer (only element_type=1)
+ cdp, dom_flat, vw, vh
+ )
+ print(f" → Visual graph: {G_vis.number_of_nodes()} nodes, "
+ f"{G_vis.number_of_edges()} edges")
+
+ await cdp.send_raw("DOM.disable")
+ await cdp.send_raw("CSS.disable")
+
+ await browser.close()
+
+ # ── Merge into heterogeneous graph ────────────────────────────────────
+ print("\n[HAG] Merging layers …")
+ HAG = nx.MultiDiGraph()
+ HAG.add_nodes_from(G_dom.nodes(data=True))
+ HAG.add_edges_from(G_dom.edges(data=True))
+ HAG.add_nodes_from(G_ax.nodes(data=True))
+ HAG.add_edges_from(G_ax.edges(data=True))
+ HAG.add_nodes_from(G_vis.nodes(data=True))
+ HAG.add_edges_from(G_vis.edges(data=True))
+
+ print("[HAG] Adding cross-edges …")
+ add_cross_edges(HAG, dom_backend_map, ax_backend_map, vis_backend_map)
+
+ return HAG
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Summary
+# ─────────────────────────────────────────────────────────────────────────────
+
+def print_summary(HAG: nx.MultiDiGraph) -> None:
+ print(f"\n{'='*65}")
+ print(" Heterogeneous Accessibility Graph (HAG) — Summary")
+ print(f"{'='*65}")
+ print(f" Total nodes : {HAG.number_of_nodes()}")
+ print(f" Total edges : {HAG.number_of_edges()}")
+
+ # Per-layer node counts
+ layer_counts: dict[str, int] = {}
+ for _, attrs in HAG.nodes(data=True):
+ lay = attrs.get("layer", "unknown")
+ layer_counts[lay] = layer_counts.get(lay, 0) + 1
+ print("\n Nodes per layer:")
+ for lay, cnt in sorted(layer_counts.items()):
+ print(f" {lay:<12} {cnt}")
+
+ # Per-relation edge counts
+ rel_counts: dict[str, int] = {}
+ for _, _, attrs in HAG.edges(data=True):
+ rel = attrs.get("relation", "unknown")
+ rel_counts[rel] = rel_counts.get(rel, 0) + 1
+ print("\n Edges per relation:")
+ for rel, cnt in sorted(rel_counts.items(), key=lambda x: -x[1]):
+ print(f" {rel:<25} {cnt}")
+
+ # AX role distribution (non-ignored)
+ roles: dict[str, int] = {}
+ for _, attrs in HAG.nodes(data=True):
+ if attrs.get("layer") != "ax" or attrs.get("ignored"):
+ continue
+ r = str(attrs.get("role") or "none")
+ roles[r] = roles.get(r, 0) + 1
+ if roles:
+ print("\n AX roles (non-ignored, top 15):")
+ for role, cnt in sorted(roles.items(), key=lambda x: -x[1])[:15]:
+ print(f" {role:<28} {cnt}")
+
+ # Visual layer stats
+ prominences = [
+ attrs["prominence"]
+ for _, attrs in HAG.nodes(data=True)
+ if attrs.get("layer") == "visual"
+ ]
+ if prominences:
+ print(f"\n Visual nodes : {len(prominences)}")
+ print(f" Max prominence : {max(prominences):.4f}")
+ print(f" Avg prominence : {sum(prominences)/len(prominences):.4f}")
+
+ components = nx.number_connected_components(HAG.to_undirected())
+ dag = nx.is_directed_acyclic_graph(HAG)
+ print(f"\n Connected components: {components}")
+ print(f" is DAG : {'yes' if dag else 'no (cycles detected)'}")
+ print(f"{'='*60}\n")
+
+
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Export
+# ─────────────────────────────────────────────────────────────────────────────
+
+def save_graph(HAG: nx.MultiDiGraph, path: str) -> None:
+ if path.endswith(".json"):
+ data = nx.node_link_data(HAG)
+ with open(path, "w", encoding="utf-8") as f:
+ json.dump(data, f, indent=2, default=str)
+ print(f"[*] Saved → {path} (node-link JSON)")
+ else:
+ # GraphML: coerce all attrs to str
+ G_exp = HAG.copy()
+ for _, attrs in G_exp.nodes(data=True):
+ for k in list(attrs):
+ attrs[k] = str(attrs[k])
+ for _, _, attrs in G_exp.edges(data=True):
+ for k in list(attrs):
+ attrs[k] = str(attrs[k])
+ nx.write_graphml(G_exp, path)
+ print(f"[*] Saved → {path} (GraphML)")
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Drawing
+# ─────────────────────────────────────────────────────────────────────────────
+
+def draw_graph(HAG: nx.MultiDiGraph, out: str = "hag.png",
+ max_nodes: int = 120) -> None:
+ try:
+ import matplotlib.pyplot as plt
+ import matplotlib.patches as mpatches
+ except ImportError:
+ print("[!] matplotlib not installed — skip draw")
+ return
+
+ # Colour scheme per layer
+ LAYER_COLOR = {
+ "dom": "#90CAF9", # blue
+ "ax": "#A5D6A7", # green
+ "visual": "#FFCC80", # orange
+ }
+ LAYER_SHAPE = {"dom": "s", "ax": "o", "visual": "D"}
+ EDGE_COLOR = {
+ "child": "#B0BEC5",
+ "dom_to_ax": "#7986CB",
+ "dom_to_visual": "#FF8A65",
+ "ax_to_visual": "#4DB6AC",
+ "visual_sibling": "#FFD54F",
+ "visual_overlap": "#EF9A9A",
+ }
+
+ if HAG.number_of_nodes() > max_nodes:
+ # Keep the most connected nodes
+ top = sorted(HAG.nodes(), key=lambda n: HAG.degree(n), reverse=True)
+ HAG = HAG.subgraph(top[:max_nodes]).copy()
+ print(f"[*] Drawing subgraph: top {max_nodes} nodes by degree")
+
+ try:
+ pos = nx.nx_agraph.graphviz_layout(HAG, prog="dot")
+ except Exception:
+ pos = nx.spring_layout(HAG, seed=42, k=1.8)
+
+ fig, ax = plt.subplots(figsize=(24, 16))
+
+ # Draw edges per relation type
+ all_edges = list(HAG.edges(data=True))
+ for rel, color in EDGE_COLOR.items():
+ subset = [(u, v) for u, v, d in all_edges if d.get("relation") == rel]
+ if subset:
+ nx.draw_networkx_edges(
+ HAG, pos, edgelist=subset,
+ edge_color=color, width=0.8 if rel == "child" else 1.4,
+ alpha=0.6, arrows=True, ax=ax,
+ arrowsize=10,
+ )
+
+ # Draw nodes per layer
+ for layer, color in LAYER_COLOR.items():
+ subset = [n for n, d in HAG.nodes(data=True) if d.get("layer") == layer]
+ if subset:
+ nx.draw_networkx_nodes(
+ HAG, pos, nodelist=subset,
+ node_color=color, node_size=350, alpha=0.9, ax=ax,
+ )
+
+ # Labels (role/tag + name snippet)
+ labels = {}
+ for nid, attrs in HAG.nodes(data=True):
+ layer = attrs.get("layer", "")
+ if layer == "ax":
+ role = str(attrs.get("role") or "?")
+ name = str(attrs.get("name") or "")[:18]
+ labels[nid] = f"{role}\n{name}" if name else role
+ elif layer == "dom":
+ labels[nid] = str(attrs.get("tag") or "?")
+ elif layer == "visual":
+ labels[nid] = f"[{attrs.get('tag','?')}]\n{attrs.get('reading_rank','')}"
+ nx.draw_networkx_labels(HAG, pos, labels=labels, font_size=5, ax=ax)
+
+ # Legend
+ patches = [mpatches.Patch(color=c, label=l)
+ for l, c in LAYER_COLOR.items()]
+ edge_patches = [mpatches.Patch(color=c, label=l)
+ for l, c in EDGE_COLOR.items() if l != "child"]
+ ax.legend(handles=patches + edge_patches, loc="upper left",
+ fontsize=8, framealpha=0.8)
+ ax.set_title("Heterogeneous Accessibility Graph (HAG)", fontsize=14)
+ ax.axis("off")
+ plt.tight_layout()
+ plt.savefig(out, dpi=150, bbox_inches="tight")
+ print(f"[*] Graph image saved → {out}")
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# CLI
+# ─────────────────────────────────────────────────────────────────────────────
+
+def _parse_args() -> argparse.Namespace:
+ p = argparse.ArgumentParser(
+ description="Build a Heterogeneous Accessibility Graph (DOM + AX + Visual)"
+ )
+ p.add_argument("url", help="URL to analyse (http/https or file://)")
+ p.add_argument("--output", "-o", default="hag.graphml",
+ help="Output path: .graphml or .json [default: hag.graphml]")
+ p.add_argument("--draw", action="store_true",
+ help="Render and save hag.png")
+ p.add_argument("--no-ignored", action="store_true",
+ help="Remove ignored AX nodes (and their cross-edges)")
+ return p.parse_args()
+
+
+async def main() -> None:
+ args = _parse_args()
+ HAG = await build_heterogeneous_graph(args.url)
+
+ if args.no_ignored:
+ to_remove = [
+ n for n, a in HAG.nodes(data=True)
+ if a.get("layer") == "ax" and a.get("ignored")#remove only ignored AX nodes, keep the others
+ ]
+ HAG.remove_nodes_from(to_remove)
+ print(f"[*] Removed {len(to_remove)} ignored AX nodes")
+
+ print_summary(HAG)
+ save_graph(HAG, args.output)
+
+ if args.draw:
+ draw_graph(HAG)
+
+
+if __name__ == "__main__":
+ asyncio.run(main())