Files
office_translator/translators/excel_translator.py
sepehr b4e873ad2c
Some checks failed
Deploy to Production / Build and Deploy (push) Failing after 2m14s
feat(review,teams): review foundation — segments, side-by-side editor, rebuild, XLIFF, team workspaces
Foundations:
- TranslationSegment model + migration f7e8d9c0b1a2 (segments, workspaces,
  workspace_members, glossaries.workspace_id)
- SegmentRecorder injected into all 4 translators: unique (source,
  translation) pairs captured per job and persisted (best-effort)
- set_segment_overrides: human-reviewed translations applied verbatim on
  rebuild — top priority over TM and provider, zero API calls

Review API (routes/review_routes.py):
- GET /translations/{id}/segments (owner or job token)
- PATCH /segments/{id} edit/approve — feeds the per-user TM so approved
  translations are reused in later jobs
- POST /translations/{id}/rebuild — rebuild document with reviewed text
- GET/POST /translations/{id}/xliff — XLIFF 1.2 export/import (edited
  segments export their reviewed text)

Review editor (frontend /dashboard/reviews/[jobId]):
- side-by-side source/translation table, inline edit, approve (single or
  all), rebuild & download (auth blob), XLIFF export/import, 13 locales
- 'Relire et corriger' link on the translation-complete screen

Team workspaces (routes/workspace_routes.py + /dashboard/teams):
- Workspace/WorkspaceMember models, roles owner/admin/member
- create (Business plan), list with seat usage, invite by email with
  seat-limit enforcement (Business=5, Enterprise unlimited), removal
- shared glossaries: workspace members can use a glossary shared to their
  workspace (access check extended)

Tests: 1184 passed / 0 failed (11 new: recorder, overrides, docx
capture->rebuild e2e, XLIFF structure/escaping, seats, workspace CRUD,
shared glossary access)
2026-08-29 19:04:32 +02:00

1225 lines
51 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._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))
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
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
) -> 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]],
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()