diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 10bb90d0..0d39f32c 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -2,8 +2,15 @@ name: Tests on: push: + branches: [main] pull_request: +# A new commit on a pull request cancels the run of the one before it. Runs on main are +# never cancelled: each has its own group. +concurrency: + group: tests-${{ github.event.pull_request.number || github.sha }} + cancel-in-progress: ${{ github.event_name == 'pull_request' }} + jobs: test: runs-on: ubuntu-latest diff --git a/src/hwpx/_document/headings.py b/src/hwpx/_document/headings.py index 252db073..9d0f387e 100644 --- a/src/hwpx/_document/headings.py +++ b/src/hwpx/_document/headings.py @@ -170,13 +170,11 @@ def bind_outline_level( from . import layout as _layout _check_level(level) - target = paragraph.element - for index, candidate in enumerate(document.paragraphs): - if candidate.element is target: - _layout.set_paragraph_format( - document, paragraph_index=index, outline_level=level - ) - return + # The paragraph itself is the target: looking its index up in the list of every body paragraph, + # and resolving that index back, made a run of headings quadratic. + if _layout._body_indexes(document, [paragraph])[0] is not None: + _layout.set_paragraph_format(document, paragraphs=[paragraph], outline_level=level) + return # 방금 만든 문단이라 항상 찾힌다. 못 찾으면 조용히 넘어가는 대신 말한다 — # 개요 수준이 빠진 문단은 "번호가 안 붙는 제목"이라 무음 실패가 가장 나쁘다. raise HwpxLookupError( # pragma: no cover - 방어 diff --git a/src/hwpx/_document/layout.py b/src/hwpx/_document/layout.py index a2684493..1697be55 100644 --- a/src/hwpx/_document/layout.py +++ b/src/hwpx/_document/layout.py @@ -232,6 +232,48 @@ def _resolve_paragraph_objects( return resolved +def _body_indexes( + doc: "HwpxDocument", objects: Sequence[HwpxOxmlParagraph] +) -> list[int | None]: + """Where each paragraph object sits in ``doc.paragraphs``; ``None`` for one outside the body. + + A few objects are placed by counting the paragraphs before them in their section, which reads + no other paragraph; wrapping every body paragraph to place one made a run of single-paragraph + calls (a heading per call) quadratic. Many objects, or plain ElementTree elements, go through + the list of body paragraphs once. + """ + + elements: list[Any] = [paragraph.element for paragraph in objects] + if len(elements) > 8 or not all(hasattr(element, "itersiblings") for element in elements): + body_index = {paragraph.element: index for index, paragraph in enumerate(doc.paragraphs)} + return [body_index.get(element) for element in elements] + + section_roots: list[Any] = [section.element for section in doc.sections] + starts: list[int] = [] # body index of each section's first paragraph, filled as far as needed + + def start_of(section_position: int) -> int: + while len(starts) <= section_position: + previous = len(starts) - 1 + starts.append( + 0 if previous < 0 + else starts[previous] + sum(1 for _ in section_roots[previous].iterchildren("{*}p")) + ) + return starts[section_position] + + indexes: list[int | None] = [] + for element in elements: + parent = element.getparent() + position = next( + (index for index, root in enumerate(section_roots) if root is parent), None + ) + if position is None or not isinstance(element.tag, str) or element.tag.rsplit("}", 1)[-1] != "p": + indexes.append(None) + continue + before = sum(1 for _ in element.itersiblings("{*}p", preceding=True)) + indexes.append(start_of(position) + before) + return indexes + + def set_paragraph_format( doc: "HwpxDocument", *, @@ -376,9 +418,7 @@ def set_paragraph_format( suggestion="Pass only one.", ) objects = _resolve_paragraph_objects(doc, paragraphs) - body = doc.paragraphs - body_index = {paragraph.element: index for index, paragraph in enumerate(body)} - targets = [(body_index.get(paragraph.element), paragraph) for paragraph in objects] + targets = list(zip(_body_indexes(doc, objects), objects)) else: targets = list(_resolve_paragraph_targets(doc, paragraph_index=paragraph_index, diff --git a/src/hwpx/_document/persistence.py b/src/hwpx/_document/persistence.py index ab8fbc5d..1b60d158 100644 --- a/src/hwpx/_document/persistence.py +++ b/src/hwpx/_document/persistence.py @@ -435,6 +435,7 @@ def save_report( before=before, edit_mask=edit_mask, ledger=ledger, + open_safety=_serialize_open_safety(doc, archive_bytes), reference_document=doc, source_label="document.save_report", ) @@ -475,6 +476,25 @@ def to_bytes( return data +def _serialize_open_safety(doc: "HwpxDocument", archive_bytes: bytes) -> OpenSafetyReport | None: + """The open-safety verdict the package's save reached for exactly *archive_bytes*, in the form + the SavePipeline composes from its own run; ``None`` when the package did not validate these + bytes (the pipeline then validates them itself). + + The same validation of the same bytes gives the same report, so ``save_report`` hands the + pipeline this one instead of running it a second time. + """ + + package = doc._package + report = getattr(package, "_verified_archive_report", None) + digest = getattr(package, "_verified_archive_digest", None) + if report is None or digest is None or not report.ok: + return None + if digest != hashlib.sha256(archive_bytes).digest(): + return None + return OpenSafetyReport(ok=report.ok, summary=report.summary, detail=report.to_dict()) + + def _to_bytes_raw( doc: "HwpxDocument", *, diff --git a/src/hwpx/opc/package.py b/src/hwpx/opc/package.py index 98bbf432..ce652255 100644 --- a/src/hwpx/opc/package.py +++ b/src/hwpx/opc/package.py @@ -441,6 +441,8 @@ def __init__( # SHA-256 of the last archive bytes this package's save verified as # editor-open safe (see _save_to_zip); None until such a save. self._verified_archive_digest: bytes | None = None + # The report of that validation, for a caller that needs its evidence for the same bytes. + self._verified_archive_report: "EditorOpenSafetyReport | None" = None self._validate_structure() @staticmethod @@ -1230,11 +1232,9 @@ def _save_to_zip( report = self._verify_editor_open_safe_archive(payload) # Remember which exact bytes passed, so the document-level # check of the same save need not run the validation again. - self._verified_archive_digest = ( - hashlib.sha256(payload).digest() - if report is not None and report.ok - else None - ) + verified = report is not None and report.ok + self._verified_archive_digest = hashlib.sha256(payload).digest() if verified else None + self._verified_archive_report = report if verified else None _write_stream_or_rollback(pkg_file, payload) if mark_clean and version_was_dirty: self._version.mark_clean() diff --git a/src/hwpx/oxml/header_part.py b/src/hwpx/oxml/header_part.py index 9e13675b..0c0a9956 100644 --- a/src/hwpx/oxml/header_part.py +++ b/src/hwpx/oxml/header_part.py @@ -87,6 +87,9 @@ def __init__(self, part_name: str, element: ET.Element, document: "HwpxOxmlDocum self._element = element self._dirty = False self._document = document + # parsed lookups (paragraph shapes, styles, tab definitions, border fills ...), see _parsed() + self._lookup_cache: dict[tuple[object, ET.Element], tuple[bytes, object]] = {} + self._para_pr_index: tuple[object, dict[str, ET.Element]] | None = None @property def element(self) -> ET.Element: @@ -1225,7 +1228,7 @@ def border_fills(self) -> dict[str, GenericElement]: if element is None: return {} - fill_list = parse_border_fills(self._convert_to_lxml(element)) + fill_list = self._parsed(element, parse_border_fills) mapping: dict[str, GenericElement] = {} for border_fill in fill_list.fills: raw_id = border_fill.attributes.get("id") @@ -1248,8 +1251,32 @@ def border_fill(self, border_fill_id_ref: int | str | None) -> GenericElement | @staticmethod def _convert_to_lxml(element: ET.Element) -> LET._Element: + # The header tree is lxml already; only a plain ElementTree element needs the round trip. + if isinstance(element, LET._Element): + return element return LET.fromstring(ET.tostring(element, encoding="utf-8")) + def _parsed(self, element: ET.Element, parse: Callable[[LET._Element], T]) -> T: + """``parse(element)``, kept until the header is marked dirty or the element's XML differs. + + The kept result is compared by the element's serialized bytes, so an edit made on the tree + without ``mark_dirty()`` is seen as well. Callers get the kept objects themselves, as + ``char_property`` gives its cached styles: treat them as read-only. + """ + + if not isinstance(element, LET._Element): + return parse(self._convert_to_lxml(element)) + stamp = LET.tostring(element) + key = (parse, element) + kept = self._lookup_cache.get(key) + if kept is not None and kept[0] == stamp: + return kept[1] # type: ignore[return-value] + value = parse(element) + if len(self._lookup_cache) >= 4096: + self._lookup_cache.clear() + self._lookup_cache[key] = (stamp, value) + return value + @staticmethod def _lookup_by_id(mapping: dict[str, T], identifier: int | str | None) -> T | None: if identifier is None: @@ -1397,7 +1424,7 @@ def bullets(self) -> dict[str, Bullet]: if bullets_element is None: return {} - bullet_list = parse_bullets(self._convert_to_lxml(bullets_element)) + bullet_list = self._parsed(bullets_element, parse_bullets) return bullet_list.as_dict() def bullet(self, bullet_id_ref: int | str | None) -> Bullet | None: @@ -1409,9 +1436,7 @@ def paragraph_properties(self) -> dict[str, ParagraphProperty]: if para_props_element is None: return {} - para_properties = parse_paragraph_properties( - self._convert_to_lxml(para_props_element) - ) + para_properties = self._parsed(para_props_element, parse_paragraph_properties) return para_properties.as_dict() def paragraph_property( @@ -1419,22 +1444,30 @@ def paragraph_property( ) -> ParagraphProperty | None: # Parse only the hh:paraPr asked for, under the keys paragraph_properties gives it; parsing # them all on each lookup cost most of a second in a header of many paragraph shapes. - elements: dict[str, ET.Element] = {} parent = self._para_properties_element() - for child in [] if parent is None else parent: - raw = child.get("id") if _element_local_name(child) == "paraPr" else None - for key in (raw, str(int(raw)) if raw and raw.strip().lstrip("+-").isdigit() else None): - if key and key not in elements: - elements[key] = child + children = [] if parent is None else list(parent) + # The id -> element index is kept while the list has the same children with the same ids. + stamp = (children, [(child.tag, child.get("id")) for child in children]) + kept = self._para_pr_index + if kept is not None and kept[0] == stamp: + elements = kept[1] + else: + elements = {} + for child in children: + raw = child.get("id") if _element_local_name(child) == "paraPr" else None + for key in (raw, str(int(raw)) if raw and raw.strip().lstrip("+-").isdigit() else None): + if key and key not in elements: + elements[key] = child + self._para_pr_index = (stamp, elements) found = self._lookup_by_id(elements, para_pr_id_ref) - return None if found is None else parse_paragraph_property(self._convert_to_lxml(found)) + return None if found is None else self._parsed(found, parse_paragraph_property) @property def tab_properties(self) -> dict[str, TabDefinition]: element = self._tab_properties_element() if element is None: return {} - return parse_tab_definitions(self._convert_to_lxml(element)).as_dict() + return self._parsed(element, parse_tab_definitions).as_dict() def tab_property(self, tab_pr_id_ref: int | str | None) -> TabDefinition | None: return self._lookup_by_id(self.tab_properties, tab_pr_id_ref) @@ -1445,7 +1478,7 @@ def styles(self) -> dict[str, Style]: if styles_element is None: return {} - style_list = parse_styles(self._convert_to_lxml(styles_element)) + style_list = self._parsed(styles_element, parse_styles) return style_list.as_dict() def style(self, style_id_ref: int | str | None) -> Style | None: @@ -1457,7 +1490,7 @@ def track_changes(self) -> dict[str, TrackChange]: if changes_element is None: return {} - change_list = parse_track_changes(self._convert_to_lxml(changes_element)) + change_list = self._parsed(changes_element, parse_track_changes) return change_list.as_dict() def track_change(self, change_id_ref: int | str | None) -> TrackChange | None: @@ -1469,9 +1502,7 @@ def track_change_authors(self) -> dict[str, TrackChangeAuthor]: if authors_element is None: return {} - author_list = parse_track_change_authors( - self._convert_to_lxml(authors_element) - ) + author_list = self._parsed(authors_element, parse_track_change_authors) return author_list.as_dict() def track_change_author( @@ -1485,6 +1516,8 @@ def dirty(self) -> bool: def mark_dirty(self) -> None: self._dirty = True + self._lookup_cache.clear() + self._para_pr_index = None def reset_dirty(self) -> None: self._dirty = False diff --git a/src/hwpx/oxml/section.py b/src/hwpx/oxml/section.py index f9af005d..1dfef86d 100644 --- a/src/hwpx/oxml/section.py +++ b/src/hwpx/oxml/section.py @@ -357,10 +357,14 @@ def _inherited_paragraph_refs( every following paragraph turning into a list item without this guard. """ - existing = self.paragraphs - if not existing: + # Only the last paragraph is read: wrapping every paragraph of the section for each append + # made a long run of appends quadratic. + last_element = next( + (child for child in reversed(self._element) if tag_local_name(child.tag) == "p"), None + ) + if last_element is None: return None, None, None - last = existing[-1] + last = HwpxOxmlParagraph(last_element, self) prev_para_ref = last.para_pr_id_ref prev_style_ref = last.style_id_ref prev_char_ref = last.char_pr_id_ref diff --git a/src/hwpx/oxml/table.py b/src/hwpx/oxml/table.py index 5a25a726..29ee4d86 100644 --- a/src/hwpx/oxml/table.py +++ b/src/hwpx/oxml/table.py @@ -36,7 +36,7 @@ from .body import Label, parse_label_element from .objects import Caption, _read_caption, _remove_caption, _write_caption -from .table_merge import collapse_merged_grid +from .table_merge import TableGridCache as _TableGridCache, collapse_merged_grid if TYPE_CHECKING: from hwpx.form_fit.policy import FitPolicy @@ -293,6 +293,7 @@ def set_span(self, row_span: int, col_span: int) -> None: span = self._span_element() span.set("rowSpan", str(max(row_span, 1))) span.set("colSpan", str(max(col_span, 1))) + self.table._invalidate_grid() self.table.mark_dirty() @property @@ -311,6 +312,7 @@ def set_size(self, width: int | None = None, height: int | None = None) -> None: size.set("width", str(max(width, 0))) if height is not None: size.set("height", str(max(height, 0))) + self.table._invalidate_grid() self.table.mark_dirty() margins = property(_table_sizes.cell_margins) @@ -588,6 +590,7 @@ def col_span(self) -> int: return self.span[1] + class HwpxOxmlTableRow: """Represents a table row.""" @@ -609,6 +612,7 @@ class HwpxOxmlTable(TableStructureEdits): def __init__(self, element: ET.Element, paragraph: "HwpxOxmlParagraph"): self.element = element self.paragraph = paragraph + self._grid_cache: _TableGridCache | None = None def __repr__(self) -> str: """Return a compact and safe summary of table geometry.""" @@ -883,6 +887,24 @@ def column_count(self) -> int: def rows(self) -> list[HwpxOxmlTableRow]: return [HwpxOxmlTableRow(row, self) for row in self.element.findall(f"{_HP}tr")] + def _invalidate_grid(self) -> None: + """Forget the built grid. Every edit that moves, adds, removes, merges or resizes cells calls this.""" + + self._grid_cache = None + + def _cell_grid(self) -> dict[tuple[int, int], HwpxTableGridPosition]: + """The grid of the table as it is now: the kept one when no row, cell, address, span or size + changed since it was built (checked against the XML, so an edit made outside this object is + seen too), else a new one. Callers must not change the returned mapping.""" + + cache = getattr(self, "_grid_cache", None) + if cache is not None and cache.still_valid(self.element): + return cache.mapping + self._grid_cache = None + mapping = self._build_cell_grid() + self._grid_cache = _TableGridCache(mapping, self.element) + return mapping + def _build_cell_grid(self) -> dict[tuple[int, int], HwpxTableGridPosition]: mapping: dict[tuple[int, int], HwpxTableGridPosition] = {} @@ -961,7 +983,16 @@ def _grid_entry(self, row_index: int, col_index: int) -> HwpxTableGridPosition: % (row_index, col_index, row_count, col_count) ) - entry = self._build_cell_grid().get((row_index, col_index)) + entry = None + cache = getattr(self, "_grid_cache", None) + if cache is not None and cache.fast and cache.same_shape(self.element): + # The kept grid answers when the table has the same rows and cells and the cell found still + # has the address, span and size it was built with; anything else goes through the full check. + entry = cache.mapping.get((row_index, col_index)) + if entry is not None and not cache.cell_unchanged(entry.cell.element): + entry = None + if entry is None: + entry = self._cell_grid().get((row_index, col_index)) if entry is None: raise IndexError( "cell coordinates (%d, %d) are covered by a merged cell" @@ -973,7 +1004,7 @@ def _grid_entry(self, row_index: int, col_index: int) -> HwpxTableGridPosition: def iter_grid(self) -> Iterator[HwpxTableGridPosition]: """Yield grid-aware mappings for every logical table position.""" - mapping = self._build_cell_grid() + mapping = self._cell_grid() row_count = self.row_count col_count = self.column_count for row_index in range(row_count): @@ -1449,6 +1480,7 @@ def split_merged_cell( row_offset, col_offset, logical_row, logical_col, row_height, col_width, ) + self._invalidate_grid() self.mark_dirty() return self.cell(row_index, col_index) @@ -1508,7 +1540,7 @@ def _scan_merge_region( total_height = 0 in_order: list[ET.Element] = [] seen: set[ET.Element] = set() - grid = self._build_cell_grid() + grid = self._cell_grid() for row_index in range(start_row, end_row + 1): for col_index in range(start_col, end_col + 1): @@ -1612,6 +1644,7 @@ def merge_cells( # line caches no longer describe it target._clear_own_layout_caches() collapse_merged_grid(self.element) + self._invalidate_grid() self.mark_dirty() return target diff --git a/src/hwpx/oxml/table_merge.py b/src/hwpx/oxml/table_merge.py index e9992aaf..e0db2e0b 100644 --- a/src/hwpx/oxml/table_merge.py +++ b/src/hwpx/oxml/table_merge.py @@ -4,15 +4,19 @@ ``HwpxOxmlTable.merge_cells`` calls :func:`collapse_merged_grid` once the merged cell spans its rectangle. It lives here so that ``table.py`` stays inside the owner-file line budget. + +:class:`TableGridCache` is here for the same reason: ``HwpxOxmlTable`` keeps one built cell +grid and asks it whether the grid still describes the table's XML. """ from __future__ import annotations +from typing import Any import xml.etree.ElementTree as ET from ._document_primitives import _HP -__all__ = ["collapse_merged_grid"] +__all__ = ["TableGridCache", "collapse_merged_grid"] def _own_cells(table: ET.Element) -> list[tuple[ET.Element, ET.Element]]: @@ -80,3 +84,83 @@ def collapse_merged_grid(table: ET.Element) -> None: pass while _drop_unused_column(table): pass + +_TC_ADDR = f"{_HP}cellAddr" +_TC_SPAN = f"{_HP}cellSpan" +_TC_SIZE = f"{_HP}cellSz" + + +def _cell_grid_signature(cell_element: ET.Element) -> tuple[Any, Any, Any]: + """The raw attributes of one ``hp:tc`` that decide where it sits in the grid.""" + + addr = cell_element.find(_TC_ADDR) + span = cell_element.find(_TC_SPAN) + size = cell_element.find(_TC_SIZE) + return ( + None if addr is None else (addr.get("rowAddr"), addr.get("colAddr")), + None if span is None else (span.get("rowSpan"), span.get("colSpan")), + None if size is None else (size.get("width"), size.get("height")), + ) + + +def _is_zero_single_cell(signature: tuple[Any, Any, Any]) -> bool: + """Whether the grid builder may treat the cell as a deactivated placeholder (then its text matters).""" + + _, span, size = signature + try: + if span is not None and (int(span[0] or 1), int(span[1] or 1)) != (1, 1): + return False + if size is not None and (int(size[0] or 0) != 0 or int(size[1] or 0) != 0): + return False + except (TypeError, ValueError): + return True + return True + + +class TableGridCache: + """One built grid with what it was built from, to tell whether the table changed since. + + ``rows``/``row_lengths``/``cells`` are the table's ``hp:tr`` elements, their child counts and, + per physical cell, its element, row and raw address/span/size attributes (plus whether it holds + text when it is a zero-sized 1x1 cell, the one case where text changes the grid). + """ + + __slots__ = ("mapping", "rows", "row_lengths", "cells", "by_cell", "fast") + + def __init__(self, mapping: dict[tuple[int, int], Any], table_element: ET.Element): + self.mapping = mapping + self.rows, self.row_lengths, self.cells, text_matters = self._snapshot(table_element) + self.by_cell = {id(cell): (row, signature) for cell, row, signature, _ in self.cells} + # Coordinate lookups check only the table's shape and the cell they return. That is not enough + # when text decides the grid, or without ``getparent`` (a plain ElementTree element). + self.fast = not text_matters and hasattr(table_element, "getparent") + + @staticmethod + def _snapshot(table_element: ET.Element) -> tuple[list[ET.Element], list[int], list[tuple[Any, ...]], bool]: + rows = table_element.findall(f"{_HP}tr") + cells: list[tuple[Any, ...]] = [] + text_matters = False + for row in rows: + for cell in row.findall(f"{_HP}tc"): + signature = _cell_grid_signature(cell) + has_text: bool | None = None + if _is_zero_single_cell(signature): + text_matters = True + has_text = any(node.text for node in cell.findall(f".//{_HP}t")) + cells.append((cell, row, signature, has_text)) + return rows, [len(row) for row in rows], cells, text_matters + + def same_shape(self, table_element: ET.Element) -> bool: + rows = table_element.findall(f"{_HP}tr") + return rows == self.rows and [len(row) for row in rows] == self.row_lengths + + def cell_unchanged(self, cell_element: ET.Element) -> bool: + known = self.by_cell.get(id(cell_element)) + if known is None: + return False + row, signature = known + return cell_element.getparent() is row and _cell_grid_signature(cell_element) == signature # type: ignore[attr-defined] + + def still_valid(self, table_element: ET.Element) -> bool: + rows, row_lengths, cells, _ = self._snapshot(table_element) + return rows == self.rows and row_lengths == self.row_lengths and cells == self.cells diff --git a/src/hwpx/oxml/table_structure.py b/src/hwpx/oxml/table_structure.py index 369e9f14..2d4a9e3a 100644 --- a/src/hwpx/oxml/table_structure.py +++ b/src/hwpx/oxml/table_structure.py @@ -136,6 +136,7 @@ def _restructure(self, op: dict[str, Any]) -> None: element.attrib.update(new.attrib) element.text = new.text element.extend(list(new)) + self._grid_cache = None # the rows and cells were replaced (HwpxOxmlTable keeps its grid here) self._grow_height(before) _clear_paragraph_layout_cache(self.paragraph.element) # its line held the table as it was section.mark_dirty() diff --git a/src/hwpx/table_patch.py b/src/hwpx/table_patch.py index 2cce692b..db284b34 100644 --- a/src/hwpx/table_patch.py +++ b/src/hwpx/table_patch.py @@ -839,14 +839,20 @@ def fill_cells( table_spans = _iter_table_spans(section_xml) # accumulate byte edits over the whole section (table region splices) section_edits: list[tuple[int, int, bytes]] = [] - occupied: list[tuple[int, int]] = [] # filled cell spans (merge-collision guard) + occupied: set[tuple[int, int]] = set() # filled cell spans (merge-collision guard) + # One grid per table: the section's bytes stay as read until every edit is collected, + # so each address in the same table reads the same table bytes. + grids: dict[int, tuple[bytes, dict[tuple[int, int], _Cell], GridReport]] = {} for ti, row, col, text, mx in section_specs: if ti < 0 or ti >= len(table_spans): skipped.append(CellSkipped(section_path, ti, row, col, "table_index out of range")) continue ts, te = table_spans[ti] - table = section_xml[ts:te] - grid, report = build_grid(table) + built = grids.get(ti) + if built is None: + table = section_xml[ts:te] + built = grids[ti] = (table, *build_grid(table)) + table, grid, report = built if not report.ok: skipped.append(CellSkipped(section_path, ti, row, col, "table grid is invalid: " + "; ".join(report.issues[:5]))) continue @@ -879,7 +885,7 @@ def fill_cells( if span in occupied: skipped.append(CellSkipped(section_path, ti, row, col, "cell shares a merged region already filled by an earlier address")) continue - occupied.append(span) + occupied.add(span) cell_edits = [(ts + cell.start + s, ts + cell.start + e, b) for s, e, b in edits] # A template's short placeholder may legitimately fit under # lineWrap=SQUEEZE, while a real filled value is compressed until diff --git a/src/hwpx/tools/mail_merge.py b/src/hwpx/tools/mail_merge.py index 15b5ab62..37545788 100644 --- a/src/hwpx/tools/mail_merge.py +++ b/src/hwpx/tools/mail_merge.py @@ -273,10 +273,11 @@ def merge_template_rows( fit_fields.append(_fit_field_report(key, fit_result)) replaced_count += _replace_token(document, token, apply_value) document.save_to_path(out_path) + saved_report = _open_safety_of_saved_file(document, out_path) finally: document.close() - open_safety = validate_editor_open_safety(out_path).to_dict() + open_safety = (saved_report or validate_editor_open_safety(out_path)).to_dict() unresolved = _unresolved_placeholders(out_path) reasons: list[str] = [] if missing_keys: @@ -385,6 +386,28 @@ def _find_placeholders(text: str) -> list[dict[str, str]]: return found +def _open_safety_of_saved_file(document: HwpxDocument, path: Path) -> Any | None: + """The open-safety report the save itself reached, when the file holds exactly the bytes it + validated; ``None`` otherwise (the caller then validates the file). + + The save validates the archive before writing it, so validating the written file again repeats + the same checks on the same bytes for every row. + """ + + import hashlib + + package = getattr(document, "_package", None) + report = getattr(package, "_verified_archive_report", None) + digest = getattr(package, "_verified_archive_digest", None) + if report is None or digest is None: + return None + try: + written = Path(path).read_bytes() + except OSError: + return None + return report if hashlib.sha256(written).digest() == digest else None + + def _unresolved_placeholders(path: Path) -> list[str]: document = HwpxDocument.open(path) try: diff --git a/src/hwpx/tools/object_finder.py b/src/hwpx/tools/object_finder.py index 2bb7c1b9..00773832 100644 --- a/src/hwpx/tools/object_finder.py +++ b/src/hwpx/tools/object_finder.py @@ -28,6 +28,7 @@ _resolve_control_nested_text, _resolve_hyperlink_target, _resolve_note_text, + _ElementPaths, build_parent_map, describe_element_path, strip_namespace, @@ -115,7 +116,7 @@ def iter( for section in extractor.iter_sections(): if section_filter is not None and not section_filter(section): continue - parent_map = build_parent_map(section.element) + paths = _ElementPaths(build_parent_map(section.element)) if xpath is not None: candidates = section.element.findall(xpath, namespaces=self.namespaces) else: @@ -125,7 +126,7 @@ def iter( continue if attrs and not self._match_attributes(element, attrs): continue - path = describe_element_path(element, parent_map) + path = paths.describe(element) yield FoundElement(section=section, path=path, element=element) if limit is not None: limit -= 1 diff --git a/src/hwpx/tools/table_navigation.py b/src/hwpx/tools/table_navigation.py index 377149da..b7357b2f 100644 --- a/src/hwpx/tools/table_navigation.py +++ b/src/hwpx/tools/table_navigation.py @@ -247,19 +247,38 @@ def _collect_document_tables(document: HwpxDocument) -> list[_IndexedTable]: ] -def _cell_text(table: HwpxOxmlTable, row_index: int, col_index: int) -> str: - cell = table.cell(row_index, col_index) +def _text_of_cell(cell: object) -> str: paragraphs = list(getattr(cell, "paragraphs", []) or []) if paragraphs: return "\n".join(paragraph.text or "" for paragraph in paragraphs) - return cell.text + return cell.text # type: ignore[attr-defined] + + +def _cell_text(table: HwpxOxmlTable, row_index: int, col_index: int) -> str: + return _text_of_cell(table.cell(row_index, col_index)) + + +def _grid_cells(table: HwpxOxmlTable) -> list[tuple[int, int, object]]: + """Every logical position with its cell, row by row, from one pass over the grid. + + A grid with a position no cell covers is read coordinate by coordinate instead, so the + error names that position as ``table.cell`` does. + """ + + try: + return [(position.row, position.column, position.cell) for position in table.iter_grid()] + except IndexError: + return [ + (row_index, col_index, table.cell(row_index, col_index)) + for row_index in range(table.row_count) + for col_index in range(table.column_count) + ] def _table_is_empty(table: HwpxOxmlTable) -> bool: - for row_index in range(table.row_count): - for col_index in range(table.column_count): - if _cell_text(table, row_index, col_index).strip(): - return False + for _, _, cell in _grid_cells(table): + if _text_of_cell(cell).strip(): + return False return True @@ -290,38 +309,36 @@ def _table_cell_paragraph_location( def _table_cells(table_ref: _IndexedTable) -> list[dict[str, object]]: cells: list[dict[str, object]] = [] - for row_index in range(table_ref.table.row_count): - for col_index in range(table_ref.table.column_count): - cell = table_ref.table.cell(row_index, col_index) - paragraphs = list(getattr(cell, "paragraphs", []) or []) - paragraph_payloads: list[dict[str, object]] = [] - for cell_paragraph_index, paragraph in enumerate(paragraphs): - paragraph_payloads.append( - { - "cell_paragraph_index": cell_paragraph_index, - "text": paragraph.text or "", - "location": _table_cell_paragraph_location( - table_ref.table_index, - row_index, - col_index, - cell_paragraph_index, - ), - } - ) - cells.append( + for row_index, col_index, cell in _grid_cells(table_ref.table): + paragraphs = list(getattr(cell, "paragraphs", []) or []) + paragraph_payloads: list[dict[str, object]] = [] + for cell_paragraph_index, paragraph in enumerate(paragraphs): + paragraph_payloads.append( { - "row": row_index, - "col": col_index, - "text": _cell_text(table_ref.table, row_index, col_index), - "paragraphs": paragraph_payloads, - "location": { - "kind": "table_cell", - "table_index": table_ref.table_index, - "row": row_index, - "col": col_index, - }, + "cell_paragraph_index": cell_paragraph_index, + "text": paragraph.text or "", + "location": _table_cell_paragraph_location( + table_ref.table_index, + row_index, + col_index, + cell_paragraph_index, + ), } ) + cells.append( + { + "row": row_index, + "col": col_index, + "text": "\n".join(paragraph.text or "" for paragraph in paragraphs) if paragraphs else cell.text, # type: ignore[attr-defined] + "paragraphs": paragraph_payloads, + "location": { + "kind": "table_cell", + "table_index": table_ref.table_index, + "row": row_index, + "col": col_index, + }, + } + ) return cells diff --git a/src/hwpx/tools/text_extractor.py b/src/hwpx/tools/text_extractor.py index 6b882a73..16563c6e 100644 --- a/src/hwpx/tools/text_extractor.py +++ b/src/hwpx/tools/text_extractor.py @@ -240,8 +240,9 @@ def iter_paragraphs( if strip_namespace(child.tag) == "p" ] + paths = _ElementPaths(parent_map) for index, element in enumerate(paragraph_elements): - path = describe_element_path(element, parent_map) + path = paths.describe(element) hierarchy = tuple(path.split("/")) yield ParagraphInfo( section=section, @@ -833,6 +834,48 @@ def describe_element_path( return "/".join(reversed(parts)) +class _ElementPaths: + """:func:`describe_element_path` for many elements of one tree. + + Gives the same strings. The position of every child among its same-named siblings is worked + out once per parent and the path of a parent is kept, so describing each paragraph of a section + no longer lists the section's paragraphs again (which made it quadratic). + """ + + def __init__(self, parent_map: Mapping[ET.Element, ET.Element]) -> None: + self._parent_map = parent_map + self._steps: Dict[ET.Element, Dict[ET.Element, str]] = {} + self._paths: Dict[ET.Element, str] = {} + + def _step(self, parent: ET.Element, element: ET.Element) -> str: + steps = self._steps.get(parent) + if steps is None: + by_name: Dict[str, list[ET.Element]] = {} + for child in parent: + by_name.setdefault(strip_namespace(child.tag), []).append(child) + steps = {} + for local, siblings in by_name.items(): + if len(siblings) > 1: + for index, child in enumerate(siblings): + steps[child] = f"{local}[{index}]" + else: + steps[siblings[0]] = local + self._steps[parent] = steps + return steps[element] + + def describe(self, element: ET.Element) -> str: + known = self._paths.get(element) + if known is not None: + return known + parent = self._parent_map.get(element) + if parent is None: + path = strip_namespace(element.tag) + else: + path = f"{self.describe(parent)}/{self._step(parent, element)}" + self._paths[element] = path + return path + + def _section_sort_key(name: str) -> Tuple[int, str]: match = _SECTION_PATTERN.match(PurePosixPath(name).name) if match: diff --git a/tests/test_append_paragraph_cost.py b/tests/test_append_paragraph_cost.py new file mode 100644 index 00000000..4acb3922 --- /dev/null +++ b/tests/test_append_paragraph_cost.py @@ -0,0 +1,61 @@ +"""Appending a paragraph, a heading or a formatted paragraph does not wrap every body paragraph.""" + +from __future__ import annotations + +import pytest + +from hwpx import HwpxDocument +from hwpx._document import layout +from hwpx.oxml.section import HwpxOxmlSection + + +@pytest.fixture +def body_lists(monkeypatch: pytest.MonkeyPatch) -> list[int]: + calls: list[int] = [] + original = HwpxOxmlSection.paragraphs.fget + assert original is not None + + def counting(self: HwpxOxmlSection): + calls.append(1) + return original(self) + + monkeypatch.setattr(HwpxOxmlSection, "paragraphs", property(counting)) + return calls + + +def test_add_paragraph_reads_only_the_last_paragraph(body_lists: list[int]) -> None: + document = HwpxDocument.new() + first = document.add_paragraph("하나", para_pr_id_ref=1, style_id_ref=2) + second = document.add_paragraph("둘") + + assert (second.para_pr_id_ref, second.style_id_ref) == (first.para_pr_id_ref, first.style_id_ref) + assert body_lists == [] + + +def test_heading_and_object_format_do_not_list_the_body(body_lists: list[int]) -> None: + document = HwpxDocument.new() + for index in range(5): + document.add_paragraph(f"문단 {index}") + body_lists.clear() + + heading = document.add_heading("제목", level=2) + result = document.styles.apply_paragraph_format(paragraphs=[heading], keep_with_next=True) + + assert result.paragraphs == (6,) + assert body_lists == [] + + +def test_body_indexes_match_the_body_list() -> None: + document = HwpxDocument.new() + for index in range(6): + document.add_paragraph(f"문단 {index}") + table = document.add_table(2, 2) + body = document.paragraphs + cell_paragraph = table.cell(0, 0).paragraphs[0] + other = HwpxDocument.new().add_paragraph("다른 문서") + + few = [body[4], cell_paragraph, body[0], other, body[-1]] + assert layout._body_indexes(document, few) == [4, None, 0, None, len(body) - 1] + + many = list(body) + [cell_paragraph] + assert layout._body_indexes(document, many) == [*range(len(body)), None] diff --git a/tests/test_coverage_ledger.py b/tests/test_coverage_ledger.py index 803b0731..24b12243 100644 --- a/tests/test_coverage_ledger.py +++ b/tests/test_coverage_ledger.py @@ -10,6 +10,7 @@ from __future__ import annotations +import copy import importlib.util import json import re @@ -73,20 +74,56 @@ def _load_ledger() -> dict: # --------------------------------------------------------------------------- -def test_check_is_green() -> None: +@pytest.fixture(scope="module") +def built_ledger(): + """원장을 한 번만 만든다. ``build_ledger``는 스키마·말뭉치 집계·소스 전체를 훑는 무거운 + 계산이라, 아래 세 테스트가 각자 만들면 테스트 시간의 큰 몫을 쓴다(CI에서 330초). + 테스트는 이 결과의 사본을 받는다.""" + + module = _module() + return module, module.build_ledger(module.DEFAULT_CENSUS_PATH) + + +def _use_built_ledger(monkeypatch: pytest.MonkeyPatch, built_ledger) -> tuple: + """``main()``이 다시 만드는 대신 한 번 만든 원장의 사본을 쓰게 한다.""" + + module, ledger = built_ledger + monkeypatch.setattr(module, "build_ledger", lambda _census: copy.deepcopy(ledger)) + return module, copy.deepcopy(ledger) + + +def test_check_is_green( + built_ledger, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: """커밋된 원장(JSON+MD)이 재생성본과 바이트 동일하다 — CI 게이트 본체.""" + module, _ = _use_built_ledger(monkeypatch, built_ledger) + monkeypatch.setattr(sys, "argv", ["coverage_ledger.py", "--check"]) + + status = module.main() + captured = capsys.readouterr() + + assert status == 0, captured.out + captured.err + assert "in sync" in captured.out + + +def test_script_starts_as_a_process() -> None: + """스크립트가 프로세스로 뜬다(원장은 만들지 않는다) — 위 게이트가 프로세스 안에서 도는 대신.""" + proc = subprocess.run( - [sys.executable, str(SCRIPT), "--check"], + [sys.executable, "-X", "utf8", str(SCRIPT), "--help"], capture_output=True, text=True, + encoding="utf-8", cwd=ROOT, ) assert proc.returncode == 0, proc.stdout + proc.stderr - assert "in sync" in proc.stdout + assert "--check" in proc.stdout -def test_check_fails_closed_on_drift(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: +def test_check_fails_closed_on_drift( + built_ledger, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: """--check가 실제로 드리프트를 감지하는지 — 위 그린 테스트의 대조군. 커밋된 ``docs/coverage-ledger.json``/``.md``는 절대 건드리지 않는다 — @@ -99,8 +136,7 @@ def test_check_fails_closed_on_drift(tmp_path: Path, monkeypatch: pytest.MonkeyP 레포를 읽으므로 진짜 코드 경로를 검증한다. """ - module = _module() - ledger = module.build_ledger(module.DEFAULT_CENSUS_PATH) + module, ledger = _use_built_ledger(monkeypatch, built_ledger) fake_json = tmp_path / "coverage-ledger.json" fake_md = tmp_path / "coverage-ledger.md" @@ -116,12 +152,11 @@ def test_check_fails_closed_on_drift(tmp_path: Path, monkeypatch: pytest.MonkeyP assert module.main() == 1 -def test_generated_from_names_its_inputs_with_forward_slashes() -> None: +def test_generated_from_names_its_inputs_with_forward_slashes(built_ledger) -> None: """원장이 적는 입력 경로는 어느 운영체제에서 만들어도 ``/``로 적힌다 — Windows의 ``str(Path)``는 역슬래시를 써서, 커밋된 원장과 어긋나 --check가 드리프트로 본다.""" - module = _module() - generated = module.build_ledger(module.DEFAULT_CENSUS_PATH)["generatedFrom"] + generated = built_ledger[1]["generatedFrom"] paths = [generated["corpusCensus"], *generated["openrateReports"]] assert paths diff --git a/tests/test_header_lookup_cache.py b/tests/test_header_lookup_cache.py new file mode 100644 index 00000000..f8b2efac --- /dev/null +++ b/tests/test_header_lookup_cache.py @@ -0,0 +1,91 @@ +"""Header lookups keep their parsed result and follow every change to the header XML.""" + +from __future__ import annotations + +import pytest + +from hwpx import HwpxDocument +from hwpx.oxml import header_part + +_HH = "{http://www.hancom.co.kr/hwpml/2011/head}" + + +@pytest.fixture +def parses(monkeypatch: pytest.MonkeyPatch) -> list[str]: + calls: list[str] = [] + for name in ("parse_paragraph_property", "parse_paragraph_properties", "parse_styles", "parse_tab_definitions", "parse_border_fills"): + original = getattr(header_part, name) + + def counting(node, _original=original, _name=name): + calls.append(_name) + return _original(node) + + monkeypatch.setattr(header_part, name, counting) + return calls + + +def test_repeated_lookups_parse_once(parses: list[str]) -> None: + document = HwpxDocument.new() + header = document.oxml.headers[0] + + first = header.paragraph_property(0) + for _ in range(5): + assert header.paragraph_property(0) == first + assert header.style(0) is not None + assert header.tab_property(0) is not None + assert header.border_fill(1) is not None + assert header.paragraph_properties + + assert sorted(parses) == sorted( + ["parse_paragraph_property", "parse_styles", "parse_tab_definitions", "parse_border_fills", "parse_paragraph_properties"] + ) + + +def test_lookup_follows_an_edit_through_the_api() -> None: + document = HwpxDocument.new() + before = dict(document.styles.paragraph_properties) + + new_id = document.oxml.headers[0].ensure_paragraph_format(base_para_pr_id=0, alignment="CENTER") + + after = document.styles.paragraph_properties + assert str(new_id) in after and str(new_id) not in before + assert document.styles.paragraph_property(new_id) is not None + + +def test_lookup_follows_a_raw_xml_edit_without_mark_dirty() -> None: + document = HwpxDocument.new() + header = document.oxml.headers[0] + assert header.paragraph_property(0).tab_pr_id_ref == 0 + assert header.style(0).name + assert header.paragraph_properties["0"].tab_pr_id_ref == 0 + + para_pr = next(child for child in header._para_properties_element() if child.get("id") == "0") + para_pr.set("tabPrIDRef", "1") + style = next(child for child in header._styles_element() if child.get("id") == "0") + style.set("name", "changed outside") + + assert header.paragraph_property(0).tab_pr_id_ref == 1 + assert header.paragraph_properties["0"].tab_pr_id_ref == 1 + assert header.style(0).name == "changed outside" + assert document.styles.get(0).name == "changed outside" + + +def test_mark_dirty_drops_the_kept_results(parses: list[str]) -> None: + document = HwpxDocument.new() + header = document.oxml.headers[0] + header.paragraph_property(0) + header.mark_dirty() + header.paragraph_property(0) + + assert parses.count("parse_paragraph_property") == 2 + + +def test_mappings_are_new_dicts_each_time() -> None: + document = HwpxDocument.new() + header = document.oxml.headers[0] + + styles = header.styles + styles.clear() + + assert header.styles + assert header.paragraph_properties is not header.paragraph_properties diff --git a/tests/test_table_grid_cache.py b/tests/test_table_grid_cache.py new file mode 100644 index 00000000..3eb19b8a --- /dev/null +++ b/tests/test_table_grid_cache.py @@ -0,0 +1,127 @@ +"""The table keeps its cell grid between coordinate reads and rebuilds it when the table changes.""" + +from __future__ import annotations + +import pytest + +from hwpx import HwpxDocument +from hwpx.oxml.table import HwpxOxmlTable + +_HP = "{http://www.hancom.co.kr/hwpml/2011/paragraph}" + + +def _table(rows: int = 4, cols: int = 4) -> HwpxOxmlTable: + document = HwpxDocument.new() + table = document.add_table(rows, cols) + for position in table.iter_grid(): + position.cell.text = f"r{position.row}c{position.column}" + return table + + +@pytest.fixture +def builds(monkeypatch: pytest.MonkeyPatch) -> list[int]: + calls: list[int] = [] + original = HwpxOxmlTable._build_cell_grid + + def counting(self: HwpxOxmlTable): + calls.append(1) + return original(self) + + monkeypatch.setattr(HwpxOxmlTable, "_build_cell_grid", counting) + return calls + + +def test_coordinate_reads_build_the_grid_once(builds: list[int]) -> None: + table = _table() # builds the grid to fill the cells + + texts = [table.cell(row, col).text for row in range(4) for col in range(4)] + for row in range(4): + for col in range(4): + table.set_cell_text(row, col, f"new {row},{col}") + list(table.iter_grid()) + table.get_cell_map() + + assert texts[5] == "r1c1" + assert table.cell(3, 3).text == "new 3,3" + assert len(builds) == 1 + + +def test_merge_and_split_rebuild_the_grid() -> None: + table = _table() + assert table.cell(1, 1).address == (1, 1) + + merged = table.merge_cells(0, 0, 1, 1) + assert table.cell(1, 1).element is merged.element + assert table.cell(1, 1).span == (2, 2) + + table.split_merged_cell(1, 1) + assert table.cell(1, 1).address == (1, 1) + assert table.cell(0, 0).span == (1, 1) + + +def test_row_and_column_edits_rebuild_the_grid() -> None: + table = _table() + assert table.cell(3, 3).text == "r3c3" + + table.insert_rows(0, 1) + assert table.row_count == 5 + assert table.cell(4, 3).text == "r3c3" + + table.delete_columns(0) + assert table.cell(4, 2).text == "r3c3" + with pytest.raises(IndexError): + table.cell(4, 3) + + +def test_edit_through_another_wrapper_of_the_same_table_is_seen() -> None: + table = _table() + assert table.cell(0, 1).address == (0, 1) + + other = HwpxOxmlTable(table.element, table.paragraph) + merged = other.merge_cells(0, 0, 0, 1) + + assert table.cell(0, 1).element is merged.element + assert [position.anchor for position in table.iter_grid()][:2] == [(0, 0), (0, 0)] + + +def test_raw_xml_edits_are_seen() -> None: + table = _table() + first_row = table.element.findall(f"{_HP}tr")[0] + first, second = first_row.findall(f"{_HP}tc")[:2] + assert table.cell(0, 1).element is second + + # the first cell grows over the second, which is taken out, as a merge saved by Hancom looks + first.find(f"{_HP}cellSpan").set("colSpan", "2") + first_row.remove(second) + assert table.cell(0, 1).element is first + assert table.cell(0, 0).span == (1, 2) + + # and back + first.find(f"{_HP}cellSpan").set("colSpan", "1") + first_row.insert(1, second) + assert table.cell(0, 1).element is second + + # a cell that moves to another address + second.find(f"{_HP}cellAddr").set("colAddr", "0") + with pytest.raises(ValueError, match="overlapping"): + table.cell(0, 1) + second.find(f"{_HP}cellAddr").set("colAddr", "1") + assert table.cell(0, 1).element is second + + +def test_text_in_a_zero_sized_covered_cell_is_seen() -> None: + table = _table() + first_row = table.element.findall(f"{_HP}tr")[0] + first, second = first_row.findall(f"{_HP}tc")[:2] + table.cell(0, 1).text = "" + first.find(f"{_HP}cellSpan").set("colSpan", "2") + second.find(f"{_HP}cellSz").set("width", "0") + second.find(f"{_HP}cellSz").set("height", "0") + + # an empty zero-sized cell under a merged one is a placeholder: the merged cell answers + assert table.cell(0, 1).element is first + + # with text it is a real cell again, and the two overlap + second.find(f".//{_HP}t").text = "x" + with pytest.raises(ValueError, match="overlapping"): + table.cell(0, 1)