"""App-owned adapters: capture upstream paragraph identity without editing packages."""

import inspect

import ast

import textwrap

import json

import math

import os

import re

import sys

import threading

from time import perf_counter

# Set before native imports, including multiprocessing capture workers.

os.environ["ORT_DISABLE_TELEMETRY"] = "1"

STARTED = perf_counter()

from pathlib import Path

import pymupdf

PROGRESS_PREFIX = "PDFMATH_PROGRESS:"

MAX_PROGRESS_STAGE_LENGTH = 160

_progress_output_broken = False

def emit_kernel_progress(stage, completed, total, percent):

"""Write one bounded progress record without affecting translation."""

global _progress_output_broken

if _progress_output_broken or not isinstance(stage, str):

return

stage = re.sub(r"[\r\n]+", " ", stage).strip()[:MAX_PROGRESS_STAGE_LENGTH]

if not stage:

return

values = (completed, total) if percent is None else (completed, total, percent)

if any(isinstance(value, bool) or not isinstance(value, (int, float)) for value in values):

return

if any(not math.isfinite(float(value)) for value in values):

return

if (

total <= 0

or completed < 0

or completed > total

or (percent is not None and (percent < 0 or percent > 100))

):

return

payload = {

"stage": stage,

"completed": completed,

"total": total,

"percent": percent,

}

try:

print(

PROGRESS_PREFIX + json.dumps(payload, ensure_ascii=False, separators=(",", ":")),

flush=True,

)

except (BrokenPipeError, OSError):

# A closed parent stdout must not change the translation result.

_progress_output_broken = True

def emit_fast_progress(progress):

"""Forward Fast's stage-local tqdm callback state."""

completed = getattr(progress, "n", None)

total = getattr(progress, "total", None)

if not isinstance(completed, (int, float)) or not isinstance(total, (int, float)):

return

if total <= 0:

return

stage = getattr(progress, "desc", None) or "Translate pages"

if stage == "Translate pages":

emit_kernel_progress("Analyze page", completed, total, None)

else:

emit_kernel_progress(stage, completed, total, completed * 100 / total)

def fast_progress_map(executor, worker, paragraphs):

"""Count completed native paragraph tasks, including ordinary Fast mode."""

paragraphs = list(paragraphs)

total = len(paragraphs)

if not total:

return executor.map(worker, paragraphs)

completed = 0

lock = threading.Lock()

emit_kernel_progress("Translate paragraphs", 0, total, 0)

def counted(paragraph):

nonlocal completed

result = worker(paragraph)

with lock:

completed += 1

emit_kernel_progress("Translate paragraphs", completed, total, completed * 100 / total)

return result

return executor.map(counted, paragraphs)

def fast_combine_paragraph_text(texts):

"""Join source fragments while preserving word and CJK boundaries."""

cjk = re.compile(r"[\u3400-\u4dbf\u4e00-\u9fff\u3040-\u30ff\uac00-\ud7af]")

result = ""

for value in texts:

next_text = str(value or "").strip()

if not next_text:

continue

if not result:

result = next_text

continue

if (

len(result) > 1

and result[-1] in "-\u00ad"

and result[-2].isalpha()

and next_text[0].isalpha()

):

result = result[:-1] + next_text

elif cjk.search(result[-1]) and cjk.search(next_text[0]):

result += next_text

else:

result += " " + next_text

return result

def fast_paragraph_continuation_candidate(before_text, after_text, before, after):

"""Bound Jev requests to adjacent prose with compatible page geometry."""

if not isinstance(before_text, str) or not isinstance(after_text, str):

return False

before_text, after_text = before_text.strip(), after_text.strip()

if not before_text or not after_text:

return False

if len(before_text) > 6000 or len(after_text) > 6000:

return False

formula = re.compile(r"^\{+\s*v\d+\s*\}+$", re.IGNORECASE)

if formula.fullmatch(before_text) or formula.fullmatch(after_text):

return False

barrier = re.compile(

r"^\s*(?:[•●▪◦]|\d+[.)、]\s|[（(]\d+[）)]\s|fig(?:ure)?\.?\s*\d|table\s*\d|图\s*\d|表\s*\d)",

re.IGNORECASE,

)

if barrier.search(before_text) or barrier.search(after_text):

