| 1 | #!/usr/bin/env python3 |
| 2 | """ |
| 3 | PPT Master - SVG Authoring View |
| 4 | |
| 5 | Create a lightweight, non-destructive editable IR from PPTX-imported SVG |
| 6 | files. The source SVG remains the native-payload authority; the authoring copy |
| 7 | keeps visible SVG content, compact shape intent, and stable source references |
| 8 | while hiding bulky import-only payloads and duplicate hidden geometry carriers. |
| 9 | |
| 10 | Usage: |
| 11 | python3 scripts/svg_authoring_view.py <svg-file-or-directory> \ |
| 12 | -o <output-dir> --projection-kind <kind> |
| 13 | |
| 14 | Examples: |
| 15 | python3 scripts/svg_authoring_view.py analysis/source_svg_import/svg \ |
| 16 | -o analysis/authoring-svg --projection-kind layered |
| 17 | python3 scripts/svg_authoring_view.py imported/slide_06.svg \ |
| 18 | -o /tmp/slide-authoring-view --projection-kind generic |
| 19 | |
| 20 | Dependencies: |
| 21 | None (standard library only). |
| 22 | |
| 23 | The output directory is an authoring bundle: editable SVGs plus one |
| 24 | model-readable `authoring_summary.json` and one tool-only |
| 25 | `authoring_manifest.json` provenance sidecar. It is the template-creation |
| 26 | input, not a release SVG directory; final templates are materialized from this |
| 27 | IR. Directory runs prepare and stage the complete batch before publishing it, |
| 28 | so a failed page leaves the existing destination set unchanged. |
| 29 | """ |
| 30 | |
| 31 | from __future__ import annotations |
| 32 | |
| 33 | import argparse |
| 34 | import hashlib |
| 35 | import json |
| 36 | import os |
| 37 | import shutil |
| 38 | import sys |
| 39 | import tempfile |
| 40 | from collections import Counter |
| 41 | from dataclasses import dataclass, field |
| 42 | from pathlib import Path |
| 43 | from typing import Optional |
| 44 | from urllib.parse import urlsplit, urlunsplit |
| 45 | from xml.etree import ElementTree as ET |
| 46 | |
| 47 | from compact_svg_coordinates import compact_svg_tree |
| 48 | from console_encoding import configure_utf8_stdio |
| 49 | |
| 50 | configure_utf8_stdio() |
| 51 | |
| 52 | SVG_NS = "http://www.w3.org/2000/svg" |
| 53 | XLINK_NS = "http://www.w3.org/1999/xlink" |
| 54 | AUTHORING_MANIFEST_NAME = "authoring_manifest.json" |
| 55 | AUTHORING_SUMMARY_NAME = "authoring_summary.json" |
| 56 | AUTHORING_SCHEMA = "ppt-master.svg-authoring-ir.v1" |
| 57 | AUTHORING_SUMMARY_SCHEMA = "ppt-master.svg-authoring-summary.v1" |
| 58 | SOURCE_REF_ATTRIBUTE = "data-pptx-source-ref" |
| 59 | _DRAWABLE_TAGS = frozenset({ |
| 60 | "circle", |
| 61 | "ellipse", |
| 62 | "line", |
| 63 | "path", |
| 64 | "polygon", |
| 65 | "polyline", |
| 66 | "rect", |
| 67 | }) |
| 68 | |
| 69 | ET.register_namespace("", SVG_NS) |
| 70 | ET.register_namespace("xlink", XLINK_NS) |
| 71 | |
| 72 | # These fields identify the source OOXML object or guard its exact imported |
| 73 | # fallback. They belong in the complete import SVG, not its lightweight view. |
| 74 | IMPORT_SOURCE_ATTRIBUTES = { |
| 75 | "data-name", |
| 76 | "data-pptx-preview-sha256", |
| 77 | "data-pptx-shape-id", |
| 78 | "data-pptx-shape-name", |
| 79 | "data-pptx-shape-scope", |
| 80 | "data-pptx-shape-style", |
| 81 | } |
| 82 | |
| 83 | # Compact native-shape intent is intentionally not in the removal set: |
| 84 | # data-pptx-object, data-pptx-prst, and data-pptx-frame remain useful while |
| 85 | # reviewing the visible fallback. Structural markers also pass through |
| 86 | # unchanged; the IR records identity but never decides payload-restoration |
| 87 | # policy. |
| 88 | |
| 89 | |
| 90 | def _local_name(name: object) -> str: |
| 91 | return name.rsplit("}", 1)[-1] if isinstance(name, str) else "" |
| 92 | |
| 93 | |
| 94 | @dataclass |
| 95 | class SourceReference: |
| 96 | source_ref: str |
| 97 | source_path: tuple[int, ...] |
| 98 | initial_authoring_subtree_sha256: str | None = None |
| 99 | |
| 100 | def as_dict(self) -> dict[str, object]: |
| 101 | return { |
| 102 | "source_path": list(self.source_path), |
| 103 | "initial_authoring_subtree_sha256": self.initial_authoring_subtree_sha256, |
| 104 | } |
| 105 | |
| 106 | |
| 107 | def _semantic_text(value: str | None) -> str | None: |
| 108 | if value is None or not value.strip(): |
| 109 | return None |
| 110 | return value |
| 111 | |
| 112 | |
| 113 | def _stable_tag_name(tag: object) -> str: |
| 114 | if isinstance(tag, str): |
| 115 | return tag |
| 116 | if tag is ET.Comment: |
| 117 | return "#comment" |
| 118 | if tag is ET.ProcessingInstruction: |
| 119 | return "#processing-instruction" |
| 120 | raise ValueError(f"Unsupported XML node type in source object: {tag!r}") |
| 121 | |
| 122 | |
| 123 | def semantic_subtree_sha256( |
| 124 | element: ET.Element, |
| 125 | *, |
| 126 | ignored_attributes: frozenset[str] = frozenset(), |
| 127 | ) -> str: |
| 128 | """Hash parsed SVG semantics without attribute order or indentation noise.""" |
| 129 | digest = hashlib.sha256() |
| 130 | |
| 131 | def visit(item: ET.Element) -> None: |
| 132 | digest.update(_stable_tag_name(item.tag).encode("utf-8")) |
| 133 | for name, value in sorted(item.attrib.items()): |
| 134 | if name in ignored_attributes: |
| 135 | continue |
| 136 | digest.update(b"\0a") |
| 137 | digest.update(name.encode("utf-8")) |
| 138 | digest.update(b"\0") |
| 139 | digest.update(value.encode("utf-8")) |
| 140 | text = _semantic_text(item.text) |
| 141 | if text is not None: |
| 142 | digest.update(b"\0t") |
| 143 | digest.update(text.encode("utf-8")) |
| 144 | for child in item: |
| 145 | digest.update(b"\0c") |
| 146 | visit(child) |
| 147 | tail = _semantic_text(child.tail) |
| 148 | if tail is not None: |
| 149 | digest.update(b"\0l") |
| 150 | digest.update(tail.encode("utf-8")) |
| 151 | digest.update(b"\0e") |
| 152 | |
| 153 | visit(element) |
| 154 | return digest.hexdigest() |
| 155 | |
| 156 | |
| 157 | def _iter_element_paths( |
| 158 | root: ET.Element, |
| 159 | ) -> list[tuple[tuple[int, ...], ET.Element]]: |
| 160 | indexed: list[tuple[tuple[int, ...], ET.Element]] = [] |
| 161 | |
| 162 | def walk(element: ET.Element, path: tuple[int, ...]) -> None: |
| 163 | indexed.append((path, element)) |
| 164 | for index, child in enumerate(element): |
| 165 | walk(child, (*path, index)) |
| 166 | |
| 167 | walk(root, ()) |
| 168 | return indexed |
| 169 | |
| 170 | |
| 171 | def _source_reference(element: ET.Element) -> str | None: |
| 172 | if not element.get("id") or not element.get("data-pptx-object"): |
| 173 | return None |
| 174 | scope = element.get("data-pptx-shape-scope") |
| 175 | shape_id = element.get("data-pptx-shape-id") |
| 176 | if not scope or not shape_id: |
| 177 | return None |
| 178 | return f"{scope}:{shape_id}" |
| 179 | |
| 180 | |
| 181 | def _stamp_source_references(root: ET.Element) -> list[SourceReference]: |
| 182 | existing = [ |
| 183 | element |
| 184 | for _, element in _iter_element_paths(root) |
| 185 | if element.get(SOURCE_REF_ATTRIBUTE) is not None |
| 186 | ] |
| 187 | if existing: |
| 188 | raise ValueError( |
| 189 | f"Input already contains reserved {SOURCE_REF_ATTRIBUTE}; " |
| 190 | "project from the lossless import SVG instead" |
| 191 | ) |
| 192 | |
| 193 | references: list[SourceReference] = [] |
| 194 | seen: set[str] = set() |
| 195 | for path, element in _iter_element_paths(root): |
| 196 | source_ref = _source_reference(element) |
| 197 | if source_ref is None: |
| 198 | continue |
| 199 | if source_ref in seen: |
| 200 | raise ValueError(f"Duplicate source object identity: {source_ref}") |
| 201 | seen.add(source_ref) |
| 202 | references.append( |
| 203 | SourceReference( |
| 204 | source_ref=source_ref, |
| 205 | source_path=path, |
| 206 | ) |
| 207 | ) |
| 208 | element.set(SOURCE_REF_ATTRIBUTE, source_ref) |
| 209 | return references |
| 210 | |
| 211 | |
| 212 | def _index_initial_authoring_references( |
| 213 | root: ET.Element, |
| 214 | references: list[SourceReference], |
| 215 | ) -> None: |
| 216 | by_ref = {reference.source_ref: reference for reference in references} |
| 217 | seen: set[str] = set() |
| 218 | for element in root.iter(): |
| 219 | source_ref = element.get(SOURCE_REF_ATTRIBUTE) |
| 220 | if source_ref is None: |
| 221 | continue |
| 222 | if source_ref in seen: |
| 223 | raise ValueError(f"Duplicate authoring source reference: {source_ref}") |
| 224 | reference = by_ref.get(source_ref) |
| 225 | if reference is None: |
| 226 | raise ValueError(f"Unknown authoring source reference: {source_ref}") |
| 227 | seen.add(source_ref) |
| 228 | reference.initial_authoring_subtree_sha256 = semantic_subtree_sha256( |
| 229 | element, |
| 230 | ignored_attributes=frozenset({SOURCE_REF_ATTRIBUTE}), |
| 231 | ) |
| 232 | |
| 233 | missing = sorted(set(by_ref) - seen) |
| 234 | if missing: |
| 235 | raise ValueError( |
| 236 | "Authoring projection dropped source-referenced object(s): " |
| 237 | + ", ".join(missing[:5]) |
| 238 | ) |
| 239 | |
| 240 | |
| 241 | @dataclass |
| 242 | class ProjectionStats: |
| 243 | txbody_metadata: int = 0 |
| 244 | hidden_geometry_carriers: int = 0 |
| 245 | geometry_preview_wrappers: int = 0 |
| 246 | geometry_detail_markers: int = 0 |
| 247 | asset_references_rewritten: int = 0 |
| 248 | coordinate_attributes_compacted: int = 0 |
| 249 | source_attributes: Counter[str] = field(default_factory=Counter) |
| 250 | |
| 251 | def as_dict(self) -> dict[str, object]: |
| 252 | return { |
| 253 | "txbody_metadata": self.txbody_metadata, |
| 254 | "hidden_geometry_carriers": self.hidden_geometry_carriers, |
| 255 | "geometry_preview_wrappers": self.geometry_preview_wrappers, |
| 256 | "geometry_detail_markers": self.geometry_detail_markers, |
| 257 | "source_attributes": dict(sorted(self.source_attributes.items())), |
| 258 | "asset_references_rewritten": self.asset_references_rewritten, |
| 259 | "coordinate_attributes_compacted": self.coordinate_attributes_compacted, |
| 260 | } |
| 261 | |
| 262 | def merge(self, other: "ProjectionStats") -> None: |
| 263 | self.txbody_metadata += other.txbody_metadata |
| 264 | self.hidden_geometry_carriers += other.hidden_geometry_carriers |
| 265 | self.geometry_preview_wrappers += other.geometry_preview_wrappers |
| 266 | self.geometry_detail_markers += other.geometry_detail_markers |
| 267 | self.asset_references_rewritten += other.asset_references_rewritten |
| 268 | self.coordinate_attributes_compacted += ( |
| 269 | other.coordinate_attributes_compacted |
| 270 | ) |
| 271 | self.source_attributes.update(other.source_attributes) |
| 272 | |
| 273 | |
| 274 | @dataclass |
| 275 | class ProjectionReport: |
| 276 | source: Path |
| 277 | output: Path |
| 278 | original_bytes: int |
| 279 | projected_bytes: int |
| 280 | stats: ProjectionStats |
| 281 | source_sha256: str |
| 282 | initial_authoring_sha256: str |
| 283 | source_references: list[SourceReference] |
| 284 | |
| 285 | def as_dict(self) -> dict[str, object]: |
| 286 | saved = self.original_bytes - self.projected_bytes |
| 287 | reduction = (saved / self.original_bytes * 100) if self.original_bytes else 0.0 |
| 288 | return { |
| 289 | "source": str(self.source), |
| 290 | "output": str(self.output), |
| 291 | "original_bytes": self.original_bytes, |
| 292 | "projected_bytes": self.projected_bytes, |
| 293 | "bytes_saved": saved, |
| 294 | "reduction_percent": round(reduction, 2), |
| 295 | "source_sha256": self.source_sha256, |
| 296 | "initial_authoring_sha256": self.initial_authoring_sha256, |
| 297 | "source_ref_count": len(self.source_references), |
| 298 | "removed": self.stats.as_dict(), |
| 299 | } |
| 300 | |
| 301 | |
| 302 | def _is_hidden_geometry_carrier(element: ET.Element) -> bool: |
| 303 | if element.get("data-pptx-part") != "geometry": |
| 304 | return False |
| 305 | visibility = (element.get("visibility") or "").strip().lower() |
| 306 | display = (element.get("display") or "").strip().lower() |
| 307 | style = (element.get("style") or "").replace(" ", "").lower() |
| 308 | return ( |
| 309 | visibility == "hidden" |
| 310 | or display == "none" |
| 311 | or "visibility:hidden" in style |
| 312 | or "display:none" in style |
| 313 | ) |
| 314 | |
| 315 | |
| 316 | def _append_tail(parent: ET.Element, index: int, tail: str | None) -> None: |
| 317 | if not tail: |
| 318 | return |
| 319 | if index > 0: |
| 320 | previous = list(parent)[index - 1] |
| 321 | previous.tail = (previous.tail or "") + tail |
| 322 | else: |
| 323 | parent.text = (parent.text or "") + tail |
| 324 | |
| 325 | |
| 326 | def _remove_child(parent: ET.Element, child: ET.Element) -> None: |
| 327 | children = list(parent) |
| 328 | index = children.index(child) |
| 329 | tail = child.tail |
| 330 | parent.remove(child) |
| 331 | _append_tail(parent, index, tail) |
| 332 | |
| 333 | |
| 334 | def _unwrap_preview(parent: ET.Element, wrapper: ET.Element) -> bool: |
| 335 | """Promote a marker-only preview wrapper without changing its geometry.""" |
| 336 | if wrapper.attrib or (wrapper.text and wrapper.text.strip()): |
| 337 | return False |
| 338 | |
| 339 | siblings = list(parent) |
| 340 | index = siblings.index(wrapper) |
| 341 | promoted = list(wrapper) |
| 342 | wrapper_tail = wrapper.tail |
| 343 | for child in promoted: |
| 344 | wrapper.remove(child) |
| 345 | parent.remove(wrapper) |
| 346 | |
| 347 | for offset, child in enumerate(promoted): |
| 348 | parent.insert(index + offset, child) |
| 349 | |
| 350 | if promoted: |
| 351 | promoted[-1].tail = (promoted[-1].tail or "") + (wrapper_tail or "") |
| 352 | else: |
| 353 | _append_tail(parent, index, wrapper_tail) |
| 354 | return True |
| 355 | |
| 356 | |
| 357 | def _strip_import_attributes(element: ET.Element, stats: ProjectionStats) -> None: |
| 358 | for name in list(element.attrib): |
| 359 | if name not in IMPORT_SOURCE_ATTRIBUTES: |
| 360 | continue |
| 361 | stats.source_attributes[name] += 1 |
| 362 | del element.attrib[name] |
| 363 | |
| 364 | |
| 365 | def _project_subtree(parent: ET.Element, stats: ProjectionStats) -> None: |
| 366 | for child in list(parent): |
| 367 | part = child.get("data-pptx-part") |
| 368 | tag = _local_name(child.tag) |
| 369 | |
| 370 | if tag == "metadata" and part == "txbody": |
| 371 | stats.txbody_metadata += 1 |
| 372 | _remove_child(parent, child) |
| 373 | continue |
| 374 | |
| 375 | if _is_hidden_geometry_carrier(child): |
| 376 | stats.hidden_geometry_carriers += 1 |
| 377 | _remove_child(parent, child) |
| 378 | continue |
| 379 | |
| 380 | _project_subtree(child, stats) |
| 381 | _strip_import_attributes(child, stats) |
| 382 | |
| 383 | if part == "geometry-preview": |
| 384 | child.attrib.pop("data-pptx-part", None) |
| 385 | if _unwrap_preview(parent, child): |
| 386 | stats.geometry_preview_wrappers += 1 |
| 387 | elif part == "geometry-detail": |
| 388 | child.attrib.pop("data-pptx-part", None) |
| 389 | stats.geometry_detail_markers += 1 |
| 390 | |
| 391 | |
| 392 | def _rewrite_asset_reference(value: str, source_dir: Path, output_dir: Path) -> str: |
| 393 | if not value or value.startswith("#"): |
| 394 | return value |
| 395 | parsed = urlsplit(value) |
| 396 | if parsed.scheme or parsed.netloc or not parsed.path: |
| 397 | return value |
| 398 | |
| 399 | resolved = (source_dir / parsed.path).resolve() |
| 400 | try: |
| 401 | relative = os.path.relpath(resolved, output_dir).replace(os.sep, "/") |
| 402 | except ValueError: |
| 403 | relative = resolved.as_uri() |
| 404 | return urlunsplit(("", "", relative, parsed.query, parsed.fragment)) |
| 405 | |
| 406 | |
| 407 | def _rewrite_asset_references( |
| 408 | root: ET.Element, |
| 409 | source_dir: Path, |
| 410 | output_dir: Path, |
| 411 | stats: ProjectionStats, |
| 412 | ) -> None: |
| 413 | for element in root.iter(): |
| 414 | for name in ("href", f"{{{XLINK_NS}}}href"): |
| 415 | current = element.get(name) |
| 416 | if current is None: |
| 417 | continue |
| 418 | rewritten = _rewrite_asset_reference(current, source_dir, output_dir) |
| 419 | if rewritten != current: |
| 420 | element.set(name, rewritten) |
| 421 | stats.asset_references_rewritten += 1 |
| 422 | |
| 423 | |
| 424 | def _render_projection(source: Path, output: Path) -> tuple[ProjectionReport, bytes]: |
| 425 | """Build one projection in memory without changing source or destination.""" |
| 426 | original = source.read_bytes() |
| 427 | parser = ET.XMLParser( |
| 428 | target=ET.TreeBuilder(insert_comments=True, insert_pis=True), |
| 429 | ) |
| 430 | root = ET.fromstring(original, parser=parser) |
| 431 | if _local_name(root.tag) != "svg": |
| 432 | raise ValueError(f"Root element is not <svg>: {source}") |
| 433 | |
| 434 | source_references = _stamp_source_references(root) |
| 435 | stats = ProjectionStats() |
| 436 | _project_subtree(root, stats) |
| 437 | _strip_import_attributes(root, stats) |
| 438 | _rewrite_asset_references(root, source.parent, output.parent, stats) |
| 439 | stats.coordinate_attributes_compacted = compact_svg_tree( |
| 440 | root, |
| 441 | ).changed_attributes |
| 442 | _index_initial_authoring_references(root, source_references) |
| 443 | |
| 444 | projected = ET.tostring(root, encoding="utf-8", xml_declaration=False) |
| 445 | if not projected.endswith(b"\n"): |
| 446 | projected += b"\n" |
| 447 | |
| 448 | report = ProjectionReport( |
| 449 | source=source, |
| 450 | output=output, |
| 451 | original_bytes=len(original), |
| 452 | projected_bytes=len(projected), |
| 453 | stats=stats, |
| 454 | source_sha256=hashlib.sha256(original).hexdigest(), |
| 455 | initial_authoring_sha256=hashlib.sha256(projected).hexdigest(), |
| 456 | source_references=source_references, |
| 457 | ) |
| 458 | return report, projected |
| 459 | |
| 460 | |
| 461 | def _portable_path(path: Path, base: Path) -> str: |
| 462 | try: |
| 463 | return os.path.relpath(path, base).replace(os.sep, "/") |
| 464 | except ValueError: |
| 465 | return path.resolve().as_uri() |
| 466 | |
| 467 | |
| 468 | def _authoring_manifest_bytes( |
| 469 | reports: list[ProjectionReport], |
| 470 | source_root: Path, |
| 471 | output_dir: Path, |
| 472 | projection_kind: str, |
| 473 | ) -> bytes: |
| 474 | documents = [] |
| 475 | for report in sorted(reports, key=lambda item: item.output.as_posix()): |
| 476 | documents.append({ |
| 477 | "source": report.source.relative_to(source_root).as_posix(), |
| 478 | "authoring": report.output.relative_to(output_dir).as_posix(), |
| 479 | "source_sha256": report.source_sha256, |
| 480 | "initial_authoring_sha256": report.initial_authoring_sha256, |
| 481 | "source_refs": { |
| 482 | reference.source_ref: reference.as_dict() |
| 483 | for reference in sorted( |
| 484 | report.source_references, |
| 485 | key=lambda item: item.source_ref, |
| 486 | ) |
| 487 | }, |
| 488 | }) |
| 489 | |
| 490 | payload = { |
| 491 | "schema": AUTHORING_SCHEMA, |
| 492 | "projection_kind": projection_kind, |
| 493 | "source_root": _portable_path(source_root, output_dir), |
| 494 | "authoring_root": ".", |
| 495 | "source_ref_attribute": SOURCE_REF_ATTRIBUTE, |
| 496 | "file_count": len(documents), |
| 497 | "source_ref_count": sum(len(report.source_references) for report in reports), |
| 498 | "documents": documents, |
| 499 | } |
| 500 | return ( |
| 501 | json.dumps(payload, ensure_ascii=False, indent=2) + "\n" |
| 502 | ).encode("utf-8") |
| 503 | |
| 504 | |
| 505 | def _authoring_document_kind(path: Path) -> str: |
| 506 | name = path.name |
| 507 | if name.startswith("master_"): |
| 508 | return "master" |
| 509 | if name.startswith("layout_"): |
| 510 | return "layout" |
| 511 | if name.startswith("slide_"): |
| 512 | return "slide" |
| 513 | return "generic" |
| 514 | |
| 515 | |
| 516 | def _authoring_summary_document( |
| 517 | path: Path, |
| 518 | relative_name: str, |
| 519 | ) -> dict[str, object]: |
| 520 | try: |
| 521 | root = ET.parse(path).getroot() |
| 522 | except (OSError, ET.ParseError) as exc: |
| 523 | raise ValueError(f"Cannot summarize authoring SVG {path}: {exc}") from exc |
| 524 | if _local_name(root.tag) != "svg": |
| 525 | raise ValueError(f"Authoring document root is not <svg>: {path}") |
| 526 | |
| 527 | elements = list(root.iter()) |
| 528 | icon_references = sorted({ |
| 529 | icon_name |
| 530 | for element in elements |
| 531 | if (icon_name := element.get("data-icon")) |
| 532 | }) |
| 533 | text_elements = [ |
| 534 | element for element in elements |
| 535 | if _local_name(element.tag) == "text" |
| 536 | ] |
| 537 | return { |
| 538 | "file": relative_name, |
| 539 | "kind": _authoring_document_kind(path), |
| 540 | "bytes": path.stat().st_size, |
| 541 | "viewBox": root.get("viewBox"), |
| 542 | "elements": len(elements), |
| 543 | "top_level_elements": len(root), |
| 544 | "drawables": sum( |
| 545 | _local_name(element.tag) in _DRAWABLE_TAGS |
| 546 | for element in elements |
| 547 | ), |
| 548 | "text_elements": len(text_elements), |
| 549 | "text_characters": sum( |
| 550 | len("".join(element.itertext())) |
| 551 | for element in text_elements |
| 552 | ), |
| 553 | "images": sum( |
| 554 | _local_name(element.tag) == "image" |
| 555 | for element in elements |
| 556 | ), |
| 557 | "icon_uses": sum( |
| 558 | element.get("data-icon") is not None |
| 559 | for element in elements |
| 560 | ), |
| 561 | "icon_refs": icon_references, |
| 562 | "placeholders": sum( |
| 563 | element.get("data-pptx-placeholder") is not None |
| 564 | for element in elements |
| 565 | ), |
| 566 | "inline_source_refs": sum( |
| 567 | element.get(SOURCE_REF_ATTRIBUTE) is not None |
| 568 | for element in elements |
| 569 | ), |
| 570 | } |
| 571 | |
| 572 | |
| 573 | def _load_authoring_summary_manifest( |
| 574 | authoring_dir: Path, |
| 575 | ) -> tuple[dict[str, object], list[str]]: |
| 576 | manifest_path = authoring_dir / AUTHORING_MANIFEST_NAME |
| 577 | try: |
| 578 | manifest = json.loads(manifest_path.read_text(encoding="utf-8")) |
| 579 | except FileNotFoundError as exc: |
| 580 | raise ValueError( |
| 581 | f"Authoring manifest not found: {manifest_path}" |
| 582 | ) from exc |
| 583 | except (OSError, json.JSONDecodeError) as exc: |
| 584 | raise ValueError( |
| 585 | f"Cannot decode authoring manifest {manifest_path}: {exc}" |
| 586 | ) from exc |
| 587 | if not isinstance(manifest, dict) or manifest.get("schema") != AUTHORING_SCHEMA: |
| 588 | raise ValueError( |
| 589 | f"Unsupported authoring manifest schema in {manifest_path}" |
| 590 | ) |
| 591 | documents = manifest.get("documents") |
| 592 | if not isinstance(documents, list): |
| 593 | raise ValueError( |
| 594 | f"Authoring manifest documents must be an array: {manifest_path}" |
| 595 | ) |
| 596 | |
| 597 | names: list[str] = [] |
| 598 | for index, document in enumerate(documents): |
| 599 | if not isinstance(document, dict): |
| 600 | raise ValueError( |
| 601 | f"Authoring manifest documents[{index}] must be an object" |
| 602 | ) |
| 603 | name = document.get("authoring") |
| 604 | relative = Path(name) if isinstance(name, str) else Path() |
| 605 | if ( |
| 606 | not isinstance(name, str) |
| 607 | or not name |
| 608 | or relative.is_absolute() |
| 609 | or any(part in {"", ".", ".."} for part in relative.parts) |
| 610 | or relative.suffix.lower() != ".svg" |
| 611 | ): |
| 612 | raise ValueError( |
| 613 | f"Authoring manifest documents[{index}].authoring is invalid" |
| 614 | ) |
| 615 | names.append(name) |
| 616 | if len(names) != len(set(names)): |
| 617 | raise ValueError("Authoring manifest contains duplicate document names") |
| 618 | |
| 619 | actual_names = sorted( |
| 620 | path.relative_to(authoring_dir).as_posix() |
| 621 | for path in authoring_dir.rglob("*.svg") |
| 622 | if path.is_file() |
| 623 | ) |
| 624 | if sorted(names) != actual_names: |
| 625 | raise ValueError( |
| 626 | "Authoring manifest/file roster differs while building summary" |
| 627 | ) |
| 628 | return manifest, sorted(names) |
| 629 | |
| 630 | |
| 631 | def _authoring_summary_bytes(authoring_dir: Path) -> bytes: |
| 632 | manifest, document_names = _load_authoring_summary_manifest(authoring_dir) |
| 633 | documents = [ |
| 634 | _authoring_summary_document(authoring_dir / name, name) |
| 635 | for name in document_names |
| 636 | ] |
| 637 | total_icon_assets = { |
| 638 | icon_name |
| 639 | for document in documents |
| 640 | for icon_name in document["icon_refs"] |
| 641 | } |
| 642 | totals = { |
| 643 | "svg_bytes": sum(int(document["bytes"]) for document in documents), |
| 644 | "elements": sum(int(document["elements"]) for document in documents), |
| 645 | "top_level_elements": sum( |
| 646 | int(document["top_level_elements"]) |
| 647 | for document in documents |
| 648 | ), |
| 649 | "drawables": sum(int(document["drawables"]) for document in documents), |
| 650 | "text_elements": sum( |
| 651 | int(document["text_elements"]) |
| 652 | for document in documents |
| 653 | ), |
| 654 | "text_characters": sum( |
| 655 | int(document["text_characters"]) |
| 656 | for document in documents |
| 657 | ), |
| 658 | "images": sum(int(document["images"]) for document in documents), |
| 659 | "icon_uses": sum(int(document["icon_uses"]) for document in documents), |
| 660 | "unique_icon_assets": len(total_icon_assets), |
| 661 | "placeholders": sum( |
| 662 | int(document["placeholders"]) |
| 663 | for document in documents |
| 664 | ), |
| 665 | "inline_source_refs": sum( |
| 666 | int(document["inline_source_refs"]) |
| 667 | for document in documents |
| 668 | ), |
| 669 | "machine_source_refs": manifest.get("source_ref_count"), |
| 670 | } |
| 671 | payload = { |
| 672 | "schema": AUTHORING_SUMMARY_SCHEMA, |
| 673 | "projection_kind": manifest.get("projection_kind"), |
| 674 | "authoring_root": ".", |
| 675 | "machine_manifest": AUTHORING_MANIFEST_NAME, |
| 676 | "machine_manifest_policy": "tool-only; do not load into model context", |
| 677 | "file_count": len(documents), |
| 678 | "totals": totals, |
| 679 | "documents": documents, |
| 680 | } |
| 681 | return ( |
| 682 | json.dumps(payload, ensure_ascii=False, indent=2) + "\n" |
| 683 | ).encode("utf-8") |
| 684 | |
| 685 | |
| 686 | def write_authoring_summary(authoring_dir: Path) -> Path: |
| 687 | """Regenerate the model-readable summary from the current authoring SVGs.""" |
| 688 | authoring_dir = Path(authoring_dir).resolve() |
| 689 | if not authoring_dir.is_dir(): |
| 690 | raise ValueError(f"Authoring directory not found: {authoring_dir}") |
| 691 | payload = _authoring_summary_bytes(authoring_dir) |
| 692 | summary_path = authoring_dir / AUTHORING_SUMMARY_NAME |
| 693 | with tempfile.NamedTemporaryFile( |
| 694 | prefix=f".{AUTHORING_SUMMARY_NAME}.", |
| 695 | suffix=".tmp", |
| 696 | dir=authoring_dir, |
| 697 | delete=False, |
| 698 | ) as handle: |
| 699 | temporary_path = Path(handle.name) |
| 700 | handle.write(payload) |
| 701 | try: |
| 702 | temporary_path.chmod(0o644) |
| 703 | temporary_path.replace(summary_path) |
| 704 | except OSError: |
| 705 | temporary_path.unlink(missing_ok=True) |
| 706 | raise |
| 707 | return summary_path |
| 708 | |
| 709 | |
| 710 | def _nearest_existing_directory(path: Path) -> Path: |
| 711 | candidate = path |
| 712 | while not os.path.lexists(candidate): |
| 713 | parent = candidate.parent |
| 714 | if parent == candidate: |
| 715 | break |
| 716 | candidate = parent |
| 717 | if not candidate.is_dir(): |
| 718 | raise NotADirectoryError(f"Output parent is not a directory: {candidate}") |
| 719 | return candidate |
| 720 | |
| 721 | |
| 722 | def _ensure_directory(path: Path, created: list[Path]) -> None: |
| 723 | missing: list[Path] = [] |
| 724 | candidate = path |
| 725 | while not candidate.exists(): |
| 726 | if os.path.lexists(candidate): |
| 727 | raise NotADirectoryError(f"Output parent is not a directory: {candidate}") |
| 728 | missing.append(candidate) |
| 729 | parent = candidate.parent |
| 730 | if parent == candidate: |
| 731 | raise NotADirectoryError(f"Cannot resolve output parent: {path}") |
| 732 | candidate = parent |
| 733 | if not candidate.is_dir(): |
| 734 | raise NotADirectoryError(f"Output parent is not a directory: {candidate}") |
| 735 | |
| 736 | for directory in reversed(missing): |
| 737 | directory.mkdir() |
| 738 | created.append(directory) |
| 739 | |
| 740 | |
| 741 | def _remove_created_directories(created: list[Path]) -> list[str]: |
| 742 | errors: list[str] = [] |
| 743 | for directory in reversed(created): |
| 744 | try: |
| 745 | directory.rmdir() |
| 746 | except FileNotFoundError: |
| 747 | continue |
| 748 | except OSError as exc: |
| 749 | errors.append(f"could not remove {directory}: {exc}") |
| 750 | return errors |
| 751 | |
| 752 | |
| 753 | def _rollback_published_files( |
| 754 | published: list[tuple[Path, Path | None]], |
| 755 | created: list[Path], |
| 756 | ) -> list[str]: |
| 757 | errors: list[str] = [] |
| 758 | for target, backup in reversed(published): |
| 759 | try: |
| 760 | if backup is None: |
| 761 | target.unlink(missing_ok=True) |
| 762 | else: |
| 763 | backup.replace(target) |
| 764 | except OSError as exc: |
| 765 | errors.append(f"could not restore {target}: {exc}") |
| 766 | errors.extend(_remove_created_directories(created)) |
| 767 | return errors |
| 768 | |
| 769 | |
| 770 | def _publish_existing_directory( |
| 771 | staged: list[tuple[Path, Path]], |
| 772 | staging_root: Path, |
| 773 | *, |
| 774 | force: bool, |
| 775 | ) -> None: |
| 776 | backup_root = staging_root / "previous" |
| 777 | backups: dict[Path, Path | None] = {} |
| 778 | |
| 779 | for index, (target, _) in enumerate(staged): |
| 780 | if not os.path.lexists(target): |
| 781 | backups[target] = None |
| 782 | continue |
| 783 | if not force: |
| 784 | raise FileExistsError(f"Output file already exists: {target}") |
| 785 | if target.is_dir() and not target.is_symlink(): |
| 786 | raise IsADirectoryError(f"Output target is a directory: {target}") |
| 787 | |
| 788 | backup = backup_root / f"{index:06d}.bak" |
| 789 | backup.parent.mkdir(parents=True, exist_ok=True) |
| 790 | shutil.copy2(target, backup, follow_symlinks=False) |
| 791 | backups[target] = backup |
| 792 | |
| 793 | created: list[Path] = [] |
| 794 | published: list[tuple[Path, Path | None]] = [] |
| 795 | try: |
| 796 | for target, _ in staged: |
| 797 | _ensure_directory(target.parent, created) |
| 798 | |
| 799 | staging_device = staging_root.stat().st_dev |
| 800 | for target, _ in staged: |
| 801 | if target.parent.stat().st_dev != staging_device: |
| 802 | raise OSError( |
| 803 | f"Cannot atomically publish across filesystems: {target}" |
| 804 | ) |
| 805 | if backups[target] is None and os.path.lexists(target): |
| 806 | raise FileExistsError( |
| 807 | f"Output appeared while projections were staged: {target}" |
| 808 | ) |
| 809 | |
| 810 | for target, staged_file in staged: |
| 811 | staged_file.replace(target) |
| 812 | published.append((target, backups[target])) |
| 813 | except OSError as exc: |
| 814 | rollback_errors = _rollback_published_files(published, created) |
| 815 | if rollback_errors: |
| 816 | details = "; ".join(rollback_errors) |
| 817 | raise RuntimeError( |
| 818 | f"Batch publish failed ({exc}); rollback was incomplete: {details}" |
| 819 | ) from exc |
| 820 | raise |
| 821 | |
| 822 | |
| 823 | def project_svg_batch( |
| 824 | mapping: list[tuple[Path, Path]], |
| 825 | source_root: Path, |
| 826 | output_dir: Path, |
| 827 | *, |
| 828 | force: bool, |
| 829 | projection_kind: str, |
| 830 | ) -> list[ProjectionReport]: |
| 831 | """Build and publish one complete authoring bundle transactionally.""" |
| 832 | rendered = [_render_projection(source, output) for source, output in mapping] |
| 833 | staging_parent = _nearest_existing_directory(output_dir.parent) |
| 834 | |
| 835 | with tempfile.TemporaryDirectory( |
| 836 | prefix=".svg-authoring-view-", |
| 837 | dir=staging_parent, |
| 838 | ) as temporary: |
| 839 | staging_root = Path(temporary) |
| 840 | new_root = staging_root / "projected" |
| 841 | staged: list[tuple[Path, Path]] = [] |
| 842 | |
| 843 | for report, projected in rendered: |
| 844 | relative = report.output.relative_to(output_dir) |
| 845 | staged_file = new_root / relative |
| 846 | staged_file.parent.mkdir(parents=True, exist_ok=True) |
| 847 | staged_file.write_bytes(projected) |
| 848 | staged.append((report.output, staged_file)) |
| 849 | |
| 850 | manifest_path = output_dir / AUTHORING_MANIFEST_NAME |
| 851 | staged_manifest = new_root / AUTHORING_MANIFEST_NAME |
| 852 | staged_manifest.write_bytes( |
| 853 | _authoring_manifest_bytes( |
| 854 | [report for report, _ in rendered], |
| 855 | source_root, |
| 856 | output_dir, |
| 857 | projection_kind, |
| 858 | ) |
| 859 | ) |
| 860 | staged.append((manifest_path, staged_manifest)) |
| 861 | staged_summary = write_authoring_summary(new_root) |
| 862 | staged.append( |
| 863 | ( |
| 864 | output_dir / AUTHORING_SUMMARY_NAME, |
| 865 | staged_summary, |
| 866 | ) |
| 867 | ) |
| 868 | |
| 869 | if not output_dir.exists(): |
| 870 | created: list[Path] = [] |
| 871 | try: |
| 872 | _ensure_directory(output_dir.parent, created) |
| 873 | if os.path.lexists(output_dir): |
| 874 | raise FileExistsError( |
| 875 | f"Output directory appeared while projections were staged: {output_dir}" |
| 876 | ) |
| 877 | if output_dir.parent.stat().st_dev != staging_root.stat().st_dev: |
| 878 | raise OSError( |
| 879 | f"Cannot atomically publish across filesystems: {output_dir}" |
| 880 | ) |
| 881 | new_root.replace(output_dir) |
| 882 | except OSError as exc: |
| 883 | cleanup_errors = _remove_created_directories(created) |
| 884 | if cleanup_errors: |
| 885 | details = "; ".join(cleanup_errors) |
| 886 | raise RuntimeError( |
| 887 | f"Batch publish failed ({exc}); cleanup was incomplete: {details}" |
| 888 | ) from exc |
| 889 | raise |
| 890 | else: |
| 891 | _publish_existing_directory( |
| 892 | staged, |
| 893 | staging_root, |
| 894 | force=force, |
| 895 | ) |
| 896 | |
| 897 | return [report for report, _ in rendered] |
| 898 | |
| 899 | |
| 900 | def _is_within(path: Path, parent: Path) -> bool: |
| 901 | try: |
| 902 | path.relative_to(parent) |
| 903 | except ValueError: |
| 904 | return False |
| 905 | return True |
| 906 | |
| 907 | |
| 908 | def _source_mapping(input_path: Path, output_dir: Path) -> list[tuple[Path, Path]]: |
| 909 | if input_path.is_file(): |
| 910 | if input_path.suffix.lower() != ".svg": |
| 911 | raise ValueError(f"Input file must use the .svg extension: {input_path}") |
| 912 | return [(input_path, output_dir / input_path.name)] |
| 913 | |
| 914 | sources = sorted( |
| 915 | path for path in input_path.rglob("*") |
| 916 | if path.is_file() and path.suffix.lower() == ".svg" |
| 917 | ) |
| 918 | if not sources: |
| 919 | raise ValueError(f"No SVG files found under: {input_path}") |
| 920 | return [(source, output_dir / source.relative_to(input_path)) for source in sources] |
| 921 | |
| 922 | |
| 923 | def build_parser() -> argparse.ArgumentParser: |
| 924 | parser = argparse.ArgumentParser( |
| 925 | description=( |
| 926 | "Create lightweight editable IR bundles from PPTX-imported SVG files." |
| 927 | ), |
| 928 | formatter_class=argparse.RawDescriptionHelpFormatter, |
| 929 | ) |
| 930 | parser.add_argument("input", type=Path, help="SVG file or directory to project") |
| 931 | parser.add_argument( |
| 932 | "-o", |
| 933 | "--output-dir", |
| 934 | type=Path, |
| 935 | help="Explicit destination directory for projected SVG copies", |
| 936 | ) |
| 937 | parser.add_argument( |
| 938 | "--refresh-summary", |
| 939 | action="store_true", |
| 940 | help=( |
| 941 | "Regenerate authoring_summary.json for an existing authoring " |
| 942 | "bundle; input must be that bundle directory and -o is omitted" |
| 943 | ), |
| 944 | ) |
| 945 | parser.add_argument( |
| 946 | "--force", |
| 947 | action="store_true", |
| 948 | help=( |
| 949 | "Replace authoring files/manifest that already exist " |
| 950 | "(never changes source files)" |
| 951 | ), |
| 952 | ) |
| 953 | parser.add_argument( |
| 954 | "--projection-kind", |
| 955 | choices=("layered", "flat", "generic"), |
| 956 | default="generic", |
| 957 | help="Record the IR representation kind in bundle metadata", |
| 958 | ) |
| 959 | return parser |
| 960 | |
| 961 | |
| 962 | def main(argv: Optional[list[str]] = None) -> int: |
| 963 | parser = build_parser() |
| 964 | args = parser.parse_args(argv) |
| 965 | input_path = args.input.resolve() |
| 966 | |
| 967 | if not input_path.exists(): |
| 968 | print(f"Error: input does not exist: {input_path}", file=sys.stderr) |
| 969 | return 1 |
| 970 | if args.refresh_summary: |
| 971 | if args.output_dir is not None: |
| 972 | print( |
| 973 | "Error: --refresh-summary does not accept -o/--output-dir", |
| 974 | file=sys.stderr, |
| 975 | ) |
| 976 | return 1 |
| 977 | try: |
| 978 | summary_path = write_authoring_summary(input_path) |
| 979 | except (OSError, ValueError) as exc: |
| 980 | print(f"Error: {exc}", file=sys.stderr) |
| 981 | return 1 |
| 982 | print(json.dumps({ |
| 983 | "authoring_dir": str(input_path), |
| 984 | "summary": str(summary_path), |
| 985 | "summary_bytes": summary_path.stat().st_size, |
| 986 | }, ensure_ascii=False, indent=2)) |
| 987 | return 0 |
| 988 | if args.output_dir is None: |
| 989 | parser.error("-o/--output-dir is required unless --refresh-summary is used") |
| 990 | output_dir = args.output_dir.resolve() |
| 991 | |
| 992 | if output_dir.exists() and not output_dir.is_dir(): |
| 993 | print(f"Error: output path is not a directory: {output_dir}", file=sys.stderr) |
| 994 | return 1 |
| 995 | if input_path.is_dir() and _is_within(output_dir, input_path): |
| 996 | print("Error: output directory must not be inside the input directory", file=sys.stderr) |
| 997 | return 1 |
| 998 | |
| 999 | try: |
| 1000 | mapping = _source_mapping(input_path, output_dir) |
| 1001 | except ValueError as exc: |
| 1002 | print(f"Error: {exc}", file=sys.stderr) |
| 1003 | return 1 |
| 1004 | |
| 1005 | same_file = [source for source, target in mapping if source.resolve() == target.resolve()] |
| 1006 | if same_file: |
| 1007 | print(f"Error: output would overwrite source SVG: {same_file[0]}", file=sys.stderr) |
| 1008 | return 1 |
| 1009 | |
| 1010 | collisions = [target for _, target in mapping if os.path.lexists(target)] |
| 1011 | manifest_path = output_dir / AUTHORING_MANIFEST_NAME |
| 1012 | if os.path.lexists(manifest_path): |
| 1013 | collisions.append(manifest_path) |
| 1014 | summary_path = output_dir / AUTHORING_SUMMARY_NAME |
| 1015 | if os.path.lexists(summary_path): |
| 1016 | collisions.append(summary_path) |
| 1017 | if collisions and not args.force: |
| 1018 | print( |
| 1019 | f"Error: {len(collisions)} output file(s) already exist; " |
| 1020 | "use --force to replace the authoring bundle. " |
| 1021 | f"First collision: {collisions[0]}", |
| 1022 | file=sys.stderr, |
| 1023 | ) |
| 1024 | return 1 |
| 1025 | |
| 1026 | reports: list[ProjectionReport] = [] |
| 1027 | try: |
| 1028 | source_root = input_path if input_path.is_dir() else input_path.parent |
| 1029 | reports = project_svg_batch( |
| 1030 | mapping, |
| 1031 | source_root, |
| 1032 | output_dir, |
| 1033 | force=args.force, |
| 1034 | projection_kind=args.projection_kind, |
| 1035 | ) |
| 1036 | except (ET.ParseError, OSError, RuntimeError, ValueError) as exc: |
| 1037 | print(f"Error: {exc}", file=sys.stderr) |
| 1038 | return 1 |
| 1039 | |
| 1040 | total_stats = ProjectionStats() |
| 1041 | original_bytes = 0 |
| 1042 | projected_bytes = 0 |
| 1043 | for report in reports: |
| 1044 | original_bytes += report.original_bytes |
| 1045 | projected_bytes += report.projected_bytes |
| 1046 | total_stats.merge(report.stats) |
| 1047 | |
| 1048 | bytes_saved = original_bytes - projected_bytes |
| 1049 | reduction = (bytes_saved / original_bytes * 100) if original_bytes else 0.0 |
| 1050 | result = { |
| 1051 | "input": str(input_path), |
| 1052 | "output_dir": str(output_dir), |
| 1053 | "manifest": str(output_dir / AUTHORING_MANIFEST_NAME), |
| 1054 | "summary": str(output_dir / AUTHORING_SUMMARY_NAME), |
| 1055 | "projection_kind": args.projection_kind, |
| 1056 | "file_count": len(reports), |
| 1057 | "files": [report.as_dict() for report in reports], |
| 1058 | "totals": { |
| 1059 | "original_bytes": original_bytes, |
| 1060 | "projected_bytes": projected_bytes, |
| 1061 | "bytes_saved": bytes_saved, |
| 1062 | "reduction_percent": round(reduction, 2), |
| 1063 | "removed": total_stats.as_dict(), |
| 1064 | }, |
| 1065 | } |
| 1066 | print(json.dumps(result, ensure_ascii=False, indent=2)) |
| 1067 | return 0 |
| 1068 | |
| 1069 | |
| 1070 | if __name__ == "__main__": |
| 1071 | raise SystemExit(main()) |
| 1072 |