DeepL Write 批量改写：按请求体大小拆分文本

官方提醒单次请求体约 10 KiB。批量润色邮件/段落时要拆包。这篇 batch_rephrase.py 按 JSON 序列化大小贪心装箱，再顺序调用 rephrase。

先看清成品

• dry-run 显示 batches；mock 写出 3 行 jsonl。
• 2 项 pytest 通过。

合规提醒：DeepL Write API 目前需要 API Pro（以官方说明为准），Free 计划可能不可用。密钥用 Authorization: DeepL-Auth-Key ...，勿提交到公开仓库。Write 是同语种润色，不是翻译；不要用它处理你无权处理的保密文本（注意 Pro 数据保护条款）。

运行说明：无 DeepL Pro 密钥。A/B/C 均用本地 mock 验证；对 api-free 用无效密钥实测 Write 返回 HTTP 403 鉴权失败。2026-10-23（Asia/Shanghai）pytest 通过。正式调用基址一般为 https://api.deepl.com 。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

98-deepl-write-C-src/
├── batch_rephrase.py
├── sample_lines.txt
├── mock_deepl_write.py
└── test_98c.py

第 1 步：批量脚本

#!/usr/bin/env python3
"""Batch DeepL Write rephrase while respecting ~10 KiB request body limit."""
from __future__ import annotations

import argparse
import json
import os
import sys
import urllib.error
import urllib.request
from pathlib import Path

LIMIT = 10 * 1024


def batches(texts: list[str], target_lang: str, style: str | None) -> list[list[str]]:
    """Greedy pack texts so JSON body stays under LIMIT."""
    out: list[list[str]] = []
    cur: list[str] = []

    def size(ts: list[str]) -> int:
        payload = {"text": ts, "target_lang": target_lang}
        if style:
            payload["writing_style"] = style
        return len(json.dumps(payload).encode())

    for t in texts:
        if size([t]) > LIMIT:
            raise ValueError(f"single text alone exceeds {LIMIT} bytes")
        trial = cur + [t]
        if cur and size(trial) > LIMIT:
            out.append(cur)
            cur = [t]
        else:
            cur = trial
    if cur:
        out.append(cur)
    return out


def call(base, key, texts, target_lang, style):
    payload = {"text": texts, "target_lang": target_lang}
    if style:
        payload["writing_style"] = style
    req = urllib.request.Request(
        f"{base.rstrip('/')}/v2/write/rephrase",
        data=json.dumps(payload).encode(),
        headers={"Authorization": f"DeepL-Auth-Key {key}", "Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=60) as resp:
            return json.loads(resp.read().decode())
    except urllib.error.HTTPError as e:
        raise SystemExit(f"HTTP {e.code}: {e.read().decode(errors='replace')}") from e


def main(argv=None) -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("input", help="每行一段文本的 UTF-8 文件")
    ap.add_argument("--base", default=os.environ.get("DEEPL_BASE_URL", "https://api.deepl.com"))
    ap.add_argument("--key", default=os.environ.get("DEEPL_API_KEY", ""))
    ap.add_argument("--lang", default="en-US")
    ap.add_argument("--style", default=None)
    ap.add_argument("--dry-run", action="store_true")
    ap.add_argument("-o", "--out", default="improved.jsonl")
    args = ap.parse_args(argv)
    texts = [ln.strip() for ln in Path(args.input).read_text(encoding="utf-8").splitlines() if ln.strip()]
    packs = batches(texts, args.lang, args.style)
    print(f"segments={len(texts)} batches={len(packs)}")
    if args.dry_run:
        for i, p in enumerate(packs):
            print(f"batch{i}: n={len(p)} bytes~={len(json.dumps({'text': p, 'target_lang': args.lang}).encode())}")
        return 0
    if not args.key:
        print("设置 DEEPL_API_KEY", file=sys.stderr)
        return 2
    out_path = Path(args.out)
    with out_path.open("w", encoding="utf-8") as f:
        for pack in packs:
            r = call(args.base, args.key, pack, args.lang, args.style)
            for src, imp in zip(pack, r.get("improvements") or []):
                f.write(json.dumps({"source": src, "improved": imp.get("text")}, ensure_ascii=False) + "\n")
    print(f"wrote {out_path}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

第 2 步：测试

import subprocess
import sys
import time
from pathlib import Path
from batch_rephrase import batches

HERE = Path(__file__).resolve().parent


def test_pack_under_limit():
    texts = ["hello world " * 50] * 20
    packs = batches(texts, "en-US", None)
    assert len(packs) >= 1
    import json
    for p in packs:
        assert len(json.dumps({"text": p, "target_lang": "en-US"}).encode()) <= 10 * 1024


def test_cli_mock(tmp_path):
    port = 18982
    proc = subprocess.Popen(
        [sys.executable, str(HERE / "mock_deepl_write.py"), "--port", str(port)],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    try:
        time.sleep(0.3)
        out = tmp_path / "out.jsonl"
        r = subprocess.run(
            [
                sys.executable,
                str(HERE / "batch_rephrase.py"),
                str(HERE / "sample_lines.txt"),
                "--base",
                f"http://127.0.0.1:{port}",
                "--key",
                "deepl-test-key",
                "-o",
                str(out),
            ],
            capture_output=True,
            text=True,
            check=True,
        )
        assert "batches=" in r.stdout
        lines = out.read_text().strip().splitlines()
        assert len(lines) == 3
    finally:
        proc.terminate()
        proc.wait(timeout=5)

第 3 步：运行

segments=3 batches=1
batch0: n=3 bytes~=132
..                                                                       [100%]
2 passed in 0.38s

怎么检查结果？

• 单段过大应直接报错而非静默截断。
• jsonl 行数等于输入非空行数。

调试示例

按字符数估算会翻车。 多字节 UTF-8 与 JSON 转义会占更多字节。本脚本用 len(json.dumps(...).encode()) 估真实请求体。

常见问题

模板规则

• 尊重 10KiB。
• 日志勿打印全文机密。
• Pro 配额以控制台为准。

项目资源

批量的第一课是限流与限体积。