return False

heading = re.compile(

r"^(?:abstract|introduction|keywords|references|conclusions?|discussion|results|methods|acknowledg(?:e)?ments|摘要|关键词|参考文献|结论)\s*[:：]?$",

re.IGNORECASE,

)

if heading.fullmatch(before_text) or heading.fullmatch(after_text):

return False

fields = ("x0", "x1", "y0", "y1", "size")

values = [getattr(item, field, None) for item in (before, after) for field in fields]

if any(isinstance(value, bool) or not isinstance(value, (int, float)) for value in values):

return False

if any(not math.isfinite(float(value)) for value in values):

return False

before_size, after_size = float(before.size), float(after.size)

size = max(1.0, before_size, after_size)

if abs(before_size - after_size) > max(1.0, size * 0.12):

return False

# pdfminer uses a bottom-left page frame. A same-line split is adjacent in x;

# a wrapped continuation is below it and starts in the same column.

y_overlap = min(float(before.y1), float(after.y1)) - max(float(before.y0), float(after.y0))

x_gap = float(after.x0) - float(before.x1)

same_line_height = abs(float(after.y1) - (float(before.y0) + size))

if y_overlap >= size * 0.5 and same_line_height <= size * 0.5 and -size <= x_gap <= size * 1.5:

return True

vertical_gap = float(before.y0) - float(after.y1)

aligned = abs(float(after.x0) - float(before.x0)) <= size * 4

return -size * 0.25 <= vertical_gap <= size * 3 and aligned

def fast_classify_paragraph_continuation(before, after, url, token, layout=None, timeout=7):

"""Ask the app-owned gateway whether two bounded fragments are continuous."""

if (

not isinstance(url, str)

or not url.strip()

or not isinstance(token, str)

or not token.strip()

):

return False

if not isinstance(before, str) or not isinstance(after, str):

return False

if not before.strip() or not after.strip() or len(before) > 6000 or len(after) > 6000:

return False

import urllib.request

request = urllib.request.Request(

url,

data=json.dumps(

{"before": before, "after": after, "layout": layout}, ensure_ascii=False

).encode("utf-8"),

headers={"Authorization": "Bearer " + token, "Content-Type": "application/json"},

method="POST",

)

try:

with urllib.request.urlopen(request, timeout=timeout) as response:

if getattr(response, "status", 200) != 200:

return False

value = json.loads(response.read().decode("utf-8"))

except Exception:

return False

return (

isinstance(value, dict)

and value.get("status") == "classified"

and value.get("combine") is True

)

def fast_group_paragraphs(texts, paragraphs, url, token, variables=None):

"""Resolve bounded same-page groups before Fast submits translation work."""

texts, paragraphs = list(texts), list(paragraphs)

if len(texts) != len(paragraphs) or len(texts) < 2 or not url or not token:

return texts, paragraphs

# Fast sometimes gives space glyphs their own layout class and paragraph.

# They are invisible in the captured text layer and must not hide adjacency

# between real prose fragments. Formula placeholders remain barriers.

nonempty = [(text, paragraph) for text, paragraph in zip(texts, paragraphs) if text.strip()]

if not nonempty:

return texts, paragraphs

texts, paragraphs = map(list, zip(*nonempty))

from concurrent.futures import ThreadPoolExecutor, as_completed

def judgment_text(text):

def restore(match):

index = int(match.group(1))

if variables is None or index >= len(variables):

return match.group(0)

return "".join(character.get_text() for character in variables[index])

return re.sub(r"\{+\s*v(\d+)\s*\}+", restore, text)

def layout_context(before, after):

size = max(1, float(before.size), float(after.size))

aligned = abs(float(after.x0) - float(before.x0)) <= size * 4

return {

"sameColumn": aligned,

"gapInEm": max(-0.25, min(3, (float(before.y0) - float(after.y1)) / size)),

"afterIndentInEm": max(-4, min(4, (float(after.x) - float(before.x0)) / size)),

}

candidates = {

index: (

judgment_text(texts[index]),

judgment_text(texts[index + 1]),

layout_context(paragraphs[index], paragraphs[index + 1]),

)

for index in range(len(texts) - 1)

if fast_paragraph_continuation_candidate(

texts[index], texts[index + 1], paragraphs[index], paragraphs[index + 1]

)

}

