sepehr
d92bbf0fa6
Le defaut le plus sournois du pipeline etait silencieux : de l'arabe livre pour une cible persane (meme ecriture, mauvaise langue) passe inapercu et part chez le lecteur. Desormais chaque lot traduit est verifie par le detecteur d'ecritures, et chaque segment fautif est redemande une fois au moteur avec une consigne renforcee (nom de la langue + lettres specifiques, ex. persan پ چ ژ گ). La seconde tentative ne remplace la premiere que si elle passe le meme controle. - services/quality/script_detector.py : extraction d'un controle script_issue() reutilisable ; detect_arabic_variant signale des ormais un long texte en ecriture arabe sans aucune lettre specifique de la langue cible (arabe pur livre pour du persan/ourdou/pachto) - translators/segments.py : retry_wrong_script() + construction de la consigne renforcee, sans jamais faire echouer le travail - Word, Excel, PDF : branchement apres la memoire de traduction et les validations humaines ; le texte inchange (chiffres, noms propres) ne declenche jamais de retentative - 14 tests nouveaux (tests/test_translators/test_script_retry.py)
1276 lines
53 KiB
Python
1276 lines
53 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.languages import is_rtl
|
|
|
|
|
|
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._tm_scope = None # set via set_tm_scope (per-user translation memory)
|
|
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 set_tm_scope(self, user_id, prompt=None) -> None:
|
|
"""Enable the per-user translation memory for this job."""
|
|
from services.translation_tm import TMScope
|
|
|
|
self._tm_scope = TMScope.from_prompt(
|
|
user_id, prompt or getattr(self, "_custom_prompt", None)
|
|
)
|
|
|
|
def set_segment_recorder(self, recorder) -> None:
|
|
"""Attach a segment recorder (review foundation)."""
|
|
self._segment_recorder = recorder
|
|
|
|
def set_segment_overrides(self, overrides) -> None:
|
|
"""Human-reviewed translations applied verbatim on rebuild."""
|
|
self._segment_overrides = overrides or {}
|
|
|
|
def get_recorded_segments(self):
|
|
recorder = getattr(self, "_segment_recorder", None)
|
|
return recorder.get_pairs() if recorder is not None else []
|
|
|
|
|
|
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,
|
|
}
|
|
)
|
|
|
|
# Record the position where sheet names start in text_elements.
|
|
# We need this to extract the right slice of translated_texts
|
|
# AFTER chart text has been appended. Otherwise the offset
|
|
# calculation drifts and sheet names get chart translations.
|
|
sheet_name_offset = len(text_elements)
|
|
|
|
for sheet_name in sheet_names_to_translate:
|
|
text_elements.append((sheet_name, None))
|
|
|
|
# Collect chart text from ZIP (appended AFTER sheet names)
|
|
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 was captured before sheet names + chart text
|
|
# were appended, so it already points at the start of the sheet
|
|
# name block. No re-derivation needed.
|
|
|
|
_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 + chart translations. The structure of
|
|
# text_elements is:
|
|
# [0 .. sheet_name_offset) -> cell/header/footer setters
|
|
# [sheet_name_offset .. +N) -> sheet names (setter=None)
|
|
# [sheet_name_offset + N .. total_texts) -> chart setters
|
|
# The old code only iterated the first slice, silently skipping
|
|
# every chart setter. We now concatenate the two non-None slices.
|
|
_n_sheet_names = len(sheet_names_to_translate)
|
|
_chart_slice = text_elements[sheet_name_offset + _n_sheet_names:]
|
|
_chart_trans_slice = translated_texts[sheet_name_offset + _n_sheet_names:]
|
|
for i, ((original_text, setter), translated) in enumerate(
|
|
zip(
|
|
text_elements[:sheet_name_offset] + _chart_slice,
|
|
translated_texts[:sheet_name_offset] + _chart_trans_slice,
|
|
)
|
|
):
|
|
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,
|
|
)
|
|
|
|
# openpyxl does NOT rewrite references on rename: cell
|
|
# formulas, defined names and chart refs pointing at the old
|
|
# sheet would break (#REF!/#NAME?). Charts are handled later
|
|
# via ZIP re-injection; fix cells + defined names here.
|
|
if sheet_name_mapping:
|
|
self._rewrite_sheet_refs_in_workbook(workbook, sheet_name_mapping)
|
|
|
|
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))
|
|
|
|
# RTL targets: flip the sheet view so every sheet reads
|
|
# right-to-left (column A renders rightmost). Content and
|
|
# formatting are untouched — only the reading direction.
|
|
# Chartsheets have no `sheet_view` and must never break the
|
|
# save: each sheet is guarded individually.
|
|
if is_rtl(target_language):
|
|
flipped = 0
|
|
for sheet_name in workbook.sheetnames:
|
|
ws = workbook[sheet_name]
|
|
try:
|
|
if getattr(ws, "sheet_view", None) is not None:
|
|
ws.sheet_view.rightToLeft = True
|
|
flipped += 1
|
|
except Exception as e:
|
|
_log_error(
|
|
"excel_rtl_view_failed", sheet=sheet_name, error=str(e)
|
|
)
|
|
if flipped:
|
|
_log_info("excel_rtl_view_applied", sheets=flipped)
|
|
|
|
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.
|
|
# We always pass the sheet_name_mapping so chart <c:f> references
|
|
# are updated when sheet names were translated — otherwise the
|
|
# chart would keep pointing at the old (now-missing) sheet and
|
|
# render as empty/zero series.
|
|
self._apply_chart_translations(
|
|
output_path,
|
|
chart_translations,
|
|
sheet_name_mapping=sheet_name_mapping or None,
|
|
)
|
|
|
|
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)
|
|
|
|
from services.translation_tm import translate_with_tm
|
|
|
|
provider_name = (
|
|
self._provider.get_name() if hasattr(self._provider, "get_name")
|
|
else type(self._provider).__name__
|
|
) if self._provider is not None else "legacy"
|
|
|
|
if self._provider is not None:
|
|
def _do_translate(miss_texts):
|
|
return self._translate_with_provider(
|
|
miss_texts, target_language, source_language
|
|
)
|
|
else:
|
|
def _do_translate(miss_texts):
|
|
return self._translate_with_legacy(
|
|
miss_texts, target_language, source_language
|
|
)
|
|
|
|
from translators.segments import apply_overrides
|
|
|
|
# Reviewer overrides (approved/edited segments) win over everything:
|
|
# no TM lookup, no provider call, zero drift from the reviewed text.
|
|
ov_hits, ov_misses = apply_overrides(
|
|
texts, getattr(self, "_segment_overrides", None)
|
|
)
|
|
miss_texts = [texts[i] for i in ov_misses]
|
|
|
|
if miss_texts:
|
|
# Translation memory: reuse this user's previous translations
|
|
# (identical context/prompt) before hitting the provider.
|
|
miss_translated = translate_with_tm(
|
|
miss_texts, target_language, source_language,
|
|
provider_name, getattr(self, "_tm_scope", None), _do_translate,
|
|
)
|
|
else:
|
|
miss_translated = []
|
|
|
|
translated = []
|
|
miss_pos = 0
|
|
for i in range(len(texts)):
|
|
if i in ov_hits:
|
|
translated.append(ov_hits[i])
|
|
else:
|
|
translated.append(
|
|
miss_translated[miss_pos]
|
|
if miss_pos < len(miss_translated)
|
|
else texts[i]
|
|
)
|
|
miss_pos += 1
|
|
|
|
# Wrong-script guard: re-ask once (with a reinforced instruction)
|
|
# for translations delivered in the wrong alphabet — e.g. Arabic
|
|
# delivered for a Persian target.
|
|
from translators.segments import retry_wrong_script
|
|
|
|
def _retry_translate(retry_texts, hint):
|
|
if self._provider is not None:
|
|
return self._translate_with_provider(
|
|
retry_texts, target_language, source_language, extra_prompt=hint
|
|
)
|
|
return self._translate_with_legacy(
|
|
retry_texts, target_language, source_language
|
|
)
|
|
|
|
translated, _script_fixed = retry_wrong_script(
|
|
texts, translated, target_language, _retry_translate
|
|
)
|
|
|
|
recorder = getattr(self, "_segment_recorder", None)
|
|
if recorder is not None:
|
|
recorder.record_pairs(zip(texts, translated))
|
|
|
|
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,
|
|
extra_prompt: Optional[str] = None,
|
|
) -> 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)
|
|
if extra_prompt:
|
|
custom_prompt = (
|
|
f"{custom_prompt}\n\n{extra_prompt}" if custom_prompt else extra_prompt
|
|
)
|
|
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).
|
|
|
|
Chartsheets are skipped: they have no cell grid (no iter_rows)
|
|
— their text lives in the chart XML, collected separately from
|
|
the ZIP.
|
|
"""
|
|
if not hasattr(worksheet, "iter_rows"):
|
|
return
|
|
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]],
|
|
sheet_name_mapping: Optional[Dict[str, str]] = None,
|
|
) -> 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.
|
|
|
|
Also rewrites `<c:f>` references inside chart XML when sheet names
|
|
were translated — otherwise the chart keeps pointing at the old
|
|
sheet name and renders empty (zero) data.
|
|
"""
|
|
if not chart_translations and not sheet_name_mapping:
|
|
return
|
|
|
|
translated_entries = [e for e in chart_translations if 'translated' in e and e['translated']]
|
|
if not translated_entries and not sheet_name_mapping:
|
|
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)
|
|
|
|
# If we have sheet renames, we must walk every chart file (even ones
|
|
# with no text translations) so the <c:f> references are updated.
|
|
if sheet_name_mapping:
|
|
# We need the list of chart files; peek into the zip to find them.
|
|
try:
|
|
with zipfile.ZipFile(output_path, 'r') as zf_peek:
|
|
for name in zf_peek.namelist():
|
|
if name.startswith('xl/charts/') and name.endswith('.xml'):
|
|
chart_files_to_update.setdefault(name, [])
|
|
except Exception:
|
|
pass
|
|
|
|
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
|
|
|
|
# Rewrite sheet references in <c:f> when sheet
|
|
# names were translated. Without this, charts
|
|
# point at sheets that no longer exist and
|
|
# render as zero/empty series.
|
|
if sheet_name_mapping:
|
|
self._rewrite_sheet_refs_in_chart(chart_xml, sheet_name_mapping)
|
|
|
|
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),
|
|
sheet_refs_rewritten=len(sheet_name_mapping or {}),
|
|
)
|
|
|
|
except Exception as e:
|
|
_log_error("excel_chart_zip_rewrite_error", error=str(e))
|
|
|
|
@staticmethod
|
|
def _quote_sheet_name_for_ref(name: str) -> str:
|
|
"""Return the sheet name as it must appear inside a chart <c:f> reference.
|
|
|
|
Excel requires single-quoting when the name contains spaces or any of
|
|
`:\\/?*[]`, and any single quote inside the name must be doubled.
|
|
"""
|
|
if not name:
|
|
return name
|
|
needs_quote = any(c in name for c in " :\\/?*[]") or " " in name
|
|
escaped = name.replace("'", "''")
|
|
return f"'{escaped}'" if needs_quote else escaped
|
|
|
|
@classmethod
|
|
def _rewrite_sheet_ref(cls, ref: str, sheet_name_mapping: Dict[str, str]) -> str:
|
|
"""Rewrite a single chart <c:f> reference to use translated sheet names.
|
|
|
|
Handles both quoted (`'Sheet 1'!A1`) and unquoted (`Sheet1!A1`) forms.
|
|
If the sheet name is not in the mapping, returns ref unchanged.
|
|
"""
|
|
if not sheet_name_mapping or not ref or '!' not in ref:
|
|
return ref
|
|
|
|
if ref.startswith("'"):
|
|
# Find the closing single quote that ends the sheet name.
|
|
# Per the ECMA-376 grammar, the inner `'` is escaped as `''`,
|
|
# so we can safely scan for the next unescaped `'`.
|
|
i = 1
|
|
while i < len(ref):
|
|
if ref[i] == "'" and i + 1 < len(ref) and ref[i + 1] == "'":
|
|
i += 2 # skip escaped apostrophe
|
|
continue
|
|
if ref[i] == "'":
|
|
break
|
|
i += 1
|
|
if i >= len(ref) or ref[i] != "'":
|
|
return ref
|
|
sheet_name = ref[1:i]
|
|
rest = ref[i + 1:] # starts with "!<cellref>"
|
|
else:
|
|
bang = ref.index('!')
|
|
sheet_name = ref[:bang]
|
|
rest = ref[bang:]
|
|
|
|
new_name = sheet_name_mapping.get(sheet_name)
|
|
if not new_name or new_name == sheet_name:
|
|
return ref
|
|
|
|
return cls._quote_sheet_name_for_ref(new_name) + rest
|
|
|
|
@classmethod
|
|
def _rewrite_sheet_refs_in_chart(
|
|
cls, chart_xml, sheet_name_mapping: Dict[str, str]
|
|
) -> int:
|
|
"""Walk all <c:f> in the chart and rewrite those pointing at renamed sheets.
|
|
|
|
Returns the number of references rewritten (for logging/tests).
|
|
"""
|
|
_NS_C = "http://schemas.openxmlformats.org/drawingml/2006/chart"
|
|
rewritten = 0
|
|
for f_elem in chart_xml.iter(f'{{{_NS_C}}}f'):
|
|
original = f_elem.text
|
|
if not original or '!' not in original:
|
|
continue
|
|
updated = cls._rewrite_sheet_ref(original, sheet_name_mapping)
|
|
if updated != original:
|
|
f_elem.text = updated
|
|
rewritten += 1
|
|
return rewritten
|
|
|
|
@staticmethod
|
|
def _rewrite_sheet_refs_in_formula(
|
|
formula: str, sheet_name_mapping: Dict[str, str]
|
|
) -> str:
|
|
"""Rewrite every sheet reference inside a formula string.
|
|
|
|
Handles quoted ('Ventes 2026'!), unquoted (Ventes!) and 3D refs
|
|
(Sheet1!A1:Sheet2!B2 — each end is rewritten independently).
|
|
Longest names are replaced first so a name that is a prefix of
|
|
another is not corrupted.
|
|
"""
|
|
if not formula or not sheet_name_mapping or "!" not in formula:
|
|
return formula
|
|
|
|
result = formula
|
|
# longest first to avoid partial-name collisions
|
|
for old in sorted(sheet_name_mapping, key=len, reverse=True):
|
|
new = sheet_name_mapping[old]
|
|
if new == old:
|
|
continue
|
|
new_quoted = ExcelTranslator._quote_sheet_name_for_ref(new)
|
|
# Quoted form: inner apostrophes are escaped as ''
|
|
escaped = old.replace("'", "''")
|
|
result = re.sub(
|
|
rf"'{re.escape(escaped)}'!",
|
|
new_quoted + "!",
|
|
result,
|
|
)
|
|
# Unquoted form: only when the old name needs no quotes; guard
|
|
# against matching the tail of a longer name.
|
|
if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_.]*", old):
|
|
result = re.sub(
|
|
rf"(?<![A-Za-z0-9_.']){re.escape(old)}!",
|
|
new_quoted + "!",
|
|
result,
|
|
)
|
|
return result
|
|
|
|
@classmethod
|
|
def _rewrite_sheet_refs_in_workbook(
|
|
cls, workbook, sheet_name_mapping: Dict[str, str]
|
|
) -> None:
|
|
"""After a sheet rename, fix every reference that still points at the
|
|
old names: cell formulas, defined names, data validations and
|
|
conditional-formatting rules. openpyxl does none of this on rename.
|
|
|
|
Best-effort: individual failures are logged and skipped — a broken
|
|
rename must never lose the whole file.
|
|
"""
|
|
if not sheet_name_mapping:
|
|
return
|
|
|
|
cells_fixed = names_fixed = validations_fixed = cf_fixed = 0
|
|
|
|
try:
|
|
for worksheet in workbook.worksheets:
|
|
# 1) Cell formulas
|
|
for row in worksheet.iter_rows():
|
|
for cell in row:
|
|
value = cell.value
|
|
if isinstance(value, str) and value.startswith("=") and "!" in value:
|
|
updated = cls._rewrite_sheet_refs_in_formula(
|
|
value, sheet_name_mapping
|
|
)
|
|
if updated != value:
|
|
cell.value = updated
|
|
cells_fixed += 1
|
|
|
|
# 2) Data validations (list sources / custom formulas)
|
|
try:
|
|
for dv in worksheet.data_validations.dataValidation:
|
|
for attr in ("formula1", "formula2"):
|
|
fx = getattr(dv, attr, None)
|
|
if isinstance(fx, str) and "!" in fx:
|
|
updated = cls._rewrite_sheet_refs_in_formula(
|
|
fx, sheet_name_mapping
|
|
)
|
|
if updated != fx:
|
|
setattr(dv, attr, updated)
|
|
validations_fixed += 1
|
|
except Exception as e:
|
|
_log_warning(
|
|
"excel_sheet_refs_validations_failed",
|
|
sheet=worksheet.title,
|
|
error=str(e),
|
|
)
|
|
|
|
# 3) Conditional formatting rules
|
|
try:
|
|
for cf in worksheet.conditional_formatting:
|
|
for rule in cf.rules:
|
|
formulas = getattr(rule, "formula", None) or []
|
|
for idx, fx in enumerate(formulas):
|
|
if isinstance(fx, str) and "!" in fx:
|
|
updated = cls._rewrite_sheet_refs_in_formula(
|
|
fx, sheet_name_mapping
|
|
)
|
|
if updated != fx:
|
|
formulas[idx] = updated
|
|
cf_fixed += 1
|
|
except Exception as e:
|
|
_log_warning(
|
|
"excel_sheet_refs_condfmt_failed",
|
|
sheet=worksheet.title,
|
|
error=str(e),
|
|
)
|
|
|
|
# 4) Workbook-level defined names
|
|
try:
|
|
for name in list(workbook.defined_names.values()):
|
|
attr_text = getattr(name, "attr_text", None)
|
|
if isinstance(attr_text, str) and "!" in attr_text:
|
|
updated = cls._rewrite_sheet_refs_in_formula(
|
|
attr_text, sheet_name_mapping
|
|
)
|
|
if updated != attr_text:
|
|
name.attr_text = updated
|
|
names_fixed += 1
|
|
except Exception as e:
|
|
_log_warning("excel_sheet_refs_names_failed", error=str(e))
|
|
|
|
except Exception as e:
|
|
_log_error("excel_sheet_refs_rewrite_failed", error=str(e))
|
|
return
|
|
|
|
if cells_fixed or names_fixed or validations_fixed or cf_fixed:
|
|
_log_info(
|
|
"excel_sheet_refs_rewritten",
|
|
cells=cells_fixed,
|
|
defined_names=names_fixed,
|
|
validations=validations_fixed,
|
|
conditional_formats=cf_fixed,
|
|
)
|
|
|
|
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()
|