diff --git a/docutranslate/translator/ai_translator/docx_translator.py b/docutranslate/translator/ai_translator/docx_translator.py index 979b908..f1defae 100644 --- a/docutranslate/translator/ai_translator/docx_translator.py +++ b/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 diff --git a/tests/test_docx_translator.py b/tests/test_docx_translator.py index 27789f2..5723f5e 100644 --- a/tests/test_docx_translator.py +++ b/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"))