# Keep a pathological page from creating an unbounded pre-translation queue.

candidates = dict(list(candidates.items())[:256])

decisions = {}

with ThreadPoolExecutor(max_workers=4) as executor:

futures = {

executor.submit(

fast_classify_paragraph_continuation, before, after, url, token, layout

): index

for index, (before, after, layout) in candidates.items()

}

for future in as_completed(futures):

index = futures[future]

try:

decisions[index] = future.result()

except Exception:

decisions[index] = False

from copy import copy

def build(indices):

if len(indices) == 1:

return texts[indices[0]], paragraphs[indices[0]]

merged = copy(paragraphs[indices[0]])

merged.x0 = min(float(paragraphs[index].x0) for index in indices)

merged.x1 = max(float(paragraphs[index].x1) for index in indices)

merged.y0 = min(float(paragraphs[index].y0) for index in indices)

merged.y1 = max(float(paragraphs[index].y1) for index in indices)

merged.brk = any(bool(getattr(paragraphs[index], "brk", False)) for index in indices)

if not merged.brk:

merged.brk = any(

not fast_paragraph_continuation_candidate(

texts[index], texts[index + 1], paragraphs[index], paragraphs[index + 1]

)

or min(float(paragraphs[index].y1), float(paragraphs[index + 1].y1))

- max(float(paragraphs[index].y0), float(paragraphs[index + 1].y0))

< -max(1.0, float(paragraphs[index].size) * 0.25)

for index in indices[:-1]

)

merged._preview_source_indices = list(indices)

merged._preview_source_texts = [texts[index] for index in indices]

return fast_combine_paragraph_text(merged._preview_source_texts), merged

grouped_texts, grouped_paragraphs = [], []

current = [0]

for index in range(1, len(texts)):

candidate = decisions.get(index - 1, False)

combined = fast_combine_paragraph_text([texts[item] for item in (*current, index)])

if (

candidate

and len(current) < 16

and len(combined) <= 20000

and fast_paragraph_continuation_candidate(

texts[index - 1], texts[index], paragraphs[index - 1], paragraphs[index]

)

):

current.append(index)

else:

text, paragraph = build(current)

grouped_texts.append(text)

grouped_paragraphs.append(paragraph)

current = [index]

text, paragraph = build(current)

grouped_texts.append(text)

grouped_paragraphs.append(paragraph)

return grouped_texts, grouped_paragraphs

def emit_precise_progress(event):

"""Forward BabelDOC's native overall progress and stage counters."""

if not isinstance(event, dict):

return

emit_kernel_progress(

event.get("stage"),

event.get("stage_current"),

event.get("stage_total"),

event.get("overall_progress"),

)

# BabelDOC's OpenAI-style rich-text placeholders are <style id='N'>...</style>, and its

# parser only accepts that literal tag with single quotes. LLMs sometimes translate the

# tag name (<样式 id='1'>) or requote it. Only tags that carry a numeric id are treated as

# placeholders; their closing tags are matched by the same name.

PRECISE_RICH_TEXT_LEFT = re.compile(

r"<\s*([^\s<>/='\"]+)\s+id\s*=\s*(['\"]?)\s*(\d+)\s*\2\s*>", re.IGNORECASE

)

def precise_canonical_rich_text_tags(text):

"""Rewrite placeholder tags to the form BabelDOC's parse_translate_output matches."""

if not isinstance(text, str):

return text

# BabelDOC reads the inner {vN} of {{vN}} and would leave the outer braces as prose.

if "{" in text:

text = re.sub(r"\{+(\s*v\s*\d+\s*)\}+", r"{\1}", text, flags=re.IGNORECASE)

if "<" not in text:

return text

names = {match.group(1) for match in PRECISE_RICH_TEXT_LEFT.finditer(text)}

if not names:

return text

text = PRECISE_RICH_TEXT_LEFT.sub(lambda match: f"<style id='{match.group(3)}'>", text)

closing = "|".join(re.escape(name) for name in sorted(names, key=len, reverse=True))

return re.sub(rf"<\s*/\s*(?:{closing})\s*>", "</style>", text, flags=re.IGNORECASE)

def precise_composition_text(obj):

"""Plain text of a BabelDOC paragraph composition: styled runs and formulas inline."""

if obj is None:

return ""

for name in ("pdf_line", "pdf_formula", "pdf_same_style_characters"):

value = getattr(obj, name, None)

if value is not None:

return "".join(

precise_composition_text(c) for c in getattr(value, "pdf_character", []) or []

)

for name in ("pdf_character", "pdf_same_style_unicode_characters"):

value = getattr(obj, name, None)

if value is not None:

return precise_composition_text(value)

# PdfCharacter stores char_unicode; PdfSameStyleUnicodeCharacters stores unicode.

return getattr(obj, "char_unicode", None) or getattr(obj, "unicode", None) or ""

# pdf2zh's converter reads formulas with this pattern and skips IDs it does not own.

FAST_FORMULA_PLACEHOLDER = re.compile(r"\{\s*v([\d\s]+)\}", re.IGNORECASE)

def fast_formula_placeholders(text, source):

"""Give pdf2zh's typesetter only formula placeholder shapes its parser reads."""

if not isinstance(text, str) or "{" not in text:

return text

# {{v0}} would otherwise typeset its outer braces as prose.

text = re.sub(r"\{+(\s*v[\d\s]+)\}+", r"{\1}", text, flags=re.IGNORECASE)

# {v*} is prompt notation, not an upstream formula ID. Recover only when the

# source paragraph identifies exactly one original run; otherwise keep the source.

wildcard = re.compile(r"\{+\s*v\s*\*\s*\}+", re.IGNORECASE)

if wildcard.search(text):

ids = {

int(match.replace(" ", ""))

for match in FAST_FORMULA_PLACEHOLDER.findall(source or "")

if match.strip()

}

if len(ids) != 1:

return source

text = wildcard.sub(lambda _: "{v%d}" % next(iter(ids)), text)

return text

def fast_paragraph_indent(text, x, x0, language):

"""Replace inherited source indentation with two CJK full-width spaces."""

# Protected tables and formulas are emitted as one placeholder run. Their

# first glyph is not a prose indent: moving it moves every retained glyph

# and rule in the run, which can push the whole table beyond the page edge.

if not re.sub(r"\{+\s*v\d+\s*\}+", "", text).strip():

return text, x

if language.lower().split("-")[0] in ("zh", "ja", "ko") and x > x0 + 0.1:

return "\u3000\u3000" + text.lstrip(" \t\u3000"), x0

return text, x

def fast_typography_advance(converter, text, ptr, advance, size, language):

"""Reserve trailing CJK punctuation together with the preceding character."""

if language.lower().split("-")[0] != "zh":

return advance

end = ptr

while end < len(text) and text[end] in "，。！？；：、）】》”’":

end += 1

if end > ptr:

advance += sum(converter.noto.char_lengths(text[ptr:end], size))

return advance

def fast_italic_prose_characters(page):

"""Recognize ordinary italic prose without relaxing math-font detection."""

protected = set()

run = []

def finish():

text = "".join(char.get_text() for char in run)

words = re.findall(r"[A-Za-z]+", text)

if (

re.fullmatch(r"[A-Za-z\s.,:'’()\-]+", text)

and words

and max(map(len, words)) >= 4

and (len(words) >= 2 or max(map(len, words)) >= 5)

):

protected.update(id(char) for char in run)

run.clear()

for char in page:

font = getattr(char, "fontname", "")

if isinstance(font, bytes):

font = font.decode("utf-8", errors="ignore")

font = font.split("+")[-1]

# CM/TeX, symbol, code and explicitly configured formula fonts stay intact.

if not re.match(

r"(?:Times|Helvetica|Arial|Georgia|Palatino|Minion|Garamond|Liberation|Nimbus).*"

r"(?:Italic|Oblique)",

font,

re.IGNORECASE,

):

finish()

continue

if run and (

char.fontname != run[-1].fontname

or abs(char.y0 - run[-1].y0) > max(1, char.size * 0.2)

or char.x0 - run[-1].x1 > char.size

or char.x0 < run[-1].x0

):

finish()

run.append(char)

finish()

return protected

def fast_resolve_text_overlaps(context):

"""Try moving translated lines below retained runs, inside the paragraph box."""

from pdfminer.pdffont import PDFCIDFont

