v0.6-wip: Python-based Phase 4 translation pipeline

架构变更:把 dr-translator 从 opencode agent 降级为 Python 脚本编排下的 LLM
调用。根本原因是 agent 一次性处理 19k 英文词整文,单次 output token 接近
Sonnet 4.6 上限(~32k),多次重跑都卡在同一个坑里——问题是架构本身,不是
prompt。

新架构:

scripts/lib/zenmux_client.py     HTTP 客户端,指数退避重试、token 统计
                                  JSONL 日志、secrets.env 自动加载
scripts/lib/markdown_chunker.py   按 H1/H2 切块,稳定 anchor ID(order+title
                                  sha1),支持合并/统计
scripts/prompts/translate_system.txt  英译中 prompt,用自定义 <<<TRANSLATION>>>
                                       分隔符格式(规避 Markdown-in-JSON 问题)
scripts/prompts/polish_system.txt     中文润色 prompt(留给下一步 polish.py)
scripts/translate.py              主入口:章节级切块 → 逐块翻译 → 拼接

关键设计:
- 0 依赖 LLM 遵从性:Python 控制切块/循环/重试,LLM 只做单块翻译
- 断点续传:每块翻译完立即写 phase4/zh_chunks/<order>-<anchor>.md
- 术语表累积:每块的 glossary_patch 合并回 phase4/glossary.json
- 失败隔离:单块失败不影响其他块,重跑只补缺
- 调试友好:--only N,M / --limit K / --force

实测(dual-target-rnai-pipeline-2026):
- 63 块全部成功,17 分钟,$1.70
- 33,441 中文字(符合"研究类 ≥30,000 字"硬标准)
- 310 条双语术语
- 翻译质量:接近母语咨询分析师写作

下一步:polish.py(按 H2 section 润色)、merge_chapters.py(从 phase2/drafts
合并生成 final_en.md)、重构 dr-editor-in-chief 调度脚本、更新 /dr-finalize。

Co-authored-by: User <human>
This commit is contained in:
kai
2026-04-22 10:43:43 +08:00
co-authored by User <human>
parent 701bc1887e
commit 1b47b50d3c
74 changed files with 2792 additions and 3 deletions
+150
View File
@@ -0,0 +1,150 @@
"""Markdown 切块 / 合并工具。
把一篇 Markdown 按标题层级(# / ##)切成"翻译单元""润色单元"
每个单元带稳定的 ID,便于断点续传和按需重跑。
核心约定:
- H1(`# `)是一级块,通常对应 Chapter / 封面 / 前置件
- H2(`## `)是二级块,对应一个 section 或独立前置件(Disclaimer / Executive Summary / Abstract / Glossary / References
- 没有任何标题的文件头(frontmatter 区)归到第 0 块
切块粒度默认到 H2;如果某个 H2 下的正文特别长可以再按 H3 切,但这是 polish 的事情,
translate 一般不需要。
"""
from __future__ import annotations
import hashlib
import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Iterable
HEADER_RE = re.compile(r"^(#{1,6})\s+(.+?)\s*$", re.MULTILINE)
@dataclass
class MarkdownBlock:
"""一个翻译/润色单元。"""
order: int # 在原文中的顺序(0-based
level: int # 0 = frontmatter; 1/2/... = H1/H2/...
title: str # 标题原文(不含 # 号);frontmatter 为空串
anchor: str # 稳定 ID,用于断点续传(order + title hash
content: str # 完整内容(包含标题行本身,除 frontmatter 块外)
parent_order: int | None = None # H2 的父 H1 orderH1 为 None
word_count: int = 0 # 英文 word count 估算(只含 a-z
char_count: int = 0 # 字符数(含中文)
meta: dict = field(default_factory=dict)
@property
def short_title(self) -> str:
t = self.title.strip()
if len(t) <= 50:
return t
return t[:47] + "..."
def _count_words(text: str) -> int:
return len(re.findall(r"[A-Za-z]+(?:[-'][A-Za-z]+)*", text))
def _stable_anchor(order: int, title: str) -> str:
"""order + title 生成稳定短 ID。同一文件改顺序不变,改标题重算。"""
h = hashlib.sha1(title.strip().encode("utf-8")).hexdigest()[:8]
return f"b{order:03d}-{h}"
def split_by_headers(text: str, max_level: int = 2) -> list[MarkdownBlock]:
"""把 Markdown 按 H1..H{max_level} 切块。
返回的 block 按出现顺序排列。第 0 块可能是 frontmatterlevel=0,无 title)。
"""
lines = text.splitlines(keepends=False)
# 先找出所有 header 位置
header_positions: list[tuple[int, int, str]] = [] # (line_idx, level, title)
in_code_block = False
for i, ln in enumerate(lines):
if ln.lstrip().startswith("```"):
in_code_block = not in_code_block
continue
if in_code_block:
continue
m = re.match(r"^(#{1,6})\s+(.+?)\s*$", ln)
if m:
level = len(m.group(1))
if level <= max_level:
header_positions.append((i, level, m.group(2)))
blocks: list[MarkdownBlock] = []
order = 0
# frontmatter:第一个 header 前的所有内容
first_header_line = header_positions[0][0] if header_positions else len(lines)
frontmatter_text = "\n".join(lines[:first_header_line]).rstrip()
if frontmatter_text.strip():
blocks.append(
MarkdownBlock(
order=order,
level=0,
title="",
anchor=_stable_anchor(order, "__frontmatter__"),
content=frontmatter_text,
word_count=_count_words(frontmatter_text),
char_count=len(frontmatter_text),
)
)
order += 1
# 为每个 header 建一个 block:内容 = 本 header 行 + 下一 header 行前所有内容
last_h1_order: int | None = None
for idx, (line_idx, level, title) in enumerate(header_positions):
end_line = (
header_positions[idx + 1][0]
if idx + 1 < len(header_positions)
else len(lines)
)
content = "\n".join(lines[line_idx:end_line]).rstrip()
block = MarkdownBlock(
order=order,
level=level,
title=title,
anchor=_stable_anchor(order, title),
content=content,
parent_order=last_h1_order if level > 1 else None,
word_count=_count_words(content),
char_count=len(content),
)
blocks.append(block)
if level == 1:
last_h1_order = order
order += 1
return blocks
def merge_blocks(blocks: Iterable[MarkdownBlock], separator: str = "\n\n") -> str:
"""按 order 拼回完整 Markdown。"""
return separator.join(b.content for b in sorted(blocks, key=lambda x: x.order))
def chapter_stem_from_title(title: str) -> str:
"""从 `# Chapter 1 — foo bar` 生成文件名 stem 如 `ch01`。找不到就 fallback。"""
m = re.search(r"chapter\s+(\d+)", title, re.IGNORECASE)
if m:
return f"ch{int(m.group(1)):02d}"
# Executive Summary / Abstract / Glossary / Disclaimer / References / Version History
slug = re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-")
return slug[:40] or "untitled"
def count_chinese_chars(text: str) -> int:
return sum(1 for c in text if "\u4e00" <= c <= "\u9fff")
if __name__ == "__main__":
import sys
src = Path(sys.argv[1])
for b in split_by_headers(src.read_text(encoding="utf-8")):
print(f"[{b.order:3d}] L{b.level} {b.word_count:>5}w {b.char_count:>6}c {b.short_title}")