修复DOCX表格漏译并增加残留校验

docutranslate/translator/ai_translator/docx_translator.py

@@ -21,6 +21,7 @@ from docx.table import _Cell, Table
 from lxml import etree
 
 from docutranslate.agents.segments_agent import SegmentsTranslateAgentConfig, SegmentsTranslateAgent
+from docutranslate.glossary.lang_mapping import lang_to_col
 from docutranslate.ir.document import Document
 from docutranslate.translator.ai_translator.base import AiTranslatorConfig, AiTranslator
 
@@ -87,6 +88,7 @@ class DocxTranslator(AiTranslator):
     def __init__(self, config: DocxTranslatorConfig):
         super().__init__(config=config)
         self.chunk_size = config.chunk_size
+        self.to_lang = config.to_lang
         self.translate_agent = None
         glossary_dict = self.glossary.glossary_dict if self.glossary else None
         if not self.skip_translate:
@@ -338,10 +340,10 @@ class DocxTranslator(AiTranslator):
                 seen_cells = set()
                 for row in table.rows:
                     for cell in row.cells:
-                        cell_id = id(cell._tc)
-                        if cell_id in seen_cells:
+                        if cell._tc in seen_cells:
                             continue
-                        seen_cells.add(cell_id)
+                        # Retain XML proxies: bare ids can be reused after empty cells are released.
+                        seen_cells.add(cell._tc)
                         self._traverse_container(cell, elements, texts)
             elif child_element.tag.endswith('sdt'):
                 sdt_content = child_element.find(qn('w:sdtContent'))
@@ -388,10 +390,9 @@ class DocxTranslator(AiTranslator):
             else:
                 parent_element = None
             if parent_element is not None:
-                element_id = id(parent_element)
-                if element_id in seen_container_elements:
+                if parent_element in seen_container_elements:
                     return
-                seen_container_elements.add(element_id)
+                seen_container_elements.add(parent_element)
             self._traverse_container(container, elements, texts)
 
         traverse_once(doc)
@@ -418,6 +419,61 @@ class DocxTranslator(AiTranslator):
     def _looks_translatable_text(text: str) -> bool:
         return len(re.findall(r"\b[A-Za-z]{4,}\b", text)) >= 2
 
+    def _incomplete_translation_indices(self, originals: List[str], translated: List[str]) -> List[int]:
+        if len(originals) != len(translated):
+            raise ValueError("DOCX 译文数量与原文不一致，已停止生成文件，请重新翻译。")
+
+        check_chinese = lang_to_col(self.to_lang) in {"en", "id"}
+        incomplete = []
+        for index, (original, result) in enumerate(zip(originals, translated)):
+            if not isinstance(result, str) or (original.strip() and not result.strip()):
+                incomplete.append(index)
+                continue
+            if not check_chinese:
+                continue
+            # Explicit glossary targets may intentionally retain Chinese names or terms.
+            remaining = result
+            for source, target in self.glossary.glossary_dict.items():
+                if source and target and source.casefold() in original.casefold():
+                    remaining = remaining.replace(target, "")
+            if re.search(r"[\u3400-\u4dbf\u4e00-\u9fff\uf900-\ufaff]", remaining):
+                incomplete.append(index)
+        return incomplete
+
+    def _require_complete_translations(self, originals: List[str], translated: List[str]) -> None:
+        remaining = self._incomplete_translation_indices(originals, translated)
+        if remaining:
+            raise ValueError(
+                f"DOCX 仍有 {len(remaining)} 个片段未完成翻译（中文残留或空译文），"
+                "已定向重试 2 次，未生成不完整文件。请重试；如需保留中文，请在术语表中指定。"
+            )
+
+    def _repair_incomplete_translations(self, originals: List[str], translated: List[str]) -> None:
+        for attempt in range(2):
+            incomplete = self._incomplete_translation_indices(originals, translated)
+            if not incomplete:
+                return
+            self.logger.warning(f"DOCX translation audit: retry {attempt + 1}/2 for {len(incomplete)} incomplete segments.")
+            for index in incomplete:
+                source = [originals[index]]
+                result = self.translate_agent.send_segments(source, self.chunk_size)
+                self._incomplete_translation_indices(source, result)
+                translated[index] = result[0]
+        self._require_complete_translations(originals, translated)
+
+    async def _repair_incomplete_translations_async(self, originals: List[str], translated: List[str]) -> None:
+        for attempt in range(2):
+            incomplete = self._incomplete_translation_indices(originals, translated)
+            if not incomplete:
+                return
+            self.logger.warning(f"DOCX translation audit: retry {attempt + 1}/2 for {len(incomplete)} incomplete segments.")
+            for index in incomplete:
+                source = [originals[index]]
+                result = await self.translate_agent.send_segments_async(source, self.chunk_size)
+                self._incomplete_translation_indices(source, result)
+                translated[index] = result[0]
+        self._require_complete_translations(originals, translated)
+
     def _log_translation_quality(self, originals: List[str], translated: List[str]) -> None:
         if not originals or not translated:
             return
