From de567823917bd36ddb1684cf6d7cf59e55f48c9e Mon Sep 17 00:00:00 2001 From: DazedAnon Date: Fri, 12 Jun 2026 16:39:46 -0500 Subject: [PATCH] LEN: Add batch mode and some optimizations --- README.md | 48 +++ modules/main.py | 219 +++++++++---- util/translation.py | 740 ++++++++++++++++++++++++++++++++++++++------ 3 files changed, 841 insertions(+), 166 deletions(-) diff --git a/README.md b/README.md index 5ace3e1..4a1cad5 100644 --- a/README.md +++ b/README.md @@ -16,6 +16,7 @@ An AI-powered game translation tool with a GUI. Translate RPG Maker, Ren'Py, Tyr - [Using the GUI](#using-the-gui) - [Vocab & Prompt](#vocab--prompt) - [Tips](#tips) +- [Batch Translation (Anthropic, 50% off)](#batch-translation-anthropic-50-off) - [Folder Structure](#folder-structure) - [Finding Untranslated Text (Snipping Tool OCR)](#finding-untranslated-text-snipping-tool-ocr) - [RPG Maker Translation Workflow](#rpg-maker-translation-workflow) @@ -185,6 +186,53 @@ This is the system prompt sent to the AI. A default `prompt.txt` is included and --- +## Batch Translation (Anthropic, 50% off) + +When using a Claude model, the CLI offers a third mode that translates through the +[Anthropic Message Batches API](https://platform.claude.com/docs/en/build-with-claude/batch-processing.md) +— every token (input, output, and prompt-cache reads/writes) is billed at **50% of the live price**. +Batches usually finish within an hour (24h worst case), so use it for large jobs where you don't +need results immediately. + +``` +python start.py + -> 3. Batch Translate (Anthropic Batches API, 50% off) +``` + +How it works (all engine modules are supported automatically): + +1. **Pass 1 (collect)** — files are processed normally, but instead of calling the API each + request is queued to `log/batch_requests.json`. Requests are byte-identical to live ones: + the static `prompt.txt` block is cached with a 1h TTL, matched vocab and translation + history ride along per request, and structured output enforces the exact line count. + Speaker/variable names still translate live during this pass (they get embedded into the + dialogue payloads, so both passes must resolve them identically) — they're a tiny share + of the volume. +2. **Cost estimate** — before anything is submitted you get a cost breakdown + (batch + cache / batch worst-case / live price) and a y/n confirmation. +3. **Submit / poll / fetch** — the batch is submitted, polled until it ends + (`batchPollInterval` env var controls the interval, default 60s), and the results are + saved to `log/batch_results.json`. Ctrl-C while polling is safe — the batch keeps + processing server-side. +4. **Pass 2 (consume)** — files are processed again; every payload is filled from the batch + results through the normal validation pipeline (line counts, placeholders, content + checks). Anything the batch missed or that fails validation falls back to the live API + automatically, so the output is always complete. + +Context note: in live mode the rolling translation history contains the previous batch's +English lines; in batch mode requests are independent, so the history carries the previous +batch's *source* lines instead. The model still sees the surrounding scene, and `vocab.txt` +keeps names and terms consistent — so keep your vocab file in good shape for batch runs. + +Cost tracking is exact: per-file and total costs printed after the consume pass use the real +billed token counts (cache reads at 0.1x, cache writes at 2x, output at the output rate) with +the 50% batch discount applied. + +`python selftest_batch.py` round-trips the whole flow offline (no API key needed) if you want +to verify the pipeline after making changes. + +--- + ## Folder Structure | Folder | Purpose | diff --git a/modules/main.py b/modules/main.py index e2e07b4..040e017 100644 --- a/modules/main.py +++ b/modules/main.py @@ -88,22 +88,50 @@ tqdm.write( def main(): - # Clear the translation cache at the start of the run from util.translation import clear_cache - clear_cache() - + estimate = "" + batch_mode = False speaker_parse = False # Deferred until after engine select while estimate == "": - estimate = input("Select Mode:\n\n 1. Translate\n 2. Estimate\n") + estimate = input("Select Mode:\n\n 1. Translate\n 2. Estimate\n 3. Batch Translate (Anthropic Batches API, 50% off)\n") match estimate: case "1": estimate = False case "2": estimate = True + case "3": + estimate = False + batch_mode = True case _: estimate = "" + resume_state = None + if batch_mode: + from util.translation import isClaudeNative, batchRunState + if not isClaudeNative(os.getenv("model", "")): + tqdm.write( + Fore.RED + + "Batch Translate requires a Claude model with the 'api' env var unset or pointing at anthropic.com." + + Fore.RESET + ) + return + # An interrupted batch run can be resumed instead of re-collecting + # (a second submission would be billed again). + resume_state = batchRunState() + if resume_state: + confirm = "" + while confirm not in ("y", "n"): + confirm = input(f"A previous batch run was interrupted ({resume_state}). Resume it? (y/n)\n").strip().lower() + if confirm == "n": + resume_state = None + + # Clear the translation cache at the start of the run. Kept when resuming a + # batch so names translated during collect stay consistent with the queued + # payloads in the consume pass. + if not resume_state: + clear_cache() + version = "" while True: tqdm.write("Select game engine:\n") @@ -125,7 +153,7 @@ files to translate are in the /files folder and that you picked the right game e # If translating RPGMaker MV/MZ, prompt for speaker parse mode speaker_parse = False - if version == 0 and not estimate: + if version == 0 and not estimate and not batch_mode: sub = "" while sub == "": sub = input("RPGMaker MV/MZ options:\n\n 1. Standard Translate\n 2. Parse Speakers (collect speaker names only)\n") @@ -180,78 +208,133 @@ files to translate are in the /files folder and that you picked the right game e except Exception: pass - # Use single worker for estimate mode to prevent race conditions - max_workers = 1 if estimate else THREADS - with ThreadPoolExecutor(max_workers=max_workers) as executor: - futures = [] - files_root = "files" + def runFiles(estimate): + runCost = totalCost + # Use single worker for estimate mode to prevent race conditions + max_workers = 1 if estimate else THREADS + with ThreadPoolExecutor(max_workers=max_workers) as executor: + futures = [] + files_root = "files" - # Special-case: Images engine expects a folder, not a file; schedule per directory containing assets - if MODULES[version][0] == "Images": - for root, dirs, filenames in os.walk(files_root): - # Skip hidden/system directories - dirs[:] = [d for d in dirs if d not in {".git", "__pycache__"}] + # Special-case: Images engine expects a folder, not a file; schedule per directory containing assets + if MODULES[version][0] == "Images": + for root, dirs, filenames in os.walk(files_root): + # Skip hidden/system directories + dirs[:] = [d for d in dirs if d not in {".git", "__pycache__"}] - # Skip the root 'files' itself to avoid processing everything twice - # We'll still allow scheduling for root if it contains assets + # Skip the root 'files' itself to avoid processing everything twice + # We'll still allow scheduling for root if it contains assets - # Only schedule directories that contain potential assets - has_assets = any(fn.lower().endswith((".png", ".txt")) for fn in filenames) - if not has_assets: - continue - - # Compute relative directory path and ensure translated mirror exists - rel_dir = os.path.relpath(root, files_root).replace(os.sep, "/") - if rel_dir == ".": - # Represent root as empty string so handler creates files under translated/ directly - rel_dir = "" - try: - target_dir = os.path.join("translated", rel_dir.replace("/", os.sep)) if rel_dir else "translated" - os.makedirs(target_dir, exist_ok=True) - except Exception: - pass - - futures.append( - executor.submit(MODULES[version][2], rel_dir, estimate) - ) - else: - # Gather all candidate files recursively - for root, dirs, filenames in os.walk(files_root): - # Skip hidden/system directories if any - dirs[:] = [d for d in dirs if d not in {".git", "__pycache__"}] - - for fname in filenames: - if fname == ".gitkeep": + # Only schedule directories that contain potential assets + has_assets = any(fn.lower().endswith((".png", ".txt")) for fn in filenames) + if not has_assets: continue - abs_path = os.path.join(root, fname) - # Build relative path from 'files' root using POSIX-style separators so handlers can do 'files/' + rel - rel_path = os.path.relpath(abs_path, files_root) - rel_path_posix = rel_path.replace(os.sep, "/") + # Compute relative directory path and ensure translated mirror exists + rel_dir = os.path.relpath(root, files_root).replace(os.sep, "/") + if rel_dir == ".": + # Represent root as empty string so handler creates files under translated/ directly + rel_dir = "" + try: + target_dir = os.path.join("translated", rel_dir.replace("/", os.sep)) if rel_dir else "translated" + os.makedirs(target_dir, exist_ok=True) + except Exception: + pass - # Check extension match for the selected module version - for m in MODULES[version][1]: - if rel_path_posix.endswith(m): - # Ensure the corresponding directory exists under 'translated' - rel_dir = os.path.dirname(rel_path_posix) - if rel_dir: - try: - os.makedirs(os.path.join("translated", rel_dir.replace("/", os.sep)), exist_ok=True) - except Exception: - # Best-effort; handler may attempt write and fail if permissions are insufficient - pass + futures.append( + executor.submit(MODULES[version][2], rel_dir, estimate) + ) + else: + # Gather all candidate files recursively + for root, dirs, filenames in os.walk(files_root): + # Skip hidden/system directories if any + dirs[:] = [d for d in dirs if d not in {".git", "__pycache__"}] - futures.append( - executor.submit(MODULES[version][2], rel_path_posix, estimate) - ) - break # Avoid double-adding if multiple ext entries match + for fname in filenames: + if fname == ".gitkeep": + continue - for future in as_completed(futures): + abs_path = os.path.join(root, fname) + # Build relative path from 'files' root using POSIX-style separators so handlers can do 'files/' + rel + rel_path = os.path.relpath(abs_path, files_root) + rel_path_posix = rel_path.replace(os.sep, "/") + + # Check extension match for the selected module version + for m in MODULES[version][1]: + if rel_path_posix.endswith(m): + # Ensure the corresponding directory exists under 'translated' + rel_dir = os.path.dirname(rel_path_posix) + if rel_dir: + try: + os.makedirs(os.path.join("translated", rel_dir.replace("/", os.sep)), exist_ok=True) + except Exception: + # Best-effort; handler may attempt write and fail if permissions are insufficient + pass + + futures.append( + executor.submit(MODULES[version][2], rel_path_posix, estimate) + ) + break # Avoid double-adding if multiple ext entries match + + for future in as_completed(futures): + try: + runCost = future.result() + except Exception as e: + tracebackLineNo = str(traceback.extract_tb(sys.exc_info()[2])[-1].lineno) + tqdm.write(Fore.RED + str(e) + "|" + tracebackLineNo + Fore.RESET) + return runCost + + if batch_mode: + from util.translation import ( + set_batch_phase, + clearBatchFiles, + pendingBatchRequests, + estimateBatchCost, + runTranslationBatches, + ) + poll = int(os.getenv("batchPollInterval", "60") or 60) + run_consume = True + + if resume_state is None: + clearBatchFiles() + + # Pass 1 — collect every needed request without calling the API. + tqdm.write(Fore.CYAN + "[BATCH] Pass 1/2: collecting requests..." + Fore.RESET) + set_batch_phase("collect") try: - totalCost = future.result() - except Exception as e: - tracebackLineNo = str(traceback.extract_tb(sys.exc_info()[2])[-1].lineno) - tqdm.write(Fore.RED + str(e) + "|" + tracebackLineNo + Fore.RESET) + totalCost = runFiles(False) + finally: + set_batch_phase(None) + + if pendingBatchRequests() == 0: + tqdm.write("[BATCH] No requests queued — nothing needed the API.") + run_consume = False + else: + estimateBatchCost() + confirm = "" + while confirm not in ("y", "n"): + confirm = input("Submit batch? (y/n)\n").strip().lower() + if confirm == "n": + tqdm.write("[BATCH] Not submitted. The queue is kept in log/batch_requests.json.") + return + runTranslationBatches(poll) + elif resume_state == "submitted": + tqdm.write(Fore.CYAN + "[BATCH] Resuming the submitted batch..." + Fore.RESET) + runTranslationBatches(poll) + else: # "fetched" — results already downloaded, just write the files + tqdm.write(Fore.CYAN + "[BATCH] Resuming from fetched results..." + Fore.RESET) + + if run_consume: + # Pass 2 — write the translated files from the fetched results. + # Anything the batch missed falls back to the live API. + tqdm.write(Fore.CYAN + "[BATCH] Pass 2/2: writing translated files..." + Fore.RESET) + set_batch_phase("consume") + try: + totalCost = runFiles(False) + finally: + set_batch_phase(None) + else: + totalCost = runFiles(estimate) # Finalize speaker parse mode by writing collected speakers to vocab if speaker_parse: diff --git a/util/translation.py b/util/translation.py index 9a67263..0d9c9a0 100644 --- a/util/translation.py +++ b/util/translation.py @@ -95,6 +95,29 @@ def _normalize_openai_base_url(url: str) -> str: _url += "/" return _url + +def isClaudeModel(model): + """True when the model name looks like an Anthropic Claude model.""" + return bool(model) and any(x in model.lower() for x in ("claude", "sonnet", "haiku", "opus")) + + +def isClaudeNative(model): + """True when this model routes to the native Anthropic SDK. + + Mirrors the routing check in translateText: the model must look like Claude + AND the configured API URL must be unset or point at anthropic.com. Any + other custom URL (e.g. DeepSeek, OpenAI proxy) uses the OpenAI-compatible + path even for Claude-named models. + """ + live_api = os.getenv("api", "").strip() + return isClaudeModel(model) and (not live_api or "anthropic" in live_api.lower()) + + +# Models that REJECT sampling params (temperature/top_p/top_k) with a 400 — +# Claude Opus 4.7 and up retired them in favour of adaptive thinking. Matches +# opus-4-7, opus-4-8, opus-4-10+ and fable, but NOT opus-4-6 or Sonnet/Haiku. +_NO_SAMPLING_RE = re.compile(r"opus-4-(?:[7-9]\b|[1-9]\d)|fable", re.I) + # Tracks which distinct batch sizes have already been cache-written during this estimate run. # Each unique numLines value maps to a distinct output_config schema → one write per size. # Persisted to disk so sequential GUI subprocesses share state. @@ -620,6 +643,429 @@ def set_var_translations_batch(pairs): _save_var_map() +# ===== Anthropic Message Batches (50% off all token usage) ===== +# Batch translation is a two-pass flow driven by the batch phase (kept in the +# BATCH_PHASE env var so GUI subprocesses inherit it): +# collect: translateAI builds each cache-missed request (byte-identical to a +# live request, including the cached system block) and queues it +# instead of calling the API. Text is left untranslated. +# consume: translateAI feeds the fetched batch responses through the normal +# validation/restore path; anything missing or invalid falls back +# to the live API automatically. +# Between the passes, submit/poll/fetch the queue with runTranslationBatches(). +BATCH_QUEUE_FILE = Path("log/batch_requests.json") +BATCH_STATE_FILE = Path("log/batch_state.json") +BATCH_RESULTS_FILE = Path("log/batch_results.json") +BATCH_LOCK_FILE = Path("log/batch_files.lock") +BATCH_LOCK = threading.RLock() +# API limits are 100,000 requests / 256 MB per batch; stay safely under both. +BATCH_MAX_REQUESTS = 100_000 +BATCH_MAX_BYTES = 200 * 1024 * 1024 + +_batch_phase = None +_batch_results = None # in-memory copy of BATCH_RESULTS_FILE (read-only during consume) +_batch_queue_pending = {} # process-local queue entries not yet flushed to disk + + +def set_batch_phase(phase): + """Set the batch phase ('collect', 'consume' or None) for this process and + any subprocesses it spawns (the GUI runs one per file).""" + global _batch_phase, _batch_results + _batch_phase = phase if phase in ("collect", "consume") else None + if _batch_phase: + os.environ["BATCH_PHASE"] = _batch_phase + else: + os.environ.pop("BATCH_PHASE", None) + _batch_results = None # phase change invalidates the in-memory results copy + + +def get_batch_phase(): + """Current batch phase, or None when batch translation is off.""" + if _batch_phase: + return _batch_phase + phase = os.getenv("BATCH_PHASE", "").strip().lower() + return phase if phase in ("collect", "consume") else None + + +@contextmanager +def _batch_file_lock(): + """Cross-process lock for the batch queue/state/results files.""" + BATCH_LOCK_FILE.parent.mkdir(parents=True, exist_ok=True) + with open(BATCH_LOCK_FILE, "a+b") as lock_file: + if os.name == "nt": + import msvcrt + lock_file.seek(0) + msvcrt.locking(lock_file.fileno(), msvcrt.LK_LOCK, 1) + try: + yield + finally: + lock_file.seek(0) + msvcrt.locking(lock_file.fileno(), msvcrt.LK_UNLCK, 1) + else: + import fcntl + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) + try: + yield + finally: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + + +def _read_batch_file(path): + """Read a batch JSON file; return an empty dict if it is unavailable.""" + try: + if path.exists(): + with open(path, "r", encoding="utf-8") as f: + data = json.load(f) + return data if isinstance(data, dict) else {} + except Exception: + pass + return {} + + +def _write_batch_file(path, data): + """Atomically write a batch JSON file (no indent — queues can be large).""" + path.parent.mkdir(parents=True, exist_ok=True) + tmp_file = path.with_name(f"{path.name}.{os.getpid()}.{threading.get_ident()}.tmp") + with open(tmp_file, "w", encoding="utf-8") as f: + json.dump(data, f, ensure_ascii=False) + tmp_file.replace(path) + + +def peek_cached_translation(payload, language): + """Cache lookup that never blocks or writes a pending marker. + + The collect pass uses this instead of get_cached_translation so abandoned + pending markers can't stall the consume pass for CACHE_PENDING_TTL.""" + key = get_cache_key(payload, language) + with CACHE_LOCK: + with _translation_cache_file_lock(): + cache = _read_cache_from_disk() + if _cache: + cache = _merge_translation_caches(cache, _cache) + entry = cache.get(key) + if entry is None or _is_pending_cache_entry(entry): + return None + return entry + + +def queue_batch_request(payload, language, params): + """Queue one Batches API request during the collect pass. + + Deduped by the same key the translation cache uses, so identical payloads + across files are only paid for once. Entries are buffered in memory and + merged to disk by flush_batch_queue() at the end of each translateAI call. + """ + key = get_cache_key(payload, language) + with BATCH_LOCK: + _batch_queue_pending[key] = {"payload": payload, "language": language, "params": params} + return key + + +def flush_batch_queue(): + """Merge this process's pending queue entries into the on-disk queue.""" + global _batch_queue_pending + with BATCH_LOCK: + if not _batch_queue_pending: + return + pending, _batch_queue_pending = _batch_queue_pending, {} + try: + with _batch_file_lock(): + queue = _read_batch_file(BATCH_QUEUE_FILE) + for key, entry in pending.items(): + queue.setdefault(key, entry) + _write_batch_file(BATCH_QUEUE_FILE, queue) + except Exception: + _batch_queue_pending.update(pending) # keep entries for the next flush + + +def take_batch_result(payload, language): + """Return the fetched batch response dict for a payload, or None. + + The results file is loaded once per process — it is written before the + consume pass starts and never changes mid-consume.""" + global _batch_results + if _batch_results is None: + with BATCH_LOCK: + if _batch_results is None: + with _batch_file_lock(): + _batch_results = _read_batch_file(BATCH_RESULTS_FILE) + return _batch_results.get(get_cache_key(payload, language)) + + +def pendingBatchRequests(): + """Number of queued batch requests (call after the collect pass).""" + flush_batch_queue() + with _batch_file_lock(): + return len(_read_batch_file(BATCH_QUEUE_FILE)) + + +def batchRunState(): + """'submitted' when a batch is still in flight, 'fetched' when results are + waiting to be consumed, else None. Lets an interrupted batch run resume + instead of re-collecting and paying for a second submission.""" + with _batch_file_lock(): + if _read_batch_file(BATCH_STATE_FILE).get("batches"): + return "submitted" + if _read_batch_file(BATCH_RESULTS_FILE): + return "fetched" + return None + + +def clearBatchFiles(): + """Remove queue/state/results left over from any previous batch run.""" + global _batch_results, _batch_queue_pending + with BATCH_LOCK: + _batch_results = None + _batch_queue_pending = {} + with _batch_file_lock(): + for path in (BATCH_QUEUE_FILE, BATCH_STATE_FILE, BATCH_RESULTS_FILE): + try: + if path.exists(): + path.unlink() + except Exception: + pass + + +def _get_anthropic_client(): + key = os.getenv("key", "").strip() + if not key: + raise Exception("Batch translation requires the 'key' env var (see .env).") + return anthropic.Anthropic(api_key=key) + + +def estimateBatchCost(model=None): + """Print a cost estimate for the queued batch requests and return it. + + Cached-prefix accounting mirrors what Anthropic bills: each distinct cached + prefix is written once (2x input rate at the 1h TTL) and read by every other + request that shares it (0.10x); everything is then halved by the batch + discount. Cache hits inside a batch are best-effort, so the no-cache batch + figure is the worst-case bound. + """ + flush_batch_queue() + with _batch_file_lock(): + queue = _read_batch_file(BATCH_QUEUE_FILE) + if not queue: + print("[BATCH] No batch requests queued.", flush=True) + return None + + enc = tiktoken.encoding_for_model("gpt-4") + prefix_count = {} # cached prefix text -> how many requests reuse it + prefix_tokens = {} # cached prefix text -> token count + dynamic_tokens = 0 + output_tokens = 0 + models = set() + for entry in queue.values(): + params = entry.get("params", {}) + if params.get("model"): + models.add(params["model"]) + blocks = params.get("system") or [] + cut = 0 # split system blocks at the cache breakpoint (0 = nothing cached) + for i, b in enumerate(blocks): + if "cache_control" in b: + cut = i + 1 + break + prefix = "".join(b.get("text", "") for b in blocks[:cut]) + dyn = "".join(b.get("text", "") for b in blocks[cut:]) + if prefix: + prefix_count[prefix] = prefix_count.get(prefix, 0) + 1 + if prefix not in prefix_tokens: + prefix_tokens[prefix] = len(enc.encode(prefix)) + msg_text = "\n".join(str(m.get("content", "")) for m in params.get("messages", [])) + dynamic_tokens += len(enc.encode(dyn)) + len(enc.encode(msg_text)) + 8 + # Output heuristic mirrors countTokens(): payload tokens x 2.5 covers + # the echoed JSON scaffold plus EN expansion. + output_tokens += round(len(enc.encode(str(entry.get("payload", "")))) * 2.5) + + est_model = model or next(iter(models), None) or os.getenv("model", "") + # Use Anthropic's count_tokens for the exact cached-prefix size when possible. + if prefix_tokens: + try: + client = _get_anthropic_client() + for prefix in list(prefix_tokens.keys()): + resp = client.messages.count_tokens( + model=est_model, + system=[{"type": "text", "text": prefix}], + messages=[{"role": "user", "content": "x"}], + ) + prefix_tokens[prefix] = resp.input_tokens + except Exception: + pass # tiktoken estimate already in place + + cache_write_tok = sum(prefix_tokens.values()) + cache_read_tok = sum(prefix_tokens[p] * (prefix_count[p] - 1) for p in prefix_count) + raw_input_tok = sum(prefix_tokens[p] * prefix_count[p] for p in prefix_count) + dynamic_tokens + + pricing = getPricingConfig(est_model) + in_rate = pricing["inputAPICost"] / 1_000_000 + out_rate = pricing["outputAPICost"] / 1_000_000 + + # Batch = 50% off every token, including cache writes (2x, 1h TTL) and reads (0.10x). + batch_cached = (cache_write_tok * in_rate * 2.00 + + cache_read_tok * in_rate * 0.10 + + dynamic_tokens * in_rate + + output_tokens * out_rate) * 0.50 + batch_nocache = (raw_input_tok * in_rate + output_tokens * out_rate) * 0.50 + live_cost = raw_input_tok * in_rate + output_tokens * out_rate + + n_reread = sum(prefix_count.values()) - len(prefix_count) + print(f"[BATCH] {len(queue)} requests queued for {est_model}", flush=True) + print(f"[BATCH] cached prefix: {cache_write_tok:,} tokens (written once, re-read by {n_reread:,} requests)", flush=True) + print(f"[BATCH] dynamic input: {dynamic_tokens:,} tokens | estimated output: {output_tokens:,} tokens", flush=True) + print(f"[BATCH] estimated cost: ${batch_cached:.2f} (batch + prompt cache)", flush=True) + print(f"[BATCH] ${batch_nocache:.2f} (batch, worst case no cache hits)", flush=True) + print(f"[BATCH] ${live_cost:.2f} (live API, no batch discount)", flush=True) + return { + "requests": len(queue), + "model": est_model, + "cache_write_tokens": cache_write_tok, + "cache_read_tokens": cache_read_tok, + "dynamic_tokens": dynamic_tokens, + "output_tokens": output_tokens, + "batch_cached_cost": batch_cached, + "batch_nocache_cost": batch_nocache, + "live_cost": live_cost, + } + + +def submitTranslationBatches(): + """Submit the queued requests to the Anthropic Message Batches API. + + Splits at the API limits and saves the custom_id -> cache-key mapping so + fetchTranslationBatches can route results back. Returns the batch ids.""" + flush_batch_queue() + with _batch_file_lock(): + queue = _read_batch_file(BATCH_QUEUE_FILE) + if not queue: + print("[BATCH] No batch requests queued.", flush=True) + return [] + + client = _get_anthropic_client() + batches = [] + requests, id_map, size = [], {}, 0 + + def _submit(): + nonlocal requests, id_map, size + if not requests: + return + batch = client.messages.batches.create(requests=requests) + batches.append({"id": batch.id, "custom_ids": id_map}) + print(f"[BATCH] submitted {batch.id} ({len(requests)} requests)", flush=True) + requests, id_map, size = [], {}, 0 + + for i, (key, entry) in enumerate(queue.items()): + custom_id = f"req-{i:06d}" + params = entry["params"] + requests.append({"custom_id": custom_id, "params": params}) + id_map[custom_id] = key + size += len(json.dumps(params, ensure_ascii=False).encode("utf-8")) + if len(requests) >= BATCH_MAX_REQUESTS or size >= BATCH_MAX_BYTES: + _submit() + _submit() + + with BATCH_LOCK: + with _batch_file_lock(): + _write_batch_file(BATCH_STATE_FILE, {"batches": batches}) + return [b["id"] for b in batches] + + +def checkTranslationBatches(): + """Print the processing status of submitted batches. True when all ended.""" + with _batch_file_lock(): + state = _read_batch_file(BATCH_STATE_FILE) + if not state.get("batches"): + print("[BATCH] No submitted batches — submit the queue first.", flush=True) + return False + client = _get_anthropic_client() + all_ended = True + for info in state["batches"]: + b = client.messages.batches.retrieve(info["id"]) + counts = getattr(b, "request_counts", None) + suffix = f" counts: {counts}" if counts else "" + print(f"[BATCH] {time.strftime('%H:%M:%S')} {b.id}: {b.processing_status}{suffix}", flush=True) + if b.processing_status != "ended": + all_ended = False + return all_ended + + +def fetchTranslationBatches(): + """Download finished batch results into the local results store. + + Successes are stored keyed by the payload cache key for the consume pass; + errored/expired requests are reported and simply fall back to the live API + during consume. Returns (succeeded, errored) counts.""" + global _batch_results + with _batch_file_lock(): + state = _read_batch_file(BATCH_STATE_FILE) + if not state.get("batches"): + print("[BATCH] No submitted batches — nothing to fetch.", flush=True) + return 0, 0 + client = _get_anthropic_client() + results, errored = {}, [] + for info in state["batches"]: + id_map = info.get("custom_ids", {}) + for r in client.messages.batches.results(info["id"]): + key = id_map.get(r.custom_id) + if key is None: + continue + res = r.result + if res.type != "succeeded": + detail = res.type + err = getattr(res, "error", None) + if err is not None: + inner = getattr(err, "error", err) + detail = f"{res.type} | {getattr(inner, 'type', '')}: {str(getattr(inner, 'message', '') or err)[:200]}" + errored.append((r.custom_id, detail)) + continue + msg = res.message + text = "".join(getattr(b, "text", "") or "" for b in msg.content) + u = msg.usage + cr = getattr(u, "cache_read_input_tokens", 0) or 0 + cw = getattr(u, "cache_creation_input_tokens", 0) or 0 + inp = getattr(u, "input_tokens", 0) or 0 + out = getattr(u, "output_tokens", 0) or 0 + results[key] = { + "text": text, + # prompt_tokens matches _AnthropicCompat: total incl. cache fields. + "prompt_tokens": inp + cr + cw, + "completion_tokens": out, + "cache_read_input_tokens": cr, + "cache_creation_input_tokens": cw, + } + with BATCH_LOCK: + with _batch_file_lock(): + merged = _read_batch_file(BATCH_RESULTS_FILE) + merged.update(results) + _write_batch_file(BATCH_RESULTS_FILE, merged) + # Queue and state are consumed; only the results store remains. + for path in (BATCH_QUEUE_FILE, BATCH_STATE_FILE): + try: + if path.exists(): + path.unlink() + except Exception: + pass + _batch_results = None + print(f"[BATCH] fetched {len(results)} results ({len(errored)} errored).", flush=True) + for cid, why in errored[:20]: + print(f"[BATCH] ! {cid}: {why}", flush=True) + if len(errored) > 20: + print(f"[BATCH] ... ({len(errored) - 20} more)", flush=True) + return len(results), len(errored) + + +def runTranslationBatches(poll=60): + """Submit the queue (unless already submitted), poll to completion, fetch.""" + with _batch_file_lock(): + state = _read_batch_file(BATCH_STATE_FILE) + if not state.get("batches"): + if not submitTranslationBatches(): + return 0, 0 + print(f"[BATCH] polling every {poll}s (Ctrl-C is safe — resume later with fetchTranslationBatches)...", flush=True) + while not checkTranslationBatches(): + time.sleep(poll) + return fetchTranslationBatches() + + class TranslationConfig: """Configuration class to hold all translation settings""" @@ -1103,6 +1549,85 @@ def createTranslationSchema(numLines): } +class _AnthropicCompat: + """OpenAI-shaped wrapper around an Anthropic response (text + usage).""" + class _Usage: + def __init__(self, prompt, completion, cr, cw): + self.prompt_tokens = prompt + self.completion_tokens = completion + self.cache_read_input_tokens = cr + self.cache_creation_input_tokens = cw + @property + def model_extra(self): + return { + "cache_read_input_tokens": self.cache_read_input_tokens, + "cache_creation_input_tokens": self.cache_creation_input_tokens, + } + class _Choice: + class _Msg: + def __init__(self, c): self.content = c + def __init__(self, c): + self.message = _AnthropicCompat._Choice._Msg(c) + def __init__(self, text, prompt, output, cr, cw): + self.choices = [_AnthropicCompat._Choice(text)] + self.usage = _AnthropicCompat._Usage(prompt, output, cr, cw) + + +def buildClaudeRequest(system, user, history, formatType, model, numLines=None, vocab_text=""): + """Build the native Anthropic request kwargs. + + Shared by live calls (translateText) and batch collection (translateAI) so + both produce byte-identical requests and share the same prompt cache. Only + the static prompt is cached — 1h TTL so async batch requests processed + minutes apart still hit it; vocab and history ride in messages so they + never bust the prefix cache (the entire system parameter is the cache key). + """ + if DISABLE_CACHE: + # No cache_control — sends as a plain content block for a real uncached run. + combined_system = system + vocab_text + ant_system = [{"type": "text", "text": f"```\n{combined_system}\n```"}] + else: + ant_system = [{"type": "text", "text": f"```\n{system}\n```", "cache_control": {"type": "ephemeral", "ttl": "1h"}}] + + native_msgs = [{"role": "user", "content": f"```\n{user}\n```"}] + + # Vocab goes into messages as a user turn so it doesn't bust the prefix cache. + if vocab_text and vocab_text.strip(): + native_msgs.insert(0, {"role": "user", "content": vocab_text.strip()}) + native_msgs.insert(1, {"role": "assistant", "content": "Understood."}) + + # History also goes into messages, NOT ant_system. + if isinstance(history, list): + history_items = [str(h) for h in history if h and str(h).strip()] + else: + history_items = [str(history)] if history and str(history).strip() else [] + if history_items: + history_block = "Translation History:\n```\n" + "\n".join(history_items) + "\n```" + native_msgs.insert(0, {"role": "user", "content": history_block}) + native_msgs.insert(1, {"role": "assistant", "content": "Understood."}) + + ant_kwargs = dict( + model=model, + max_tokens=16384, + system=ant_system, + messages=native_msgs, + ) + if formatType == "json" and numLines is not None: + ant_kwargs["output_config"] = { + "format": { + "type": "json_schema", + "schema": createTranslationSchema(numLines), + } + } + elif not _NO_SAMPLING_RE.search(model or ""): + # Plain completions still allow explicit sampling params (except on + # Opus 4.7+ which rejects them with a 400). + ant_kwargs["temperature"] = 0 + # Do not pass temperature with output_config: newer Claude (e.g. Opus 4.7) + # returns errors such as "temperature is not supported" for structured outputs. + return ant_kwargs + + def translateText(system, user, history, penalty, formatType, model, numLines=None, vocab_text=""): """Send translation request to the selected API. @@ -1239,59 +1764,12 @@ def translateText(system, user, history, penalty, formatType, model, numLines=No # Use native Anthropic SDK — the OpenAI compat endpoint strips cache_control # and never returns cache_read/creation_input_tokens. if _is_claude: - # system blocks were built above (with cache_control on static prompt). - ant_system = list(msg[0]["content"]) - - # Convert remaining messages to native format. - # Skip system wrapper lines; collect assistant history and user content. - history_items = [] - native_msgs = [] - for m in msg[1:]: - role = m["role"] - content = m.get("content", "") - if role == "system": - pass # skip "Translation History:\n```" / "```" wrapper lines - elif role == "assistant": - history_items.append(str(content)) - elif role == "user": - native_msgs.append({"role": "user", "content": str(content)}) - - # Vocab goes into messages as a user turn so it doesn't bust the - # Anthropic prefix cache (the entire system parameter is the key). - if vocab_text and vocab_text.strip(): - native_msgs.insert(0, {"role": "user", "content": vocab_text.strip()}) - native_msgs.insert(1, {"role": "assistant", "content": "Understood."}) - - # History also goes into messages, NOT ant_system. - if history_items: - history_block = "Translation History:\n```\n" + "\n".join(history_items) + "\n```" - native_msgs.insert(0, {"role": "user", "content": history_block}) - native_msgs.insert(1, {"role": "assistant", "content": "Understood."}) - - if not native_msgs: - raise ValueError("No user message found for Anthropic native call") + # Built by the shared builder so live and batch requests are + # byte-identical and share the same prompt cache. + ant_kwargs = buildClaudeRequest(system, user, history, formatType, model, numLines, vocab_text=vocab_text) ant_client = anthropic.Anthropic(api_key=openai.api_key) try: - ant_kwargs = dict( - model=model, - max_tokens=16384, - system=ant_system, - messages=native_msgs, - ) - if formatType == "json" and numLines is not None: - ant_kwargs["output_config"] = { - "format": { - "type": "json_schema", - "schema": createTranslationSchema(numLines), - } - } - else: - # Plain completions still allow explicit sampling params. - ant_kwargs["temperature"] = 0 - # Do not pass temperature with output_config: newer Claude (e.g. Opus 4.7) - # returns errors such as "temperature is not supported" for structured outputs. - ant_resp = ant_client.messages.create(**ant_kwargs) except Exception as e: raise Exception(f"Anthropic API error: {e}") @@ -1306,28 +1784,6 @@ def translateText(system, user, history, penalty, formatType, model, numLines=No # input_tokens (native SDK) = non-cached portion; add cache fields for true total. _total_prompt = _inp + _cr + _cw - class _AnthropicCompat: - class _Usage: - def __init__(self, prompt, completion, cr, cw): - self.prompt_tokens = prompt - self.completion_tokens = completion - self.cache_read_input_tokens = cr - self.cache_creation_input_tokens = cw - @property - def model_extra(self): - return { - "cache_read_input_tokens": self.cache_read_input_tokens, - "cache_creation_input_tokens": self.cache_creation_input_tokens, - } - class _Choice: - class _Msg: - def __init__(self, c): self.content = c - def __init__(self, c): - self.message = _AnthropicCompat._Choice._Msg(c) - def __init__(self, text, prompt, output, cr, cw): - self.choices = [_AnthropicCompat._Choice(text)] - self.usage = _AnthropicCompat._Usage(prompt, output, cr, cw) - compat_response = _AnthropicCompat(_ant_text, _total_prompt, _out, _cr, _cw) _write_request_debug_log("anthropic", ant_kwargs, compat_response.usage) return compat_response @@ -1578,14 +2034,24 @@ def calculateCost(inputTokens, outputTokens, model): cw = getattr(_thread_local, 'file_cache_write', 0) reg = getattr(_thread_local, 'file_regular', 0) out = getattr(_thread_local, 'file_output', 0) + bcr = getattr(_thread_local, 'file_batch_read', 0) + bcw = getattr(_thread_local, 'file_batch_write', 0) + breg = getattr(_thread_local, 'file_batch_regular', 0) + bout = getattr(_thread_local, 'file_batch_output', 0) pricing = getPricingConfig(model) br = pricing["inputAPICost"] / 1_000_000 orr = pricing["outputAPICost"] / 1_000_000 - cost = cr * br * 0.10 + cw * br * 2.00 + reg * br + out * orr + cost = (cr * br * 0.10 + cw * br * 2.00 + reg * br + out * orr + # Batch API tokens: same rates, then the 50% batch discount. + + (bcr * br * 0.10 + bcw * br * 2.00 + breg * br + bout * orr) * 0.50) _thread_local.file_cache_read = 0 _thread_local.file_cache_write = 0 _thread_local.file_regular = 0 _thread_local.file_output = 0 + _thread_local.file_batch_read = 0 + _thread_local.file_batch_write = 0 + _thread_local.file_batch_regular = 0 + _thread_local.file_batch_output = 0 _thread_local.file_cost_ready = False return cost # TOTAL call (flag already cleared): return the cross-thread running total. @@ -1693,12 +2159,28 @@ def translateAI(text, history, config, filename=None, pbar=None, lock=None, mism _thread_local.file_cache_write = 0 _thread_local.file_regular = 0 _thread_local.file_output = 0 + # Batch API usage is billed at 50% so it is accumulated separately. + if not hasattr(_thread_local, 'file_batch_read'): + _thread_local.file_batch_read = 0 + _thread_local.file_batch_write = 0 + _thread_local.file_batch_regular = 0 + _thread_local.file_batch_output = 0 # Snapshot accumulators so end-of-call delta only counts tokens from this call. _prev_cr = _thread_local.file_cache_read _prev_cw = _thread_local.file_cache_write _prev_reg = _thread_local.file_regular _prev_out = _thread_local.file_output + _prev_bcr = _thread_local.file_batch_read + _prev_bcw = _thread_local.file_batch_write + _prev_breg = _thread_local.file_batch_regular + _prev_bout = _thread_local.file_batch_output _thread_local.file_cost_ready = False # will be set True at end of translateAI + + # Anthropic batch phase ('collect'/'consume'); None when off, in estimate + # mode, or when the model doesn't route to the native Anthropic SDK. + batch_phase = get_batch_phase() + if batch_phase and (config.estimateMode or not isClaudeNative(config.model)): + batch_phase = None if isinstance(text, list): formatType = "json" @@ -1835,8 +2317,20 @@ def translateAI(text, history, config, filename=None, pbar=None, lock=None, mism else: subbedT = json.dumps({"Line1": protected_items}, indent=4, ensure_ascii=False) - # Check cache for this exact payload - cached_result = get_cached_translation(subbedT, config.language) + # Batch collect queues list payloads only. Single strings (speaker and + # variable names) translate live — modules memoize them and embed the + # results into later payloads, so they must resolve identically in both + # passes or the consume pass couldn't match the queued payload keys. + # This is the names-first phase; names are a tiny share of the volume. + queue_for_batch = batch_phase == "collect" and isinstance(tItem, list) + + # Check cache for this exact payload (the collect pass uses a + # non-blocking peek so no pending markers are left behind for the + # consume pass to wait on) + if queue_for_batch: + cached_result = peek_cached_translation(subbedT, config.language) + else: + cached_result = get_cached_translation(subbedT, config.language) if cached_result is not None: # In estimate mode, never replace tList[index] from cache — the cached value # may have been stored for a batch with a different number of skip_indices, @@ -1865,6 +2359,21 @@ def translateAI(text, history, config, filename=None, pbar=None, lock=None, mism # vocab_text is the per-batch matched vocabulary (dynamic). static_system, vocab_text, user = createContext(config, subbedT, formatType, history) + # Batch collect pass: queue the request (built exactly like a live one) + # instead of calling the API. The text stays untranslated; the consume + # pass fills it in from the fetched batch results. History carries the + # preceding source lines so the model still sees scene context. + if queue_for_batch: + numLines = len(clean_tItem) if isinstance(tItem, list) else 1 + params = buildClaudeRequest(static_system, user, history, formatType, + config.model, numLines, vocab_text=vocab_text) + queue_batch_request(subbedT, config.language, params) + if lock and pbar is not None: + with lock: + pbar.update(len(tItem)) + history = tItem[-config.maxHistory:] + continue + # Calculate estimate if in estimate mode if config.estimateMode: estimate = countTokens(static_system + vocab_text, user, history) @@ -1944,24 +2453,39 @@ def translateAI(text, history, config, filename=None, pbar=None, lock=None, mism if pbar: pbar.write(f"Retrying translation... (Attempt {attempt + 1}/{max_retries + 1})") - # Translate - try: - response = translateText(static_system, current_user, history, 0.05, formatType, config.model, numLines, vocab_text=vocab_text) - except Exception as api_err: - err_msg = f"[API_ERROR] {api_err}" - # Print to stdout so the GUI captures it immediately - print(err_msg, flush=True) - if pbar: - pbar.write(err_msg) - # Also write to the translation log file for persistence + # Translate — the consume pass tries the fetched batch result first; + # a missing or invalid result falls through to the live API. + from_batch = False + if batch_phase == "consume" and attempt == 0: + batch_result = take_batch_result(subbedT, config.language) + if batch_result is not None: + response = _AnthropicCompat( + batch_result.get("text", ""), + batch_result.get("prompt_tokens", 0) or 0, + batch_result.get("completion_tokens", 0) or 0, + batch_result.get("cache_read_input_tokens", 0) or 0, + batch_result.get("cache_creation_input_tokens", 0) or 0, + ) + from_batch = True + _write_request_debug_log("anthropic-batch", {"payload": subbedT}, response.usage) + if not from_batch: try: - Path(config.logFilePath).parent.mkdir(parents=True, exist_ok=True) - with open(config.logFilePath, "a", encoding="utf-8") as _lf: - _lf.write(f"{err_msg}\n") - _lf.flush() - except Exception: - pass - raise # Let retry decorator handle it + response = translateText(static_system, current_user, history, 0.05, formatType, config.model, numLines, vocab_text=vocab_text) + except Exception as api_err: + err_msg = f"[API_ERROR] {api_err}" + # Print to stdout so the GUI captures it immediately + print(err_msg, flush=True) + if pbar: + pbar.write(err_msg) + # Also write to the translation log file for persistence + try: + Path(config.logFilePath).parent.mkdir(parents=True, exist_ok=True) + with open(config.logFilePath, "a", encoding="utf-8") as _lf: + _lf.write(f"{err_msg}\n") + _lf.flush() + except Exception: + pass + raise # Let retry decorator handle it translatedText = response.choices[0].message.content last_raw_translation = translatedText @@ -1987,11 +2511,18 @@ def translateAI(text, history, config, filename=None, pbar=None, lock=None, mism batch_regular = max(0, batch_prompt_total - batch_cache_read - batch_cache_write) batch_output = getattr(usage, "completion_tokens", 0) or 0 - # Accumulate into per-file thread-local counters. - _thread_local.file_cache_read += batch_cache_read - _thread_local.file_cache_write += batch_cache_write - _thread_local.file_regular += batch_regular - _thread_local.file_output += batch_output + # Accumulate into per-file thread-local counters. Batch API + # usage is billed at 50% so it goes into its own counters. + if from_batch: + _thread_local.file_batch_read += batch_cache_read + _thread_local.file_batch_write += batch_cache_write + _thread_local.file_batch_regular += batch_regular + _thread_local.file_batch_output += batch_output + else: + _thread_local.file_cache_read += batch_cache_read + _thread_local.file_cache_write += batch_cache_write + _thread_local.file_regular += batch_regular + _thread_local.file_output += batch_output # --- Debug Token Logging --- if DEBUG: @@ -2217,6 +2748,10 @@ def translateAI(text, history, config, filename=None, pbar=None, lock=None, mism # Save cache after processing (for both estimate and translation modes) save_cache() + # Batch collect pass: merge this call's queued requests into the disk queue. + if batch_phase == "collect": + flush_batch_queue() + # For Claude: accumulate only this call's delta into the cross-thread total. # file_* accumulators hold full per-file totals; calculateCost() reads them. _is_claude_final = config.model and any(x in config.model.lower() for x in ("claude", "sonnet", "haiku", "opus")) @@ -2229,11 +2764,20 @@ def translateAI(text, history, config, filename=None, pbar=None, lock=None, mism _delta_cw = getattr(_thread_local, 'file_cache_write', 0) - _prev_cw _delta_reg = getattr(_thread_local, 'file_regular', 0) - _prev_reg _delta_out = getattr(_thread_local, 'file_output', 0) - _prev_out + _delta_bcr = getattr(_thread_local, 'file_batch_read', 0) - _prev_bcr + _delta_bcw = getattr(_thread_local, 'file_batch_write', 0) - _prev_bcw + _delta_breg = getattr(_thread_local, 'file_batch_regular', 0) - _prev_breg + _delta_bout = getattr(_thread_local, 'file_batch_output', 0) - _prev_bout _call_cost = ( _delta_cr * _br * 0.10 + _delta_cw * _br * 2.00 + _delta_reg * _br + - _delta_out * _or + _delta_out * _or + + # Batch API tokens: same rates, then the 50% batch discount. + (_delta_bcr * _br * 0.10 + + _delta_bcw * _br * 2.00 + + _delta_breg * _br + + _delta_bout * _or) * 0.50 ) global _global_accurate_cost with _global_accurate_cost_lock: