diff --git a/backend/services/document_export.py b/backend/services/document_export.py index c6329f7..1becaca 100644 --- a/backend/services/document_export.py +++ b/backend/services/document_export.py @@ -1,4 +1,5 @@ from io import BytesIO +from copy import deepcopy from docx import Document from docx.document import Document as DocumentObject @@ -38,36 +39,103 @@ def _clear_paragraph(paragraph: Paragraph): element.remove(child) -def _append_paragraph_after(paragraph: Paragraph, text: str, style_name: str | None = None) -> Paragraph: +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: - new_para.add_run(text) + run = new_para.add_run(text) + _copy_run_format(run, template_paragraph) return new_para -def _append_table_after(paragraph: Paragraph, rows: list[list[str]], headers: list[str] | None = None): +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=1, cols=max(len(headers or []), len(rows[0]) if rows else 1)) + table = container.add_table(rows=max(len(matrix), 1), cols=max(len(headers or []), len(rows[0]) if rows else 1)) if headers: - header_cells = table.rows[0].cells - for index, value in enumerate(headers): - header_cells[index].text = value - else: - if rows: - first = rows.pop(0) - for index, value in enumerate(first): - table.rows[0].cells[index].text = value - for row in rows: - new_row = table.add_row().cells - for index, value in enumerate(row): - new_row[index].text = value + 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) @@ -87,19 +155,25 @@ def _append_empty_paragraph_after_table(table: Table, style_name: str | None = N return new_para -def _find_heading_paragraph(document: DocumentObject, heading_text: str) -> Paragraph | None: +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(): - return block + 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): - heading = _find_heading_paragraph(document, heading_title) +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 + 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: @@ -109,9 +183,14 @@ def _replace_section_content(document: DocumentObject, heading_title: str, conte 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): - blocks_to_remove.append(Table(current, heading._parent)) + 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: @@ -124,17 +203,31 @@ def _replace_section_content(document: DocumentObject, heading_title: str, conte 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 = _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", "") - insert_after = _append_paragraph_after(insert_after, text, first_body_style) + 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: - _replace_section_content(document, item["title"], item["content"]) + last_heading_element = _replace_section_content( + document, + item["title"], + item["content"], + last_heading_element, + ) output = BytesIO() document.save(output) diff --git a/docs/tasks/task_detail_2026_07_02.md b/docs/tasks/task_detail_2026_07_02.md index 315fe32..b0be953 100644 --- a/docs/tasks/task_detail_2026_07_02.md +++ b/docs/tasks/task_detail_2026_07_02.md @@ -208,3 +208,14 @@ 2. 调整参考附件序列化方法，将 `created_at` 统一转为 ISO 字符串后再写入任务快照 JSON。 3. 执行后端语法检查，确认修复稳定。 - **执行结果**: 当前生成任务提交时不会再因为附件记录中的 `datetime` 字段导致 JSON 序列化失败。 + +## 会话 ID: local-20260702182751 +- [2026-07-02 18:27:51] +- **执行原因**: 用户反馈导出的模板与原模板几乎一致，AI 结果没有正确写入，同时模板样式在导出后发生明显偏移。 +- **执行过程**: + 1. 检查导出链路，确认原实现采用“删除原内容后新建普通段落/表格”的方式回写，容易导致标题命中不稳和样式丢失。 + 2. 重构导出服务的段落替换逻辑，改为按标题顺序匹配模板中的章节，减少重复标题导致的误替换风险。 + 3. 调整导出时的文本回写方式，优先复用模板原有段落的段落属性与首个 run 的字体样式，再写入 AI 返回文本。 + 4. 调整导出时的表格回写方式，优先克隆模板原有表格结构并填充新数据，尽量保留表格外观和基础样式。 + 5. 执行后端语法检查，确认导出服务改动稳定。 +- **执行结果**: 当前导出链路已改为“尽量复用模板原始段落/表格样式后写入 AI 内容”，比此前的新建空白内容块方式更接近原模板样式，也更容易把 AI 结果正确写回文档。