converter = context["self"]

paragraph = context["pstk"][context["id"]]

size, leading = context["size"], context["line_height"]

step = size * leading

translated_fonts = {"tiro", converter.noto_name}

items = [item for item in context["ops_vals"] if item.get("type").value == "text"]

def bounds(item):

font = item["font"]

encoded = item["rtxt"]

if font == converter.noto_name:

width = len(encoded) / 4 * item["size"]

else:

pdf_font = converter.fontmap[font]

stride = 4 if isinstance(pdf_font, PDFCIDFont) else 2

width = (

sum(

pdf_font.char_width(int(encoded[i : i + stride], 16))

for i in range(0, len(encoded), stride)

)

* item["size"]

)

baseline = context["y"] + item["dy"] - item["lidx"] * step

return (

item["x"],

baseline - item["size"] * 0.2,

item["x"] + width,

baseline + item["size"] * 0.8,

)

retained = [bounds(item) for item in items if item["font"] not in translated_fonts]

moved = 0

for line in sorted({item["lidx"] for item in items if item["font"] in translated_fonts}):

current = [

item for item in items if item["font"] in translated_fonts and item["lidx"] == line

]

for _ in range(3):

collision = any(

min(a[2], b[2]) - max(a[0], b[0]) > size * 0.15

and min(a[3], b[3]) - max(a[1], b[1]) > size * 0.15

for a in map(bounds, current)

for b in retained

)

if not collision:

break

following = [

item for item in items if item["font"] in translated_fonts and item["lidx"] >= line

]

if min(bounds(item)[1] for item in following) - step < paragraph.y0:

break

for item in following:

item["dy"] -= step

moved += 1

return moved

def layout_box(values, source_page, kind):

# Fast's interpreter applies its CropBox/rotation CTM before layout.

# Its coordinates are already in the visible page's bottom-left frame.

# Applying the raw PDF transform again subtracts the CropBox twice.

if kind == "pdf_math_fast":

matrix = pymupdf.Matrix(1, 0, 0, -1, 0, source_page.rect.height)

else:

matrix = source_page.transformation_matrix * source_page.rotation_matrix

rect = pymupdf.Rect(*values) * matrix

return {"x": rect.x0, "y": rect.y0, "width": max(1, rect.width), "height": max(1, rect.height)}

def install_glossary(kind, entries):

"""Enforce terms at the translator boundary for every service."""

if not entries:

return

if kind == "pdf_math_fast":

from pdf2zh.translator import BaseTranslator

else:

from pdf2zh_next.translator.base_translator import BaseTranslator

targets = {entry["source"]: entry["target"] for entry in entries}

pattern = re.compile(

"|".join(re.escape(source) for source in sorted(targets, key=len, reverse=True))

)

def wrap(original, structured=False):

def translate(self, text, *args, **kwargs):

prefix = "PMRGLOSSARY"

while prefix in text:

prefix += "X"

replacements = []

def protect(match):

token = f"{prefix}{len(replacements)}END"

replacements.append((token, targets[match.group()]))

return token

protected = pattern.sub(protect, text)

if replacements:

if args:

args = (True, *args[1:])

else:

kwargs["ignore_cache"] = True

output = original(self, protected, *args, **kwargs)

for token, target in replacements:

if token not in output:

raise ValueError(

"Translation did not preserve a required glossary term. Retry translation."

)

# LLM-only output is JSON; escape inserted values to keep it valid.

replacement = json.dumps(target, ensure_ascii=False)[1:-1] if structured else target

output = output.replace(token, replacement)

return output

return translate

def patch_overrides(cls):

for child in cls.__subclasses__():

if "translate" in child.__dict__:

child.translate = wrap(child.translate)

if "llm_translate" in child.__dict__:

child.llm_translate = wrap(child.llm_translate, structured=True)

patch_overrides(child)

patch_overrides(BaseTranslator)

BaseTranslator.translate = wrap(BaseTranslator.translate)

if hasattr(BaseTranslator, "llm_translate"):

BaseTranslator.llm_translate = wrap(BaseTranslator.llm_translate, structured=True)

def configure_gpu_inference():

import onnxruntime as ort

available = ort.get_available_providers()