@@ -750,6 +806,8 @@ class DocxTranslator(AiTranslator):
 
         translated = self.translate_agent.send_segments(originals,
                                                         self.chunk_size) if self.translate_agent else originals
+        if self.translate_agent:
+            self._repair_incomplete_translations(originals, translated)
         document.content = self._after_translate(doc, elements, translated, originals)
         return self
 
@@ -774,5 +832,7 @@ class DocxTranslator(AiTranslator):
 
         translated = await self.translate_agent.send_segments_async(originals,
                                                                     self.chunk_size) if self.translate_agent else originals
+        if self.translate_agent:
+            await self._repair_incomplete_translations_async(originals, translated)
         document.content = await asyncio.to_thread(self._after_translate, doc, elements, translated, originals)
         return self

tests/test_docx_translator.py

@@ -1,3 +1,4 @@
+import asyncio
 from io import BytesIO
 import unittest
 from zipfile import ZipFile
@@ -266,10 +267,190 @@ def _build_fixture_docx():
     return _docx_bytes(document)
 
 
+def _build_signature_table_docx():
+    document = docx.Document()
+    table = document.add_table(rows=2, cols=11)
+    table.cell(0, 0).text = "日期"
+    paragraph = table.cell(1, 0).paragraphs[0]
+    paragraph.add_run("姓名&签名")
+    paragraph.add_run("  ")
+    return _docx_bytes(document)
+
+
+class ScriptedAgent:
+    def __init__(self, responses):
+        self.responses = iter(responses)
+        self.calls = []
+
+    def send_segments(self, segments, chunk_size):
+        self.calls.append(list(segments))
+        return next(self.responses)
+
+    async def send_segments_async(self, segments, chunk_size):
+        return self.send_segments(segments, chunk_size)
+
+
 class DocxTranslatorTest(unittest.TestCase):
     def _translator(self, skip_translate=True):
         return DocxTranslator(DocxTranslatorConfig(skip_translate=skip_translate))
 
