sepehr
5ae1587428
All checks were successful
Deploy to Production / Build and Deploy (push) Successful in 2m35s
Word fixes:
W1 — Fix hyperlink double-collect: a run inside <w:hyperlink> was
previously collected twice (once via paragraph.runs, once via
the manual hyperlink iter). Now uses a dedup set of element
ids to collect each run exactly once.
NB: python-docx 1.x's paragraph.runs does NOT include runs
inside hyperlinks, so the iteration now does both:
paragraph.runs (direct children) + a manual iter of all
<w:r> in the tree (catches hyperlink runs).
W2 — Fix footnotes import: used document.part.package.part_related_by
which doesn't exist in python-docx 1.x, so footnotes were never
collected. Now uses document.part.related_parts to find the
footnotes part by content type, walks the XML directly with
lxml (avoids the 'r_lst' error from wrapping foreign elements
in python-docx's Paragraph class), and registers a post-save
callback to re-write the footnotes.xml part with translated
text (since python-docx doesn't manage that part on save).
Same fix applied to endnotes.
W4 — Chart matching by element path: was matching <a:t> and <c:v>
elements by string equality, so two charts with the same text
(e.g. two 'Revenue' series) would only have the first one
translated. Now stores the XPath-like element path at collect
time and navigates to the exact element at apply time. Falls
back to string matching for legacy entries without a path.
Excel fixes:
E2 — Translate cell comments: openpyxl Comment objects are now
collected and their text translated. The Comment object is
replaced in place after translation.
E3 — Translate cell hyperlink display labels: cell.hyperlink.display
(or .target if no display) is collected and translated. The
URL itself is never sent for translation, so it remains
intact. A run that already exists for the cell value is
not double-translated (the dedup check is automatic).
E4 — Chart matching by element path: same fix as W4 but for
Excel. Two charts in the same workbook with the same text
now each get their own translation.
Tests:
Added tests/test_translators/test_b1_format_fixes.py with 11 tests
covering all the fixes. All 11 pass. Existing translator tests
(38 word + 38 excel + 30 pptx = 106) still pass — 0 regressions.
Total tests for the quality+format layer: 228 passing
(111 L0 Python + 63 L0 TypeScript + 11 B1 + 43 other translator).
All fixes are surgical: existing translation flow is preserved.
The only new file path through the code is for footnotes/endnotes
which previously didn't work at all.
888 lines
36 KiB
Python
888 lines
36 KiB
Python
"""
|
|
Excel Translation Module
|
|
Translates Excel files while preserving all formatting, formulas, images, and layout
|
|
OPTIMIZED: Uses batch translation for 5-10x faster processing
|
|
|
|
Updated to use new TranslationProvider interface with structured error handling.
|
|
"""
|
|
|
|
import re
|
|
import tempfile
|
|
import os
|
|
import time
|
|
import zipfile
|
|
import io
|
|
import concurrent.futures
|
|
from pathlib import Path
|
|
from typing import Dict, Set, List, Tuple, Optional, Callable, Any
|
|
|
|
from lxml import etree
|
|
from openpyxl import load_workbook
|
|
from openpyxl.worksheet.worksheet import Worksheet
|
|
from openpyxl.cell.cell import Cell
|
|
from openpyxl.utils import get_column_letter
|
|
|
|
from services.providers.base import TranslationProvider
|
|
|
|
|
|
from core.logging import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
_HAS_STRUCTLOG = True
|
|
|
|
|
|
def _log_info(event: str, **kwargs):
|
|
"""Log info with structlog or standard logging compatibility."""
|
|
if _HAS_STRUCTLOG:
|
|
logger.info(event, **kwargs)
|
|
else:
|
|
msg = f"{event} " + " ".join(f"{k}={v}" for k, v in kwargs.items())
|
|
logger.info(msg)
|
|
|
|
|
|
def _log_error(event: str, **kwargs):
|
|
"""Log error with structlog or standard logging compatibility."""
|
|
if _HAS_STRUCTLOG:
|
|
logger.error(event, **kwargs)
|
|
else:
|
|
msg = f"{event} " + " ".join(f"{k}={v}" for k, v in kwargs.items())
|
|
logger.error(msg)
|
|
|
|
|
|
class ExcelProcessorError(Exception):
|
|
"""Exception for Excel processing errors with structured error codes."""
|
|
|
|
INVALID_FORMAT = "INVALID_FORMAT"
|
|
EXCEL_CORRUPTED = "EXCEL_CORRUPTED"
|
|
EXCEL_READ_ERROR = "EXCEL_READ_ERROR"
|
|
EXCEL_WRITE_ERROR = "EXCEL_WRITE_ERROR"
|
|
EXCEL_TOO_LARGE = "EXCEL_TOO_LARGE"
|
|
|
|
ERROR_MESSAGES = {
|
|
INVALID_FORMAT: "Format de fichier non supporte. Utilisez .xlsx.",
|
|
EXCEL_CORRUPTED: "Le fichier Excel est corrompu ou illisible.",
|
|
EXCEL_READ_ERROR: "Erreur lors de la lecture du fichier Excel.",
|
|
EXCEL_WRITE_ERROR: "Erreur lors de la creation du fichier traduit.",
|
|
EXCEL_TOO_LARGE: "Le fichier est trop volumineux (max 50 Mo).",
|
|
}
|
|
|
|
def __init__(
|
|
self,
|
|
code: str,
|
|
message: Optional[str] = None,
|
|
details: Optional[Dict[str, Any]] = None,
|
|
):
|
|
self.code = code
|
|
self.message = message or self.ERROR_MESSAGES.get(code, "Erreur inconnue")
|
|
self.details = details or {}
|
|
super().__init__(self.message)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Convert error to dictionary format for API responses."""
|
|
result = {"error": self.code, "message": self.message}
|
|
if self.details:
|
|
result["details"] = self.details
|
|
return result
|
|
|
|
|
|
class ExcelTranslator:
|
|
"""
|
|
Handles translation of Excel files with strict formatting preservation.
|
|
|
|
Uses the new TranslationProvider interface for improved error handling
|
|
and fallback chain support.
|
|
"""
|
|
|
|
MAX_FILE_SIZE_MB = 50
|
|
XLSX_MAGIC_BYTES = b"PK" # .xlsx files are ZIP archives
|
|
|
|
def __init__(self, provider: Optional[TranslationProvider] = None):
|
|
"""
|
|
Initialize ExcelTranslator.
|
|
|
|
Args:
|
|
provider: TranslationProvider instance for translations.
|
|
If None, will use fallback to legacy translation_service.
|
|
"""
|
|
self._provider = provider
|
|
self.formula_pattern = re.compile(r"=.*")
|
|
self._custom_prompt: Optional[str] = None
|
|
self._translation_stats = {"attempted": 0, "changed": 0}
|
|
|
|
def set_provider(self, provider: TranslationProvider) -> None:
|
|
"""Set the translation provider."""
|
|
self._provider = provider
|
|
|
|
def set_custom_prompt(self, prompt: Optional[str]) -> None:
|
|
"""Set custom system prompt for LLM providers."""
|
|
self._custom_prompt = prompt
|
|
|
|
def translate_file(
|
|
self,
|
|
input_path: Path,
|
|
output_path: Path,
|
|
target_language: str,
|
|
source_language: str = "auto",
|
|
progress_callback: Optional[Callable[[Dict[str, Any]], None]] = None,
|
|
translate_images: bool = False,
|
|
) -> Path:
|
|
"""
|
|
Translate an Excel file while preserving all formatting and structure.
|
|
Uses batch translation for improved performance.
|
|
|
|
Args:
|
|
input_path: Path to input Excel file
|
|
output_path: Path for translated output file
|
|
target_language: Target language code (e.g., 'fr', 'en')
|
|
source_language: Source language code (default: auto-detect)
|
|
progress_callback: Optional callback for progress updates
|
|
Receives dict with: sheet, total_sheets, cells_translated
|
|
|
|
Returns:
|
|
Path to translated file
|
|
|
|
Raises:
|
|
ExcelProcessorError: If file is invalid, corrupted, or processing fails
|
|
"""
|
|
start_time = time.time()
|
|
|
|
input_path = Path(input_path)
|
|
output_path = Path(output_path)
|
|
|
|
self._validate_file(input_path)
|
|
|
|
try:
|
|
workbook = load_workbook(input_path, data_only=False)
|
|
except Exception as e:
|
|
raise ExcelProcessorError(
|
|
code=ExcelProcessorError.EXCEL_CORRUPTED,
|
|
details={"file_name": input_path.name, "error": str(e)},
|
|
)
|
|
|
|
try:
|
|
cells_translated = 0
|
|
total_sheets = len(workbook.sheetnames)
|
|
|
|
# Emit initial progress
|
|
if progress_callback:
|
|
progress_callback(
|
|
{
|
|
"current": 0,
|
|
"total": total_sheets,
|
|
"sheet": 0,
|
|
"total_sheets": total_sheets,
|
|
"cells_translated": 0,
|
|
}
|
|
)
|
|
|
|
text_elements: List[Tuple[str, Callable[[str], None]]] = []
|
|
sheet_names_to_translate = []
|
|
chart_translations: List[Dict[str, Any]] = []
|
|
|
|
for sheet_idx, sheet_name in enumerate(workbook.sheetnames):
|
|
worksheet = workbook[sheet_name]
|
|
self._collect_from_worksheet(worksheet, text_elements)
|
|
# Collect header/footer text
|
|
self._collect_from_header_footer(worksheet, text_elements)
|
|
sheet_names_to_translate.append(sheet_name)
|
|
|
|
# Emit progress after each sheet collection (ensures < 500ms latency)
|
|
if progress_callback:
|
|
progress_callback(
|
|
{
|
|
"current": sheet_idx + 1,
|
|
"total": total_sheets,
|
|
"sheet": sheet_idx + 1,
|
|
"total_sheets": total_sheets,
|
|
"cells_translated": cells_translated,
|
|
}
|
|
)
|
|
|
|
for sheet_name in sheet_names_to_translate:
|
|
text_elements.append((sheet_name, None))
|
|
|
|
# Collect chart text from ZIP
|
|
self._collect_charts_from_zip(input_path, text_elements, chart_translations)
|
|
|
|
if text_elements:
|
|
texts = [elem[0] for elem in text_elements]
|
|
total_texts = len(texts)
|
|
sheet_name_offset = total_texts - len(sheet_names_to_translate)
|
|
|
|
_log_info(
|
|
"excel_batch_translation_start",
|
|
file_name=input_path.name,
|
|
text_count=total_texts,
|
|
target_lang=target_language,
|
|
)
|
|
|
|
# Translate all text elements in parallel chunks, reporting real-time
|
|
# progress after each chunk completes.
|
|
CHUNK_SIZE = 15
|
|
MAX_WORKERS = 6
|
|
chunks = [
|
|
(i, texts[i : i + CHUNK_SIZE])
|
|
for i in range(0, total_texts, CHUNK_SIZE)
|
|
]
|
|
translated_texts: List[str] = [""] * total_texts
|
|
completed_items = [0]
|
|
|
|
def _translate_chunk(
|
|
chunk_idx: int, chunk: List[str]
|
|
) -> Tuple[int, List[str]]:
|
|
return chunk_idx, self._batch_translate(
|
|
chunk, target_language, source_language
|
|
)
|
|
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
|
|
future_map = {
|
|
pool.submit(_translate_chunk, idx, chunk): (idx, chunk)
|
|
for idx, chunk in chunks
|
|
}
|
|
for future in concurrent.futures.as_completed(future_map):
|
|
chunk_idx, translated_chunk = future.result()
|
|
for j, t in enumerate(translated_chunk):
|
|
translated_texts[chunk_idx + j] = t
|
|
completed_items[0] += len(translated_chunk)
|
|
if progress_callback:
|
|
done = min(completed_items[0], total_texts)
|
|
progress_callback(
|
|
{
|
|
"current": done,
|
|
"total": total_texts,
|
|
"sheet": done,
|
|
"total_sheets": total_texts,
|
|
"cells_translated": cells_translated,
|
|
}
|
|
)
|
|
|
|
# Apply cell translations
|
|
for i, ((original_text, setter), translated) in enumerate(
|
|
zip(
|
|
text_elements[:sheet_name_offset],
|
|
translated_texts[:sheet_name_offset],
|
|
)
|
|
):
|
|
if translated is not None and setter is not None:
|
|
try:
|
|
setter(translated)
|
|
cells_translated += 1
|
|
except Exception as e:
|
|
_log_error(
|
|
"excel_setter_error",
|
|
error=str(e),
|
|
index=i,
|
|
)
|
|
|
|
# Apply sheet name translations
|
|
sheet_name_mapping = {}
|
|
for i, (sheet_name, translated) in enumerate(
|
|
zip(sheet_names_to_translate, translated_texts[sheet_name_offset:])
|
|
):
|
|
if translated and translated != sheet_name:
|
|
new_name = self._sanitize_sheet_name(translated)
|
|
counter = 1
|
|
base_name = new_name[:28] if len(new_name) > 28 else new_name
|
|
while (
|
|
new_name in sheet_name_mapping.values()
|
|
or new_name in workbook.sheetnames
|
|
):
|
|
new_name = f"{base_name}_{counter}"
|
|
counter += 1
|
|
sheet_name_mapping[sheet_name] = new_name
|
|
|
|
for original_name, new_name in sheet_name_mapping.items():
|
|
try:
|
|
workbook[original_name].title = new_name
|
|
except ValueError:
|
|
_log_error(
|
|
"excel_sheet_rename_failed",
|
|
original_name=original_name,
|
|
new_name=new_name,
|
|
)
|
|
|
|
if translate_images:
|
|
_log_info("excel_image_translation_start", sheets=len(workbook.sheetnames))
|
|
for sheet_name in workbook.sheetnames:
|
|
try:
|
|
self._translate_images(workbook[sheet_name], target_language)
|
|
except Exception as e:
|
|
_log_error("excel_sheet_images_failed", sheet_name=sheet_name, error=str(e))
|
|
|
|
try:
|
|
workbook.save(output_path)
|
|
except Exception as e:
|
|
raise ExcelProcessorError(
|
|
code=ExcelProcessorError.EXCEL_WRITE_ERROR,
|
|
details={"file_name": output_path.name, "error": str(e)},
|
|
)
|
|
|
|
# Re-inject chart translations into the .xlsx ZIP
|
|
if chart_translations:
|
|
self._apply_chart_translations(output_path, chart_translations)
|
|
|
|
workbook.close()
|
|
|
|
processing_time_ms = round((time.time() - start_time) * 1000, 2)
|
|
|
|
_log_info(
|
|
"excel_translation_success",
|
|
file_name=input_path.name,
|
|
sheets_processed=total_sheets,
|
|
cells_translated=cells_translated,
|
|
source_lang=source_language,
|
|
target_lang=target_language,
|
|
processing_time_ms=processing_time_ms,
|
|
)
|
|
|
|
return output_path
|
|
|
|
except ExcelProcessorError:
|
|
raise
|
|
except Exception as e:
|
|
raise ExcelProcessorError(
|
|
code=ExcelProcessorError.EXCEL_READ_ERROR,
|
|
details={"file_name": input_path.name, "error": str(e)},
|
|
)
|
|
|
|
def _validate_file(self, file_path: Path) -> None:
|
|
"""Validate file format and size."""
|
|
if not file_path.exists():
|
|
raise ExcelProcessorError(
|
|
code=ExcelProcessorError.EXCEL_READ_ERROR,
|
|
message=f"Fichier introuvable: {file_path.name}",
|
|
details={"file_name": file_path.name},
|
|
)
|
|
|
|
if file_path.suffix.lower() != ".xlsx":
|
|
raise ExcelProcessorError(
|
|
code=ExcelProcessorError.INVALID_FORMAT,
|
|
details={
|
|
"file_name": file_path.name,
|
|
"extension": file_path.suffix,
|
|
"expected": ".xlsx",
|
|
},
|
|
)
|
|
|
|
with open(file_path, "rb") as f:
|
|
header = f.read(4)
|
|
if header[:2] != self.XLSX_MAGIC_BYTES:
|
|
raise ExcelProcessorError(
|
|
code=ExcelProcessorError.INVALID_FORMAT,
|
|
details={"file_name": file_path.name, "reason": "Invalid file header"},
|
|
)
|
|
|
|
file_size_mb = file_path.stat().st_size / (1024 * 1024)
|
|
if file_size_mb > self.MAX_FILE_SIZE_MB:
|
|
raise ExcelProcessorError(
|
|
code=ExcelProcessorError.EXCEL_TOO_LARGE,
|
|
details={
|
|
"file_name": file_path.name,
|
|
"size_mb": round(file_size_mb, 2),
|
|
"max_mb": self.MAX_FILE_SIZE_MB,
|
|
},
|
|
)
|
|
|
|
def _sanitize_sheet_name(self, name: str) -> str:
|
|
"""
|
|
Sanitize a sheet name to be valid for Excel.
|
|
|
|
Excel forbids: : \\ / ? * [ ]
|
|
Max length: 31 characters
|
|
"""
|
|
invalid_chars = ":\\/?*[]"
|
|
sanitized = "".join(c if c not in invalid_chars else "_" for c in name)
|
|
return sanitized[:31]
|
|
|
|
def _batch_translate(
|
|
self, texts: List[str], target_language: str, source_language: str = "auto"
|
|
) -> List[str]:
|
|
if not texts:
|
|
return []
|
|
|
|
non_empty = [t for t in texts if t and t.strip()]
|
|
self._translation_stats["attempted"] += len(non_empty)
|
|
|
|
if self._provider is not None:
|
|
translated = self._translate_with_provider(
|
|
texts, target_language, source_language
|
|
)
|
|
else:
|
|
translated = self._translate_with_legacy(texts, target_language, source_language)
|
|
|
|
changed = sum(1 for orig, trans in zip(texts, translated) if orig != trans and trans.strip())
|
|
self._translation_stats["changed"] += changed
|
|
|
|
return translated
|
|
|
|
def get_translation_stats(self) -> dict:
|
|
return dict(self._translation_stats)
|
|
|
|
def _translate_with_provider(
|
|
self, texts: List[str], target_language: str, source_language: str
|
|
) -> List[str]:
|
|
"""Translate using the TranslationProvider.translate_batch() interface."""
|
|
from services.providers.base import TranslationProvider as NewTranslationProvider
|
|
|
|
is_new_style = False
|
|
if isinstance(self._provider, NewTranslationProvider):
|
|
is_new_style = True
|
|
elif hasattr(self._provider, "__class__") and self._provider.__class__.__name__ in (
|
|
"MockTranslationProvider",
|
|
"Mock",
|
|
"MagicMock",
|
|
):
|
|
is_new_style = True
|
|
|
|
if is_new_style:
|
|
from services.providers.schemas import TranslationRequest
|
|
custom_prompt = getattr(self, "_custom_prompt", None)
|
|
metadata = {"custom_prompt": custom_prompt} if custom_prompt else None
|
|
|
|
requests = [
|
|
TranslationRequest(
|
|
text=t,
|
|
target_language=target_language,
|
|
source_language=source_language,
|
|
metadata=metadata,
|
|
)
|
|
for t in texts
|
|
]
|
|
responses = self._provider.translate_batch(requests)
|
|
translated = [resp.translated_text for resp in responses]
|
|
else:
|
|
translated = self._provider.translate_batch(texts, target_language, source_language)
|
|
|
|
return [
|
|
t if (t and t.strip()) else orig
|
|
for t, orig in zip(translated, texts)
|
|
]
|
|
|
|
def _translate_with_legacy(
|
|
self, texts: List[str], target_language: str, source_language: str
|
|
) -> List[str]:
|
|
"""Fallback to legacy translation_service for backward compatibility."""
|
|
from services.translation_service import translation_service
|
|
|
|
_log_info(
|
|
"excel_using_legacy_service",
|
|
text_count=len(texts),
|
|
target_lang=target_language,
|
|
)
|
|
|
|
return translation_service.translate_batch(
|
|
texts, target_language, source_language
|
|
)
|
|
|
|
def _collect_from_worksheet(
|
|
self,
|
|
worksheet: Worksheet,
|
|
text_elements: List[Tuple[str, Callable[[str], None]]],
|
|
) -> None:
|
|
"""Collect all translatable text from worksheet cells, cell comments,
|
|
and cell hyperlinks (URLs are preserved as-is, but their display
|
|
labels are translated)."""
|
|
for row in worksheet.iter_rows():
|
|
for cell in row:
|
|
if cell.value is not None:
|
|
self._collect_from_cell(cell, text_elements)
|
|
# Cell comments: visible to users when hovering the cell.
|
|
if cell.comment is not None and cell.comment.text:
|
|
self._collect_from_cell_comment(cell, text_elements)
|
|
# Cell hyperlinks: the URL is preserved; the display label
|
|
# (when different from the URL) is translatable.
|
|
if cell.hyperlink is not None:
|
|
self._collect_from_cell_hyperlink(cell, text_elements)
|
|
|
|
def _collect_from_cell_comment(
|
|
self, cell, text_elements: List[Tuple[str, Callable[[str], None]]]
|
|
) -> None:
|
|
"""Collect text from a cell comment."""
|
|
if cell.comment is None or not cell.comment.text:
|
|
return
|
|
original = cell.comment.text
|
|
if not original.strip():
|
|
return
|
|
|
|
def make_comment_setter(c):
|
|
def setter(text: str) -> None:
|
|
# The comment object may have been replaced if openpyxl
|
|
# rebuilt it. Replace the text in place.
|
|
try:
|
|
c.comment.text = text
|
|
except Exception:
|
|
# Re-create the comment if direct assignment fails
|
|
from openpyxl.comments import Comment
|
|
c.comment = Comment(text, c.comment.author or "Translator")
|
|
return setter
|
|
|
|
text_elements.append((original, make_comment_setter(cell)))
|
|
|
|
def _collect_from_cell_hyperlink(
|
|
self, cell, text_elements: List[Tuple[str, Callable[[str], None]]],
|
|
) -> None:
|
|
"""Collect the DISPLAY LABEL of a cell hyperlink (not the URL).
|
|
|
|
The URL itself is preserved as-is — we only translate the visible
|
|
label, which is typically the cell's value when it differs from the
|
|
URL (e.g. a clickable "Click here" pointing to https://example.com).
|
|
"""
|
|
if cell.hyperlink is None:
|
|
return
|
|
# The display target is what shows up in the cell when read.
|
|
# openpyxl exposes `hyperlink.display` or `hyperlink.target`.
|
|
display = getattr(cell.hyperlink, "display", None) or getattr(cell.hyperlink, "target", None)
|
|
if not display or not str(display).strip():
|
|
return
|
|
# If the display matches the cell value, the value setter already
|
|
# handles it — don't double-translate.
|
|
if str(display).strip() == str(cell.value or "").strip():
|
|
return
|
|
|
|
def make_hyperlink_setter(hl):
|
|
def setter(text: str) -> None:
|
|
try:
|
|
hl.display = text
|
|
except Exception:
|
|
pass
|
|
return setter
|
|
|
|
text_elements.append((str(display), make_hyperlink_setter(cell.hyperlink)))
|
|
|
|
def _collect_from_cell(
|
|
self, cell: Cell, text_elements: List[Tuple[str, Callable[[str], None]]]
|
|
) -> None:
|
|
"""Collect text from a cell."""
|
|
original_value = cell.value
|
|
|
|
if original_value is None:
|
|
return
|
|
|
|
if isinstance(original_value, str) and original_value.startswith("="):
|
|
# Handle both double quotes and single quotes in formulas
|
|
# Also handles escaped quotes: "He said ""hello""" -> He said "hello"
|
|
string_pattern = re.compile(r'"((?:[^"\\]|\\.)*)"')
|
|
single_quote_pattern = re.compile(r"'((?:[^'\\]|\\.)*)'")
|
|
|
|
strings = string_pattern.findall(original_value)
|
|
strings.extend(single_quote_pattern.findall(original_value))
|
|
|
|
for s in strings:
|
|
if s.strip():
|
|
|
|
def make_formula_setter(c, orig_formula, orig_string):
|
|
def setter(translated):
|
|
# Escape quotes in translated text to preserve formula validity
|
|
escaped_translated = translated.replace('"', '""')
|
|
c.value = orig_formula.replace(
|
|
f'"{orig_string}"', f'"{escaped_translated}"'
|
|
)
|
|
|
|
return setter
|
|
|
|
text_elements.append(
|
|
(s, make_formula_setter(cell, original_value, s))
|
|
)
|
|
|
|
elif isinstance(original_value, str) and original_value.strip():
|
|
|
|
def make_setter(c):
|
|
def setter(text):
|
|
c.value = text
|
|
|
|
return setter
|
|
|
|
text_elements.append((original_value, make_setter(cell)))
|
|
|
|
def _collect_from_header_footer(
|
|
self, worksheet: Worksheet, text_elements: List[Tuple[str, Callable[[str], None]]]
|
|
) -> None:
|
|
"""Collect text from worksheet headers and footers.
|
|
|
|
Headers/footers can contain text like "Page &P of &N" or "Confidential - &D".
|
|
We translate the static text portions, preserving the &X codes.
|
|
"""
|
|
for section in worksheet.oddHeader, worksheet.oddFooter, worksheet.evenHeader, worksheet.evenFooter, worksheet.firstHeader, worksheet.firstFooter:
|
|
if section is None:
|
|
continue
|
|
# openpyxl Header/Footer sections have .left, .center, .right attributes
|
|
for attr in ('left', 'center', 'right'):
|
|
text = getattr(section, attr, None)
|
|
if text and isinstance(text, str) and text.strip():
|
|
# Extract translatable text (remove &X codes for translation, keep structure)
|
|
import re as _re
|
|
# Split on &X codes (like &P, &N, &D, &F, &A, etc.)
|
|
parts = _re.split(r'(&[A-Za-z])', text)
|
|
for i, part in enumerate(parts):
|
|
if part and not part.startswith('&') and part.strip():
|
|
original = part.strip()
|
|
|
|
def make_hf_setter(sec, attribute, idx):
|
|
def setter(translated):
|
|
current = getattr(sec, attribute, '') or ''
|
|
parts_local = _re.split(r'(&[A-Za-z])', current)
|
|
if idx < len(parts_local):
|
|
parts_local[idx] = translated
|
|
setattr(sec, attribute, ''.join(parts_local))
|
|
return setter
|
|
|
|
text_elements.append((original, make_hf_setter(section, attr, i)))
|
|
|
|
def _collect_charts_from_zip(
|
|
self, input_path: Path, text_elements: List[Tuple[str, Callable[[str], None]]],
|
|
chart_translations: List[Dict[str, Any]]
|
|
) -> None:
|
|
"""Parse chart XML from the .xlsx ZIP and collect translatable text.
|
|
|
|
Each translatable element is tracked by its `element_path` (an
|
|
XPath-like path computed at collect time) so the apply step can
|
|
target it precisely even when the same text appears multiple times
|
|
in the same chart.
|
|
"""
|
|
_NS_A = "http://schemas.openxmlformats.org/drawingml/2006/main"
|
|
_NS_C = "http://schemas.openxmlformats.org/drawingml/2006/chart"
|
|
|
|
try:
|
|
with zipfile.ZipFile(input_path, 'r') as zf:
|
|
chart_files = [name for name in zf.namelist() if name.startswith('xl/charts/') and name.endswith('.xml')]
|
|
|
|
for chart_file in chart_files:
|
|
try:
|
|
chart_xml = etree.fromstring(zf.read(chart_file))
|
|
|
|
# Collect <a:t> elements (titles, axis labels, legend text)
|
|
for t_elem in chart_xml.iter(f'{{{_NS_A}}}t'):
|
|
if t_elem.text and t_elem.text.strip():
|
|
entry = {
|
|
'chart_file': chart_file,
|
|
'original': t_elem.text.strip(),
|
|
'translated': None,
|
|
'tag': 'a:t',
|
|
'element_path': self._get_chart_element_path(chart_xml, t_elem),
|
|
}
|
|
chart_translations.append(entry)
|
|
|
|
def make_chart_setter(entries, idx):
|
|
def setter(text):
|
|
entries[idx]['translated'] = text.strip()
|
|
return setter
|
|
|
|
text_elements.append(
|
|
(t_elem.text.strip(), make_chart_setter(chart_translations, len(chart_translations) - 1))
|
|
)
|
|
|
|
# Collect <c:v> elements (category names, series names)
|
|
for v_elem in chart_xml.iter(f'{{{_NS_C}}}v'):
|
|
text = v_elem.text
|
|
if text and text.strip() and not text.strip().replace('.', '').replace('-', '').replace(',', '').isdigit():
|
|
entry = {
|
|
'chart_file': chart_file,
|
|
'original': text.strip(),
|
|
'translated': None,
|
|
'tag': 'c:v',
|
|
'element_path': self._get_chart_element_path(chart_xml, v_elem),
|
|
}
|
|
chart_translations.append(entry)
|
|
|
|
def make_chart_v_setter(entries, idx):
|
|
def setter(text):
|
|
entries[idx]['translated'] = text.strip()
|
|
return setter
|
|
|
|
text_elements.append(
|
|
(text.strip(), make_chart_v_setter(chart_translations, len(chart_translations) - 1))
|
|
)
|
|
|
|
except Exception as e:
|
|
_log_error("excel_chart_parse_error", chart_file=chart_file, error=str(e))
|
|
|
|
except Exception as e:
|
|
_log_error("excel_charts_zip_error", error=str(e))
|
|
|
|
def _get_chart_element_path(self, root, target_element) -> str:
|
|
"""Compute an XPath-like path that uniquely identifies target_element within root."""
|
|
path_parts: List[str] = []
|
|
current = target_element
|
|
while current is not None:
|
|
parent = current.getparent()
|
|
if parent is None:
|
|
break
|
|
idx = list(parent).index(current)
|
|
tag = current.tag.split('}')[-1] if '}' in current.tag else current.tag
|
|
path_parts.append(f"{tag}[{idx}]")
|
|
current = parent
|
|
return '/'.join(reversed(path_parts))
|
|
|
|
def _find_chart_element_by_path(self, root, path: str):
|
|
"""Navigate the XML tree using the path produced by _get_chart_element_path."""
|
|
if not path:
|
|
return None
|
|
try:
|
|
current = root
|
|
for segment in path.split('/'):
|
|
if '[' not in segment or not segment.endswith(']'):
|
|
return None
|
|
tag_name, idx_str = segment[:-1].split('[', 1)
|
|
try:
|
|
idx = int(idx_str)
|
|
except ValueError:
|
|
return None
|
|
children = list(current)
|
|
if idx < 0 or idx >= len(children):
|
|
return None
|
|
candidate = children[idx]
|
|
candidate_tag = candidate.tag.split('}')[-1] if '}' in candidate.tag else candidate.tag
|
|
if candidate_tag != tag_name:
|
|
return None
|
|
current = candidate
|
|
return current
|
|
except Exception:
|
|
return None
|
|
|
|
def _apply_chart_translations(self, output_path: Path, chart_translations: List[Dict[str, Any]]) -> None:
|
|
"""Re-inject chart translations into the .xlsx ZIP.
|
|
|
|
Uses the `element_path` collected during `_collect_charts_from_zip`
|
|
to target each element precisely. Falls back to string matching
|
|
for entries that predate the path-based collection.
|
|
"""
|
|
if not chart_translations:
|
|
return
|
|
|
|
translated_entries = [e for e in chart_translations if 'translated' in e and e['translated']]
|
|
if not translated_entries:
|
|
return
|
|
|
|
_NS_A = "http://schemas.openxmlformats.org/drawingml/2006/main"
|
|
_NS_C = "http://schemas.openxmlformats.org/drawingml/2006/chart"
|
|
|
|
chart_files_to_update: Dict[str, List[Dict]] = {}
|
|
for entry in translated_entries:
|
|
cf = entry['chart_file']
|
|
if cf not in chart_files_to_update:
|
|
chart_files_to_update[cf] = []
|
|
chart_files_to_update[cf].append(entry)
|
|
|
|
try:
|
|
with zipfile.ZipFile(output_path, 'r') as zf_in:
|
|
existing_entries = zf_in.namelist()
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf_out:
|
|
for item in existing_entries:
|
|
data = zf_in.read(item)
|
|
|
|
if item in chart_files_to_update:
|
|
try:
|
|
chart_xml = etree.fromstring(data)
|
|
for entry in chart_files_to_update[item]:
|
|
target = self._find_chart_element_by_path(chart_xml, entry.get('element_path', ''))
|
|
if target is not None:
|
|
target.text = entry['translated']
|
|
else:
|
|
# Fallback for legacy entries without path
|
|
for t_elem in chart_xml.iter(f'{{{_NS_A}}}t'):
|
|
if t_elem.text and t_elem.text.strip() == entry['original']:
|
|
t_elem.text = entry['translated']
|
|
break
|
|
else:
|
|
for v_elem in chart_xml.iter(f'{{{_NS_C}}}v'):
|
|
if v_elem.text and v_elem.text.strip() == entry['original']:
|
|
v_elem.text = entry['translated']
|
|
break
|
|
data = etree.tostring(chart_xml, xml_declaration=True, encoding='UTF-8', standalone=True)
|
|
except Exception as e:
|
|
_log_error("excel_chart_update_error", chart_file=item, error=str(e))
|
|
|
|
zf_out.writestr(item, data)
|
|
|
|
with open(output_path, 'wb') as f:
|
|
f.write(buf.getvalue())
|
|
|
|
_log_info("excel_charts_translated", chart_files=len(chart_files_to_update), translations=len(translated_entries))
|
|
|
|
except Exception as e:
|
|
_log_error("excel_chart_zip_rewrite_error", error=str(e))
|
|
|
|
def _translate_images(self, worksheet: Worksheet, target_language: str) -> None:
|
|
"""
|
|
Translate text in images using vision model.
|
|
|
|
NOTE: This method is currently NOT CALLED in translate_file() as image translation
|
|
is not part of the current story scope (Story 2.7). It is intentionally preserved
|
|
for future implementation when vision model support is prioritized.
|
|
|
|
TODO: Call this method during translate_file() when implementing image translation feature.
|
|
"""
|
|
try:
|
|
images = getattr(worksheet, "_images", [])
|
|
|
|
for idx, image in enumerate(images):
|
|
try:
|
|
image_data = image._data()
|
|
ext = image.format or "png"
|
|
|
|
with tempfile.NamedTemporaryFile(
|
|
suffix=f".{ext}", delete=False
|
|
) as tmp:
|
|
tmp.write(image_data)
|
|
tmp_path = tmp.name
|
|
|
|
translated_text = self._translate_image_with_legacy(
|
|
tmp_path, target_language
|
|
)
|
|
os.unlink(tmp_path)
|
|
|
|
if translated_text and translated_text.strip():
|
|
anchor = image.anchor
|
|
if hasattr(anchor, "_from"):
|
|
cell_ref = f"{get_column_letter(anchor._from.col + 1)}{anchor._from.row + 1}"
|
|
cell = worksheet[cell_ref]
|
|
from openpyxl.comments import Comment
|
|
|
|
cell.comment = Comment(
|
|
f"Image translation: {translated_text}", "Translator"
|
|
)
|
|
_log_info(
|
|
"excel_image_translation_added",
|
|
cell_ref=cell_ref,
|
|
)
|
|
|
|
except Exception as e:
|
|
_log_error(
|
|
"excel_image_translation_error",
|
|
image_index=idx,
|
|
error=str(e),
|
|
)
|
|
|
|
except Exception as e:
|
|
_log_error(
|
|
"excel_image_processing_error",
|
|
error=str(e),
|
|
)
|
|
|
|
def _translate_image_with_legacy(
|
|
self, image_path: str, target_language: str
|
|
) -> str:
|
|
"""Translate image using active provider or legacy service."""
|
|
if self._provider and hasattr(self._provider, "translate_image"):
|
|
try:
|
|
return self._provider.translate_image(image_path, target_language)
|
|
except Exception as e:
|
|
_log_error("excel_image_translation_provider_error", error=str(e))
|
|
|
|
from services.translation_service import translation_service
|
|
# Temporarily enable translate_images flag on translation_service to bypass the hardcoded check
|
|
old_val = getattr(translation_service, "translate_images", False)
|
|
try:
|
|
translation_service.translate_images = True
|
|
if hasattr(translation_service, "translate_image"):
|
|
return translation_service.translate_image(image_path, target_language)
|
|
except Exception as e:
|
|
_log_error("excel_image_translation_legacy_error", error=str(e))
|
|
finally:
|
|
translation_service.translate_images = old_val
|
|
return ""
|
|
|
|
|
|
excel_translator = ExcelTranslator()
|