preferred = [

name

for name in ("DmlExecutionProvider", "CoreMLExecutionProvider", "CUDAExecutionProvider")

if name in available

]

providers = [*preferred, "CPUExecutionProvider"]

original = ort.InferenceSession.__init__

def initialize(

self, path_or_bytes, sess_options=None, providers=None, provider_options=None, **kwargs

):

selected = [*preferred, "CPUExecutionProvider"]

if sess_options is None:

sess_options = ort.SessionOptions()

if "DmlExecutionProvider" in selected:

sess_options.enable_mem_pattern = False

sess_options.execution_mode = ort.ExecutionMode.ORT_SEQUENTIAL

# CoreML compiled nodes cannot be serialized as an optimized ONNX graph.

if "CoreMLExecutionProvider" in selected:

sess_options.optimized_model_filepath = ""

return original(self, path_or_bytes, sess_options, selected, None, **kwargs)

ort.InferenceSession.__init__ = initialize

print("GPU inference providers: " + ", ".join(providers), file=sys.stderr)

def main(capture_only=False):

if capture_only:

kind, sidecar, selected, input_path, args = json.loads(

os.environ["PREVIEW_CAPTURE_CONTEXT"]

)

else:

kind, sidecar, selected, input_path, *args = sys.argv[1:]

if "--prefer-gpu" in args:

os.environ["PDFMATHREADER_PREFER_GPU"] = "1"

args = [arg for arg in args if arg != "--prefer-gpu"]

if os.environ.get("PDFMATHREADER_PREFER_GPU") == "1":

configure_gpu_inference()

selected = int(selected)

glossary_path = os.environ.get("PDFMATHREADER_GLOSSARY_PATH")

if glossary_path:

install_glossary(kind, json.loads(Path(glossary_path).read_text(encoding="utf-8")))

if os.environ.get("PDFMATHREADER_LOCAL_TRANSLATION") == "1":

# The native API receives source text, never an LLM instruction template.

import urllib.request

def local_translate(self, text, *unused, **options):

request = urllib.request.Request(

os.environ["OPENAI_BASE_URL"] + "/translate",

data=json.dumps({"text": text}).encode(),

headers={

"Authorization": "Bearer " + os.environ["OPENAI_API_KEY"],

"Content-Type": "application/json",

},

)

with urllib.request.urlopen(request, timeout=60) as response:

return json.load(response)["translation"]

if kind == "pdf_math_fast":

from pdf2zh.translator import OpenAITranslator

else:

from pdf2zh_next.translator.translator_impl.openai import OpenAITranslator

def no_llm(self, *args, **kwargs):

raise NotImplementedError("Apple Translation is a text translation API")

OpenAITranslator.do_llm_translate = no_llm

OpenAITranslator.do_translate = local_translate

timings = {}

checkpoint = perf_counter()

def step(name):

nonlocal checkpoint

now = perf_counter()

timings[name] = timings.get(name, 0) + (now - checkpoint) * 1000

checkpoint = now

source_pdf = pymupdf.open(input_path)

source_page = source_pdf[selected - 1]

records = []

def box(values):

return layout_box(values, source_page, kind)

if kind == "pdf_math_fast":

from pdf2zh.converter import TranslateConverter

output_language = next(

(args[i + 1] for i, arg in enumerate(args[:-1]) if arg in ("-lo", "--lang-out")), ""

)

import pdf2zh.high_level as high_level

original_stream = high_level.translate_stream

original_document = high_level.Document

stream_lines, stream_start = inspect.getsourcelines(original_stream)

stream_tree = ast.parse(textwrap.dedent("".join(stream_lines)))

source_document_lines = {

stream_start + node.lineno - 1

for node in ast.walk(stream_tree)

if isinstance(node, ast.Assign)

and any(

isinstance(target, ast.Name) and target.id == "doc_en" for target in node.targets

)

and isinstance(node.value, ast.Call)

and isinstance(node.value.func, ast.Name)

and node.value.func.id == "Document"

}

class Timer:

def add(self, name, seconds):

timings[name] = timings.get(name, 0) + seconds * 1000

def step(self, name):

step(name)

def timed_stream(*positional, **kwargs):

step("model_and_cli_setup")

kwargs["perf"] = Timer()

callback = kwargs.get("callback")

