diff --git a/pyproject.toml b/pyproject.toml index b63deccd..1030e084 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "xgen-edit2docs" -version = "0.23.0" +version = "0.24.0" description = "AI-agent-native document engine: generate and chat-edit DOCX, XLSX and PPTX as a Python library, agent tool set, MCP server or hosted service. English-first with first-class Korean support. Sister project of edit2ppt." readme = "README.md" requires-python = ">=3.12" @@ -31,17 +31,12 @@ dependencies = [ # Core engine dependencies (from ppt-master) "python-pptx>=0.6.21", "python-docx>=1.1.0", - "PyMuPDF>=1.23.0", + # PDF engine: pdfium-based xgen-pdf (BSD/Apache). PyMuPDF (AGPL) is gone. + "xgen-pdf @ https://github.com/PlateerLab/xgen-pdf/releases/download/v0.1.1/xgen_pdf-0.1.1-py3-none-any.whl", "mammoth>=1.6.0", "openpyxl>=3.1.0", - # Lossless raw OOXML layer (xgen_contextifier.open_raw): surgical xlsx/docx - # edits and PPTX native-content preservation that keep charts, - # sparklines, custom XML, cached formula values and native - # charts/tables byte-identical through an edit. - # ⚠ 내부 패키지는 **URL 직접참조**로 고정한다. 우리 패키지는 어느 인덱스에도 - # 없으므로 이름으로만 요구하면 `pip install -e .` 이 해석에 실패한다 — - # 실제로 이 저장소는 개발 설치조차 되지 않는 상태였다. - "xgen-contextifier @ https://github.com/PlateerLab/xgen-contextifier/releases/download/v0.9.0/xgen_contextifier-0.9.0-py3-none-any.whl", + # Lossless raw OOXML layer: vendored in xgen_edit2docs.raw (from xgen-contextifier 0.9.0), + # so no dependency on that package (which would pull in PyMuPDF). # Native chart export (data-pptx-native markers) generates the chart's # embedded XLSX workbook with XlsxWriter (upstream 53c6cad5). "XlsxWriter>=3.0", @@ -49,6 +44,8 @@ dependencies = [ "markdownify>=0.11.6", "ebooklib>=0.18", "beautifulsoup4>=4.12.0", + # HWP/OLE containers (was an indirect dependency via xgen-contextifier). + "olefile>=0.47", "requests>=2.31.0", "Pillow>=9.0.0", "numpy>=1.20.0", diff --git a/src/xgen_edit2docs/__init__.py b/src/xgen_edit2docs/__init__.py index 7b85b29c..1bff1801 100644 --- a/src/xgen_edit2docs/__init__.py +++ b/src/xgen_edit2docs/__init__.py @@ -19,7 +19,7 @@ import importlib from typing import Any -__version__ = "0.18.0" +__version__ = "0.24.0" _LAZY: dict[str, str] = { # Unified, extension-dispatched verbs (docx / xlsx / pptx) diff --git a/src/xgen_edit2docs/core/docs/conversion.md b/src/xgen_edit2docs/core/docs/conversion.md index e795158a..be346dd4 100644 --- a/src/xgen_edit2docs/core/docs/conversion.md +++ b/src/xgen_edit2docs/core/docs/conversion.md @@ -33,7 +33,7 @@ Prefer MinerU or another OCR/layout tool when: Dependency: ```bash -pip install PyMuPDF +pip install xgen-pdf ``` ## `source_to_md/doc_to_md.py` diff --git a/src/xgen_edit2docs/core/docs/troubleshooting.md b/src/xgen_edit2docs/core/docs/troubleshooting.md index f53ef502..54ff821c 100644 --- a/src/xgen_edit2docs/core/docs/troubleshooting.md +++ b/src/xgen_edit2docs/core/docs/troubleshooting.md @@ -76,5 +76,5 @@ Important optional packages: - `edge-tts` for `notes_to_audio.py` recorded narration audio - `Pillow` for image utilities - `numpy` for watermark removal -- `PyMuPDF` for PDF conversion +- `xgen-pdf` for PDF conversion - `google-genai` / `openai` for image generation backends diff --git a/src/xgen_edit2docs/core/source_to_md/pdf_to_md.py b/src/xgen_edit2docs/core/source_to_md/pdf_to_md.py index c24377c4..adbb15d8 100644 --- a/src/xgen_edit2docs/core/source_to_md/pdf_to_md.py +++ b/src/xgen_edit2docs/core/source_to_md/pdf_to_md.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 """ PDF to Markdown Converter -Uses PyMuPDF to extract PDF text content and convert to Markdown format. +Uses xgen-pdf (pdfium) to extract PDF text content and convert to Markdown format. Supports heading levels, bold, italic, and list detection. """ @@ -14,9 +14,9 @@ from collections import Counter try: - import fitz # PyMuPDF + import xgen_pdf as fitz # pdfium-based engine (xgen-pdf) except ImportError: - print("[ERROR] PyMuPDF not installed. Run: pip install PyMuPDF", file=sys.stderr) + print("[ERROR] xgen-pdf not installed. Run: pip install xgen-pdf", file=sys.stderr) sys.exit(1) FONT_BODY_SIZE = 12 @@ -356,7 +356,7 @@ def should_keep_image( """Filter out small, decorative, or duplicate images. Args: - block: Image block extracted from PyMuPDF. + block: Image block from ``page.get_text("dict")``. page_rect: Current page rectangle. seen_hashes: Optional set used to deduplicate image payloads. @@ -618,7 +618,7 @@ def clean_text(text: str) -> str: def merge_adjacent_formatting(text: str) -> str: """Merge adjacent same-style formatted spans split across PDF tokens. - PyMuPDF often emits a phrase as several spans, so per-span wrapping in + The text engine may emit a phrase as several spans, so per-span wrapping in ``format_span_text`` produces ``**X****Y**`` (bold) or ``***X******Y***`` (bold-italic) where one phrase is intended. Collapse the abutting markers so the run reads as a single phrase: ``**X Y**`` / ``***X Y***``. diff --git a/src/xgen_edit2docs/documents/arrange.py b/src/xgen_edit2docs/documents/arrange.py index d9cc4bcd..74ee3f33 100644 --- a/src/xgen_edit2docs/documents/arrange.py +++ b/src/xgen_edit2docs/documents/arrange.py @@ -42,7 +42,7 @@ def apply_arrange( per op ``{op, target, to?, name?, status, message}``; ``warnings`` is a list of ``{code, message}`` (e.g. a rename that leaves formula references dangling).""" - from xgen_contextifier import open_raw + from xgen_edit2docs.raw import open_raw results: list[dict] = [] warnings: list[dict] = [] diff --git a/src/xgen_edit2docs/documents/chart_edit.py b/src/xgen_edit2docs/documents/chart_edit.py index 08d6c7b3..5fd8b311 100644 --- a/src/xgen_edit2docs/documents/chart_edit.py +++ b/src/xgen_edit2docs/documents/chart_edit.py @@ -73,7 +73,7 @@ def list_charts(content: bytes, fmt: str) -> list[dict]: the address source for :func:`apply_chart_edits`. """ try: - from xgen_contextifier import open_raw + from xgen_edit2docs.raw import open_raw raw = open_raw(content, extension=fmt) out: list[dict] = [] @@ -110,8 +110,8 @@ def apply_chart_edits( deterministic editors. The package is only re-serialized when at least one edit applied; untouched parts stay byte-identical. """ - from xgen_contextifier import open_raw - from xgen_contextifier.raw.opc import RawUnsupportedError + from xgen_edit2docs.raw import open_raw + from xgen_edit2docs.raw.opc import RawUnsupportedError raw = open_raw(content, extension=fmt) charts = _charts_of(raw, fmt) diff --git a/src/xgen_edit2docs/documents/docx_engine.py b/src/xgen_edit2docs/documents/docx_engine.py index 26a28520..aa62226f 100644 --- a/src/xgen_edit2docs/documents/docx_engine.py +++ b/src/xgen_edit2docs/documents/docx_engine.py @@ -318,7 +318,7 @@ def _chart_outline(content: bytes) -> list[dict]: """Read-only chart summaries via xgen_contextifier (best-effort: outline must never fail because a chart part is exotic).""" try: - from xgen_contextifier import open_raw + from xgen_edit2docs.raw import open_raw raw = open_raw(content, extension="docx") return [ @@ -379,7 +379,7 @@ def apply_docx_edits(content: bytes, edits: Iterable[DocxEdit]) -> tuple[bytes, Per-edit soft failures (like the PPTX text editor): ``old_text`` guards replaces with a whitespace-normalized comparison. """ - from xgen_contextifier import open_raw + from xgen_edit2docs.raw import open_raw try: raw = open_raw(content, extension="docx") @@ -477,7 +477,7 @@ def _fragment_blocks(fragment: bytes) -> list: grafting this replaces).""" import zipfile - from xgen_contextifier.raw import qn + from xgen_edit2docs.raw import qn from lxml import etree with zipfile.ZipFile(io.BytesIO(fragment)) as zf: diff --git a/src/xgen_edit2docs/documents/docx_pages.py b/src/xgen_edit2docs/documents/docx_pages.py index 16d399ec..262f4667 100644 --- a/src/xgen_edit2docs/documents/docx_pages.py +++ b/src/xgen_edit2docs/documents/docx_pages.py @@ -5,7 +5,7 @@ line wrap (``xgen_edit2docs.render.fonts`` — the same fonts resvg rasterizes with), lays out tables/images/headers/footers, and emits one self-contained SVG per page. ``render_doc`` feeds these to the -resvg/PyMuPDF raster layer for PNG/PDF — the piece LibreOffice used to +resvg/xgen-pdf raster layer for PNG/PDF — the piece LibreOffice used to provide. Fidelity scope (deliberate): body paragraphs (runs with bold/italic/ diff --git a/src/xgen_edit2docs/documents/xlsx_engine.py b/src/xgen_edit2docs/documents/xlsx_engine.py index 0e719f22..96d9b970 100644 --- a/src/xgen_edit2docs/documents/xlsx_engine.py +++ b/src/xgen_edit2docs/documents/xlsx_engine.py @@ -187,7 +187,7 @@ def _chart_outline(content: bytes) -> list[dict]: """Read-only chart summaries via xgen_contextifier (best-effort: outline must never fail because a chart part is exotic).""" try: - from xgen_contextifier import open_raw + from xgen_edit2docs.raw import open_raw raw = open_raw(content, extension="xlsx") return [ @@ -266,7 +266,7 @@ def apply_xlsx_edits(content: bytes, edits: Iterable[XlsxEdit]) -> tuple[bytes, formula values all survive byte-identical (the old openpyxl load→save round-trip destroyed every one of those on EVERY edit). """ - from xgen_contextifier import open_raw + from xgen_edit2docs.raw import open_raw try: raw = open_raw(content, extension="xlsx") @@ -369,7 +369,7 @@ def _add_raw_sheet(raw, title: str) -> None: rels entry, and the ```` element in ``xl/workbook.xml``. Everything else in the package stays byte-identical. """ - from xgen_contextifier.raw import qn + from xgen_edit2docs.raw import qn package = raw.package n = 1 diff --git a/src/xgen_edit2docs/documents/xml_edit.py b/src/xgen_edit2docs/documents/xml_edit.py index 9a7c50bb..e8100fe5 100644 --- a/src/xgen_edit2docs/documents/xml_edit.py +++ b/src/xgen_edit2docs/documents/xml_edit.py @@ -51,7 +51,7 @@ class XmlEditResult: def _open_package(content: bytes): - from xgen_contextifier.raw.opc import OpcPackage + from xgen_edit2docs.raw.opc import OpcPackage return OpcPackage.open(content) diff --git a/src/xgen_edit2docs/raw/__init__.py b/src/xgen_edit2docs/raw/__init__.py new file mode 100644 index 00000000..85d6904d --- /dev/null +++ b/src/xgen_edit2docs/raw/__init__.py @@ -0,0 +1,93 @@ +# xgen_contextifier/raw +""" +Raw document access — the lossless twin of the extraction pipeline. + +Contextifier has two ways to look at a document: + +* ``DocumentProcessor.extract_text()`` / ``.process()`` — the existing + pipeline that renders an **AI-friendly** view (clean text, normalized + tables/charts) and throws the rest away. +* ``open_raw()`` (this package) — a **lossless, addressable, writable** + view of the same file. Nothing is discarded: every OPC part stays + available, XML is parsed lazily, and edits are *surgical* — when you + save, untouched parts are written back **byte-identical** (the + byte-preservation contract), so charts, pivot tables, sparklines, + custom XML, styles and anything else the higher-level libraries can't + model all survive. + +Usage:: + + from xgen_edit2docs.raw import open_raw + + raw = open_raw("report.xlsx") # XlsxRawDocument + raw.sheets["Sales"].set_cell("B3", 142) + raw.charts[0].set_data(categories=["Q1", "Q2"], series=[("Sales", [1, 2])]) + raw.save("report-edited.xlsx") # or raw.to_bytes() + + raw = open_raw("deck.pptx") # PptxRawDocument + raw = open_raw("paper.docx") # DocxRawDocument + +Every format model also exposes ``.package`` (the raw +:class:`~xgen_edit2docs.raw.opc.OpcPackage`) for part-level work, so the +"easy" interface never locks you out of the full container. + +Supported today: the OOXML trio (.xlsx / .docx / .pptx). Other handlers +raise :class:`RawUnsupportedError`. +""" + +from __future__ import annotations + +from xgen_edit2docs.raw.opc import OpcPackage, OpcPart, RawUnsupportedError +from xgen_edit2docs.raw.xmlpart import NS, XmlPart, qn + +__all__ = [ + "OpcPackage", + "OpcPart", + "RawUnsupportedError", + "XmlPart", + "NS", + "qn", + "open_raw", +] + + +def open_raw(source, *, extension: str | None = None): + """Open a document for lossless, writable access. + + Args: + source: path (str/Path), bytes, or a binary file object. + extension: override the format sniff (e.g. ``"xlsx"``); by default + the file extension (for paths) or the package content is used. + + Returns: + ``XlsxRawDocument`` / ``DocxRawDocument`` / ``PptxRawDocument``. + + Raises: + RawUnsupportedError: format has no raw model yet. + """ + from pathlib import Path + + ext = (extension or "").lower().lstrip(".") + if not ext and isinstance(source, (str, Path)): + ext = Path(source).suffix.lower().lstrip(".") + + package = OpcPackage.open(source) + if not ext: + ext = package.sniff_format() or "" + + if ext == "xlsx": + from xgen_edit2docs.raw.xlsx import XlsxRawDocument + + return XlsxRawDocument(package) + if ext == "docx": + from xgen_edit2docs.raw.docx import DocxRawDocument + + return DocxRawDocument(package) + if ext == "pptx": + from xgen_edit2docs.raw.pptx import PptxRawDocument + + return PptxRawDocument(package) + raise RawUnsupportedError( + f"No raw model for {ext or 'unknown format'!r} yet " + "(supported: xlsx, docx, pptx)" + ) diff --git a/src/xgen_edit2docs/raw/base.py b/src/xgen_edit2docs/raw/base.py new file mode 100644 index 00000000..c3db0c1c --- /dev/null +++ b/src/xgen_edit2docs/raw/base.py @@ -0,0 +1,172 @@ +# xgen_edit2docs/raw/base.py (vendored from xgen-contextifier 0.9.0, Apache-2.0) +""" +Shared base for format raw-document models (xlsx / docx / pptx). + +A format model owns a set of :class:`~xgen_edit2docs.raw.xmlpart.XmlPart` +facades over the parts it understands. ``save()`` flushes every dirty +facade into the package, then serializes the package under the +byte-preservation contract. Parts the model does NOT understand are +never touched at all. +""" + +from __future__ import annotations + +import posixpath +from pathlib import Path +from typing import BinaryIO + +from xgen_edit2docs.raw.opc import OpcPackage, OpcPart +from xgen_edit2docs.raw.xmlpart import XmlPart, qn + +__all__ = ["RawDocumentBase"] + + +class RawDocumentBase: + """Common plumbing: part registry, flush-on-save, byte export.""" + + #: subclasses set this ("xlsx" / "docx" / "pptx") + format: str = "" + + def __init__(self, package: OpcPackage): + self.package = package + self._xml_parts: dict[str, XmlPart] = {} + + # -- part facades ------------------------------------------------------------ + + def xml_part(self, name: str) -> XmlPart: + """The (cached) XmlPart facade for a package part.""" + if name not in self._xml_parts: + self._xml_parts[name] = XmlPart(self.package.get_part(name)) + return self._xml_parts[name] + + def raw_part(self, name: str) -> OpcPart: + """Direct part access — the escape hatch for anything the model + doesn't cover.""" + return self.package.get_part(name) + + # -- persistence ---------------------------------------------------------- + + def flush(self) -> None: + """Serialize every dirty XML facade into its package part.""" + for xp in self._xml_parts.values(): + xp.flush() + + def to_bytes(self) -> bytes: + self.flush() + return self.package.to_bytes() + + def save(self, target: str | Path | BinaryIO | None = None) -> bytes: + """Write the package; untouched parts stay byte-identical.""" + self.flush() + return self.package.save(target) + + def close(self) -> None: + self.package.close() + + def __enter__(self): + return self + + def __exit__(self, *exc) -> None: + self.close() + + # -- part removal & orphan sweep ------------------------------------------- + # Shared by PptxRawDocument.remove_slide and XlsxRawDocument.delete_sheet: + # a structural delete drops one anchor part, then reference-counts every + # part it transitively pulled in against everything still in the package + # and deletes the now-orphaned ones (charts, embedded workbooks, images, + # notes …), leaving surviving parts byte-identical. + + def _delete_part(self, name: str) -> list[str]: + """Remove *name* and its ``.rels`` part; returns what was removed.""" + removed: list[str] = [] + if self.package.has_part(name): + self.package.remove_part(name) + removed.append(name) + rels_name = OpcPackage._rels_name_for(name) + if self.package.has_part(rels_name): + self.package.remove_part(rels_name) + removed.append(rels_name) + self._xml_parts.pop(name, None) + self._xml_parts.pop(rels_name, None) + return removed + + def _referenced_parts(self) -> set[str]: + """Internal targets of every relationships part still present.""" + referenced: set[str] = set() + for rels_name in list(self.package.part_names): + if not rels_name.endswith(".rels"): + continue + directory, base = posixpath.split(rels_name) + owner_dir = posixpath.dirname(directory) + owner_base = base[: -len(".rels")] + owner = posixpath.join(owner_dir, owner_base) if owner_base else "" + rels = self.package.rels_for(owner) + if rels is None: + continue + for rel in rels: + if rel["mode"] == "External": + continue + referenced.add(rels.resolve(owner, rel["target"])) + return referenced + + def _drop_content_type_overrides(self, part_names: list[str]) -> None: + from lxml import etree + + ct_part = self.package.get_part("[Content_Types].xml") + root = etree.fromstring(ct_part.read()) + doomed = {f"/{name}" for name in part_names} + changed = False + for el in list(root): + if el.tag == qn("ct:Override") and el.get("PartName") in doomed: + root.remove(el) + changed = True + if changed: + ct_part.write( + etree.tostring( + root, xml_declaration=True, encoding="UTF-8", standalone=True + ) + ) + + def _sweep_orphans(self, doomed: list[str]) -> list[str]: + """Delete *doomed* parts, then every part only they referenced. + + Candidates (everything transitively reachable from *doomed*) are + collected while their rels still exist; after the doomed parts go, + any candidate no surviving relationship points at is deleted, to a + fixpoint (removing a chart un-anchors its embedded workbook, …). + Returns every part name removed (for content-type cleanup).""" + candidates: set[str] = set() + stack, visited = list(doomed), set(doomed) + while stack: + src = stack.pop() + rels = self.package.rels_for(src) + if rels is None: + continue + for rel in rels: + if rel["mode"] == "External": + continue + target = rels.resolve(src, rel["target"]) + if target not in visited: + visited.add(target) + candidates.add(target) + stack.append(target) + + removed: list[str] = [] + for name in doomed: + removed += self._delete_part(name) + + while True: + referenced = self._referenced_parts() + orphans = [ + c + for c in sorted(candidates) + if self.package.has_part(c) and c not in referenced + ] + if not orphans: + break + for name in orphans: + removed += self._delete_part(name) + candidates.discard(name) + + self._drop_content_type_overrides(removed) + return removed diff --git a/src/xgen_edit2docs/raw/chart.py b/src/xgen_edit2docs/raw/chart.py new file mode 100644 index 00000000..311fc139 --- /dev/null +++ b/src/xgen_edit2docs/raw/chart.py @@ -0,0 +1,771 @@ +# xgen_edit2docs/raw/chart.py (vendored from xgen-contextifier 0.9.0, Apache-2.0) +""" +ChartModel — read & write DrawingML charts, shared across all three +OOXML formats. + +A chart is a ``chartN.xml`` part (classic ``c:`` namespace, or ``cx:`` +chartEx for the 2016+ types) referenced from a sheet drawing (xlsx), a +document drawing (docx) or a slide graphicFrame (pptx), usually paired +with an embedded ``.xlsx`` workbook holding the source table. This +module is format-agnostic: it only needs the chart part and the +package, so the three format models can share one implementation. + +Reading:: + + chart = raw.charts[0] + chart.kind # "bar" | "line" | "pie" | ... | "chartex:" + chart.title # str | None + chart.series # [ChartSeriesData(name, categories, values), ...] + +Writing:: + + chart.set_title("Q3 Sales") + chart.set_data( + categories=["Q1", "Q2", "Q3"], + series=[("Sales", [120, 135, 150]), ("Cost", [80, 90, 95])], + ) + raw.save("out.xlsx") + +``set_data`` must rewrite BOTH the caches inside the chart XML +(``c:cat/c:strRef/c:strCache``, ``c:val/c:numRef/c:numCache`` per +series — creating/removing ``c:ser`` elements as the series count +changes) AND the embedded workbook part (if present) so that +double-click-edit in Office shows the same numbers. Formula references +(``c:f``) should be regenerated against the embedded workbook's sheet +("Sheet1!$B$2:$B$4" style). For xlsx-hosted charts whose series +reference the HOST workbook's own cells, ``set_data`` rewrites caches +and leaves the ``c:f`` references pointing at the host sheet (values +there are the caller's responsibility — typically edited through +``sheet.set_cell`` alongside). + +Embedded workbook regeneration +------------------------------ +When the chart has an embedded ``.../embeddings/*.xlsx`` workbook (the +usual case for pptx/docx-hosted charts), ``set_data`` REGENERATES that +part from scratch with openpyxl: a single sheet holding the plain data +table (header row ``[None, name1, name2, ...]`` followed by one +``[category, v1, v2, ...]`` row per category). The embedded workbook is +chart-internal source data, not user content, so replacing it wholesale +is safe — it is exactly what keeps Office's "Edit Data" view in sync +with the rewritten caches. + +chartEx (cx:) limitations in this milestone +------------------------------------------- +chartEx charts are fully **readable** (``kind`` / ``title`` / +``series``), but ``set_title`` and ``set_data`` raise +:class:`~xgen_edit2docs.raw.opc.RawUnsupportedError` for them — the cx: +write path is deferred (documented limitation, see the class contract). +""" + +from __future__ import annotations + +import copy +import io +import posixpath +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Sequence + +from lxml import etree + +from xgen_edit2docs.raw.opc import RawUnsupportedError +from xgen_edit2docs.raw.xmlpart import NS, XmlPart, qn + +if TYPE_CHECKING: # pragma: no cover + from lxml.etree import _Element + + from xgen_edit2docs.raw.opc import OpcPackage, OpcPart + +__all__ = ["ChartSeriesData", "ChartModel", "find_chart_parts", "load_chart"] + + +@dataclass +class ChartSeriesData: + """One series as read from (or written to) the chart caches.""" + + name: str | None + categories: list[str] = field(default_factory=list) + values: list[float | None] = field(default_factory=list) + + +# Classic plot-type element (local name) → kind. bar/bar3D are handled +# separately because they split on c:barDir ("bar" vs "column"). +_PLOT_KIND: dict[str, str] = { + "lineChart": "line", + "line3DChart": "line", + "pieChart": "pie", + "pie3DChart": "pie", + "areaChart": "area", + "area3DChart": "area", + "scatterChart": "scatter", + "doughnutChart": "doughnut", + "radarChart": "radar", + "bubbleChart": "bubble", + "ofPieChart": "of_pie", + "surfaceChart": "surface", + "surface3DChart": "surface", + "stockChart": "stock", +} + +# Plot types whose series carry c:xVal/c:yVal instead of c:cat/c:val. +_XY_PLOTS = {"scatterChart", "bubbleChart"} + +# c:ser children that must stay AFTER c:cat/c:val (schema tail). +_SER_TAIL_TAGS = ("c:smooth", "c:shape", "c:bubbleSize", "c:bubble3D", "c:extLst") + + +def _local(el: "_Element") -> str: + """Local (namespace-stripped) tag name; '' for comments/PIs.""" + tag = el.tag + if not isinstance(tag, str): + return "" + return tag.rsplit("}", 1)[-1] + + +def _col_letter(n: int) -> str: + """1-based column index → spreadsheet letters (1→A, 27→AA).""" + letters = "" + while n > 0: + n, rem = divmod(n - 1, 26) + letters = chr(ord("A") + rem) + letters + return letters + + +def _num_text(value: float) -> str: + """Float/int → the shortest cache text Excel reads back exactly.""" + f = float(value) + if f.is_integer(): + return str(int(f)) + return repr(f) + + +def _unquote_sheet(ref_prefix: str) -> str: + """``"'My Sheet'"`` → ``"My Sheet"`` (c:f quoting → real sheet name).""" + if ref_prefix.startswith("'") and ref_prefix.endswith("'"): + return ref_prefix[1:-1].replace("''", "'") + return ref_prefix + + +class ChartModel: + """Read/write view over one chart part. + + Contract (implemented in this module; consumed by the xlsx/docx/pptx + models — do not change signatures without updating all three): + + * ``ChartModel(xml_part, package)`` — *xml_part* is the chart part + facade; *package* is the owning :class:`OpcPackage` (used to reach + the embedded workbook through the chart part's rels). + * ``kind: str`` (property) — first plot type found: ``"bar"``, + ``"line"``, ``"pie"``, ``"area"``, ``"scatter"``, ``"doughnut"``, + ``"radar"``, ``"bubble"``, ... or ``"chartex:funnel"`` etc. for + cx: charts. + * ``title: str | None`` (property) — concatenated ``a:t`` runs of the + chart title, if any. + * ``series: list[ChartSeriesData]`` (property) — parsed from + str/num caches (classic) or ``cx:strDim``/``cx:numDim`` (chartEx). + Missing cache points yield ``None`` values. + * ``set_title(text: str) -> None`` — replace/insert the title text, + preserving existing run formatting where present. + * ``set_data(categories, series) -> None`` — see module docstring. + ``series`` accepts ``[(name, values), ...]`` or + ``[ChartSeriesData, ...]``. Raises ``ValueError`` on ragged input + (series length != len(categories)). chartEx write support may + raise ``RawUnsupportedError`` (documented limitation) in v0.4. + * ``embedded_workbook_part() -> OpcPart | None`` — the + ``.../embeddings/*.xlsx`` part referenced by this chart's rels. + + Implementation notes (behavior, not signatures): + + * Parsing is lazy — ``__init__`` stores the facade; the tree is + first parsed on the first property access. + * Mutators (``set_title`` / ``set_data``) mark the XML facade dirty + AND flush it immediately, so ``package.to_bytes()`` alone reflects + the edit; format-model ``save()`` flushing stays a no-op for it. + * chartEx: read-only in this milestone — ``set_title`` and + ``set_data`` raise :class:`RawUnsupportedError` for cx: charts. + * Charts without value caches (e.g. openpyxl-authored charts, which + write only ``c:f`` references into the host workbook) read as + series with ``name=None``, empty categories and ``values`` of + ``None`` per cached ``ptCount`` (or empty when no cache exists). + """ + + def __init__(self, xml_part: "XmlPart", package: "OpcPackage"): + self.xml = xml_part + self.package = package + + # -- detection --------------------------------------------------------------- + + @property + def _is_chartex(self) -> bool: + return self.xml.root.tag == qn("cx:chartSpace") + + def _plot_elements(self) -> list["_Element"]: + """Plot-type children of c:plotArea (barChart, lineChart, ...).""" + plot_area = self.xml.root.find("c:chart/c:plotArea", NS) + if plot_area is None: + return [] + return [el for el in plot_area if _local(el).endswith("Chart")] + + # -- reading ----------------------------------------------------------------- + + @property + def kind(self) -> str: + if self._is_chartex: + ser = self.xml.root.find( + "cx:chart/cx:plotArea/cx:plotAreaRegion/cx:series", NS + ) + if ser is None: + ser = next(self.xml.root.iter(qn("cx:series")), None) + layout = ser.get("layoutId") if ser is not None else None + return f"chartex:{layout or 'unknown'}" + for plot in self._plot_elements(): + name = _local(plot) + if name in ("barChart", "bar3DChart"): + bar_dir = plot.find("c:barDir", NS) + val = bar_dir.get("val", "col") if bar_dir is not None else "col" + return "bar" if val == "bar" else "column" + return _PLOT_KIND.get(name, name) + return "unknown" + + @property + def title(self) -> str | None: + root = self.xml.root + if self._is_chartex: + title_el = root.find("cx:chart/cx:title", NS) + else: + chart = root.find("c:chart", NS) + if chart is None: + return None + atd = chart.find("c:autoTitleDeleted", NS) + if atd is not None and atd.get("val", "1") in ("1", "true"): + return None + title_el = chart.find("c:title", NS) + if title_el is None: + return None + text = "".join(t.text for t in title_el.iter(qn("a:t")) if t.text) + return text or None + + @property + def series(self) -> list[ChartSeriesData]: + if self._is_chartex: + return self._series_chartex() + out: list[ChartSeriesData] = [] + for plot in self._plot_elements(): + uses_xy = _local(plot) in _XY_PLOTS + for ser in plot.findall("c:ser", NS): + out.append(self._read_ser(ser, uses_xy)) + return out + + def _read_ser(self, ser: "_Element", uses_xy: bool) -> ChartSeriesData: + name = None + name_el = ser.find("c:tx/c:strRef/c:strCache/c:pt/c:v", NS) + if name_el is None: + name_el = ser.find("c:tx/c:v", NS) + if name_el is not None and name_el.text: + name = name_el.text + cat_holder = ser.find("c:xVal" if uses_xy else "c:cat", NS) + val_holder = ser.find("c:yVal" if uses_xy else "c:val", NS) + return ChartSeriesData( + name=name, + categories=self._cache_texts(cat_holder), + values=self._cache_values(val_holder), + ) + + @staticmethod + def _find_cache(holder: "_Element | None") -> "_Element | None": + """The cache element inside a c:cat/c:val/c:xVal/c:yVal holder.""" + if holder is None: + return None + for path in ( + "c:strRef/c:strCache", + "c:numRef/c:numCache", + "c:strLit", + "c:numLit", + # multi-level categories: first c:lvl = leaf labels + "c:multiLvlStrRef/c:multiLvlStrCache/c:lvl", + ): + cache = holder.find(path, NS) + if cache is not None: + return cache + return None + + @classmethod + def _cache_texts(cls, holder: "_Element | None") -> list[str]: + """Cache points as strings in idx order; gaps are skipped.""" + cache = cls._find_cache(holder) + if cache is None: + return [] + pts: list[tuple[int, str]] = [] + for pt in cache.findall("c:pt", NS): + v = pt.find("c:v", NS) + if v is None: + continue + try: + idx = int(pt.get("idx", str(len(pts)))) + except ValueError: + idx = len(pts) + pts.append((idx, v.text or "")) + return [text for _, text in sorted(pts, key=lambda p: p[0])] + + @classmethod + def _cache_values(cls, holder: "_Element | None") -> list[float | None]: + """Numeric cache; length = ptCount, gaps/unparseable → None.""" + cache = cls._find_cache(holder) + if cache is None: + return [] + by_idx: dict[int, float | None] = {} + for pt in cache.findall("c:pt", NS): + v = pt.find("c:v", NS) + try: + idx = int(pt.get("idx", "0")) + except ValueError: + continue + if v is None or v.text is None: + by_idx[idx] = None + continue + try: + by_idx[idx] = float(v.text) + except ValueError: + by_idx[idx] = None + pt_count_el = cache.find("c:ptCount", NS) + try: + count = int(pt_count_el.get("val", "")) if pt_count_el is not None else -1 + except ValueError: + count = -1 + if count < 0: + count = (max(by_idx) + 1) if by_idx else 0 + return [by_idx.get(i) for i in range(count)] + + def _series_chartex(self) -> list[ChartSeriesData]: + root = self.xml.root + data_by_id = { + data.get("id"): data for data in root.findall("cx:chartData/cx:data", NS) + } + series_els = root.findall( + "cx:chart/cx:plotArea/cx:plotAreaRegion/cx:series", NS + ) or list(root.iter(qn("cx:series"))) + out: list[ChartSeriesData] = [] + if not series_els: + # No series elements: expose each data block as an unnamed series. + for data in data_by_id.values(): + cats, vals = self._chartex_dims(data) + out.append(ChartSeriesData(name=None, categories=cats, values=vals)) + return out + for ser in series_els: + name_el = ser.find("cx:tx/cx:txData/cx:v", NS) + name = name_el.text if name_el is not None and name_el.text else None + data_id_el = ser.find("cx:dataId", NS) + data = ( + data_by_id.get(data_id_el.get("val")) + if data_id_el is not None + else None + ) + cats: list[str] = [] + vals: list[float | None] = [] + if data is not None: + cats, vals = self._chartex_dims(data) + out.append(ChartSeriesData(name=name, categories=cats, values=vals)) + return out + + @staticmethod + def _chartex_dims(data: "_Element") -> tuple[list[str], list[float | None]]: + """(categories, values) from a cx:data block's dims.""" + + def pick(dims: list["_Element"], wanted_type: str) -> "_Element | None": + for dim in dims: + if dim.get("type") == wanted_type: + return dim + return dims[0] if dims else None + + cats: list[str] = [] + vals: list[float | None] = [] + str_dim = pick(data.findall("cx:strDim", NS), "cat") + if str_dim is not None: + lvl = str_dim.find("cx:lvl", NS) + if lvl is not None: + pts = [] + for pt in lvl.findall("cx:pt", NS): + try: + idx = int(pt.get("idx", str(len(pts)))) + except ValueError: + idx = len(pts) + pts.append((idx, pt.text or "")) + cats = [text for _, text in sorted(pts, key=lambda p: p[0])] + num_dim = pick(data.findall("cx:numDim", NS), "val") + if num_dim is not None: + lvl = num_dim.find("cx:lvl", NS) + if lvl is not None: + by_idx: dict[int, float | None] = {} + for pt in lvl.findall("cx:pt", NS): + try: + idx = int(pt.get("idx", "0")) + except ValueError: + continue + try: + by_idx[idx] = float(pt.text) if pt.text else None + except ValueError: + by_idx[idx] = None + try: + count = int(lvl.get("ptCount", "")) + except ValueError: + count = (max(by_idx) + 1) if by_idx else 0 + vals = [by_idx.get(i) for i in range(count)] + return cats, vals + + # -- writing: title ---------------------------------------------------------- + + def set_title(self, text: str) -> None: + """Replace (or insert) the chart title text. + + Keeps the first existing run's ``a:rPr`` formatting; extra runs + and extra paragraphs of the old title are removed so the title + reads exactly *text*. Removes ``c:autoTitleDeleted val="1"`` so + the explicit title is shown. chartEx charts are read-only in + this milestone → :class:`RawUnsupportedError`. + """ + if self._is_chartex: + raise RawUnsupportedError( + "set_title is not supported for chartEx (cx:) charts yet; " + "classic c: charts only" + ) + chart = self.xml.root.find("c:chart", NS) + if chart is None: + raise RawUnsupportedError("chart part has no c:chart element") + + atd = chart.find("c:autoTitleDeleted", NS) + if atd is not None and atd.get("val", "1") in ("1", "true"): + chart.remove(atd) + + title_el = chart.find("c:title", NS) + if title_el is None: + title_el = etree.Element(qn("c:title")) + chart.insert(0, title_el) # c:title is the first child per schema + tx = title_el.find("c:tx", NS) + if tx is None: + tx = etree.Element(qn("c:tx")) + title_el.insert(0, tx) # c:tx is the first child of c:title + rich = tx.find("c:rich", NS) + if rich is None: + # Replace whatever c:tx held (e.g. a c:strRef) with rich text. + for child in list(tx): + tx.remove(child) + rich = etree.SubElement(tx, qn("c:rich")) + etree.SubElement(rich, qn("a:bodyPr")) + etree.SubElement(rich, qn("a:lstStyle")) + + paragraphs = rich.findall("a:p", NS) + if paragraphs: + para = paragraphs[0] + for extra in paragraphs[1:]: + rich.remove(extra) + else: + para = etree.SubElement(rich, qn("a:p")) + + run_props = None + first_run = para.find("a:r", NS) + if first_run is not None: + run_props = first_run.find("a:rPr", NS) + for el in list(para): + if el.tag in (qn("a:r"), qn("a:fld"), qn("a:br")): + para.remove(el) + run = etree.Element(qn("a:r")) + if run_props is not None: + run.append(copy.deepcopy(run_props)) + t = etree.SubElement(run, qn("a:t")) + t.text = text + end_props = para.find("a:endParaRPr", NS) + if end_props is not None: + end_props.addprevious(run) + else: + para.append(run) + self._commit() + + # -- writing: data ----------------------------------------------------------- + + def set_data( + self, + categories: Sequence[object], + series: Sequence["tuple[str | None, Sequence[float | None]] | ChartSeriesData"], + ) -> None: + """Rewrite the chart's cached data (and its embedded workbook). + + *series* accepts ``[(name, values), ...]`` or + ``[ChartSeriesData, ...]`` (whose ``.categories`` are ignored — + the *categories* argument is canonical for every series). + + Raises ``ValueError`` on ragged input and + :class:`RawUnsupportedError` for chartEx (cx:) charts. + """ + if self._is_chartex: + raise RawUnsupportedError( + "set_data is not supported for chartEx (cx:) charts in this " + "version; classic c: charts only" + ) + cats = list(categories) + normalized: list[tuple[str | None, list[float | None]]] = [] + for item in series: + if isinstance(item, ChartSeriesData): + normalized.append((item.name, list(item.values))) + else: + name, values = item + normalized.append((name, list(values))) + if not normalized: + raise ValueError("set_data requires at least one series") + for name, values in normalized: + if len(values) != len(cats): + raise ValueError( + f"series {name!r} has {len(values)} values for " + f"{len(cats)} categories (lengths must match)" + ) + + sers: list[tuple["_Element", "_Element"]] = [ + (plot, ser) + for plot in self._plot_elements() + for ser in plot.findall("c:ser", NS) + ] + if not sers: + raise RawUnsupportedError( + "chart has no existing c:ser element to rewrite or clone from" + ) + + sheet_ref = self._sheet_ref() + + # Shrink: drop trailing series. + while len(sers) > len(normalized): + plot, ser = sers.pop() + plot.remove(ser) + # Grow: clone the LAST series (deep copy keeps its styling). + if len(sers) < len(normalized): + used_indices = [] + for _, ser in sers: + for tag in ("c:idx", "c:order"): + el = ser.find(tag, NS) + if el is not None: + try: + used_indices.append(int(el.get("val", ""))) + except ValueError: + pass + next_index = (max(used_indices) + 1) if used_indices else len(sers) + plot, last = sers[-1] + while len(sers) < len(normalized): + clone = copy.deepcopy(last) + self._set_ser_index(clone, next_index) + last.addnext(clone) + sers.append((plot, clone)) + last = clone + next_index += 1 + + n = len(cats) + for i, ((plot, ser), (name, values)) in enumerate(zip(sers, normalized)): + col = _col_letter(i + 2) # data columns start at B + self._rewrite_ser( + ser, + name, + cats, + values, + uses_xy=_local(plot) in _XY_PLOTS, + cat_ref=f"{sheet_ref}!$A$2:$A${n + 1}", + val_ref=f"{sheet_ref}!${col}$2:${col}${n + 1}", + name_ref=f"{sheet_ref}!${col}$1", + ) + + workbook_part = self.embedded_workbook_part() + if workbook_part is not None: + workbook_part.write( + self._workbook_bytes(_unquote_sheet(sheet_ref), cats, normalized) + ) + self._commit() + + def _sheet_ref(self) -> str: + """Sheet prefix (verbatim, quotes kept) of the first c:f found.""" + for f in self.xml.root.iter(qn("c:f")): + text = f.text or "" + if "!" in text: + return text.rsplit("!", 1)[0] + return "Sheet1" + + @staticmethod + def _set_ser_index(ser: "_Element", index: int) -> None: + for tag in ("c:idx", "c:order"): + el = ser.find(tag, NS) + if el is None: + el = etree.Element(qn(tag)) + ser.insert(0 if tag == "c:idx" else 1, el) + el.set("val", str(index)) + + def _rewrite_ser( + self, + ser: "_Element", + name: str | None, + cats: list[object], + values: list[float | None], + *, + uses_xy: bool, + cat_ref: str, + val_ref: str, + name_ref: str, + ) -> None: + if name is not None: + self._rewrite_ser_name(ser, name, name_ref) + cat_tag = "c:xVal" if uses_xy else "c:cat" + val_tag = "c:yVal" if uses_xy else "c:val" + numeric_cats = all( + isinstance(c, (int, float)) and not isinstance(c, bool) for c in cats + ) + if uses_xy and numeric_cats: + new_cat = self._build_num_holder(cat_tag, cat_ref, cats) # type: ignore[arg-type] + else: + new_cat = self._build_str_holder(cat_tag, cat_ref, cats) + new_val = self._build_num_holder(val_tag, val_ref, values) + self._replace_ser_child(ser, cat_tag, new_cat, also_before=(val_tag,)) + self._replace_ser_child(ser, val_tag, new_val) + + def _rewrite_ser_name(self, ser: "_Element", name: str, name_ref: str) -> None: + tx = ser.find("c:tx", NS) + if tx is None: + tx = etree.Element(qn("c:tx")) + anchor = ser.find("c:order", NS) + if anchor is None: + anchor = ser.find("c:idx", NS) + if anchor is not None: + anchor.addnext(tx) + else: + ser.insert(0, tx) + literal = tx.find("c:v", NS) + if literal is not None and tx.find("c:strRef", NS) is None: + literal.text = name + return + for child in list(tx): + tx.remove(child) + str_ref = etree.SubElement(tx, qn("c:strRef")) + f = etree.SubElement(str_ref, qn("c:f")) + f.text = name_ref + cache = etree.SubElement(str_ref, qn("c:strCache")) + pt_count = etree.SubElement(cache, qn("c:ptCount")) + pt_count.set("val", "1") + pt = etree.SubElement(cache, qn("c:pt")) + pt.set("idx", "0") + v = etree.SubElement(pt, qn("c:v")) + v.text = name + + @staticmethod + def _build_str_holder(tag: str, ref: str, items: list[object]) -> "_Element": + holder = etree.Element(qn(tag)) + str_ref = etree.SubElement(holder, qn("c:strRef")) + f = etree.SubElement(str_ref, qn("c:f")) + f.text = ref + cache = etree.SubElement(str_ref, qn("c:strCache")) + pt_count = etree.SubElement(cache, qn("c:ptCount")) + pt_count.set("val", str(len(items))) + for idx, item in enumerate(items): + if item is None: + continue # gap + pt = etree.SubElement(cache, qn("c:pt")) + pt.set("idx", str(idx)) + v = etree.SubElement(pt, qn("c:v")) + v.text = str(item) + return holder + + @staticmethod + def _build_num_holder(tag: str, ref: str, values: list[float | None]) -> "_Element": + holder = etree.Element(qn(tag)) + num_ref = etree.SubElement(holder, qn("c:numRef")) + f = etree.SubElement(num_ref, qn("c:f")) + f.text = ref + cache = etree.SubElement(num_ref, qn("c:numCache")) + format_code = etree.SubElement(cache, qn("c:formatCode")) + format_code.text = "General" + pt_count = etree.SubElement(cache, qn("c:ptCount")) + pt_count.set("val", str(len(values))) + for idx, value in enumerate(values): + if value is None: + continue # gap + pt = etree.SubElement(cache, qn("c:pt")) + pt.set("idx", str(idx)) + v = etree.SubElement(pt, qn("c:v")) + v.text = _num_text(value) + return holder + + @staticmethod + def _replace_ser_child( + ser: "_Element", + tag: str, + new_el: "_Element", + also_before: tuple[str, ...] = (), + ) -> None: + """Swap the *tag* child in place, or insert it at a schema-valid + position (before the c:ser tail elements).""" + existing = ser.find(tag, NS) + if existing is not None: + existing.addprevious(new_el) + ser.remove(existing) + return + stop_tags = {qn(t) for t in (*_SER_TAIL_TAGS, *also_before)} + for child in ser: + if child.tag in stop_tags: + child.addprevious(new_el) + return + ser.append(new_el) + + # -- embedded workbook --------------------------------------------------------- + + def embedded_workbook_part(self) -> "OpcPart | None": + """The ``.../embeddings/*.xlsx`` part behind this chart, if any.""" + rels = self.package.rels_for(self.xml.name) + if rels is None: + return None + for rel in rels: + if rel["mode"] == "External": + continue + rel_type = rel["type"] or "" + if not (rel_type.endswith("/package") or rel_type.endswith("/oleObject")): + continue + target = rel["target"] or "" + if posixpath.splitext(target)[1].lower() != ".xlsx": + continue + name = rels.resolve(self.xml.name, target) + if self.package.has_part(name): + return self.package.get_part(name) + return None + + @staticmethod + def _workbook_bytes( + sheet_name: str, + cats: list[object], + normalized: list[tuple[str | None, list[float | None]]], + ) -> bytes: + """A fresh data-only workbook mirroring the rewritten caches.""" + from openpyxl import Workbook + + wb = Workbook() + ws = wb.active + ws.title = sheet_name + ws.append([None, *(name for name, _ in normalized)]) + for i, cat in enumerate(cats): + ws.append([cat, *(values[i] for _, values in normalized)]) + buf = io.BytesIO() + wb.save(buf) + return buf.getvalue() + + # -- persistence ----------------------------------------------------------- + + def _commit(self) -> None: + """Mark the facade dirty and serialize into the package part.""" + self.xml.mark_dirty() + self.xml.flush() + + +def load_chart(package: "OpcPackage", chart_part_name: str) -> ChartModel: + """Convenience: chart part name → :class:`ChartModel` facade.""" + return ChartModel(XmlPart(package.get_part(chart_part_name)), package) + + +def find_chart_parts(package: "OpcPackage", from_part: str) -> list[str]: + """Chart part names referenced (directly) by *from_part*'s rels. + + Format models use this from a drawing/slide part: + ``rels.by_type("/chart")`` → resolve targets → chart part names. + """ + rels = package.rels_for(from_part) + if rels is None: + return [] + return [rels.resolve(from_part, rel["target"]) for rel in rels.by_type("/chart")] diff --git a/src/xgen_edit2docs/raw/docx.py b/src/xgen_edit2docs/raw/docx.py new file mode 100644 index 00000000..5dd8bf6d --- /dev/null +++ b/src/xgen_edit2docs/raw/docx.py @@ -0,0 +1,567 @@ +# xgen_edit2docs/raw/docx.py (vendored from xgen-contextifier 0.9.0, Apache-2.0) +""" +DocxRawDocument — the raw (lossless, addressable, writable) model for +WordprocessingML documents. + +Addressing invariant +-------------------- +``paragraphs[i]`` is the *i*-th direct ``w:p`` child of ``w:body`` and +``tables[t]`` is the *t*-th direct ``w:tbl`` child — exactly the +elements python-docx exposes as ``document.paragraphs`` / +``document.tables``. Table cells use **grid addressing** with the same +semantics as python-docx ``row.cells``: a horizontally merged cell +(``w:gridSpan``) occupies every grid column it spans, and a vertically +merged continuation (``w:vMerge``) resolves to its start cell. This +keeps raw addresses interchangeable with edit2docs' existing addresses. + +Run-preserving text replacement (the P0-3 fix) +---------------------------------------------- +``set_paragraph_text`` never rebuilds a paragraph. Within the +paragraph, runs are classified: + +* **text runs** — ``w:r`` containing ``w:t`` and *no* protected content + (``w:drawing`` / ``w:pict`` / ``w:object`` / ``mc:AlternateContent``); +* **protected elements** — runs holding drawings/pictures/OLE, plus + every non-run child (bookmarks, math, comment ranges, revision + containers, ...). These are left in place, in order, untouched. + +The new text goes into the **first direct text run** (its ``w:rPr`` is +kept; ``w:t`` gets ``xml:space="preserve"`` when the text has leading +or trailing whitespace); the other direct pure-text runs are deleted. + +Hyperlink policy: ``w:hyperlink`` elements are never deleted by a text +replace. If the paragraph has no direct text run, the new text goes +into the *first hyperlink's first text run* (the hyperlink itself is +preserved). All other pure-text runs inside hyperlinks are emptied, but +the (now empty) hyperlink elements remain — callers that want them gone +use :meth:`DocxRawDocument.strip_empty_hyperlinks`, which also drops +the hyperlink relationship when nothing else references it. + +``RawCell.set_text`` applies the same rules per paragraph: the first +paragraph that owns a text run becomes the carrier, paragraphs with +protected content (inline images, ...) are left completely untouched, +and the remaining pure-text paragraphs are emptied — never removed, so +cell layout, nested tables and images survive (the edit2docs +cell-image-destruction fix). +""" + +from __future__ import annotations + +import copy +from typing import TYPE_CHECKING + +from xgen_edit2docs.raw.base import RawDocumentBase +from xgen_edit2docs.raw.chart import ChartModel, find_chart_parts +from xgen_edit2docs.raw.xmlpart import NS, qn + +if TYPE_CHECKING: # pragma: no cover + from lxml.etree import _Element + + from xgen_edit2docs.raw.opc import OpcPackage + +__all__ = ["DocxRawDocument", "RawParagraph", "RawTable", "RawCell"] + +_DOCUMENT_PART = "word/document.xml" + +#: Content that makes a run (or a paragraph) non-replaceable. +_PROTECTED_TAGS = ( + qn("w:drawing"), + qn("w:pict"), + qn("w:object"), + qn("mc:AlternateContent"), +) +_TEXTUAL_TAGS = (qn("w:t"), qn("w:tab"), qn("w:br"), qn("w:cr")) + + +# -- element-level helpers --------------------------------------------------- + + +def _contains_protected(el: "_Element") -> bool: + return next(el.iter(*_PROTECTED_TAGS), None) is not None + + +def _is_text_run(el: "_Element") -> bool: + """A pure-text run: ``w:r`` with a ``w:t`` and no protected content.""" + return ( + el.tag == qn("w:r") + and next(el.iter(qn("w:t")), None) is not None + and not _contains_protected(el) + ) + + +def _has_del_ancestor(node: "_Element", stop: "_Element") -> bool: + parent = node.getparent() + while parent is not None and parent is not stop: + if parent.tag == qn("w:del"): + return True + parent = parent.getparent() + return False + + +def _element_text(el: "_Element") -> str: + """Visible text of *el* (python-docx ``.text`` semantics). + + Concatenates ``w:t`` anywhere below *el* — including inside + ``w:hyperlink`` and ``w:ins`` — excluding deleted (``w:del``) + content; run-level ``w:tab``/``w:br``/``w:cr`` render as + ``"\\t"``/``"\\n"``. + """ + parts: list[str] = [] + for node in el.iter(*_TEXTUAL_TAGS): + if _has_del_ancestor(node, el): + continue + if node.tag == qn("w:t"): + parts.append(node.text or "") + elif node.getparent() is not None and node.getparent().tag == qn("w:r"): + parts.append("\t" if node.tag == qn("w:tab") else "\n") + return "".join(parts) + + +def _set_run_text(run: "_Element", text: str) -> None: + """Replace the run's content with a single ``w:t``, keeping ``w:rPr``.""" + from lxml import etree + + for child in list(run): + if child.tag != qn("w:rPr"): + run.remove(child) + t = etree.SubElement(run, qn("w:t")) + t.text = text + if text != text.strip(): + t.set("{http://www.w3.org/XML/1998/namespace}space", "preserve") + + +def _clear_run_text(run: "_Element") -> None: + """Empty a pure-text run (drop everything but ``w:rPr``).""" + for child in list(run): + if child.tag != qn("w:rPr"): + run.remove(child) + + +def _paragraph_text_runs(p_el: "_Element") -> tuple[list, list[tuple]]: + """(direct pure-text runs, [(hyperlink, its pure-text runs), ...]).""" + direct = [r for r in p_el.findall(qn("w:r")) if _is_text_run(r)] + in_links = [ + (h, [r for r in h.findall(qn("w:r")) if _is_text_run(r)]) + for h in p_el.findall(qn("w:hyperlink")) + ] + return direct, in_links + + +def _has_text_run(p_el: "_Element") -> bool: + direct, in_links = _paragraph_text_runs(p_el) + return bool(direct) or any(runs for _, runs in in_links) + + +def _replace_paragraph_text(p_el: "_Element", text: str) -> None: + """The run-preserving replace documented in the module docstring.""" + from lxml import etree + + direct, in_links = _paragraph_text_runs(p_el) + + carrier = direct[0] if direct else None + if carrier is None: + for _, runs in in_links: + if runs: + carrier = runs[0] + break + if carrier is None: # no text run anywhere — append a fresh one + carrier = etree.SubElement(p_el, qn("w:r")) + + _set_run_text(carrier, text) + for run in direct: + if run is not carrier: + p_el.remove(run) + for _, runs in in_links: + for run in runs: + if run is not carrier: + _clear_run_text(run) + + +def _clear_paragraph_text(p_el: "_Element") -> None: + """Empty a paragraph's text: delete direct pure-text runs, empty the + text runs inside hyperlinks (hyperlink elements stay).""" + direct, in_links = _paragraph_text_runs(p_el) + for run in direct: + p_el.remove(run) + for _, runs in in_links: + for run in runs: + _clear_run_text(run) + + +def _clear_row_copy(tr_el: "_Element") -> None: + """Blank a deep-copied row template: clear every paragraph's text and + drop the (copied) hyperlink elements. Cell count, ``w:trPr``, + ``w:tcPr`` (shading, spans, merges) all stay.""" + for p_el in tr_el.iter(qn("w:p")): + _clear_paragraph_text(p_el) + for h in list(p_el.findall(qn("w:hyperlink"))): + p_el.remove(h) + + +# -- model classes ------------------------------------------------------------- + + +class RawParagraph: + """A body-level paragraph (positional snapshot — re-fetch + ``document.paragraphs`` after structural edits).""" + + __slots__ = ("element", "index") + + def __init__(self, element: "_Element", index: int): + self.element = element + self.index = index + + @property + def text(self) -> str: + return _element_text(self.element) + + @property + def style(self) -> str: + """The ``w:pStyle`` style *id* (e.g. ``"Heading1"``), or ``"Normal"``.""" + style = self.element.find(f"{qn('w:pPr')}/{qn('w:pStyle')}") + return style.get(qn("w:val")) if style is not None else "Normal" + + def __repr__(self) -> str: # pragma: no cover - debug aid + return f"" + + +class RawCell: + """One (grid-resolved) table cell.""" + + __slots__ = ("element", "_doc") + + def __init__(self, element: "_Element", doc: "DocxRawDocument"): + self.element = element + self._doc = doc + + def _paragraph_els(self) -> list: + return self.element.findall(qn("w:p")) + + @property + def text(self) -> str: + """python-docx parity: direct paragraphs joined with newlines + (nested-table text is *not* included).""" + return "\n".join(_element_text(p) for p in self._paragraph_els()) + + @property + def paragraph_count(self) -> int: + return len(self._paragraph_els()) + + def set_text(self, text: str) -> None: + """Run- and layout-preserving text replace (see module docstring). + + The carrier is the first paragraph that owns a text run (else + the first paragraph without protected content, else a new + trailing paragraph). Paragraphs containing drawings / pictures / + OLE stay byte-for-byte untouched; other pure-text paragraphs are + emptied but never removed, so images and nested tables keep + their positions. + """ + from lxml import etree + + paras = self._paragraph_els() + target = next((p for p in paras if _has_text_run(p)), None) + if target is None: + target = next((p for p in paras if not _contains_protected(p)), None) + if target is None: + target = etree.SubElement(self.element, qn("w:p")) + _replace_paragraph_text(target, text) + for p in paras: + if p is not target and not _contains_protected(p): + _clear_paragraph_text(p) + self._doc._mark_document_dirty() + + def __repr__(self) -> str: # pragma: no cover - debug aid + return f"" + + +class RawTable: + """A table (body-level or nested) with grid-resolved cell access.""" + + __slots__ = ("element", "index", "_doc") + + def __init__(self, element: "_Element", index: int, doc: "DocxRawDocument"): + self.element = element + self.index = index + self._doc = doc + + def _row_els(self) -> list: + return self.element.findall(qn("w:tr")) + + @property + def n_rows(self) -> int: + return len(self._row_els()) + + @property + def n_cols(self) -> int: + grid = self.element.find(qn("w:tblGrid")) + return len(grid.findall(qn("w:gridCol"))) if grid is not None else 0 + + # -- grid resolution ----------------------------------------------------- + + def _grid(self) -> list[list]: + """``matrix[r][c]`` → the ``w:tc`` owning grid position (r, c). + + gridSpan repeats a cell across the columns it spans; a vMerge + continuation resolves to the start cell of the merge — the same + answers python-docx gives for ``table.rows[r].cells[c]``. + """ + rows = self._row_els() + n_cols = self.n_cols + matrix: list[list] = [[None] * n_cols for _ in rows] + for ri, tr in enumerate(rows): + ci = 0 + before = tr.find(f"{qn('w:trPr')}/{qn('w:gridBefore')}") + if before is not None: + ci = int(before.get(qn("w:val"), 0)) + for tc in tr.findall(qn("w:tc")): + tc_pr = tc.find(qn("w:tcPr")) + span, resolved = 1, tc + if tc_pr is not None: + grid_span = tc_pr.find(qn("w:gridSpan")) + if grid_span is not None: + span = int(grid_span.get(qn("w:val"), 1)) + v_merge = tc_pr.find(qn("w:vMerge")) + if ( + v_merge is not None + and v_merge.get(qn("w:val"), "continue") == "continue" + and ri > 0 + and ci < n_cols + and matrix[ri - 1][ci] is not None + ): + resolved = matrix[ri - 1][ci] + for k in range(span): + if ci + k < n_cols: + matrix[ri][ci + k] = resolved + ci += span + return matrix + + def _tc_at(self, r: int, c: int) -> "_Element": + if not (0 <= r < self.n_rows and 0 <= c < self.n_cols): + raise IndexError( + f"cell ({r}, {c}) out of range for {self.n_rows}x{self.n_cols} table" + ) + tc = self._grid()[r][c] + if tc is None: + raise IndexError(f"grid position ({r}, {c}) has no cell") + return tc + + def cell(self, r: int, c: int) -> RawCell: + return RawCell(self._tc_at(r, c), self._doc) + + def nested_tables(self, r: int, c: int) -> list["RawTable"]: + """Tables directly inside cell (r, c), addressable like any table.""" + return [ + RawTable(el, i, self._doc) + for i, el in enumerate(self._tc_at(r, c).findall(qn("w:tbl"))) + ] + + # -- row editing -------------------------------------------------------- + + def insert_row(self, idx: int) -> None: + """Insert a blank row at *idx* (0 ≤ idx ≤ n_rows), deep-copying the + row above (row 0 for idx=0) as the template: cell count and all + row/cell properties carry over, text is cleared.""" + rows = self._row_els() + if not rows: + raise ValueError("cannot insert into a table with no rows") + if not 0 <= idx <= len(rows): + raise IndexError(f"row index {idx} out of range 0..{len(rows)}") + template = rows[idx - 1] if idx > 0 else rows[0] + new_tr = copy.deepcopy(template) + _clear_row_copy(new_tr) + if idx == len(rows): + rows[-1].addnext(new_tr) + else: + rows[idx].addprevious(new_tr) + self._doc._mark_document_dirty() + + def delete_row(self, idx: int) -> None: + rows = self._row_els() + if not 0 <= idx < len(rows): + raise IndexError(f"row index {idx} out of range 0..{len(rows) - 1}") + self.element.remove(rows[idx]) + self._doc._mark_document_dirty() + + def __repr__(self) -> str: # pragma: no cover - debug aid + return f"" + + +class DocxRawDocument(RawDocumentBase): + """Raw model over ``word/document.xml`` (plus header/footer/chart + parts). Everything the model doesn't touch round-trips + byte-identically via the OPC container.""" + + format = "docx" + + def __init__(self, package: "OpcPackage"): + super().__init__(package) + if not package.has_part(_DOCUMENT_PART): + from xgen_edit2docs.raw.opc import RawUnsupportedError + + raise RawUnsupportedError("Package has no word/document.xml — not a docx") + self._document = self.xml_part(_DOCUMENT_PART) + self._charts: list[ChartModel] | None = None + + # -- internals ----------------------------------------------------------- + + @property + def _body(self) -> "_Element": + return self._document.find("w:body") + + def _mark_document_dirty(self) -> None: + self._document.mark_dirty() + + def _paragraph_els(self) -> list: + return self._body.findall(qn("w:p")) + + def _paragraph_el(self, index: int) -> "_Element": + els = self._paragraph_els() + if not 0 <= index < len(els): + raise IndexError(f"paragraph index {index} out of range 0..{len(els) - 1}") + return els[index] + + # -- paragraphs ----------------------------------------------------------- + + @property + def paragraphs(self) -> list[RawParagraph]: + """Body-level paragraphs, indexed like python-docx + ``document.paragraphs``.""" + return [RawParagraph(el, i) for i, el in enumerate(self._paragraph_els())] + + def set_paragraph_text(self, index: int, text: str) -> None: + """Run-preserving text replace (see module docstring): first text + run carries the new text with its formatting, protected runs + (images, OLE, bookmarks, math, ...) stay in place, hyperlink + elements survive (possibly with empty text).""" + _replace_paragraph_text(self._paragraph_el(index), text) + self._mark_document_dirty() + + def strip_empty_hyperlinks(self, index: int) -> int: + """Remove the paragraph's ``w:hyperlink`` children whose text is + empty (e.g. after :meth:`set_paragraph_text`). Relationships no + longer referenced anywhere in the document part are dropped too. + Returns the number of hyperlinks removed.""" + p_el = self._paragraph_el(index) + removed_ids: list[str] = [] + removed = 0 + for h in list(p_el.findall(qn("w:hyperlink"))): + if _element_text(h) == "": + rid = h.get(qn("r:id")) + p_el.remove(h) + removed += 1 + if rid: + removed_ids.append(rid) + if removed: + self._mark_document_dirty() + if removed_ids: + rels = self.package.rels_for(_DOCUMENT_PART) + r_ns = "{%s}" % NS["r"] + for rid in removed_ids: + still_used = any( + value == rid + for el in self._document.root.iter() + for key, value in el.attrib.items() + if key.startswith(r_ns) + ) + if rels is not None and not still_used: + rels.remove(rid) + return removed + + def insert_paragraph_after( + self, index: int, text: str, style: str | None = None + ) -> RawParagraph: + """Insert a new paragraph after ``paragraphs[index]``; ``index=-1`` + inserts at the very start of the body. Safe with a trailing body + ``w:sectPr`` (insertion is always anchored to an existing + paragraph, never appended past it).""" + from lxml import etree + + new_p = etree.Element(qn("w:p")) + if style: + p_pr = etree.SubElement(new_p, qn("w:pPr")) + p_style = etree.SubElement(p_pr, qn("w:pStyle")) + p_style.set(qn("w:val"), style) + run = etree.SubElement(new_p, qn("w:r")) + _set_run_text(run, text) + + if index == -1: + self._body.insert(0, new_p) + new_index = 0 + else: + self._paragraph_el(index).addnext(new_p) + new_index = index + 1 + self._mark_document_dirty() + return RawParagraph(new_p, new_index) + + def delete_paragraph(self, index: int) -> None: + el = self._paragraph_el(index) + el.getparent().remove(el) + self._mark_document_dirty() + + # -- tables --------------------------------------------------------------- + + @property + def tables(self) -> list[RawTable]: + """Body-level tables, indexed like python-docx ``document.tables`` + (nested tables are reached via :meth:`RawTable.nested_tables`).""" + return [ + RawTable(el, i, self) + for i, el in enumerate(self._body.findall(qn("w:tbl"))) + ] + + # -- charts --------------------------------------------------------------- + + @property + def chart_part_names(self) -> list[str]: + """Chart parts referenced from the document part's rels.""" + return find_chart_parts(self.package, _DOCUMENT_PART) + + @property + def charts(self) -> list[ChartModel]: + """Lazy :class:`ChartModel` per chart part (shared C3 contract).""" + if self._charts is None: + self._charts = [ + ChartModel(self.xml_part(name), self.package) + for name in self.chart_part_names + ] + return self._charts + + # -- headers / footers ------------------------------------------------------ + + def _header_footer_text(self, type_suffix: str) -> dict[str, str]: + rels = self.package.rels_for(_DOCUMENT_PART) + if rels is None: + return {} + out: dict[str, str] = {} + for rel in rels.by_type(type_suffix): + name = rels.resolve(_DOCUMENT_PART, rel["target"]) + root = self.xml_part(name).root + out[name] = "\n".join(_element_text(p) for p in root.iter(qn("w:p"))) + return out + + @property + def headers(self) -> dict[str, str]: + """Read-only ``{part name: text}`` for the header parts.""" + return self._header_footer_text("/header") + + @property + def footers(self) -> dict[str, str]: + """Read-only ``{part name: text}`` for the footer parts.""" + return self._header_footer_text("/footer") + + # -- structure --------------------------------------------------------------- + + def body_order(self) -> list[tuple[str, int]]: + """Document order of body children as ``("p"|"tbl", index)`` pairs — + the skeleton outline builders walk.""" + out: list[tuple[str, int]] = [] + p_i = t_i = 0 + for child in self._body: + if child.tag == qn("w:p"): + out.append(("p", p_i)) + p_i += 1 + elif child.tag == qn("w:tbl"): + out.append(("tbl", t_i)) + t_i += 1 + return out diff --git a/src/xgen_edit2docs/raw/errors.py b/src/xgen_edit2docs/raw/errors.py new file mode 100644 index 00000000..17dc4269 --- /dev/null +++ b/src/xgen_edit2docs/raw/errors.py @@ -0,0 +1,5 @@ +"""Error base class for the raw OOXML layer (kept for API compatibility).""" + + +class ContextifierError(Exception): + """Base class for raw-layer errors.""" diff --git a/src/xgen_edit2docs/raw/opc.py b/src/xgen_edit2docs/raw/opc.py new file mode 100644 index 00000000..a5e0f71a --- /dev/null +++ b/src/xgen_edit2docs/raw/opc.py @@ -0,0 +1,489 @@ +# xgen_edit2docs/raw/opc.py (vendored from xgen-contextifier 0.9.0, Apache-2.0) +""" +OPC (Open Packaging Conventions) container with a byte-preservation +write contract. + +An OOXML file is a ZIP of *parts* plus ``[Content_Types].xml`` and +``_rels/*.rels`` relationship graphs. The higher-level Office libraries +(openpyxl & friends) parse the whole package into their own object model +and re-serialize everything on save — silently destroying whatever they +don't model. This container does the opposite: + +**Byte-preservation contract** — on :meth:`OpcPackage.save`, every part +that was not explicitly written through :meth:`OpcPart.write` (or +removed/added) is emitted with its original decompressed bytes, +unchanged. Only dirty parts are re-serialized. The ZIP container itself +may be re-encoded (compression is not part of the contract); part +*content* is. + +The container knows nothing about spreadsheets or slides — format +models (:mod:`xgen_contextifier.raw.xlsx` etc.) sit on top. +""" + +from __future__ import annotations + +import io +import posixpath +import re +import zipfile +from pathlib import Path +from typing import BinaryIO, Callable, Iterable, Iterator + +from xgen_edit2docs.raw.errors import ContextifierError + +__all__ = [ + "OpcPackage", + "OpcPart", + "Relationships", + "RawUnsupportedError", + "make_part_renamer", +] + +_TRAILING_DIGITS = re.compile(r"^(.*?)(\d+)$") + + +def make_part_renamer(package: "OpcPackage") -> Callable[[str], str]: + """A fresh-name allocator for cloning parts. + + Given an existing part name it returns an unused one in the SAME + directory, keeping the non-numeric stem prefix and bumping the + trailing number (``ppt/slides/slide3.xml`` → ``ppt/slides/slide4.xml`` + if 4 is free, else the next gap). Names it has already handed out are + reserved even before they are added to the package, so a single clone + of a whole part subtree never collides with itself. + """ + allocated: set[str] = set() + + def rename(old: str) -> str: + directory = posixpath.dirname(old) + base = posixpath.basename(old) + stem, dot, ext = base.partition(".") + m = _TRAILING_DIGITS.match(stem) + prefix = m.group(1) if m else stem + n = int(m.group(2)) + 1 if m else 1 + while True: + cand_base = f"{prefix}{n}{dot}{ext}" + cand = posixpath.join(directory, cand_base) if directory else cand_base + if not package.has_part(cand) and cand not in allocated: + allocated.add(cand) + return cand + n += 1 + + return rename + + +_CONTENT_TYPES = "[Content_Types].xml" + +# Mirrors the pipeline's zip-bomb guard (handlers validate the same cap). +_MAX_UNCOMPRESSED = 1 << 30 # 1 GiB + + +class RawUnsupportedError(ContextifierError): + """The requested raw capability does not exist for this format.""" + + +class OpcPart: + """One part (file entry) inside the package. + + Parts are lazy: bytes are read from the source ZIP on first access. + Writing replaces the content and marks the part dirty; clean parts + round-trip byte-identically. + """ + + __slots__ = ("name", "_package", "_data", "dirty", "is_new") + + def __init__(self, name: str, package: "OpcPackage", *, is_new: bool = False): + self.name = name + self._package = package + self._data: bytes | None = None + self.dirty = is_new + self.is_new = is_new + + def read(self) -> bytes: + if self._data is None: + self._data = self._package._read_source(self.name) + return self._data + + def write(self, data: bytes) -> None: + self._data = bytes(data) + self.dirty = True + + def __repr__(self) -> str: # pragma: no cover - debug aid + state = "new" if self.is_new else ("dirty" if self.dirty else "clean") + return f"" + + +class Relationships: + """A part's ``_rels`` graph (``_rels/.rels``). + + Minimal, allocation-light XML handling: relationships are parsed with + lxml on demand and re-serialized only when mutated. + """ + + NS = "http://schemas.openxmlformats.org/package/2006/relationships" + + def __init__(self, part: OpcPart): + from lxml import etree + + self._part = part + self._root = etree.fromstring(part.read()) + + def __iter__(self) -> Iterator[dict]: + for rel in self._root: + if rel.tag == f"{{{self.NS}}}Relationship": + yield { + "id": rel.get("Id"), + "type": rel.get("Type"), + "target": rel.get("Target"), + "mode": rel.get("TargetMode", "Internal"), + } + + def target_of(self, rel_id: str) -> str | None: + for rel in self: + if rel["id"] == rel_id: + return rel["target"] + return None + + def by_type(self, type_suffix: str) -> list[dict]: + """Relationships whose Type ends with *type_suffix* (e.g. ``/chart``).""" + return [rel for rel in self if rel["type"].endswith(type_suffix)] + + def resolve(self, base_part: str, target: str) -> str: + """Resolve a relationship target to an absolute part name.""" + if target.startswith("/"): + return target[1:] + base_dir = posixpath.dirname(base_part) + return posixpath.normpath(posixpath.join(base_dir, target)) + + def add( + self, rel_id: str, rel_type: str, target: str, *, external: bool = False + ) -> None: + from lxml import etree + + rel = etree.SubElement(self._root, f"{{{self.NS}}}Relationship") + rel.set("Id", rel_id) + rel.set("Type", rel_type) + rel.set("Target", target) + if external: + rel.set("TargetMode", "External") + self.flush() + + def remove(self, rel_id: str) -> bool: + for rel in list(self._root): + if rel.get("Id") == rel_id: + self._root.remove(rel) + self.flush() + return True + return False + + def next_id(self) -> str: + used = {rel["id"] for rel in self} + n = len(used) + 1 + while f"rId{n}" in used: + n += 1 + return f"rId{n}" + + def flush(self) -> None: + from lxml import etree + + self._part.write( + etree.tostring( + self._root, xml_declaration=True, encoding="UTF-8", standalone=True + ) + ) + + +class OpcPackage: + """The package: parts + content types + relationship graphs. + + Open with :meth:`open`, mutate parts, then :meth:`save` / + :meth:`to_bytes`. Untouched parts keep their exact original bytes. + """ + + def __init__(self, source_bytes: bytes): + self._source = source_bytes + self._zip = zipfile.ZipFile(io.BytesIO(source_bytes)) + total = sum(i.file_size for i in self._zip.infolist()) + if total > _MAX_UNCOMPRESSED: + raise ContextifierError( + f"Package inflates to {total} bytes (> {_MAX_UNCOMPRESSED}); refusing" + ) + self._parts: dict[str, OpcPart] = { + info.filename: OpcPart(info.filename, self) + for info in self._zip.infolist() + if not info.is_dir() + } + self._removed: set[str] = set() + self._rels_cache: dict[str, Relationships] = {} + + # -- construction -------------------------------------------------------- + + @classmethod + def open(cls, source: str | Path | bytes | BinaryIO) -> "OpcPackage": + if isinstance(source, (str, Path)): + data = Path(source).read_bytes() + elif isinstance(source, (bytes, bytearray)): + data = bytes(source) + else: + data = source.read() + if data[:4] != b"PK\x03\x04": + raise ContextifierError("Not a ZIP/OPC package (bad magic)") + return cls(data) + + def sniff_format(self) -> str | None: + """Best-effort format detection from the package layout.""" + if self.has_part("xl/workbook.xml"): + return "xlsx" + if self.has_part("word/document.xml"): + return "docx" + if self.has_part("ppt/presentation.xml"): + return "pptx" + return None + + # -- part access ---------------------------------------------------------- + + def _read_source(self, name: str) -> bytes: + return self._zip.read(name) + + @property + def part_names(self) -> list[str]: + return sorted(self._parts) + + def has_part(self, name: str) -> bool: + return name in self._parts + + def get_part(self, name: str) -> OpcPart: + try: + return self._parts[name] + except KeyError: + raise KeyError(f"No such part: {name!r}") from None + + __getitem__ = get_part + + def add_part(self, name: str, data: bytes) -> OpcPart: + part = OpcPart(name, self, is_new=True) + part.write(data) + self._parts[name] = part + self._removed.discard(name) + return part + + def remove_part(self, name: str) -> None: + self._parts.pop(name, None) + self._rels_cache.pop(self._rels_name_for(name), None) + self._removed.add(name) + + # -- relationships & content types --------------------------------------- + + @staticmethod + def _rels_name_for(part_name: str) -> str: + if part_name == "": # package-level rels + return "_rels/.rels" + directory = posixpath.dirname(part_name) + base = posixpath.basename(part_name) + return ( + posixpath.join(directory, "_rels", f"{base}.rels") + if directory + else f"_rels/{base}.rels" + ) + + def rels_for(self, part_name: str) -> Relationships | None: + """The relationships of *part_name* ('' = package root), or None.""" + rels_name = self._rels_name_for(part_name) + if rels_name in self._rels_cache: + return self._rels_cache[rels_name] + if not self.has_part(rels_name): + return None + rels = Relationships(self.get_part(rels_name)) + self._rels_cache[rels_name] = rels + return rels + + def content_type_of(self, part_name: str) -> str | None: + from lxml import etree + + root = etree.fromstring(self.get_part(_CONTENT_TYPES).read()) + ns = "http://schemas.openxmlformats.org/package/2006/content-types" + for el in root: + if el.tag == f"{{{ns}}}Override" and el.get("PartName") == f"/{part_name}": + return el.get("ContentType") + ext = part_name.rsplit(".", 1)[-1].lower() + for el in root: + if ( + el.tag == f"{{{ns}}}Default" + and (el.get("Extension") or "").lower() == ext + ): + return el.get("ContentType") + return None + + def content_type_override_of(self, part_name: str) -> str | None: + """The *explicit* ```` content type for *part_name*, or + None. Unlike :meth:`content_type_of` this does NOT fall back to a + by-extension ```` — a cloned part only needs its own + Override copied (Defaults already cover it by extension).""" + from lxml import etree + + root = etree.fromstring(self.get_part(_CONTENT_TYPES).read()) + ns = "http://schemas.openxmlformats.org/package/2006/content-types" + for el in root: + if el.tag == f"{{{ns}}}Override" and el.get("PartName") == f"/{part_name}": + return el.get("ContentType") + return None + + def clone_part_graph( + self, + src_part: str, + *, + rename: Callable[[str], str], + share_types: Iterable[str] = (), + stop_types: Iterable[str] = (), + ) -> tuple[str, dict[str, str]]: + """Deep-copy *src_part* and the part subtree it references. + + Every part reachable from *src_part* through internal + relationships is copied to a fresh name from *rename*, unless its + relationship Type ends with one of: + + * a ``share_types`` suffix — the copy REFERENCES the original + target (not copied); use for read-only shared assets (images, + slide layouts, notes masters). + * a ``stop_types`` suffix — the relationship is dropped from the + copy entirely. + + Each copied part's ```` content type is duplicated, and + its ``.rels`` is rebuilt with the SAME relationship ids (so the + copied part bodies, which reference those ids, need no rewrite) — + only internal targets are retargeted to the clones. A relationship + that resolves to a part already cloned in this call reuses that + clone (so a notes-slide's back-reference to its slide points at the + NEW slide, and diamonds don't double-copy). + + Returns ``(new_src_part_name, {old_part: new_part})``. + """ + from lxml import etree + + share = tuple(share_types) + stop = tuple(stop_types) + rel_ns = Relationships.NS + mapping: dict[str, str] = {} + + def _matches(rtype: str | None, suffixes: tuple[str, ...]) -> bool: + return bool(rtype) and any(rtype.endswith(s) for s in suffixes) + + def _clone(old: str) -> str: + if old in mapping: + return mapping[old] + new = rename(old) + mapping[old] = new # reserve BEFORE recursing (breaks cycles) + self.add_part(new, self.get_part(old).read()) + override = self.content_type_override_of(old) + if override is not None: + self.set_content_type_override(new, override) + old_rels = self.rels_for(old) + if old_rels is None: + return new + new_dir = posixpath.dirname(new) + new_root = etree.Element(f"{{{rel_ns}}}Relationships", nsmap={None: rel_ns}) + for rel in old_rels: + attrs = {"Id": rel["id"], "Type": rel["type"], "Target": rel["target"]} + if rel["mode"] == "External": + attrs["TargetMode"] = "External" + elif _matches(rel["type"], stop): + continue + else: + abs_tgt = old_rels.resolve(old, rel["target"]) + dest = abs_tgt if _matches(rel["type"], share) else _clone(abs_tgt) + attrs["Target"] = ( + posixpath.relpath(dest, new_dir) if new_dir else dest + ) + sub = etree.SubElement(new_root, f"{{{rel_ns}}}Relationship") + for k, v in attrs.items(): + sub.set(k, v) + self.add_part( + self._rels_name_for(new), + etree.tostring( + new_root, xml_declaration=True, encoding="UTF-8", standalone=True + ), + ) + return new + + new_src = _clone(src_part) + return new_src, mapping + + def set_content_type_override(self, part_name: str, content_type: str) -> None: + from lxml import etree + + ct = self.get_part(_CONTENT_TYPES) + root = etree.fromstring(ct.read()) + ns = "http://schemas.openxmlformats.org/package/2006/content-types" + for el in root: + if el.tag == f"{{{ns}}}Override" and el.get("PartName") == f"/{part_name}": + el.set("ContentType", content_type) + break + else: + override = etree.SubElement(root, f"{{{ns}}}Override") + override.set("PartName", f"/{part_name}") + override.set("ContentType", content_type) + ct.write( + etree.tostring( + root, xml_declaration=True, encoding="UTF-8", standalone=True + ) + ) + + # -- saving ---------------------------------------------------------------- + + def to_bytes(self) -> bytes: + """Serialize the package. Clean parts are byte-identical.""" + buf = io.BytesIO() + original_infos = {i.filename: i for i in self._zip.infolist()} + with zipfile.ZipFile(buf, "w") as out: + # Original entry order first (Office is order-tolerant, but + # keeping it minimizes diffs), then new parts. + for name in [ + *(n for n in original_infos if n in self._parts), + *(n for n in self._parts if n not in original_infos), + ]: + part = self._parts[name] + src_info = original_infos.get(name) + if part.dirty or src_info is None: + data = part.read() + compress = ( + src_info.compress_type + if src_info is not None + else zipfile.ZIP_DEFLATED + ) + out.writestr( + zipfile.ZipInfo(name, date_time=(1980, 1, 1, 0, 0, 0)), + data, + compress_type=compress, + ) + else: + # Clean part: original decompressed bytes, original + # compression method, original timestamp. COPY the + # ZipInfo — writestr mutates it (header_offset/sizes), + # which would corrupt the source reader on a second + # to_bytes() call. + info_copy = zipfile.ZipInfo( + src_info.filename, date_time=src_info.date_time + ) + info_copy.compress_type = src_info.compress_type + info_copy.external_attr = src_info.external_attr + info_copy.internal_attr = src_info.internal_attr + info_copy.create_system = src_info.create_system + out.writestr(info_copy, self._zip.read(name)) + return buf.getvalue() + + def save(self, target: str | Path | BinaryIO | None = None) -> bytes: + data = self.to_bytes() + if isinstance(target, (str, Path)): + Path(target).write_bytes(data) + elif target is not None: + target.write(data) + return data + + def close(self) -> None: + self._zip.close() + + def __enter__(self) -> "OpcPackage": + return self + + def __exit__(self, *exc) -> None: + self.close() diff --git a/src/xgen_edit2docs/raw/pptx.py b/src/xgen_edit2docs/raw/pptx.py new file mode 100644 index 00000000..243d76af --- /dev/null +++ b/src/xgen_edit2docs/raw/pptx.py @@ -0,0 +1,657 @@ +# xgen_edit2docs/raw/pptx.py (vendored from xgen-contextifier 0.9.0, Apache-2.0) +""" +PptxRawDocument — the raw (lossless, writable) model for .pptx decks. + +Slides are thin views over their XML parts: reading never dirties, and +edits are surgical, so untouched parts round-trip byte-identical (the +OPC byte-preservation contract). Two operations exist specifically to +fix what template-driven pipelines (edit2docs-style) get wrong today: + +* :meth:`RawSlide.replace_content` — swap a slide's XML for freshly + generated markup while *pulling the original native objects along*: + chart / table / diagram graphicFrames (and optionally pictures) are + lifted out of the old tree and re-appended into the new one, ids + renumbered, relationships untouched. Native charts stay native + instead of being rasterized or dropped. +* :meth:`PptxRawDocument.remove_slide` — deletes the slide *and* + reference-counts every part it pulled in (charts, embedded workbooks, + images, notes), removing the ones no remaining slide uses. No more + orphan-part bloat in the package. +""" + +from __future__ import annotations + +import copy +import posixpath +from dataclasses import dataclass +from typing import TYPE_CHECKING, Iterator + +from xgen_edit2docs.raw.base import RawDocumentBase +from xgen_edit2docs.raw.chart import ChartModel, find_chart_parts +from xgen_edit2docs.raw.opc import make_part_renamer +from xgen_edit2docs.raw.xmlpart import NS, qn + +if TYPE_CHECKING: # pragma: no cover + from lxml.etree import _Element + +__all__ = [ + "PptxRawDocument", + "RawSlide", + "RawShapeInfo", + "RawTable", + "RawTableCell", +] + +_PRESENTATION = "ppt/presentation.xml" + +#: graphicData/@uri tail → shape kind +_GRAPHIC_KIND = { + "table": "table", + "chart": "chart", + "chartex": "chart", + "diagram": "diagram", +} + +#: uri tails of graphicFrames replace_content must always carry over +_NATIVE_FRAME_TAILS = ("table", "chart", "chartex", "diagram") + +#: nv*Pr wrappers whose first p:cNvPr identifies the shape +_NV_PR_TAGS = frozenset( + qn(t) + for t in ( + "p:nvSpPr", + "p:nvPicPr", + "p:nvGraphicFramePr", + "p:nvGrpSpPr", + "p:nvCxnSpPr", + ) +) + + +@dataclass +class RawShapeInfo: + """Inventory entry for one shape on a slide.""" + + id: int + name: str + kind: str # "text" | "picture" | "table" | "chart" | "group" | "diagram" | "other" + text: str | None + + +# -- XML helpers --------------------------------------------------------------- + + +def _para_text(para: "_Element") -> str: + return "".join(t.text or "" for t in para.iter(qn("a:t"))) + + +def _body_text(tx_body: "_Element") -> str: + return "\n".join(_para_text(p) for p in tx_body.findall(qn("a:p"))) + + +def _replace_para_text(para: "_Element", new_text: str) -> None: + """Set a paragraph's text under the format-preserving contract. + + The first ``a:r`` keeps its ``a:rPr`` (formatting) and receives the + new text; every *other* plain-text run is dropped; anything that is + not a plain run — ``a:fld``, ``a:br``, ``a:pPr``, ``a:endParaRPr`` — + stays exactly where it was. + """ + from lxml import etree + + a_r, a_t = qn("a:r"), qn("a:t") + runs = [child for child in para if child.tag == a_r] + if runs: + first = runs[0] + t = first.find(a_t) + if t is None: + t = etree.SubElement(first, a_t) + t.text = new_text + for extra in runs[1:]: + para.remove(extra) + return + run = etree.SubElement(para, a_r) + etree.SubElement(run, a_t).text = new_text + end = para.find(qn("a:endParaRPr")) + if end is not None: + end.addprevious(run) + + +def _shape_cnvpr(shape_el: "_Element") -> "_Element | None": + """The shape's own ``p:cNvPr`` (never a nested child's).""" + for child in shape_el: + if child.tag in _NV_PR_TAGS: + return child.find(qn("p:cNvPr")) + return None + + +def _graphic_kind(frame_el: "_Element") -> str | None: + data = frame_el.find("a:graphic/a:graphicData", NS) + if data is None: + return None + uri = data.get("uri") or "" + return _GRAPHIC_KIND.get(uri.rsplit("/", 1)[-1]) + + +def _walk_shapes(container: "_Element") -> Iterator["_Element"]: + """Shape elements in document order, descending into p:grpSp.""" + shape_tags = ( + qn("p:sp"), + qn("p:pic"), + qn("p:graphicFrame"), + qn("p:grpSp"), + qn("p:cxnSp"), + ) + for child in container: + if child.tag in shape_tags: + yield child + if child.tag == qn("p:grpSp"): + yield from _walk_shapes(child) + + +def _first_rid(el: "_Element", attr: str) -> str | None: + """First ``attr`` (a qualified name) found on *el* or a descendant.""" + for node in el.iter(): + rid = node.get(attr) + if rid: + return rid + return None + + +# -- tables -------------------------------------------------------------------- + + +class RawTableCell: + """One ``a:tc`` — read/format-preserving-write of its text.""" + + def __init__(self, table: "RawTable", tc_el: "_Element"): + self._table = table + self._tc = tc_el + + @property + def text(self) -> str: + tx = self._tc.find(qn("a:txBody")) + return _body_text(tx) if tx is not None else "" + + def set_text(self, text: str) -> None: + """Replace the first paragraph's text (first run's ``a:rPr`` and + any non-text elements are preserved — same rules as + :meth:`RawSlide.set_text`).""" + from lxml import etree + + tx = self._tc.find(qn("a:txBody")) + if tx is None: + tx = etree.SubElement(self._tc, qn("a:txBody")) + etree.SubElement(tx, qn("a:bodyPr")) + etree.SubElement(tx, qn("a:lstStyle")) + para = tx.find(qn("a:p")) + if para is None: + para = etree.SubElement(tx, qn("a:p")) + _replace_para_text(para, text) + self._table._slide._mark_dirty() + + +class RawTable: + """A native DrawingML table (``a:tbl``) hosted in a graphicFrame. + + v0.4 scope: cell text + row insert/delete. Column operations would + require ``a:gridCol`` surgery and are deliberately out of scope. + """ + + def __init__(self, slide: "RawSlide", frame_el: "_Element"): + self._slide = slide + self._frame = frame_el + cnvpr = _shape_cnvpr(frame_el) + self.shape_id: int = int(cnvpr.get("id")) if cnvpr is not None else -1 + tbl = frame_el.find("a:graphic/a:graphicData/a:tbl", NS) + if tbl is None: # pragma: no cover - guarded by caller's uri check + raise ValueError("graphicFrame does not contain an a:tbl") + self._tbl = tbl + + # -- geometry --------------------------------------------------------------- + + @property + def _rows(self) -> list["_Element"]: + return self._tbl.findall(qn("a:tr")) + + @property + def n_rows(self) -> int: + return len(self._rows) + + @property + def n_cols(self) -> int: + return len(self._tbl.findall("a:tblGrid/a:gridCol", NS)) + + # -- cells ------------------------------------------------------------------ + + def cell(self, r: int, c: int) -> RawTableCell: + rows = self._rows + if not 0 <= r < len(rows): + raise IndexError(f"row {r} out of range (table has {len(rows)} rows)") + cells = rows[r].findall(qn("a:tc")) + if not 0 <= c < len(cells): + raise IndexError(f"col {c} out of range (row has {len(cells)} cells)") + return RawTableCell(self, cells[c]) + + # -- rows ------------------------------------------------------------------- + + def insert_row(self, idx: int) -> None: + """Insert an empty row at *idx*, cloning the row above (or the + first row) as the style template. Column count stays consistent + with ``a:tblGrid`` because the template row already matches it.""" + from lxml import etree + + rows = self._rows + if not rows: + raise ValueError("cannot insert into a table with no template row") + if not 0 <= idx <= len(rows): + raise IndexError(f"insert index {idx} out of range (0..{len(rows)})") + template = rows[idx - 1] if idx > 0 else rows[0] + new_row = copy.deepcopy(template) + for tc in new_row.findall(qn("a:tc")): + tx = tc.find(qn("a:txBody")) + if tx is not None: + for para in tx.findall(qn("a:p")): + tx.remove(para) + etree.SubElement(tx, qn("a:p")) + if idx == len(rows): + rows[-1].addnext(new_row) + else: + rows[idx].addprevious(new_row) + self._slide._mark_dirty() + + def delete_row(self, idx: int) -> None: + rows = self._rows + if not 0 <= idx < len(rows): + raise IndexError(f"row {idx} out of range (table has {len(rows)} rows)") + self._tbl.remove(rows[idx]) + self._slide._mark_dirty() + + +# -- slides -------------------------------------------------------------------- + + +class RawSlide: + """One slide part, addressed by shape id.""" + + def __init__(self, doc: "PptxRawDocument", part_name: str, index: int): + self._doc = doc + self.part_name = part_name + self.index = index + + def __repr__(self) -> str: # pragma: no cover - debug aid + return f"" + + @property + def _xp(self): + return self._doc.xml_part(self.part_name) + + def _mark_dirty(self) -> None: + self._xp.mark_dirty() + + @property + def _sp_tree(self) -> "_Element": + tree = self._xp.find("p:cSld/p:spTree") + if tree is None: + raise ValueError(f"{self.part_name} has no p:cSld/p:spTree") + return tree + + # -- inventory --------------------------------------------------------------- + + @property + def shapes(self) -> list[RawShapeInfo]: + """All shapes (document order, groups flattened in place).""" + out: list[RawShapeInfo] = [] + for el in _walk_shapes(self._sp_tree): + cnvpr = _shape_cnvpr(el) + shape_id = int(cnvpr.get("id")) if cnvpr is not None else -1 + name = (cnvpr.get("name") or "") if cnvpr is not None else "" + kind, text = "other", None + if el.tag == qn("p:sp"): + tx = el.find(qn("p:txBody")) + if tx is not None: + kind, text = "text", _body_text(tx) + elif el.tag == qn("p:pic"): + kind = "picture" + elif el.tag == qn("p:grpSp"): + kind = "group" + elif el.tag == qn("p:graphicFrame"): + kind = _graphic_kind(el) or "other" + out.append(RawShapeInfo(id=shape_id, name=name, kind=kind, text=text)) + return out + + def _find_shape(self, shape_id: int) -> "_Element": + for el in _walk_shapes(self._sp_tree): + cnvpr = _shape_cnvpr(el) + if cnvpr is not None and cnvpr.get("id") == str(shape_id): + return el + raise KeyError(f"No shape with id={shape_id} on {self.part_name}") + + # -- text -------------------------------------------------------------------- + + def get_text(self, shape_id: int) -> str: + el = self._find_shape(shape_id) + tx = el.find(qn("p:txBody")) + if tx is None: + raise ValueError(f"Shape id={shape_id} has no text body") + return _body_text(tx) + + def set_text(self, shape_id: int, new_text: str, para: int = 0) -> None: + """Replace paragraph *para*'s text, preserving the first run's + formatting (``a:rPr``) and any non-text elements (``a:fld``, + ``a:br``); other plain-text runs in the paragraph are removed.""" + el = self._find_shape(shape_id) + tx = el.find(qn("p:txBody")) + if tx is None: + raise ValueError(f"Shape id={shape_id} has no text body") + paras = tx.findall(qn("a:p")) + if not 0 <= para < len(paras): + raise IndexError( + f"paragraph {para} out of range (shape has {len(paras)} paragraphs)" + ) + _replace_para_text(paras[para], new_text) + self._mark_dirty() + + # -- tables / charts / notes --------------------------------------------------- + + @property + def tables(self) -> list[RawTable]: + return [ + RawTable(self, el) + for el in _walk_shapes(self._sp_tree) + if el.tag == qn("p:graphicFrame") and _graphic_kind(el) == "table" + ] + + @property + def chart_part_names(self) -> list[str]: + """Chart parts referenced from this slide's relationships.""" + return find_chart_parts(self._doc.package, self.part_name) + + @property + def charts(self) -> list[ChartModel]: + """Lazy :class:`ChartModel` views (implemented in milestone C3).""" + return [ + ChartModel(self._doc.xml_part(name), self._doc.package) + for name in self.chart_part_names + ] + + @property + def notes_text(self) -> str | None: + """Text of the notes slide's body placeholder, or ``None``.""" + rels = self._doc.package.rels_for(self.part_name) + if rels is None: + return None + notes_rels = rels.by_type("/notesSlide") + if not notes_rels: + return None + notes_part = rels.resolve(self.part_name, notes_rels[0]["target"]) + xp = self._doc.xml_part(notes_part) + for sp in xp.findall(".//p:sp"): + ph = sp.find("p:nvSpPr/p:nvPr/p:ph", NS) + tx = sp.find(qn("p:txBody")) + if tx is not None and ph is not None and ph.get("type") == "body": + return _body_text(tx) + texts = [_body_text(tx) for tx in xp.root.iter(qn("p:txBody"))] + return "\n".join(t for t in texts if t) + + # -- content replacement -------------------------------------------------------- + + def replace_content( + self, + new_slide_xml: bytes, + *, + preserve_native: bool = True, + preserve_pictures: bool = True, + ) -> list[str]: + """Replace this slide's XML with *new_slide_xml*, carrying the + original native objects over. + + With ``preserve_native=True`` every chart / table / diagram + graphicFrame of the ORIGINAL slide (and, when + ``preserve_pictures``, every ``p:pic`` that references embedded + media) is deep-copied into the new tree's ``p:spTree``, with + their ``p:cNvPr/@id`` renumbered past any id used by the new + XML. The slide keeps its part name, so its relationships part is + untouched and every carried-over ``r:id`` / ``r:embed`` keeps + resolving. + + *new_slide_xml* must be a complete ```` document and may + only reference relationship ids that already exist in this + slide's rels (or none at all) — this method never edits the rels + part, so unknown ``r:id`` / ``r:embed`` values in the new XML + would dangle. + + Returns descriptions of the preserved elements, e.g. + ``["table", "chart:chart1.xml", "picture:image1.png"]``. + """ + from lxml import etree + + new_root = etree.fromstring(new_slide_xml) + if new_root.tag != qn("p:sld"): + raise ValueError("new_slide_xml must be a complete document") + new_tree = new_root.find("p:cSld/p:spTree", NS) + if new_tree is None: + raise ValueError("new_slide_xml has no p:cSld/p:spTree") + + preserved: list[str] = [] + if preserve_native: + rels = self._doc.package.rels_for(self.part_name) + keep: list[tuple["_Element", str]] = [] + for el in _walk_shapes(self._sp_tree): + if el.tag == qn("p:graphicFrame"): + data = el.find("a:graphic/a:graphicData", NS) + uri_tail = ( + (data.get("uri") or "").rsplit("/", 1)[-1] + if data is not None + else "" + ) + if uri_tail not in _NATIVE_FRAME_TAILS: + continue + desc = _GRAPHIC_KIND[uri_tail] + if desc == "chart": + rid = _first_rid(data, qn("r:id")) + target = rels.target_of(rid) if (rels and rid) else None + if target: + desc = f"chart:{posixpath.basename(target)}" + keep.append((el, desc)) + elif el.tag == qn("p:pic") and preserve_pictures: + blip = el.find("p:blipFill/a:blip", NS) + rid = blip.get(qn("r:embed")) if blip is not None else None + if not rid: + continue + target = rels.target_of(rid) if rels else None + desc = ( + f"picture:{posixpath.basename(target)}" if target else "picture" + ) + keep.append((el, desc)) + + used_ids = { + int(c.get("id")) + for c in new_root.iter(qn("p:cNvPr")) + if (c.get("id") or "").isdigit() + } + next_id = max(used_ids, default=1) + 1 + for el, desc in keep: + clone = copy.deepcopy(el) + for cnvpr in clone.iter(qn("p:cNvPr")): + cnvpr.set("id", str(next_id)) + next_id += 1 + new_tree.append(clone) + preserved.append(desc) + + xp = self._xp + xp._root = new_root # swap the facade's tree in place + xp.mark_dirty() + return preserved + + +# -- document ------------------------------------------------------------------ + + +class PptxRawDocument(RawDocumentBase): + """Raw model for a .pptx package.""" + + format = "pptx" + + @property + def slides(self) -> list[RawSlide]: + """Slides in presentation order (``p:sldIdLst``).""" + pres = self.xml_part(_PRESENTATION) + rels = self.package.rels_for(_PRESENTATION) + sld_id_lst = pres.find("p:sldIdLst") + if sld_id_lst is None or rels is None: + return [] + out: list[RawSlide] = [] + for sld_id in sld_id_lst.findall(qn("p:sldId")): + rid = sld_id.get(qn("r:id")) + target = rels.target_of(rid) if rid else None + if target is None: + continue + out.append(RawSlide(self, rels.resolve(_PRESENTATION, target), len(out))) + return out + + # -- slide removal ------------------------------------------------------------ + + def remove_slide(self, index: int) -> None: + """Remove the slide at *index* and everything only it used. + + Beyond dropping the ``p:sldId`` entry and the slide part itself + (plus its rels and notes slide), this reference-counts every + part transitively reachable from the removed slide — charts, + embedded chart workbooks, images, chart colors/style parts — + against the relationships of everything still in the package, + and deletes the now-orphaned ones, including their content-type + overrides. Parts shared with surviving slides (or anchored by + the presentation / masters, like layouts and the notes master) + are untouched, and surviving slide parts stay byte-identical. + """ + slides = self.slides + if not 0 <= index < len(slides): + raise IndexError(f"slide index {index} out of range (0..{len(slides) - 1})") + part_name = slides[index].part_name + + pres = self.xml_part(_PRESENTATION) + pres_rels = self.package.rels_for(_PRESENTATION) + if pres_rels is None: # pragma: no cover - malformed package + raise ValueError("presentation has no relationships part") + rid = next( + ( + rel["id"] + for rel in pres_rels.by_type("/slide") + if pres_rels.resolve(_PRESENTATION, rel["target"]) == part_name + ), + None, + ) + sld_id_lst = pres.find("p:sldIdLst") + if sld_id_lst is not None and rid is not None: + for sld_id in list(sld_id_lst): + if sld_id.get(qn("r:id")) == rid: + sld_id_lst.remove(sld_id) + pres.mark_dirty() + if rid is not None: + pres_rels.remove(rid) + + # The slide dies; so does its notes slide. + doomed = [part_name] + slide_rels = self.package.rels_for(part_name) + if slide_rels is not None: + doomed += [ + slide_rels.resolve(part_name, rel["target"]) + for rel in slide_rels.by_type("/notesSlide") + ] + + # Delete the doomed parts and sweep every part only they anchored + # (charts, embedded workbooks, images, notes) — shared base logic. + self._sweep_orphans(doomed) + + # -- reordering & duplication --------------------------------------------- + + #: relationship Types a duplicated slide REFERENCES (shares) instead of + #: copying: read-only or globally-anchored assets. Everything else + #: (charts + their embedded workbooks/colors/styles, and the notes + #: slide) is cloned, so editing the copy never mutates the original. + _SHARE_ON_DUPLICATE = ( + "/slideLayout", + "/notesMaster", + "/image", + "/audio", + "/video", + "/tags", + ) + + def move_slide(self, index: int, to: int) -> None: + """Move the slide at *index* to position *to*. + + A pure reorder of ``p:sldIdLst`` in ``ppt/presentation.xml`` — no + part is copied, renamed or deleted, so every slide part (and + everything else in the package) stays byte-identical.""" + pres = self.xml_part(_PRESENTATION) + sld_id_lst = pres.find("p:sldIdLst") + if sld_id_lst is None: + raise ValueError("presentation has no p:sldIdLst") + ids = sld_id_lst.findall(qn("p:sldId")) + n = len(ids) + if not 0 <= index < n: + raise IndexError(f"slide index {index} out of range (0..{n - 1})") + if not 0 <= to < n: + raise IndexError(f"destination {to} out of range (0..{n - 1})") + if index == to: + return + el = ids[index] + sld_id_lst.remove(el) + remaining = sld_id_lst.findall(qn("p:sldId")) + if to >= len(remaining): + sld_id_lst.append(el) + else: + remaining[to].addprevious(el) + pres.mark_dirty() + + def duplicate_slide(self, index: int, *, at: int | None = None) -> int: + """Insert an independent copy of slide *index* at position *at* + (default: right after the source). Returns the new slide's index. + + The slide's XML and its non-shared referenced parts (charts, + embedded workbooks, chart colors/styles, notes slide) are deep- + copied under fresh names via ``clone_part_graph``; images, layouts + and the notes master are shared (see ``_SHARE_ON_DUPLICATE``). The + notes slide's back-reference is retargeted to the copy. A new + ``p:sldId`` (id = max + 1, ≥ 256) and presentation relationship are + added; existing slide parts stay byte-identical.""" + slides = self.slides + if not 0 <= index < len(slides): + raise IndexError(f"slide index {index} out of range (0..{len(slides) - 1})") + src_part = slides[index].part_name + + renamer = make_part_renamer(self.package) + new_part, _ = self.package.clone_part_graph( + src_part, rename=renamer, share_types=self._SHARE_ON_DUPLICATE + ) + + pres = self.xml_part(_PRESENTATION) + pres_rels = self.package.rels_for(_PRESENTATION) + if pres_rels is None: # pragma: no cover - malformed package + raise ValueError("presentation has no relationships part") + slide_rel_type = next( + (rel["type"] for rel in pres_rels.by_type("/slide")), + "http://schemas.openxmlformats.org/officeDocument/2006/relationships/slide", + ) + new_rid = pres_rels.next_id() + target = posixpath.relpath(new_part, posixpath.dirname(_PRESENTATION)) + pres_rels.add(new_rid, slide_rel_type, target) + + sld_id_lst = pres.find("p:sldIdLst") + if sld_id_lst is None: # pragma: no cover - malformed package + raise ValueError("presentation has no p:sldIdLst") + existing = sld_id_lst.findall(qn("p:sldId")) + used_ids = [int(s.get("id")) for s in existing if (s.get("id") or "").isdigit()] + new_el = sld_id_lst.makeelement( + qn("p:sldId"), + {"id": str(max(used_ids, default=255) + 1), qn("r:id"): new_rid}, + ) + pos = index + 1 if at is None else max(0, min(at, len(existing))) + if pos >= len(existing): + sld_id_lst.append(new_el) + else: + existing[pos].addprevious(new_el) + pres.mark_dirty() + return pos diff --git a/src/xgen_edit2docs/raw/xlsx.py b/src/xgen_edit2docs/raw/xlsx.py new file mode 100644 index 00000000..b1ef6060 --- /dev/null +++ b/src/xgen_edit2docs/raw/xlsx.py @@ -0,0 +1,825 @@ +# xgen_edit2docs/raw/xlsx.py (vendored from xgen-contextifier 0.9.0, Apache-2.0) +""" +XLSX raw document model — SpreadsheetML semantics on the OPC container. + +:class:`XlsxRawDocument` gives addressable, *surgical* read/write access +to a workbook. Only the worksheet XML you actually touch (plus, when +needed, ``xl/workbook.xml`` for ``calcPr``) is ever re-serialized — +every other part rides the byte-preservation contract untouched, so +charts, chart styles, pivot tables, sparkline ``extLst`` blocks, custom +XML, themes and styles all survive a save (openpyxl round-trips destroy +all of these). + +Design decisions: + +* **String writes use inline strings** (``t="inlineStr"``), never the + shared-string table — so ``xl/sharedStrings.xml`` is never rewritten + (and never *created*; a workbook without one stays without one). +* **Overwriting a formula cell removes the formula.** ``set_cell`` is + "the user supplies the value now": the ```` element is dropped and + the literal is written. Use :meth:`RawSheet.get_formula` to inspect + formulas before overwriting. +* **Stale caches recalculate on open.** Whenever a ``set_cell`` / + ``append_rows`` overwrites a formula cell — or any formula exists + anywhere in the workbook (its cached result may now be stale) — the + workbook's ```` gains ``fullCalcOnLoad="1"`` (created after + the last of sheets/definedNames if absent) so Excel recalculates. +* **Values pass through as stored.** No date-system handling (1900 vs + 1904 is ignored); date cells read back as their serial numbers. +""" + +from __future__ import annotations + +import math +import posixpath +import re +from typing import TYPE_CHECKING, Iterator + +from xgen_edit2docs.raw.base import RawDocumentBase +from xgen_edit2docs.raw.opc import make_part_renamer +from xgen_edit2docs.raw.xmlpart import XmlPart, qn + +#: a schema-minimal worksheet (CT_Worksheet only requires ) +_MINIMAL_WORKSHEET_XML = ( + b'\n' + b'' + b"" +) +_WORKSHEET_CONTENT_TYPE = ( + "application/vnd.openxmlformats-officedocument.spreadsheetml.worksheet+xml" +) +_WORKSHEET_REL_TYPE = ( + "http://schemas.openxmlformats.org/officeDocument/2006/relationships/worksheet" +) +#: characters Excel forbids in a sheet tab name +_FORBIDDEN_SHEET_CHARS = set(r":\/?*[]") + +if TYPE_CHECKING: # pragma: no cover + from lxml.etree import _Element + + from xgen_edit2docs.raw.chart import ChartModel + from xgen_edit2docs.raw.opc import OpcPackage + +__all__ = ["XlsxRawDocument", "RawSheet", "SheetCollection"] + +#: attribute key for ``xml:space`` +_XML_SPACE = "{http://www.w3.org/XML/1998/namespace}space" + +_REF_RE = re.compile(r"^\$?([A-Za-z]{1,3})\$?([1-9][0-9]*)$") +_RANGE_RE = re.compile( + r"^\$?([A-Za-z]{1,3})\$?([1-9][0-9]*)(?::\$?([A-Za-z]{1,3})\$?([1-9][0-9]*))?$" +) + +#: workbook children that must precede (CT_Workbook sequence) +_PRE_CALC_PR = ( + "fileVersion", + "fileSharing", + "workbookPr", + "workbookProtection", + "bookViews", + "sheets", + "functionGroups", + "externalReferences", + "definedNames", +) + + +# -- A1-reference helpers ---------------------------------------------------- + + +def col_letters_to_index(letters: str) -> int: + """``"A"`` → 1, ``"B"`` → 2, ..., ``"AA"`` → 27.""" + idx = 0 + for ch in letters.upper(): + idx = idx * 26 + (ord(ch) - 64) + return idx + + +def col_index_to_letters(idx: int) -> str: + """1 → ``"A"``, 27 → ``"AA"``.""" + if idx < 1: + raise ValueError(f"Column index must be >= 1, got {idx}") + out: list[str] = [] + while idx: + idx, rem = divmod(idx - 1, 26) + out.append(chr(65 + rem)) + return "".join(reversed(out)) + + +def parse_ref(ref: str) -> tuple[int, int]: + """``"B3"`` → ``(3, 2)`` (row, column), 1-based. ``$`` anchors OK.""" + m = _REF_RE.match(ref.strip()) + if m is None: + raise ValueError(f"Not a cell reference: {ref!r}") + return int(m.group(2)), col_letters_to_index(m.group(1)) + + +def _row_number(row_el: "_Element", prev: int) -> int: + r = row_el.get("r") + return int(r) if r else prev + 1 + + +def _cell_column(c_el: "_Element", prev: int) -> int: + r = c_el.get("r") + if r: + m = _REF_RE.match(r) + if m is not None: + return col_letters_to_index(m.group(1)) + return prev + 1 + + +class RawSheet: + """One worksheet: addressable cells over the live lxml tree. + + Reads never dirty the part; every mutation marks it dirty so + ``save()`` re-serializes exactly this worksheet and nothing else. + """ + + def __init__(self, doc: "XlsxRawDocument", name: str, part_name: str): + self._doc = doc + self.name = name + self.part_name = part_name + + def __repr__(self) -> str: # pragma: no cover - debug aid + return f"" + + @property + def _xp(self) -> XmlPart: + return self._doc.xml_part(self.part_name) + + def _sheet_data(self) -> "_Element": + sd = self._xp.find("s:sheetData") + if sd is None: # degenerate but legal: create the container + sd = self._xp.root.makeelement(qn("s:sheetData"), {}) + self._xp.root.append(sd) + return sd + + # -- reading --------------------------------------------------------------- + + def get_cell(self, ref: str) -> object | None: + """The cell's value (formula cells: the *cached* value), or None. + + Types follow the cell's ``t``: shared/inline strings → str, + ``b`` → bool, ``str``/``e`` → str, numeric → int when integral + else float. Missing row/cell (or a value-less cell) → None. + """ + row_idx, col_idx = parse_ref(ref) + row = self._find_row(row_idx) + if row is None: + return None + cell = self._find_cell(row, col_idx) + if cell is None: + return None + return self._read_value(cell) + + def get_formula(self, ref: str) -> str | None: + """The cell's formula text (````, without the leading ``=``), + or None if the cell has no formula.""" + row_idx, col_idx = parse_ref(ref) + row = self._find_row(row_idx) + if row is None: + return None + cell = self._find_cell(row, col_idx) + if cell is None: + return None + f = cell.find(qn("s:f")) + return None if f is None else (f.text or "") + + @property + def dimensions(self) -> tuple[int, int]: + """``(max_row, max_col)`` actually present in sheetData; (0, 0) + for an empty sheet.""" + max_row = max_col = 0 + prev_row = 0 + for row_el in self._sheet_data(): + if row_el.tag != qn("s:row"): + continue + prev_row = _row_number(row_el, prev_row) + max_row = max(max_row, prev_row) + prev_col = 0 + for c_el in row_el: + if c_el.tag != qn("s:c"): + continue + prev_col = _cell_column(c_el, prev_col) + max_col = max(max_col, prev_col) + return (max_row, max_col) + + @property + def merged_ranges(self) -> list[str]: + """The sheet's merged ranges, e.g. ``["A5:B5"]``.""" + return [ + mc.get("ref") + for mc in self._xp.findall("s:mergeCells/s:mergeCell") + if mc.get("ref") + ] + + def iter_rows( + self, min_row: int = 1, max_row: int | None = None + ) -> Iterator[tuple[str, object | None]]: + """Yield ``(ref, value)`` for every stored cell in row order.""" + prev_row = 0 + for row_el in self._sheet_data(): + if row_el.tag != qn("s:row"): + continue + prev_row = _row_number(row_el, prev_row) + if prev_row < min_row: + continue + if max_row is not None and prev_row > max_row: + break # rows are kept sorted by @r + prev_col = 0 + for c_el in row_el: + if c_el.tag != qn("s:c"): + continue + prev_col = _cell_column(c_el, prev_col) + ref = c_el.get("r") or f"{col_index_to_letters(prev_col)}{prev_row}" + yield ref, self._read_value(c_el) + + # -- writing --------------------------------------------------------------- + + def set_cell(self, ref: str, value: object) -> None: + """Surgically write one cell, preserving everything else. + + * str → ``t="inlineStr"`` (sharedStrings is never touched) + * bool → ``t="b"``, int/float → plain numeric, None → value + removed (the cell and its style stay) + * The cell's style attribute ``s`` is preserved. + * An existing ```` formula is **removed** — set_cell means + "this literal is the value now". ``fullCalcOnLoad`` is then + ensured so Excel recalculates any dependents on open. + """ + row_idx, col_idx = parse_ref(ref) + row = self._find_row(row_idx) + if row is None: + row = self._insert_row(row_idx) + cell = self._find_cell(row, col_idx) + if cell is None: + cell = self._insert_cell(row, row_idx, col_idx) + had_formula = cell.find(qn("s:f")) is not None + self._write_value(cell, value) + self._xp.mark_dirty() + self._extend_dimension(row_idx, col_idx) + if had_formula or self._doc._any_formula_exists(): + self._doc._ensure_full_calc_on_load() + + def append_rows(self, rows: list[list]) -> None: + """Append *rows* after the last existing row. ``None`` entries + leave their cell unstored (sparse), matching Excel semantics.""" + sd = self._sheet_data() + last = self.dimensions[0] + for offset, values in enumerate(rows, start=1): + row_idx = last + offset + row_el = sd.makeelement(qn("s:row"), {"r": str(row_idx)}) + sd.append(row_el) + for col_idx, value in enumerate(values, start=1): + if value is None: + continue + ref = f"{col_index_to_letters(col_idx)}{row_idx}" + c_el = row_el.makeelement(qn("s:c"), {"r": ref}) + row_el.append(c_el) + self._write_value(c_el, value) + if values: + self._extend_dimension(row_idx, len(values)) + else: + self._extend_dimension(row_idx, 1) + if rows: + self._xp.mark_dirty() + if self._doc._any_formula_exists(): + self._doc._ensure_full_calc_on_load() + + # -- internals --------------------------------------------------------------- + + def _find_row(self, row_idx: int) -> "_Element | None": + prev = 0 + for row_el in self._sheet_data(): + if row_el.tag != qn("s:row"): + continue + prev = _row_number(row_el, prev) + if prev == row_idx: + return row_el + if prev > row_idx: + return None # rows sorted by @r + return None + + def _find_cell(self, row_el: "_Element", col_idx: int) -> "_Element | None": + prev = 0 + for c_el in row_el: + if c_el.tag != qn("s:c"): + continue + prev = _cell_column(c_el, prev) + if prev == col_idx: + return c_el + if prev > col_idx: + return None # cells sorted by column + return None + + def _insert_row(self, row_idx: int) -> "_Element": + """Create ```` keeping sheetData sorted by @r + (Excel requires ascending rows).""" + sd = self._sheet_data() + new = sd.makeelement(qn("s:row"), {"r": str(row_idx)}) + prev = 0 + for row_el in sd: + if row_el.tag != qn("s:row"): + continue + prev = _row_number(row_el, prev) + if prev > row_idx: + row_el.addprevious(new) + return new + sd.append(new) + return new + + def _insert_cell( + self, row_el: "_Element", row_idx: int, col_idx: int + ) -> "_Element": + """Create ```` keeping the row's cells column-sorted.""" + ref = f"{col_index_to_letters(col_idx)}{row_idx}" + new = row_el.makeelement(qn("s:c"), {"r": ref}) + prev = 0 + for c_el in row_el: + if c_el.tag != qn("s:c"): + continue + prev = _cell_column(c_el, prev) + if prev > col_idx: + c_el.addprevious(new) + return new + row_el.append(new) + return new + + def _read_value(self, c_el: "_Element") -> object | None: + t = c_el.get("t", "n") + if t == "inlineStr": + is_el = c_el.find(qn("s:is")) + if is_el is None: + return None + return "".join(tel.text or "" for tel in is_el.iter(qn("s:t"))) + v = c_el.find(qn("s:v")) + if v is None or v.text is None: + return None + text = v.text + if t == "s": + strings = self._doc._shared_strings() + try: + return strings[int(text)] + except (ValueError, IndexError): + return None + if t == "b": + return text.strip() in ("1", "true", "TRUE") + if t in ("str", "e"): + return text + # default: numeric + try: + return int(text) + except ValueError: + f = float(text) + return int(f) if f.is_integer() else f + + def _write_value(self, c_el: "_Element", value: object) -> None: + """Replace the cell's content, preserving its style (@s). + + Any existing formula is removed — overwriting means the caller + supplies the value from now on.""" + for tag in ("s:f", "s:v", "s:is"): + el = c_el.find(qn(tag)) + if el is not None: + c_el.remove(el) + if value is None: + c_el.attrib.pop("t", None) + return + if isinstance(value, bool): # before int: bool is an int subclass + c_el.set("t", "b") + v = c_el.makeelement(qn("s:v"), {}) + v.text = "1" if value else "0" + c_el.append(v) + elif isinstance(value, (int, float)): + if isinstance(value, float) and not math.isfinite(value): + raise ValueError(f"Cannot store non-finite float {value!r}") + c_el.attrib.pop("t", None) # default numeric type + v = c_el.makeelement(qn("s:v"), {}) + v.text = repr(value) if isinstance(value, float) else str(value) + c_el.append(v) + elif isinstance(value, str): + c_el.set("t", "inlineStr") + is_el = c_el.makeelement(qn("s:is"), {}) + t_el = is_el.makeelement(qn("s:t"), {_XML_SPACE: "preserve"}) + t_el.text = value + is_el.append(t_el) + c_el.append(is_el) + else: + raise TypeError( + f"Unsupported cell value type {type(value).__name__!r} " + "(str, bool, int, float or None)" + ) + + def _extend_dimension(self, row_idx: int, col_idx: int) -> None: + """Grow the sheet's ```` to include the cell.""" + dim = self._xp.find("s:dimension") + if dim is None: + return + m = _RANGE_RE.match(dim.get("ref") or "") + if m is None: + return + min_c, min_r = col_letters_to_index(m.group(1)), int(m.group(2)) + if m.group(3): + max_c, max_r = col_letters_to_index(m.group(3)), int(m.group(4)) + else: + max_c, max_r = min_c, min_r + new_min_c, new_min_r = min(min_c, col_idx), min(min_r, row_idx) + new_max_c, new_max_r = max(max_c, col_idx), max(max_r, row_idx) + if (new_min_c, new_min_r, new_max_c, new_max_r) != ( + min_c, + min_r, + max_c, + max_r, + ): + dim.set( + "ref", + f"{col_index_to_letters(new_min_c)}{new_min_r}" + f":{col_index_to_letters(new_max_c)}{new_max_r}", + ) + self._xp.mark_dirty() + + +class SheetCollection: + """Mapping-like sheet accessor: by name (``raw.sheets["Sales"]``) or + by position (``raw.sheets[0]``).""" + + def __init__(self, doc: "XlsxRawDocument"): + self._doc = doc + + def __getitem__(self, key: int | str) -> RawSheet: + entries = self._doc._sheet_entries() + if isinstance(key, int): + name, part = entries[key] # IndexError speaks for itself + else: + for name, part in entries: + if name == key: + break + else: + raise KeyError(f"No sheet named {key!r}") + return self._doc._sheet(name, part) + + def __len__(self) -> int: + return len(self._doc._sheet_entries()) + + def __iter__(self) -> Iterator[RawSheet]: + for name, part in self._doc._sheet_entries(): + yield self._doc._sheet(name, part) + + def __contains__(self, name: object) -> bool: + return any(n == name for n, _ in self._doc._sheet_entries()) + + +class XlsxRawDocument(RawDocumentBase): + """Lossless, writable view of an .xlsx workbook.""" + + format = "xlsx" + + def __init__(self, package: "OpcPackage"): + super().__init__(package) + self._workbook_name = self._locate_workbook() + self._sheet_cache: dict[str, RawSheet] = {} + self._shared_cache: list[str] | None = None + self._formulas_exist: bool | None = None + + def _locate_workbook(self) -> str: + rels = self.package.rels_for("") + if rels is not None: + for rel in rels.by_type("/officeDocument"): + return rels.resolve("", rel["target"]) + return "xl/workbook.xml" + + @property + def workbook(self) -> XmlPart: + """The ``xl/workbook.xml`` facade.""" + return self.xml_part(self._workbook_name) + + # -- sheets --------------------------------------------------------------- + + @property + def sheet_names(self) -> list[str]: + """Sheet names in workbook (tab) order.""" + return [s.get("name") or "" for s in self.workbook.findall("s:sheets/s:sheet")] + + @property + def sheets(self) -> SheetCollection: + return SheetCollection(self) + + def _sheet_entries(self) -> list[tuple[str, str]]: + """``(name, worksheet part name)`` in workbook order.""" + rels = self.package.rels_for(self._workbook_name) + entries: list[tuple[str, str]] = [] + for sheet in self.workbook.findall("s:sheets/s:sheet"): + rid = sheet.get(qn("r:id")) + target = rels.target_of(rid) if (rels is not None and rid) else None + if target is None: + continue + entries.append( + (sheet.get("name") or "", rels.resolve(self._workbook_name, target)) + ) + return entries + + def _sheet(self, name: str, part_name: str) -> RawSheet: + if part_name not in self._sheet_cache: + self._sheet_cache[part_name] = RawSheet(self, name, part_name) + return self._sheet_cache[part_name] + + # -- sheet structure (add / copy / move / rename / delete) ------------------ + + def add_sheet(self, name: str, *, at: int | None = None) -> RawSheet: + """Create an empty worksheet named *name* at tab position *at* + (default: last). Wires the four OPC touchpoints — worksheet part, + content-type Override, workbook rel, ```` element — leaving + everything else byte-identical.""" + self._validate_new_sheet_name(name) + part = self._fresh_worksheet_part() + self.package.add_part(part, _MINIMAL_WORKSHEET_XML) + self.package.set_content_type_override(part, _WORKSHEET_CONTENT_TYPE) + rid = self._add_workbook_rel(part) + self._insert_sheet_element(name, rid, at) + self._invalidate() + return self._sheet(name, part) + + def copy_sheet( + self, key: int | str, new_name: str, *, at: int | None = None + ) -> RawSheet: + """Insert an independent copy of a sheet (by name or index) as + *new_name* at tab position *at* (default: after the source). + + The worksheet part and its drawing/chart/table subtree are deep- + copied (``clone_part_graph``); images are SHARED, so editing the + copy's cells or chart data never touches the original. v1 copies + cells + drawings + charts + native tables; it does NOT rewrite + formulas or defined names that reference the source tab by name.""" + self._validate_new_sheet_name(new_name) + _, src_part = self._resolve_sheet(key) + pos = self._tab_index(src_part) + 1 if at is None else at + new_ws, _ = self.package.clone_part_graph( + src_part, rename=make_part_renamer(self.package), share_types=("/image",) + ) + rid = self._add_workbook_rel(new_ws) + self._insert_sheet_element(new_name, rid, pos) + self._invalidate() + return self._sheet(new_name, new_ws) + + def move_sheet(self, key: int | str, to: int) -> None: + """Move a sheet (by name or index) to tab position *to*. Reorders + the ```` element in ``xl/workbook.xml`` only; no part is + touched (bookViews/activeTab is left as-is).""" + name, _ = self._resolve_sheet(key) + sheets = self._sheets_el() + els = sheets.findall(qn("s:sheet")) + n = len(els) + if not 0 <= to < n: + raise IndexError(f"destination {to} out of range (0..{n - 1})") + idx = next(i for i, s in enumerate(els) if s.get("name") == name) + if idx == to: + return + el = els[idx] + sheets.remove(el) + remaining = sheets.findall(qn("s:sheet")) + if to >= len(remaining): + sheets.append(el) + else: + remaining[to].addprevious(el) + self.workbook.mark_dirty() + self._invalidate() + + def rename_sheet(self, key: int | str, new_name: str) -> None: + """Rename a sheet's tab. v1 sets the ```` attribute + only — it does NOT rewrite formulas / defined names that reference + the old name (the edit2docs verb surfaces a warning when any do).""" + old_name, _ = self._resolve_sheet(key) + if new_name == old_name: + return + self._validate_new_sheet_name(new_name) + for s in self._sheets_el().findall(qn("s:sheet")): + if s.get("name") == old_name: + s.set("name", new_name) + self.workbook.mark_dirty() + break + self._invalidate() + + def delete_sheet(self, key: int | str) -> None: + """Remove a sheet and everything only it anchored (its drawing, + charts, embeddings, native tables), reference-counted against the + surviving sheets — the same orphan sweep as + ``PptxRawDocument.remove_slide``. Refuses to delete the last sheet + (a workbook needs at least one).""" + if len(self._sheet_entries()) <= 1: + raise ValueError("cannot delete the only sheet in a workbook") + name, part = self._resolve_sheet(key) + sheets = self._sheets_el() + rid = None + for s in sheets.findall(qn("s:sheet")): + if s.get("name") == name: + rid = s.get(qn("r:id")) + sheets.remove(s) + break + self.workbook.mark_dirty() + rels = self.package.rels_for(self._workbook_name) + if rid is not None and rels is not None: + rels.remove(rid) + self._sweep_orphans([part]) + self._invalidate() + + # -- sheet-structure internals --------------------------------------------- + + def _resolve_sheet(self, key: int | str) -> tuple[str, str]: + """``(name, worksheet part)`` for a sheet by tab index or name.""" + entries = self._sheet_entries() + if isinstance(key, int): + if not 0 <= key < len(entries): + raise IndexError( + f"sheet index {key} out of range (0..{len(entries) - 1})" + ) + return entries[key] + for name, part in entries: + if name == key: + return name, part + raise KeyError(f"No sheet named {key!r}") + + def _tab_index(self, part_name: str) -> int: + for i, (_, part) in enumerate(self._sheet_entries()): + if part == part_name: + return i + return len(self._sheet_entries()) + + def _validate_new_sheet_name(self, name: str) -> None: + if not name or len(name) > 31: + raise ValueError(f"sheet name must be 1..31 chars (got {len(name)})") + bad = set(name) & _FORBIDDEN_SHEET_CHARS + if bad: + raise ValueError( + f"sheet name {name!r} has forbidden characters: {sorted(bad)}" + ) + if name in self.sheet_names: + raise ValueError(f"a sheet named {name!r} already exists") + + def _fresh_worksheet_part(self) -> str: + n = 1 + while self.package.has_part(f"xl/worksheets/sheet{n}.xml"): + n += 1 + return f"xl/worksheets/sheet{n}.xml" + + def _add_workbook_rel(self, worksheet_part: str) -> str: + rels = self.package.rels_for(self._workbook_name) + if rels is None: + raise ValueError("workbook has no relationships part") + rid = rels.next_id() + rels.add( + rid, + _WORKSHEET_REL_TYPE, + posixpath.relpath(worksheet_part, posixpath.dirname(self._workbook_name)), + ) + return rid + + def _sheets_el(self): + sheets = self.workbook.find("s:sheets") + if sheets is None: + raise ValueError("workbook.xml has no element") + return sheets + + def _insert_sheet_element(self, name: str, rid: str, at: int | None) -> None: + sheets = self._sheets_el() + existing = sheets.findall(qn("s:sheet")) + ids = [ + int(s.get("sheetId", "0")) + for s in existing + if (s.get("sheetId") or "0").isdigit() + ] + el = sheets.makeelement(qn("s:sheet"), {}) + el.set("name", name) + el.set("sheetId", str(max(ids, default=0) + 1)) + el.set(qn("r:id"), rid) + if at is None or at >= len(existing): + sheets.append(el) + else: + existing[max(0, at)].addprevious(el) + self.workbook.mark_dirty() + + def _invalidate(self) -> None: + self._sheet_cache.clear() + self._shared_cache = None + self._formulas_exist = None + + # -- workbook-level reads ---------------------------------------------------- + + @property + def defined_names(self) -> dict[str, str]: + """``definedName`` → refers-to text (read-only view).""" + out: dict[str, str] = {} + for dn in self.workbook.findall("s:definedNames/s:definedName"): + name = dn.get("name") + if name: + out[name] = dn.text or "" + return out + + # -- charts --------------------------------------------------------------- + + @property + def chart_part_names(self) -> list[str]: + """Chart part names reachable from the sheets' drawings, in + sheet order (duplicates removed).""" + from xgen_edit2docs.raw.chart import find_chart_parts + + names: list[str] = [] + seen: set[str] = set() + for _, ws_name in self._sheet_entries(): + if not self.package.has_part(ws_name): + continue + drawing = self.xml_part(ws_name).find("s:drawing") + if drawing is None: + continue + rid = drawing.get(qn("r:id")) + rels = self.package.rels_for(ws_name) + target = rels.target_of(rid) if (rels is not None and rid) else None + if target is None: + continue + drawing_part = rels.resolve(ws_name, target) + for chart_name in find_chart_parts(self.package, drawing_part): + if chart_name not in seen: + seen.add(chart_name) + names.append(chart_name) + return names + + @property + def charts(self) -> list["ChartModel"]: + """ChartModel per chart part (raises NotImplementedError until + the C3 chart milestone lands; use :attr:`chart_part_names` for + discovery in the meantime).""" + from xgen_edit2docs.raw.chart import ChartModel + + models: list["ChartModel"] = [] + for name in self.chart_part_names: + try: + models.append(ChartModel(self.xml_part(name), self.package)) + except NotImplementedError: + raise NotImplementedError( + "ChartModel is not implemented yet (milestone C3); " + "chart_part_names still lists the chart parts" + ) from None + return models + + # -- shared strings ------------------------------------------------------------ + + def _shared_strings(self) -> list[str]: + """The shared-string table (read-only, cached). A workbook with + no sharedStrings part simply yields an empty table — writes use + inline strings, so the part is never created.""" + if self._shared_cache is None: + name = "xl/sharedStrings.xml" + rels = self.package.rels_for(self._workbook_name) + if rels is not None: + for rel in rels.by_type("/sharedStrings"): + name = rels.resolve(self._workbook_name, rel["target"]) + break + if not self.package.has_part(name): + self._shared_cache = [] + else: + root = self.xml_part(name).root + self._shared_cache = [ + "".join(t.text or "" for t in si.iter(qn("s:t"))) + for si in root.findall(qn("s:si")) + ] + return self._shared_cache + + # -- formula staleness --------------------------------------------------------- + + def _any_formula_exists(self) -> bool: + """Whether any worksheet contains a formula (cached; used to + decide if edits make cached results stale).""" + if self._formulas_exist is None: + self._formulas_exist = any( + self._part_has_formula(part) for _, part in self._sheet_entries() + ) + return self._formulas_exist + + def _part_has_formula(self, part_name: str) -> bool: + if not self.package.has_part(part_name): + return False + xp = self._xml_parts.get(part_name) + if xp is not None and xp.loaded: + return next(xp.root.iter(qn("s:f")), None) is not None + # Unparsed part: cheap byte scan (an element in the default + # spreadsheetml namespace; false positives merely add calcPr). + data = self.package.get_part(part_name).read() + return b"" in data or b" None: + """Guarantee ```` in workbook.xml so + Excel recalculates stale formula caches on open. No-op (and no + rewrite) when the flag is already set.""" + wb = self.workbook + calc = wb.find("s:calcPr") + if calc is None: + calc = wb.root.makeelement(qn("s:calcPr"), {"fullCalcOnLoad": "1"}) + preceding = {qn(f"s:{tag}") for tag in _PRE_CALC_PR} + pos = 0 + for i, child in enumerate(wb.root): + if child.tag in preceding: + pos = i + 1 + wb.root.insert(pos, calc) + wb.mark_dirty() + elif calc.get("fullCalcOnLoad") != "1": + calc.set("fullCalcOnLoad", "1") + wb.mark_dirty() diff --git a/src/xgen_edit2docs/raw/xmlpart.py b/src/xgen_edit2docs/raw/xmlpart.py new file mode 100644 index 00000000..d4ed48e5 --- /dev/null +++ b/src/xgen_edit2docs/raw/xmlpart.py @@ -0,0 +1,121 @@ +# xgen_edit2docs/raw/xmlpart.py (vendored from xgen-contextifier 0.9.0, Apache-2.0) +""" +Lazy lxml view over an :class:`~xgen_edit2docs.raw.opc.OpcPart`. + +Format models hold ``XmlPart`` facades for the parts they understand. +The tree is parsed on first access and serialized back into the package +only when the model mutated it (``mark_dirty()`` + ``flush()``), so +opening a document and reading a few values never rewrites anything — +the byte-preservation contract stays intact by construction. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: # pragma: no cover + from lxml.etree import _Element + + from xgen_edit2docs.raw.opc import OpcPart + +__all__ = ["NS", "qn", "XmlPart"] + +# The OOXML namespace registry shared by the raw layer. Prefixes follow +# the conventional short names used across ECMA-376 documentation. +NS: dict[str, str] = { + # wordprocessingml + "w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main", + # drawingml + "a": "http://schemas.openxmlformats.org/drawingml/2006/main", + "c": "http://schemas.openxmlformats.org/drawingml/2006/chart", + "cx": "http://schemas.microsoft.com/office/drawing/2014/chartex", + "xdr": "http://schemas.openxmlformats.org/drawingml/2006/spreadsheetDrawing", + "wp": "http://schemas.openxmlformats.org/drawingml/2006/wordprocessingDrawing", + # presentationml + "p": "http://schemas.openxmlformats.org/presentationml/2006/main", + # spreadsheetml + "s": "http://schemas.openxmlformats.org/spreadsheetml/2006/main", + # shared + "r": "http://schemas.openxmlformats.org/officeDocument/2006/relationships", + "ct": "http://schemas.openxmlformats.org/package/2006/content-types", + "pr": "http://schemas.openxmlformats.org/package/2006/relationships", + "mc": "http://schemas.openxmlformats.org/markup-compatibility/2006", +} + + +def qn(tag: str) -> str: + """``"w:p"`` → ``"{http://…/wordprocessingml/2006/main}p"``.""" + prefix, _, local = tag.partition(":") + if not local: + return tag + try: + return f"{{{NS[prefix]}}}{local}" + except KeyError: + raise KeyError(f"Unknown XML namespace prefix: {prefix!r}") from None + + +class XmlPart: + """A parsed XML part with explicit dirty tracking. + + ``root`` parses lazily; call :meth:`mark_dirty` after mutating the + tree and :meth:`flush` (usually via the owning document's ``save``) + to serialize back into the package part. + """ + + __slots__ = ("part", "_root", "_dirty") + + def __init__(self, part: "OpcPart"): + self.part = part + self._root: "_Element | None" = None + self._dirty = False + + @property + def name(self) -> str: + return self.part.name + + @property + def root(self) -> "_Element": + if self._root is None: + from lxml import etree + + self._root = etree.fromstring(self.part.read()) + return self._root + + @property + def loaded(self) -> bool: + return self._root is not None + + @property + def dirty(self) -> bool: + return self._dirty + + def mark_dirty(self) -> None: + self._dirty = True + + def flush(self) -> None: + """Serialize the tree into the part iff this facade dirtied it.""" + if self._dirty and self._root is not None: + from lxml import etree + + self.part.write( + etree.tostring( + self._root, xml_declaration=True, encoding="UTF-8", standalone=True + ) + ) + self._dirty = False + + # -- conveniences ----------------------------------------------------------- + + def find(self, path: str): + """`find` with the shared namespace map (``"s:sheetData/s:row"``).""" + return self.root.find(path, NS) + + def findall(self, path: str): + return self.root.findall(path, NS) + + def iter(self, tag: str): + return self.root.iter(qn(tag)) + + def __repr__(self) -> str: # pragma: no cover - debug aid + state = "dirty" if self._dirty else ("loaded" if self.loaded else "lazy") + return f"" diff --git a/src/xgen_edit2docs/render/__init__.py b/src/xgen_edit2docs/render/__init__.py index 6d65510e..edf6380c 100644 --- a/src/xgen_edit2docs/render/__init__.py +++ b/src/xgen_edit2docs/render/__init__.py @@ -8,7 +8,7 @@ - :func:`svg_to_png` / :func:`svgs_to_pngs` — resvg (self-contained Rust rasterizer wheel; gradients/clipPath/CJK verified) at a chosen DPI. -- :func:`svgs_to_pdf` — raster pages assembled into a PDF by PyMuPDF +- :func:`svgs_to_pdf` — raster pages assembled into a PDF by xgen-pdf (already a core dependency). - :class:`FontResolver` — system-font discovery + fontTools metrics so layout code can measure real advance widths instead of the fixed diff --git a/src/xgen_edit2docs/render/rasterize.py b/src/xgen_edit2docs/render/rasterize.py index 2a6eb1b0..04e27b05 100644 --- a/src/xgen_edit2docs/render/rasterize.py +++ b/src/xgen_edit2docs/render/rasterize.py @@ -1,11 +1,11 @@ -"""SVG → PNG / PDF rasterization (resvg + PyMuPDF). +"""SVG → PNG / PDF rasterization (resvg + xgen-pdf). Backend choice is deliberate (verified by spike, 2026-07-05): - ``resvg`` (via the self-contained ``resvg-py`` wheel) renders the SVG our converters emit — linear/radial gradients, clipPath, stroke-dasharray, CJK text — faithfully, with no system libraries. -- PyMuPDF's built-in SVG parser was disqualified: it rasterizes +- A PDF library's built-in SVG parser was disqualified: it rasterizes gradient/pattern fills as black boxes. It is still the right tool for the *assembly* step (PNG pages → one PDF), which is pure image placement. @@ -100,7 +100,7 @@ def svgs_to_pdf( backend (cairosvg, system cairo) can slot in later behind the same signature — see docs/native-render-plan.md. """ - import fitz # PyMuPDF — core dependency + import xgen_pdf as fitz # pdfium-based engine (xgen-pdf) if not svgs: raise ValueError("svgs_to_pdf needs at least one SVG page") diff --git a/src/xgen_edit2docs/simple.py b/src/xgen_edit2docs/simple.py index 61c197d5..05d1811e 100644 --- a/src/xgen_edit2docs/simple.py +++ b/src/xgen_edit2docs/simple.py @@ -510,7 +510,7 @@ def render_doc( """Render a document to page images or a PDF — no LibreOffice. Native raster pipeline (docs/native-render-plan.md): per-page SVG → - resvg PNG → (for ``to="pdf"``) PyMuPDF assembly. Deterministic, no + resvg PNG → (for ``to="pdf"``) xgen-pdf assembly. Deterministic, no LLM, no subprocess. Args: diff --git a/src/xgen_edit2docs/tools/edit_deck.py b/src/xgen_edit2docs/tools/edit_deck.py index 0ad90974..83d6c178 100644 --- a/src/xgen_edit2docs/tools/edit_deck.py +++ b/src/xgen_edit2docs/tools/edit_deck.py @@ -449,7 +449,7 @@ def _native_inventory(pptx: bytes) -> list[_SlideNatives]: planner simply loses the annotation, never the turn. """ try: - from xgen_contextifier import open_raw + from xgen_edit2docs.raw import open_raw except Exception: # pragma: no cover - xgen_contextifier is a hard dep return [] try: diff --git a/tests/unit/test_render.py b/tests/unit/test_render.py index fe28a6f4..04cb5216 100644 --- a/tests/unit/test_render.py +++ b/tests/unit/test_render.py @@ -81,7 +81,7 @@ def test_svgs_to_pngs_naming_and_stale_cleanup(self, tmp_path): assert not (out / "page-9.png").exists() # shrunk docs drop old pages def test_svgs_to_pdf_page_count_and_size(self): - import fitz + import xgen_pdf as fitz pdf = svgs_to_pdf([GRADIENT_SVG, GRADIENT_SVG, GRADIENT_SVG], dpi=144) with fitz.open(stream=pdf, filetype="pdf") as doc: @@ -106,7 +106,7 @@ def test_pptx_to_png_pages(self, deck_path): assert result.paths[0].read_bytes().startswith(PNG_MAGIC) def test_pptx_to_pdf(self, deck_path, tmp_path): - import fitz + import xgen_pdf as fitz result = render_doc(deck_path, to="pdf", out_dir=tmp_path / "pdf") assert result.paths[0].name == "deck.pdf"