2

scripts/test_step5.py

@@ -473,6 +473,7 @@ def _safe_copy_and_clean_chapter(source, target, has_packages, remove_keys=None,
             if body is not None:
                 from step5_reviewing.table_dedup import (
                     clear_duplicate_signature_blocks, clear_table_repetitions,
+                    remove_empty_body_paragraphs,
                 )
                 clear_duplicate_signature_blocks(body, _TITLE_PREFIX_RE)
                 protected_postamble = set()
@@ -556,6 +557,11 @@ def _safe_copy_and_clean_chapter(source, target, has_packages, remove_keys=None,
                                 text_nodes[0].text = f"{label}：{value}"
                             break
                     _apply_body_format(paragraph, W_NS)
+                styles_root = (
+                    ET.fromstring(source_zip.read("word/styles.xml"))
+                    if "word/styles.xml" in names else None
+                )
+                remove_empty_body_paragraphs(body, styles_root)
             document_out = ET.tostring(
                 root, encoding="utf-8", xml_declaration=True
             )

scripts/tests/test_step4_response_table.py

@@ -0,0 +1,134 @@
+"""响应表覆盖必须使用Word网格，原要求单元格不可压平或移位。"""
+import os
+import tempfile
+import unittest
+import xml.etree.ElementTree as ET
+
+from docx import Document
+from docx.enum.text import WD_ALIGN_PARAGRAPH
+
+from models import Chapter, ExtractedTable, TableCell, TenderAnalysis
+from step4_writing import (
+    _fill_current_response_table, _overlay_table_xml_text,
+    _save_chapter_docx, _w_tag,
+)
+
+
+def text(element):
+    return ''.join(t.text or '' for t in element.iter(_w_tag('t')))
+
+
+def canonical_cell(cell):
+    return ET.tostring(ET.fromstring(cell._tc.xml))
+
+
+class ResponseTableTests(unittest.TestCase):
+    def fixture(self):
+        doc = Document()
+        table = doc.add_table(rows=4, cols=5)
+        headers = ['类别', '序号', '资格要求', '是否响应', '证明材料']
+        for i, value in enumerate(headers):
+            table.cell(0, i).text = value
+        table.cell(1, 0).merge(table.cell(3, 0)).text = '资格'
+        cells = [[TableCell(text=value, col=i) for i, value in enumerate(headers)]]
+        for r in range(1, 4):
+            table.cell(r, 1).text = str(r)
+            p = table.cell(r, 2).paragraphs[0]
+            p.alignment = WD_ALIGN_PARAGRAPH.CENTER
+            p.add_run(f'{r}、提供').bold = True
+            p.add_run('示例证明。').underline = True
+            table.cell(r, 2).add_paragraph('原说明必须保留。')
+            cells.append([
+                TableCell(text='资格' if r == 1 else '', row=r, col=0,
+                          rowspan=3 if r == 1 else 1,
+                          is_merged_origin=r == 1, is_merged_continuation=r > 1),
+                TableCell(text=str(r), row=r, col=1),
+                TableCell(text=f'{r}、提供示例证明。\n原说明必须保留。', row=r, col=2),
+                TableCell(row=r, col=3), TableCell(row=r, col=4),
+            ])
+        source = ExtractedTable(rows=4, cols=5, cells=cells,
+                                table_type='qualification_response', source_type='tender_pdf')
+        return doc, table, source
+
+    def test_vertical_merge_does_not_shift_requirements_or_flatten_runs(self):
+        _, table, source = self.fixture()
+        xml = ET.fromstring(table._tbl.xml)
+        before = [[ET.tostring(tc) for tc in row.findall(_w_tag('tc'))[:3]]
+                  for row in xml.findall(_w_tag('tr'))]
+        filled = _fill_current_response_table(source, '资格条件响应表')
+        _overlay_table_xml_text(xml, filled)
+        after = [[ET.tostring(tc) for tc in row.findall(_w_tag('tc'))[:3]]
+                 for row in xml.findall(_w_tag('tr'))]
+        self.assertEqual(before, after, '原要求、合并续格及多段混合格式必须原样保留')
+        for row in xml.findall(_w_tag('tr'))[1:]:
+            self.assertEqual(text(row.findall(_w_tag('tc'))[3]), '是')
+
+    def test_horizontal_and_vertical_merge_keep_response_origin_only(self):
+        _, table, source = self.fixture()
+        table.cell(1, 3).merge(table.cell(2, 3))
+        source.cells[1][3].rowspan = 2
+        source.cells[1][3].is_merged_origin = True
+        source.cells[2][3].is_merged_continuation = True
+        table.cell(3, 1).merge(table.cell(3, 2))
+        source.cells[3][1].text = '3\n3、提供示例证明。\n原说明必须保留。'
+        source.cells[3][1].colspan = 2
+        source.cells[3][1].is_merged_origin = True
+        source.cells[3][2].is_merged_continuation = True
+        filled = _fill_current_response_table(source, '资格条件响应表')
+        self.assertEqual(filled.cells[2][3].text, '', '合并续格不得被填充')
+        xml = ET.fromstring(table._tbl.xml)
+        _overlay_table_xml_text(xml, filled)
+        rows = xml.findall(_w_tag('tr'))
+        self.assertEqual(text(rows[2].findall(_w_tag('tc'))[3]), '')
+        self.assertEqual(text(rows[3].findall(_w_tag('tc'))[2]), '是')
+
+    def test_actual_chapter_write_preserves_step1_requirement_xml(self):
+        doc, table, source = self.fixture()
+        with tempfile.TemporaryDirectory() as directory:
+            source.artifact_path = os.path.join(directory, 'source.docx')
+            doc.save(source.artifact_path)
+            template = Document()
+            template.add_paragraph('第一章 资格', style='Heading 1')
+            template.add_paragraph('%%资格条件响应表%%')
+            template.add_paragraph('第二章 报价', style='Heading 1')
+            template_path = os.path.join(directory, 'template.docx')
+            template.save(template_path)
+            path = _save_chapter_docx(
+                Chapter(id='1', title='资格', template_chapter_id='1'), directory,
+                template_path=template_path,
+                analysis=TenderAnalysis(project_name='示例项目', tender_tables=[source]),
+            )
+            result = Document(path).tables[0]
+            for r in range(1, 4):
+                self.assertEqual(result.cell(r, 2).text, table.cell(r, 2).text)
+                self.assertEqual(canonical_cell(result.cell(r, 2)), canonical_cell(table.cell(r, 2)))
+                self.assertEqual(result.cell(r, 3).text, '是')
+            self.assertEqual(source.cells[1][3].text, '', '填表不得污染Step1元数据')
+
+    def test_grid_before_and_existing_response_are_preserved(self):
+        _, table, source = self.fixture()
+        table.cell(1, 3).text = '已有响应'
+        table.cell(2, 3).paragraphs[0].alignment = WD_ALIGN_PARAGRAPH.RIGHT
+        table.cell(2, 3).paragraphs[0].add_run(' ').underline = True
+        xml = ET.fromstring(table._tbl.xml)
+        rows = xml.findall(_w_tag('tr'))
+        first_response = ET.tostring(rows[1].findall(_w_tag('tc'))[3])
+        rows[2].remove(rows[2].find(_w_tag('tc')))
+        pr = ET.Element(_w_tag('trPr'))
+        ET.SubElement(pr, _w_tag('gridBefore')).set(_w_tag('val'), '1')
+        rows[2].insert(0, pr)
+        filled = _fill_current_response_table(source, '资格条件响应表')
+        _overlay_table_xml_text(xml, filled)
+        self.assertEqual(first_response, ET.tostring(rows[1].findall(_w_tag('tc'))[3]))
+        response = rows[2].findall(_w_tag('tc'))[2]
+        self.assertEqual(text(response), '是')
+        self.assertEqual(response.find('.//' + _w_tag('jc')).get(_w_tag('val')), 'right')
+        self.assertIsNotNone(response.find('.//' + _w_tag('u')))
+        self.assertEqual(len(response.findall(_w_tag('p'))), 1, '空单元格沿用原段落')
+        once = ET.tostring(xml)
+        _overlay_table_xml_text(xml, filled)
+        self.assertEqual(once, ET.tostring(xml), '重复写出须幂等')
+
+
+if __name__ == '__main__':
+    unittest.main()

scripts/tests/test_step5_table_dedup.py

@@ -1,6 +1,8 @@
 """Synthetic DOCX regressions for response-table postambles (no LLM)."""
 
 import tempfile
+import base64
+import io
 import sys
 import unittest
 import xml.etree.ElementTree as ET
@@ -8,6 +10,7 @@ import zipfile
 from pathlib import Path
 
 from docx import Document
+from docx.enum.style import WD_STYLE_TYPE
 from docx.oxml import OxmlElement
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 from scripts.test_step5 import _safe_copy_and_clean_chapter
@@ -18,6 +21,57 @@ RESPONSE = "我方具备履行本项目合同所必需的设备和专业技术
 
 
 class TablePostambleTests(unittest.TestCase):
+    def test_sweep_removes_original_and_new_plain_empty_paragraphs(self):
+        doc = Document()
+        doc.add_paragraph("")
+        doc.add_paragraph("  \t\u3000 ")
+        doc.add_paragraph("这段已被审核确定需要删除的冗余正文。")
+        doc.add_heading("资格条件响应表", 2)
+        doc.add_table(rows=1, cols=1)
+        for _ in range(3):
+            doc.add_paragraph("注：本表应附相关资格证明材料，具体以附件为准。")
+        doc.add_paragraph("")
+        lines = self.clean(doc, {"这段已被审核确定需要删除的冗余正文。"})
+        self.assertEqual(lines, ["资格条件响应表", "注：本表应附相关资格证明材料，具体以附件为准。"],
+                         "Sweep must remove old whitespace and newly cleared duplicates, without skipping adjacent paragraphs")
+        self.assertEqual(len(list(self.last_body.iter(W + "tbl"))), 1)
+        self.assertEqual(len(list(self.last_body.iter(W + "tc"))[0].findall(W + "p")), 1)
+
+    def test_empty_image_bookmark_section_and_field_carriers_remain_identical(self):
+        doc = Document()
+        doc.add_paragraph("")
+        png = base64.b64decode(
+            "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+aPl8AAAAASUVORK5CYII="
+        )
+        doc.add_paragraph().add_run().add_picture(io.BytesIO(png))
+        for tag in ("bookmarkStart", "bookmarkEnd", "fldChar", "instrText",
+                    "fldSimple", "commentRangeStart", "commentRangeEnd", "footnoteReference"):
+            doc.add_paragraph()._p.append(OxmlElement("w:" + tag))
+        for tag in ("sectPr", "numPr", "pageBreakBefore", "keepNext"):
+            doc.add_paragraph()._p.get_or_add_pPr().append(OxmlElement("w:" + tag))
+        doc.add_paragraph().add_run().add_break()
+        # Save exact XML for every carrier; only the leading plain blank may go.
+        self.clean(doc)
+        expected = [ET.tostring(p) for p in self.original_body if p.tag == W + "p"][1:]
+        actual = [ET.tostring(p) for p in self.last_body if p.tag == W + "p"]
+        self.assertEqual(actual, expected, "Do not move or rewrite empty structural/image anchor paragraphs")
+
+    def test_empty_paragraph_inherited_pagination_and_numbering_are_preserved(self):
+        doc = Document()
+        base = doc.styles.add_style("LayoutBase", WD_STYLE_TYPE.PARAGRAPH)
+        base.paragraph_format.page_break_before = True
+        derived = doc.styles.add_style("LayoutDerived", WD_STYLE_TYPE.PARAGRAPH)
+        derived.base_style = base
+        numbering = doc.styles.add_style("NumberedEmpty", WD_STYLE_TYPE.PARAGRAPH)
+        numbering.element.get_or_add_pPr().append(OxmlElement("w:numPr"))
+        doc.add_paragraph("", style=derived)
+        doc.add_paragraph("", style=numbering)
+        doc.add_paragraph("")
+        self.clean(doc)
+        self.assertEqual([p.find(W + "pPr/" + W + "pStyle").get(W + "val")
+                          for p in self.last_body if p.tag == W + "p"],
+                         ["LayoutDerived", "NumberedEmpty"])
+
     def add_signature(self, doc, company="", date="年月日", media=False):
         doc.add_paragraph("投标人授权代表签字（盖章）：")
         p = doc.add_paragraph("投标人（公章）：" + company)
@@ -76,6 +130,7 @@ class TablePostambleTests(unittest.TestCase):
             self.assertEqual([ET.tostring(t) for t in original.iter(W + "drawing")],
                              [ET.tostring(t) for t in result.iter(W + "drawing")])
             self.last_body = result.find(W + "body")
+            self.original_body = original.find(W + "body")
             return ["".join(t.text or "" for t in p.iter(W + "t"))
                     for p in result.find(W + "body") if p.tag == W + "p"]
 

src/step4_writing/__init__.py

@@ -2180,10 +2180,16 @@ def _fill_current_response_table(source_table: Any, key: str) -> Any:
         return table
 
     for row in cells[1:]:
-        if response_col < len(row) and not str(row[response_col].text or "").strip():
+        if (
+            response_col < len(row)
+            and not getattr(row[response_col], "is_merged_continuation", False)
+            and not str(row[response_col].text or "").strip()
+        ):
             row[response_col].text = "是"
         if evidence_col < 0 or evidence_col >= len(row):
             continue
+        if getattr(row[evidence_col], "is_merged_continuation", False):
+            continue
         if str(row[evidence_col].text or "").strip():
             continue
         item_name = str(getattr(row[0], "text", "") or "") if row else ""
@@ -2200,23 +2206,37 @@ def _fill_current_response_table(source_table: Any, key: str) -> Any:
 
 
 def _overlay_table_xml_text(table_xml, source_table: Any):
-    """在来源原生表 XML 上覆盖单元格文字，保留表格及单元格样式。"""
+    """按Word网格只填空白响应/证明格，其他原生XML保持不变。
+
+    横向合并省略物理tc，纵向合并续格却仍有tc；不能过滤元数据续格后zip。
+    Step1原文不应再用扁平元数据覆盖，否则会压平段落、run并丢失格式。
+    """
     rows = list(getattr(source_table, "cells", []) or [])
+    fill_columns = {col for col in _response_table_columns(source_table) if col >= 0}
     xml_rows = table_xml.findall(_w_tag("tr"))
-    for xml_row, row in zip(xml_rows, rows):
-        logical_cells = [
-            cell for cell in row
-            if not getattr(cell, "is_merged_continuation", False)
-        ]
-        for xml_cell, cell in zip(xml_row.findall(_w_tag("tc")), logical_cells):
+    for xml_row, row in zip(xml_rows[1:], rows[1:]):
+        before = xml_row.find(_w_tag("trPr") + "/" + _w_tag("gridBefore"))
+        col = int(before.get(_w_tag("val"), "0")) if before is not None else 0
+        for xml_cell in xml_row.findall(_w_tag("tc")):
+            span = xml_cell.find(_w_tag("tcPr") + "/" + _w_tag("gridSpan"))
+            width = max(1, int(span.get(_w_tag("val"), "1"))) if span is not None else 1
+            cell_col = col
+            col += width
+            merge = xml_cell.find(_w_tag("tcPr") + "/" + _w_tag("vMerge"))
+            if merge is not None and merge.get(_w_tag("val")) != "restart":
+                continue
+            if cell_col not in fill_columns or cell_col >= len(row):
+                continue
+            cell = row[cell_col]
+            if getattr(cell, "is_merged_continuation", False):
+                continue
             value = str(getattr(cell, "text", "") or "")
-            text_nodes = list(xml_cell.iter(_w_tag("t")))
-            if text_nodes:
-                text_nodes[0].text = value
-                for node in text_nodes[1:]:
-                    node.text = ""
-            elif value:
-                xml_cell.append(_new_docx_paragraph(value))
+            if not value.strip() or _docx_xml_text(xml_cell).strip():
+                continue
+            paragraph = xml_cell.find(_w_tag("p"))
+            if paragraph is None:
+                paragraph = ET.SubElement(xml_cell, _w_tag("p"))
+            _fill_docx_paragraph_text(paragraph, value)
     return table_xml
 
 

src/step5_reviewing/table_dedup.py

@@ -22,6 +22,61 @@ def _key(text):
     return re.sub(r"[\s\u3000]+", "", text)
 
 
+def remove_empty_body_paragraphs(body, styles_root=None):
+    """Remove plain empty body paragraphs, retaining layout/content carriers.
+
+    Do not descend into tables: Word requires paragraph nodes in cells.
+    Inherited paragraph styles may carry numbering or pagination even when
+    the paragraph itself only contains whitespace.
+    """
+    protected = _STRUCTURAL | {W + name for name in (
+        "fldSimple", "footnoteReference", "endnoteReference", "commentReference",
+        "permStart", "permEnd", "sdt", "customXml", "ins", "del", "moveFrom",
+        "moveTo", "hyperlink", "sym", "br", "cr", "lastRenderedPageBreak",
+        "numPr", "framePr", "outlineLvl",
+    )}
+    styles = {} if styles_root is None else {
+        style.get(W + "styleId"): style for style in styles_root.findall(W + "style")
+    }
+    defaults = [] if styles_root is None else styles_root.findall(
+        W + "docDefaults/" + W + "pPrDefault/" + W + "pPr"
+    )
+    default_style = next((style for style in styles.values()
+                          if style.get(W + "type") == "paragraph"
+                          and style.get(W + "default") in {"1", "true", "on"}), None)
+
+    def carries_layout(element):
+        for node in element.iter():
+            if node.tag in protected:
+                return True
+            if node.tag in {W + "pageBreakBefore", W + "keepNext", W + "keepLines"}:
+                if node.get(W + "val", "1").lower() not in {"0", "false", "off"}:
+                    return True
+        return False
+
+    removed = 0
+    for paragraph in list(body):
+        if paragraph.tag != W + "p" or _text(paragraph).strip():
+            continue
+        if carries_layout(paragraph) or any(carries_layout(p) for p in defaults):
+            continue
+        p_style = paragraph.find(W + "pPr/" + W + "pStyle")
+        style = styles.get(p_style.get(W + "val")) if p_style is not None else default_style
+        visited = set()
+        keep = False
+        while style is not None and id(style) not in visited:
+            visited.add(id(style))
+            if carries_layout(style):
+                keep = True
+                break
+            based_on = style.find(W + "basedOn")
+            style = styles.get(based_on.get(W + "val")) if based_on is not None else None
+        if not keep:
+            body.remove(paragraph)
+            removed += 1
+    return removed
+
+
 def _signature_field(text):
     label, sep, value = re.sub(r"\s+", "", text).partition("：")
     if not sep: