proposa-ai/scripts/tests/test_step5_table_dedup.py

"""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
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


W = "{http://schemas.openxmlformats.org/wordprocessingml/2006/main}"
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)
        if media:
            p.add_run()._r.append(OxmlElement("w:drawing"))
        doc.add_paragraph("日期：" + date)

    def test_complete_blank_signature_block_yields_to_filled_block(self):
        doc = Document()
        for title in ("资格条件响应表", "实质性要求响应表"):
            doc.add_heading(title, 2)
            doc.add_table(rows=1, cols=1)
            self.add_signature(doc)
            doc.add_paragraph("")
            doc.add_paragraph().add_run()._r.append(OxmlElement("w:drawing"))
            self.add_signature(doc, "测试单位")
        lines = self.clean(doc)
        self.assertEqual(lines.count("投标人（公章）：测试单位"), 2)
        self.assertNotIn("投标人（公章）：", lines,
                         "Remove only the redundant blank complete block after each table")
        self.assertEqual(lines.count("日期：年月日"), 2)

    def test_conflicting_or_image_signature_blocks_are_preserved(self):
        for conflict, media in ((True, False), (False, True)):
            with self.subTest(conflict=conflict, media=media):
                doc = Document()
                doc.add_table(rows=1, cols=1)
                self.add_signature(doc, "测试单位甲", media=media)
                self.add_signature(doc, "测试单位乙" if conflict else "测试单位甲")
                lines = self.clean(doc)
                self.assertEqual(lines.count("日期：年月日"), 2)

    def clean(self, doc, remove_keys=None):
        with tempfile.TemporaryDirectory() as folder:
            source = Path(folder) / "source.docx"
            target = Path(folder) / "clean.docx"
            again = Path(folder) / "again.docx"
            doc.save(source)
            _safe_copy_and_clean_chapter(source, target, False, remove_keys)
            _safe_copy_and_clean_chapter(target, again, False, remove_keys)
            with zipfile.ZipFile(source) as z:
                original = ET.fromstring(z.read("word/document.xml"))
            with zipfile.ZipFile(target) as z, zipfile.ZipFile(again) as a:
                result = ET.fromstring(z.read("word/document.xml"))
                self.assertEqual(z.read("word/document.xml"), a.read("word/document.xml"),
                                 "Step5 cleanup must be idempotent")
                for name in z.namelist():
                    if name != "word/document.xml":
                        with zipfile.ZipFile(source) as s:
                            self.assertEqual(s.read(name), z.read(name))
            self.assertEqual(
                [ET.tostring(t) for t in original.iter(W + "tbl")],
                [ET.tostring(t) for t in result.iter(W + "tbl")],
                "Never change native table XML to remove postamble duplicates",
            )
            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"]

    def test_both_response_tables_remove_cell_and_numbered_echoes(self):
        doc = Document()
        for title in ("资格条件响应表", "实质性要求响应表"):
            doc.add_heading(title, 2)
            table = doc.add_table(rows=1, cols=1)
            table.cell(0, 0).text = "1、" + RESPONSE
            doc.add_paragraph("1、" + RESPONSE)
            doc.add_paragraph("此处为独有的补充材料位置说明，应当完整保留。")
        lines = self.clean(doc)
        self.assertNotIn("1、" + RESPONSE, lines,
                         "Include table text in dedup; numbered responses are not headings")
        self.assertEqual(lines.count("此处为独有的补充材料位置说明，应当完整保留。"), 2)

    def test_postamble_repeated_project_values_and_notes(self):
        doc = Document()
        doc.add_heading("资格条件响应表", 2)
        doc.add_table(rows=1, cols=1).cell(0, 0).text = RESPONSE
        for _ in range(2):
            doc.add_paragraph("项目名称：测试")
            doc.add_paragraph("项目编号：A1")
            doc.add_paragraph("注：本表应附相关资格证明材料，具体以附件为准。")
        lines = self.clean(doc)
        self.assertEqual(lines.count("项目名称：测试"), 1,
                         "Filled project labels must not bypass duplicate detection")
        self.assertEqual(lines.count("项目编号：A1"), 1)
        self.assertEqual(lines.count("注：本表应附相关资格证明材料，具体以附件为准。"), 1)

    def test_signatures_notes_and_next_section_survive(self):
        doc = Document()
        doc.add_heading("资格条件响应表", 2)
        note = "注：本表应附相关资格证明材料，具体以附件为准。"
        doc.add_table(rows=1, cols=1).cell(0, 0).text = RESPONSE + "\n" + note
        doc.add_paragraph(note)
        for _ in range(2):
            doc.add_paragraph("投标人（公章）：测试单位")
            doc.add_paragraph("日期：2026年9月11日")
        doc.add_heading("另一节", 2)
        doc.add_paragraph(RESPONSE)
        lines = self.clean(doc)
        self.assertIn(note, lines)
        self.assertEqual(lines.count("投标人（公章）：测试单位"), 2)
        self.assertEqual(lines.count("日期：2026年9月11日"), 2)
        self.assertIn(RESPONSE, lines, "Do not deduplicate across heading boundaries")

    def test_new_table_replaces_corpus_and_keeps_different_response(self):
        doc = Document()
        doc.add_heading("资格条件响应表", 2)
        doc.add_table(rows=1, cols=1).cell(0, 0).text = RESPONSE
        doc.add_paragraph("我方不具备履行本项目合同所必需的设备和专业技术能力。")
        doc.add_table(rows=1, cols=1).cell(0, 0).text = "另一张独立表格，载有不同的响应内容。"
        doc.add_paragraph(RESPONSE)
        lines = self.clean(doc)
        self.assertIn(RESPONSE, lines)
        self.assertTrue(any("不具备" in line for line in lines))

    def test_bookmark_and_drawing_paragraphs_are_not_table_echoes(self):
        doc = Document()
        doc.add_table(rows=1, cols=1).cell(0, 0).text = RESPONSE
        p = doc.add_paragraph(RESPONSE)
        p._p.append(OxmlElement("w:bookmarkStart"))
        p = doc.add_paragraph("配图说明文字属于原生图片块，不应因表内相同文字而删除。")
        p.add_run()._r.append(OxmlElement("w:drawing"))
        lines = self.clean(doc)
        self.assertIn(RESPONSE, lines)

    def test_equivalent_postamble_keeps_later_template_paragraph(self):
        doc = Document()
        doc.add_table(rows=1, cols=1)
        note = "注：本表应附相关资格证明材料，具体以附件为准。"
        for marker in ("11111111", "22222222"):
            p = doc.add_paragraph(note)
            p._p.set("{http://schemas.microsoft.com/office/word/2010/wordml}paraId", marker)
        self.clean(doc, {note})
        survivors = [p for p in self.last_body if ''.join(
            n.text or '' for n in p.iter(W + 't')) == note]
        self.assertEqual(len(survivors), 1, "LLM text removals must not delete the survivor")
        self.assertEqual(survivors[0].get(
            "{http://schemas.microsoft.com/office/word/2010/wordml}paraId"), "22222222")

    def test_richer_postamble_survives_in_either_order(self):
        short = "注：本表应附相关资格证明材料，具体以附件为准。"
        rich = short + "证明材料须注明有效期，并提供查询方式。"
        unique = "表格自带的独有说明，应同时提交纸质复印件。"
        for values in ((short, rich), (rich, short)):
            with self.subTest(values=values):
                doc = Document()
                doc.add_table(rows=1, cols=1)
                doc.add_paragraph(values[0])
                doc.add_paragraph(unique)
                doc.add_paragraph(values[1])
                lines = self.clean(doc, {rich, short, unique})
                self.assertIn(rich, lines, "Retain all additional clauses, regardless of order")
                self.assertNotIn(short, lines)
                self.assertIn(unique, lines, "A unique source-table note must never be deleted")

    def test_different_or_conflicting_sentences_are_not_length_ranked(self):
        doc = Document()
        doc.add_table(rows=1, cols=1)
        values = ["我方提交全部资格证明材料。", "我方不提交全部资格证明材料。",
                  "资格证明材料有效期为三年。", "资格证明材料有效期为五年，并包含附加证明。"]
        for value in values:
            doc.add_paragraph(value)
        lines = self.clean(doc, set(values))
        for value in values:
            self.assertIn(value, lines, "Length or shared keywords are not proof of duplication")

    def test_signature_keeps_later_location_and_complementary_values(self):
        doc = Document()
        doc.add_table(rows=1, cols=1)
        self.add_signature(doc, "测试单位")
        note = "源表独有说明：请保留本说明及随附证明。"
        doc.add_paragraph(note)
        self.add_signature(doc, date="2026年9月14日")
        lines = self.clean(doc, {note})
        self.assertEqual(lines.count("投标人授权代表签字（盖章）："), 1)
        self.assertIn("投标人（公章）：测试单位", lines)
        self.assertIn("日期：2026年9月14日", lines)
        self.assertGreater(lines.index("投标人（公章）：测试单位"), lines.index(note),
                           "Retain the template block's later position with all filled values")

    def test_equal_short_project_fields_keep_later_and_empty_yields_to_filled(self):
        doc = Document()
        doc.add_table(rows=1, cols=1)
        doc.add_paragraph("项目名称：")
        doc.add_paragraph("项目名称：测试")
        doc.add_paragraph("项目编号：A1")
        doc.add_paragraph("此处为前后两份字段之间的独有说明。")
        doc.add_paragraph("项目编号：A1")
        lines = self.clean(doc, {"项目名称：测试", "项目编号：A1"})
        self.assertNotIn("项目名称：", lines)
        self.assertEqual(lines.count("项目名称：测试"), 1)
        self.assertEqual(lines.count("项目编号：A1"), 1)
        self.assertGreater(lines.index("项目编号：A1"),
                           lines.index("此处为前后两份字段之间的独有说明。"))


if __name__ == "__main__":
    unittest.main()