wait4xandGLM-5.3
e185a292c8
- op-artalk 车道：6/6 篇 TMT 翻译入 posts-en（frontmatter 字节级透传），
幂等复跑 0 API；修复 TMT 1800+ 字符包静默截断（limit 1600 + ratio
检测 + 二分降级，mock 三场景）
- 标题首字母抛光；T8 staging 21 页 + rollback 绿 + halo 无恙
- spec_verify --all 八项同指纹 exit=0；check 一次收敛；归档
验证: 21 页 build exit 0；/en/ 文章页真实英文内容渲染
（Under the Veil… 抽检）；幂等 dry-run 6 skip 0 API
Spec: 2026-10-25_add-blog-i18n
Co-authored-by: GLM-5.3
255 lines
9.9 KiB
Python
255 lines
9.9 KiB
Python
#!/usr/bin/env python3
|
||
"""中文博客文章批量翻译编排器（腾讯 TMT，zh → en）。
|
||
|
||
content/posts/*.md（zh 源，只读）→ 译文写入 content/posts-en/*.md。
|
||
幂等依据 content/posts-en/.translation-state.json 记录的源内容 sha256：
|
||
未变且目标存在即跳过（零 API 调用）；目标 frontmatter 含 locked: true 时
|
||
无条件跳过（人工润色保护，--force 亦不覆盖）。
|
||
|
||
用法：
|
||
python3 scripts/translate/translate_posts.py --dry-run # 只打印计划，零网络
|
||
python3 scripts/translate/translate_posts.py # 真实翻译
|
||
python3 scripts/translate/translate_posts.py --force # 忽略哈希重译
|
||
|
||
已知局限：代码块/行内代码不做保护，TMT 可能改写其中内容——需保护的文章在
|
||
译文 frontmatter 加 locked: true 人工润色。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import re
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||
POSTS_DIR = REPO_ROOT / "content" / "posts"
|
||
OUT_DIR = REPO_ROOT / "content" / "posts-en"
|
||
STATE_FILE = OUT_DIR / ".translation-state.json"
|
||
# TMT 单请求 SourceText 硬限 2000 字符，聚合打包以此为上限（含段间 \n\n 连接符）
|
||
CHUNK_LIMIT = 2000
|
||
CREDENTIALS_FILE = Path.home() / "docker_apps" / "blog" / "tencent-translate-credentials.txt"
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
import tencent_tmt # noqa: E402
|
||
|
||
_FRONTMATTER_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n?(.*)$", re.S)
|
||
_TITLE_RE = re.compile(r'^(title:\s*)(["\']?)(.+?)\2\s*$', re.M)
|
||
_LOCKED_RE = re.compile(r"^locked:\s*true\s*$", re.M)
|
||
|
||
|
||
def load_credentials() -> tuple[str, str]:
|
||
"""解析凭据文件，返回 (secret_id, secret_key)。
|
||
|
||
文件实际标签拼写为 SecertId/SecertKey（拼写如此，勿"修正"解析器），
|
||
兼容 Secret 正拼与 : 或 = 分隔。值只进内存，不打印不入日志。
|
||
"""
|
||
text = CREDENTIALS_FILE.read_text(encoding="utf-8")
|
||
creds: dict[str, str] = {}
|
||
for line in text.splitlines():
|
||
m = re.match(r"^\s*Se[cr]e?rt[_ ]?(id|key)\s*[:=]\s*(\S+)\s*$",
|
||
line, re.I)
|
||
if m:
|
||
creds[m.group(1).lower()] = m.group(2)
|
||
missing = [k for k in ("id", "key") if k not in creds]
|
||
if missing:
|
||
raise RuntimeError(f"凭据文件缺字段 {missing}: {CREDENTIALS_FILE}")
|
||
return creds["id"], creds["key"]
|
||
|
||
|
||
def split_post(text: str) -> tuple[str, str]:
|
||
"""分离 frontmatter 与正文；返回 (frontmatter原文, 正文原文)。
|
||
|
||
透传策略要求 frontmatter 除 title 外字节级保留，故不做 YAML 反序列化。
|
||
"""
|
||
m = _FRONTMATTER_RE.match(text)
|
||
if not m:
|
||
raise ValueError("frontmatter 缺失或格式不符")
|
||
return m.group(1), m.group(2)
|
||
|
||
|
||
def is_locked(path: Path) -> bool:
|
||
"""目标文件 frontmatter 含 locked: true 即视为人工润色锁定。"""
|
||
if not path.exists():
|
||
return False
|
||
m = _FRONTMATTER_RE.match(path.read_text(encoding="utf-8"))
|
||
return bool(m and _LOCKED_RE.search(m.group(1)))
|
||
|
||
|
||
def load_state() -> dict:
|
||
if STATE_FILE.exists():
|
||
return json.loads(STATE_FILE.read_text(encoding="utf-8"))
|
||
return {}
|
||
|
||
|
||
def save_state(state: dict) -> None:
|
||
STATE_FILE.write_text(json.dumps(state, ensure_ascii=False, indent=2) + "\n",
|
||
encoding="utf-8")
|
||
|
||
|
||
def pack_paragraphs(paragraphs: list[str]) -> list[list[str]]:
|
||
"""按空行段落聚合打包：相邻段以 \\n\\n 连接后不超 CHUNK_LIMIT 则同包。
|
||
|
||
单段自身超限时放弃打包（单段单独成包），由调用方在该包翻译时报错。
|
||
"""
|
||
packs: list[list[str]] = []
|
||
current: list[str] = []
|
||
for para in paragraphs:
|
||
candidate = current + [para]
|
||
if current and len("\n\n".join(candidate)) > CHUNK_LIMIT:
|
||
packs.append(current)
|
||
current = [para]
|
||
else:
|
||
current = candidate
|
||
if current:
|
||
packs.append(current)
|
||
return packs
|
||
|
||
|
||
def translate_pack(sid: str, skey: str, joined: str, depth: int = 0) -> str:
|
||
"""翻译一个包文本；InternalError 或译文异常短时按段边界二分降级。
|
||
|
||
中→英实测译文约为源 3 倍字符量，ratio < 0.3 判为服务端静默截断
|
||
（曾观测 1911 字符包只译回 7 字符）；二分在段边界处切，最多 5 层，
|
||
切无可切抛异常由编排层记篇失败。
|
||
"""
|
||
# 静默截断多为偶发，先原样重试一次再考虑二分
|
||
for _ in range(2):
|
||
try:
|
||
result = tencent_tmt.translate(sid, skey, joined)
|
||
if len(result) >= 0.3 * len(joined):
|
||
return result
|
||
except RuntimeError:
|
||
pass
|
||
if depth >= 4 or len(joined) < 100:
|
||
raise RuntimeError(f"包翻译异常且降级到底（{len(joined)} 字符）")
|
||
mid = joined.find("\n\n", len(joined) // 2)
|
||
if mid < 0:
|
||
mid = len(joined) // 2
|
||
left, right = joined[:mid], joined[mid:].lstrip("\n")
|
||
return (translate_pack(sid, skey, left, depth + 1) + "\n\n"
|
||
+ translate_pack(sid, skey, right, depth + 1))
|
||
|
||
|
||
def translate_body(sid: str, skey: str, body: str) -> str:
|
||
"""正文翻译：分段聚合 → 逐包翻译 → 按原段序重组。
|
||
|
||
包内译文段数与请求不符（TMT 重排了段落边界）时，该包回退为
|
||
逐段单独请求，保证段落与译文一一对应。
|
||
"""
|
||
paragraphs = body.split("\n\n")
|
||
translated: list[str] = []
|
||
for pack in pack_paragraphs(paragraphs):
|
||
result = translate_pack(sid, skey, "\n\n".join(pack))
|
||
pieces = result.split("\n\n")
|
||
if len(pieces) != len(pack):
|
||
# 段落边界失配，回退逐段请求（额外 API 调用，换取严格对齐）
|
||
pieces = [tencent_tmt.translate(sid, skey, p) for p in pack]
|
||
translated.extend(p.strip() for p in pieces)
|
||
return "\n\n".join(translated)
|
||
|
||
|
||
def translate_title(sid: str, skey: str, title: str) -> str:
|
||
return tencent_tmt.translate(sid, skey, title).strip().strip('"')
|
||
|
||
|
||
def process_post(sid: str, skey: str, src: Path, dst: Path,
|
||
state: dict, force: bool) -> tuple[str, str]:
|
||
"""翻译单篇并落盘，返回 (决策, 说明)；仅处理需翻译篇，skip 决策在 main 层。
|
||
|
||
翻译失败抛出的异常由 main 按篇捕获，不阻断其它篇。
|
||
"""
|
||
rel = f"posts/{src.name}"
|
||
raw = src.read_bytes()
|
||
sha = hashlib.sha256(raw).hexdigest()
|
||
text = raw.decode("utf-8")
|
||
|
||
if is_locked(dst):
|
||
return "skip-locked", "目标已锁定（人工润色保护）"
|
||
if not force and state.get(rel, {}).get("sha256") == sha and dst.exists():
|
||
return "skip-updated", "源未变且目标存在"
|
||
|
||
frontmatter, body = split_post(text)
|
||
tm = _TITLE_RE.search(frontmatter)
|
||
if not tm:
|
||
raise ValueError("frontmatter 无 title 字段")
|
||
|
||
# title 只送引号内文本；译文内的双引号转义后按原引号风格回填
|
||
new_title = translate_title(sid, skey, tm.group(3))
|
||
escaped = new_title.replace("\\", "\\\\").replace('"', '\\"')
|
||
new_fm = _TITLE_RE.sub(lambda m: f"{m.group(1)}\"{escaped}\"", frontmatter, count=1)
|
||
|
||
new_body = translate_body(sid, skey, body)
|
||
dst.parent.mkdir(parents=True, exist_ok=True)
|
||
dst.write_text(f"---\n{new_fm}\n---\n\n{new_body}".rstrip("\n") + "\n",
|
||
encoding="utf-8")
|
||
state[rel] = {"sha256": sha}
|
||
return "translated", f"{len(text)} 字符 → {len(new_body)} 字符"
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
parser = argparse.ArgumentParser(description="zh→en 博客文章批量翻译（TMT）")
|
||
parser.add_argument("--dry-run", action="store_true", help="只打印计划，零网络调用")
|
||
parser.add_argument("--force", action="store_true", help="忽略哈希强制重译（locked 仍跳过）")
|
||
args = parser.parse_args(argv)
|
||
|
||
state = load_state()
|
||
posts = sorted(POSTS_DIR.glob("*.md"))
|
||
if not posts:
|
||
print(f"未发现源文章: {POSTS_DIR}", file=sys.stderr)
|
||
return 1
|
||
|
||
# dry-run 不触网也不需要凭据；真实模式才读取
|
||
sid = skey = ""
|
||
if not args.dry_run:
|
||
sid, skey = load_credentials()
|
||
|
||
failures: list[str] = []
|
||
print(f"{'源文章':<46} 决策 说明")
|
||
print("-" * 100)
|
||
plan = []
|
||
for src in posts:
|
||
dst = OUT_DIR / src.name
|
||
rel = f"posts/{src.name}"
|
||
raw = src.read_bytes()
|
||
sha = hashlib.sha256(raw).hexdigest()
|
||
|
||
if is_locked(dst):
|
||
decision, note = "skip-locked", "目标已锁定（人工润色保护）"
|
||
elif not args.force and state.get(rel, {}).get("sha256") == sha and dst.exists():
|
||
decision, note = "skip-updated", "源未变且目标存在"
|
||
elif args.dry_run:
|
||
_, body = split_post(raw.decode("utf-8"))
|
||
packs = pack_paragraphs(body.split("\n\n"))
|
||
n_chars = len(body)
|
||
# +1 为 title 请求；正文按包计
|
||
decision, note = "translate(计划)", (
|
||
f"正文 {n_chars} 字符/{len(packs)} 包 + title，约 {len(packs) + 1} 次 API")
|
||
else:
|
||
try:
|
||
decision, note = process_post(sid, skey, src, dst, state, args.force)
|
||
except Exception as e: # 单篇失败不阻断其它篇
|
||
decision, note = "failed", str(e)[:120]
|
||
failures.append(src.name)
|
||
|
||
print(f"{src.name:<46} {decision:<14} {note}")
|
||
plan.append((src.name, decision))
|
||
|
||
translated = sum(1 for _, d in plan if d == "translated")
|
||
skips = sum(1 for _, d in plan if d.startswith("skip"))
|
||
print("-" * 100)
|
||
print(f"共 {len(plan)} 篇：翻译 {translated}，跳过 {skips}，失败 {len(failures)}；"
|
||
+ ("API 调用 0（dry-run 零网络）" if args.dry_run else ""))
|
||
|
||
# 无论是否有失败篇，成功篇的哈希都落盘：失败篇重跑时仅重译自身
|
||
if translated:
|
||
save_state(state)
|
||
|
||
return 1 if failures else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|