def progress_callback(progress):

emit_fast_progress(progress)

if callback:

callback(progress)

# pdf2zh.high_level.translate_patch invokes this callback after

# each native page, paragraph, or typesetting update. Keep the

# caller's callback and add the app-owned stdout record beside it.

kwargs["callback"] = progress_callback

def mono_document(*args, **options):

doc = original_document(*args, **options)

caller = sys._getframe(1)

if (

caller.f_code is original_stream.__code__

and caller.f_lineno in source_document_lines

):

# doc_en is only used for the unused bilingual output after

# its working copy is saved. Keep source save/font behavior;

# avoid a second font subset and duplicate PDF serialization.

doc.insert_file = lambda *a, **k: None

doc.move_page = lambda *a, **k: None

doc.subset_fonts = lambda *a, **k: None

doc.write = lambda *a, **k: b""

return doc

high_level.Document = mono_document

try:

result = original_stream(*positional, **kwargs)

emit_kernel_progress("Finalize PDF", 0, 1, None)

return result

finally:

high_level.Document = original_document

high_level.translate_stream = timed_stream

# Retain the layout detector's explicit formula boxes. Protected table

# placeholders must not be mistaken for formula paragraphs.

from pdf2zh.doclayout import OnnxModel

original_predict = OnnxModel.predict

def formula_predict(model, image, *args, **kwargs):

results = original_predict(model, image, *args, **kwargs)

for result in results:

for detected in result.boxes:

label = result.names[int(detected.cls)]

if label not in ("isolate_formula", "formula", "equation"):

continue

x0, y0, x1, y1 = map(float, detected.xyxy.squeeze())

sx = source_page.rect.width / image.shape[1]

sy = source_page.rect.height / image.shape[0]

region = {

"x": max(0, x0 * sx),

"y": max(0, y0 * sy),

"width": (min(source_page.rect.width, x1 * sx) - max(0, x0 * sx)),

"height": (min(source_page.rect.height, y1 * sy) - max(0, y0 * sy)),

}

if region["width"] <= 0 or region["height"] <= 0:

continue

records.append(

{

"id": f"fast-formula-{selected}-{len(records)}",

"page": selected,

"text": "",

"translation": "",

"sourceBox": region,

"translatedBox": dict(region),

"fontSize": 12,

"isFormula": True,

"layoutLabel": label,

"layoutSource": "pdf2zh.doclayout",

}

)

return results

OnnxModel.predict = formula_predict

step("imports")

# OCR pages bypass TranslateConverter.receive_layout entirely. Reuse

# upstream's paragraph reconstruction and typesetting, while tagging

# each translated string so the reader receives the same sidecar

# records as native-text pages.

import pdf2zh.ocr as ocr_module

original_ocr_paragraphs = ocr_module.ocr_paragraphs

original_translate_ocr_page = ocr_module.translate_ocr_page

def capture_ocr_page(page, detection, translator, font, thread, cancellation_event=None):

paragraphs = original_ocr_paragraphs(page, detection)

if not paragraphs:

return original_translate_ocr_page(

page, detection, translator, font, thread, cancellation_event

)

translations = [None] * len(paragraphs)

class TaggedText(str):

def __new__(cls, value, index):

result = super().__new__(cls, value)

result.index = index

return result

tagged = [

(bounds, TaggedText(text, index), size, words)

for index, (bounds, text, size, words) in enumerate(paragraphs)

]

class CapturingTranslator:

def __init__(self, delegate):

self.delegate = delegate

def translate(self, text, *args, **kwargs):

output = self.delegate.translate(str(text), *args, **kwargs)

translations[text.index] = output

return output

# translate_ocr_page resolves ocr_paragraphs from its module

# globals. Return the captured list so OCR is reconstructed once.

ocr_module.ocr_paragraphs = lambda *_args, **_kwargs: tagged

try:

result = original_translate_ocr_page(

page,

detection,

CapturingTranslator(translator),

font,

thread,

cancellation_event,

)

finally:

ocr_module.ocr_paragraphs = original_ocr_paragraphs

for index, (bounds, text, size, _words) in enumerate(paragraphs):

text = str(text)

if not text.strip():

continue

region = {

"x": float(bounds.x0),

"y": float(bounds.y0),

"width": max(1, float(bounds.width)),

"height": max(1, float(bounds.height)),

}

