diff --git a/.gitignore b/.gitignore index d834a8b..9c0dd20 100644 --- a/.gitignore +++ b/.gitignore @@ -6,7 +6,6 @@ dist/ wheels/ *.egg-info tests/resource/ -tests/ docutranslate/output/ # Shared glossary runtime data data/ diff --git a/docutranslate/translator/ai_translator/docx_translator.py b/docutranslate/translator/ai_translator/docx_translator.py index f3348c5..979b908 100644 --- a/docutranslate/translator/ai_translator/docx_translator.py +++ b/docutranslate/translator/ai_translator/docx_translator.py @@ -1,6 +1,7 @@ # SPDX-FileCopyrightText: 2025 QinHan # SPDX-License-Identifier: MPL-2.0 import asyncio +import re from collections import defaultdict from copy import deepcopy from dataclasses import dataclass @@ -17,6 +18,7 @@ from docx.section import _Header, _Footer from docx.text.paragraph import Paragraph from docx.text.run import Run from docx.table import _Cell, Table +from lxml import etree from docutranslate.agents.segments_agent import SegmentsTranslateAgentConfig, SegmentsTranslateAgent from docutranslate.ir.document import Document @@ -29,7 +31,8 @@ from docutranslate.translator.ai_translator.base import AiTranslatorConfig, AiTr def is_image_run(run: Run) -> bool: """检查一个 Run 是否包含图片。""" xml = getattr(run.element, 'xml', '') - return ' bool: @@ -146,7 +149,7 @@ class DocxTranslator(AiTranslator): merged = [] group_start = 0 for i in range(1, len(runs)): - if self._run_format_key(runs[i]) != self._run_format_key(runs[group_start]): + if not self._can_merge_runs(runs[i - 1], runs[i], runs[group_start]): # Format boundary: finalize the current group if i - group_start > 1: # Merge: accumulate all text into first run, delete the rest @@ -165,6 +168,111 @@ class DocxTranslator(AiTranslator): return merged + def _can_merge_runs(self, previous_run: Run, current_run: Run, group_start_run: Run) -> bool: + """Only merge runs that are same-style and adjacent in the same XML parent.""" + if self._run_format_key(current_run) != self._run_format_key(group_start_run): + return False + + previous_parent = previous_run.element.getparent() + current_parent = current_run.element.getparent() + if previous_parent is None or previous_parent is not current_parent: + return False + + try: + return current_parent.index(current_run.element) == current_parent.index(previous_run.element) + 1 + except ValueError: + return False + + @staticmethod + def _is_run_in_paragraph(run_element, paragraph_element) -> bool: + """Return True when the run belongs to this paragraph, not a nested textbox paragraph.""" + parent = run_element.getparent() + while parent is not None: + if parent.tag == qn('w:p'): + return parent is paragraph_element + parent = parent.getparent() + return False + + @staticmethod + def _has_ancestor(element, tags) -> bool: + parent = element.getparent() + while parent is not None: + if parent.tag in tags: + return True + parent = parent.getparent() + return False + + @staticmethod + def _is_deleted_revision_run(run: Run) -> bool: + """Skip tracked-deletion text; inserted revision text is still visible/translatable.""" + return ( + DocxTranslator._has_ancestor(run.element, {qn('w:del'), qn('w:moveFrom')}) + or run.element.find(qn('w:delText')) is not None + ) + + def _iter_paragraph_runs(self, para: Paragraph): + """Yield direct runs and runs nested in same-paragraph wrappers such as hyperlinks.""" + paragraph_element = para._p + for run_element in paragraph_element.iter(qn('w:r')): + if self._is_run_in_paragraph(run_element, paragraph_element): + yield Run(run_element, para) + + @staticmethod + def _is_toc_paragraph(para: Paragraph) -> bool: + style_name = para.style.name if para.style is not None else "" + style_id = "" + p_pr = para._p.pPr + if p_pr is not None and p_pr.pStyle is not None: + style_id = p_pr.pStyle.val or "" + return ( + style_name.replace(" ", "").upper().startswith("TOC") + or style_id.replace(" ", "").upper().startswith("TOC") + ) + + @staticmethod + def _has_tab(run: Run) -> bool: + return run.element.find(qn('w:tab')) is not None + + @staticmethod + def _field_char_type(run: Run) -> Optional[str]: + field_char = run.element.find(qn('w:fldChar')) + return field_char.get(qn('w:fldCharType')) if field_char is not None else None + + @staticmethod + def _is_page_field_start(runs: List[Run], start_index: int) -> bool: + instruction_parts = [] + for run in runs[start_index + 1:]: + field_type = DocxTranslator._field_char_type(run) + if field_type in ("begin", "end", "separate"): + break + instr_text = run.element.find(qn('w:instrText')) + if instr_text is not None and instr_text.text: + instruction_parts.append(instr_text.text) + instruction = "".join(instruction_parts).upper() + return "PAGE" in instruction or "NUMPAGES" in instruction + + def _append_text_runs_element( + self, + runs: List[Run], + para: Paragraph, + top_level_para: Paragraph, + elements: List[Dict[str, Any]], + texts: List[str], + ) -> None: + if not runs: + return + runs = self._merge_adjacent_runs(runs) + full_text = "".join(r.text for r in runs) + if not full_text.strip(): + return + elements.append({ + "type": "text_runs", + "runs": list(runs), + "paragraph": para, + "top_level_paragraph": top_level_para + }) + texts.append(full_text) + def _process_paragraph(self, para: Paragraph, elements: List[Dict[str, Any]], texts: List[str], top_level_para: Paragraph = None): """ @@ -174,25 +282,49 @@ class DocxTranslator(AiTranslator): top_level_para = para text_runs = [] - for run in para.runs: - if is_image_run(run) or is_instr_text_run(run): + is_toc_paragraph = self._is_toc_paragraph(para) + reached_toc_page_number = False + paragraph_runs = list(self._iter_paragraph_runs(para)) + skip_page_field = False + in_page_field_result = False + + for run_index, run in enumerate(paragraph_runs): + field_type = self._field_char_type(run) + if field_type == "begin": + skip_page_field = self._is_page_field_start(paragraph_runs, run_index) + in_page_field_result = False continue - if not run.text.strip(): + if field_type == "separate": + if skip_page_field: + in_page_field_result = True + continue + if field_type == "end": + if skip_page_field or in_page_field_result: + skip_page_field = False + in_page_field_result = False + continue + if skip_page_field or in_page_field_result: + continue + + if self._is_deleted_revision_run(run) or is_image_run(run) or is_instr_text_run(run): + continue + + run_text = run.text + if is_toc_paragraph and self._has_tab(run): + reached_toc_page_number = True + if not run_text.replace("\t", "").strip(): + continue + if reached_toc_page_number and run_text.strip().isdigit(): + continue + if not run_text: continue text_runs.append(run) - if text_runs: - # Merge adjacent runs with identical formatting to reduce fragmentation - text_runs = self._merge_adjacent_runs(text_runs) - full_text = "".join(r.text for r in text_runs) - if full_text.strip(): - elements.append({ - "type": "text_runs", - "runs": list(text_runs), - "paragraph": para, - "top_level_paragraph": top_level_para - }) - texts.append(full_text) + self._append_text_runs_element(text_runs, para, top_level_para, elements, texts) + + for textbox_paragraph_element in para._p.xpath('.//w:txbxContent//w:p'): + nested_paragraph = Paragraph(textbox_paragraph_element, para._parent) + self._process_paragraph(nested_paragraph, elements, texts, top_level_para=top_level_para) # ---------------------- 修改结束 ---------------------- @@ -203,8 +335,13 @@ class DocxTranslator(AiTranslator): self._process_paragraph(Paragraph(child_element, container), elements, texts) elif child_element.tag.endswith('tbl'): table = Table(child_element, container) + seen_cells = set() for row in table.rows: for cell in row.cells: + cell_id = id(cell._tc) + if cell_id in seen_cells: + continue + seen_cells.add(cell_id) self._traverse_container(cell, elements, texts) elif child_element.tag.endswith('sdt'): sdt_content = child_element.find(qn('w:sdtContent')) @@ -228,7 +365,7 @@ class DocxTranslator(AiTranslator): self.logger.warning(f"跳过未知类型的容器: {type(container)}") return - if parent_element is not None and parent_element.tag in [qn('w:footnotes'), qn('w:endnotes')]: + if parent_element is not None and parent_element.tag in [qn('w:footnotes'), qn('w:endnotes'), qn('w:comments')]: for note_element in parent_element: self._process_body_elements(note_element, container, elements, texts) elif parent_element is not None: @@ -239,23 +376,70 @@ class DocxTranslator(AiTranslator): doc = docx.Document(BytesIO(content)) elements, texts = [], [] - self._traverse_container(doc, elements, texts) + seen_container_elements = set() + + def traverse_once(container): + if container is None: + return + if isinstance(container, (DocumentObject, Part)): + parent_element = container.element.body if hasattr(container.element, 'body') else container.element + elif isinstance(container, (_Cell, _Header, _Footer)): + parent_element = container._element + else: + parent_element = None + if parent_element is not None: + element_id = id(parent_element) + if element_id in seen_container_elements: + return + seen_container_elements.add(element_id) + self._traverse_container(container, elements, texts) + + traverse_once(doc) for section in doc.sections: - self._traverse_container(section.header, elements, texts) - self._traverse_container(section.first_page_header, elements, texts) - self._traverse_container(section.even_page_header, elements, texts) - self._traverse_container(section.footer, elements, texts) - self._traverse_container(section.first_page_footer, elements, texts) - self._traverse_container(section.even_page_footer, elements, texts) + traverse_once(section.header) + traverse_once(section.first_page_header) + traverse_once(section.even_page_header) + traverse_once(section.footer) + traverse_once(section.first_page_footer) + traverse_once(section.even_page_footer) if hasattr(doc.part, 'footnotes_part') and doc.part.footnotes_part is not None: - self._traverse_container(doc.part.footnotes_part, elements, texts) + traverse_once(doc.part.footnotes_part) if hasattr(doc.part, 'endnotes_part') and doc.part.endnotes_part is not None: - self._traverse_container(doc.part.endnotes_part, elements, texts) + traverse_once(doc.part.endnotes_part) + comments_part = getattr(doc.part, '_comments_part', None) + if comments_part is not None: + traverse_once(comments_part) return doc, elements, texts + @staticmethod + def _looks_translatable_text(text: str) -> bool: + return len(re.findall(r"\b[A-Za-z]{4,}\b", text)) >= 2 + + def _log_translation_quality(self, originals: List[str], translated: List[str]) -> None: + if not originals or not translated: + return + + comparable_count = min(len(originals), len(translated)) + unchanged = 0 + empty = 0 + for original, result in zip(originals[:comparable_count], translated[:comparable_count]): + if not str(result).strip() and str(original).strip(): + empty += 1 + if original == result and self._looks_translatable_text(original): + unchanged += 1 + + if empty: + self.logger.warning(f"DOCX translation audit: {empty}/{comparable_count} non-empty segments returned empty translations.") + if unchanged: + ratio = unchanged / comparable_count + level = self.logger.warning if ratio >= 0.05 else self.logger.info + level( + f"DOCX translation audit: {unchanged}/{comparable_count} translatable-looking segments are unchanged." + ) + def _apply_translation(self, element_info: Dict[str, Any], final_text: str): if element_info["type"] == "text_runs": runs = element_info["runs"] @@ -315,6 +499,55 @@ class DocxTranslator(AiTranslator): self.logger.debug(f"尝试删除一个不存在的run元素。这通常是安全的。") # ---------- FIX START: 新增用于清理副本段落的辅助方法 ---------- + @staticmethod + def _section_signature_without_break_type(sect_pr) -> bytes: + """Return a section-property signature that ignores only the break type.""" + sect_copy = deepcopy(sect_pr) + for type_element in sect_copy.findall(qn('w:type')): + sect_copy.remove(type_element) + return etree.tostring(sect_copy, with_tail=False) + + def _normalize_redundant_next_page_section_breaks(self, doc: DocumentObject) -> None: + """ + Convert redundant next-page section breaks to continuous breaks. + + Some source DOCX files contain empty-paragraph section breaks after many + subsections even when the neighboring sections have identical layout. + Once translated text expands, those breaks create large blank areas. + """ + normalized_count = 0 + previous_signature = None + + for sect_pr in doc._element.xpath('.//w:sectPr'): + parent = sect_pr.getparent() + paragraph_element = parent.getparent() if parent is not None and parent.tag == qn('w:pPr') else None + if paragraph_element is None or paragraph_element.tag != qn('w:p'): + continue + + signature = self._section_signature_without_break_type(sect_pr) + type_element = sect_pr.find(qn('w:type')) + break_type = type_element.get(qn('w:val')) if type_element is not None else 'nextPage' + paragraph_text = ''.join(paragraph_element.xpath('.//w:t/text()')).strip() + + if ( + previous_signature is not None + and signature == previous_signature + and break_type == 'nextPage' + and not paragraph_text + ): + if type_element is None: + type_element = OxmlElement('w:type') + sect_pr.insert(0, type_element) + type_element.set(qn('w:val'), 'continuous') + normalized_count += 1 + + previous_signature = signature + + if normalized_count: + self.logger.info( + f"DOCX layout cleanup: converted {normalized_count} redundant next-page section breaks to continuous." + ) + def _prune_unwanted_elements_from_copy(self, p_element: OxmlElement): """ 从复制的段落元素中移除包含图片或页码字段的 Run。 @@ -328,7 +561,9 @@ class DocxTranslator(AiTranslator): run_element = runs[i] # 检查图片 - if run_element.find(qn('w:drawing')) is not None or run_element.find(qn('w:pict')) is not None: + has_drawing = run_element.find(qn('w:drawing')) is not None or run_element.find(qn('w:pict')) is not None + has_textbox = run_element.find('.//' + qn('w:txbxContent')) is not None + if has_drawing and not has_textbox: runs_to_remove.append(run_element) i += 1 continue @@ -381,6 +616,9 @@ class DocxTranslator(AiTranslator): def _after_translate(self, doc: DocumentObject, elements: List[Dict[str, Any]], translated: List[str], originals: List[str]) -> bytes: + if not self.skip_translate: + self._log_translation_quality(originals, translated) + if len(elements) != len(translated): self.logger.error( f"翻译数量不匹配！原文: {len(originals)}, 译文: {len(translated)}. 将只处理公共部分。") @@ -390,6 +628,8 @@ class DocxTranslator(AiTranslator): if self.insert_mode == "replace": for info, trans in zip(elements, translated): self._apply_translation(info, trans) + if not self.skip_translate: + self._normalize_redundant_next_page_section_breaks(doc) else: paragraph_segments = defaultdict(list) # [FIX] 按顶级段落对所有片段进行分组，以确保形状等嵌套内容与主段落一起处理 @@ -535,4 +775,4 @@ class DocxTranslator(AiTranslator): translated = await self.translate_agent.send_segments_async(originals, self.chunk_size) if self.translate_agent else originals document.content = await asyncio.to_thread(self._after_translate, doc, elements, translated, originals) - return self \ No newline at end of file + return self diff --git a/run.bat b/run.bat index f896351..be6b154 100644 --- a/run.bat +++ b/run.bat @@ -18,7 +18,7 @@ REM 如果传了参数，就用参数；否则默认用当前目录 if "%~1"=="" ( python docutranslate/cli.py -i --host 0.0.0.0 ) else ( - python docutranslate/cli.py -i "%~1" + python docutranslate/cli.py -i --host 0.0.0.0 %* ) set EXIT_CODE=%ERRORLEVEL% @@ -32,4 +32,4 @@ if not "%EXIT_CODE%"=="0" ( pause ) -exit /b %EXIT_CODE% \ No newline at end of file +exit /b %EXIT_CODE% diff --git a/tests/test_docx_translator.py b/tests/test_docx_translator.py new file mode 100644 index 0000000..27789f2 --- /dev/null +++ b/tests/test_docx_translator.py @@ -0,0 +1,437 @@ +from io import BytesIO +import unittest +from zipfile import ZipFile + +import docx +from docx.oxml import OxmlElement, parse_xml +from docx.oxml.ns import nsdecls, qn +from docx.shared import Inches +from lxml import etree + +from docutranslate.ir.document import Document +from docutranslate.translator.ai_translator.docx_translator import ( + DocxTranslator, + DocxTranslatorConfig, +) + + +NS = {"w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main"} + + +def _docx_bytes(document): + stream = BytesIO() + document.save(stream) + return stream.getvalue() + + +def _text_values(docx_bytes): + with ZipFile(BytesIO(docx_bytes)) as zf: + tree = etree.fromstring(zf.read("word/document.xml")) + return tree.xpath("//w:t/text()", namespaces=NS) + + +def _comments_text_values(docx_bytes): + with ZipFile(BytesIO(docx_bytes)) as zf: + if "word/comments.xml" not in zf.namelist(): + return [] + tree = etree.fromstring(zf.read("word/comments.xml")) + return tree.xpath("//w:t/text()", namespaces=NS) + + +def _header_text_values(docx_bytes): + values = [] + with ZipFile(BytesIO(docx_bytes)) as zf: + for name in zf.namelist(): + if name.startswith("word/header") and name.endswith(".xml"): + tree = etree.fromstring(zf.read(name)) + values.extend(tree.xpath("//w:t/text()", namespaces=NS)) + return values + + +def _paragraph_and_table_counts(docx_bytes): + with ZipFile(BytesIO(docx_bytes)) as zf: + tree = etree.fromstring(zf.read("word/document.xml")) + return ( + len(tree.xpath("//w:p", namespaces=NS)), + len(tree.xpath("//w:tbl", namespaces=NS)), + ) + + +def _section_type_counts(docx_bytes): + with ZipFile(BytesIO(docx_bytes)) as zf: + tree = etree.fromstring(zf.read("word/document.xml")) + counts = {} + for sect_pr in tree.xpath("//w:sectPr", namespaces=NS): + value = sect_pr.xpath("./w:type/@w:val", namespaces=NS) + key = value[0] if value else "nextPage(default)" + counts[key] = counts.get(key, 0) + 1 + return counts + + +def _add_toc_paragraph(document): + paragraph = document.add_paragraph() + p_style = OxmlElement("w:pStyle") + p_style.set(qn("w:val"), "TOC1") + paragraph._p.get_or_add_pPr().append(p_style) + + hyperlink = OxmlElement("w:hyperlink") + hyperlink.set(qn("w:anchor"), "_TocFixture") + + title_run = OxmlElement("w:r") + title_text = OxmlElement("w:t") + title_text.text = "Purpose" + title_run.append(title_text) + + tab_run = OxmlElement("w:r") + tab_run.append(OxmlElement("w:tab")) + + page_run = OxmlElement("w:r") + page_text = OxmlElement("w:t") + page_text.text = "3" + page_run.append(page_text) + + hyperlink.extend([title_run, tab_run, page_run]) + paragraph._p.append(hyperlink) + + +def _add_internal_hyperlink_paragraph(document): + paragraph = document.add_paragraph() + paragraph.add_run("Read ") + hyperlink = OxmlElement("w:hyperlink") + hyperlink.set(qn("w:anchor"), "_Manual") + link_run = OxmlElement("w:r") + link_text = OxmlElement("w:t") + link_text.text = "manual" + link_run.append(link_text) + hyperlink.append(link_run) + paragraph._p.append(hyperlink) + paragraph.add_run(" now") + + +def _add_textbox_paragraph(document): + paragraph = document.add_paragraph() + paragraph.add_run("Diagram: ") + textbox_run = parse_xml( + f""" + + + + + + + + + + Flow label + + + + + + + + + + """ + ) + paragraph._p.append(textbox_run) + + +def _add_inline_sdt(paragraph, text): + sdt = OxmlElement("w:sdt") + sdt_content = OxmlElement("w:sdtContent") + run = OxmlElement("w:r") + text_element = OxmlElement("w:t") + text_element.text = text + run.append(text_element) + sdt_content.append(run) + sdt.append(sdt_content) + paragraph._p.append(sdt) + + +def _add_block_sdt(document, text): + sdt = OxmlElement("w:sdt") + sdt_content = OxmlElement("w:sdtContent") + paragraph = OxmlElement("w:p") + run = OxmlElement("w:r") + text_element = OxmlElement("w:t") + text_element.text = text + run.append(text_element) + paragraph.append(run) + sdt_content.append(paragraph) + sdt.append(sdt_content) + document._body._element.append(sdt) + + +def _add_page_field_paragraph(document): + paragraph = document.add_paragraph("Page ") + for fld_type in ("begin", "separate", "end"): + run = OxmlElement("w:r") + if fld_type == "begin": + fld = OxmlElement("w:fldChar") + fld.set(qn("w:fldCharType"), "begin") + run.append(fld) + instr_run = OxmlElement("w:r") + instr = OxmlElement("w:instrText") + instr.text = " PAGE " + instr_run.append(instr) + paragraph._p.append(run) + paragraph._p.append(instr_run) + continue + if fld_type == "separate": + fld = OxmlElement("w:fldChar") + fld.set(qn("w:fldCharType"), "separate") + run.append(fld) + cached_run = OxmlElement("w:r") + cached_text = OxmlElement("w:t") + cached_text.text = "1" + cached_run.append(cached_text) + paragraph._p.append(run) + paragraph._p.append(cached_run) + continue + fld = OxmlElement("w:fldChar") + fld.set(qn("w:fldCharType"), "end") + run.append(fld) + paragraph._p.append(run) + + +def _add_tracked_changes_paragraph(document): + paragraph = document.add_paragraph("Revision ") + + deleted = OxmlElement("w:del") + deleted.set(qn("w:id"), "1") + deleted_run = OxmlElement("w:r") + deleted_text = OxmlElement("w:delText") + deleted_text.text = "old text" + deleted_run.append(deleted_text) + deleted.append(deleted_run) + paragraph._p.append(deleted) + + inserted = OxmlElement("w:ins") + inserted.set(qn("w:id"), "2") + inserted_run = OxmlElement("w:r") + inserted_text = OxmlElement("w:t") + inserted_text.text = "new text" + inserted_run.append(inserted_text) + inserted.append(inserted_run) + paragraph._p.append(inserted) + + +def _add_redundant_section_break(document): + document.add_paragraph("Section One") + for _ in range(2): + section = document.add_section() + section.start_type = docx.enum.section.WD_SECTION.NEW_PAGE + section.top_margin = Inches(1) + section.bottom_margin = Inches(1) + section.left_margin = Inches(1) + section.right_margin = Inches(1) + document.add_paragraph("Section Two") + + +def _build_fixture_docx(): + document = docx.Document() + for section in document.sections: + section.top_margin = Inches(1) + section.bottom_margin = Inches(1) + section.left_margin = Inches(1) + section.right_margin = Inches(1) + document.add_heading("Fixture", level=1) + document.add_paragraph("Normal paragraph") + _add_toc_paragraph(document) + _add_internal_hyperlink_paragraph(document) + _add_textbox_paragraph(document) + inline_sdt_paragraph = document.add_paragraph("Controlled ") + _add_inline_sdt(inline_sdt_paragraph, "Approval Required") + _add_block_sdt(document, "Block Controlled Text") + _add_page_field_paragraph(document) + _add_tracked_changes_paragraph(document) + + table = document.add_table(rows=2, cols=2) + merged = table.cell(0, 0).merge(table.cell(0, 1)) + merged.text = "Merged Cell" + table.cell(1, 0).text = "Left Cell" + table.cell(1, 1).text = "Right Cell" + nested_table = table.cell(1, 0).add_table(rows=1, cols=1) + nested_table.cell(0, 0).text = "Nested Cell" + + comment_anchor = document.add_paragraph().add_run("Comment anchor") + document.add_comment(comment_anchor, text="Review this note", author="QA") + + _add_redundant_section_break(document) + for sect_pr in document._element.xpath(".//w:sectPr"): + for child in list(sect_pr): + if child.tag in {qn("w:headerReference"), qn("w:footerReference")}: + sect_pr.remove(child) + return _docx_bytes(document) + + +class DocxTranslatorTest(unittest.TestCase): + def _translator(self, skip_translate=True): + return DocxTranslator(DocxTranslatorConfig(skip_translate=skip_translate)) + + def test_pre_translate_extracts_docx_boundary_text(self): + translator = self._translator() + _, _, originals = translator._pre_translate(Document(content=_build_fixture_docx(), suffix=".docx")) + + self.assertIn("Purpose", originals) + self.assertIn("Flow label", originals) + self.assertIn("Controlled Approval Required", originals) + self.assertIn("Block Controlled Text", originals) + self.assertIn("Revision new text", originals) + self.assertNotIn("Revision old textnew text", originals) + self.assertTrue(all("old text" not in original for original in originals)) + self.assertIn("Merged Cell", originals) + self.assertIn("Nested Cell", originals) + self.assertIn("Review this note", originals) + self.assertNotIn("Purpose3", originals) + self.assertNotIn("PAGE", originals) + self.assertIn("Page ", originals) + self.assertNotIn("Page 1", originals) + + def test_replace_mode_writes_boundary_text_without_structural_loss(self): + input_bytes = _build_fixture_docx() + translator = self._translator() + doc, elements, originals = translator._pre_translate(Document(content=input_bytes, suffix=".docx")) + translations = { + "Purpose": "Tujuan", + "Flow label": "Label alur", + "Controlled Approval Required": "Persetujuan diperlukan", + "Block Controlled Text": "Teks kontrol blok", + "Revision new text": "Revisi teks baru", + "Merged Cell": "Sel gabungan", + "Nested Cell": "Sel bersarang", + "Review this note": "Tinjau catatan ini", + } + translated = [translations.get(text, text) for text in originals] + + output_bytes = translator._after_translate(doc, elements, translated, originals) + + body_text = _text_values(output_bytes) + comment_text = _comments_text_values(output_bytes) + self.assertIn("Tujuan", body_text) + self.assertIn("3", body_text) + self.assertIn("Label alur", body_text) + self.assertIn("Persetujuan diperlukan", "".join(body_text)) + self.assertIn("Teks kontrol blok", body_text) + self.assertIn("Revisi teks baru", "".join(body_text)) + self.assertIn("Sel gabungan", body_text) + self.assertIn("Sel bersarang", body_text) + self.assertIn("Tinjau catatan ini", comment_text) + self.assertEqual(_paragraph_and_table_counts(input_bytes), _paragraph_and_table_counts(output_bytes)) + + def test_append_mode_keeps_textbox_mapping(self): + input_bytes = _build_fixture_docx() + translator = DocxTranslator(DocxTranslatorConfig(skip_translate=True, insert_mode="append")) + doc, elements, originals = translator._pre_translate(Document(content=input_bytes, suffix=".docx")) + translated = ["Label alur" if text == "Flow label" else text for text in originals] + + output_bytes = translator._after_translate(doc, elements, translated, originals) + + body_text = _text_values(output_bytes) + self.assertIn("Flow label", body_text) + self.assertIn("Label alur", body_text) + + def test_real_translation_mode_normalizes_redundant_section_breaks(self): + input_bytes = _build_fixture_docx() + translator = DocxTranslator( + DocxTranslatorConfig( + skip_translate=False, + api_key="dummy", + base_url="http://127.0.0.1", + model_id="dummy", + ) + ) + doc, elements, originals = translator._pre_translate(Document(content=input_bytes, suffix=".docx")) + for sect_pr in doc._element.xpath(".//w:sectPr"): + for child in list(sect_pr): + if child.tag in {qn("w:headerReference"), qn("w:footerReference")}: + sect_pr.remove(child) + + output_bytes = translator._after_translate(doc, elements, originals, originals) + + source_counts = _section_type_counts(input_bytes) + output_counts = _section_type_counts(output_bytes) + self.assertGreater(source_counts.get("nextPage(default)", 0), output_counts.get("nextPage(default)", 0)) + self.assertGreater(output_counts.get("continuous", 0), source_counts.get("continuous", 0)) + + def test_hyperlink_runs_are_not_merged_across_parent_boundary(self): + input_bytes = _build_fixture_docx() + translator = self._translator() + doc, elements, originals = translator._pre_translate(Document(content=input_bytes, suffix=".docx")) + translated = ["Baca manual sekarang" if text == "Read manual now" else text for text in originals] + + output_bytes = translator._after_translate(doc, elements, translated, originals) + + with ZipFile(BytesIO(output_bytes)) as zf: + tree = etree.fromstring(zf.read("word/document.xml")) + self.assertEqual(len(tree.xpath("//w:hyperlink", namespaces=NS)), 2) + + def test_shared_headers_are_extracted_once_across_sections(self): + document = docx.Document() + document.sections[0].header.paragraphs[0].text = "Shared Header" + document.add_paragraph("Body One") + document.add_section() + document.add_paragraph("Body Two") + input_bytes = _docx_bytes(document) + + translator = self._translator() + _, _, originals = translator._pre_translate(Document(content=input_bytes, suffix=".docx")) + + self.assertEqual(originals.count("Shared Header"), 1) + + def test_shared_header_is_written_once_and_preserved(self): + document = docx.Document() + document.sections[0].header.paragraphs[0].text = "Shared Header" + document.add_paragraph("Body One") + document.add_section() + document.add_paragraph("Body Two") + input_bytes = _docx_bytes(document) + + translator = self._translator() + doc, elements, originals = translator._pre_translate(Document(content=input_bytes, suffix=".docx")) + translated = ["Header Bersama" if text == "Shared Header" else text for text in originals] + output_bytes = translator._after_translate(doc, elements, translated, originals) + + self.assertEqual(_header_text_values(output_bytes).count("Header Bersama"), 1) + + def test_translate_method_end_to_end_with_fake_agent(self): + class FakeAgent: + def send_segments(self, segments, chunk_size): + translations = { + "Purpose": "Tujuan", + "Flow label": "Label alur", + "Controlled Approval Required": "Persetujuan diperlukan", + "Block Controlled Text": "Teks kontrol blok", + "Revision new text": "Revisi teks baru", + "Nested Cell": "Sel bersarang", + "Review this note": "Tinjau catatan ini", + } + return [translations.get(segment, segment) for segment in segments] + + document = Document(content=_build_fixture_docx(), suffix=".docx") + translator = DocxTranslator( + DocxTranslatorConfig( + skip_translate=False, + api_key="dummy", + base_url="http://127.0.0.1", + model_id="dummy", + ) + ) + translator.translate_agent = FakeAgent() + + translator.translate(document) + + body_text = _text_values(document.content) + comments_text = _comments_text_values(document.content) + self.assertIn("Tujuan", body_text) + self.assertIn("Label alur", body_text) + self.assertIn("Persetujuan diperlukan", "".join(body_text)) + self.assertIn("Teks kontrol blok", body_text) + self.assertIn("Revisi teks baru", "".join(body_text)) + self.assertIn("Sel bersarang", body_text) + self.assertIn("Tinjau catatan ini", comments_text) + + +if __name__ == "__main__": + unittest.main()