feat(i18n): T6 真实翻译入仓与 Phase 3 收敛归档
- 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
This commit is contained in:
1 parent
4aa79fca06
commit
e185a292c8
15 files changed
+1204
-15
No files matched your search
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,221 @@
|
||||
#!/usr/bin/env python3
|
||||
"""腾讯云 TMT (机器翻译) TC3-HMAC-SHA256 签名客户端。
|
||||
|
||||
零第三方依赖(hashlib/hmac/urllib),供 translate_posts.py 编排调用;
|
||||
--selftest 用官方签名文档 (cloud.tencent.com/document/api/213/30654) 公布的
|
||||
cvm 示例中间量做逐字节断言,并打印 tmt 请求的完整签名链中间量(固定假密钥,
|
||||
不触网),供与官方 API Explorer「签名串生成」交叉比对。
|
||||
|
||||
已知约束:单请求 SourceText 上限 2000 字符,分块由编排器负责。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import sys
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from datetime import datetime, timezone
|
||||
|
||||
HOST = "tmt.tencentcloudapi.com"
|
||||
SERVICE = "tmt"
|
||||
ACTION = "TextTranslate"
|
||||
VERSION = "2018-03-21"
|
||||
REGION = "ap-guangzhou" # TMT 为全球服务;该头不参与签名,仅路由用
|
||||
CT = "application/json; charset=utf-8"
|
||||
SIGNED_HEADERS = "content-type;host" # 官方要求的最低签名头集合
|
||||
SOURCE_TEXT_LIMIT = 2000
|
||||
TIMEOUT = 30
|
||||
|
||||
|
||||
def _sha256_hex(s: str) -> str:
|
||||
return hashlib.sha256(s.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def _hmac(key: bytes, msg: str) -> bytes:
|
||||
return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()
|
||||
|
||||
|
||||
def sign(secret_id: str, secret_key: str, action: str, service: str, host: str,
|
||||
payload: str, timestamp: int,
|
||||
extra_signed_headers: dict[str, str] | None = None) -> tuple[dict, dict]:
|
||||
"""按 TC3 规范签名,返回 (请求头, 中间量字典)。
|
||||
|
||||
extra_signed_headers: 除 content-type/host 外追加参与签名的头(key 小写),
|
||||
仅供 selftest 复现官方文档示例;正常调用不传。
|
||||
中间量字典含 canonical_request / string_to_sign / signature 等,
|
||||
供 selftest 展示;正常调用方可忽略。
|
||||
"""
|
||||
date = datetime.fromtimestamp(timestamp, tz=timezone.utc).strftime("%Y-%m-%d")
|
||||
# Date 必须由时间戳换算 UTC 日期:带本地时区会在跨日时段必然签名失败(官方文档陷阱)
|
||||
credential_scope = f"{date}/{service}/tc3_request"
|
||||
|
||||
signed = {"content-type": CT, "host": host}
|
||||
if extra_signed_headers:
|
||||
signed.update({k.lower(): v for k, v in extra_signed_headers.items()})
|
||||
signed_headers = ";".join(sorted(signed))
|
||||
# 头按 key 的 ASCII 升序拼接,每行 key:value\n 结尾(官方拼接规则)
|
||||
canonical_headers = "".join(f"{k}:{v}\n" for k, v in sorted(signed.items()))
|
||||
hashed_payload = _sha256_hex(payload)
|
||||
canonical_request = (
|
||||
f"POST\n/\n\n{canonical_headers}\n{signed_headers}\n{hashed_payload}"
|
||||
)
|
||||
|
||||
hashed_canonical_request = _sha256_hex(canonical_request)
|
||||
string_to_sign = (
|
||||
f"TC3-HMAC-SHA256\n{timestamp}\n{credential_scope}\n{hashed_canonical_request}"
|
||||
)
|
||||
|
||||
secret_date = _hmac(f"TC3{secret_key}".encode("utf-8"), date)
|
||||
secret_service = _hmac(secret_date, service)
|
||||
secret_signing = _hmac(secret_service, "tc3_request")
|
||||
signature = hmac.new(secret_signing, string_to_sign.encode("utf-8"),
|
||||
hashlib.sha256).hexdigest()
|
||||
|
||||
authorization = (
|
||||
f"TC3-HMAC-SHA256 Credential={secret_id}/{credential_scope}, "
|
||||
f"SignedHeaders={SIGNED_HEADERS}, Signature={signature}"
|
||||
)
|
||||
headers = {
|
||||
"Authorization": authorization,
|
||||
"Content-Type": CT,
|
||||
"Host": host,
|
||||
"X-TC-Action": action,
|
||||
"X-TC-Timestamp": str(timestamp),
|
||||
"X-TC-Version": VERSION,
|
||||
"X-TC-Region": REGION,
|
||||
}
|
||||
debug = {
|
||||
"timestamp": timestamp,
|
||||
"date": date,
|
||||
"credential_scope": credential_scope,
|
||||
"canonical_request": canonical_request,
|
||||
"hashed_request_payload": hashed_payload,
|
||||
"hashed_canonical_request": hashed_canonical_request,
|
||||
"string_to_sign": string_to_sign,
|
||||
"secret_date_hex": secret_date.hex(),
|
||||
"secret_service_hex": secret_service.hex(),
|
||||
"secret_signing_hex": secret_signing.hex(),
|
||||
"signature": signature,
|
||||
"authorization": authorization,
|
||||
}
|
||||
return headers, debug
|
||||
|
||||
|
||||
def translate(secret_id: str, secret_key: str, text: str,
|
||||
source: str = "zh", target: str = "en",
|
||||
retries: int = 3) -> str:
|
||||
"""调用 TextTranslate 并重试(指数退避 1/2/4s),返回 TargetText。
|
||||
|
||||
网络异常、HTTP 非 2xx、响应含 Response.Error 均触发重试;
|
||||
耗尽后抛 RuntimeError(调用方按篇捕获,不阻断其它篇)。
|
||||
"""
|
||||
if len(text) > SOURCE_TEXT_LIMIT:
|
||||
raise ValueError(f"SourceText 超限: {len(text)} > {SOURCE_TEXT_LIMIT}")
|
||||
|
||||
payload = json.dumps({"Source": source, "Target": target,
|
||||
"ProjectId": 0, "SourceText": text})
|
||||
last_err: Exception | None = None
|
||||
for attempt in range(retries):
|
||||
if attempt:
|
||||
time.sleep(2 ** (attempt - 1))
|
||||
try:
|
||||
headers, _ = sign(secret_id, secret_key, ACTION, SERVICE, HOST,
|
||||
payload, int(time.time()))
|
||||
req = urllib.request.Request(f"https://{HOST}/", data=payload.encode("utf-8"),
|
||||
headers=headers, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
|
||||
body = json.loads(resp.read().decode("utf-8"))
|
||||
if "Error" in body.get("Response", {}):
|
||||
err = body["Response"]["Error"]
|
||||
raise RuntimeError(f"TMT {err.get('Code')}: {err.get('Message')}")
|
||||
return body["Response"]["TargetText"]
|
||||
except (urllib.error.URLError, urllib.error.HTTPError, OSError,
|
||||
RuntimeError, KeyError, ValueError) as e:
|
||||
last_err = e
|
||||
raise RuntimeError(f"重试 {retries} 次仍失败: {last_err}")
|
||||
|
||||
|
||||
# ---- selftest:官方文档公布值断言 + 假密钥全链路展示(不触网) ----
|
||||
|
||||
# 官方签名文档 cvm 示例的输入与公布中间量(密钥打码故仅断言无密钥依赖的三组量)
|
||||
_DOC_PAYLOAD = ('{"Limit": 1, "Filters": [{"Values": ["\\u672a\\u547d\\u540d"], '
|
||||
'"Name": "instance-name"}]}')
|
||||
_DOC_HASHED_PAYLOAD = "35e9c5b0e3ae67532d3c9f17ead6c90222632e5b1ff7f6e89887f1398934f064"
|
||||
_DOC_CANONICAL_REQUEST = (
|
||||
"POST\n/\n\n"
|
||||
"content-type:application/json; charset=utf-8\n"
|
||||
"host:cvm.tencentcloudapi.com\n"
|
||||
"x-tc-action:describeinstances\n"
|
||||
"\n"
|
||||
"content-type;host;x-tc-action\n"
|
||||
+ _DOC_HASHED_PAYLOAD
|
||||
)
|
||||
_DOC_HASHED_CANONICAL = "7019a55be8395899b900fb5564e4200d984910f34794a27cb3fb7d10ff6a1e84"
|
||||
_DOC_STRING_TO_SIGN = (
|
||||
"TC3-HMAC-SHA256\n1551113065\n2019-02-25/cvm/tc3_request\n" + _DOC_HASHED_CANONICAL
|
||||
)
|
||||
|
||||
|
||||
def selftest() -> int:
|
||||
"""断言签名拼接与官方文档示例一致;失败 exit 1,全过 exit 0。"""
|
||||
failures = []
|
||||
|
||||
# 官方签名文档 cvm 示例(三头规格)直接走 sign() 代码路径,逐字节复现规范串
|
||||
_, doc_debug = sign("AKIDDemo", "DemoKey", "DescribeInstances", "cvm",
|
||||
"cvm.tencentcloudapi.com", _DOC_PAYLOAD, 1551113065,
|
||||
extra_signed_headers={"x-tc-action": "describeinstances"})
|
||||
if doc_debug["canonical_request"] != _DOC_CANONICAL_REQUEST:
|
||||
failures.append(f"sign() 生成的 CanonicalRequest 与文档不符:\n{doc_debug['canonical_request']!r}")
|
||||
|
||||
hp = _sha256_hex(_DOC_PAYLOAD)
|
||||
if hp != _DOC_HASHED_PAYLOAD:
|
||||
failures.append(f"HashedRequestPayload 不符: {hp}")
|
||||
|
||||
hcr = _sha256_hex(doc_debug["canonical_request"])
|
||||
if hcr != _DOC_HASHED_CANONICAL:
|
||||
failures.append(f"HashedCanonicalRequest 不符: {hcr}")
|
||||
|
||||
sts = f"TC3-HMAC-SHA256\n1551113065\n2019-02-25/cvm/tc3_request\n{hcr}"
|
||||
if sts != _DOC_STRING_TO_SIGN:
|
||||
failures.append(f"StringToSign 不符: {sts!r}")
|
||||
|
||||
if failures:
|
||||
print("[SELFTEST-FAIL]")
|
||||
for f in failures:
|
||||
print(" -", f)
|
||||
return 1
|
||||
|
||||
print("[SELFTEST-PASS] 官方文档 cvm 示例中间量逐字节一致:"
|
||||
"HashedRequestPayload / HashedCanonicalRequest / StringToSign")
|
||||
print()
|
||||
|
||||
# 固定假密钥 + tmt 真实参数,打印全链路中间量(可与官方 API Explorer 比对)
|
||||
fake_id, fake_key = "AKIDFAKEexample0000000000000000000000", "FakeSecretKeyDoNotUse00000"
|
||||
payload = json.dumps({"Source": "zh", "Target": "en", "ProjectId": 0,
|
||||
"SourceText": "你好,世界"}, ensure_ascii=False)
|
||||
headers, debug = sign(fake_id, fake_key, ACTION, SERVICE, HOST, payload, 1551113065)
|
||||
print("[TMT-REQUEST 签名链中间量](假密钥,供与官方 API Explorer 比对)")
|
||||
for k in ("timestamp", "date", "credential_scope", "canonical_request",
|
||||
"hashed_request_payload", "hashed_canonical_request", "string_to_sign",
|
||||
"secret_date_hex", "secret_service_hex", "secret_signing_hex",
|
||||
"signature", "authorization"):
|
||||
print(f"--- {k} ---")
|
||||
print(debug[k])
|
||||
print()
|
||||
print("--- 实际发送 HTTP 头 ---")
|
||||
for k, v in headers.items():
|
||||
print(f"{k}: {v}")
|
||||
print(f"(body) {payload}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if "--selftest" in sys.argv:
|
||||
sys.exit(selftest())
|
||||
print("用法: python3 tencent_tmt.py --selftest;正常使用经 translate_posts.py 编排",
|
||||
file=sys.stderr)
|
||||
sys.exit(2)
|
||||
@@ -0,0 +1,254 @@
|
||||
#!/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())
|
||||
Reference in new issue
Block a user