# Utility digest - part 4/8 (scope: workspace/agents/opencode) ================ File: workspace/agents/opencode/engine/skills/diagram-design/scripts/drawio_extract.py ================ #!/usr/bin/env python3 """Extract a normalized intermediate representation (IR) from a draw.io file. The deterministic half of the draw.io import flow: this script never makes a design decision. It decodes whatever draw.io wrote (raw XML, deflate+base64 payloads, PNG/SVG files with an embedded ``mxfile``), flattens the mxGraphModel into absolute-positioned nodes and edges, and reports structural signals — hubs, containers, depth, cycles, leaf clusters — that the skill uses to pick a diagram type and a level of detail. Usage: python3 drawio_extract.py [--page N|NAME] [--json] [--max-rows N] [--out PATH] Default output is a compact Markdown digest for local review. ``--json`` emits the full IR instead (every node, every edge, every style). Exit codes: 0 ok, 2 unreadable / unsupported input. """ from __future__ import annotations import argparse import base64 import html import json import math import re import struct import sys import zlib from dataclasses import dataclass, field, asdict from pathlib import Path from typing import Any from urllib.parse import unquote from xml.etree import ElementTree as ET # -------------------------------------------------------------------------- # container / payload decoding # -------------------------------------------------------------------------- PNG_MAGIC = b"\x89PNG\r\n\x1a\n" MAX_INPUT_BYTES = 32 * 1024 * 1024 MAX_XML_BYTES = 64 * 1024 * 1024 MAX_PAGES = 100 MAX_CELLS_PER_PAGE = 10000 def _configure_stdout_utf8() -> None: """Emit digests as UTF-8 even when Windows selects a legacy codepage.""" reconfigure = getattr(sys.stdout, "reconfigure", None) if reconfigure is not None: reconfigure(encoding="utf-8", errors="strict") class PayloadTooLarge(ValueError): """Raised when compressed metadata expands beyond the supported limit.""" def _fail(msg: str) -> "NoReturn": # type: ignore[valid-type] print(f"drawio_extract: {msg}", file=sys.stderr) raise SystemExit(2) def _reject_unsafe_xml(xml: str, source: str) -> None: """Reject declarations that can make XML parsing expand external data.""" upper = xml.upper() if " bytes: """Decompress without allowing a small payload to expand without bound.""" decompressor = zlib.decompressobj(wbits) output = bytearray() chunk = data while chunk: remaining = limit + 1 - len(output) if remaining <= 0: raise PayloadTooLarge(f"decoded payload exceeds {limit} bytes") output.extend(decompressor.decompress(chunk, remaining)) if len(output) > limit: raise PayloadTooLarge(f"decoded payload exceeds {limit} bytes") chunk = decompressor.unconsumed_tail if not decompressor.eof: raise zlib.error("incomplete compressed payload") remaining = limit + 1 - len(output) if remaining <= 0: raise PayloadTooLarge(f"decoded payload exceeds {limit} bytes") output.extend(decompressor.flush(remaining)) if len(output) > limit: raise PayloadTooLarge(f"decoded payload exceeds {limit} bytes") return bytes(output) def _inflate(payload: str) -> str | None: """Undo draw.io's base64 + raw-deflate + URL-encoding pipeline.""" try: raw = base64.b64decode(payload, validate=False) except Exception: return None for wbits in (-15, 15, 47): try: text = _decompress_limited(raw, wbits).decode("utf-8", "replace") except PayloadTooLarge: _fail( f"decoded diagram exceeds the {MAX_XML_BYTES // (1024 * 1024)} MiB limit" ) except Exception: continue # draw.io URL-encodes before deflating; unquote is a no-op if it didn't. return unquote(text) return None def _png_embedded_xml(data: bytes) -> str | None: """Pull the ``mxfile`` tEXt/zTXt chunk out of a draw.io-exported PNG.""" pos = len(PNG_MAGIC) while pos + 8 <= len(data): (length,) = struct.unpack(">I", data[pos : pos + 4]) ctype = data[pos + 4 : pos + 8] body_end = pos + 8 + length chunk_end = body_end + 4 if chunk_end > len(data): _fail("PNG has a truncated metadata chunk") body = data[pos + 8 : body_end] pos = chunk_end if ctype not in (b"tEXt", b"zTXt", b"iTXt"): if ctype == b"IEND": break continue key, _, rest = body.partition(b"\x00") if key.lower() != b"mxfile": continue try: if ctype == b"tEXt": value = rest elif ctype == b"zTXt": value = _decompress_limited(rest[1:], 15) else: # iTXt: compression flag, method, lang, translated key, text flag = rest[0:1] tail = rest[2:].split(b"\x00", 2)[-1] value = _decompress_limited(tail, 15) if flag == b"\x01" else tail except PayloadTooLarge: _fail( f"embedded PNG diagram exceeds the " f"{MAX_XML_BYTES // (1024 * 1024)} MiB limit" ) except (IndexError, ValueError, zlib.error): _fail("PNG has invalid compressed draw.io metadata") return unquote(value.decode("utf-8", "replace")) return None def _svg_embedded_xml(text: str) -> str | None: for match in re.finditer( r"\bcontent\s*=\s*([\"'])(.*?)\1", text, flags=re.IGNORECASE | re.DOTALL ): candidate = html.unescape(match.group(2)) if " str: """Return the ```` (or bare ````) XML for any input.""" size = path.stat().st_size if size > MAX_INPUT_BYTES: _fail( f"{path.name}: input is {size} bytes; maximum is " f"{MAX_INPUT_BYTES // (1024 * 1024)} MiB" ) data = path.read_bytes() if data.startswith(PNG_MAGIC): xml = _png_embedded_xml(data) if not xml: _fail(f"{path.name}: PNG has no embedded draw.io diagram") return xml text = data.decode("utf-8-sig", "replace").strip() if "||", re.IGNORECASE) HTML_COMMENT_RE = re.compile(r"", re.DOTALL) # Tags draw.io actually writes. `List` and `a < b > c` are text, and a # catch-all `<[^>]+>` scan deletes them after XML decoding turns `<` into `<`. _HTML_TAG = ( "a|b|blockquote|br|code|del|div|em|font|h[1-6]|hr|i|img|ins|li|mark|ol|p|pre|" "s|small|span|strike|strong|sub|sup|table|tbody|td|th|thead|tr|u|ul|wbr" ) TAG_RE = re.compile( rf"]*>|]*>", re.IGNORECASE, ) def parse_style(style: str | None) -> dict[str, str]: out: dict[str, str] = {} if not style: return out for part in style.split(";"): part = part.strip() if not part: continue key, sep, value = part.partition("=") out[key.strip()] = value.strip() if sep else "1" return out def clean_label(value: str | None, *, is_html: bool = False) -> str: """Flatten HTML-enabled cells; preserve angle brackets in plain labels.""" if not value: return "" text = value if is_html: text = HTML_COMMENT_RE.sub("", text) text = BR_RE.sub("\n", text) text = TAG_RE.sub("", text) text = html.unescape(text) text = text.replace("\xa0", " ") lines = [re.sub(r"[ \t]+", " ", ln).strip() for ln in text.split("\n")] return "\n".join(ln for ln in lines if ln).strip() SHAPE_FAMILIES = ( ("mxgraph.aws", "aws"), ("mxgraph.azure", "azure"), ("mxgraph.gcp", "gcp"), ("mxgraph.kubernetes", "kubernetes"), ("mxgraph.cisco", "network"), ("mxgraph.veeam", "infra"), ("mxgraph.flowchart", "flowchart"), ("mxgraph.bpmn", "bpmn"), ("mxgraph.er", "er"), ("mxgraph.sysml", "uml"), ("mxgraph.archimate", "archimate"), ) # style key -> canonical shape name, checked in order SHAPE_KEYS = ( ("swimlane", "swimlane"), ("ellipse", "ellipse"), ("rhombus", "rhombus"), ("triangle", "triangle"), ("cylinder", "cylinder"), ("cylinder3", "cylinder"), ("hexagon", "hexagon"), ("cloud", "cloud"), ("actor", "actor"), ("umlActor", "actor"), ("note", "note"), ("card", "card"), ("step", "step"), ("process", "process"), ("parallelogram", "parallelogram"), ("document", "document"), ("datastore", "cylinder"), ("umlLifeline", "lifeline"), ("umlFrame", "frame"), ("table", "table"), ("tableRow", "table-row"), ("partialRectangle", "table-row"), ("image", "image"), ("text", "text"), ("group", "group"), ) def classify_shape(style: dict[str, str]) -> str: raw = style.get("shape", "") if raw: for key, name in SHAPE_KEYS: if raw == key or raw.startswith(key): return name for prefix, family in SHAPE_FAMILIES: if raw.startswith(prefix): return f"icon:{family}" return f"shape:{raw}" for key, name in SHAPE_KEYS: if key in style: return name if style.get("ellipse") == "1": return "ellipse" return "rect" def shape_family(shape: str) -> str: if shape.startswith("icon:"): return shape.split(":", 1)[1] if shape.startswith("shape:"): return "custom" return shape # -------------------------------------------------------------------------- # IR model # -------------------------------------------------------------------------- @dataclass class Node: id: str label: str = "" shape: str = "rect" parent: str | None = None depth: int = 0 x: float = 0.0 y: float = 0.0 w: float = 0.0 h: float = 0.0 fill: str = "" stroke: str = "" font_color: str = "" dashed: bool = False rounded: bool = False container: bool = False children: list[str] = field(default_factory=list) link: str = "" attrs: dict[str, str] = field(default_factory=dict) in_degree: int = 0 out_degree: int = 0 @dataclass class Edge: id: str source: str | None target: str | None label: str = "" dashed: bool = False bidirectional: bool = False undirected: bool = False style_name: str = "" waypoints: int = 0 stroke: str = "" @dataclass class Page: id: str name: str index: int nodes: list[Node] = field(default_factory=list) edges: list[Edge] = field(default_factory=list) @property def node_map(self) -> dict[str, Node]: return {n.id: n for n in self.nodes} def _num(geom: ET.Element | None, key: str) -> float: if geom is None: return 0.0 try: value = float(geom.get(key, "0") or 0) except (OverflowError, ValueError): return 0.0 if not math.isfinite(value): _fail(f"invalid geometry: {key} must be finite") return value def parse_page(diagram: ET.Element, index: int) -> Page: name = diagram.get("name") or f"Page-{index + 1}" page = Page(id=diagram.get("id") or f"page-{index}", name=name, index=index) model = diagram.find(".//mxGraphModel") if model is None: text = (diagram.text or "").strip() inflated = _inflate(text) if text else None if not inflated: return page _reject_unsafe_xml(inflated, f"page {index}") model = ET.fromstring(inflated) if model.tag != "mxGraphModel": found = model.find(".//mxGraphModel") if found is None: return page model = found root = model.find("root") if root is None: return page if len(root) > MAX_CELLS_PER_PAGE: _fail( f"page {index}: cell limit exceeded (max {MAX_CELLS_PER_PAGE})" ) # Pass 1: collect raw cells, unwrapping / containers. raw: dict[str, dict[str, Any]] = {} order: list[str] = [] for element in root: if element.tag in ("object", "UserObject"): cell = element.find("mxCell") if cell is None: continue attrs = { k: v for k, v in element.attrib.items() if k not in ("id", "label", "placeholders") } cid = element.get("id") or cell.get("id") or "" value = element.get("label", "") elif element.tag == "mxCell": cell = element attrs = {} cid = cell.get("id") or "" value = cell.get("value", "") else: continue if not cid: continue if cid in raw: _fail(f"page {index}: duplicate cell id") raw[cid] = {"cell": cell, "attrs": attrs, "value": value} order.append(cid) # draw.io hides a cell, a group or a whole layer with visible="0"; what the author hid is not # part of the diagram, so drop it together with everything parented beneath it. hidden: dict[str, bool] = {} def is_hidden(cid: str) -> bool: chain: list[str] = [] visited: set[str] = set() current, result = cid, False while current in raw and current not in visited: if current in hidden: result = hidden[current] break chain.append(current) visited.add(current) if raw[current]["cell"].get("visible") == "0": result = True break current = raw[current]["cell"].get("parent") or "" for seen in chain: hidden[seen] = result return result # Pass 2: vertices (absolute geometry resolved after the pass). edge_label_parts: dict[str, list[str]] = {} relative_offsets: dict[str, tuple[float, float]] = {} for cid in order: entry = raw[cid] cell = entry["cell"] style = parse_style(cell.get("style")) parent = cell.get("parent") if is_hidden(cid) or cell.get("edge") == "1": continue if cell.get("vertex") != "1": continue # An edge label is a vertex parented to an edge; fold it into the edge. parent_entry = raw.get(parent or "") parent_is_edge = bool( parent_entry and parent_entry["cell"].get("edge") == "1" ) if parent_is_edge or "edgeLabel" in style: if parent: text = clean_label(entry["value"], is_html=style.get("html") == "1") if text: edge_label_parts.setdefault(parent, []).append(text) continue geom = cell.find("mxGeometry") if geom is not None and geom.get("relative") == "1": offset = geom.find("mxPoint[@as='offset']") relative_offsets[cid] = (_num(offset, "x"), _num(offset, "y")) node = Node( id=cid, label=clean_label(entry["value"], is_html=style.get("html") == "1"), shape=classify_shape(style), parent=parent, x=_num(geom, "x"), y=_num(geom, "y"), w=_num(geom, "width"), h=_num(geom, "height"), fill=style.get("fillColor", ""), stroke=style.get("strokeColor", ""), font_color=style.get("fontColor", ""), dashed=style.get("dashed") == "1", rounded=style.get("rounded") == "1", container=style.get("container") == "1" or "swimlane" in style, link=entry["attrs"].get("link", ""), attrs={ k: v for k, v in entry["attrs"].items() if k not in ("link", "tooltip") }, ) page.nodes.append(node) node_map = page.node_map # Resolve absolute geometry + depth iteratively and reuse resolved prefixes. resolved_by_id: dict[str, tuple[float, float, int]] = {} for node in page.nodes: chain: list[Node] = [] active: set[str] = set() current = node while current.id not in resolved_by_id: if current.id in active: _fail(f"page {index}: parent cycle") active.add(current.id) chain.append(current) parent = node_map.get(current.parent or "") if parent is None: position = (0.0, 0.0, -1) break current = parent else: position = resolved_by_id[current.id] px, py, pdepth = position while chain: current = chain.pop() dx, dy = current.x, current.y parent = node_map.get(current.parent or "") if parent is not None and current.id in relative_offsets: ox, oy = relative_offsets[current.id] dx, dy = dx * parent.w + ox, dy * parent.h + oy px, py, pdepth = dx + px, dy + py, pdepth + 1 if not math.isfinite(px) or not math.isfinite(py): _fail(f"page {index}: geometry overflow") resolved_by_id[current.id] = (px, py, pdepth) resolved = [resolved_by_id[node.id] for node in page.nodes] for node, (ax, ay, depth) in zip(page.nodes, resolved): node.x, node.y, node.depth = ax, ay, depth parent = node_map.get(node.parent or "") if parent is not None: parent.children.append(node.id) parent.container = True # Pass 3: edges. for cid in order: entry = raw[cid] cell = entry["cell"] if is_hidden(cid) or cell.get("edge") != "1": continue style = parse_style(cell.get("style")) geom = cell.find("mxGeometry") waypoints = 0 if geom is not None: waypoints = len( [p for p in geom.findall(".//mxPoint") if p.get("as") is None] ) label = clean_label(entry["value"], is_html=style.get("html") == "1") extra = edge_label_parts.get(cid, []) if extra: label = " / ".join([p for p in ([label] + extra) if p]) source = cell.get("source") target = cell.get("target") start_head = style.get("startArrow", "none") not in ("none", "0", "") end_head = style.get("endArrow", "classic") not in ("none", "0", "") if start_head and not end_head: source, target = target, source page.edges.append( Edge( id=cid, source=source if source in node_map else None, target=target if target in node_map else None, label=label, dashed=style.get("dashed") == "1", bidirectional=start_head and end_head, undirected=not start_head and not end_head, style_name=style.get("shape", "") or ("orthogonal" if style.get("edgeStyle") else ""), waypoints=waypoints, stroke=style.get("strokeColor", ""), ) ) for edge in page.edges: source, target = node_map.get(edge.source or ""), node_map.get(edge.target or "") if edge.bidirectional or edge.undirected: for endpoint in (source, target): if endpoint is not None: endpoint.in_degree += 1 endpoint.out_degree += 1 else: if source is not None: source.out_degree += 1 if target is not None: target.in_degree += 1 return page def parse_file(path: Path) -> list[Page]: xml = load_mxfile(path) _reject_unsafe_xml(xml, path.name) try: root = ET.fromstring(xml) except ET.ParseError as exc: _fail(f"{path.name}: malformed XML ({exc})") if root.tag == "mxGraphModel": wrapper = ET.Element("diagram", {"name": path.stem, "id": "single"}) wrapper.append(root) return [parse_page(wrapper, 0)] diagrams = root.findall("diagram") if not diagrams: _fail(f"{path.name}: mxfile contains no pages") if len(diagrams) > MAX_PAGES: _fail(f"page limit exceeded (max {MAX_PAGES})") return [parse_page(d, i) for i, d in enumerate(diagrams)] # -------------------------------------------------------------------------- # structural analysis — signals, not decisions # -------------------------------------------------------------------------- def _has_cycle(nodes: list[Node], edges: list[Edge]) -> bool: adjacency: dict[str, list[str]] = {n.id: [] for n in nodes} for edge in edges: if edge.source and edge.target and edge.source in adjacency: adjacency[edge.source].append(edge.target) if edge.bidirectional and edge.target in adjacency: adjacency[edge.target].append(edge.source) WHITE, GREY, BLACK = 0, 1, 2 color = {n.id: WHITE for n in nodes} def visit(start: str) -> bool: stack = [(start, iter(adjacency.get(start, [])))] color[start] = GREY while stack: nid, it = stack[-1] advanced = False for nxt in it: state = color.get(nxt, BLACK) if state == GREY: return True if state == WHITE: color[nxt] = GREY stack.append((nxt, iter(adjacency.get(nxt, [])))) advanced = True break if not advanced: color[nid] = BLACK stack.pop() return False return any(color[n.id] == WHITE and visit(n.id) for n in nodes) def _aligned(boxes: list[Node], tolerance: float = 8.0) -> bool: """True when the boxes stack as lanes — shared left edge or shared top edge.""" if len(boxes) < 2: return False same_x = max(n.x for n in boxes) - min(n.x for n in boxes) <= tolerance same_w = max(n.w for n in boxes) - min(n.w for n in boxes) <= tolerance same_y = max(n.y for n in boxes) - min(n.y for n in boxes) <= tolerance same_h = max(n.h for n in boxes) - min(n.h for n in boxes) <= tolerance return (same_x and same_w) or (same_y and same_h) def analyze(page: Page) -> dict[str, Any]: nodes = page.nodes edges = page.edges drawable = [n for n in nodes if n.shape not in ("text",) and (n.label or n.children)] containers = [n for n in nodes if n.children] leaves = [n for n in nodes if not n.children] shapes: dict[str, int] = {} for node in nodes: shapes[shape_family(node.shape)] = shapes.get(shape_family(node.shape), 0) + 1 def name_of(node: Node) -> str: return (node.label.replace("\n", " · ") or node.id) ranked = sorted( leaves, key=lambda n: (n.in_degree + n.out_degree), reverse=True ) hubs = [ {"id": n.id, "label": name_of(n), "degree": n.in_degree + n.out_degree} for n in ranked[:5] if (n.in_degree + n.out_degree) > 0 ] sources = [name_of(n) for n in leaves if n.out_degree and not n.in_degree] sinks = [name_of(n) for n in leaves if n.in_degree and not n.out_degree] orphans = [name_of(n) for n in leaves if not n.in_degree and not n.out_degree] # Type candidates, strongest signal first. Advisory only. candidates: list[str] = [] if shapes.get("lifeline"): candidates.append("sequence") if shapes.get("table") or shapes.get("er"): candidates.append("er") lanes = [n for n in nodes if n.shape == "swimlane" and n.children] if len(lanes) >= 2 and _aligned(lanes): candidates.append("swimlane") if shapes.get("rhombus"): candidates.append("flowchart") if shapes.get("ellipse", 0) >= max(2, len(leaves) // 3) and edges: candidates.append("state") if any(f in shapes for f in ("aws", "azure", "gcp", "kubernetes", "network")): candidates.append("architecture") if containers and not shapes.get("swimlane"): candidates.append("nested") if edges and not _has_cycle(nodes, edges) and len(sources) == 1: candidates.append("tree") if edges: candidates.append("architecture") if not candidates: candidates.append("architecture") seen: set[str] = set() candidates = [c for c in candidates if not (c in seen or seen.add(c))] # Collapse candidates: containers whose children are all leaves, and # fan-out clusters — the first things to merge when simplifying. collapsible = [ { "id": c.id, "label": name_of(c), "children": len(c.children), "child_labels": [ name_of(page.node_map[cid]) for cid in c.children if page.node_map.get(cid) and page.node_map[cid].label ][:8], } for c in containers if c.children and all(not page.node_map[cid].children for cid in c.children) ] collapsible.sort(key=lambda c: c["children"], reverse=True) return { "nodes_total": len(nodes), "nodes_drawable": len(drawable), "containers": len(containers), "leaves": len(leaves), "edges_total": len(edges), "edges_labeled": sum(1 for e in edges if e.label), "edges_dangling": sum(1 for e in edges if not (e.source and e.target)), "max_depth": max((n.depth for n in nodes), default=0), "shapes": dict(sorted(shapes.items(), key=lambda kv: -kv[1])), "has_cycle": _has_cycle(nodes, edges), "hubs": hubs, "entry_points": sources[:6], "terminals": sinks[:6], "orphans": orphans[:6], "type_candidates": candidates[:3], "collapsible_groups": collapsible[:8], "over_node_budget": len(drawable) > 9, "over_edge_budget": len(edges) > 12, } # -------------------------------------------------------------------------- # rendering the digest # -------------------------------------------------------------------------- def _escape_markdown(text: str) -> str: encoded = html.escape(text, quote=False) return re.sub(r"([\\`*{}\[\]()#+\-.!_|>])", r"\\\1", encoded) def _fold_lines(text: str, replacement: str) -> str: return replacement.join(text.splitlines()) def _escape_inline(text: str) -> str: return _escape_markdown(_fold_lines(text, " · ")) def _escape_table(text: str) -> str: return _escape_markdown(_fold_lines(text, " ⏎ ")) def page_bounds(page: Page) -> tuple[float, float, float, float]: boxes = [(n.x, n.y, n.x + n.w, n.y + n.h) for n in page.nodes if n.w and n.h] if any(not math.isfinite(value) for box in boxes for value in box): _fail("invalid geometry: bounding box overflow") if not boxes: return (0.0, 0.0, 0.0, 0.0) bounds = ( min(b[0] for b in boxes), min(b[1] for b in boxes), max(b[2] for b in boxes), max(b[3] for b in boxes), ) if not math.isfinite(bounds[2] - bounds[0]) or not math.isfinite( bounds[3] - bounds[1] ): _fail("invalid geometry: canvas span overflow") return bounds def digest(path: Path, pages: list[Page], selected: list[Page], max_rows: int) -> str: out: list[str] = [] out.append(f"# draw.io IR — {_escape_inline(path.name)}") out.append("") out.append( f"{len(pages)} page(s): " + ", ".join( f"[{p.index}] {_escape_inline(p.name)} ({len(p.nodes)}n/{len(p.edges)}e)" for p in pages ) ) for page in selected: info = analyze(page) x0, y0, x1, y1 = page_bounds(page) out.append("") out.append(f"## Page {page.index} — {_escape_inline(page.name)}") out.append("") out.append( f"- source canvas: {int(x1 - x0)}×{int(y1 - y0)} px " f"(aspect {((x1 - x0) / (y1 - y0)):.2f})" if y1 > y0 else "- source canvas: empty" ) out.append( f"- nodes: {info['nodes_total']} total / {info['nodes_drawable']} drawable " f"/ {info['containers']} containers, depth {info['max_depth']}" ) out.append( f"- edges: {info['edges_total']} ({info['edges_labeled']} labeled, " f"{info['edges_dangling']} dangling), cycle: {info['has_cycle']}" ) out.append(f"- shapes: {info['shapes']}") out.append(f"- type candidates: {', '.join(info['type_candidates'])}") out.append( f"- budget: nodes {'OVER' if info['over_node_budget'] else 'ok'} (max 9), " f"edges {'OVER' if info['over_edge_budget'] else 'ok'} (max 12)" ) if info["hubs"]: hubs = ", ".join( f"{_escape_inline(h['label'] or h['id'])}({h['degree']})" for h in info["hubs"] ) out.append(f"- hubs (focal candidates): {hubs}") if info["entry_points"]: out.append( f"- entry points: {', '.join(_escape_inline(label) for label in info['entry_points'])}" ) if info["terminals"]: out.append( f"- terminals: {', '.join(_escape_inline(label) for label in info['terminals'])}" ) if info["orphans"]: out.append( f"- unconnected: {', '.join(_escape_inline(label) for label in info['orphans'])}" ) if info["collapsible_groups"]: out.append("- collapsible groups (simplify here first):") for group in info["collapsible_groups"]: kids = ", ".join(_escape_inline(label) for label in group["child_labels"]) out.append( f" - {_escape_inline(group['label'])} — " f"{group['children']} children: {kids}" ) out.append("") out.append("### Nodes") out.append("") out.append("| id | label | shape | depth | parent | deg | box |") out.append("|---|---|---|---|---|---|---|") listed = [n for n in page.nodes if n.label or n.children] for node in listed[:max_rows]: out.append( f"| {_escape_table(node.id)} | {_escape_table(node.label)} | " f"{_escape_table(node.shape)} | {node.depth} | " f"{_escape_table(node.parent or '-')} | {node.in_degree}/{node.out_degree} | " f"{int(node.x)},{int(node.y)} {int(node.w)}×{int(node.h)} |" ) if len(listed) > max_rows: out.append(f"| … | +{len(listed) - max_rows} more (use --json) | | | | | |") out.append("") out.append("### Edges") out.append("") out.append("| source | target | label | style |") out.append("|---|---|---|---|") names = {n.id: (n.label.split("\n")[0] or n.id) for n in page.nodes} for edge in page.edges[:max_rows]: marks = [] if edge.dashed: marks.append("dashed") if edge.bidirectional: marks.append("bidir") if edge.undirected: marks.append("undirected") out.append( f"| {_escape_table(names.get(edge.source or '', '?'))} | " f"{_escape_table(names.get(edge.target or '', '?'))} | " f"{_escape_table(edge.label) or '-'} | {' '.join(marks) or '-'} |" ) if len(page.edges) > max_rows: out.append(f"| … | +{len(page.edges) - max_rows} more (use --json) | | |") out.append("") return "\n".join(out) def to_json(path: Path, pages: list[Page], selected: list[Page]) -> str: payload = { "source": str(path), "pages_total": len(pages), "pages": [ { "id": p.id, "name": p.name, "index": p.index, "bounds": dict(zip(("x0", "y0", "x1", "y1"), page_bounds(p))), "analysis": analyze(p), "nodes": [asdict(n) for n in p.nodes], "edges": [asdict(e) for e in p.edges], } for p in selected ], } return json.dumps(payload, indent=2, ensure_ascii=False) def select_pages(pages: list[Page], selector: str | None) -> list[Page]: if selector is None: return pages if len(pages) == 1 else pages[:1] if selector == "all": return pages if selector.isdecimal(): index = int(selector) match = [p for p in pages if p.index == index] if not match: _fail(f"no page with index {index} (have 0..{len(pages) - 1})") return match match = [p for p in pages if p.name.lower() == selector.lower()] if not match: names = ", ".join(p.name for p in pages) _fail(f"no page named {selector!r} (have: {names})") return match def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__.split("\n")[0]) parser.add_argument("file", help=".drawio / .xml / .drawio.png / .drawio.svg") parser.add_argument( "--page", help="page index, page name, or 'all' (default: first page)", ) parser.add_argument("--json", action="store_true", help="emit the full IR as JSON") parser.add_argument( "--max-rows", type=int, default=40, help="rows per table in the Markdown digest (default 40)", ) parser.add_argument("--out", help="write to this path instead of stdout") args = parser.parse_args(argv) if args.max_rows < 1: parser.error("--max-rows must be at least 1") path = Path(args.file) if not path.is_file(): _fail(f"{path}: no such file") pages = parse_file(path) selected = select_pages(pages, args.page) text = ( to_json(path, pages, selected) if args.json else digest(path, pages, selected, args.max_rows) ) if args.out: try: Path(args.out).write_text(text, encoding="utf-8") except OSError as error: _fail(f"cannot write {args.out}: {error}") print(f"wrote {args.out} ({len(text)} bytes)") else: sys.stdout.write(text if text.endswith("\n") else text + "\n") return 0 if __name__ == "__main__": _configure_stdout_utf8() raise SystemExit(main()) ================ File: workspace/agents/opencode/engine/skills/diagram-design/scripts/excalidraw_extract.py ================ #!/usr/bin/env python3 """Extract a normalized intermediate representation (IR) from an Excalidraw scene. Trust boundary: this program parses bounded JSON. It never renders, fetches, or executes anything the scene carries. Element links, embed URLs, and binary file payloads (``files``, ``fileId``, ``dataURL``) are counted and discarded; labels are emitted only as inert, escaped text. Prompt-looking label content is diagram data, never an instruction. The input is a saved ``.excalidraw`` (or ``.excalidraw.json``) scene. PNG/SVG exports are rejected with a named failure — ask for the source scene instead of scraping pixels. Usage: python3 excalidraw_extract.py [--json] [--max-rows N] [--out PATH] Default output is a compact Markdown digest meant to be read into context. ``--json`` emits the full IR instead. Exit codes: 0 ok, 2 unreadable / unsupported / over limits. """ from __future__ import annotations import argparse import html import json import math import re import sys from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Any, NoReturn MAX_INPUT_BYTES = 16 * 1024 * 1024 MAX_ELEMENTS = 10000 MAX_NODES = 2000 MAX_EDGES = 5000 EXCALIDRAW_SUFFIXES = (".excalidraw", ".json") EXPORT_SUFFIXES = (".png", ".svg") # Excalidraw element type -> IR shape family. Everything else is either an # edge (arrow, line), folded (text), counted-and-dropped (freedraw, image # pixels), or reported as unknown. NODE_SHAPES = { "rectangle": "rect", "ellipse": "ellipse", "diamond": "rhombus", "image": "image", "embeddable": "embed", "iframe": "embed", } EDGE_TYPES = {"arrow", "line"} CONTAINER_TYPES = {"frame", "magicframe"} def _configure_stdout_utf8() -> None: """Emit digests as UTF-8 even when Windows selects a legacy codepage.""" reconfigure = getattr(sys.stdout, "reconfigure", None) if reconfigure is not None: reconfigure(encoding="utf-8", errors="strict") def _fail(message: str) -> NoReturn: print(f"excalidraw_extract: {message}", file=sys.stderr) raise SystemExit(2) # -------------------------------------------------------------------------- # IR model — mirrors the draw.io / Mermaid extractors' digest shape # -------------------------------------------------------------------------- @dataclass class Node: id: str label: str = "" shape: str = "rect" parent: str | None = None depth: int = 0 x: float = 0.0 y: float = 0.0 w: float = 0.0 h: float = 0.0 fill: str = "" stroke: str = "" dashed: bool = False container: bool = False children: list[str] = field(default_factory=list) groups: list[str] = field(default_factory=list) in_degree: int = 0 out_degree: int = 0 @dataclass class Edge: id: str source: str | None target: str | None label: str = "" dashed: bool = False bidirectional: bool = False undirected: bool = False waypoints: int = 0 stroke: str = "" @dataclass class Scene: name: str nodes: list[Node] = field(default_factory=list) edges: list[Edge] = field(default_factory=list) discarded: dict[str, int] = field( default_factory=lambda: { "freedraw_strokes": 0, "image_payloads": 0, "links": 0, "embeds": 0, "deleted_elements": 0, "unknown_elements": 0, } ) @property def node_map(self) -> dict[str, Node]: return {node.id: node for node in self.nodes} def clean_label(value: Any) -> str: """Excalidraw text is plain, but normalize whitespace defensively.""" if not isinstance(value, str): return "" text = value.replace("\r\n", "\n").replace("\r", "\n") text = "".join(ch for ch in text if ch == "\n" or ch >= " " or ch == "\t") lines = [re.sub(r"[ \t]+", " ", line).strip() for line in text.split("\n")] return "\n".join(line for line in lines if line).strip() def _num(element: dict[str, Any], key: str) -> float: """A geometry field as a finite float; absent or non-numeric reads as 0. A scene is untrusted input. An out-of-range int and the JSON tokens ``Infinity``/``NaN`` both survive as floats that only blow up later, in the digest's integer formatting, as an uncaught OverflowError or ValueError rather than the promised exit-2 diagnostic. Reject them here instead. """ value = element.get(key, 0) if isinstance(value, bool) or not isinstance(value, (int, float)): return 0.0 try: value = float(value) except OverflowError: _fail(f"invalid geometry: {key} is out of range") if not math.isfinite(value): _fail(f"invalid geometry: {key} must be finite") return value def load_scene_data(path: Path) -> dict[str, Any]: suffix = path.suffix.casefold() if suffix in EXPORT_SUFFIXES: _fail( f"{path.name}: PNG/SVG exports are not supported; save the scene " "as .excalidraw and import that file" ) if suffix not in EXCALIDRAW_SUFFIXES: _fail(f"{path.name}: not an Excalidraw file") try: with path.open("rb") as source: data = source.read(MAX_INPUT_BYTES + 1) except OSError as error: _fail(f"{path}: {error}") if len(data) > MAX_INPUT_BYTES: _fail( f"source exceeds the {MAX_INPUT_BYTES // (1024 * 1024)} MiB limit" ) try: document = json.loads( data.decode("utf-8"), parse_constant=lambda token: _fail( f"invalid geometry: {token} is not finite" ), ) except RecursionError: _fail(f"{path.name}: Excalidraw JSON nesting exceeds the parser limit") except (UnicodeDecodeError, ValueError): _fail(f"{path.name}: not valid Excalidraw JSON") if not isinstance(document, dict) or document.get("type") != "excalidraw": _fail(f"{path.name}: not an Excalidraw scene (missing type: excalidraw)") elements = document.get("elements") if not isinstance(elements, list): _fail(f"{path.name}: scene has no elements array") if len(elements) > MAX_ELEMENTS: _fail(f"element limit exceeded (max {MAX_ELEMENTS})") return document def parse_scene(path: Path, document: dict[str, Any]) -> Scene: scene = Scene(name=path.stem.removesuffix(".excalidraw")) elements = [ element for element in document["elements"] if isinstance(element, dict) and isinstance(element.get("id"), str) ] seen_ids: set[str] = set() for element in elements: element_id = element["id"] if element_id in seen_ids: _fail("duplicate element id") seen_ids.add(element_id) live: list[dict[str, Any]] = [] for element in elements: if not isinstance(element.get("type"), str): _fail("invalid element type: expected a string") if element.get("isDeleted"): scene.discarded["deleted_elements"] += 1 continue live.append(element) foldable_ids = { element["id"] for element in live if element.get("type") in NODE_SHAPES or element.get("type") in EDGE_TYPES } # Pass 1: fold bound text into its container (node label or edge label). bound_labels: dict[str, list[str]] = {} for element in live: if element.get("type") != "text": continue container = element.get("containerId") if isinstance(container, str) and container in foldable_ids: text = clean_label(element.get("text")) if text: bound_labels.setdefault(container, []).append(text) def label_for(element: dict[str, Any]) -> str: return "\n".join(bound_labels.get(element["id"], [])) # Pass 2: nodes (frames become containers; standalone text stays a node). frame_ids: set[str] = set() for element in live: kind = element.get("type") element_id = element["id"] if isinstance(element.get("link"), str) and element.get("link"): scene.discarded["links"] += 1 if kind in EDGE_TYPES: continue if kind == "text": if isinstance(element.get("containerId"), str) and element["containerId"] in foldable_ids: continue # folded into its container in pass 1 shape, label = "text", clean_label(element.get("text")) elif kind in CONTAINER_TYPES: frame_ids.add(element_id) shape = "frame" label = clean_label(element.get("name")) or f"Frame {len(frame_ids)}" elif kind in NODE_SHAPES: shape = NODE_SHAPES[kind] label = label_for(element) if kind == "image": # The pixels never cross the trust boundary; only the box does. scene.discarded["image_payloads"] += 1 if kind in ("embeddable", "iframe"): scene.discarded["embeds"] += 1 elif kind == "freedraw": scene.discarded["freedraw_strokes"] += 1 continue elif kind in ("selection", "laser"): continue # editor-only artifacts, no content else: scene.discarded["unknown_elements"] += 1 continue if len(scene.nodes) >= MAX_NODES: _fail(f"node limit exceeded (max {MAX_NODES})") groups = [ group for group in element.get("groupIds", []) if isinstance(group, str) ] if isinstance(element.get("groupIds"), list) else [] scene.nodes.append( Node( id=element_id, label=label, shape=shape, x=_num(element, "x"), y=_num(element, "y"), w=_num(element, "width"), h=_num(element, "height"), fill=element.get("backgroundColor", "") if isinstance(element.get("backgroundColor"), str) else "", stroke=element.get("strokeColor", "") if isinstance(element.get("strokeColor"), str) else "", dashed=element.get("strokeStyle") in ("dashed", "dotted"), container=shape == "frame", groups=groups, ) ) node_map = scene.node_map # Frame membership: Excalidraw frames don't nest, so members sit at depth 1. for element in live: frame = element.get("frameId") if not isinstance(frame, str) or frame not in frame_ids: continue member = node_map.get(element["id"]) parent = node_map.get(frame) if member is None or parent is None or member.id == parent.id: continue member.parent = frame member.depth = 1 parent.children.append(member.id) # Pass 3: edges. Arrows are directed unless both arrowheads vanish; lines # are undirected unless the author added arrowheads. def binding_id(element: dict[str, Any], key: str) -> str | None: binding = element.get(key) if isinstance(binding, dict) and isinstance(binding.get("elementId"), str): bound = binding["elementId"] return bound if bound in node_map else None return None for element in live: kind = element.get("type") if kind not in EDGE_TYPES: continue if len(scene.edges) >= MAX_EDGES: _fail(f"edge limit exceeded (max {MAX_EDGES})") default_end = "arrow" if kind == "arrow" else None start_head = element.get("startArrowhead") end_head = element.get("endArrowhead", default_end) if start_head is not None and not isinstance(start_head, str): _fail("invalid startArrowhead: expected a string or null") if end_head is not None and not isinstance(end_head, str): _fail("invalid endArrowhead: expected a string or null") start_bound = binding_id(element, "startBinding") end_bound = binding_id(element, "endBinding") # The semantic source is the tail, not necessarily startBinding: a # start-only arrowhead points from the end binding back to the start. source, target = ( (end_bound, start_bound) if start_head and not end_head else (start_bound, end_bound) ) points = element.get("points") waypoints = max(len(points) - 2, 0) if isinstance(points, list) else 0 scene.edges.append( Edge( id=element["id"], source=source, target=target, label=label_for(element), dashed=element.get("strokeStyle") in ("dashed", "dotted"), bidirectional=bool(start_head) and bool(end_head), undirected=not start_head and not end_head, waypoints=waypoints, stroke=element.get("strokeColor", "") if isinstance(element.get("strokeColor"), str) else "", ) ) for edge in scene.edges: source = node_map.get(edge.source or "") target = node_map.get(edge.target or "") if edge.bidirectional or edge.undirected: for endpoint in (source, target): if endpoint is not None: endpoint.in_degree += 1 endpoint.out_degree += 1 else: if source is not None: source.out_degree += 1 if target is not None: target.in_degree += 1 return scene # -------------------------------------------------------------------------- # structural analysis — signals, not decisions # -------------------------------------------------------------------------- def _has_cycle(nodes: list[Node], edges: list[Edge]) -> bool: adjacency: dict[str, list[str]] = {node.id: [] for node in nodes} for edge in edges: if edge.source and edge.target and edge.source in adjacency: adjacency[edge.source].append(edge.target) if edge.bidirectional and edge.target in adjacency: adjacency[edge.target].append(edge.source) WHITE, GREY, BLACK = 0, 1, 2 color = {node.id: WHITE for node in nodes} def visit(start: str) -> bool: stack = [(start, iter(adjacency.get(start, [])))] color[start] = GREY while stack: node_id, targets = stack[-1] advanced = False for target in targets: state = color.get(target, BLACK) if state == GREY: return True if state == WHITE: color[target] = GREY stack.append((target, iter(adjacency.get(target, [])))) advanced = True break if not advanced: color[node_id] = BLACK stack.pop() return False return any(color[node.id] == WHITE and visit(node.id) for node in nodes) def analyze(scene: Scene) -> dict[str, Any]: nodes = scene.nodes edges = scene.edges drawable = [ node for node in nodes if node.shape != "text" and (node.label or node.children) ] containers = [node for node in nodes if node.children] leaves = [node for node in nodes if not node.children] shapes: dict[str, int] = {} for node in nodes: shapes[node.shape] = shapes.get(node.shape, 0) + 1 def name_of(node: Node) -> str: return node.label.replace("\n", " · ") or node.id ranked = sorted(leaves, key=lambda n: n.in_degree + n.out_degree, reverse=True) hubs = [ {"id": node.id, "label": name_of(node), "degree": node.in_degree + node.out_degree} for node in ranked[:5] if node.in_degree + node.out_degree > 0 ] sources = [name_of(n) for n in leaves if n.out_degree and not n.in_degree] sinks = [name_of(n) for n in leaves if n.in_degree and not n.out_degree] orphans = [ name_of(n) for n in leaves if not n.in_degree and not n.out_degree and n.shape != "frame" ] candidates: list[str] = [] if shapes.get("rhombus"): candidates.append("flowchart") if shapes.get("ellipse", 0) >= max(2, len(leaves) // 3) and edges: candidates.append("state") if containers: candidates.append("nested") if edges and not _has_cycle(nodes, edges) and len(sources) == 1: candidates.append("tree") candidates.append("architecture") seen: set[str] = set() candidates = [c for c in candidates if not (c in seen or seen.add(c))] # Collapse candidates: frames whose members are all leaves, then explicit # groups — the first things to merge when simplifying. collapsible = [ { "id": container.id, "label": name_of(container), "children": len(container.children), "child_labels": [ name_of(scene.node_map[child]) for child in container.children if scene.node_map.get(child) and scene.node_map[child].label ][:8], } for container in containers if container.children and all(not scene.node_map[child].children for child in container.children) ] # Excalidraw orders group ids from deepest to shallowest (inner to outer). # Count every membership: selecting either endpoint alone loses a cluster. group_members: dict[str, list[Node]] = {} for node in nodes: seen_groups: set[str] = set() for group_id in node.groups: if not group_id or group_id in seen_groups: continue seen_groups.add(group_id) group_members.setdefault(group_id, []).append(node) for group_id, members in group_members.items(): if len(members) < 2: continue collapsible.append( { "id": group_id, "label": f"group ({len(members)} members)", "children": len(members), "child_labels": [name_of(m) for m in members if m.label][:8], } ) collapsible.sort(key=lambda c: c["children"], reverse=True) return { "nodes_total": len(nodes), "nodes_drawable": len(drawable), "containers": len(containers), "leaves": len(leaves), "edges_total": len(edges), "edges_labeled": sum(1 for e in edges if e.label), "edges_dangling": sum(1 for e in edges if not (e.source and e.target)), "max_depth": max((n.depth for n in nodes), default=0), "shapes": dict(sorted(shapes.items(), key=lambda kv: (-kv[1], kv[0]))), "has_cycle": _has_cycle(nodes, edges), "hubs": hubs, "entry_points": sources[:6], "terminals": sinks[:6], "orphans": orphans[:6], "type_candidates": candidates[:3], "collapsible_groups": collapsible[:8], "over_node_budget": len(drawable) > 9, "over_edge_budget": len(edges) > 12, } # -------------------------------------------------------------------------- # rendering the digest # -------------------------------------------------------------------------- def _escape_markdown(text: str) -> str: encoded = html.escape(text, quote=False) return re.sub(r"([\\`*{}\[\]()#+\-.!_|>])", r"\\\1", encoded) def _fold_lines(text: str, replacement: str) -> str: return replacement.join(text.splitlines()) def _escape_inline(text: str) -> str: return _escape_markdown(_fold_lines(text, " · ")) def _escape_table(text: str) -> str: return _escape_markdown(_fold_lines(text, " ⏎ ")) def scene_bounds(scene: Scene) -> tuple[float, float, float, float]: boxes = [(n.x, n.y, n.x + n.w, n.y + n.h) for n in scene.nodes if n.w and n.h] if any(not math.isfinite(value) for box in boxes for value in box): _fail("invalid geometry: bounding box overflow") if not boxes: return (0.0, 0.0, 0.0, 0.0) bounds = ( min(b[0] for b in boxes), min(b[1] for b in boxes), max(b[2] for b in boxes), max(b[3] for b in boxes), ) width, height = bounds[2] - bounds[0], bounds[3] - bounds[1] ratio = width / height if height > 0 else 0 if not all(math.isfinite(value) for value in (width, height, ratio)): _fail("invalid geometry: canvas arithmetic overflow") return bounds def digest(path: Path, scene: Scene, max_rows: int) -> str: info = analyze(scene) x0, y0, x1, y1 = scene_bounds(scene) out: list[str] = [] out.append(f"# Excalidraw IR — {_escape_inline(path.name)}") out.append("") out.append( f"1 scene: {_escape_inline(scene.name)} " f"({len(scene.nodes)}n/{len(scene.edges)}e)" ) out.append("") out.append(f"## Scene — {_escape_inline(scene.name)}") out.append("") out.append( f"- source canvas: {int(x1 - x0)}×{int(y1 - y0)} px " f"(aspect {((x1 - x0) / (y1 - y0)):.2f})" if y1 > y0 else "- source canvas: empty" ) out.append( f"- nodes: {info['nodes_total']} total / {info['nodes_drawable']} drawable " f"/ {info['containers']} containers, depth {info['max_depth']}" ) out.append( f"- edges: {info['edges_total']} ({info['edges_labeled']} labeled, " f"{info['edges_dangling']} dangling), cycle: {info['has_cycle']}" ) out.append(f"- shapes: {info['shapes']}") out.append(f"- type candidates: {', '.join(info['type_candidates'])}") out.append( f"- budget: nodes {'OVER' if info['over_node_budget'] else 'ok'} (max 9), " f"edges {'OVER' if info['over_edge_budget'] else 'ok'} (max 12)" ) dropped = {key: count for key, count in scene.discarded.items() if count} if dropped: out.append( "- discarded: " + ", ".join( f"{count} {key.replace('_', ' ')}" for key, count in dropped.items() ) ) if info["hubs"]: hubs = ", ".join( f"{_escape_inline(h['label'] or h['id'])}({h['degree']})" for h in info["hubs"] ) out.append(f"- hubs (focal candidates): {hubs}") if info["entry_points"]: out.append( f"- entry points: {', '.join(_escape_inline(label) for label in info['entry_points'])}" ) if info["terminals"]: out.append( f"- terminals: {', '.join(_escape_inline(label) for label in info['terminals'])}" ) if info["orphans"]: out.append( f"- unconnected: {', '.join(_escape_inline(label) for label in info['orphans'])}" ) if info["collapsible_groups"]: out.append("- collapsible groups (simplify here first):") for group in info["collapsible_groups"]: kids = ", ".join(_escape_inline(label) for label in group["child_labels"]) out.append( f" - {_escape_inline(group['label'])} — " f"{group['children']} children: {kids}" ) out.append("") out.append("### Nodes") out.append("") out.append("| id | label | shape | depth | parent | deg | box |") out.append("|---|---|---|---|---|---|---|") listed = [n for n in scene.nodes if n.label or n.children] for node in listed[:max_rows]: out.append( f"| {_escape_table(node.id)} | {_escape_table(node.label)} | " f"{_escape_table(node.shape)} | {node.depth} | " f"{_escape_table(node.parent or '-')} | {node.in_degree}/{node.out_degree} | " f"{int(node.x)},{int(node.y)} {int(node.w)}×{int(node.h)} |" ) if len(listed) > max_rows: out.append(f"| … | +{len(listed) - max_rows} more (use --json) | | | | | |") out.append("") out.append("### Edges") out.append("") out.append("| source | target | label | style |") out.append("|---|---|---|---|") names = {n.id: (n.label.split("\n")[0] or n.id) for n in scene.nodes} for edge in scene.edges[:max_rows]: marks = [] if edge.dashed: marks.append("dashed") if edge.bidirectional: marks.append("bidir") if edge.undirected: marks.append("undirected") out.append( f"| {_escape_table(names.get(edge.source or '', '?'))} | " f"{_escape_table(names.get(edge.target or '', '?'))} | " f"{_escape_table(edge.label) or '-'} | {' '.join(marks) or '-'} |" ) if len(scene.edges) > max_rows: out.append(f"| … | +{len(scene.edges) - max_rows} more (use --json) | | |") out.append("") return "\n".join(out) def to_json(path: Path, scene: Scene) -> str: payload = { "source": str(path), "scene": { "name": scene.name, "bounds": dict(zip(("x0", "y0", "x1", "y1"), scene_bounds(scene))), "analysis": analyze(scene), "discarded": scene.discarded, "nodes": [asdict(node) for node in scene.nodes], "edges": [asdict(edge) for edge in scene.edges], }, } return json.dumps(payload, indent=2, ensure_ascii=False) def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__.split("\n")[0]) parser.add_argument("file", help=".excalidraw / .excalidraw.json scene") parser.add_argument("--json", action="store_true", help="emit the full IR as JSON") parser.add_argument( "--max-rows", type=int, default=40, help="rows per table in the Markdown digest (default 40)", ) parser.add_argument("--out", help="write to this path instead of stdout") args = parser.parse_args(argv) if args.max_rows < 1: _fail("--max-rows must be at least 1") path = Path(args.file) if not path.is_file(): _fail(f"{path}: no such file") scene = parse_scene(path, load_scene_data(path)) text = to_json(path, scene) if args.json else digest(path, scene, args.max_rows) if args.out: try: Path(args.out).write_text(text, encoding="utf-8") except OSError as error: _fail(f"cannot write {args.out}: {error}") print(f"wrote {args.out} ({len(text)} bytes)") else: sys.stdout.write(text if text.endswith("\n") else text + "\n") return 0 if __name__ == "__main__": _configure_stdout_utf8() raise SystemExit(main()) ================ File: workspace/agents/opencode/engine/skills/diagram-design/scripts/export_png.py ================ #!/usr/bin/env python3 """Rasterize the first SVG in a diagram HTML file with Playwright.""" from __future__ import annotations import argparse import math import pathlib import sys from playwright.sync_api import TimeoutError as PlaywrightTimeoutError, sync_playwright NETWORK_IDLE_TIMEOUT_MS = 15_000 FALLBACK_SETTLE_MS = 4_000 def exact_size_scale(value: str) -> float: try: scale = float(value) except ValueError as exc: raise argparse.ArgumentTypeError("scale must be a number") from exc if not math.isfinite(scale) or not 1 <= scale <= 4: raise argparse.ArgumentTypeError("scale must be between 1 and 4") return scale def rasterize( src: pathlib.Path, out: pathlib.Path, scale: float = 2, *, network_idle_timeout_ms: int = NETWORK_IDLE_TIMEOUT_MS, fallback_settle_ms: int = FALLBACK_SETTLE_MS, ) -> None: with sync_playwright() as playwright: browser = playwright.chromium.launch() try: page = browser.new_page(device_scale_factor=scale) page.goto( src.resolve().as_uri() + "?motion=static", wait_until="domcontentloaded", ) try: page.wait_for_load_state("networkidle", timeout=network_idle_timeout_ms) except PlaywrightTimeoutError: # Proxied networks may stall webfont subrequests past any deadline. # Cancel the outstanding load so capture cannot block on it again. page.evaluate("window.stop()") page.wait_for_timeout(fallback_settle_ms) print( "warning: webfont request stalled - captured with fallback " "typography; the PNG may not match the intended fonts", file=sys.stderr, ) page.evaluate("document.fonts.ready") if not page.locator("[data-motion-root]").evaluate_all( "roots => roots.every(root => root.dataset.frame === 'static')" ): raise RuntimeError("motion export requires a complete static frame") svg = page.locator("svg").first # Capture at one CSS pixel per viewBox unit before applying the multiplier. svg.evaluate( """el => { const box = el.viewBox.baseVal; if (![box.width, box.height].every(n => Number.isFinite(n) && n > 0)) throw new Error('PNG export requires a nonzero finite viewBox'); for (const [axis, size] of [['width', box.width], ['height', box.height]]) for (const prefix of ['', 'min-', 'max-']) el.style.setProperty(prefix + axis, size + 'px', 'important'); }""" ) # Release every clipping ancestor so a wide SVG is captured whole. svg.evaluate( "el => { for (let a = el.parentElement; a; a = a.parentElement) " "a.style.setProperty('overflow', 'visible', 'important'); }" ) svg.screenshot(path=out, omit_background=True) finally: browser.close() def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("src", type=pathlib.Path, help="diagram HTML input") parser.add_argument("out", type=pathlib.Path, help="PNG output") parser.add_argument("scale", nargs="?", type=exact_size_scale, default=2) args = parser.parse_args() rasterize(args.src, args.out, args.scale) return 0 if __name__ == "__main__": raise SystemExit(main()) ================ File: workspace/agents/opencode/engine/skills/diagram-design/scripts/export_svg.py ================ #!/usr/bin/env python3 """Export a diagram HTML file to a standalone, inline-safe SVG. Ships inside the skill so an installed agent can produce a portable SVG without re-deriving the transform: python3 /scripts/export_svg.py my-diagram.html python3 /scripts/export_svg.py my-diagram.html out.svg Fixes the export gaps that make a fragment unsafe to open or inline: 1. Class-styled diagrams keep their page ``" defs_match = re.search(r"]*>", svg, re.IGNORECASE) if defs_match: insert_at = defs_match.end() return svg[:insert_at] + "\n " + style_tag + svg[insert_at:] # Keep the authored first-child title/desc ahead of a newly created defs. open_match = re.match(r"]*>", svg, re.IGNORECASE) assert open_match is not None insert_at = leading_accessibility_end(svg, open_match.end()) return ( svg[:insert_at] + f"\n \n {style_tag}\n " + svg[insert_at:] ) def find_defs_ids(svg: str) -> list[str]: """IDs on referenceable elements living under , including comments.""" defs_blocks = re.findall(r"]*>(.*?)", svg, re.IGNORECASE | re.DOTALL) allowed_tags = {kind.lower() for kind in DEFS_ID_TAGS} found: list[str] = [] seen: set[str] = set() for block in defs_blocks: for tag in START_TAG_RE.finditer(block): name, attrs, _ = tag.groups() if name.lower() not in allowed_tags: continue for attribute in TAG_ATTR_RE.finditer(attrs): _, name, _, quoted = attribute.groups() if name != "id" or quoted is None or quoted[0] not in "\"'": continue ident = decode_xml_references(quoted[1:-1]) if ident not in seen: seen.add(ident) found.append(ident) break return found # Markup that can carry defs references: