kaxil commented on code in PR #73926:
URL: https://github.com/apache/airflow/pull/73926#discussion_r4243913697
##########
providers/common/ai/src/airflow/providers/common/ai/operators/document_loader.py:
##########
@@ -450,21 +453,77 @@ def _parse_pdf_stream(self, stream: BinaryIO) ->
list[dict[str, Any]]:
documents.append({"text": text, "metadata": {"page_number":
page_num + 1}})
return documents
- def _parse_docx_stream(self, stream: BinaryIO) -> list[dict[str, Any]]:
+ def _parse_docx_stream(self, stream: BinaryIO, *, source_hint: str) ->
list[dict[str, Any]]:
"""
- Parse a DOCX stream into documents.
+ Parse a DOCX stream into a single document.
- Extracts paragraph text only. Tables, headers, footers, and footnotes
- are not included. For richer DOCX parsing, plug in a dedicated
- extraction tool (``Unstructured``, ``docling``) as a custom parser
- backend.
+ Paragraphs and tables in the document body are extracted in document
+ order. Each table row becomes one "| cell | cell |" line, and a nested
+ table is flattened into its cell. Headers, footers, footnotes, content
+ controls, text boxes, and pending tracked insertions are not included.
"""
try:
from docx import Document
+ from docx.oxml.exceptions import InvalidXmlError
+ from docx.table import Table
except ImportError as e:
raise AirflowOptionalProviderFeatureException(e)
doc = Document(stream)
- paragraphs = [p.text for p in doc.paragraphs if p.text.strip()]
- text = "\n\n".join(paragraphs)
+ blocks = []
+ for block in doc.iter_inner_content():
+ if isinstance(block, Table):
+ try:
+ rows = self._get_docx_table_rows(block, Table)
+ except (ValueError, InvalidXmlError, RecursionError) as e:
+ # Malformed table XML raises ValueError (a vMerge with no
cell above) or
+ # InvalidXmlError (a missing required attribute).
python-docx walks vertical
+ # merges up recursively, so one spanning about 1000 rows
raises RecursionError.
+ self.log.warning(
+ "Skipping a table in %s that python-docx could not
read: %r", source_hint, e
+ )
+ continue
+ text = "\n".join(f"| {' | '.join(cells)} |" for cells in rows)
+ else:
+ text = block.text
+ if text.strip():
+ blocks.append(text)
+ text = "\n\n".join(blocks)
return [{"text": text, "metadata": {}}]
+
+ def _get_docx_table_rows(self, table: Table, table_cls: type[Table]) ->
list[list[str]]:
+ # w:gridBefore, w:gridAfter and w:gridSpan come straight from the XML,
so the empty
+ # cells added for them are capped at the table's column count.
+ column_count = len(table.columns)
+
+ rows = []
+ for row in table.rows:
+ cells: list[str] = [""] * min(row.grid_cols_before, column_count)
+ previous_cell = None
+ for cell in row.cells:
Review Comment:
For `w:gridSpan` the cap bounds the text but not the work. python-docx's
`row.cells` yields the cell once per spanned column and returns a tuple, so the
whole span is built before the `len(cells) < column_count` check runs. On a 36
KB file with a two-column table, `gridSpan w:val="100000000"` took about 4 s
and 950 MB peak on both 1.1.2 and 1.2.0 to produce the correct `| x | |`, and
it grows linearly from there, so a larger value ends in a `MemoryError` (not in
the except tuple) or an OOM-killed worker. Checking `tc.grid_span` for each
`tc` in `row._tr.tc_lst` before calling `row.cells`, and raising `ValueError`
when one exceeds `column_count`, would send it down the existing
skip-with-warning path. The `grid-span` test uses `val = 1000`, which passes
either way. With the pre-check in place it could use a large value and assert
the warning.
##########
providers/common/ai/src/airflow/providers/common/ai/operators/document_loader.py:
##########
@@ -450,21 +453,77 @@ def _parse_pdf_stream(self, stream: BinaryIO) ->
list[dict[str, Any]]:
documents.append({"text": text, "metadata": {"page_number":
page_num + 1}})
return documents
- def _parse_docx_stream(self, stream: BinaryIO) -> list[dict[str, Any]]:
+ def _parse_docx_stream(self, stream: BinaryIO, *, source_hint: str) ->
list[dict[str, Any]]:
"""
- Parse a DOCX stream into documents.
+ Parse a DOCX stream into a single document.
- Extracts paragraph text only. Tables, headers, footers, and footnotes
- are not included. For richer DOCX parsing, plug in a dedicated
- extraction tool (``Unstructured``, ``docling``) as a custom parser
- backend.
+ Paragraphs and tables in the document body are extracted in document
+ order. Each table row becomes one "| cell | cell |" line, and a nested
+ table is flattened into its cell. Headers, footers, footnotes, content
+ controls, text boxes, and pending tracked insertions are not included.
"""
try:
from docx import Document
+ from docx.oxml.exceptions import InvalidXmlError
+ from docx.table import Table
except ImportError as e:
raise AirflowOptionalProviderFeatureException(e)
doc = Document(stream)
- paragraphs = [p.text for p in doc.paragraphs if p.text.strip()]
- text = "\n\n".join(paragraphs)
+ blocks = []
+ for block in doc.iter_inner_content():
+ if isinstance(block, Table):
+ try:
+ rows = self._get_docx_table_rows(block, Table)
+ except (ValueError, InvalidXmlError, RecursionError) as e:
+ # Malformed table XML raises ValueError (a vMerge with no
cell above) or
+ # InvalidXmlError (a missing required attribute).
python-docx walks vertical
+ # merges up recursively, so one spanning about 1000 rows
raises RecursionError.
+ self.log.warning(
+ "Skipping a table in %s that python-docx could not
read: %r", source_hint, e
+ )
+ continue
+ text = "\n".join(f"| {' | '.join(cells)} |" for cells in rows)
+ else:
+ text = block.text
+ if text.strip():
+ blocks.append(text)
+ text = "\n\n".join(blocks)
return [{"text": text, "metadata": {}}]
+
+ def _get_docx_table_rows(self, table: Table, table_cls: type[Table]) ->
list[list[str]]:
+ # w:gridBefore, w:gridAfter and w:gridSpan come straight from the XML,
so the empty
+ # cells added for them are capped at the table's column count.
+ column_count = len(table.columns)
+
+ rows = []
+ for row in table.rows:
+ cells: list[str] = [""] * min(row.grid_cols_before, column_count)
+ previous_cell = None
+ for cell in row.cells:
+ # python-docx repeats the same _Cell object for every grid
column a horizontal
+ # merge spans: keep its text once and leave the other columns
empty so later
+ # cells stay under their headers. A vertical merge repeats on
each row it spans.
+ if cell is previous_cell:
+ if len(cells) < column_count:
+ cells.append("")
+ continue
+ previous_cell = cell
+ cells.append(self._get_docx_cell_text(cell, table_cls))
+ cells.extend([""] * min(row.grid_cols_after, column_count))
Review Comment:
Each padding source is capped on its own, so a row can still come out wider
than the table. With `gridBefore` and `gridAfter` both at 1000 on a two-column
table the row renders as `| | | x | | |`, and the
`grid-before`/`grid-after` cases in
`test_docx_grid_values_capped_at_column_count` expect three cells under a
two-column header. Capping against the remaining width, e.g. `max(0,
column_count - len(cells))` here, would keep rows at the table width.
##########
providers/common/ai/src/airflow/providers/common/ai/operators/document_loader.py:
##########
@@ -450,21 +453,77 @@ def _parse_pdf_stream(self, stream: BinaryIO) ->
list[dict[str, Any]]:
documents.append({"text": text, "metadata": {"page_number":
page_num + 1}})
return documents
- def _parse_docx_stream(self, stream: BinaryIO) -> list[dict[str, Any]]:
+ def _parse_docx_stream(self, stream: BinaryIO, *, source_hint: str) ->
list[dict[str, Any]]:
"""
- Parse a DOCX stream into documents.
+ Parse a DOCX stream into a single document.
- Extracts paragraph text only. Tables, headers, footers, and footnotes
- are not included. For richer DOCX parsing, plug in a dedicated
- extraction tool (``Unstructured``, ``docling``) as a custom parser
- backend.
+ Paragraphs and tables in the document body are extracted in document
+ order. Each table row becomes one "| cell | cell |" line, and a nested
+ table is flattened into its cell. Headers, footers, footnotes, content
+ controls, text boxes, and pending tracked insertions are not included.
"""
try:
from docx import Document
+ from docx.oxml.exceptions import InvalidXmlError
+ from docx.table import Table
except ImportError as e:
raise AirflowOptionalProviderFeatureException(e)
doc = Document(stream)
- paragraphs = [p.text for p in doc.paragraphs if p.text.strip()]
- text = "\n\n".join(paragraphs)
+ blocks = []
+ for block in doc.iter_inner_content():
+ if isinstance(block, Table):
+ try:
+ rows = self._get_docx_table_rows(block, Table)
+ except (ValueError, InvalidXmlError, RecursionError) as e:
+ # Malformed table XML raises ValueError (a vMerge with no
cell above) or
+ # InvalidXmlError (a missing required attribute).
python-docx walks vertical
+ # merges up recursively, so one spanning about 1000 rows
raises RecursionError.
+ self.log.warning(
+ "Skipping a table in %s that python-docx could not
read: %r", source_hint, e
+ )
+ continue
+ text = "\n".join(f"| {' | '.join(cells)} |" for cells in rows)
+ else:
+ text = block.text
+ if text.strip():
+ blocks.append(text)
+ text = "\n\n".join(blocks)
return [{"text": text, "metadata": {}}]
+
+ def _get_docx_table_rows(self, table: Table, table_cls: type[Table]) ->
list[list[str]]:
+ # w:gridBefore, w:gridAfter and w:gridSpan come straight from the XML,
so the empty
+ # cells added for them are capped at the table's column count.
+ column_count = len(table.columns)
+
+ rows = []
+ for row in table.rows:
+ cells: list[str] = [""] * min(row.grid_cols_before, column_count)
+ previous_cell = None
+ for cell in row.cells:
+ # python-docx repeats the same _Cell object for every grid
column a horizontal
+ # merge spans: keep its text once and leave the other columns
empty so later
+ # cells stay under their headers. A vertical merge repeats on
each row it spans.
+ if cell is previous_cell:
+ if len(cells) < column_count:
+ cells.append("")
+ continue
+ previous_cell = cell
+ cells.append(self._get_docx_cell_text(cell, table_cls))
+ cells.extend([""] * min(row.grid_cols_after, column_count))
+ if any(cells):
+ rows.append(cells)
+ return rows
+
+ def _get_docx_cell_text(self, cell: _Cell, table_cls: type[Table]) -> str:
+ parts = []
+ for item in cell.iter_inner_content():
+ if isinstance(item, table_cls):
+ text = "; ".join(" / ".join(cells) for cells in
self._get_docx_table_rows(item, table_cls))
Review Comment:
The skip only wraps top-level tables, so an unreadable nested table takes
the whole outer table with it. A paragraph followed by a 2x2 table whose last
cell holds a nested table with a first-row `vMerge` continuation loads as just
the paragraph: Name, Qty and Apple are lost, and the warning only says a table
was skipped. Catching the same exceptions around this call and leaving out just
the nested table would keep the readable outer rows. A case next to
`test_docx_unreadable_table_skipped_with_warning` would cover it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
