| 1 | #!/usr/bin/env python3 |
| 2 | """ |
| 3 | PPT Master - Native SVG Metadata Store |
| 4 | |
| 5 | Deduplicate opaque PowerPoint-native payloads and repeated restoration |
| 6 | attributes into one deterministic gzip-compressed workspace store, then |
| 7 | hydrate legacy inline metadata on demand. |
| 8 | |
| 9 | Usage: |
| 10 | Imported by mirror materialization, SVG validation, and SVG-to-PPTX export. |
| 11 | |
| 12 | Examples: |
| 13 | from native_payloads import hydrate_native_payload_refs |
| 14 | |
| 15 | Dependencies: |
| 16 | None (standard library only). |
| 17 | """ |
| 18 | |
| 19 | from __future__ import annotations |
| 20 | |
| 21 | import base64 |
| 22 | import binascii |
| 23 | import gzip |
| 24 | import hashlib |
| 25 | import json |
| 26 | import re |
| 27 | from dataclasses import dataclass |
| 28 | from pathlib import Path, PurePosixPath |
| 29 | from xml.etree import ElementTree as ET |
| 30 | |
| 31 | |
| 32 | LEGACY_PAYLOAD_STORE_SCHEMA = "ppt-master.native-payload-store.v1" |
| 33 | PAYLOAD_STORE_SCHEMA = "ppt-master.native-payload-store.v2" |
| 34 | PAYLOAD_STORE_FILENAME = "native_payloads.json.gz" |
| 35 | PAYLOAD_STORE_RELATIVE_PATH = Path("templates") / PAYLOAD_STORE_FILENAME |
| 36 | PAYLOAD_STORE_REFERENCE_PREFIX = ( |
| 37 | f"project:{PAYLOAD_STORE_RELATIVE_PATH.as_posix()}#sha256:" |
| 38 | ) |
| 39 | |
| 40 | NATIVE_RECORD_REF_ATTRIBUTE = "data-pptx-native-ref" |
| 41 | TXBODY_REF_ATTRIBUTE = "data-pptx-ref" |
| 42 | SHAPE_STYLE_ATTRIBUTE = "data-pptx-shape-style" |
| 43 | SHAPE_STYLE_REF_ATTRIBUTE = "data-pptx-shape-style-ref" |
| 44 | CUSTOM_GEOMETRY_ATTRIBUTE = "data-pptx-custgeom" |
| 45 | CUSTOM_GEOMETRY_REF_ATTRIBUTE = "data-pptx-custgeom-ref" |
| 46 | |
| 47 | _SHA256_RE = re.compile(r"[0-9a-f]{64}") |
| 48 | _NATIVE_RECORD_ID_RE = re.compile(r"r(?:0|[1-9][0-9]*)") |
| 49 | _NATIVE_RECORD_ATTRIBUTES = frozenset({ |
| 50 | "data-pptx-custgeom-ref", |
| 51 | "data-pptx-frame", |
| 52 | "data-pptx-geometry-kind", |
| 53 | "data-pptx-geometry-sha256", |
| 54 | "data-pptx-object", |
| 55 | "data-pptx-part", |
| 56 | "data-pptx-preview-sha256", |
| 57 | "data-pptx-prst", |
| 58 | "data-pptx-ref", |
| 59 | "data-pptx-shape-id", |
| 60 | "data-pptx-shape-name", |
| 61 | "data-pptx-shape-scope", |
| 62 | "data-pptx-shape-style-ref", |
| 63 | "data-pptx-text-sha256", |
| 64 | }) |
| 65 | _NATIVE_RECORD_PREFIXES = ( |
| 66 | "data-pptx-av-", |
| 67 | "data-pptx-end-", |
| 68 | "data-pptx-start-", |
| 69 | ) |
| 70 | |
| 71 | |
| 72 | class NativePayloadError(ValueError): |
| 73 | """Reject malformed, missing, or contradictory native metadata transport.""" |
| 74 | |
| 75 | |
| 76 | @dataclass |
| 77 | class NativePayloadStats: |
| 78 | """Count externalized native metadata and its original inline bytes.""" |
| 79 | |
| 80 | txbody_count: int = 0 |
| 81 | shape_style_count: int = 0 |
| 82 | custom_geometry_count: int = 0 |
| 83 | inline_bytes: int = 0 |
| 84 | native_record_count: int = 0 |
| 85 | native_attribute_bytes: int = 0 |
| 86 | |
| 87 | def merge(self, other: "NativePayloadStats") -> None: |
| 88 | self.txbody_count += other.txbody_count |
| 89 | self.shape_style_count += other.shape_style_count |
| 90 | self.custom_geometry_count += other.custom_geometry_count |
| 91 | self.inline_bytes += other.inline_bytes |
| 92 | self.native_record_count += other.native_record_count |
| 93 | self.native_attribute_bytes += other.native_attribute_bytes |
| 94 | |
| 95 | def as_dict(self) -> dict[str, int]: |
| 96 | return { |
| 97 | "txbody_count": self.txbody_count, |
| 98 | "shape_style_count": self.shape_style_count, |
| 99 | "custom_geometry_count": self.custom_geometry_count, |
| 100 | "inline_bytes": self.inline_bytes, |
| 101 | "native_record_count": self.native_record_count, |
| 102 | "native_attribute_bytes": self.native_attribute_bytes, |
| 103 | } |
| 104 | |
| 105 | |
| 106 | @dataclass(frozen=True) |
| 107 | class NativePayloadStore: |
| 108 | """Validated payload and native-attribute records loaded from one store.""" |
| 109 | |
| 110 | payloads: dict[str, bytes] |
| 111 | native_records: dict[str, dict[str, str]] |
| 112 | |
| 113 | |
| 114 | _STORE_CACHE: dict[Path, tuple[int, int, NativePayloadStore]] = {} |
| 115 | |
| 116 | |
| 117 | def _local_name(name: object) -> str: |
| 118 | return name.rsplit("}", 1)[-1] if isinstance(name, str) else "" |
| 119 | |
| 120 | |
| 121 | def _sha256(payload: bytes) -> str: |
| 122 | return hashlib.sha256(payload).hexdigest() |
| 123 | |
| 124 | |
| 125 | def _decode_base64(value: str, *, context: str) -> bytes: |
| 126 | try: |
| 127 | return base64.b64decode(value.strip(), validate=True) |
| 128 | except (ValueError, binascii.Error) as exc: |
| 129 | raise NativePayloadError(f"{context} is not valid base64: {exc}") from exc |
| 130 | |
| 131 | |
| 132 | def _register_payload(payloads: dict[str, bytes], raw: bytes) -> str: |
| 133 | digest = _sha256(raw) |
| 134 | existing = payloads.get(digest) |
| 135 | if existing is not None and existing != raw: |
| 136 | raise NativePayloadError( |
| 137 | f"SHA-256 collision while registering native payload {digest}" |
| 138 | ) |
| 139 | payloads[digest] = raw |
| 140 | return PAYLOAD_STORE_REFERENCE_PREFIX + digest |
| 141 | |
| 142 | |
| 143 | def _is_native_record_attribute(name: str) -> bool: |
| 144 | return ( |
| 145 | name in _NATIVE_RECORD_ATTRIBUTES |
| 146 | or name.startswith(_NATIVE_RECORD_PREFIXES) |
| 147 | ) |
| 148 | |
| 149 | |
| 150 | def _native_record_attributes(element: ET.Element) -> dict[str, str]: |
| 151 | return { |
| 152 | name: value |
| 153 | for name, value in element.attrib.items() |
| 154 | if _is_native_record_attribute(name) |
| 155 | } |
| 156 | |
| 157 | |
| 158 | def _native_record_key(attributes: dict[str, str]) -> str: |
| 159 | return json.dumps( |
| 160 | attributes, |
| 161 | ensure_ascii=True, |
| 162 | separators=(",", ":"), |
| 163 | sort_keys=True, |
| 164 | ) |
| 165 | |
| 166 | |
| 167 | def collect_native_attribute_record_keys(root: ET.Element) -> set[str]: |
| 168 | """Return canonical record keys for compressible native attributes.""" |
| 169 | keys: set[str] = set() |
| 170 | for element in root.iter(): |
| 171 | if NATIVE_RECORD_REF_ATTRIBUTE in element.attrib: |
| 172 | raise NativePayloadError( |
| 173 | "native metadata is already externalized; source attributes " |
| 174 | "must be hydrated before building a new store" |
| 175 | ) |
| 176 | attributes = _native_record_attributes(element) |
| 177 | if attributes: |
| 178 | keys.add(_native_record_key(attributes)) |
| 179 | return keys |
| 180 | |
| 181 | |
| 182 | def build_native_attribute_records( |
| 183 | record_keys: set[str], |
| 184 | ) -> tuple[dict[str, str], dict[str, dict[str, str]]]: |
| 185 | """Assign deterministic short ids to canonical native-attribute records.""" |
| 186 | ids_by_key: dict[str, str] = {} |
| 187 | records: dict[str, dict[str, str]] = {} |
| 188 | for index, key in enumerate(sorted(record_keys)): |
| 189 | try: |
| 190 | attributes = json.loads(key) |
| 191 | except json.JSONDecodeError as exc: |
| 192 | raise NativePayloadError( |
| 193 | f"Cannot decode canonical native-attribute record: {exc}" |
| 194 | ) from exc |
| 195 | if not isinstance(attributes, dict) or not attributes: |
| 196 | raise NativePayloadError( |
| 197 | "Canonical native-attribute record must be a non-empty object" |
| 198 | ) |
| 199 | if any( |
| 200 | not isinstance(name, str) |
| 201 | or not isinstance(value, str) |
| 202 | or not _is_native_record_attribute(name) |
| 203 | for name, value in attributes.items() |
| 204 | ): |
| 205 | raise NativePayloadError( |
| 206 | "Canonical native-attribute record contains an unsupported field" |
| 207 | ) |
| 208 | record_id = f"r{index}" |
| 209 | ids_by_key[key] = record_id |
| 210 | records[record_id] = attributes |
| 211 | return ids_by_key, records |
| 212 | |
| 213 | |
| 214 | def externalize_native_attribute_records( |
| 215 | root: ET.Element, |
| 216 | ids_by_key: dict[str, str], |
| 217 | ) -> NativePayloadStats: |
| 218 | """Replace supported native-attribute groups with deterministic short ids.""" |
| 219 | stats = NativePayloadStats() |
| 220 | for element in root.iter(): |
| 221 | if NATIVE_RECORD_REF_ATTRIBUTE in element.attrib: |
| 222 | raise NativePayloadError( |
| 223 | "native metadata is already externalized; source attributes " |
| 224 | "must be hydrated before building a new store" |
| 225 | ) |
| 226 | attributes = _native_record_attributes(element) |
| 227 | if not attributes: |
| 228 | continue |
| 229 | record_id = ids_by_key.get(_native_record_key(attributes)) |
| 230 | if record_id is None: |
| 231 | raise NativePayloadError( |
| 232 | "Native-attribute record was not registered before externalization" |
| 233 | ) |
| 234 | for name in attributes: |
| 235 | element.attrib.pop(name) |
| 236 | element.set(NATIVE_RECORD_REF_ATTRIBUTE, record_id) |
| 237 | stats.native_record_count += 1 |
| 238 | stats.native_attribute_bytes += sum( |
| 239 | len(name) + len(value) + 4 |
| 240 | for name, value in attributes.items() |
| 241 | ) |
| 242 | return stats |
| 243 | |
| 244 | |
| 245 | def externalize_native_payloads( |
| 246 | root: ET.Element, |
| 247 | payloads: dict[str, bytes], |
| 248 | ) -> NativePayloadStats: |
| 249 | """Move supported large inline native payloads into ``payloads``. |
| 250 | |
| 251 | Repeated restoration attributes are handled separately by |
| 252 | ``externalize_native_attribute_records``. The supported opaque payload |
| 253 | classes are: |
| 254 | |
| 255 | - ``metadata[data-pptx-part="txbody"]`` |
| 256 | - ``data-pptx-shape-style`` |
| 257 | - ``data-pptx-custgeom`` |
| 258 | """ |
| 259 | stats = NativePayloadStats() |
| 260 | for element in root.iter(): |
| 261 | if ( |
| 262 | _local_name(element.tag) == "metadata" |
| 263 | and element.get("data-pptx-part") == "txbody" |
| 264 | ): |
| 265 | has_reference = TXBODY_REF_ATTRIBUTE in element.attrib |
| 266 | reference = element.get(TXBODY_REF_ATTRIBUTE) or "" |
| 267 | encoded = (element.text or "").strip() |
| 268 | encoding = element.get("data-pptx-encoding") |
| 269 | if has_reference: |
| 270 | if not reference: |
| 271 | raise NativePayloadError( |
| 272 | "txbody metadata has an empty payload reference" |
| 273 | ) |
| 274 | raise NativePayloadError( |
| 275 | "txbody metadata is already externalized; source payload " |
| 276 | "must be hydrated before building a new store" |
| 277 | ) |
| 278 | elif encoded: |
| 279 | if encoding != "base64": |
| 280 | raise NativePayloadError( |
| 281 | "txbody metadata must use base64 before externalization" |
| 282 | ) |
| 283 | raw = _decode_base64(encoded, context="txbody metadata") |
| 284 | element.text = None |
| 285 | element.attrib.pop("data-pptx-encoding", None) |
| 286 | element.set(TXBODY_REF_ATTRIBUTE, _register_payload(payloads, raw)) |
| 287 | stats.txbody_count += 1 |
| 288 | stats.inline_bytes += len(encoded) |
| 289 | else: |
| 290 | raise NativePayloadError( |
| 291 | "txbody metadata requires inline base64 data or a payload reference" |
| 292 | ) |
| 293 | |
| 294 | has_shape_reference = SHAPE_STYLE_REF_ATTRIBUTE in element.attrib |
| 295 | shape_reference = element.get(SHAPE_STYLE_REF_ATTRIBUTE) or "" |
| 296 | shape_encoded = element.get(SHAPE_STYLE_ATTRIBUTE) |
| 297 | if has_shape_reference and not shape_reference: |
| 298 | raise NativePayloadError("shape-style metadata has an empty payload reference") |
| 299 | if has_shape_reference: |
| 300 | raise NativePayloadError( |
| 301 | "shape-style metadata is already externalized; source payload " |
| 302 | "must be hydrated before building a new store" |
| 303 | ) |
| 304 | if shape_encoded: |
| 305 | raw = _decode_base64(shape_encoded, context="shape-style metadata") |
| 306 | element.attrib.pop(SHAPE_STYLE_ATTRIBUTE, None) |
| 307 | element.set( |
| 308 | SHAPE_STYLE_REF_ATTRIBUTE, |
| 309 | _register_payload(payloads, raw), |
| 310 | ) |
| 311 | stats.shape_style_count += 1 |
| 312 | stats.inline_bytes += len(shape_encoded) |
| 313 | |
| 314 | has_geometry_reference = CUSTOM_GEOMETRY_REF_ATTRIBUTE in element.attrib |
| 315 | geometry_reference = element.get(CUSTOM_GEOMETRY_REF_ATTRIBUTE) or "" |
| 316 | geometry_encoded = element.get(CUSTOM_GEOMETRY_ATTRIBUTE) |
| 317 | if has_geometry_reference and not geometry_reference: |
| 318 | raise NativePayloadError( |
| 319 | "custom-geometry metadata has an empty payload reference" |
| 320 | ) |
| 321 | if has_geometry_reference: |
| 322 | raise NativePayloadError( |
| 323 | "custom-geometry metadata is already externalized; source payload " |
| 324 | "must be hydrated before building a new store" |
| 325 | ) |
| 326 | if geometry_encoded: |
| 327 | raw = _decode_base64(geometry_encoded, context="custom-geometry metadata") |
| 328 | element.attrib.pop(CUSTOM_GEOMETRY_ATTRIBUTE, None) |
| 329 | element.set( |
| 330 | CUSTOM_GEOMETRY_REF_ATTRIBUTE, |
| 331 | _register_payload(payloads, raw), |
| 332 | ) |
| 333 | stats.custom_geometry_count += 1 |
| 334 | stats.inline_bytes += len(geometry_encoded) |
| 335 | return stats |
| 336 | |
| 337 | |
| 338 | def serialize_native_payload_store( |
| 339 | payloads: dict[str, bytes], |
| 340 | native_records: dict[str, dict[str, str]] | None = None, |
| 341 | ) -> bytes: |
| 342 | """Return one deterministic gzip-compressed native metadata store.""" |
| 343 | encoded_payloads: dict[str, str] = {} |
| 344 | for digest, raw in sorted(payloads.items()): |
| 345 | if _SHA256_RE.fullmatch(digest) is None or _sha256(raw) != digest: |
| 346 | raise NativePayloadError( |
| 347 | f"Native payload store key does not match its content: {digest!r}" |
| 348 | ) |
| 349 | encoded_payloads[digest] = base64.b64encode(raw).decode("ascii") |
| 350 | encoded_records: dict[str, dict[str, str]] = {} |
| 351 | for record_id, attributes in sorted((native_records or {}).items()): |
| 352 | if _NATIVE_RECORD_ID_RE.fullmatch(record_id) is None: |
| 353 | raise NativePayloadError( |
| 354 | f"Native metadata store contains an invalid record id: {record_id!r}" |
| 355 | ) |
| 356 | if not isinstance(attributes, dict) or not attributes: |
| 357 | raise NativePayloadError( |
| 358 | f"Native metadata record {record_id} must be a non-empty object" |
| 359 | ) |
| 360 | if any( |
| 361 | not isinstance(name, str) |
| 362 | or not isinstance(value, str) |
| 363 | or not _is_native_record_attribute(name) |
| 364 | for name, value in attributes.items() |
| 365 | ): |
| 366 | raise NativePayloadError( |
| 367 | f"Native metadata record {record_id} contains an unsupported field" |
| 368 | ) |
| 369 | encoded_records[record_id] = dict(sorted(attributes.items())) |
| 370 | document = { |
| 371 | "schema": PAYLOAD_STORE_SCHEMA, |
| 372 | "hash": "sha256", |
| 373 | "payloads": encoded_payloads, |
| 374 | "native_records": encoded_records, |
| 375 | } |
| 376 | raw_json = json.dumps( |
| 377 | document, |
| 378 | ensure_ascii=True, |
| 379 | separators=(",", ":"), |
| 380 | sort_keys=True, |
| 381 | ).encode("utf-8") |
| 382 | return gzip.compress(raw_json, compresslevel=9, mtime=0) |
| 383 | |
| 384 | |
| 385 | def _parse_reference(value: str) -> tuple[PurePosixPath, str]: |
| 386 | prefix = "project:" |
| 387 | marker = "#sha256:" |
| 388 | if not value.startswith(prefix) or marker not in value: |
| 389 | raise NativePayloadError( |
| 390 | f"Unsupported native payload reference: {value!r}" |
| 391 | ) |
| 392 | path_text, digest = value[len(prefix):].split(marker, 1) |
| 393 | relative = PurePosixPath(path_text) |
| 394 | if ( |
| 395 | not path_text |
| 396 | or relative.is_absolute() |
| 397 | or any(part in {"", ".", ".."} for part in relative.parts) |
| 398 | ): |
| 399 | raise NativePayloadError( |
| 400 | f"Native payload reference must use a safe project-relative path: {value!r}" |
| 401 | ) |
| 402 | if _SHA256_RE.fullmatch(digest) is None: |
| 403 | raise NativePayloadError( |
| 404 | f"Native payload reference has an invalid SHA-256 digest: {value!r}" |
| 405 | ) |
| 406 | return relative, digest |
| 407 | |
| 408 | |
| 409 | def _resolve_store_path(svg_path: Path, relative: PurePosixPath) -> Path: |
| 410 | start = Path(svg_path).expanduser().resolve().parent |
| 411 | relative_path = Path(*relative.parts) |
| 412 | for base in (start, *start.parents): |
| 413 | candidate = base / relative_path |
| 414 | if candidate.is_file(): |
| 415 | return candidate.resolve() |
| 416 | raise NativePayloadError( |
| 417 | f"Native payload store not found for {svg_path}: {relative.as_posix()}" |
| 418 | ) |
| 419 | |
| 420 | |
| 421 | def _load_store(path: Path) -> NativePayloadStore: |
| 422 | resolved = path.resolve() |
| 423 | try: |
| 424 | stat = resolved.stat() |
| 425 | except OSError as exc: |
| 426 | raise NativePayloadError( |
| 427 | f"Cannot inspect native payload store {resolved}: {exc}" |
| 428 | ) from exc |
| 429 | cached = _STORE_CACHE.get(resolved) |
| 430 | if cached and cached[0] == stat.st_mtime_ns and cached[1] == stat.st_size: |
| 431 | return cached[2] |
| 432 | |
| 433 | try: |
| 434 | document = json.loads(gzip.decompress(resolved.read_bytes()).decode("utf-8")) |
| 435 | except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: |
| 436 | raise NativePayloadError( |
| 437 | f"Cannot decode native payload store {resolved}: {exc}" |
| 438 | ) from exc |
| 439 | if not isinstance(document, dict) or document.get("schema") not in { |
| 440 | LEGACY_PAYLOAD_STORE_SCHEMA, |
| 441 | PAYLOAD_STORE_SCHEMA, |
| 442 | }: |
| 443 | raise NativePayloadError( |
| 444 | f"Unsupported native payload store schema in {resolved}" |
| 445 | ) |
| 446 | if document.get("hash") != "sha256": |
| 447 | raise NativePayloadError( |
| 448 | f"Unsupported native payload store hash algorithm in {resolved}" |
| 449 | ) |
| 450 | encoded_payloads = document.get("payloads") |
| 451 | if not isinstance(encoded_payloads, dict): |
| 452 | raise NativePayloadError( |
| 453 | f"Native payload store must contain a payload object: {resolved}" |
| 454 | ) |
| 455 | |
| 456 | payloads: dict[str, bytes] = {} |
| 457 | for digest, encoded in encoded_payloads.items(): |
| 458 | if not isinstance(digest, str) or _SHA256_RE.fullmatch(digest) is None: |
| 459 | raise NativePayloadError( |
| 460 | f"Native payload store contains an invalid digest key: {digest!r}" |
| 461 | ) |
| 462 | if not isinstance(encoded, str): |
| 463 | raise NativePayloadError( |
| 464 | f"Native payload {digest} must be a base64 string" |
| 465 | ) |
| 466 | raw = _decode_base64( |
| 467 | encoded, |
| 468 | context=f"native payload {digest} in {resolved}", |
| 469 | ) |
| 470 | if _sha256(raw) != digest: |
| 471 | raise NativePayloadError( |
| 472 | f"Native payload {digest} failed content-hash validation" |
| 473 | ) |
| 474 | payloads[digest] = raw |
| 475 | schema = document.get("schema") |
| 476 | if schema == PAYLOAD_STORE_SCHEMA and "native_records" not in document: |
| 477 | raise NativePayloadError( |
| 478 | f"Native payload store is missing native_records: {resolved}" |
| 479 | ) |
| 480 | native_records_value = document.get("native_records", {}) |
| 481 | if schema == PAYLOAD_STORE_SCHEMA and not isinstance(native_records_value, dict): |
| 482 | raise NativePayloadError( |
| 483 | f"Native payload store must contain a native_records object: {resolved}" |
| 484 | ) |
| 485 | if schema == LEGACY_PAYLOAD_STORE_SCHEMA: |
| 486 | native_records_value = {} |
| 487 | |
| 488 | native_records: dict[str, dict[str, str]] = {} |
| 489 | for record_id, attributes in native_records_value.items(): |
| 490 | if ( |
| 491 | not isinstance(record_id, str) |
| 492 | or _NATIVE_RECORD_ID_RE.fullmatch(record_id) is None |
| 493 | ): |
| 494 | raise NativePayloadError( |
| 495 | f"Native payload store contains an invalid record id: {record_id!r}" |
| 496 | ) |
| 497 | if not isinstance(attributes, dict) or not attributes: |
| 498 | raise NativePayloadError( |
| 499 | f"Native metadata record {record_id} must be a non-empty object" |
| 500 | ) |
| 501 | validated: dict[str, str] = {} |
| 502 | for name, value in attributes.items(): |
| 503 | if ( |
| 504 | not isinstance(name, str) |
| 505 | or not isinstance(value, str) |
| 506 | or not _is_native_record_attribute(name) |
| 507 | ): |
| 508 | raise NativePayloadError( |
| 509 | f"Native metadata record {record_id} contains an unsupported field" |
| 510 | ) |
| 511 | validated[name] = value |
| 512 | native_records[record_id] = validated |
| 513 | |
| 514 | store = NativePayloadStore( |
| 515 | payloads=payloads, |
| 516 | native_records=native_records, |
| 517 | ) |
| 518 | _STORE_CACHE[resolved] = (stat.st_mtime_ns, stat.st_size, store) |
| 519 | return store |
| 520 | |
| 521 | |
| 522 | def _payload_for_reference(value: str, svg_path: Path) -> bytes: |
| 523 | relative, digest = _parse_reference(value) |
| 524 | store_path = _resolve_store_path(svg_path, relative) |
| 525 | payload = _load_store(store_path).payloads.get(digest) |
| 526 | if payload is None: |
| 527 | raise NativePayloadError( |
| 528 | f"Native payload {digest} is missing from {store_path}" |
| 529 | ) |
| 530 | return payload |
| 531 | |
| 532 | |
| 533 | def _native_record_for_reference( |
| 534 | value: str, |
| 535 | svg_path: Path, |
| 536 | ) -> dict[str, str]: |
| 537 | if _NATIVE_RECORD_ID_RE.fullmatch(value) is None: |
| 538 | raise NativePayloadError( |
| 539 | f"Native metadata reference has an invalid record id: {value!r}" |
| 540 | ) |
| 541 | store_path = _resolve_store_path( |
| 542 | svg_path, |
| 543 | PurePosixPath(PAYLOAD_STORE_RELATIVE_PATH.as_posix()), |
| 544 | ) |
| 545 | record = _load_store(store_path).native_records.get(value) |
| 546 | if record is None: |
| 547 | raise NativePayloadError( |
| 548 | f"Native metadata record {value} is missing from {store_path}" |
| 549 | ) |
| 550 | return record |
| 551 | |
| 552 | |
| 553 | def hydrate_native_payload_refs(root: ET.Element, svg_path: Path) -> int: |
| 554 | """Restore compact native records and payloads as legacy inline metadata. |
| 555 | |
| 556 | The operation preflights every reference before mutating the tree, so an |
| 557 | invalid store leaves the caller's parsed SVG unchanged. |
| 558 | """ |
| 559 | record_operations: list[tuple[ET.Element, dict[str, str]]] = [] |
| 560 | payload_operations: list[tuple[str, ET.Element, bytes]] = [] |
| 561 | for element in root.iter(): |
| 562 | effective_attributes = dict(element.attrib) |
| 563 | has_native_record = NATIVE_RECORD_REF_ATTRIBUTE in element.attrib |
| 564 | native_record_ref = element.get(NATIVE_RECORD_REF_ATTRIBUTE) or "" |
| 565 | if has_native_record: |
| 566 | if not native_record_ref: |
| 567 | raise NativePayloadError( |
| 568 | "native metadata has an empty record reference" |
| 569 | ) |
| 570 | record = _native_record_for_reference(native_record_ref, svg_path) |
| 571 | conflicts = sorted(set(record) & set(element.attrib)) |
| 572 | if conflicts: |
| 573 | raise NativePayloadError( |
| 574 | "native metadata cannot carry both inline fields and a record: " |
| 575 | + ", ".join(conflicts) |
| 576 | ) |
| 577 | effective_attributes.update(record) |
| 578 | record_operations.append((element, record)) |
| 579 | |
| 580 | if ( |
| 581 | _local_name(element.tag) == "metadata" |
| 582 | and effective_attributes.get("data-pptx-part") == "txbody" |
| 583 | ): |
| 584 | has_reference = TXBODY_REF_ATTRIBUTE in effective_attributes |
| 585 | reference = effective_attributes.get(TXBODY_REF_ATTRIBUTE) or "" |
| 586 | if has_reference: |
| 587 | if not reference: |
| 588 | raise NativePayloadError( |
| 589 | "txbody metadata has an empty payload reference" |
| 590 | ) |
| 591 | if (element.text or "").strip() or element.get("data-pptx-encoding"): |
| 592 | raise NativePayloadError( |
| 593 | "txbody metadata cannot carry both inline data and a reference" |
| 594 | ) |
| 595 | payload_operations.append( |
| 596 | ("txbody", element, _payload_for_reference(reference, svg_path)) |
| 597 | ) |
| 598 | |
| 599 | has_shape_reference = SHAPE_STYLE_REF_ATTRIBUTE in effective_attributes |
| 600 | shape_reference = effective_attributes.get(SHAPE_STYLE_REF_ATTRIBUTE) or "" |
| 601 | if has_shape_reference: |
| 602 | if not shape_reference: |
| 603 | raise NativePayloadError( |
| 604 | "shape-style metadata has an empty payload reference" |
| 605 | ) |
| 606 | if effective_attributes.get(SHAPE_STYLE_ATTRIBUTE): |
| 607 | raise NativePayloadError( |
| 608 | "shape-style metadata cannot carry both inline data and a reference" |
| 609 | ) |
| 610 | payload_operations.append( |
| 611 | ( |
| 612 | "shape-style", |
| 613 | element, |
| 614 | _payload_for_reference(shape_reference, svg_path), |
| 615 | ) |
| 616 | ) |
| 617 | |
| 618 | has_geometry_reference = CUSTOM_GEOMETRY_REF_ATTRIBUTE in effective_attributes |
| 619 | geometry_reference = ( |
| 620 | effective_attributes.get(CUSTOM_GEOMETRY_REF_ATTRIBUTE) or "" |
| 621 | ) |
| 622 | if has_geometry_reference: |
| 623 | if not geometry_reference: |
| 624 | raise NativePayloadError( |
| 625 | "custom-geometry metadata has an empty payload reference" |
| 626 | ) |
| 627 | if effective_attributes.get(CUSTOM_GEOMETRY_ATTRIBUTE): |
| 628 | raise NativePayloadError( |
| 629 | "custom-geometry metadata cannot carry both inline data and a reference" |
| 630 | ) |
| 631 | payload_operations.append( |
| 632 | ( |
| 633 | "custom-geometry", |
| 634 | element, |
| 635 | _payload_for_reference(geometry_reference, svg_path), |
| 636 | ) |
| 637 | ) |
| 638 | |
| 639 | for element, record in record_operations: |
| 640 | element.attrib.pop(NATIVE_RECORD_REF_ATTRIBUTE) |
| 641 | element.attrib.update(record) |
| 642 | |
| 643 | for kind, element, raw in payload_operations: |
| 644 | encoded = base64.b64encode(raw).decode("ascii") |
| 645 | if kind == "txbody": |
| 646 | element.text = encoded |
| 647 | element.set("data-pptx-encoding", "base64") |
| 648 | element.attrib.pop(TXBODY_REF_ATTRIBUTE, None) |
| 649 | elif kind == "shape-style": |
| 650 | element.set(SHAPE_STYLE_ATTRIBUTE, encoded) |
| 651 | element.attrib.pop(SHAPE_STYLE_REF_ATTRIBUTE, None) |
| 652 | else: |
| 653 | element.set(CUSTOM_GEOMETRY_ATTRIBUTE, encoded) |
| 654 | element.attrib.pop(CUSTOM_GEOMETRY_REF_ATTRIBUTE, None) |
| 655 | return len(record_operations) + len(payload_operations) |
| 656 |