返回 ppt-master
diagram_read.py
1 """Read SmartArt content and structure from DrawingML diagram parts.
2
3 The reader exposes source facts only. It does not edit DiagramML or promise
4 native SmartArt regeneration; generated decks continue to redraw the extracted
5 content through the SVG-to-DrawingML shape pipeline.
6 """
7
8 from __future__ import annotations
9
10 import zipfile
11 from typing import Any
12 from xml.etree import ElementTree as ET
13
14 from .ooxml import (
15 MC_NS,
16 NS,
17 _container_geometry,
18 _normalize_part,
19 _paragraph_texts,
20 _qn,
21 _read_xml,
22 _rels_name_for_part,
23 _shape_identity,
24 _slide_relationships,
25 )
26
27
28 DIAGRAM_NS = "http://schemas.openxmlformats.org/drawingml/2006/diagram"
29 DIAGRAM_DRAWING_NS = "http://schemas.microsoft.com/office/drawing/2008/diagram"
30 DIAGRAM_URI = "http://schemas.openxmlformats.org/drawingml/2006/diagram"
31
32 _DIAGRAM_REL_SUFFIXES = {
33 "colors": "diagramColors",
34 "data": "diagramData",
35 "layout": "diagramLayout",
36 "quick_style": "diagramQuickStyle",
37 }
38 _REL_ATTRS = {
39 "colors": "cs",
40 "data": "dm",
41 "layout": "lo",
42 "quick_style": "qs",
43 }
44 _CONTENT_POINT_TYPES = {"asst", "node"}
45
46 DIAGRAM_NS_MAP = {
47 **NS,
48 "dgm": DIAGRAM_NS,
49 "dsp": DIAGRAM_DRAWING_NS,
50 "mc": MC_NS,
51 }
52
53
54 def _relationship_matches(rel_type: str, suffix: str) -> bool:
55 return rel_type.rsplit("/", 1)[-1] == suffix
56
57
58 def _read_optional_xml(
59 zf: zipfile.ZipFile,
60 part_name: str | None,
61 ) -> tuple[ET.Element | None, str | None]:
62 if not part_name:
63 return None, "missing-part-reference"
64 try:
65 return _read_xml(zf, part_name), None
66 except RuntimeError:
67 return None, "missing-part"
68 except ET.ParseError:
69 return None, "invalid-xml"
70
71
72 def _diagram_containers(slide_root: ET.Element) -> list[ET.Element]:
73 containers: list[ET.Element] = []
74 for frame in slide_root.findall(".//p:graphicFrame", DIAGRAM_NS_MAP):
75 graphic_data = frame.find("a:graphic/a:graphicData", DIAGRAM_NS_MAP)
76 if graphic_data is not None and graphic_data.attrib.get("uri") == DIAGRAM_URI:
77 containers.append(frame)
78 return containers
79
80
81 def _fallback_preview_shape_ids(slide_root: ET.Element) -> set[str]:
82 shape_ids: set[str] = set()
83 for alternate in slide_root.findall(".//mc:AlternateContent", DIAGRAM_NS_MAP):
84 frame = alternate.find("mc:Choice//p:graphicFrame", DIAGRAM_NS_MAP)
85 if frame is None:
86 continue
87 graphic_data = frame.find("a:graphic/a:graphicData", DIAGRAM_NS_MAP)
88 if graphic_data is None or graphic_data.attrib.get("uri") != DIAGRAM_URI:
89 continue
90 if alternate.find("mc:Fallback//p:pic", DIAGRAM_NS_MAP) is None:
91 continue
92 shape_id, _shape_name = _shape_identity(frame, len(shape_ids) + 1)
93 shape_ids.add(shape_id)
94 return shape_ids
95
96
97 def _diagram_parts(
98 rel_ids: ET.Element | None,
99 relationships: dict[str, dict[str, str]],
100 slide_part: str,
101 ) -> tuple[dict[str, str], str | None]:
102 if rel_ids is None:
103 return {}, "missing-rel-ids"
104
105 parts: dict[str, str] = {}
106 data_error: str | None = None
107 for key, attr_name in _REL_ATTRS.items():
108 rel_id = rel_ids.attrib.get(_qn(NS["r"], attr_name), "")
109 if not rel_id:
110 if key == "data":
111 data_error = "missing-data-relationship"
112 continue
113 relationship = relationships.get(rel_id)
114 if relationship is None:
115 if key == "data":
116 data_error = "missing-data-relationship"
117 continue
118 if not _relationship_matches(
119 relationship.get("type", ""),
120 _DIAGRAM_REL_SUFFIXES[key],
121 ):
122 if key == "data":
123 data_error = "invalid-data-relationship"
124 continue
125 parts[key] = _normalize_part(relationship["target"], slide_part)
126 return parts, data_error
127
128
129 def _persisted_drawing_part(
130 zf: zipfile.ZipFile,
131 data_part: str | None,
132 data_root: ET.Element | None,
133 slide_part: str,
134 slide_relationships: dict[str, dict[str, str]],
135 ) -> tuple[str | None, list[str]]:
136 if not data_part or data_root is None:
137 return None, []
138 data_model_ext = data_root.find(".//dsp:dataModelExt", DIAGRAM_NS_MAP)
139 rel_id = data_model_ext.attrib.get("relId", "") if data_model_ext is not None else ""
140 if not rel_id:
141 return None, []
142
143 warnings: list[str] = []
144 relationship = slide_relationships.get(rel_id)
145 relationship_owner = slide_part
146 if relationship is not None and not _relationship_matches(
147 relationship.get("type", ""),
148 "diagramDrawing",
149 ):
150 relationship = None
151 if relationship is None:
152 try:
153 data_relationships = _slide_relationships(zf, _rels_name_for_part(data_part))
154 except ET.ParseError:
155 data_relationships = {}
156 warnings.append("invalid-persisted-drawing-relationships")
157 relationship = data_relationships.get(rel_id)
158 relationship_owner = data_part
159 if relationship is not None and not _relationship_matches(
160 relationship.get("type", ""),
161 "diagramDrawing",
162 ):
163 relationship = None
164 if relationship is None:
165 warnings.append("unresolved-persisted-drawing-relationship")
166 return None, warnings
167
168 part_name = _normalize_part(relationship["target"], relationship_owner)
169 if part_name not in zf.namelist():
170 warnings.append("missing-persisted-drawing-part")
171 return None, warnings
172 return part_name, warnings
173
174
175 def _layout_info(
176 layout_root: ET.Element | None,
177 data_root: ET.Element | None,
178 ) -> dict[str, Any]:
179 unique_id = layout_root.attrib.get("uniqueId") if layout_root is not None else None
180 if not unique_id and data_root is not None:
181 document_properties = data_root.find(
182 ".//dgm:pt[@type='doc']/dgm:prSet",
183 DIAGRAM_NS_MAP,
184 )
185 if document_properties is not None:
186 unique_id = document_properties.attrib.get("loTypeId")
187 title = layout_root.find("dgm:title", DIAGRAM_NS_MAP) if layout_root is not None else None
188 name = title.attrib.get("val") if title is not None else None
189 if not name and unique_id:
190 name = unique_id.rstrip("/").rsplit("/", 1)[-1]
191 categories = (
192 [
193 category.attrib["type"]
194 for category in layout_root.findall("dgm:catLst/dgm:cat", DIAGRAM_NS_MAP)
195 if category.attrib.get("type")
196 ]
197 if layout_root is not None
198 else []
199 )
200 return {
201 "name": name,
202 "unique_id": unique_id,
203 "categories": categories,
204 }
205
206
207 def _integer_or_none(value: str | None) -> int | None:
208 if value is None:
209 return None
210 try:
211 return int(value)
212 except ValueError:
213 return None
214
215
216 def _point_text(point: ET.Element) -> str:
217 text_body = point.find("dgm:t", DIAGRAM_NS_MAP)
218 if text_body is None:
219 return ""
220 return "\n".join(_paragraph_texts(text_body)).strip()
221
222
223 def _connections(data_root: ET.Element) -> list[dict[str, Any]]:
224 connections: list[dict[str, Any]] = []
225 for connection in data_root.findall(
226 ".//dgm:cxnLst/dgm:cxn",
227 DIAGRAM_NS_MAP,
228 ):
229 source_id = connection.attrib.get("srcId", "")
230 destination_id = connection.attrib.get("destId", "")
231 if not source_id or not destination_id:
232 continue
233 connections.append(
234 {
235 "type": connection.attrib.get("type") or "parOf",
236 "source_id": source_id,
237 "destination_id": destination_id,
238 "source_order": _integer_or_none(connection.attrib.get("srcOrd")),
239 "destination_order": _integer_or_none(connection.attrib.get("destOrd")),
240 }
241 )
242 return connections
243
244
245 def _nearest_content_parent(
246 node_id: str,
247 parent_by_id: dict[str, str],
248 content_ids: set[str],
249 ) -> str | None:
250 current = parent_by_id.get(node_id)
251 visited = {node_id}
252 while current:
253 if current in content_ids:
254 return current
255 if current in visited:
256 return None
257 visited.add(current)
258 current = parent_by_id.get(current)
259 return None
260
261
262 def _break_parent_cycles(
263 parent_by_node: dict[str, str | None],
264 source_order: dict[str, int],
265 ) -> list[str]:
266 """Break cycles in the Markdown tree projection while retaining raw connections."""
267 warnings: list[str] = []
268 for start_id in sorted(parent_by_node, key=lambda node_id: source_order[node_id]):
269 path: list[str] = []
270 path_index: dict[str, int] = {}
271 current_id: str | None = start_id
272 while current_id is not None and current_id in parent_by_node:
273 if current_id in path_index:
274 cycle = path[path_index[current_id] :]
275 root_id = min(cycle, key=lambda node_id: source_order[node_id])
276 parent_by_node[root_id] = None
277 warnings.append(f"parent-cycle-broken-at:{root_id}")
278 break
279 path_index[current_id] = len(path)
280 path.append(current_id)
281 current_id = parent_by_node.get(current_id)
282 return warnings
283
284
285 def _ordered_nodes(
286 data_root: ET.Element,
287 ) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[str]]:
288 raw_points: list[dict[str, Any]] = []
289 for source_index, point in enumerate(
290 data_root.findall(".//dgm:ptLst/dgm:pt", DIAGRAM_NS_MAP),
291 ):
292 model_id = point.attrib.get("modelId", "")
293 point_type = point.attrib.get("type") or "node"
294 text = _point_text(point)
295 if not model_id or point_type not in _CONTENT_POINT_TYPES:
296 continue
297 raw_points.append(
298 {
299 "id": model_id,
300 "type": point_type,
301 "text": text,
302 "source_index": source_index,
303 }
304 )
305
306 all_connections = _connections(data_root)
307 parent_by_id: dict[str, str] = {}
308 order_by_id: dict[str, int | None] = {}
309 for connection in all_connections:
310 if connection["type"] != "parOf":
311 continue
312 destination_id = connection["destination_id"]
313 parent_by_id.setdefault(destination_id, connection["source_id"])
314 order_by_id.setdefault(destination_id, connection["source_order"])
315
316 content_ids = {point["id"] for point in raw_points}
317 source_order = {point["id"]: int(point["source_index"]) for point in raw_points}
318 parent_by_node = {
319 point["id"]: _nearest_content_parent(
320 point["id"],
321 parent_by_id,
322 content_ids,
323 )
324 for point in raw_points
325 }
326 structure_warnings = _break_parent_cycles(parent_by_node, source_order)
327 nodes_by_id: dict[str, dict[str, Any]] = {}
328 for point in raw_points:
329 node = {
330 "id": point["id"],
331 "type": point["type"],
332 "text": point["text"],
333 "parent_id": parent_by_node[point["id"]],
334 "order": order_by_id.get(point["id"]),
335 "depth": 0,
336 "_source_index": point["source_index"],
337 }
338 nodes_by_id[point["id"]] = node
339
340 children: dict[str | None, list[dict[str, Any]]] = {}
341 for node in nodes_by_id.values():
342 children.setdefault(node["parent_id"], []).append(node)
343
344 def sort_key(node: dict[str, Any]) -> tuple[int, int]:
345 order = node["order"]
346 return (
347 order if isinstance(order, int) else 1_000_000,
348 int(node["_source_index"]),
349 )
350
351 for siblings in children.values():
352 siblings.sort(key=sort_key)
353
354 ordered: list[dict[str, Any]] = []
355 visited: set[str] = set()
356
357 def visit(node: dict[str, Any], depth: int) -> None:
358 node_id = str(node["id"])
359 if node_id in visited:
360 return
361 visited.add(node_id)
362 node["depth"] = depth
363 ordered.append(node)
364 for child in children.get(node_id, []):
365 visit(child, depth + 1)
366
367 for root in children.get(None, []):
368 visit(root, 0)
369 for node in sorted(nodes_by_id.values(), key=sort_key):
370 if node["id"] not in visited:
371 visit(node, 0)
372
373 for node in ordered:
374 node.pop("_source_index", None)
375
376 visible_connections = [
377 connection
378 for connection in all_connections
379 if connection["type"] == "parOf"
380 and connection["source_id"] in content_ids
381 and connection["destination_id"] in content_ids
382 ]
383 return ordered, visible_connections, structure_warnings
384
385
386 def _read_diagram_container(
387 zf: zipfile.ZipFile,
388 container: ET.Element,
389 *,
390 slide_part: str,
391 slide_index: int,
392 order: int,
393 relationships: dict[str, dict[str, str]],
394 relationship_error: str | None,
395 fallback_shape_ids: set[str],
396 ) -> dict[str, Any]:
397 shape_id, shape_name = _shape_identity(container, order)
398 graphic_data = container.find("a:graphic/a:graphicData", DIAGRAM_NS_MAP)
399 rel_ids = (
400 graphic_data.find("dgm:relIds", DIAGRAM_NS_MAP)
401 if graphic_data is not None
402 else None
403 )
404 parts, diagram_relation_error = _diagram_parts(
405 rel_ids,
406 relationships,
407 slide_part,
408 )
409 data_root, data_error = _read_optional_xml(zf, parts.get("data"))
410 layout_root, layout_error = _read_optional_xml(zf, parts.get("layout"))
411 metadata_warnings = [f"layout:{layout_error}"] if layout_error else []
412 nodes: list[dict[str, Any]] = []
413 connections: list[dict[str, Any]] = []
414 structure_warnings: list[str] = []
415 if data_root is not None:
416 nodes, connections, structure_warnings = _ordered_nodes(data_root)
417
418 status = relationship_error or diagram_relation_error or data_error
419 if status is None and structure_warnings:
420 status = "structure-cycle"
421 persisted_drawing, drawing_warnings = _persisted_drawing_part(
422 zf,
423 parts.get("data"),
424 data_root,
425 slide_part,
426 relationships,
427 )
428 text_items = [node["text"] for node in nodes if node["text"]]
429 return {
430 "diagram_id": f"s{slide_index:02d}_dgm{shape_id}",
431 "kind": "smartart",
432 "shape_id": shape_id,
433 "shape_name": shape_name,
434 "geometry": _container_geometry(container),
435 "layout": _layout_info(layout_root, data_root),
436 "root_ids": [node["id"] for node in nodes if node["parent_id"] is None],
437 "nodes": nodes,
438 "connections": connections,
439 "text_items": text_items,
440 "node_count": len(nodes),
441 "text_count": len(text_items),
442 "connection_count": len(connections),
443 "max_depth": max((int(node["depth"]) for node in nodes), default=0),
444 "text_extracted": data_root is not None,
445 "has_persisted_drawing": persisted_drawing is not None,
446 "has_fallback_preview": shape_id in fallback_shape_ids,
447 "status": status or "ok",
448 "warnings": metadata_warnings + structure_warnings + drawing_warnings,
449 }
450
451
452 def _failed_diagram(
453 container: ET.Element,
454 *,
455 slide_index: int,
456 order: int,
457 error: Exception,
458 fallback_shape_ids: set[str],
459 ) -> dict[str, Any]:
460 shape_id, shape_name = _shape_identity(container, order)
461 return {
462 "diagram_id": f"s{slide_index:02d}_dgm{shape_id}",
463 "kind": "smartart",
464 "shape_id": shape_id,
465 "shape_name": shape_name,
466 "geometry": _container_geometry(container),
467 "layout": {"name": None, "unique_id": None, "categories": []},
468 "root_ids": [],
469 "nodes": [],
470 "connections": [],
471 "text_items": [],
472 "node_count": 0,
473 "text_count": 0,
474 "connection_count": 0,
475 "max_depth": 0,
476 "text_extracted": False,
477 "has_persisted_drawing": False,
478 "has_fallback_preview": shape_id in fallback_shape_ids,
479 "status": "diagram-read-error",
480 "warnings": [f"{type(error).__name__}:{error}"],
481 }
482
483
484 def read_smartart_diagrams(
485 zf: zipfile.ZipFile,
486 slide_part: str,
487 slide_index: int,
488 ) -> list[dict[str, Any]]:
489 """Return SmartArt source facts for one slide part."""
490 slide_root = _read_xml(zf, slide_part)
491 relationship_error: str | None = None
492 try:
493 relationships = _slide_relationships(zf, _rels_name_for_part(slide_part))
494 except ET.ParseError:
495 relationships = {}
496 relationship_error = "invalid-slide-relationships"
497 fallback_shape_ids = _fallback_preview_shape_ids(slide_root)
498 diagrams: list[dict[str, Any]] = []
499
500 for order, container in enumerate(_diagram_containers(slide_root), start=1):
501 try:
502 diagram = _read_diagram_container(
503 zf,
504 container,
505 slide_part=slide_part,
506 slide_index=slide_index,
507 order=order,
508 relationships=relationships,
509 relationship_error=relationship_error,
510 fallback_shape_ids=fallback_shape_ids,
511 )
512 except (OSError, RuntimeError, zipfile.BadZipFile, ET.ParseError, KeyError, ValueError) as exc:
513 diagram = _failed_diagram(
514 container,
515 slide_index=slide_index,
516 order=order,
517 error=exc,
518 fallback_shape_ids=fallback_shape_ids,
519 )
520 diagrams.append(diagram)
521 return diagrams
522
523
524 def smartart_to_markdown(diagram: dict[str, Any]) -> str:
525 """Render one extracted SmartArt diagram as hierarchical Markdown."""
526 name = str(diagram.get("shape_name") or diagram.get("diagram_id") or "SmartArt")
527 layout = diagram.get("layout") or {}
528 layout_name = str(layout.get("name") or "") if isinstance(layout, dict) else ""
529 heading = f"### SmartArt: {name}"
530 if layout_name and layout_name.lower() not in name.lower():
531 heading += f" — {layout_name}"
532 lines = [heading, ""]
533 nodes = diagram.get("nodes") or []
534 text_nodes = [node for node in nodes if str(node.get("text") or "").strip()]
535 if text_nodes:
536 nodes_by_id = {
537 str(node.get("id")): node
538 for node in nodes
539 if node.get("id") is not None
540 }
541 rendered_depths: dict[str, int] = {}
542
543 def rendered_depth(node: dict[str, Any]) -> int:
544 node_id = str(node.get("id") or "")
545 if node_id in rendered_depths:
546 return rendered_depths[node_id]
547 parent = nodes_by_id.get(str(node.get("parent_id") or ""))
548 if parent is None:
549 depth = 0
550 else:
551 depth = rendered_depth(parent)
552 if str(parent.get("text") or "").strip():
553 depth += 1
554 rendered_depths[node_id] = depth
555 return depth
556
557 for node in text_nodes:
558 text = " / ".join(str(node.get("text") or "").splitlines()).strip()
559 indent = " " * rendered_depth(node)
560 lines.append(f"{indent}- {text}")
561 elif nodes:
562 lines.append(f"> [SmartArt structure has {len(nodes)} node(s), but no text]")
563 else:
564 status = str(diagram.get("status") or "content-unavailable")
565 if diagram.get("text_extracted") and status == "ok":
566 lines.append("> [SmartArt data is readable but has no semantic nodes]")
567 else:
568 lines.append(f"> [SmartArt content unavailable: {status}]")
569 return "\n".join(lines).rstrip()
570
570 lines PYTHON