sepehr
3ae28dd3cb
All checks were successful
Deploy to Production / Build and Deploy (push) Successful in 2m36s
B3.6 — fix two visual bugs reported on the user's prod PDF:
1. Title 'Spécification technique : Office Translator v3.0' overflowed
its 2-line bbox and overlapped the 'Version du document...' block.
Root cause: MAX_VERTICAL_EXPANSION was 1.5x the original height,
way too small for a long French title. Bumped to 6.0x.
2. 'Avis important' blue background had white rectangular patches.
Root cause: redaction always used fill=(1,1,1) (opaque white),
which erased the colored drawing underneath the text.
Fix: detect when a block's bbox intersects a page drawing,
and use fill=None (transparent) for the redaction in that case.
The original drawing survives intact.
B3.7 — eliminate remaining block-vs-next-block overlap:
Computes each block's 'next_block_y' (the y0 of the nearest block
below it on the same page) and uses it as the ceiling for vertical
expansion. Previously the smart-fit logic used the page bottom as
the ceiling, which let long translated blocks flow into their
neighbour (e.g. 'Pour la dernière version...' overlapping
'8. Résolution des problèmes' in the TOC).
Also includes:
- 9 new tests (6 B3.6 + 3 B3.7) — total 446 tests pass, zero regression
- scripts/verify_b3_6_fix.py — visual+structural verification
- Updated sample_files/test_corpus/test_pdf_translated.pdf with the
clean B3.6+B3.7 output
1187 lines
44 KiB
Python
1187 lines
44 KiB
Python
"""
|
||
PDF Document Translation Module — Layout-Preserving
|
||
|
||
Primary strategy (layout mode):
|
||
Use PyMuPDF (fitz) for direct, in-place text replacement on each page.
|
||
|
||
For each page:
|
||
1. Extract text blocks with positions, fonts, sizes, colors
|
||
2. Translate each block as a unit (preserving context within block)
|
||
3. Redact original text area
|
||
4. Write translated text at the same position, auto-adjusting font size
|
||
|
||
This preserves:
|
||
- Page structure, images, vector graphics, backgrounds
|
||
- Text positions within original bounding boxes
|
||
- Approximate font styling (size, color)
|
||
|
||
Fallback:
|
||
If PyMuPDF direct editing fails, falls back to the pdf2docx pipeline
|
||
(PDF → DOCX → WordTranslator → PDF via LibreOffice).
|
||
|
||
Text-only mode:
|
||
Extract text, translate, generate a clean formatted PDF via reportlab.
|
||
"""
|
||
|
||
import time
|
||
import shutil
|
||
import subprocess
|
||
from pathlib import Path
|
||
from typing import Dict, Any, Optional, Callable, List
|
||
|
||
from core.logging import get_logger
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
# Minimum readable font size (points)
|
||
MIN_FONT_SIZE = 4.5
|
||
|
||
# Font size reduction factor when text overflows its bounding box
|
||
FONT_SHRINK_FACTOR = 0.93 # Track B3.5: was 0.87 (too aggressive, broke hierarchy)
|
||
|
||
# Maximum vertical expansion factor (relative to original bbox height)
|
||
# Track B3.5: was 1.5x; bumped to 3x so longer translations can flow down
|
||
# Track B3.6: bumped to 6x to handle long French titles that wrap to
|
||
# multiple lines (a 22pt title may need 4-5 lines in French).
|
||
MAX_VERTICAL_EXPANSION = 6.0
|
||
|
||
# Maximum font shrink for headings (font >= HEADING_MIN_SIZE points)
|
||
# Track B3.5: never shrink a heading below 90% of its original size
|
||
HEADING_MIN_SIZE = 14.0
|
||
HEADING_MIN_SCALE = 0.90
|
||
|
||
# Maximum font shrink for body text
|
||
BODY_MIN_SCALE = 0.75
|
||
|
||
# RTL language codes
|
||
RTL_LANGUAGES = frozenset({"ar", "he", "fa", "ur", "ku", "ps", "ug", "sd", "yi", "dv", "ckb"})
|
||
|
||
|
||
def _record_format_loss_metric(element_type: str, count: int = 1) -> None:
|
||
"""Best-effort emission of a format-loss metric (never raises).
|
||
|
||
Used by the PDF translator to track hyperlinks/annotations lost
|
||
during in-place text replacement.
|
||
"""
|
||
if count <= 0:
|
||
return
|
||
try:
|
||
from middleware.metrics import record_format_loss
|
||
for _ in range(count):
|
||
record_format_loss(format="pdf", element_type=element_type)
|
||
except Exception:
|
||
# Metrics must never break a job
|
||
pass
|
||
|
||
|
||
def _libreoffice_available() -> bool:
|
||
"""Check if the LibreOffice CLI is available on this host.
|
||
|
||
Result is cached for 5 minutes to avoid spawning a subprocess
|
||
on every PDF translation.
|
||
"""
|
||
import time
|
||
now = time.time()
|
||
cached = _libreoffice_available._cache
|
||
if cached and (now - cached[0]) < 300:
|
||
return cached[1]
|
||
try:
|
||
result = subprocess.run(
|
||
["libreoffice", "--version"],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=5,
|
||
)
|
||
ok = result.returncode == 0
|
||
except (FileNotFoundError, subprocess.TimeoutExpired, Exception):
|
||
ok = False
|
||
_libreoffice_available._cache = (now, ok)
|
||
return ok
|
||
|
||
|
||
_libreoffice_available._cache = None # type: ignore[attr-defined]
|
||
|
||
|
||
class PDFTranslator:
|
||
"""Translates PDF files with layout preservation using PyMuPDF."""
|
||
|
||
_FONT_SEARCH_PATHS = [
|
||
"/usr/share/fonts/opentype/noto/NotoSans-Regular.ttf",
|
||
"/usr/share/fonts/opentype/noto/NotoSans.ttc",
|
||
"/usr/share/fonts/truetype/noto/NotoSans-Regular.ttf",
|
||
"/usr/share/fonts/truetype/noto/NotoSans[Noto].ttf",
|
||
"/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
|
||
"/usr/share/fonts/truetype/liberation/LiberationSans-Regular.ttf",
|
||
"/usr/share/fonts/truetype/freefont/FreeSans.ttf",
|
||
"/app/fonts/NotoSans-Regular.ttf",
|
||
"C:/Windows/Fonts/arial.ttf",
|
||
"C:/Windows/Fonts/msyh.ttc",
|
||
"/System/Library/Fonts/Helvetica.ttc",
|
||
]
|
||
|
||
def __init__(self, provider=None):
|
||
self._provider = provider
|
||
self._font_path: Optional[str] = None
|
||
self._translation_stats = {"attempted": 0, "changed": 0}
|
||
self._custom_prompt: Optional[str] = None
|
||
|
||
def set_provider(self, provider) -> 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 _get_font_path(self) -> Optional[str]:
|
||
"""Resolve a Unicode-capable TTF/OTF font file."""
|
||
if self._font_path is not None:
|
||
return self._font_path
|
||
for p in self._FONT_SEARCH_PATHS:
|
||
if Path(p).exists():
|
||
self._font_path = p
|
||
return p
|
||
logger.warning("no_unicode_font_found")
|
||
return None
|
||
|
||
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,
|
||
pdf_mode: str = "layout",
|
||
translate_images: bool = False,
|
||
**kwargs,
|
||
) -> Path:
|
||
input_path = Path(input_path)
|
||
output_path = Path(output_path)
|
||
self._validate_file(input_path)
|
||
|
||
if pdf_mode == "text_only":
|
||
return self._translate_text_only(
|
||
input_path, output_path, target_language, source_language, progress_callback
|
||
)
|
||
return self._translate_preserve_layout(
|
||
input_path, output_path, target_language, source_language, progress_callback,
|
||
translate_images=translate_images,
|
||
)
|
||
|
||
# ------------------------------------------------------------------ #
|
||
# LAYOUT MODE — PyMuPDF in-place text replacement
|
||
# ------------------------------------------------------------------ #
|
||
|
||
def _translate_preserve_layout(
|
||
self,
|
||
input_path: Path,
|
||
output_path: Path,
|
||
target_language: str,
|
||
source_language: str,
|
||
progress_callback,
|
||
translate_images: bool = False,
|
||
) -> Path:
|
||
"""Translate PDF preserving layout via PyMuPDF direct text replacement."""
|
||
start_time = time.time()
|
||
|
||
try:
|
||
import fitz
|
||
except ImportError:
|
||
logger.warning("pymupdf_missing_fallback_docx")
|
||
return self._translate_preserve_layout_fallback(
|
||
input_path, output_path, target_language, source_language, progress_callback,
|
||
translate_images=translate_images,
|
||
)
|
||
|
||
doc = fitz.open(str(input_path))
|
||
total_pages = len(doc)
|
||
|
||
if total_pages == 0:
|
||
doc.close()
|
||
raise RuntimeError("PDF has no pages.")
|
||
|
||
font_path = self._get_font_path()
|
||
logger.info(
|
||
"pdf_layout_start",
|
||
pages=total_pages,
|
||
file=input_path.name,
|
||
font=font_path or "built-in",
|
||
)
|
||
|
||
if progress_callback:
|
||
progress_callback({
|
||
"current": 1, "total": total_pages,
|
||
"phase": "extracting",
|
||
"paragraph": 1, "total_paragraphs": total_pages,
|
||
})
|
||
|
||
try:
|
||
result_path = self._process_pages_inplace(
|
||
doc, total_pages, output_path,
|
||
target_language, source_language,
|
||
font_path, progress_callback,
|
||
)
|
||
|
||
processing_time_ms = round((time.time() - start_time) * 1000, 2)
|
||
logger.info(
|
||
"pdf_layout_success",
|
||
pages=total_pages,
|
||
processing_time_ms=processing_time_ms,
|
||
output=str(result_path),
|
||
)
|
||
return result_path
|
||
|
||
except Exception as e:
|
||
doc.close()
|
||
logger.warning("inplace_failed_fallback", error=str(e))
|
||
return self._translate_preserve_layout_fallback(
|
||
input_path, output_path, target_language, source_language, progress_callback,
|
||
translate_images=translate_images,
|
||
)
|
||
|
||
def _process_pages_inplace(
|
||
self,
|
||
doc,
|
||
total_pages: int,
|
||
output_path: Path,
|
||
target_language: str,
|
||
source_language: str,
|
||
font_path: Optional[str],
|
||
progress_callback,
|
||
) -> Path:
|
||
"""Core PyMuPDF in-place processing — one page at a time."""
|
||
import fitz
|
||
|
||
is_rtl = target_language.lower() in RTL_LANGUAGES
|
||
total_blocks = 0
|
||
translated_blocks = 0
|
||
|
||
for page_num in range(total_pages):
|
||
page = doc[page_num]
|
||
raw_blocks = self._extract_text_blocks(page)
|
||
|
||
if not raw_blocks:
|
||
if progress_callback:
|
||
pct = int(30 + 65 * (page_num + 1) / total_pages)
|
||
progress_callback({
|
||
"current": page_num + 1, "total": total_pages,
|
||
"phase": f"Page {page_num + 1}/{total_pages} (no text)",
|
||
"paragraph": page_num + 1,
|
||
"total_paragraphs": total_pages,
|
||
"progress_override": pct,
|
||
})
|
||
continue
|
||
|
||
# Merge adjacent blocks that form a single paragraph
|
||
blocks = self._merge_adjacent_blocks(raw_blocks, page.rect)
|
||
total_blocks += len(blocks)
|
||
|
||
# Track B3.7: compute each block's "next block" y0 so the
|
||
# smart-fit logic never expands a block into its neighbour.
|
||
# Without this, a long translated block can flow into the
|
||
# block below it (e.g. "Pour la dernière version" overlapping
|
||
# "8. Résolution des problèmes").
|
||
self._populate_next_block_y(blocks, page.rect)
|
||
|
||
# Phase 1: translate all blocks on this page
|
||
for block in blocks:
|
||
original = block["text"]
|
||
if not original.strip():
|
||
continue
|
||
|
||
try:
|
||
translated = self._translate_single(
|
||
original, target_language, source_language
|
||
)
|
||
if translated and translated.strip():
|
||
block["translated"] = translated
|
||
translated_blocks += 1
|
||
else:
|
||
logger.warning(
|
||
"block_translation_empty",
|
||
page=page_num + 1,
|
||
text_preview=original[:60],
|
||
)
|
||
except Exception as e:
|
||
logger.warning(
|
||
"block_translation_failed",
|
||
page=page_num + 1,
|
||
error=str(e),
|
||
)
|
||
|
||
# Phase 2: redact original text areas
|
||
#
|
||
# CRITICAL (Track B3.6): For blocks that contain a vector
|
||
# drawing (e.g. a colored background box), we need to use a
|
||
# TRANSPARENT redaction (fill=None) so the original drawing
|
||
# isn't erased. For all other blocks, we use the standard
|
||
# white redaction to fully erase the text.
|
||
#
|
||
# Before redaction, we also capture the page's hyperlinks
|
||
# (PDF links are not stored in text blocks; they're attached
|
||
# to page annotations). After redaction, we re-insert the
|
||
# links so the translated page remains navigable.
|
||
#
|
||
# PDF_REDACT_IMAGE_NONE is critical: it tells PyMuPDF NOT to
|
||
# rasterize image content that intersects redaction areas.
|
||
# Without this flag, the redaction would replace the text
|
||
# AND the underlying image with a blank rectangle — wiping
|
||
# out any diagrams, charts, or photos in the same area.
|
||
page_links_before: list = []
|
||
if hasattr(page, "get_links"):
|
||
try:
|
||
page_links_before = list(page.get_links())
|
||
except Exception:
|
||
page_links_before = []
|
||
|
||
# Find the rectangles of all vector drawings on the page.
|
||
# If a block's bbox intersects a drawing rect, we use
|
||
# transparent redaction to preserve the drawing.
|
||
drawing_rects: list = []
|
||
if hasattr(page, "get_drawings"):
|
||
try:
|
||
for d in page.get_drawings():
|
||
r = d.get("rect")
|
||
if r is not None and d.get("fill") is not None:
|
||
drawing_rects.append(fitz.Rect(r))
|
||
except Exception:
|
||
pass
|
||
|
||
def _block_intersects_drawing(block_bbox):
|
||
"""True if the block's bbox intersects any drawing rect."""
|
||
if not drawing_rects:
|
||
return False
|
||
b = fitz.Rect(block_bbox)
|
||
for dr in drawing_rects:
|
||
if b.intersects(dr):
|
||
return True
|
||
return False
|
||
|
||
for block in blocks:
|
||
if block.get("translated"):
|
||
# Track B3.5: single redaction per block, not per sub-bbox.
|
||
# Track B3.6: if the block contains a drawing, use
|
||
# transparent redaction (fill=None) so the drawing
|
||
# underneath isn't erased. Otherwise use white fill
|
||
# to fully remove the original text.
|
||
block_bbox = fitz.Rect(block["bbox"])
|
||
if _block_intersects_drawing(block_bbox):
|
||
# Transparent redaction: removes text but keeps
|
||
# the underlying drawing intact.
|
||
page.add_redact_annot(block_bbox, fill=None)
|
||
else:
|
||
page.add_redact_annot(block_bbox, fill=(1, 1, 1))
|
||
|
||
page.apply_redactions(images=fitz.PDF_REDACT_IMAGE_NONE)
|
||
|
||
# Re-insert hyperlinks that were inside the redacted area.
|
||
# We use the original link geometry (it's unaffected by the
|
||
# redaction). URIs are preserved verbatim; only the visible
|
||
# text changes.
|
||
if page_links_before:
|
||
reinserted = 0
|
||
lost = 0
|
||
page_rect = page.rect
|
||
for link in page_links_before:
|
||
try:
|
||
from_rect = link.get("from")
|
||
if from_rect is None:
|
||
lost += 1
|
||
continue
|
||
# Clamp the rect to the page (in case redaction
|
||
# shrunk it)
|
||
if not page_rect.intersects(from_rect):
|
||
lost += 1
|
||
continue
|
||
# Build the link insertion kwargs based on kind
|
||
if link.get("uri"):
|
||
# External URI link
|
||
page.insert_link({
|
||
"kind": fitz.LINK_URI,
|
||
"from": from_rect,
|
||
"uri": link["uri"],
|
||
})
|
||
reinserted += 1
|
||
elif link.get("page") is not None:
|
||
# Internal goto link
|
||
page.insert_link({
|
||
"kind": fitz.LINK_GOTO,
|
||
"from": from_rect,
|
||
"page": link["page"],
|
||
"to": link.get("to", fitz.Point(0, 0)),
|
||
})
|
||
reinserted += 1
|
||
else:
|
||
lost += 1
|
||
except Exception:
|
||
lost += 1
|
||
if reinserted > 0:
|
||
logger.info(
|
||
"pdf_hyperlinks_reinserted",
|
||
page=page_num + 1,
|
||
reinserted=reinserted,
|
||
lost=lost,
|
||
)
|
||
if lost > 0:
|
||
_record_format_loss_metric("hyperlink", count=lost)
|
||
|
||
# Phase 3: write translated text
|
||
for block in blocks:
|
||
if block.get("translated"):
|
||
self._write_translated_block(
|
||
page, block, font_path, is_rtl
|
||
)
|
||
|
||
if progress_callback:
|
||
pct = int(30 + 65 * (page_num + 1) / total_pages)
|
||
progress_callback({
|
||
"current": page_num + 1, "total": total_pages,
|
||
"phase": f"Translating page {page_num + 1}/{total_pages}",
|
||
"paragraph": page_num + 1,
|
||
"total_paragraphs": total_pages,
|
||
"progress_override": pct,
|
||
})
|
||
|
||
logger.info(
|
||
"pdf_blocks_processed",
|
||
total_blocks=total_blocks,
|
||
translated_blocks=translated_blocks,
|
||
)
|
||
|
||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||
doc.save(str(output_path), garbage=4, deflate=True)
|
||
doc.close()
|
||
return output_path
|
||
|
||
def _extract_text_blocks(self, page) -> List[Dict]:
|
||
"""Extract text blocks with position, font, and color information."""
|
||
import fitz
|
||
|
||
blocks = []
|
||
data = page.get_text("dict", flags=fitz.TEXT_PRESERVE_WHITESPACE)
|
||
|
||
for block in data.get("blocks", []):
|
||
if block.get("type") != 0:
|
||
continue
|
||
|
||
lines = block.get("lines", [])
|
||
if not lines:
|
||
continue
|
||
|
||
line_parts = []
|
||
spans_info = []
|
||
|
||
for line in lines:
|
||
span_parts = []
|
||
for span in line.get("spans", []):
|
||
text = span.get("text", "")
|
||
if text:
|
||
span_parts.append(text)
|
||
spans_info.append({
|
||
"size": span.get("size", 12),
|
||
"font": span.get("font", "Helvetica"),
|
||
"color": span.get("color", 0),
|
||
"flags": span.get("flags", 0),
|
||
"origin": span.get("origin", (0, 0)),
|
||
})
|
||
if span_parts:
|
||
line_parts.append("".join(span_parts))
|
||
|
||
full_text = "\n".join(line_parts).strip()
|
||
if not full_text:
|
||
continue
|
||
|
||
avg_size = (
|
||
sum(s["size"] for s in spans_info) / len(spans_info)
|
||
if spans_info
|
||
else 12.0
|
||
)
|
||
first_color = spans_info[0]["color"] if spans_info else 0
|
||
is_bold = any(s["flags"] & 16 for s in spans_info)
|
||
is_italic = any(s["flags"] & 2 for s in spans_info)
|
||
|
||
blocks.append({
|
||
"bbox": tuple(block["bbox"]),
|
||
"text": full_text,
|
||
"font_size": round(avg_size, 1),
|
||
"color": first_color,
|
||
"is_bold": is_bold,
|
||
"is_italic": is_italic,
|
||
"line_count": len(line_parts),
|
||
"translated": None,
|
||
"sub_bboxes": [tuple(block["bbox"])],
|
||
})
|
||
|
||
return blocks
|
||
|
||
def _merge_adjacent_blocks(
|
||
self, blocks: List[Dict], page_rect
|
||
) -> List[Dict]:
|
||
"""Merge consecutive text blocks that form a single paragraph.
|
||
|
||
Blocks are merged when they:
|
||
- Have the same (or very close) font size
|
||
- Are vertically adjacent (gap < 1.5× line height)
|
||
- Have the same x-origin (left-aligned) or same width
|
||
This produces larger bounding boxes for better translation context
|
||
and prevents excessive font-size reduction for multi-line paragraphs.
|
||
"""
|
||
if len(blocks) <= 1:
|
||
return blocks
|
||
|
||
merged = []
|
||
current = dict(blocks[0])
|
||
|
||
for next_block in blocks[1:]:
|
||
should_merge = self._should_merge_blocks(current, next_block)
|
||
if should_merge:
|
||
# Merge: combine text and expand bounding box
|
||
current["text"] += "\n" + next_block["text"]
|
||
current["line_count"] += next_block["line_count"]
|
||
# Expand bbox to cover both
|
||
x0 = min(current["bbox"][0], next_block["bbox"][0])
|
||
y0 = min(current["bbox"][1], next_block["bbox"][1])
|
||
x1 = max(current["bbox"][2], next_block["bbox"][2])
|
||
y1 = max(current["bbox"][3], next_block["bbox"][3])
|
||
current["bbox"] = (x0, y0, x1, y1)
|
||
current["sub_bboxes"].extend(next_block["sub_bboxes"])
|
||
else:
|
||
merged.append(current)
|
||
current = dict(next_block)
|
||
|
||
merged.append(current)
|
||
return merged
|
||
|
||
def _should_merge_blocks(self, a: Dict, b: Dict) -> bool:
|
||
"""Check if two blocks should be merged into one paragraph."""
|
||
a_bbox = a["bbox"]
|
||
b_bbox = b["bbox"]
|
||
|
||
# Must have similar font size (within 20%)
|
||
if abs(a["font_size"] - b["font_size"]) > max(a["font_size"], b["font_size"]) * 0.2:
|
||
return False
|
||
|
||
# Block b must start soon after block a ends vertically
|
||
vertical_gap = b_bbox[1] - a_bbox[3]
|
||
line_height = a["font_size"] * 1.4
|
||
if vertical_gap < 0 or vertical_gap > line_height * 1.5:
|
||
return False
|
||
|
||
# Similar horizontal position (within 15pt)
|
||
if abs(a_bbox[0] - b_bbox[0]) > 15:
|
||
return False
|
||
|
||
# Don't merge if widths are very different (likely different columns)
|
||
a_width = a_bbox[2] - a_bbox[0]
|
||
b_width = b_bbox[2] - b_bbox[0]
|
||
if a_width > 0 and abs(b_width - a_width) / a_width > 0.5:
|
||
return False
|
||
|
||
return True
|
||
|
||
def _populate_next_block_y(self, blocks: List[Dict], page_rect) -> None:
|
||
"""For each block, set ``block['_next_block_y']`` to the y0 of the
|
||
nearest block below it (or the page bottom margin if none).
|
||
|
||
Track B3.7: this lets the smart-fit logic use the next block as
|
||
a ceiling for vertical expansion, so a long translated block
|
||
never overlaps its neighbour.
|
||
"""
|
||
margin = 18
|
||
page_bottom = page_rect.y1 - margin
|
||
|
||
# Sort by y0 ascending; for each block, the next one in this list
|
||
# is its nearest downward neighbour.
|
||
sorted_blocks = sorted(blocks, key=lambda b: b["bbox"][1])
|
||
for i, block in enumerate(sorted_blocks):
|
||
if i + 1 < len(sorted_blocks):
|
||
block["_next_block_y"] = sorted_blocks[i + 1]["bbox"][1] - 2
|
||
else:
|
||
block["_next_block_y"] = page_bottom
|
||
|
||
# Keep a small visual gap so we don't crowd the next block's top edge.
|
||
|
||
def _write_translated_block(
|
||
self,
|
||
page,
|
||
block: Dict,
|
||
font_path: Optional[str],
|
||
is_rtl: bool,
|
||
) -> bool:
|
||
"""Write translated text into the block's bounding box.
|
||
|
||
Track B3.5: Smart overflow handling. Returns True if the block
|
||
was rendered cleanly, False if it was skipped (format loss).
|
||
|
||
Strategy (in order, first success wins):
|
||
0. Try original rect at original font size.
|
||
1. Expand bbox to page margins horizontally, original font size.
|
||
2. Expand bbox vertically (up to MAX_VERTICAL_EXPANSION x height),
|
||
original font size.
|
||
3. Shrink font ONCE (FONT_SHRINK_FACTOR = 0.93), with expanded bbox.
|
||
4. Shrink font AGAIN (0.87 cumulative), with expanded bbox.
|
||
5. Heading: stop at HEADING_MIN_SCALE (90%). Body: BODY_MIN_SCALE (75%).
|
||
6. If still overflow: SKIP the block, emit format_loss metric, write
|
||
a small placeholder so the user knows something was lost.
|
||
|
||
Key insight: prefer LOOSING a block over DEGRADING the entire page's
|
||
font hierarchy. A title that becomes unreadable 8.5pt is worse than
|
||
a missing TOC entry.
|
||
"""
|
||
import fitz
|
||
|
||
original_rect = fitz.Rect(block["bbox"])
|
||
translated = block["translated"]
|
||
target_size = block["font_size"]
|
||
|
||
color = self._int_to_rgb(block["color"])
|
||
align = fitz.TEXT_ALIGN_RIGHT if is_rtl else fitz.TEXT_ALIGN_LEFT
|
||
|
||
# PyMuPDF bug: fontname=None raises AttributeError. Default to 'helv'.
|
||
# If a custom font file is available, use it via fontfile (fontname ignored).
|
||
fontname = "helv"
|
||
fontfile = font_path
|
||
|
||
# Determine if this is a heading (larger font size = more visual weight)
|
||
is_heading = target_size >= HEADING_MIN_SIZE
|
||
min_scale = HEADING_MIN_SCALE if is_heading else BODY_MIN_SCALE
|
||
min_size = max(target_size * min_scale, MIN_FONT_SIZE)
|
||
|
||
page_rect = page.rect
|
||
margin = 18
|
||
|
||
# Track B3.6: try a wider bbox that respects the page margin.
|
||
# For headings, also allow horizontal expansion because long
|
||
# translated titles often don't fit in the original width.
|
||
max_x1 = page_rect.x1 - margin
|
||
expanded_h = fitz.Rect(
|
||
max(original_rect.x0, page_rect.x0 + margin),
|
||
original_rect.y0,
|
||
max_x1,
|
||
original_rect.y1,
|
||
)
|
||
|
||
# Track B3.7: cap vertical expansion at the next block's y0
|
||
# (not the page bottom), so a long translated block cannot
|
||
# overflow into its neighbour.
|
||
next_block_y = block.get("_next_block_y", page_rect.y1 - margin)
|
||
max_expand_y = min(
|
||
next_block_y - original_rect.y1,
|
||
original_rect.height * MAX_VERTICAL_EXPANSION,
|
||
)
|
||
expanded_v = fitz.Rect(
|
||
expanded_h.x0,
|
||
expanded_h.y0,
|
||
expanded_h.x1,
|
||
expanded_h.y1 + max_expand_y,
|
||
)
|
||
|
||
# Tier 0: original rect, original size
|
||
# insert_textbox returns >= 0 if text fit, < 0 if overflow (in points)
|
||
rc = self._try_insert(page, original_rect, translated, target_size, fontname, fontfile, color, align)
|
||
if rc is not None and rc >= 0:
|
||
return True
|
||
|
||
# Tier 1: expanded horizontal, original size
|
||
if expanded_h.width > original_rect.width + 1:
|
||
rc = self._try_insert(page, expanded_h, translated, target_size, fontname, fontfile, color, align)
|
||
if rc is not None and rc >= 0:
|
||
return True
|
||
|
||
# Tier 2: expanded vertical, original size
|
||
if expanded_v.height > original_rect.height + 1:
|
||
rc = self._try_insert(page, expanded_v, translated, target_size, fontname, fontfile, color, align)
|
||
if rc is not None and rc >= 0:
|
||
return True
|
||
|
||
# Tier 3: shrink once
|
||
size_1 = max(target_size * FONT_SHRINK_FACTOR, min_size)
|
||
if size_1 < target_size - 0.1: # only try if shrink is meaningful
|
||
rc = self._try_insert(page, expanded_v, translated, size_1, fontname, fontfile, color, align)
|
||
if rc is not None and rc >= 0:
|
||
return True
|
||
|
||
# Tier 4: shrink twice
|
||
size_2 = max(target_size * FONT_SHRINK_FACTOR * FONT_SHRINK_FACTOR, min_size)
|
||
if size_2 < size_1 - 0.1:
|
||
rc = self._try_insert(page, expanded_v, translated, size_2, fontname, fontfile, color, align)
|
||
if rc is not None and rc >= 0:
|
||
return True
|
||
|
||
# Tier 5: hit the min size floor
|
||
rc = self._try_insert(page, expanded_v, translated, min_size, fontname, fontfile, color, align)
|
||
if rc is not None and rc >= 0:
|
||
return True
|
||
|
||
# Tier 6: graceful failure — skip the block, log it
|
||
self._skip_block_with_marker(page, original_rect, translated, color)
|
||
_record_format_loss_metric("text_overflow")
|
||
logger.warning(
|
||
"pdf_block_skipped_overflow",
|
||
font_size=target_size,
|
||
text_preview=translated[:60],
|
||
)
|
||
return False
|
||
|
||
def _skip_block_with_marker(self, page, rect, original_text, color):
|
||
"""Mark a skipped block with a visible placeholder so the user
|
||
knows something was lost. Uses a tiny font + grey color."""
|
||
import fitz
|
||
placeholder = "[translation overflow]"
|
||
try:
|
||
page.insert_textbox(
|
||
rect,
|
||
placeholder,
|
||
fontsize=6.0,
|
||
fontname="helv",
|
||
color=(0.6, 0.6, 0.6),
|
||
align=fitz.TEXT_ALIGN_LEFT,
|
||
overlay=True,
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
def _try_insert(
|
||
self, page, rect, text, fontsize, fontname, fontfile, color, align
|
||
):
|
||
"""Attempt insert_textbox, returns rc or None on error."""
|
||
try:
|
||
return page.insert_textbox(
|
||
rect,
|
||
text,
|
||
fontsize=fontsize,
|
||
fontname=fontname,
|
||
fontfile=fontfile,
|
||
color=color,
|
||
align=align,
|
||
overlay=True,
|
||
)
|
||
except Exception as e:
|
||
# Demote this from warning to debug — insert_textbox is called
|
||
# 6+ times per block (one per tier), and these errors are common
|
||
# in normal operation (e.g. when the text really doesn't fit).
|
||
logger.debug(
|
||
"pdf_insert_textbox_error",
|
||
error=str(e)[:200],
|
||
error_type=type(e).__name__,
|
||
fontsize=fontsize,
|
||
)
|
||
return None
|
||
|
||
@staticmethod
|
||
def _int_to_rgb(color_int: int) -> tuple:
|
||
"""Convert integer color (0xRRGGBB) to (r, g, b) float tuple."""
|
||
r = ((color_int >> 16) & 0xFF) / 255.0
|
||
g = ((color_int >> 8) & 0xFF) / 255.0
|
||
b = (color_int & 0xFF) / 255.0
|
||
return (r, g, b)
|
||
|
||
# ------------------------------------------------------------------ #
|
||
# FALLBACK — pdf2docx → WordTranslator → LibreOffice
|
||
# ------------------------------------------------------------------ #
|
||
|
||
def _translate_preserve_layout_fallback(
|
||
self,
|
||
input_path: Path,
|
||
output_path: Path,
|
||
target_language: str,
|
||
source_language: str,
|
||
progress_callback,
|
||
translate_images: bool = False,
|
||
) -> Path:
|
||
"""Fallback: PDF → DOCX (pdf2docx) → WordTranslator → PDF (LibreOffice)."""
|
||
start_time = time.time()
|
||
|
||
try:
|
||
if progress_callback:
|
||
progress_callback({
|
||
"current": 1, "total": 3,
|
||
"phase": "converting",
|
||
"paragraph": 1, "total_paragraphs": 3,
|
||
})
|
||
|
||
docx_path = self._convert_pdf_to_docx(input_path)
|
||
|
||
if progress_callback:
|
||
progress_callback({
|
||
"current": 2, "total": 3,
|
||
"phase": "translating",
|
||
"paragraph": 2, "total_paragraphs": 3,
|
||
})
|
||
|
||
from translators.word_translator import WordTranslator
|
||
|
||
translated_docx = output_path.with_suffix(".docx")
|
||
wt = WordTranslator(provider=self._provider)
|
||
wt.translate_file(
|
||
docx_path, translated_docx,
|
||
target_language, source_language,
|
||
progress_callback=None,
|
||
translate_images=translate_images,
|
||
)
|
||
|
||
if progress_callback:
|
||
progress_callback({
|
||
"current": 3, "total": 3,
|
||
"phase": "converting_back",
|
||
"paragraph": 3, "total_paragraphs": 3,
|
||
})
|
||
|
||
final_path = self._convert_docx_to_pdf(translated_docx, output_path)
|
||
|
||
for tmp in [docx_path, translated_docx]:
|
||
if tmp.exists() and tmp != final_path:
|
||
try:
|
||
tmp.unlink()
|
||
except Exception:
|
||
pass
|
||
|
||
processing_time_ms = round((time.time() - start_time) * 1000, 2)
|
||
logger.info(
|
||
"pdf_layout_fallback_success",
|
||
file_name=input_path.name,
|
||
processing_time_ms=processing_time_ms,
|
||
output=str(final_path),
|
||
)
|
||
return final_path
|
||
|
||
except Exception as e:
|
||
logger.error("pdf_layout_fallback_error", file=str(input_path), error=str(e))
|
||
raise
|
||
|
||
def _convert_pdf_to_docx(self, pdf_path: Path) -> Path:
|
||
"""Convert PDF to DOCX using pdf2docx."""
|
||
try:
|
||
from pdf2docx import Converter
|
||
except ImportError:
|
||
raise RuntimeError("pdf2docx is not installed")
|
||
|
||
docx_path = pdf_path.with_suffix(".docx")
|
||
cv = Converter(str(pdf_path))
|
||
try:
|
||
cv.convert(str(docx_path))
|
||
finally:
|
||
cv.close()
|
||
|
||
if not docx_path.exists() or docx_path.stat().st_size == 0:
|
||
raise RuntimeError("PDF conversion produced empty output")
|
||
|
||
from docx import Document
|
||
doc = Document(str(docx_path))
|
||
total_text = "".join(p.text for p in doc.paragraphs).strip()
|
||
if not total_text:
|
||
raise RuntimeError("PDF appears to be scanned or contains only images")
|
||
|
||
logger.info("pdf_converted_to_docx", pages=len(doc.paragraphs))
|
||
return docx_path
|
||
|
||
def _convert_docx_to_pdf(self, docx_path: Path, target_pdf: Path) -> Path:
|
||
"""Convert DOCX → PDF using LibreOffice headless."""
|
||
if not _libreoffice_available():
|
||
# Cache miss → known-missing path
|
||
logger.warning(
|
||
"libreoffice_not_found",
|
||
hint="Install libreoffice-core on the host for DOCX→PDF fallback",
|
||
)
|
||
return None # type: ignore[return-value]
|
||
try:
|
||
result = subprocess.run(
|
||
[
|
||
"libreoffice", "--headless", "--convert-to", "pdf",
|
||
"--outdir", str(target_pdf.parent),
|
||
str(docx_path),
|
||
],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=120,
|
||
)
|
||
expected_pdf = docx_path.with_suffix(".pdf")
|
||
if expected_pdf.exists() and expected_pdf.stat().st_size > 0:
|
||
if expected_pdf != target_pdf:
|
||
shutil.move(str(expected_pdf), str(target_pdf))
|
||
logger.info("docx_to_pdf_success")
|
||
return target_pdf
|
||
logger.warning("docx_to_pdf_no_output", stderr=result.stderr[:200])
|
||
except FileNotFoundError:
|
||
logger.warning("libreoffice_not_found_runtime")
|
||
except subprocess.TimeoutExpired:
|
||
logger.warning("libreoffice_timeout")
|
||
except Exception as e:
|
||
logger.warning("docx_to_pdf_failed", error=str(e)[:200])
|
||
|
||
docx_output = target_pdf.with_suffix(".docx")
|
||
if docx_path != docx_output and docx_path.exists():
|
||
shutil.move(str(docx_path), str(docx_output))
|
||
return docx_output
|
||
|
||
# ------------------------------------------------------------------ #
|
||
# MODE: text_only — extract text, translate, clean PDF output
|
||
# ------------------------------------------------------------------ #
|
||
|
||
def _translate_text_only(
|
||
self,
|
||
input_path: Path,
|
||
output_path: Path,
|
||
target_language: str,
|
||
source_language: str,
|
||
progress_callback,
|
||
) -> Path:
|
||
"""Extract text from PDF, translate, output as a clean formatted PDF."""
|
||
import fitz
|
||
|
||
start_time = time.time()
|
||
doc = fitz.open(str(input_path))
|
||
total_pages = len(doc)
|
||
|
||
if total_pages == 0:
|
||
doc.close()
|
||
raise RuntimeError("PDF has no pages.")
|
||
|
||
logger.info("pdf_text_only_start", pages=total_pages, file=input_path.name)
|
||
|
||
pages_text = []
|
||
for page_num in range(total_pages):
|
||
page = doc[page_num]
|
||
text = page.get_text("text").strip()
|
||
pages_text.append(text)
|
||
doc.close()
|
||
|
||
non_empty_indices = [i for i, t in enumerate(pages_text) if t]
|
||
|
||
if progress_callback:
|
||
progress_callback({
|
||
"current": 1, "total": 3,
|
||
"phase": "translating",
|
||
"paragraph": 1, "total_paragraphs": 3,
|
||
})
|
||
|
||
translated_pages = list(pages_text)
|
||
|
||
for seq, page_idx in enumerate(non_empty_indices):
|
||
text = pages_text[page_idx]
|
||
if not text.strip():
|
||
continue
|
||
try:
|
||
translated = self._translate_single(text, target_language, source_language)
|
||
if translated and translated.strip():
|
||
translated_pages[page_idx] = translated
|
||
else:
|
||
logger.warning("page_translation_empty", page=page_idx + 1)
|
||
except Exception as e:
|
||
logger.warning("page_translation_failed", page=page_idx + 1, error=str(e))
|
||
|
||
if progress_callback:
|
||
pct = int(30 + 60 * (seq + 1) / len(non_empty_indices))
|
||
progress_callback({
|
||
"current": seq + 1,
|
||
"total": len(non_empty_indices),
|
||
"phase": f"Translating page {page_idx + 1}/{total_pages}",
|
||
"paragraph": seq + 1,
|
||
"total_paragraphs": len(non_empty_indices),
|
||
"progress_override": pct,
|
||
})
|
||
|
||
final_path = output_path.with_suffix(".pdf")
|
||
self._generate_clean_pdf(translated_pages, final_path, target_language)
|
||
|
||
processing_time_ms = round((time.time() - start_time) * 1000, 2)
|
||
logger.info(
|
||
"pdf_text_only_success",
|
||
file_name=input_path.name,
|
||
pages=total_pages,
|
||
processing_time_ms=processing_time_ms,
|
||
)
|
||
|
||
return final_path
|
||
|
||
def _generate_clean_pdf(
|
||
self, pages_text: List[str], output_path: Path, target_language: str = "en"
|
||
) -> None:
|
||
"""Generate a clean, well-formatted PDF from translated page texts."""
|
||
from reportlab.lib.pagesizes import A4
|
||
from reportlab.lib.styles import ParagraphStyle
|
||
from reportlab.lib.units import mm
|
||
from reportlab.lib.enums import TA_LEFT, TA_JUSTIFY, TA_RIGHT
|
||
from reportlab.lib import colors
|
||
from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, PageBreak
|
||
from reportlab.lib.styles import getSampleStyleSheet
|
||
|
||
is_rtl = target_language.lower() in RTL_LANGUAGES
|
||
alignment = TA_RIGHT if is_rtl else TA_JUSTIFY
|
||
|
||
styles = getSampleStyleSheet()
|
||
|
||
pdf_doc = SimpleDocTemplate(
|
||
str(output_path),
|
||
pagesize=A4,
|
||
leftMargin=25 * mm,
|
||
rightMargin=25 * mm,
|
||
topMargin=25 * mm,
|
||
bottomMargin=25 * mm,
|
||
)
|
||
|
||
body_style = ParagraphStyle(
|
||
"BodyText_Custom",
|
||
parent=styles["Normal"],
|
||
fontSize=11,
|
||
leading=16,
|
||
spaceAfter=6,
|
||
alignment=alignment,
|
||
textColor=colors.HexColor("#1a1a1a"),
|
||
)
|
||
|
||
page_number_style = ParagraphStyle(
|
||
"PageNumber",
|
||
parent=styles["Normal"],
|
||
fontSize=9,
|
||
textColor=colors.HexColor("#999999"),
|
||
alignment=TA_LEFT,
|
||
)
|
||
|
||
elements = []
|
||
for i, page_text in enumerate(pages_text):
|
||
if not page_text.strip():
|
||
continue
|
||
|
||
if len(pages_text) > 1:
|
||
elements.append(Paragraph(f"— Page {i + 1} —", page_number_style))
|
||
elements.append(Spacer(1, 8))
|
||
|
||
for para_text in page_text.split("\n"):
|
||
para_text = para_text.strip()
|
||
if not para_text:
|
||
elements.append(Spacer(1, 4))
|
||
continue
|
||
|
||
safe = (
|
||
para_text
|
||
.replace("&", "&")
|
||
.replace("<", "<")
|
||
.replace(">", ">")
|
||
)
|
||
|
||
try:
|
||
elements.append(Paragraph(safe, body_style))
|
||
except Exception:
|
||
elements.append(
|
||
Paragraph(
|
||
para_text.encode("ascii", "replace").decode(),
|
||
body_style,
|
||
)
|
||
)
|
||
|
||
if i < len(pages_text) - 1:
|
||
elements.append(PageBreak())
|
||
|
||
if not elements:
|
||
raise RuntimeError("No text content to generate PDF")
|
||
|
||
pdf_doc.build(elements)
|
||
|
||
# ------------------------------------------------------------------ #
|
||
# Shared helpers
|
||
# ------------------------------------------------------------------ #
|
||
|
||
def _translate_with_provider(
|
||
self, texts: List[str], target_language: str, source_language: str
|
||
) -> List[str]:
|
||
"""Translate using the TranslationProvider interface (handles old & new styles)."""
|
||
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)
|
||
|
||
# Fallback: keep original text for any empty/failed result
|
||
return [
|
||
t if (t and t.strip()) else orig
|
||
for t, orig in zip(translated, texts)
|
||
]
|
||
|
||
def _translate_single(
|
||
self, text: str, target_language: str, source_language: str
|
||
) -> str:
|
||
"""Translate a single text string."""
|
||
if self._provider is not None:
|
||
try:
|
||
results = self._translate_with_provider([text], target_language, source_language)
|
||
if results and results[0].strip():
|
||
return results[0]
|
||
except Exception as e:
|
||
logger.warning("provider_single_failed", error=str(e))
|
||
|
||
from services.translation_service import translation_service
|
||
try:
|
||
return translation_service.translate_text(text, target_language, source_language)
|
||
except Exception as e:
|
||
logger.warning("legacy_single_failed", error=str(e))
|
||
return text
|
||
|
||
def _translate_batch(
|
||
self, texts: List[str], target_language: str, source_language: str
|
||
) -> List[str]:
|
||
"""Translate a batch of texts."""
|
||
non_empty = [t for t in texts if t and t.strip()]
|
||
self._translation_stats["attempted"] += len(non_empty)
|
||
|
||
translated = None
|
||
if self._provider is not None:
|
||
try:
|
||
translated = self._translate_with_provider(texts, target_language, source_language)
|
||
except Exception as e:
|
||
logger.warning("provider_translate_failed", error=str(e))
|
||
|
||
if translated is None:
|
||
from services.translation_service import translation_service
|
||
try:
|
||
translated = translation_service.translate_batch(texts, target_language, source_language)
|
||
except Exception as e:
|
||
logger.warning("legacy_translate_failed", error=str(e))
|
||
translated = texts
|
||
|
||
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 _validate_file(self, file_path: Path) -> None:
|
||
if not file_path.exists():
|
||
raise FileNotFoundError(f"File not found: {file_path.name}")
|
||
if file_path.suffix.lower() != ".pdf":
|
||
raise ValueError(f"Expected .pdf file, got {file_path.suffix}")
|
||
with open(file_path, "rb") as f:
|
||
header = f.read(5)
|
||
if header[:4] != b"%PDF":
|
||
raise ValueError("File does not appear to be a valid PDF.")
|
||
|
||
|
||
pdf_translator = PDFTranslator()
|