+    def _quality_translator(self, responses, **options):
+        translator = DocxTranslator(DocxTranslatorConfig(
+            api_key="dummy", base_url="http://127.0.0.1", model_id="dummy",
+            **options,
+        ))
+        translator.translate_agent = ScriptedAgent(responses)
+        return translator
+
+    def test_only_incomplete_segments_are_retried(self):
+        for use_async in (False, True):
+            for incomplete in ("姓名&签名  ", "Nama & 签名", "", None):
+                with self.subTest(use_async=use_async, incomplete=incomplete):
+                    translator = self._quality_translator(
+                        [["Tanggal", incomplete], ["Nama & Tanda Tangan"]],
+                        to_lang="Bahasa Indonesia",
+                    )
+                    document = Document(content=_build_signature_table_docx(), suffix=".docx")
+                    if use_async:
+                        asyncio.run(translator.translate_async(document))
+                    else:
+                        translator.translate(document)
+                    self.assertEqual(translator.translate_agent.calls, [
+                        ["日期", "姓名&签名  "], ["姓名&签名  "],
+                    ])
+                    result = docx.Document(BytesIO(document.content))
+                    self.assertEqual(result.tables[0].cell(1, 0).text, "Nama & Tanda Tangan")
+
+    def test_incomplete_translation_fails_after_two_repairs_without_changing_input(self):
+        for use_async in (False, True):
+            with self.subTest(use_async=use_async):
+                translator = self._quality_translator(
+                    [["Tanggal", "姓名&签名"], ["Nama & 签名"], ["Nama & 签名"]],
+                    to_lang="Indonesian",
+                )
+                source = _build_signature_table_docx()
+                document = Document(content=source, suffix=".docx")
+                with self.assertRaisesRegex(ValueError, "已定向重试 2 次"):
+                    if use_async:
+                        asyncio.run(translator.translate_async(document))
+                    else:
+                        translator.translate(document)
+                self.assertEqual(document.content, source)
+                self.assertEqual(len(translator.translate_agent.calls), 3)
+
+    def test_second_repair_can_complete_translation(self):
+        translator = self._quality_translator(
+            [["Tanggal", "姓名&签名"], ["Nama & 签名"], ["Nama & Tanda Tangan"]],
+            to_lang="印尼语",
+        )
+        document = Document(content=_build_signature_table_docx(), suffix=".docx")
+        translator.translate(document)
+        self.assertEqual(len(translator.translate_agent.calls), 3)
+        self.assertIn("Nama & Tanda Tangan", "".join(_text_values(document.content)))
+
+    def test_translation_quality_respects_target_language_and_glossary(self):
+        for language in ("English", "en-US", "英文", "Indonesian", "id", "印尼语"):
+            with self.subTest(language=language):
+                translator = self._quality_translator([], to_lang=language)
+                self.assertEqual(translator._incomplete_translation_indices(
+                    ["姓名&签名"], ["Nama & 签名"]
+                ), [0])
+        for language in ("简体中文", "繁体中文", "Japanese"):
+            with self.subTest(language=language):
+                translator = self._quality_translator([], to_lang=language)
+                self.assertEqual(translator._incomplete_translation_indices(
+                    ["姓名&签名"], ["姓名&签名"]
+                ), [])
+        translator = self._quality_translator(
+            [], to_lang="English", glossary_dict={"华宝": "华宝"},
+        )
+        self.assertEqual(translator._incomplete_translation_indices(
+            ["华宝签名"], ["华宝 signature"]
+        ), [])
+        self.assertEqual(translator._incomplete_translation_indices(
+            ["华宝签名"], ["华宝签名"]
+        ), [0])
+
+    def test_wrong_translation_count_is_not_silently_truncated(self):
+        for use_async in (False, True):
+            for responses in ([["Tanggal"]], [["Tanggal", "姓名&签名"], []]):
+                with self.subTest(use_async=use_async, responses=responses):
+                    translator = self._quality_translator(responses, to_lang="Indonesian")
+                    source = _build_signature_table_docx()
+                    document = Document(content=source, suffix=".docx")
+                    with self.assertRaisesRegex(ValueError, "译文数量与原文不一致"):
+                        if use_async:
+                            asyncio.run(translator.translate_async(document))
+                        else:
+                            translator.translate(document)
+                    self.assertEqual(document.content, source)
+
+    def test_signature_label_after_empty_cells_is_always_extracted(self):
+        input_bytes = _build_signature_table_docx()
+        translator = self._translator()
+        for iteration in range(50):
+            with self.subTest(iteration=iteration):
+                _, _, originals = translator._pre_translate(
+                    Document(content=input_bytes, suffix=".docx")
+                )
+                self.assertEqual(originals, ["日期", "姓名&签名  "])
+
+    def test_empty_and_merged_cells_do_not_hide_later_labels(self):
+        document = docx.Document()
+        table = document.add_table(rows=5, cols=11)
+        table.cell(0, 0).merge(table.cell(0, 1)).text = "日期"
+        table.cell(1, 0).merge(table.cell(2, 0)).text = "姓名&签名"
+        table.cell(4, 0).text = "姓名&签名"
+        table.cell(4, 10).text = "审批<通过>"
+        input_bytes = _docx_bytes(document)
+
+        for iteration in range(20):
+            with self.subTest(iteration=iteration):
+                _, _, originals = self._translator()._pre_translate(
+                    Document(content=input_bytes, suffix=".docx")
+                )
+                self.assertEqual(
+                    originals, ["日期", "姓名&签名", "姓名&签名", "审批<通过>"]
+                )
+
+    def test_signature_label_round_trip_sync_and_async(self):
+        class FakeAgent:
+            def send_segments(self, segments, chunk_size):
+                translations = {
+                    "日期": "Tanggal",
+                    "姓名&签名  ": "Nama & Tanda Tangan  ",
+                }
+                return [translations[segment] for segment in segments]
+
+            async def send_segments_async(self, segments, chunk_size):
+                return self.send_segments(segments, chunk_size)
+
+        input_bytes = _build_signature_table_docx()
+        for use_async in (False, True):
+            for mode in ("replace", "append", "prepend"):
+                with self.subTest(use_async=use_async, mode=mode):
+                    translator = DocxTranslator(DocxTranslatorConfig(
+                        skip_translate=False, insert_mode=mode,
+                        api_key="dummy", base_url="http://127.0.0.1",
+                        model_id="dummy", to_lang="Indonesian",
+                    ))
+                    translator.translate_agent = FakeAgent()
+                    document = Document(content=input_bytes, suffix=".docx")
+                    if use_async:
+                        asyncio.run(translator.translate_async(document))
+                    else:
+                        translator.translate(document)
+                    result = docx.Document(BytesIO(document.content))
+                    self.assertEqual(len(result.tables), 1)
+                    self.assertEqual(len(result.tables[0].rows), 2)
+                    self.assertEqual(len(result.tables[0].columns), 11)
+                    self.assertIn("Tanggal", result.tables[0].cell(0, 0).text)
+                    label = result.tables[0].cell(1, 0).text
+                    self.assertIn("Nama & Tanda Tangan", label)
+                    self.assertEqual("姓名&签名" in label, mode != "replace")
+                    for row in result.tables[0].rows:
+                        self.assertTrue(all(not cell.text for cell in row.cells[1:]))
+
     def test_pre_translate_extracts_docx_boundary_text(self):
         translator = self._translator()
         _, _, originals = translator._pre_translate(Document(content=_build_fixture_docx(), suffix=".docx"))