多言語アプリの訳ブレを止める — Batch API と用語集キャッシュで Localizable.strings を一貫翻訳する設計

# 出力をツール呼び出しに固定して、自由形式テキストの混入を防ぐ
TRANSLATE_TOOL = {
    "name": "emit_translations",
    "description": "翻訳済みの key-value ペアを返す",
    "input_schema": {
        "type": "object",
        "properties": {
            "translations": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "key": {"type": "string"},
                        "value": {"type": "string"},
                    },
                    "required": ["key", "value"],
                },
            }
        },
        "required": ["translations"],
    },
}
 
def build_user_message(entries: dict) -> str:
    # entries: {"settings.wallpaper.title": "壁紙の品質", ...}
    lines = [f'{k}\t{v}' for k, v in entries.items()]
    return (
        "次の key と原文を翻訳し、emit_translations ツールで返してください。"
        "key は一切変更しないでください。\n\n" + "\n".join(lines)
    )

# build_batch.py — 言語ごとにバッチリクエストを組み立てて投入する
from anthropic.types.messages.batch_create_params import Request
from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
 
def chunk(d: dict, size: int):
    items = list(d.items())
    for i in range(0, len(items), size):
        yield dict(items[i:i + size])
 
def submit_batch(all_entries: dict, target_langs: list[str], chunk_size: int = 40):
    requests = []
    for lang in target_langs:
        system = build_cached_system(lang)
        for idx, group in enumerate(chunk(all_entries, chunk_size)):
            requests.append(Request(
                custom_id=f"{lang}-{idx}",  # 後でどの言語のどの塊か特定する
                params=MessageCreateParamsNonStreaming(
                    model="claude-opus-4-8",
                    max_tokens=4096,
                    system=system,
                    tools=[TRANSLATE_TOOL],
                    tool_choice={"type": "tool", "name": "emit_translations"},
                    messages=[{"role": "user", "content": build_user_message(group)}],
                ),
            ))
    batch = client.messages.batches.create(requests=requests)
    print(f"submitted batch: {batch.id}, requests={len(requests)}")
    return batch.id

# collect.py — バッチ完了を待って結果を Localizable.strings に書き戻す
import time, json
 
def wait_and_collect(batch_id: str) -> dict:
    while True:
        batch = client.messages.batches.retrieve(batch_id)
        if batch.processing_status == "ended":
            break
        time.sleep(30)  # 完了までポーリング
 
    results: dict[str, dict] = {}  # {lang: {key: value}}
    for entry in client.messages.batches.results(batch_id):
        lang = entry.custom_id.rsplit("-", 1)[0]
        if entry.result.type != "succeeded":
            print(f"⚠️ failed: {entry.custom_id} ({entry.result.type})")
            continue
        for block in entry.result.message.content:
            if block.type == "tool_use":
                for pair in block.input["translations"]:
                    results.setdefault(lang, {})[pair["key"]] = pair["value"]
    return results

import re
 
PLACEHOLDER = re.compile(r'%(?:\d+\$)?[@dfsu]')
 
def validate(src: str, dst: str) -> list[str]:
    errors = []
    # フォーマット指定子の個数と種類が原文と一致するか
    if sorted(PLACEHOLDER.findall(src)) != sorted(PLACEHOLDER.findall(dst)):
        errors.append("placeholder mismatch")
    # 改行コードの数が保持されているか
    if src.count("\\n") != dst.count("\\n"):
        errors.append("newline mismatch")
    # 引用符が閉じているか(.strings の構文を壊さないため)
    if dst.count('"') % 2 != 0:
        errors.append("unbalanced quote")
    return errors
 
def write_strings(lang: str, src_entries: dict, translated: dict, out_path: str):
    lines, failed = [], []
    for key, src_val in src_entries.items():
        dst_val = translated.get(key)
        if dst_val is None:
            failed.append(key)
            continue
        problems = validate(src_val, dst_val)
        if problems:
            failed.append(f"{key} ({', '.join(problems)})")
            continue
        escaped = dst_val.replace('"', '\\"')
        lines.append(f'"{key}" = "{escaped}";')
    with open(out_path, "w", encoding="utf-8") as f:
        f.write("\n".join(lines) + "\n")
    print(f"[{lang}] wrote {len(lines)} / failed {len(failed)}")
    return failed