#!/usr/bin/env python3 """Translate EPUB content in /opt/epub-translate/work via Ollama gemma4:26b. Paragraph-level (Option A): natural RU, footnotes preserved, inline italics dropped. Segment-level checkpoint -> safe to interrupt and resume. Usage: python 02_translate.py --only Text/ch01.xhtml # sample one file python 02_translate.py # full run (all non-skip files) """ import argparse, json, re, sys, time, signal, warnings from pathlib import Path from bs4 import BeautifulSoup, NavigableString, Tag, XMLParsedAsHTMLWarning import ollama warnings.filterwarnings("ignore", category=XMLParsedAsHTMLWarning) ROOT = Path("/opt/epub-translate") WORK = ROOT / "work" PROGRESS = ROOT / "progress" / "checkpoint.json" MODEL = "gemma4:26b" SKIP_FILES = {"ind.xhtml", "bib.xhtml", "copy.xhtml", "cover.xhtml", "half.xhtml"} SKIP_TAGS = {"script", "style", "code", "pre", "head", "title", "meta", "link"} BLOCK_TAGS = {"p","div","h1","h2","h3","h4","h5","h6", "li","blockquote","figcaption","td","th","dd","dt"} # Inert placeholder form (less prone to hallucination than ⟦⟧ glyphs) FN_RE = re.compile(r"\[\[FN\d+\]\]") SYSTEM = ( "You are a professional literary translator. Translate the user's text from " "English into natural, idiomatic Russian that reads as a native speaker would " "write, staying faithful to the original meaning, tone and style. " "Placeholders like [[FN1]] are footnote markers: keep them EXACTLY as-is, " "unchanged, and in a natural position. Do NOT invent new [[FN]] markers. " "Keep proper names consistent. Output ONLY the Russian translation — no notes, " "no quotes, no extra commentary. If the input is only a number, date, or " "symbol, return it unchanged." ) OPTIONS = {"temperature": 1.0, "top_p": 0.95, "top_k": 64} CKPT_FLUSH_EVERY = 5 # buffer this many new blocks between disk writes # ---------------- parsing / extraction ---------------- def parse(path): return BeautifulSoup(path.read_text(encoding="utf-8", errors="replace"), "xml") def has_letters(s): return bool(s) and bool(re.search(r"[A-Za-z\u00C0-\u024F\u0400-\u04FF]", s)) def innermost_blocks(soup): """Yield block elements that contain text but no nested block child.""" for tag in soup.find_all(BLOCK_TAGS): if any(p.name in SKIP_TAGS for p in tag.parents): continue if tag.find(BLOCK_TAGS): # has nested block -> not innermost continue if has_letters(tag.get_text()): yield tag def extract_with_placeholders(block): fns, n, parts = {}, 0, [] for child in block.children: if isinstance(child, NavigableString): parts.append(str(child)) elif isinstance(child, Tag): if child.name == "sup": n += 1 key = f"[[FN{n}]]" fns[key] = str(child) parts.append(key) else: parts.append(child.get_text()) # empty -> "" -> dropped text = re.sub(r"\s+", " ", "".join(parts)).strip() return text, fns # ---------------- model ---------------- def strip_think(s): """Remove any stray gemma4 thinking/channel tags and surrounding quotes.""" s = re.sub(r"<\|?channel\|?>.*?<\|?/?channel\|?>", "", s, flags=re.S) s = re.sub(r"<\|?/?think\|?>", "", s) return s.strip().strip('"').strip() def repair_placeholders(out, expected_fns): """Strip hallucinated [[FN]] markers; append any that went missing.""" # remove placeholders the model invented out = FN_RE.sub(lambda m: m.group(0) if m.group(0) in expected_fns else "", out) # append any expected placeholder that vanished (keeps footnotes alive) for fn in expected_fns - set(FN_RE.findall(out)): out += f" {fn}" # tidy doubled spaces left by stripping return re.sub(r"\s{2,}", " ", out).strip() def normalize_brackets(out, expected_fns): """Protect valid [[FNn]] markers, kill corrupted [[FN...]] tokens, and collapse any stray [[...]] the model emitted down to single [...] (the book's real editorial-insertion convention).""" # 1. shield the legitimate footnote placeholders shield = {fn: f"\x00{i}\x00" for i, fn in enumerate(expected_fns)} for fn, tok in shield.items(): out = out.replace(fn, tok) # 2. remove corrupted footnote tokens like [[FNло2]], [[FN3a]], [[FN ]] out = re.sub(r"\[\[FN[^\]]*\]\]", "", out) # 3. collapse remaining double brackets to single (real convention is [x]) out = out.replace("[[", "[").replace("]]", "]") # 4. restore the shielded footnote markers for fn, tok in shield.items(): out = out.replace(tok, fn) return re.sub(r"\s{2,}", " ", out).strip() def translate_text(text, prev_en, prev_ru, retries=3): """Translate one paragraph. Retries only when placeholders go MISSING.""" ctx = "" if prev_en and prev_ru: ctx = (f"Previous paragraph (context only, do NOT re-translate):\n" f"EN: {prev_en}\nRU: {prev_ru}\n\n") expected_fns = set(FN_RE.findall(text)) prompt = f"{ctx}Translate this paragraph to Russian:\n{text}" for attempt in range(retries): try: r = ollama.chat( model=MODEL, messages=[{"role": "system", "content": SYSTEM}, {"role": "user", "content": prompt}], options=OPTIONS, think=False, # thinking OFF — saves huge CPU time ) out = strip_think(r["message"]["content"]) except Exception as e: wait = 5 * (attempt + 1) print(f" ! ollama error: {e} (retry in {wait}s)", flush=True) time.sleep(wait) continue got_fns = set(FN_RE.findall(out)) missing = expected_fns - got_fns # Only a MISSING placeholder is worth a retry; extras we just strip. if missing and attempt < retries - 1: print(f" ! missing {sorted(missing)} retry {attempt+1}", flush=True) continue out = repair_placeholders(out, expected_fns) out = normalize_brackets(out, expected_fns) if out: # never return empty / whitespace return out # empty output -> retry; if last attempt, fall through # total failure: return original so the build never crashes return text # ---------------- write-back ---------------- def restore_block(block, translated, fns): """Replace block contents with translated text; re-insert footnote HTML. Tolerates placeholders with no matching fns entry (drops them safely).""" block.clear() pieces = FN_RE.split(translated) keys = FN_RE.findall(translated) for i, piece in enumerate(pieces): if piece: block.append(NavigableString(piece)) if i < len(keys): html = fns.get(keys[i]) # safe lookup -> no KeyError if not html: continue # stray/hallucinated marker -> drop try: frag = BeautifulSoup(html, "xml") node = frag.find() if node: block.append(node) except Exception: # malformed stored fragment must never crash the build continue def write_xhtml_atomic(path, soup): tmp = path.with_suffix(path.suffix + ".tmp") tmp.write_text(str(soup), encoding="utf-8") tmp.replace(path) # atomic on POSIX # ---------------- checkpoint ---------------- def load_ckpt(): if PROGRESS.exists(): try: return json.loads(PROGRESS.read_text(encoding="utf-8")) except Exception: print("! checkpoint unreadable — starting fresh", flush=True) return {"done": {}} def save_ckpt(ck): PROGRESS.parent.mkdir(parents=True, exist_ok=True) tmp = PROGRESS.with_suffix(".tmp") tmp.write_text(json.dumps(ck, ensure_ascii=False), encoding="utf-8") tmp.replace(PROGRESS) # atomic # ---------------- main ---------------- def target_files(only): files = sorted(WORK.rglob("*.xhtml")) files = [f for f in files if f.name not in SKIP_FILES] if only: files = [f for f in files if str(f.relative_to(WORK)) == only or f.name == only] if not files: sys.exit(f"--only {only}: not found / skipped") return files def main(): ap = argparse.ArgumentParser() ap.add_argument("--only", help="translate one file e.g. Text/ch01.xhtml") args = ap.parse_args() if not WORK.exists(): sys.exit("work/ not found — unzip the EPUB first") ck = load_ckpt() files = target_files(args.only) print(f"Files to process: {len(files)} (resume from checkpoint enabled)\n", flush=True) # Clean Ctrl-C: flush checkpoint and exit interrupted = {"flag": False} def _sigint(_sig, _frm): interrupted["flag"] = True print("\n! interrupt received — saving checkpoint and exiting...", flush=True) signal.signal(signal.SIGINT, _sigint) signal.signal(signal.SIGTERM, _sigint) t_start = time.time() n_done = n_total = 0 out_chars = 0 since_flush = 0 dirty = False # checkpoint has unsaved entries in memory try: for f in files: if interrupted["flag"]: break rel = str(f.relative_to(WORK)) soup = parse(f) blocks = list(innermost_blocks(soup)) n_total += len(blocks) print(f"=== {rel}: {len(blocks)} blocks ===", flush=True) prev_en = prev_ru = None changed = False for idx, block in enumerate(blocks): if interrupted["flag"]: break key = f"{rel}::{idx}" text, fns = extract_with_placeholders(block) if not has_letters(text): continue if key in ck["done"]: # resume ru = ck["done"][key] else: t0 = time.time() ru = translate_text(text, prev_en, prev_ru) dt = time.time() - t0 out_chars += len(ru) ck["done"][key] = ru dirty = True since_flush += 1 n_done += 1 # buffered, crash-safe checkpoint flush if since_flush >= CKPT_FLUSH_EVERY: save_ckpt(ck) since_flush = 0 dirty = False if n_done % 10 == 0 or dt > 30: rate = out_chars / max(1e-9, time.time() - t_start) print(f" [{key}] {dt:4.1f}s ~{rate:5.1f} chars/s " f"done={n_done}", flush=True) # validate/repair placeholders even for RESUMED entries, # then write the translated text back into the DOM ru = repair_placeholders(ru, set(fns.keys())) restore_block(block, ru, fns) changed = True prev_en, prev_ru = text, ru # write each file atomically after processing its blocks if changed: write_xhtml_atomic(f, soup) print(f" -> wrote {rel}", flush=True) finally: # ALWAYS persist whatever we have (normal end, Ctrl-C, or crash) if dirty or since_flush: save_ckpt(ck) print("checkpoint saved.", flush=True) elapsed = (time.time() - t_start) / 3600 status = "INTERRUPTED" if interrupted["flag"] else "DONE" print(f"\n{status}. translated {n_done} new blocks " f"({n_total} total seen) in {elapsed:.2f} h") print(f"Checkpoint: {PROGRESS}") if __name__ == "__main__": main()