translation = str(translations[index] or "")

records.append(

{

"id": f"fast-ocr-{selected}-{len(records)}",

"page": selected,

"text": text,

"translation": translation,

"sourceBox": region,

"translatedBox": dict(region),

"fontSize": float(size or 12),

"sourceInput": text,

"translationOutput": translation,

"layoutSource": "pdf2zh.ocr",

"translatedWidth": "source-paragraph-width",

}

)

return result

ocr_module.translate_ocr_page = capture_ocr_page

original = TranslateConverter.receive_layout

def capture(v):

overlap_adjustments = fast_resolve_text_overlaps(v)

index = v["id"]

paragraph = v["pstk"][index]

raw_source, raw_target = v["sstk"][index], v["news"][index]

variables = ["".join(c.get_text() for c in chars) for chars in v["var"]]

# Mirror the typesetter: unknown formula IDs are dropped, as upstream does.

def formula(match):

digits = match.group(1).replace(" ", "")

return variables[int(digits)] if digits and int(digits) < len(variables) else ""

def restore(text):

return FAST_FORMULA_PLACEHOLDER.sub(formula, text)

display_target, _ = fast_paragraph_indent(

fast_formula_placeholders(raw_target, raw_source),

paragraph.x,

paragraph.x0,

output_language,

)

text, translation = restore(raw_source), restore(display_target)

if text.strip():

values = [item for item in v["ops_vals"] if item.get("type").value == "text"]

low = min(

(

item["dy"]

+ v["y"]

- item["lidx"] * item["size"] * v["line_height"]

- item["size"] * 0.25

for item in values

),

default=paragraph.y0,

)

high = max(

(

item["dy"]

+ v["y"]

- item["lidx"] * item["size"] * v["line_height"]

+ item["size"]

for item in values

),

default=paragraph.y1,

)

records.append(

{

"id": f"fast-{selected}-{len(records)}",

"page": selected,

"text": text,

"translation": translation,

"sourceBox": box((paragraph.x0, paragraph.y0, paragraph.x1, paragraph.y1)),

"translatedBox": box((paragraph.x0, low, paragraph.x1, high)),

"fontSize": paragraph.size,

"sourceInput": raw_source,

"translationOutput": raw_target,

"formulaTexts": variables,

"layoutSource": "pdf2zh.converter",

"overlapAdjustments": overlap_adjustments,

"translatedWidth": "source-paragraph-width",

}

)

# Inject one capture per paragraph, instead of tracing every Python line

# during parsing, provider calls and typesetting. Upstream files stay intact.

tree = ast.parse(textwrap.dedent(inspect.getsource(original)))

class CaptureParagraph(ast.NodeTransformer):

count = 0

indent_count = 0

prose_count = 0

grouping_count = 0

def visit_FunctionDef(self, node):

self.generic_visit(node)

if node.name == "receive_layout":

node.body.insert(

0,

ast.parse(

"_preview_prose = set() if self.vfont else _preview_italic_prose(ltpage)"

).body[0],

)

worker_indices = [

index

for index, child in enumerate(node.body)

if isinstance(child, ast.FunctionDef) and child.name == "worker"

]

if len(worker_indices) == 1:

node.body.insert(

worker_indices[0] + 1,

ast.parse(

"sstk, pstk = _preview_group_paragraphs(sstk, pstk, _preview_group_url, _preview_group_token, var)"

).body[0],

)

self.grouping_count += 1

return node

def visit_Call(self, node):

self.generic_visit(node)

if (

isinstance(node.func, ast.Attribute)

and isinstance(node.func.value, ast.Name)

and node.func.value.id == "executor"

and node.func.attr == "map"

and len(node.args) == 2

and isinstance(node.args[0], ast.Name)

and node.args[0].id == "worker"

):

return ast.Call(

func=ast.Name(id="_preview_progress_map", ctx=ast.Load()),

args=[node.func.value, *node.args],

keywords=node.keywords,

)

if isinstance(node.func, ast.Name) and node.func.id == "vflag":

self.prose_count += 1

return ast.IfExp(

test=ast.parse("_preview_identity(child) in _preview_prose").body[0].value,

body=ast.Constant(value=False),