Files
doc-forge/backend/services/document_export.py
T
2026-07-02 18:34:21 +08:00

235 lines
8.0 KiB
Python

from io import BytesIO
from copy import deepcopy
from docx import Document
from docx.document import Document as DocumentObject
from docx.oxml import OxmlElement
from docx.oxml.table import CT_Tbl
from docx.oxml.text.paragraph import CT_P
from docx.table import Table, _Cell
from docx.text.paragraph import Paragraph
def _iter_block_items(parent: DocumentObject | _Cell):
parent_elm = parent.element.body if isinstance(parent, DocumentObject) else parent._tc
for child in parent_elm.iterchildren():
if isinstance(child, CT_P):
yield Paragraph(child, parent)
elif isinstance(child, CT_Tbl):
yield Table(child, parent)
def _is_heading(paragraph: Paragraph) -> bool:
style_name = paragraph.style.name if paragraph.style is not None else ""
normalized = style_name.lower().replace(" ", "")
return normalized.startswith("heading")
def _delete_block(block):
element = block._element
parent = element.getparent()
if parent is not None:
parent.remove(element)
def _clear_paragraph(paragraph: Paragraph):
element = paragraph._element
for child in list(element):
if child.tag.endswith("}r"):
element.remove(child)
def _copy_paragraph_format(target: Paragraph, source: Paragraph | None):
if source is None:
return
source_ppr = source._element.pPr
if source_ppr is not None:
target._element.insert(0, deepcopy(source_ppr))
def _copy_run_format(target_run, source_paragraph: Paragraph | None):
if source_paragraph is None:
return
for source_run in source_paragraph.runs:
if source_run._element.rPr is not None:
target_run._element.insert(0, deepcopy(source_run._element.rPr))
break
def _append_paragraph_after(
paragraph: Paragraph,
text: str,
style_name: str | None = None,
template_paragraph: Paragraph | None = None,
) -> Paragraph:
new_p = OxmlElement("w:p")
paragraph._element.addnext(new_p)
new_para = Paragraph(new_p, paragraph._parent)
_copy_paragraph_format(new_para, template_paragraph)
if style_name:
try:
new_para.style = style_name
except Exception:
pass
if text:
run = new_para.add_run(text)
_copy_run_format(run, template_paragraph)
return new_para
def _set_cell_text_with_template(cell, value: str, template_paragraph: Paragraph | None = None):
if not cell.paragraphs:
cell.text = value
return
paragraph = cell.paragraphs[0]
_clear_paragraph(paragraph)
run = paragraph.add_run(value)
_copy_run_format(run, template_paragraph)
def _resize_table_rows(table: Table, row_count: int):
current_rows = len(table.rows)
if current_rows == 0:
return
if current_rows < row_count:
template_row = table.rows[-1]._tr
for _ in range(row_count - current_rows):
table._tbl.append(deepcopy(template_row))
elif current_rows > row_count:
for _ in range(current_rows - row_count):
table._tbl.remove(table.rows[-1]._tr)
def _fill_table(table: Table, matrix: list[list[str]]):
if not matrix:
return
_resize_table_rows(table, len(matrix))
template_cell_paragraph = table.rows[0].cells[0].paragraphs[0] if table.rows and table.rows[0].cells else None
for row_index, row_values in enumerate(matrix):
row = table.rows[row_index]
for col_index, cell in enumerate(row.cells):
value = row_values[col_index] if col_index < len(row_values) else ""
_set_cell_text_with_template(cell, value, template_cell_paragraph)
def _append_table_after(
paragraph: Paragraph,
rows: list[list[str]],
headers: list[str] | None = None,
template_table: Table | None = None,
):
matrix = [headers, *rows] if headers else rows
if template_table is not None:
cloned_tbl = deepcopy(template_table._tbl)
paragraph._element.addnext(cloned_tbl)
cloned_table = Table(cloned_tbl, paragraph._parent)
_fill_table(cloned_table, matrix)
return cloned_table
container = paragraph._parent
table = container.add_table(rows=max(len(matrix), 1), cols=max(len(headers or []), len(rows[0]) if rows else 1))
if headers:
for row_index, row_values in enumerate(matrix):
for index, value in enumerate(row_values):
table.rows[row_index].cells[index].text = value
elif rows:
for row_index, row_values in enumerate(matrix):
for index, value in enumerate(row_values):
table.rows[row_index].cells[index].text = value
tbl = table._tbl
tbl.getparent().remove(tbl)
paragraph._element.addnext(tbl)
return Table(tbl, container)
def _append_empty_paragraph_after_table(table: Table, style_name: str | None = None) -> Paragraph:
new_p = OxmlElement("w:p")
table._tbl.addnext(new_p)
new_para = Paragraph(new_p, table._parent)
if style_name:
try:
new_para.style = style_name
except Exception:
pass
return new_para
def _find_heading_paragraph(document: DocumentObject, heading_text: str, after_element=None) -> Paragraph | None:
started = after_element is None
for block in _iter_block_items(document):
if isinstance(block, Paragraph) and _is_heading(block) and block.text.strip() == heading_text.strip():
if started:
return block
if after_element is not None and block._element == after_element:
started = True
return None
def _replace_section_content(document: DocumentObject, heading_title: str, content: dict, after_element=None):
heading = _find_heading_paragraph(document, heading_title, after_element)
if heading is None:
return after_element
first_body_style = None
paragraph_template = None
table_template = None
current = heading._element.getnext()
blocks_to_remove = []
while current is not None:
if isinstance(current, CT_P):
current_paragraph = Paragraph(current, heading._parent)
if _is_heading(current_paragraph):
break
if first_body_style is None and current_paragraph.style is not None:
first_body_style = current_paragraph.style.name
if paragraph_template is None:
paragraph_template = current_paragraph
blocks_to_remove.append(current_paragraph)
elif isinstance(current, CT_Tbl):
current_table = Table(current, heading._parent)
if table_template is None:
table_template = current_table
blocks_to_remove.append(current_table)
current = current.getnext()
for block in blocks_to_remove:
_delete_block(block)
insert_after = heading
content_blocks = content.get("content", [])
for block in content_blocks:
block_type = block.get("type")
if block_type == "table":
rows = [list(row) for row in block.get("rows", [])]
headers = block.get("headers") or []
table = _append_table_after(insert_after, rows, headers, table_template)
insert_after = _append_empty_paragraph_after_table(table, first_body_style)
else:
text = block.get("text", "")
text_parts = [item for item in text.split("\n") if item] or [text]
for text_part in text_parts:
insert_after = _append_paragraph_after(
insert_after,
text_part,
first_body_style,
paragraph_template,
)
return heading._element
def export_document_bytes(template_bytes: bytes, logs: list[dict]) -> bytes:
document = Document(BytesIO(template_bytes))
last_heading_element = None
for item in logs:
last_heading_element = _replace_section_content(
document,
item["title"],
item["content"],
last_heading_element,
)
output = BytesIO()
document.save(output)
return output.getvalue()