"""Per-sheet mermaid extraction from Excel drawings. Walks each worksheet's drawing XML, detects connected shape+connector components, and converts each component to a mermaid flowchart block whose top-left cell range is reported as the embed target. Self-contained: defines its own ShapeInfo / ConnectorInfo / SheetDrawing dataclasses and parser. The legacy `src.extraction_v2.connector_parser` module ignores flipV and explicit / attach refs, so we do not reuse it here. Public entry point: extract_mermaid_graphs(xlsx_path, direction="TD") -> Dict[sheet_name, Tuple[List[GraphResult], List[str]]] Each sheet returns (graphs, consumed_cell_coords). `consumed_cell_coords` are worksheet cells whose text was mined as edge-label content; callers rewriting the workbook should clear those cells so the same text does not re-appear in the downstream markdown alongside the mermaid block. """ from __future__ import annotations import bisect import logging import re import xml.etree.ElementTree as ET import zipfile from dataclasses import dataclass from typing import Dict, Iterable, List, Optional, Tuple logger = logging.getLogger(__name__) NS_XDR = "http://schemas.openxmlformats.org/drawingml/2006/spreadsheetDrawing" NS_A = "http://schemas.openxmlformats.org/drawingml/2006/main" NS_MAIN = "http://schemas.openxmlformats.org/spreadsheetml/2006/main" NS_R = "http://schemas.openxmlformats.org/officeDocument/2006/relationships" NS_PR = "http://schemas.openxmlformats.org/package/2006/relationships" _NS = {"xdr": NS_XDR, "a": NS_A} _NS_FULL = {"xdr": NS_XDR, "a": NS_A, "main": NS_MAIN, "r": NS_R} # --------------------------------------------------------------------------- # Dataclasses # --------------------------------------------------------------------------- @dataclass class ShapeInfo: id: str name: str text: str shape_type: str position: Tuple[int, int] size: Tuple[int, int] center: Tuple[int, int] sheet_name: str fill_color: Optional[str] = None @dataclass class ConnectorInfo: id: str name: str start_pos: Tuple[int, int] end_pos: Tuple[int, int] text: str has_arrow: bool sheet_name: str explicit_start_id: Optional[str] = None explicit_end_id: Optional[str] = None @dataclass class SheetDrawing: sheet_name: str sheet_xml_path: str drawing_file: Optional[str] shapes: List[ShapeInfo] connectors: List[ConnectorInfo] @dataclass class GraphResult: sheet: str location: str # "A12:M30" cell range mermaid_diagram: str # mermaid flowchart text (no fenced wrapper) # --------------------------------------------------------------------------- # Drawing XML parser # --------------------------------------------------------------------------- class ConnectorParser: """Drawing-XML parser tuned for flowchart extraction. Differences vs. the legacy connector_parser.ConnectorParser: - Honors both flipH and flipV when computing start/end positions. - Reads explicit / attach refs and prefers them over proximity when inferring connections. - Provides a per-sheet API (parse_by_sheet) that resolves workbook.xml -> sheet -> drawing relationships. """ DEFAULT_CONNECTOR_MAX_DISTANCE = 500_000 # EMU def _extract_text(self, elem: ET.Element) -> str: parts: List[str] = [] for t in elem.findall(".//a:t", _NS): if t.text: parts.append(t.text) return " ".join(p.strip() for p in parts if p.strip()) def _extract_fill_color(self, elem: ET.Element) -> Optional[str]: spPr = elem.find(".//xdr:spPr", _NS) if spPr is None: return None srgb = spPr.find(".//a:solidFill/a:srgbClr", _NS) return srgb.get("val") if srgb is not None else None def _parse_shapes(self, root: ET.Element, sheet_name: str) -> List[ShapeInfo]: shapes: List[ShapeInfo] = [] for sp in root.findall(".//xdr:sp", _NS): cNvPr = sp.find(".//xdr:cNvPr", _NS) if cNvPr is None: continue xfrm = sp.find(".//a:xfrm", _NS) if xfrm is None: continue off = xfrm.find("a:off", _NS) ext = xfrm.find("a:ext", _NS) if off is None or ext is None: continue x = int(off.get("x", "0")) y = int(off.get("y", "0")) w = int(ext.get("cx", "0")) h = int(ext.get("cy", "0")) prstGeom = sp.find(".//a:prstGeom", _NS) shape_type = prstGeom.get("prst", "unknown") if prstGeom is not None else "unknown" shapes.append(ShapeInfo( id=cNvPr.get("id", ""), name=cNvPr.get("name", ""), text=self._extract_text(sp), shape_type=shape_type, position=(x, y), size=(w, h), center=(x + w // 2, y + h // 2), sheet_name=sheet_name, fill_color=self._extract_fill_color(sp), )) return shapes def _parse_connectors(self, root: ET.Element, sheet_name: str) -> List[ConnectorInfo]: connectors: List[ConnectorInfo] = [] for cx in root.findall(".//xdr:cxnSp", _NS): cNvPr = cx.find(".//xdr:cNvPr", _NS) if cNvPr is None: continue xfrm = cx.find(".//a:xfrm", _NS) if xfrm is None: continue off = xfrm.find("a:off", _NS) ext = xfrm.find("a:ext", _NS) if off is None or ext is None: continue x = int(off.get("x", "0")) y = int(off.get("y", "0")) w = int(ext.get("cx", "0")) h = int(ext.get("cy", "0")) flip_h = xfrm.get("flipH") == "1" flip_v = xfrm.get("flipV") == "1" sx, sy, ex, ey = x, y, x + w, y + h if flip_h: sx, ex = ex, sx if flip_v: sy, ey = ey, sy tail = cx.find(".//a:tailEnd", _NS) head = cx.find(".//a:headEnd", _NS) has_arrow = tail is not None or head is not None cNvCxnSpPr = cx.find(".//xdr:cNvCxnSpPr", _NS) st_cxn = cNvCxnSpPr.find("a:stCxn", _NS) if cNvCxnSpPr is not None else None end_cxn = cNvCxnSpPr.find("a:endCxn", _NS) if cNvCxnSpPr is not None else None connectors.append(ConnectorInfo( id=cNvPr.get("id", ""), name=cNvPr.get("name", ""), start_pos=(sx, sy), end_pos=(ex, ey), text=self._extract_text(cx), has_arrow=has_arrow, sheet_name=sheet_name, explicit_start_id=st_cxn.get("id") if st_cxn is not None else None, explicit_end_id=end_cxn.get("id") if end_cxn is not None else None, )) return connectors def _parse_drawing_xml( self, xml_content: str, sheet_name: str, ) -> Tuple[List[ShapeInfo], List[ConnectorInfo]]: root = ET.fromstring(xml_content) return self._parse_shapes(root, sheet_name), self._parse_connectors(root, sheet_name) def _find_closest_shape( self, point: Tuple[int, int], shapes: List[ShapeInfo], max_distance: int, ) -> Optional[ShapeInfo]: px, py = point best: Optional[ShapeInfo] = None best_d = float("inf") for s in shapes: x0, y0 = s.position w, h = s.size x1, y1 = x0 + w, y0 + h dx = max(x0 - px, 0, px - x1) dy = max(y0 - py, 0, py - y1) d = (dx * dx + dy * dy) ** 0.5 if d < best_d and d <= max_distance: best_d = d best = s return best def find_connected_shapes( self, shapes: List[ShapeInfo], connectors: List[ConnectorInfo], ) -> List[Tuple[str, str, str]]: shape_by_id = {s.id: s for s in shapes} result: List[Tuple[str, str, str]] = [] for c in connectors: start_shape = shape_by_id.get(c.explicit_start_id) if c.explicit_start_id else None if start_shape is None: start_shape = self._find_closest_shape( c.start_pos, shapes, self.DEFAULT_CONNECTOR_MAX_DISTANCE, ) end_shape = shape_by_id.get(c.explicit_end_id) if c.explicit_end_id else None if end_shape is None: end_shape = self._find_closest_shape( c.end_pos, shapes, self.DEFAULT_CONNECTOR_MAX_DISTANCE, ) if start_shape and end_shape and start_shape.id != end_shape.id: result.append((start_shape.id, end_shape.id, c.text or "")) logger.info("Inferred %d connections from %d connectors", len(result), len(connectors)) return result def parse_by_sheet(self, xlsx_path: str) -> List[SheetDrawing]: """Walk workbook.xml -> sheet -> drawing relationships per sheet.""" out: List[SheetDrawing] = [] with zipfile.ZipFile(xlsx_path, "r") as zf: wb_xml = ET.fromstring(zf.read("xl/workbook.xml")) sheets_meta: List[Tuple[str, str]] = [] for s in wb_xml.findall(f"{{{NS_MAIN}}}sheets/{{{NS_MAIN}}}sheet"): sheets_meta.append((s.get("name", ""), s.get(f"{{{NS_R}}}id", ""))) wb_rels_root = ET.fromstring(zf.read("xl/_rels/workbook.xml.rels")) wb_rels: Dict[str, str] = {} for r in wb_rels_root.findall(f"{{{NS_PR}}}Relationship"): wb_rels[r.get("Id", "")] = r.get("Target", "") for sheet_name, rid in sheets_meta: target = wb_rels.get(rid, "") if not target: continue sheet_xml_path = "xl/" + target.lstrip("/").removeprefix("/") sheet_xml_path = sheet_xml_path.replace("xl/xl/", "xl/") sheet_dir = sheet_xml_path.rsplit("/", 1)[0] sheet_file = sheet_xml_path.rsplit("/", 1)[1] sheet_rels_path = f"{sheet_dir}/_rels/{sheet_file}.rels" drawing_file: Optional[str] = None if sheet_rels_path in zf.namelist(): rels_root = ET.fromstring(zf.read(sheet_rels_path)) for r in rels_root.findall(f"{{{NS_PR}}}Relationship"): if r.get("Type", "").endswith("/drawing"): t = r.get("Target", "").lstrip("/") if t.startswith("../"): drawing_file = "xl/" + t[3:] else: drawing_file = f"{sheet_dir}/{t}" break shapes: List[ShapeInfo] = [] connectors: List[ConnectorInfo] = [] if drawing_file and drawing_file in zf.namelist(): xml_content = zf.read(drawing_file).decode("utf-8") shapes, connectors = self._parse_drawing_xml(xml_content, sheet_name) out.append(SheetDrawing( sheet_name=sheet_name, sheet_xml_path=sheet_xml_path, drawing_file=drawing_file, shapes=shapes, connectors=connectors, )) return out # --------------------------------------------------------------------------- # Mermaid node shape mapping (OOXML prstGeom -> mermaid wrappers) # --------------------------------------------------------------------------- SHAPE_WRAP: Dict[str, Tuple[str, str]] = { "rect": ("[", "]"), "rectangle": ("[", "]"), "roundRect": ("(", ")"), "ellipse": ("([", "])"), "oval": ("([", "])"), "diamond": ("{", "}"), "parallelogram": ("[/", "/]"), "trapezoid": ("[\\", "/]"), "hexagon": ("{{", "}}"), "flowChartProcess": ("[", "]"), "flowChartDecision": ("{", "}"), "flowChartTerminator": ("([", "])"), "flowChartData": ("[/", "/]"), "flowChartInputOutput": ("[/", "/]"), "flowChartPredefinedProcess":("[[", "]]"), "flowChartInternalStorage": ("[", "]"), "flowChartDocument": ("[", "]"), "flowChartMultidocument": ("[", "]"), "flowChartDisplay": ("[", "]"), "flowChartManualInput": ("[/", "/]"), "flowChartManualOperation": ("[\\", "/]"), "flowChartConnector": ("((", "))"), "flowChartOffpageConnector": ("((", "))"), "flowChartMagneticDisk": ("[(", ")]"), "flowChartMagneticDrum": ("[(", ")]"), "flowChartOnlineStorage": ("[(", ")]"), "flowChartAlternateProcess": ("(", ")"), "flowChartPreparation": ("{{", "}}"), "flowChartDelay": ("(", ")"), "flowChartOffpageReference": ("((", "))"), } def _shape_wrap(shape_type: str) -> Tuple[str, str]: return SHAPE_WRAP.get(shape_type, ("[", "]")) # --------------------------------------------------------------------------- # Text cleanup / shape grouping # --------------------------------------------------------------------------- DECORATIVE_NAME_HINTS = ("laptop", "禁止", "arrow", "矢印") _ANNOTATION_RE = re.compile(r"^(修正|※|注|備考|Note|NOTE)\s*[\d０-９]*$") _CJK_CHAR = ( r"　-〿" r"぀-ゟ" r"゠-ヿ" r"㐀-䶿" r"一-鿿" r"＀-￯" ) _SPACE_BETWEEN_CJK = re.compile(rf"(?<=[{_CJK_CHAR}])[ \t]+(?=[{_CJK_CHAR}])") _SPACE_AFTER_OPEN = re.compile(r"([\[\(（【])[ \t]+") _SPACE_BEFORE_CLOSE = re.compile(r"[ \t]+([\]\)）】])") def _clean_shape_text(text: str) -> str: """Strip phantom spaces inserted between text runs during XML parsing.""" if not text: return text cleaned = _SPACE_BETWEEN_CJK.sub("", text) cleaned = _SPACE_AFTER_OPEN.sub(r"\1", cleaned) cleaned = _SPACE_BEFORE_CLOSE.sub(r"\1", cleaned) return cleaned def _is_decorative(shape: ShapeInfo) -> bool: if shape.text and shape.text.strip(): return False name = shape.name.lower() return any(hint in shape.name or hint in name for hint in DECORATIVE_NAME_HINTS) def _overlap_area(a: ShapeInfo, b: ShapeInfo) -> int: ax, ay = a.position; aw, ah = a.size bx, by = b.position; bw, bh = b.size ix = max(0, min(ax + aw, bx + bw) - max(ax, bx)) iy = max(0, min(ay + ah, by + bh) - max(ay, by)) return ix * iy def _edge_distance(a: ShapeInfo, b: ShapeInfo) -> float: ax, ay = a.position; aw, ah = a.size bx, by = b.position; bw, bh = b.size dx = max(0, bx - (ax + aw), ax - (bx + bw)) dy = max(0, by - (ay + ah), ay - (by + bh)) return (dx * dx + dy * dy) ** 0.5 def _build_node_groups(shapes: List[ShapeInfo]) -> Dict[str, str]: """Collapse overlapping/decorative/annotation shapes into one canonical node id.""" canonical: Dict[str, str] = {s.id: s.id for s in shapes} text_shapes = [s for s in shapes if s.text and s.text.strip()] bg_shapes = [s for s in shapes if not (s.text and s.text.strip())] # Pass 1: text shape overlapping a background shape → background wins. for tb in text_shapes: best_bg: Optional[ShapeInfo] = None best_area = 0 for bg in bg_shapes: if _is_decorative(bg): continue area = _overlap_area(tb, bg) if area > best_area: best_area = area best_bg = bg if best_bg is not None and best_area > 0: tb_area = tb.size[0] * tb.size[1] bg_area = best_bg.size[0] * best_bg.size[1] min_area = min(tb_area, bg_area) if min_area > 0 and best_area / min_area >= 0.30: canonical[tb.id] = best_bg.id # Pass 1b: annotation badges absorbed into nearest larger text shape. _ANNOTATION_ABSORB_MAX = 400_000 # EMU for badge in text_shapes: if canonical[badge.id] != badge.id: continue if not _ANNOTATION_RE.match(badge.text.strip()): continue b_area = badge.size[0] * badge.size[1] best_target: Optional[ShapeInfo] = None best_distance = float("inf") for big in text_shapes: if big.id == badge.id: continue if _ANNOTATION_RE.match(big.text.strip()): continue if big.size[0] * big.size[1] <= b_area: continue d = _edge_distance(badge, big) if d > _ANNOTATION_ABSORB_MAX: continue if d < best_distance: best_distance = d best_target = big if best_target is not None: canonical[badge.id] = canonical.get(best_target.id, best_target.id) # Pass 2: decorative shapes pulled into overlapping text shape. for deco in shapes: if not _is_decorative(deco): continue best_target = None best_area = 0 for t in text_shapes: area = _overlap_area(deco, t) if area > best_area: best_area = area best_target = t if best_target is not None and best_area > 0: canonical[deco.id] = canonical.get(best_target.id, best_target.id) # Flatten chains (a -> b -> c becomes a -> c). for sid in list(canonical): seen = set() cur = sid while canonical[cur] != cur and cur not in seen: seen.add(cur) cur = canonical[cur] canonical[sid] = cur return canonical # --------------------------------------------------------------------------- # Mermaid emission # --------------------------------------------------------------------------- @dataclass class _MermaidNode: nid: str label: str wrap: Tuple[str, str] fill: Optional[str] def _safe_id(raw_id: str) -> str: sanitized = re.sub(r"[^A-Za-z0-9_]", "_", raw_id) return f"n{sanitized}" if sanitized and sanitized[0].isdigit() else f"n_{sanitized}" def _escape_label(text: str) -> str: if not text: return "" lines = [re.sub(r"\s+", " ", ln).strip() for ln in text.splitlines() if ln.strip()] joined = "
".join(lines) return joined.replace('"', "#quot;") def _pick_label(shapes_in_group: List[ShapeInfo]) -> str: candidates = [s.text.strip() for s in shapes_in_group if s.text and s.text.strip()] if not candidates: return "" non_annotation = [c for c in candidates if not _ANNOTATION_RE.match(c)] pool = non_annotation if non_annotation else candidates return max(pool, key=len) def _pick_shape_type(shapes_in_group: List[ShapeInfo]) -> str: priority = [ lambda s: s.shape_type.startswith("flowChart") and s.shape_type != "flowChartAlternateProcess", lambda s: s.shape_type == "flowChartAlternateProcess", lambda s: s.shape_type in ("roundRect", "diamond", "hexagon", "parallelogram", "trapezoid"), lambda s: s.shape_type in ("rect", "rectangle", "ellipse", "oval"), lambda _: True, ] for pred in priority: matches = [s for s in shapes_in_group if pred(s)] if matches: return matches[0].shape_type return "rect" def _pick_fill(shapes_in_group: List[ShapeInfo]) -> Optional[str]: SKIP = {None, "", "-", "auto", "windowText", "000000"} for s in shapes_in_group: fill = (s.fill_color or "").strip() if fill and fill not in SKIP and len(fill) == 6: return fill.lower() return None def _build_nodes( shapes: List[ShapeInfo], groups: Dict[str, str], ) -> Dict[str, _MermaidNode]: by_group: Dict[str, List[ShapeInfo]] = {} for s in shapes: by_group.setdefault(groups[s.id], []).append(s) nodes: Dict[str, _MermaidNode] = {} for canonical_id, members in by_group.items(): label = _pick_label(members) if not label: continue stype = _pick_shape_type(members) nodes[canonical_id] = _MermaidNode( nid=_safe_id(canonical_id), label=_escape_label(label), wrap=_shape_wrap(stype), fill=_pick_fill(members), ) return nodes def _build_edges( connections: Iterable[Tuple[str, str, str]], groups: Dict[str, str], nodes: Dict[str, _MermaidNode], ) -> List[Tuple[str, str, str]]: seen: set = set() edges: List[Tuple[str, str, str]] = [] for from_id, to_id, label in connections: a = groups.get(from_id, from_id) b = groups.get(to_id, to_id) if a == b: continue if a not in nodes or b not in nodes: continue key = (a, b, label or "") if key in seen: continue seen.add(key) edges.append((a, b, label or "")) return edges def _emit_mermaid( nodes: Dict[str, _MermaidNode], edges: List[Tuple[str, str, str]], direction: str = "TD", ) -> str: lines: List[str] = [f"flowchart {direction}"] for node in nodes.values(): open_w, close_w = node.wrap lines.append(f' {node.nid}{open_w}"{node.label}"{close_w}') for a, b, label in edges: na = nodes[a].nid nb = nodes[b].nid if label: esc = _escape_label(label) lines.append(f' {na} -->|"{esc}"| {nb}') else: lines.append(f" {na} --> {nb}") fill_to_nodes: Dict[str, List[str]] = {} for node in nodes.values(): if node.fill: fill_to_nodes.setdefault(node.fill, []).append(node.nid) for fill, nids in sorted(fill_to_nodes.items()): cls = f"fill{fill}" lines.append(f" classDef {cls} fill:#{fill},stroke:#333,stroke-width:1px") lines.append(f" class {','.join(nids)} {cls}") return "\n".join(lines) # Docling's MarkdownTableSerializer renders the cell carrying our injected # mermaid block as a single-row Markdown table. Markdown table cells cannot # contain newlines, so the multi-line fence emitted by _emit_mermaid is # flattened into a one-liner where the `\n ` indent prefix becomes a run of # 5 whitespace chars. _MERMAID_TABLE_ROW_RE matches that flattened row and # _MERMAID_LINE_SPLIT_RE re-splits the body on the 4+ whitespace boundary so # we can restore the original lines. _MERMAID_TABLE_ROW_RE = re.compile( r"^\|\s*```mermaid\b(?P.*?)```\s*\|\s*$", re.MULTILINE, ) _MERMAID_LINE_SPLIT_RE = re.compile(r"\s{4,}") _MERMAID_TABLE_SEPARATOR_RE = re.compile(r"^\|[\s\-]+\|\s*$") def unwrap_mermaid_tables(markdown: str) -> str: """Restore proper multi-line ```mermaid``` fences from flattened table rows. Each fence sits in its own 1-cell, 1-row table emitted by Docling's MarkdownTableSerializer. We rewrite the header row as a real fenced block and drop the now-orphaned table-separator row that follows it. """ if "```mermaid" not in markdown: return markdown def _expand(match: re.Match) -> str: body = match.group("body").strip() if not body: return match.group(0) parts = [p.strip() for p in _MERMAID_LINE_SPLIT_RE.split(body) if p.strip()] if not parts: return match.group(0) # parts[0] is "flowchart " (no indent); the rest are indented stmts. lines = [parts[0]] + [f" {p}" for p in parts[1:]] return "```mermaid\n" + "\n".join(lines) + "\n```" expanded = _MERMAID_TABLE_ROW_RE.sub(_expand, markdown) out_lines: List[str] = [] prev_closed_fence = False for line in expanded.splitlines(): if prev_closed_fence and _MERMAID_TABLE_SEPARATOR_RE.match(line): prev_closed_fence = False continue prev_closed_fence = line.strip() == "```" out_lines.append(line) return "\n".join(out_lines) # --------------------------------------------------------------------------- # Worksheet grid / cell-based edge labels # --------------------------------------------------------------------------- _DEFAULT_COL_WIDTH_CHARS = 8.43 _DEFAULT_ROW_HEIGHT_PT = 15.0 _CELL_LABEL_MAX_DIST_EMU = 320_000 _CELL_LABEL_MAX_ENDPOINT_EMU = 700_000 _CELL_LABEL_CLUSTER_RADIUS_EMU = 1_200_000 _CELL_LABEL_MAX_LEN = 30 _CELL_LABEL_SKIP_NUMERIC = re.compile(r"^[\d０-９\.\s]+$") def _col_idx(letters: str) -> int: n = 0 for c in letters: n = n * 26 + (ord(c) - ord("A") + 1) return n def _col_letters(col: int) -> str: s = "" while col > 0: col, r = divmod(col - 1, 26) s = chr(ord("A") + r) + s return s def _read_sheet_cols_rows( zf: zipfile.ZipFile, sheet_xml: str, ) -> Tuple[Dict[int, float], Dict[int, float]]: with zf.open(sheet_xml) as f: root = ET.fromstring(f.read()) widths: Dict[int, float] = {} cols = root.find(f"{{{NS_MAIN}}}cols") if cols is not None: for col in cols.findall(f"{{{NS_MAIN}}}col"): w = col.get("width") if not w: continue mn, mx = int(col.get("min", "1")), int(col.get("max", "1")) for i in range(mn, mx + 1): widths[i] = float(w) heights: Dict[int, float] = {} sd = root.find(f"{{{NS_MAIN}}}sheetData") if sd is not None: for row in sd.findall(f"{{{NS_MAIN}}}row"): h = row.get("ht") if h: heights[int(row.get("r"))] = float(h) return widths, heights def _read_anchor_samples( zf: zipfile.ZipFile, drawing_xml: str, ) -> List[Tuple[int, int, int, int, int, int]]: try: with zf.open(drawing_xml) as f: root = ET.fromstring(f.read()) except KeyError: return [] out: List[Tuple[int, int, int, int, int, int]] = [] for anchor in root.findall("xdr:twoCellAnchor", _NS): frm = anchor.find("xdr:from", _NS) sp = anchor.find(".//xdr:spPr", _NS) if frm is None or sp is None: continue xfrm = sp.find("a:xfrm", _NS) if xfrm is None: continue off = xfrm.find("a:off", _NS) if off is None: continue out.append(( int(frm.findtext("xdr:col", "0", _NS)), int(frm.findtext("xdr:colOff", "0", _NS)), int(frm.findtext("xdr:row", "0", _NS)), int(frm.findtext("xdr:rowOff", "0", _NS)), int(off.get("x", "0")), int(off.get("y", "0")), )) return out def _build_grid_positions( col_widths: Dict[int, float], row_heights: Dict[int, float], anchors: List[Tuple[int, int, int, int, int, int]], max_col: int, max_row: int, ) -> Tuple[List[float], List[float]]: cum_chars = [0.0] * (max_col + 2) for i in range(1, max_col + 2): cum_chars[i] = cum_chars[i - 1] + col_widths.get(i, _DEFAULT_COL_WIDTH_CHARS) cum_pts = [0.0] * (max_row + 2) for i in range(1, max_row + 2): cum_pts[i] = cum_pts[i - 1] + row_heights.get(i, _DEFAULT_ROW_HEIGHT_PT) def _calibrate(targets: List[Tuple[float, int]], fallback: float) -> float: num = sum(u * e for u, e in targets) den = sum(u * u for u, e in targets) return num / den if den else fallback col_targets = [(cum_chars[col0], x - colOff) for col0, colOff, _, _, x, _ in anchors] row_targets = [(cum_pts[row0], y - rowOff) for _, _, row0, rowOff, _, y in anchors] emu_per_char = _calibrate(col_targets, 88929.0) emu_per_pt = _calibrate(row_targets, 12700.0) col_left_emu = [c * emu_per_char for c in cum_chars] row_top_emu = [r * emu_per_pt for r in cum_pts] return col_left_emu, row_top_emu def _compute_grid( xlsx_path: str, sheet: SheetDrawing, ) -> Tuple[List[float], List[float]]: try: zf = zipfile.ZipFile(xlsx_path) except (zipfile.BadZipFile, FileNotFoundError): return [], [] with zf: sheet_xml = sheet.sheet_xml_path.lstrip("/") if sheet.sheet_xml_path else None if not sheet_xml or sheet_xml not in zf.namelist(): return [], [] col_widths, row_heights = _read_sheet_cols_rows(zf, sheet_xml) drawing_xml = sheet.drawing_file.lstrip("/") if sheet.drawing_file else None anchors = _read_anchor_samples(zf, drawing_xml) if drawing_xml else [] max_col = max((a[0] for a in anchors), default=0) + 5 max_col = max(max_col, max(col_widths.keys(), default=0), 80) max_row = max((a[2] for a in anchors), default=0) + 5 max_row = max(max_row, max(row_heights.keys(), default=0), 200) return _build_grid_positions(col_widths, row_heights, anchors, max_col, max_row) def _coord_to_center( coord: str, col_left: List[float], row_top: List[float], ) -> Tuple[float, float]: m = re.match(r"^([A-Z]+)(\d+)$", coord) if not m: return 0.0, 0.0 col = _col_idx(m.group(1)) row = int(m.group(2)) if col >= len(col_left) - 1 or row >= len(row_top) - 1: return 0.0, 0.0 cx = (col_left[col - 1] + col_left[col]) / 2 cy = (row_top[row - 1] + row_top[row]) / 2 return cx, cy def _emu_to_cell( x_emu: float, y_emu: float, col_left: List[float], row_top: List[float], ) -> Tuple[int, int]: if not col_left or not row_top: return 1, 1 col = max(1, min(len(col_left) - 1, bisect.bisect_right(col_left, x_emu))) row = max(1, min(len(row_top) - 1, bisect.bisect_right(row_top, y_emu))) return col, row def _emu_bbox_to_range( bbox_emu: Tuple[float, float, float, float], col_left: List[float], row_top: List[float], ) -> str: x0, y0, x1, y1 = bbox_emu c0, r0 = _emu_to_cell(x0, y0, col_left, row_top) c1, r1 = _emu_to_cell(max(x0, x1 - 1), max(y0, y1 - 1), col_left, row_top) return f"{_col_letters(c0)}{r0}:{_col_letters(c1)}{r1}" def _dist_point_to_segment( px: float, py: float, ax: float, ay: float, bx: float, by: float, ) -> float: dx, dy = bx - ax, by - ay if dx == 0 and dy == 0: return ((px - ax) ** 2 + (py - ay) ** 2) ** 0.5 t = max(0.0, min(1.0, ((px - ax) * dx + (py - ay) * dy) / (dx * dx + dy * dy))) qx, qy = ax + t * dx, ay + t * dy return ((px - qx) ** 2 + (py - qy) ** 2) ** 0.5 def _score_cell_to_connector( px: float, py: float, ax: float, ay: float, bx: float, by: float, ) -> Optional[float]: d_start = ((px - ax) ** 2 + (py - ay) ** 2) ** 0.5 d_end = ((px - bx) ** 2 + (py - by) ** 2) ** 0.5 if min(d_start, d_end) > _CELL_LABEL_MAX_ENDPOINT_EMU: return None return _dist_point_to_segment(px, py, ax, ay, bx, by) def _extract_cell_edge_labels( xlsx_path: str, sheet: SheetDrawing, col_left: List[float], row_top: List[float], ) -> Tuple[Dict[int, str], List[str]]: """Return ({connector_id: label_text}, [consumed_cell_coords]) mined from worksheet cells.""" if not sheet.connectors or not col_left or not row_top: return {}, [] try: from openpyxl import load_workbook except ImportError: logger.warning("openpyxl not installed; skipping cell-based edge labels") return {}, [] if not sheet.shapes and not sheet.connectors: return {}, [] pts: List[Tuple[int, int]] = [] for s in sheet.shapes: pts.append((s.position[0], s.position[1])) pts.append((s.position[0] + s.size[0], s.position[1] + s.size[1])) for c in sheet.connectors: pts.append(c.start_pos) pts.append(c.end_pos) min_x = min(p[0] for p in pts) - 300_000 max_x = max(p[0] for p in pts) + 300_000 min_y = min(p[1] for p in pts) - 300_000 max_y = max(p[1] for p in pts) + 300_000 wb = load_workbook(xlsx_path, data_only=True) if sheet.sheet_name not in wb.sheetnames: return {}, [] ws = wb[sheet.sheet_name] candidates: List[Tuple[str, float, float, str]] = [] for row in ws.iter_rows(): for cell in row: v = cell.value if v is None: continue txt = str(v).strip() if not txt or len(txt) > _CELL_LABEL_MAX_LEN or "\n" in txt: continue if _CELL_LABEL_SKIP_NUMERIC.match(txt): continue cx, cy = _coord_to_center(cell.coordinate, col_left, row_top) if not (min_x <= cx <= max_x and min_y <= cy <= max_y): continue inside_shape = False for s in sheet.shapes: sx0, sy0 = s.position sx1, sy1 = sx0 + s.size[0], sy0 + s.size[1] if sx0 <= cx <= sx1 and sy0 <= cy <= sy1: inside_shape = True break if inside_shape: continue candidates.append((cell.coordinate, cx, cy, txt)) per_conn: Dict[int, List[Tuple[float, float, float, str]]] = {} assigned: List[Tuple[float, float, int]] = [] assigned_coords: set = set() for coord, cx, cy, txt in candidates: best_d, best_id = _CELL_LABEL_MAX_DIST_EMU, None for conn in sheet.connectors: d = _score_cell_to_connector( cx, cy, conn.start_pos[0], conn.start_pos[1], conn.end_pos[0], conn.end_pos[1], ) if d is not None and d < best_d: best_d, best_id = d, conn.id if best_id is not None: per_conn.setdefault(best_id, []).append((best_d, cy, cx, txt)) assigned.append((cx, cy, best_id)) assigned_coords.add(coord) EDGE_TOUCH_TOL = 50_000 EDGE_ZONE_OUTWARD = 2_000_000 EDGE_ZONE_PAD = 200_000 conn_by_id = {c.id: c for c in sheet.connectors} for conn in sheet.connectors: if conn.id in per_conn: continue sp = conn.start_pos best_match: Optional[Tuple[float, str, ShapeInfo]] = None for s in sheet.shapes: sx0, sy0 = s.position sx1, sy1 = sx0 + s.size[0], sy0 + s.size[1] in_y = sy0 - EDGE_TOUCH_TOL <= sp[1] <= sy1 + EDGE_TOUCH_TOL in_x = sx0 - EDGE_TOUCH_TOL <= sp[0] <= sx1 + EDGE_TOUCH_TOL edge_candidates = [] if in_y and abs(sp[0] - sx0) <= EDGE_TOUCH_TOL: edge_candidates.append((abs(sp[0] - sx0), "left")) if in_y and abs(sp[0] - sx1) <= EDGE_TOUCH_TOL: edge_candidates.append((abs(sp[0] - sx1), "right")) if in_x and abs(sp[1] - sy0) <= EDGE_TOUCH_TOL: edge_candidates.append((abs(sp[1] - sy0), "top")) if in_x and abs(sp[1] - sy1) <= EDGE_TOUCH_TOL: edge_candidates.append((abs(sp[1] - sy1), "bottom")) if edge_candidates: edge_candidates.sort() if best_match is None or edge_candidates[0][0] < best_match[0]: best_match = (edge_candidates[0][0], edge_candidates[0][1], s) if best_match is None: continue _, exit_edge, source = best_match sx0, sy0 = source.position sx1, sy1 = sx0 + source.size[0], sy0 + source.size[1] if exit_edge == "left": zone = (sx0 - EDGE_ZONE_OUTWARD, sy0 - EDGE_ZONE_PAD, sx0, sy1 + EDGE_ZONE_PAD) elif exit_edge == "right": zone = (sx1, sy0 - EDGE_ZONE_PAD, sx1 + EDGE_ZONE_OUTWARD, sy1 + EDGE_ZONE_PAD) elif exit_edge == "top": zone = (sx0 - EDGE_ZONE_PAD, sy0 - EDGE_ZONE_OUTWARD, sx1 + EDGE_ZONE_PAD, sy0) else: zone = (sx0 - EDGE_ZONE_PAD, sy1, sx1 + EDGE_ZONE_PAD, sy1 + EDGE_ZONE_OUTWARD) for coord, cx, cy, txt in candidates: if coord in assigned_coords: continue if not (zone[0] <= cx <= zone[2] and zone[1] <= cy <= zone[3]): continue per_conn.setdefault(conn.id, []).append((0.0, cy, cx, txt)) assigned.append((cx, cy, conn.id)) assigned_coords.add(coord) for coord, cx, cy, txt in candidates: if coord in assigned_coords: continue neighbors: Dict[int, float] = {} for ax, ay, conn_id in assigned: d = ((cx - ax) ** 2 + (cy - ay) ** 2) ** 0.5 if d > _CELL_LABEL_CLUSTER_RADIUS_EMU: continue if conn_id not in neighbors or d < neighbors[conn_id]: neighbors[conn_id] = d if not neighbors: continue if len(neighbors) == 1: conn_id, d = next(iter(neighbors.items())) per_conn.setdefault(conn_id, []).append((d, cy, cx, txt)) assigned_coords.add(coord) continue scored: List[Tuple[float, int]] = [] for conn_id in neighbors: conn = conn_by_id.get(conn_id) if conn is None: continue d_line = _dist_point_to_segment( cx, cy, conn.start_pos[0], conn.start_pos[1], conn.end_pos[0], conn.end_pos[1], ) scored.append((d_line, conn_id)) scored.sort() if len(scored) >= 2 and scored[0][0] > 0.9 * scored[1][0]: continue if scored: d_line, conn_id = scored[0] per_conn.setdefault(conn_id, []).append((d_line, cy, cx, txt)) assigned_coords.add(coord) out: Dict[int, str] = {} for cid, cells in per_conn.items(): cells.sort(key=lambda t: (t[1], t[2])) seen: List[str] = [] for _, _, _, t in cells: if not seen or seen[-1] != t: seen.append(t) out[cid] = "\n".join(seen) return out, sorted(assigned_coords) # --------------------------------------------------------------------------- # Component detection + per-sheet driver # --------------------------------------------------------------------------- def _detect_components( shapes: List[ShapeInfo], raw_connections: List[Tuple[str, str, str]], groups: Dict[str, str], ) -> List[List[ShapeInfo]]: parent: Dict[str, str] = {} def find(x: str) -> str: while parent[x] != x: parent[x] = parent[parent[x]] x = parent[x] return x def union(a: str, b: str) -> None: parent.setdefault(a, a) parent.setdefault(b, b) ra, rb = find(a), find(b) if ra != rb: parent[ra] = rb for sid, canonical in groups.items(): parent.setdefault(sid, sid) parent.setdefault(canonical, canonical) union(sid, canonical) for from_id, to_id, _ in raw_connections: if from_id in parent and to_id in parent: union(from_id, to_id) buckets: Dict[str, List[ShapeInfo]] = {} for s in shapes: if s.id not in parent: continue buckets.setdefault(find(s.id), []).append(s) return list(buckets.values()) def _component_bbox_emu( comp_shapes: List[ShapeInfo], comp_connectors: List[ConnectorInfo], ) -> Tuple[float, float, float, float]: xs: List[float] = [] ys: List[float] = [] for s in comp_shapes: xs.extend([s.position[0], s.position[0] + s.size[0]]) ys.extend([s.position[1], s.position[1] + s.size[1]]) for c in comp_connectors: xs.extend([c.start_pos[0], c.end_pos[0]]) ys.extend([c.start_pos[1], c.end_pos[1]]) if not xs or not ys: return 0.0, 0.0, 0.0, 0.0 return min(xs), min(ys), max(xs), max(ys) def _sheet_to_graphs( sheet: SheetDrawing, parser: ConnectorParser, direction: str, xlsx_path: str, ) -> Tuple[List[GraphResult], List[str]]: """Return (graphs, consumed_cell_coords) for one sheet.""" if not sheet.shapes: return [], [] for s in sheet.shapes: s.text = _clean_shape_text(s.text) col_left, row_top = _compute_grid(xlsx_path, sheet) cell_labels, consumed_cells = _extract_cell_edge_labels(xlsx_path, sheet, col_left, row_top) if cell_labels: for conn in sheet.connectors: new = cell_labels.get(conn.id) if new and not (conn.text and conn.text.strip()): conn.text = new logger.info( "Sheet %r: %d edge label(s) mined from %d cell(s)", sheet.sheet_name, len(cell_labels), len(consumed_cells), ) groups = _build_node_groups(sheet.shapes) raw_connections = parser.find_connected_shapes(sheet.shapes, sheet.connectors) components = _detect_components(sheet.shapes, raw_connections, groups) shape_to_comp_idx: Dict[str, int] = {} for idx, comp in enumerate(components): for s in comp: shape_to_comp_idx[s.id] = idx comp_conns: Dict[int, List[ConnectorInfo]] = {i: [] for i in range(len(components))} for conn in sheet.connectors: start = parser._find_closest_shape( conn.start_pos, sheet.shapes, parser.DEFAULT_CONNECTOR_MAX_DISTANCE, ) end = parser._find_closest_shape( conn.end_pos, sheet.shapes, parser.DEFAULT_CONNECTOR_MAX_DISTANCE, ) for ep in (start, end): if ep is None: continue idx = shape_to_comp_idx.get(ep.id) if idx is not None: comp_conns[idx].append(conn) break results: List[GraphResult] = [] for idx, comp in enumerate(components): comp_nodes = _build_nodes(comp, groups) if len(comp_nodes) < 2: continue comp_shape_ids = {s.id for s in comp} comp_raw_conns = [ (a, b, lab) for (a, b, lab) in raw_connections if a in comp_shape_ids and b in comp_shape_ids ] comp_edges = _build_edges(comp_raw_conns, groups, comp_nodes) mermaid = _emit_mermaid(comp_nodes, comp_edges, direction=direction) bbox = _component_bbox_emu(comp, comp_conns.get(idx, [])) if col_left and row_top: location = _emu_bbox_to_range(bbox, col_left, row_top) else: location = "" results.append(GraphResult( sheet=sheet.sheet_name, location=location, mermaid_diagram=mermaid, )) return results, consumed_cells # --------------------------------------------------------------------------- # Public entry point # --------------------------------------------------------------------------- def extract_mermaid_graphs( xlsx_path: str, direction: str = "TD", ) -> Dict[str, Tuple[List[GraphResult], List[str]]]: """Detect connected flowchart components per sheet and emit mermaid. Args: xlsx_path: Path to a .xlsx file (callers must convert .xls/.xlsm first). direction: Mermaid layout direction ("TD", "LR", "TB", "RL", "BT"). Returns: Dict keyed by sheet name. Each value is a tuple of: - List[GraphResult] — one mermaid block per detected component. - List[str] — worksheet cell coords whose text was mined as edge labels. Callers that rewrite the workbook should clear those cells to prevent the same text from re-appearing in markdown alongside the embedded mermaid block. Sheets with no shapes are still present in the dict with empty lists. """ parser = ConnectorParser() per_sheet = parser.parse_by_sheet(xlsx_path) out: Dict[str, Tuple[List[GraphResult], List[str]]] = {} for sd in per_sheet: graphs, consumed = _sheet_to_graphs(sd, parser, direction, xlsx_path) out[sd.sheet_name] = (graphs, consumed) if graphs: logger.info( "Sheet %r: %d shape(s), %d connector(s) -> %d graph(s); " "%d cell(s) consumed as edge labels", sd.sheet_name, len(sd.shapes), len(sd.connectors), len(graphs), len(consumed), ) return out