Compare commits

...

2 commits

Author SHA1 Message Date
300953a04d Proper batch gui that lets you manage batch history and reuse batches 2026-07-11 09:18:32 -05:00
8a0cda9e26 nah 2026-07-11 08:46:29 -05:00
8 changed files with 2254 additions and 357 deletions

102
TODO.md
View file

@ -6,106 +6,6 @@
- Add an optional **per-game custom prompt** in the UI that merges with the default (custom rules layered on top of shared instructions, not a full replacement unless explicitly chosen).
- Surface this in settings / project setup so each game folder can carry its own tone, terminology notes, or content warnings without forking the global prompt.
## Batch history tab
- Add a **Batch** tab (or dedicated panel) in the GUI to manage past Anthropic batch runs saved locally.
- Persist batch metadata locally (batch id, submit time, file set, status, cost estimate, etc.) alongside existing `log/batch_*.json` artifacts (`batch_state.json`, `batch_results.json`, `batch_requests.json`).
- Actions:
- **List batches** — browse all batches the tool has submitted or tracked for this project.
- **Redownload** — fetch results again from Anthropic for a completed batch (e.g. after a crash before consume, or to recover results). **Not in sketch yet.**
- **Cancel batch** — cancel an in-flight batch where the API allows it (`in_progress` → `canceling``ended`).
- **Usage / cost** — sum real billed tokens from batch results (input, output, cache read/write, thinking).
- Resume / link into the existing collect → submit → consume flow when a batch is still pending or fetched but not consumed.
### Sketch: CLI batch manager (adapt for DazedMTLTool)
Reference from sibling tooling (`batches.py`). Port to this repo using `_get_anthropic_client()` / `fetchTranslationBatches()` in `util/translation.py` and `log/batch_state.json` (maps to sketch's `tl/_batch_state.json`).
```python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
batches.py — list and cancel Anthropic Message Batches (the ones DazedMTLTool
submits). Handy when a run is stuck polling or you submitted by mistake.
set ANTHROPIC_API_KEY=sk-... # or key= in .env
python tooling/batches.py # list recent batches (default)
python tooling/batches.py list --limit 50
python tooling/batches.py cancel <batch_id> [<batch_id> ...]
python tooling/batches.py cancel --all # cancel every in-progress batch
python tooling/batches.py cancel --mine # cancel batch(es) in log/batch_state.json
python tooling/batches.py usage [batch_id] # real billed tokens + estimated cost
# TODO: redownload <batch_id> — re-fetch results into log/batch_results.json
Only an in-progress batch can be cancelled; cancellation finishes any in-flight
requests (in_progress -> canceling -> ended). Already-ended batches can't be
cancelled.
"""
import os
import sys
import argparse
import collections
# Adapt imports to DazedMTLTool:
# from util.translation import _get_anthropic_client, BATCH_STATE_FILE, _read_batch_file, fetchTranslationBatches
CANCELABLE = {"in_progress"}
def _counts(b):
rc = getattr(b, "request_counts", None)
if not rc:
return ""
parts = []
for k in ("processing", "succeeded", "errored", "canceled", "expired"):
v = getattr(rc, k, None)
if v:
parts.append(f"{k[:4]}={v}")
return " ".join(parts)
def _mine_batch_ids():
"""Return batch ids from log/batch_state.json (DazedMTLTool format)."""
# state = _read_batch_file(BATCH_STATE_FILE) or {}
# return [b["id"] for b in state.get("batches", [])]
...
def cmd_list(client, limit):
mine = set(_mine_batch_ids())
print(f"{'BATCH ID':<28} {'STATUS':<12} {'CREATED':<22} COUNTS")
print("-" * 90)
n = 0
for b in client.messages.batches.list(limit=limit):
n += 1
mark = " <- yours" if b.id in mine else ""
created = str(getattr(b, "created_at", "") or "")[:22]
print(f"{b.id:<28} {b.processing_status:<12} {created:<22} {_counts(b)}{mark}")
...
def cmd_cancel(client, ids, do_all, do_mine, limit):
# targets from ids, --mine (log/batch_state.json), or --all (in_progress only)
# client.messages.batches.cancel(bid)
...
def cmd_usage(client, bid, model):
# Sum usage across client.messages.batches.results(bid)
# Price with getPricingConfig(model) + 50% batch discount on I/O
...
# cmd_redownload(client, bid):
# # TODO: client.messages.batches.results(bid) -> merge into log/batch_results.json
# # Wire custom_id map from batch_state.json entries
# ...
```
**GUI tab (later):** same operations as buttons on a saved local history table — list, cancel, usage, redownload, resume consume.
## Local translation cache (game retranslate)
- Persist every successful translation locally so a full retranslate from scratch can reuse prior results instead of paying again.
@ -117,5 +17,3 @@ def cmd_usage(client, bid, model):
- **Retranslate workflow:** when the user re-imports fresh JSON and runs Translate again, cache hits should fill `translated/` automatically for unchanged source lines; only new or edited source text should incur API cost.
- **UI (later):** cache stats (hits / misses / estimated savings), export/import, clear cache for one game, optional “prefer cache” toggle.
- **Invalidation:** optional metadata (model id, prompt version) in the key or entry if we need to avoid reusing stale translations after prompt or model changes.
## According to someone from f95, the patcher breaks if a folder's name has an apostrophe

467
gui/batch_tab.py Normal file
View file

@ -0,0 +1,467 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Batch history tab - list / cancel / usage / redownload / resume."""
from __future__ import annotations
import os
from pathlib import Path
from PyQt5.QtCore import Qt, QThread, pyqtSignal, QTimer
from PyQt5.QtWidgets import (
QWidget,
QVBoxLayout,
QHBoxLayout,
QPushButton,
QLabel,
QTableWidget,
QTableWidgetItem,
QHeaderView,
QTextEdit,
QMessageBox,
QAbstractItemView,
QSplitter,
)
from dotenv import load_dotenv
class _BatchOpsWorker(QThread):
"""Run a blocking batch-history callable off the UI thread."""
done = pyqtSignal(bool, str, object) # ok, message, payload
log = pyqtSignal(str)
def __init__(self, task, project_root: Path):
super().__init__()
self._task = task
self._project_root = project_root
def run(self):
old = os.getcwd()
try:
os.chdir(str(self._project_root))
load_dotenv()
ok, msg, payload = self._task(self.log.emit)
self.done.emit(bool(ok), str(msg), payload)
except Exception as exc:
import traceback
self.log.emit(traceback.format_exc())
self.done.emit(False, f"Error: {exc}", None)
finally:
try:
os.chdir(old)
except Exception:
pass
class BatchTab(QWidget):
"""Manage past / in-flight Anthropic Message Batches for this project."""
COLUMNS = [
"Batch ID",
"Status",
"Created",
"Requests",
"Model",
"Estimate",
"Actual $",
"Files",
]
def __init__(self, parent=None):
super().__init__(parent)
self.parent_window = parent
self.project_root = Path(
getattr(parent, "project_root", None) or Path(__file__).resolve().parent.parent
)
self._worker: _BatchOpsWorker | None = None
self._rows: list[dict] = []
self._loaded_once = False
self._init_ui()
def showEvent(self, event):
super().showEvent(event)
# Refresh every time the Batches page is shown so live status stays current.
if self._worker is not None and self._worker.isRunning():
return
self._loaded_once = True
QTimer.singleShot(0, lambda: self.refresh_list(live=True))
def _init_ui(self):
layout = QVBoxLayout(self)
layout.setContentsMargins(16, 16, 16, 16)
layout.setSpacing(10)
title = QLabel("Batch History")
title.setStyleSheet("color:#ffffff;font-size:18px;font-weight:bold;")
layout.addWidget(title)
hint = QLabel(
"Local Anthropic Message Batch runs for this project. "
"Cancel / redownload / usage never submit a new batch."
)
hint.setWordWrap(True)
hint.setStyleSheet("color:#9d9d9d;font-size:12px;")
layout.addWidget(hint)
toolbar = QHBoxLayout()
toolbar.setSpacing(8)
self.refresh_btn = QPushButton("Refresh")
self.refresh_btn.setToolTip("Reload local history and refresh live status for active batches")
self.refresh_btn.clicked.connect(lambda: self.refresh_list(live=True))
self.cancel_btn = QPushButton("Cancel")
self.cancel_btn.setToolTip("Cancel selected in-progress batch(es)")
self.cancel_btn.clicked.connect(self.cancel_selected)
self.usage_btn = QPushButton("Usage")
self.usage_btn.setToolTip("Sum real billed tokens for the selected ended batch")
self.usage_btn.clicked.connect(self.usage_selected)
self.redownload_btn = QPushButton("Redownload")
self.redownload_btn.setToolTip("Re-fetch results into log/batch_results.json (no re-submit)")
self.redownload_btn.clicked.connect(self.redownload_selected)
self.resume_btn = QPushButton("Resume")
self.resume_btn.setToolTip("Activate this batch and continue poll/consume on the Translation tab")
self.resume_btn.clicked.connect(self.resume_selected)
for btn in (
self.refresh_btn,
self.cancel_btn,
self.usage_btn,
self.redownload_btn,
self.resume_btn,
):
btn.setStyleSheet(
"QPushButton{background-color:#3c3c3c;color:#cccccc;border:1px solid #555555;"
"border-radius:4px;padding:6px 12px;}"
"QPushButton:hover{border-color:#007acc;}"
"QPushButton:disabled{color:#666666;border-color:#444444;}"
)
toolbar.addWidget(btn)
toolbar.addStretch()
layout.addLayout(toolbar)
splitter = QSplitter(Qt.Vertical)
self.table = QTableWidget(0, len(self.COLUMNS))
self.table.setHorizontalHeaderLabels(self.COLUMNS)
self.table.setSelectionBehavior(QAbstractItemView.SelectRows)
self.table.setSelectionMode(QAbstractItemView.ExtendedSelection)
self.table.setEditTriggers(QAbstractItemView.NoEditTriggers)
self.table.setAlternatingRowColors(True)
self.table.verticalHeader().setVisible(False)
self.table.setStyleSheet(
"QTableWidget{background-color:#1e1e1e;color:#cccccc;gridline-color:#3a3a3a;"
"alternate-background-color:#252526;}"
"QHeaderView::section{background-color:#2d2d30;color:#cccccc;padding:4px;"
"border:1px solid #3a3a3a;}"
)
header = self.table.horizontalHeader()
header.setSectionResizeMode(0, QHeaderView.Stretch)
for i in range(1, len(self.COLUMNS)):
header.setSectionResizeMode(i, QHeaderView.ResizeToContents)
self.table.itemSelectionChanged.connect(self._update_button_states)
splitter.addWidget(self.table)
self.log = QTextEdit()
self.log.setReadOnly(True)
self.log.setPlaceholderText("Operation log…")
self.log.setStyleSheet(
"QTextEdit{background-color:#1e1e1e;color:#cccccc;border:1px solid #3a3a3a;"
"font-family:monospace;font-size:12px;}"
)
splitter.addWidget(self.log)
splitter.setStretchFactor(0, 3)
splitter.setStretchFactor(1, 1)
layout.addWidget(splitter, 1)
self._update_button_states()
def _append_log(self, text: str):
if not text:
return
self.log.append(text.rstrip())
def _selected_entries(self) -> list[dict]:
rows = sorted({idx.row() for idx in self.table.selectedIndexes()})
out = []
for r in rows:
if 0 <= r < len(self._rows):
out.append(self._rows[r])
return out
def _update_button_states(self):
busy = self._worker is not None and self._worker.isRunning()
entries = self._selected_entries()
n = len(entries)
self.refresh_btn.setEnabled(not busy)
self.cancel_btn.setEnabled(
not busy and n >= 1 and any(
(e.get("api_status") == "in_progress") or (e.get("status") == "submitted")
for e in entries
)
)
self.usage_btn.setEnabled(not busy and n == 1)
self.redownload_btn.setEnabled(
not busy
and n == 1
and entries[0].get("status") in ("ended", "fetched", "consumed", "submitted", "canceling")
)
self.resume_btn.setEnabled(
not busy
and n == 1
and entries[0].get("status")
in ("submitted", "canceling", "ended", "fetched")
)
def _set_busy(self, busy: bool):
if busy:
for btn in (
self.refresh_btn,
self.cancel_btn,
self.usage_btn,
self.redownload_btn,
self.resume_btn,
):
btn.setEnabled(False)
else:
self._update_button_states()
def _clear_worker(self):
if self.sender() is self._worker:
self._worker = None
def _run_task(self, task, on_done=None):
if self._worker is not None and self._worker.isRunning():
QMessageBox.information(self, "Busy", "A batch operation is already running.")
return
self._set_busy(True)
worker = _BatchOpsWorker(task, self.project_root)
self._worker = worker
def _finished(ok, msg, payload):
self._set_busy(False)
if msg:
self._append_log(msg)
if on_done:
try:
on_done(ok, msg, payload)
except Exception as exc:
self._append_log(f"[BATCH] UI update failed: {exc}")
worker.log.connect(self._append_log)
worker.done.connect(_finished)
# Release the reference only after run() returns. Clearing in the done
# slot aborts with "QThread: Destroyed while thread is still running".
worker.finished.connect(worker.deleteLater)
worker.finished.connect(self._clear_worker)
worker.start()
def _populate_table(self, entries: list[dict]):
self._rows = list(entries or [])
self.table.setRowCount(len(self._rows))
for r, entry in enumerate(self._rows):
est = entry.get("cost_estimate") or {}
est_s = ""
if isinstance(est, dict) and est.get("batch_cached_cost") is not None:
try:
est_s = f"${float(est['batch_cached_cost']):.2f}"
except (TypeError, ValueError):
est_s = ""
actual = entry.get("actual_cost")
actual_s = f"${actual:.4f}" if isinstance(actual, (int, float)) else ""
files = entry.get("file_set") or []
if not isinstance(files, list):
files = [str(files)]
files_s = ", ".join(str(f) for f in files[:3])
if len(files) > 3:
files_s += f" (+{len(files) - 3})"
values = [
str(entry.get("id") or ""),
str(entry.get("status") or ""),
str(entry.get("created_at") or "")[:19],
str(entry.get("request_count") or ""),
str(entry.get("model") or ""),
est_s,
actual_s,
files_s,
]
for c, val in enumerate(values):
item = QTableWidgetItem(val)
if c == 0:
item.setData(Qt.UserRole, entry.get("id"))
self.table.setItem(r, c, item)
self._update_button_states()
def refresh_list(self, live: bool = True):
def task(log):
from util.batch_history import list_local_batches
log("[BATCH] Loading local history...")
entries = list_local_batches(refresh_live=live)
log(f"[BATCH] {len(entries)} batch(es) in history.")
return True, f"Loaded {len(entries)} batch(es).", entries
def done(ok, _msg, payload):
if ok and isinstance(payload, list):
self._populate_table(payload)
self._run_task(task, on_done=done)
def cancel_selected(self):
entries = self._selected_entries()
if not entries:
return
ids = [e["id"] for e in entries if e.get("id")]
reply = QMessageBox.question(
self,
"Cancel Batch?",
"Cancel the selected in-progress batch(es)?\n\n"
"Anthropic may still finish and bill requests that were already "
"in flight when cancel is received.\n"
"This does not submit a new batch.",
QMessageBox.Yes | QMessageBox.No,
)
if reply != QMessageBox.Yes:
return
def task(log):
from util.batch_history import cancel_batches
log(f"[BATCH] Canceling {len(ids)} batch(es)…")
results = cancel_batches(ids)
lines = []
for r in results:
if r.get("ok"):
lines.append(f" canceled {r['id']} -> {r.get('api_status')}")
else:
lines.append(f" failed {r['id']}: {r.get('error')}")
for line in lines:
log(line)
return True, "Cancel finished.", results
def done(ok, _msg, _payload):
self.refresh_list(live=False)
self._run_task(task, on_done=done)
def usage_selected(self):
entries = self._selected_entries()
if len(entries) != 1:
return
bid = entries[0]["id"]
def task(log):
from util.batch_history import usage_for_batch
log(f"[BATCH] Computing usage for {bid}")
info = usage_for_batch(bid)
u = info.get("usage") or {}
log(
f"[BATCH] tokens: in={u.get('input_tokens', 0)} out={u.get('output_tokens', 0)} "
f"cache_read={u.get('cache_read_input_tokens', 0)} "
f"cache_write={u.get('cache_creation_input_tokens', 0)} "
f"thinking={u.get('thinking_tokens', 0)}"
)
cost = info.get("actual_cost")
if cost is not None:
log(f"[BATCH] estimated billed cost (batch 50% off): ${cost:.4f}")
return True, "Usage updated.", info
def done(ok, _msg, _payload):
self.refresh_list(live=False)
self._run_task(task, on_done=done)
def redownload_selected(self):
entries = self._selected_entries()
if len(entries) != 1:
return
bid = entries[0]["id"]
reply = QMessageBox.question(
self,
"Redownload Results?",
f"Re-fetch results for {bid} into log/batch_results.json?\n\n"
"This does not create a new batch or re-collect.",
QMessageBox.Yes | QMessageBox.No,
)
if reply != QMessageBox.Yes:
return
def task(log):
from util.batch_history import redownload_batch
log(f"[BATCH] Redownloading {bid}")
info = redownload_batch(bid)
log(
f"[BATCH] redownload ok={info.get('succeeded')} err={info.get('errored')} "
f"cost≈{info.get('actual_cost')}"
)
return True, "Redownload finished.", info
def done(ok, msg, payload):
self.refresh_list(live=False)
if not ok:
return
reply = QMessageBox.question(
self,
"Resume Consume?",
"Results are ready locally. Switch to Translation and resume consume now?",
QMessageBox.Yes | QMessageBox.No,
)
if reply == QMessageBox.Yes:
self._resume_with_state("fetched", entries[0])
self._run_task(task, on_done=done)
def resume_selected(self):
entries = self._selected_entries()
if len(entries) != 1:
return
entry = entries[0]
bid = entry["id"]
def task(log):
from util.batch_history import activate_for_resume
log(f"[BATCH] Activating {bid} for resume…")
state = activate_for_resume(bid)
log(f"[BATCH] active resume state: {state}")
return True, f"Activated ({state}).", {"state": state, "entry": entry}
def done(ok, _msg, payload):
self.refresh_list(live=False)
if not ok or not payload:
return
self._resume_with_state(payload["state"], payload.get("entry") or entry)
self._run_task(task, on_done=done)
def _resume_with_state(self, resume_state: str, entry: dict):
parent = self.parent_window
if parent is None or not hasattr(parent, "translation_tab"):
QMessageBox.warning(self, "Resume", "Translation tab is not available.")
return
tt = parent.translation_tab
file_set = entry.get("file_set") or []
if file_set and hasattr(tt, "select_files_by_name"):
tt.select_files_by_name(file_set)
# Switch to Translation page (index 0).
if hasattr(parent, "switch_page"):
parent.switch_page(0)
reply = QMessageBox.question(
self,
"Start Resume?",
f"Start Batch Translate resume ({resume_state}) on the Translation tab?\n\n"
"This will not clear batch files or submit a new batch.",
QMessageBox.Yes | QMessageBox.No,
)
if reply != QMessageBox.Yes:
return
tt.start_translation(forced_resume_state=resume_state)

View file

@ -595,6 +595,7 @@ from gui.config_tab import ConfigTab
from gui.translation_tab import TranslationTab
from gui.workflow_tab import WorkflowTab
from gui.wolf_workflow_tab import WolfWorkflowTab
from gui.batch_tab import BatchTab
class DazedMTLGUI(QMainWindow):
"""Main GUI window for the DazedMTLTool."""
@ -830,9 +831,16 @@ class DazedMTLGUI(QMainWindow):
sidebar_layout.addWidget(btn_workflow)
self.nav_buttons.append(btn_workflow)
# Configuration button (third)
# Batch history button (third)
btn_batches = self.create_nav_button("📦", "Batches")
btn_batches.setToolTip("Batches — Anthropic Message Batch history")
btn_batches.clicked.connect(lambda: self.switch_page(2))
sidebar_layout.addWidget(btn_batches)
self.nav_buttons.append(btn_batches)
# Configuration button (fourth)
btn_config = self.create_nav_button("⚙️", "Configuration")
btn_config.clicked.connect(lambda: self.switch_page(2))
btn_config.clicked.connect(lambda: self.switch_page(3))
sidebar_layout.addWidget(btn_config)
self.nav_buttons.append(btn_config)
@ -884,6 +892,8 @@ class DazedMTLGUI(QMainWindow):
def setup_tabs(self):
"""Set up all the tabs in the interface."""
self.project_root = PROJECT_ROOT
# Translation Execution Tab (index 0)
self.translation_tab = TranslationTab(self)
self.content_stack.addWidget(self.translation_tab)
@ -892,7 +902,11 @@ class DazedMTLGUI(QMainWindow):
# RPGMaker and Wolf guided panels while keeping a single sidebar button.
self.content_stack.addWidget(self._create_workflow_container())
# Configuration Tab (index 2)
# Batch History Tab (index 2)
self.batch_tab = BatchTab(self)
self.content_stack.addWidget(self.batch_tab)
# Configuration Tab (index 3)
self.config_tab = ConfigTab()
self.config_tab.config_changed.connect(self.on_config_changed)
self.content_stack.addWidget(self.config_tab)

File diff suppressed because it is too large Load diff

View file

@ -1,5 +1,6 @@
import sys
import os
import time
import traceback
import datetime
from pathlib import Path
@ -128,7 +129,18 @@ def main():
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 resume_state == "queued":
prompt = (
"A previous collect finished but was not submitted (queued). "
"Resume and submit that queue? (y/n)\n"
"(n discards the queue and re-collects - live collect charges again)\n"
)
else:
prompt = (
f"A previous batch run was interrupted ({resume_state}). Resume it? (y/n)\n"
"(n discards it and can bill again)\n"
)
confirm = input(prompt).strip().lower()
if confirm == "n":
resume_state = None
@ -322,21 +334,72 @@ files to translate are in the /files folder and that you picked the right game e
tqdm.write("[BATCH] No requests queued — nothing needed the API.")
run_consume = False
else:
estimateBatchCost()
est = 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.")
tqdm.write(
"[BATCH] Not submitted. The queue is kept in log/batch_requests.json "
"(resume Batch Translate later to submit without re-collecting)."
)
return
runTranslationBatches(poll)
from util.translation import submitTranslationBatches, checkTranslationBatches, fetchTranslationBatches
if not submitTranslationBatches(cost_estimate=est):
run_consume = False
else:
tqdm.write(
f"[BATCH] polling every {poll}s (Ctrl-C is safe - resume later)..."
)
while not checkTranslationBatches():
time.sleep(poll)
fetchTranslationBatches()
elif resume_state == "queued":
tqdm.write(
Fore.CYAN
+ "[BATCH] Resuming queued requests (skipping re-collect)..."
+ Fore.RESET
)
if pendingBatchRequests() == 0:
tqdm.write("[BATCH] Queue is empty - nothing to submit.")
run_consume = False
else:
est = 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. Queue kept.")
return
from util.translation import submitTranslationBatches, checkTranslationBatches, fetchTranslationBatches
if not submitTranslationBatches(cost_estimate=est):
run_consume = False
else:
tqdm.write(
f"[BATCH] polling every {poll}s (Ctrl-C is safe - resume later)..."
)
while not checkTranslationBatches():
time.sleep(poll)
fetchTranslationBatches()
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
else: # "fetched" - results already downloaded, just write the files
tqdm.write(Fore.CYAN + "[BATCH] Resuming from fetched results..." + Fore.RESET)
if run_consume:
try:
from util.batch_history import missing_result_count
present, expected = missing_result_count()
if expected and present < expected:
tqdm.write(
Fore.YELLOW
+ f"[BATCH] WARNING: only {present}/{expected} results present. "
"Missing keys fall back to the live API (full price)."
+ Fore.RESET
)
except Exception:
pass
# 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)

265
tests/test_batch_history.py Normal file
View file

@ -0,0 +1,265 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Tests for durable batch history and spend-safe ops."""
from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest import mock
import util.translation as T
import util.batch_history as BH
class BatchHistoryTestBase(unittest.TestCase):
"""Isolate batch JSON files to a temp dir."""
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
tmp = Path(self._tmp.name)
self._orig = {
"QUEUE": T.BATCH_QUEUE_FILE,
"STATE": T.BATCH_STATE_FILE,
"RESULTS": T.BATCH_RESULTS_FILE,
"LOCK": T.BATCH_LOCK_FILE,
"HISTORY": BH.BATCH_HISTORY_FILE,
"results_mem": T._batch_results,
"pending": dict(T._batch_queue_pending),
}
T.BATCH_QUEUE_FILE = tmp / "batch_requests.json"
T.BATCH_STATE_FILE = tmp / "batch_state.json"
T.BATCH_RESULTS_FILE = tmp / "batch_results.json"
T.BATCH_LOCK_FILE = tmp / "batch_files.lock"
BH.BATCH_HISTORY_FILE = tmp / "batch_history.json"
# batch_history imports queue/state/results paths at call time via T.* —
# but it also imported BATCH_* as names. Rebind module-level aliases.
BH.BATCH_QUEUE_FILE = T.BATCH_QUEUE_FILE
BH.BATCH_STATE_FILE = T.BATCH_STATE_FILE
BH.BATCH_RESULTS_FILE = T.BATCH_RESULTS_FILE
T._batch_results = None
T._batch_queue_pending = {}
def tearDown(self):
T.BATCH_QUEUE_FILE = self._orig["QUEUE"]
T.BATCH_STATE_FILE = self._orig["STATE"]
T.BATCH_RESULTS_FILE = self._orig["RESULTS"]
T.BATCH_LOCK_FILE = self._orig["LOCK"]
BH.BATCH_HISTORY_FILE = self._orig["HISTORY"]
BH.BATCH_QUEUE_FILE = T.BATCH_QUEUE_FILE
BH.BATCH_STATE_FILE = T.BATCH_STATE_FILE
BH.BATCH_RESULTS_FILE = T.BATCH_RESULTS_FILE
T._batch_results = self._orig["results_mem"]
T._batch_queue_pending = self._orig["pending"]
self._tmp.cleanup()
class BatchRunStateTests(BatchHistoryTestBase):
def test_none_when_empty(self):
self.assertIsNone(T.batchRunState())
def test_queued_when_only_queue(self):
T._write_batch_file(T.BATCH_QUEUE_FILE, {"k1": {"payload": "x", "language": "English", "params": {}}})
self.assertEqual(T.batchRunState(), "queued")
def test_submitted_when_state_has_batches(self):
T._write_batch_file(
T.BATCH_STATE_FILE,
{"batches": [{"id": "msgbatch_1", "custom_ids": {"req-000000": "k1"}}]},
)
self.assertEqual(T.batchRunState(), "submitted")
def test_fetched_when_results_present(self):
T._write_batch_file(T.BATCH_RESULTS_FILE, {"k1": {"text": "hi"}})
self.assertEqual(T.batchRunState(), "fetched")
def test_fetched_when_state_status_fetched(self):
T._write_batch_file(
T.BATCH_STATE_FILE,
{"status": "fetched", "batch_ids": ["msgbatch_1"], "batches": []},
)
self.assertEqual(T.batchRunState(), "fetched")
class HistorySurvivalTests(BatchHistoryTestBase):
def test_history_survives_fetch_marker_and_clear(self):
custom_ids = {"req-000000": "cachekey1", "req-000001": "cachekey2"}
BH.record_submit(
[{"id": "msgbatch_abc", "custom_ids": custom_ids}],
model="claude-sonnet-4-5",
file_set=["Map001.json"],
cost_estimate={"batch_cached_cost": 1.23, "model": "claude-sonnet-4-5"},
)
BH.record_fetch(["msgbatch_abc"], succeeded=2, errored=0, usage={"input_tokens": 10}, actual_cost=0.5)
T._write_batch_file(T.BATCH_RESULTS_FILE, {"cachekey1": {"text": "A"}, "cachekey2": {"text": "B"}})
T._write_batch_file(
T.BATCH_STATE_FILE,
{"status": "fetched", "batch_ids": ["msgbatch_abc"], "batches": []},
)
T.clearBatchFiles()
# Active files gone…
self.assertFalse(T.BATCH_RESULTS_FILE.exists())
self.assertFalse(T.BATCH_STATE_FILE.exists())
# …but history retains custom_ids and is marked consumed.
history = BH.read_history()
entry = history["batches"][0]
self.assertEqual(entry["id"], "msgbatch_abc")
self.assertEqual(entry["custom_ids"], custom_ids)
self.assertEqual(entry["status"], BH.STATUS_CONSUMED)
self.assertEqual(entry["file_set"], ["Map001.json"])
def test_clear_does_not_wipe_history_file(self):
BH.upsert_history_entry("msgbatch_keep", status=BH.STATUS_SUBMITTED, custom_ids={"a": "b"})
T.clearBatchFiles()
self.assertTrue(BH.BATCH_HISTORY_FILE.exists())
self.assertEqual(len(BH.read_history()["batches"]), 1)
class RedownloadTests(BatchHistoryTestBase):
def test_redownload_rebuilds_results_from_custom_ids(self):
custom_ids = {"req-000000": "keyA", "req-000001": "keyB"}
BH.upsert_history_entry(
"msgbatch_rd",
status=BH.STATUS_ENDED,
model="claude-sonnet-4-5",
custom_ids=custom_ids,
request_count=2,
)
usage = SimpleNamespace(
input_tokens=100,
output_tokens=50,
cache_read_input_tokens=10,
cache_creation_input_tokens=20,
thinking_tokens=5,
)
msg = SimpleNamespace(
content=[SimpleNamespace(text='{"Line1":"Hi"}')],
usage=usage,
)
ok_result = SimpleNamespace(type="succeeded", message=msg)
row = SimpleNamespace(custom_id="req-000000", result=ok_result)
row2_usage = SimpleNamespace(
input_tokens=80,
output_tokens=40,
cache_read_input_tokens=0,
cache_creation_input_tokens=0,
thinking_tokens=0,
)
msg2 = SimpleNamespace(content=[SimpleNamespace(text='{"Line1":"Yo"}')], usage=row2_usage)
row2 = SimpleNamespace(custom_id="req-000001", result=SimpleNamespace(type="succeeded", message=msg2))
fake_batch = SimpleNamespace(id="msgbatch_rd", processing_status="ended")
client = mock.MagicMock()
client.messages.batches.retrieve.return_value = fake_batch
client.messages.batches.results.return_value = [row, row2]
with mock.patch.object(BH, "_get_anthropic_client", return_value=client):
with mock.patch.object(BH, "getPricingConfig", return_value={"inputAPICost": 3.0, "outputAPICost": 15.0}):
info = BH.redownload_batch("msgbatch_rd")
self.assertEqual(info["succeeded"], 2)
results = T._read_batch_file(T.BATCH_RESULTS_FILE)
self.assertIn("keyA", results)
self.assertIn("keyB", results)
self.assertEqual(results["keyA"]["text"], '{"Line1":"Hi"}')
state = T._read_batch_file(T.BATCH_STATE_FILE)
self.assertEqual(state.get("status"), "fetched")
self.assertEqual(T.batchRunState(), "fetched")
entry = BH.read_history()["batches"][0]
self.assertEqual(entry["status"], BH.STATUS_FETCHED)
self.assertEqual(entry["custom_ids"], custom_ids)
class CancelTests(BatchHistoryTestBase):
def test_cancel_updates_history_and_active_state(self):
T._write_batch_file(
T.BATCH_STATE_FILE,
{
"batches": [
{"id": "msgbatch_c1", "custom_ids": {"req-000000": "k"}},
{"id": "msgbatch_c2", "custom_ids": {"req-000000": "k2"}},
]
},
)
BH.upsert_history_entry("msgbatch_c1", status=BH.STATUS_SUBMITTED, custom_ids={"req-000000": "k"})
BH.upsert_history_entry("msgbatch_c2", status=BH.STATUS_SUBMITTED, custom_ids={"req-000000": "k2"})
before = SimpleNamespace(id="msgbatch_c1", processing_status="in_progress")
after = SimpleNamespace(id="msgbatch_c1", processing_status="canceling")
client = mock.MagicMock()
client.messages.batches.retrieve.return_value = before
client.messages.batches.cancel.return_value = after
with mock.patch.object(BH, "_get_anthropic_client", return_value=client):
results = BH.cancel_batches(["msgbatch_c1"])
self.assertTrue(results[0]["ok"])
entry = next(e for e in BH.read_history()["batches"] if e["id"] == "msgbatch_c1")
self.assertEqual(entry["status"], BH.STATUS_CANCELING)
state = T._read_batch_file(T.BATCH_STATE_FILE)
ids = [b["id"] for b in state.get("batches", [])]
self.assertNotIn("msgbatch_c1", ids)
self.assertIn("msgbatch_c2", ids)
class UsageTests(BatchHistoryTestBase):
def test_usage_sums_cache_and_thinking(self):
BH.upsert_history_entry(
"msgbatch_u",
status=BH.STATUS_ENDED,
model="claude-sonnet-4-5",
custom_ids={"req-000000": "k"},
)
usage = SimpleNamespace(
input_tokens=1000,
output_tokens=200,
cache_read_input_tokens=500,
cache_creation_input_tokens=100,
thinking_tokens=50,
)
msg = SimpleNamespace(content=[SimpleNamespace(text="ok")], usage=usage)
row = SimpleNamespace(custom_id="req-000000", result=SimpleNamespace(type="succeeded", message=msg))
client = mock.MagicMock()
client.messages.batches.retrieve.return_value = SimpleNamespace(processing_status="ended")
client.messages.batches.results.return_value = [row]
with mock.patch.object(BH, "_get_anthropic_client", return_value=client):
with mock.patch.object(BH, "getPricingConfig", return_value={"inputAPICost": 3.0, "outputAPICost": 15.0}):
info = BH.usage_for_batch("msgbatch_u")
u = info["usage"]
self.assertEqual(u["input_tokens"], 1000)
self.assertEqual(u["output_tokens"], 200)
self.assertEqual(u["cache_read_input_tokens"], 500)
self.assertEqual(u["cache_creation_input_tokens"], 100)
self.assertEqual(u["thinking_tokens"], 50)
self.assertIsInstance(info["actual_cost"], float)
self.assertGreater(info["actual_cost"], 0)
entry = BH.read_history()["batches"][0]
self.assertEqual(entry["usage"]["thinking_tokens"], 50)
class ActivateResumeTests(BatchHistoryTestBase):
def test_activate_submitted_restores_state(self):
custom_ids = {"req-000000": "k"}
BH.upsert_history_entry(
"msgbatch_act",
status=BH.STATUS_SUBMITTED,
custom_ids=custom_ids,
model="claude-sonnet-4-5",
file_set=["a.json"],
)
state = BH.activate_for_resume("msgbatch_act")
self.assertEqual(state, "submitted")
disk = T._read_batch_file(T.BATCH_STATE_FILE)
self.assertEqual(disk["batches"][0]["id"], "msgbatch_act")
self.assertEqual(disk["batches"][0]["custom_ids"], custom_ids)
if __name__ == "__main__":
unittest.main()

620
util/batch_history.py Normal file
View file

@ -0,0 +1,620 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Durable Anthropic Message Batch history and spend-safe management ops.
Active run files (queue / state / results) still live in util.translation.
This module keeps an append-only index so batch ids and custom_id maps survive
fetch and clear, enabling cancel / usage / redownload / resume without
re-submitting (and re-billing) work.
"""
from __future__ import annotations
import copy
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable, Optional
from util.translation import (
BATCH_LOCK,
BATCH_QUEUE_FILE,
BATCH_RESULTS_FILE,
BATCH_STATE_FILE,
_batch_file_lock,
_get_anthropic_client,
_read_batch_file,
_write_batch_file,
getPricingConfig,
)
BATCH_HISTORY_FILE = Path("log/batch_history.json")
# Local lifecycle statuses (not 1:1 with Anthropic processing_status).
STATUS_QUEUED = "queued"
STATUS_SUBMITTED = "submitted"
STATUS_CANCELING = "canceling"
STATUS_ENDED = "ended"
STATUS_FETCHED = "fetched"
STATUS_CONSUMED = "consumed"
STATUS_CANCELED = "canceled"
STATUS_ERROR = "error"
TERMINAL_STATUSES = frozenset({
STATUS_CONSUMED,
STATUS_CANCELED,
STATUS_ERROR,
})
CANCELABLE_API = frozenset({"in_progress"})
def _utc_now() -> str:
return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
def _empty_history() -> dict:
return {"batches": []}
def _read_history_unlocked() -> dict:
"""Read history under an existing batch file lock."""
data = _read_batch_file(BATCH_HISTORY_FILE)
if not data:
return _empty_history()
if not isinstance(data.get("batches"), list):
print(f"[BATCH] Corrupt history (missing batches list): {BATCH_HISTORY_FILE}", flush=True)
return _empty_history()
return data
def _write_history_unlocked(data: dict) -> None:
_write_batch_file(BATCH_HISTORY_FILE, data)
def read_history() -> dict:
"""Return the full history document (copy-safe for callers)."""
with _batch_file_lock():
return copy.deepcopy(_read_history_unlocked())
def _find_entry(history: dict, batch_id: str) -> Optional[dict]:
for entry in history.get("batches", []):
if entry.get("id") == batch_id:
return entry
return None
def upsert_history_entry(batch_id: str, **fields: Any) -> dict:
"""Create or update one history row. Returns the updated entry (copy)."""
with BATCH_LOCK:
with _batch_file_lock():
history = _read_history_unlocked()
entry = _find_entry(history, batch_id)
now = _utc_now()
if entry is None:
entry = {
"id": batch_id,
"created_at": now,
"updated_at": now,
"status": STATUS_SUBMITTED,
"model": "",
"request_count": 0,
"file_set": [],
"cost_estimate": None,
"custom_ids": {},
"usage": None,
"actual_cost": None,
"notes": "",
"request_counts": None,
"api_status": None,
}
history.setdefault("batches", []).append(entry)
for key, value in fields.items():
if value is not None or key in ("notes", "usage", "actual_cost", "cost_estimate", "request_counts"):
entry[key] = value
entry["updated_at"] = now
_write_history_unlocked(history)
return copy.deepcopy(entry)
def record_submit(
batches: list[dict],
*,
model: str = "",
file_set: Optional[list] = None,
cost_estimate: Optional[dict] = None,
) -> None:
"""Record newly submitted Anthropic batches into durable history."""
file_set = list(file_set or [])
for info in batches:
bid = info.get("id")
if not bid:
continue
custom_ids = dict(info.get("custom_ids") or {})
upsert_history_entry(
bid,
status=STATUS_SUBMITTED,
model=model or (cost_estimate or {}).get("model") or "",
request_count=len(custom_ids),
file_set=file_set,
cost_estimate=copy.deepcopy(cost_estimate) if cost_estimate else None,
custom_ids=custom_ids,
api_status="in_progress",
notes="",
)
def record_fetch(
batch_ids: Iterable[str],
*,
succeeded: int = 0,
errored: int = 0,
usage: Optional[dict] = None,
actual_cost: Optional[float] = None,
) -> None:
"""Mark batches as fetched after results land locally."""
note = f"fetched ok={succeeded} err={errored}"
for bid in batch_ids:
fields = {
"status": STATUS_FETCHED,
"api_status": "ended",
"notes": note,
}
if usage is not None:
fields["usage"] = copy.deepcopy(usage)
if actual_cost is not None:
fields["actual_cost"] = actual_cost
upsert_history_entry(bid, **fields)
def mark_batches_consumed(batch_ids: Iterable[str]) -> None:
"""Mark history rows consumed after a successful consume + clear."""
for bid in batch_ids:
upsert_history_entry(bid, status=STATUS_CONSUMED, notes="consumed")
def mark_batches_canceled(batch_ids: Iterable[str], *, api_status: str = "canceling") -> None:
local = STATUS_CANCELED if api_status == "ended" else STATUS_CANCELING
for bid in batch_ids:
upsert_history_entry(
bid,
status=local,
api_status=api_status,
notes=f"cancel requested ({api_status})",
)
def list_local_batches(*, refresh_live: bool = False) -> list[dict]:
"""Return history entries newest-first. Optionally refresh non-terminal via API."""
history = read_history()
entries = list(history.get("batches") or [])
if refresh_live:
for entry in entries:
status = entry.get("status")
if status in TERMINAL_STATUSES or status == STATUS_CONSUMED:
continue
if status in (STATUS_FETCHED, STATUS_QUEUED):
continue
try:
refresh_batch_status(entry["id"])
except Exception as exc:
upsert_history_entry(entry["id"], notes=f"refresh failed: {exc}")
history = read_history()
entries = list(history.get("batches") or [])
entries.sort(key=lambda e: e.get("created_at") or "", reverse=True)
return entries
def refresh_batch_status(batch_id: str) -> dict:
"""Retrieve live Anthropic status and update the history row."""
client = _get_anthropic_client()
b = client.messages.batches.retrieve(batch_id)
api_status = getattr(b, "processing_status", None) or ""
counts = getattr(b, "request_counts", None)
counts_dict = None
if counts is not None:
counts_dict = {
k: getattr(counts, k, 0) or 0
for k in ("processing", "succeeded", "errored", "canceled", "expired")
}
entry = None
with _batch_file_lock():
history = _read_history_unlocked()
entry = _find_entry(history, batch_id)
prev_status = (entry or {}).get("status")
# Map API status onto local lifecycle without clobbering fetched/consumed.
local_status = None
if prev_status in (STATUS_FETCHED, STATUS_CONSUMED):
local_status = prev_status
elif api_status == "in_progress":
local_status = STATUS_SUBMITTED
elif api_status == "canceling":
local_status = STATUS_CANCELING
elif api_status == "ended":
# Ended after cancel vs normal end - inspect counts when available.
if counts_dict and (counts_dict.get("canceled") or 0) > 0 and (counts_dict.get("succeeded") or 0) == 0:
local_status = STATUS_CANCELED
else:
local_status = STATUS_ENDED
fields: dict[str, Any] = {
"api_status": api_status,
"request_counts": counts_dict,
}
if local_status:
fields["status"] = local_status
return upsert_history_entry(batch_id, **fields)
def cancel_batches(batch_ids: Iterable[str]) -> list[dict]:
"""Cancel in-progress batches. Never submits new work.
Also updates active batch_state so resume does not keep polling canceled ids.
"""
client = _get_anthropic_client()
results = []
canceled_active = []
for bid in batch_ids:
try:
b = client.messages.batches.retrieve(bid)
api_status = getattr(b, "processing_status", "") or ""
if api_status not in CANCELABLE_API:
results.append({"id": bid, "ok": False, "error": f"not cancelable ({api_status})"})
upsert_history_entry(bid, notes=f"cancel skipped: {api_status}", api_status=api_status)
continue
b = client.messages.batches.cancel(bid)
new_status = getattr(b, "processing_status", "canceling") or "canceling"
mark_batches_canceled([bid], api_status=new_status)
canceled_active.append(bid)
results.append({"id": bid, "ok": True, "api_status": new_status})
except Exception as exc:
results.append({"id": bid, "ok": False, "error": str(exc)})
upsert_history_entry(bid, notes=f"cancel failed: {exc}", status=STATUS_ERROR)
if canceled_active:
_remove_active_batch_ids(canceled_active)
return results
def _remove_active_batch_ids(batch_ids: list[str]) -> None:
"""Drop canceled ids from active state so poll/resume does not wait on them."""
id_set = set(batch_ids)
with BATCH_LOCK:
with _batch_file_lock():
state = _read_batch_file(BATCH_STATE_FILE)
batches = [b for b in (state.get("batches") or []) if b.get("id") not in id_set]
if not batches:
# No active submitted batches left.
if state.get("status") == "fetched" or state.get("batch_ids"):
state["batches"] = []
_write_batch_file(BATCH_STATE_FILE, state)
else:
try:
if BATCH_STATE_FILE.exists():
BATCH_STATE_FILE.unlink()
except Exception:
pass
else:
state["batches"] = batches
_write_batch_file(BATCH_STATE_FILE, state)
def _usage_from_message(u) -> dict:
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
# Adaptive thinking may surface under several names depending on SDK version.
thinking = (
getattr(u, "thinking_tokens", None)
or getattr(u, "output_thinking_tokens", None)
or 0
) or 0
return {
"input_tokens": inp,
"output_tokens": out,
"cache_read_input_tokens": cr,
"cache_creation_input_tokens": cw,
"thinking_tokens": thinking,
}
def _price_usage(usage: dict, model: str) -> float:
"""Price real batch usage with cache multipliers and 50% batch discount."""
pricing = getPricingConfig(model)
br = pricing["inputAPICost"] / 1_000_000
orr = pricing["outputAPICost"] / 1_000_000
cr = usage.get("cache_read_input_tokens", 0) or 0
cw = usage.get("cache_creation_input_tokens", 0) or 0
inp = usage.get("input_tokens", 0) or 0
out = usage.get("output_tokens", 0) or 0
thinking = usage.get("thinking_tokens", 0) or 0
# Thinking tokens are billed as output on Anthropic.
raw = cr * br * 0.10 + cw * br * 2.00 + inp * br + (out + thinking) * orr
return raw * 0.50
def _sum_usage_from_results(client, batch_id: str, custom_ids: dict) -> tuple[dict, int, int]:
"""Stream batch results and sum usage. Returns (usage, succeeded, errored)."""
totals = {
"input_tokens": 0,
"output_tokens": 0,
"cache_read_input_tokens": 0,
"cache_creation_input_tokens": 0,
"thinking_tokens": 0,
}
succeeded = 0
errored = 0
for r in client.messages.batches.results(batch_id):
res = r.result
if res.type != "succeeded":
errored += 1
continue
succeeded += 1
part = _usage_from_message(res.message.usage)
for k, v in part.items():
totals[k] = totals.get(k, 0) + (v or 0)
# custom_ids unused for summing but kept for API symmetry / future filters
_ = custom_ids
return totals, succeeded, errored
def usage_for_batch(batch_id: str, model: Optional[str] = None) -> dict:
"""Sum real billed tokens from Anthropic results and persist onto history."""
history = read_history()
entry = _find_entry(history, batch_id)
if entry is None:
raise ValueError(f"Unknown batch id (not in local history): {batch_id}")
client = _get_anthropic_client()
# Ensure batch has ended before streaming results.
b = client.messages.batches.retrieve(batch_id)
api_status = getattr(b, "processing_status", "") or ""
if api_status != "ended":
raise ValueError(f"Batch {batch_id} is not ended (status={api_status})")
custom_ids = dict(entry.get("custom_ids") or {})
usage, succeeded, errored = _sum_usage_from_results(client, batch_id, custom_ids)
use_model = model or entry.get("model") or ""
cost = _price_usage(usage, use_model) if use_model else None
updated = upsert_history_entry(
batch_id,
usage=usage,
actual_cost=cost,
api_status="ended",
notes=f"usage ok={succeeded} err={errored}",
)
return {
"id": batch_id,
"usage": usage,
"actual_cost": cost,
"succeeded": succeeded,
"errored": errored,
"model": use_model,
"entry": updated,
}
def _result_entry_from_message(msg) -> dict:
text = "".join(getattr(b, "text", "") or "" for b in msg.content)
u = msg.usage
part = _usage_from_message(u)
cr = part["cache_read_input_tokens"]
cw = part["cache_creation_input_tokens"]
inp = part["input_tokens"]
out = part["output_tokens"]
entry = {
"text": text,
"prompt_tokens": inp + cr + cw,
"completion_tokens": out,
"cache_read_input_tokens": cr,
"cache_creation_input_tokens": cw,
}
if part.get("thinking_tokens"):
entry["thinking_tokens"] = part["thinking_tokens"]
return entry
def download_batch_results(
batch_id: str,
custom_ids: dict,
*,
client=None,
) -> tuple[dict, list, dict]:
"""Download one ended batch into a cache-key -> result map.
Returns (results, errored_list, usage_totals).
"""
client = client or _get_anthropic_client()
results, errored = {}, []
usage_totals = {
"input_tokens": 0,
"output_tokens": 0,
"cache_read_input_tokens": 0,
"cache_creation_input_tokens": 0,
"thinking_tokens": 0,
}
for r in client.messages.batches.results(batch_id):
key = custom_ids.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', '')}: "
f"{str(getattr(inner, 'message', '') or err)[:200]}"
)
errored.append((r.custom_id, detail))
continue
msg = res.message
results[key] = _result_entry_from_message(msg)
part = _usage_from_message(msg.usage)
for k, v in part.items():
usage_totals[k] = usage_totals.get(k, 0) + (v or 0)
return results, errored, usage_totals
def redownload_batch(batch_id: str) -> dict:
"""Re-fetch results into batch_results.json using stored custom_ids. No re-submit."""
history = read_history()
entry = _find_entry(history, batch_id)
if entry is None:
raise ValueError(f"Unknown batch id (not in local history): {batch_id}")
custom_ids = dict(entry.get("custom_ids") or {})
if not custom_ids:
raise ValueError(f"Batch {batch_id} has no stored custom_ids - cannot redownload")
client = _get_anthropic_client()
b = client.messages.batches.retrieve(batch_id)
api_status = getattr(b, "processing_status", "") or ""
if api_status != "ended":
raise ValueError(f"Batch {batch_id} is not ended (status={api_status})")
results, errored, usage = download_batch_results(batch_id, custom_ids, client=client)
model = entry.get("model") or ""
cost = _price_usage(usage, model) if model else None
import util.translation as T
with T.BATCH_LOCK:
with _batch_file_lock():
merged = _read_batch_file(BATCH_RESULTS_FILE)
merged.update(results)
_write_batch_file(BATCH_RESULTS_FILE, merged)
# Activate fetched state without re-submit; keep custom_ids in history.
_write_batch_file(
BATCH_STATE_FILE,
{"status": "fetched", "batch_ids": [batch_id], "batches": []},
)
# Queue is no longer needed for consume.
try:
if BATCH_QUEUE_FILE.exists():
BATCH_QUEUE_FILE.unlink()
except Exception:
pass
T._batch_results = None
record_fetch([batch_id], succeeded=len(results), errored=len(errored), usage=usage, actual_cost=cost)
return {
"id": batch_id,
"succeeded": len(results),
"errored": len(errored),
"errors": errored[:20],
"usage": usage,
"actual_cost": cost,
}
def activate_for_resume(batch_id: str) -> str:
"""Ensure active files match a history entry for Translation-tab resume.
Returns the batchRunState string to pass as batch_resume_state:
'submitted' | 'fetched'.
"""
history = read_history()
entry = _find_entry(history, batch_id)
if entry is None:
raise ValueError(f"Unknown batch id: {batch_id}")
status = entry.get("status")
custom_ids = dict(entry.get("custom_ids") or {})
if status in (STATUS_FETCHED, STATUS_ENDED, STATUS_CONSUMED):
# Prefer local results; redownload if missing.
with _batch_file_lock():
results = _read_batch_file(BATCH_RESULTS_FILE)
if not results:
if status == STATUS_CONSUMED:
raise ValueError(
f"Batch {batch_id} was already consumed and local results are gone. "
"Use Redownload first."
)
redownload_batch(batch_id)
else:
with BATCH_LOCK:
with _batch_file_lock():
_write_batch_file(
BATCH_STATE_FILE,
{"status": "fetched", "batch_ids": [batch_id], "batches": []},
)
return "fetched"
if status in (STATUS_SUBMITTED, STATUS_CANCELING, STATUS_ENDED):
# Restore active state so poll/fetch can continue.
with BATCH_LOCK:
with _batch_file_lock():
state = _read_batch_file(BATCH_STATE_FILE)
batches = list(state.get("batches") or [])
existing = {b.get("id") for b in batches}
if batch_id not in existing:
batches.append({"id": batch_id, "custom_ids": custom_ids})
state = {
"batches": batches,
"submitted_at": state.get("submitted_at") or entry.get("created_at"),
"model": entry.get("model") or state.get("model") or "",
"file_set": entry.get("file_set") or state.get("file_set") or [],
"cost_estimate": entry.get("cost_estimate") or state.get("cost_estimate"),
}
_write_batch_file(BATCH_STATE_FILE, state)
if status == STATUS_ENDED:
# Already ended on Anthropic - fetch path via resume submitted → poll sees ended.
return "submitted"
return "submitted"
if status == STATUS_CANCELED:
raise ValueError(f"Batch {batch_id} was canceled - nothing to resume")
raise ValueError(f"Batch {batch_id} cannot be resumed from status={status}")
def missing_result_count() -> tuple[int, int]:
"""Return (present, expected) result counts for the active fetched run.
expected is summed from history request_count for active fetched batch ids.
"""
with _batch_file_lock():
results = _read_batch_file(BATCH_RESULTS_FILE)
state = _read_batch_file(BATCH_STATE_FILE)
history = _read_history_unlocked()
present = len(results)
ids = list(state.get("batch_ids") or [])
if not ids:
ids = [b.get("id") for b in (state.get("batches") or []) if b.get("id")]
expected = 0
for bid in ids:
entry = _find_entry(history, bid)
if entry:
expected += int(entry.get("request_count") or len(entry.get("custom_ids") or {}))
if not expected:
expected = present
return present, expected
def active_fetched_batch_ids() -> list[str]:
"""Batch ids associated with the current fetched active run."""
with _batch_file_lock():
state = _read_batch_file(BATCH_STATE_FILE)
ids = list(state.get("batch_ids") or [])
if ids:
return ids
return [b.get("id") for b in (state.get("batches") or []) if b.get("id")]
def on_clear_active_files(*, had_results: bool, fetched_ids: Optional[list] = None) -> None:
"""Called from clearBatchFiles - mark fetched ids consumed, never wipe history."""
if not had_results and not fetched_ids:
return
ids = list(fetched_ids or [])
if not ids and had_results:
# Fall back: mark any currently-fetched history rows as consumed.
history = read_history()
ids = [e["id"] for e in history.get("batches", []) if e.get("status") == STATUS_FETCHED]
if ids:
mark_batches_consumed(ids)

View file

@ -756,9 +756,12 @@ def _read_batch_file(path):
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
if isinstance(data, dict):
return data
print(f"[BATCH] Corrupt batch file (not a JSON object): {path}", flush=True)
except Exception as exc:
# State/history corruption must not look like an empty run - log it.
print(f"[BATCH] Failed to read {path}: {exc}", flush=True)
return {}
@ -840,30 +843,54 @@ def pendingBatchRequests():
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."""
"""Resume detector for an interrupted batch run.
Returns:
'submitted' - Anthropic batch(es) in flight (or ended but not fetched)
'fetched' - results on disk waiting for consume
'queued' - collect finished / submit declined; queue still on disk
None - nothing to resume
"""
with _batch_file_lock():
if _read_batch_file(BATCH_STATE_FILE).get("batches"):
state = _read_batch_file(BATCH_STATE_FILE)
if state.get("batches"):
return "submitted"
if _read_batch_file(BATCH_RESULTS_FILE):
if state.get("status") == "fetched" or _read_batch_file(BATCH_RESULTS_FILE):
return "fetched"
if _read_batch_file(BATCH_QUEUE_FILE):
return "queued"
return None
def clearBatchFiles():
"""Remove queue/state/results left over from any previous batch run."""
"""Remove active queue/state/results. Never deletes durable batch history."""
global _batch_results, _batch_queue_pending
with BATCH_LOCK:
_batch_results = None
_batch_queue_pending = {}
fetched_ids = []
had_results = False
with _batch_file_lock():
state = _read_batch_file(BATCH_STATE_FILE)
had_results = bool(_read_batch_file(BATCH_RESULTS_FILE)) or state.get("status") == "fetched"
if had_results:
fetched_ids = list(state.get("batch_ids") or [])
if not fetched_ids:
fetched_ids = [b.get("id") for b in (state.get("batches") or []) if b.get("id")]
for path in (BATCH_QUEUE_FILE, BATCH_STATE_FILE, BATCH_RESULTS_FILE):
try:
if path.exists():
path.unlink()
except Exception:
pass
_batch_results = None
_batch_queue_pending = {}
# Only mark history consumed when clearing after a successful fetch/consume,
# never when discarding a still-submitted or queued run.
if had_results:
try:
from util.batch_history import on_clear_active_files
on_clear_active_files(had_results=True, fetched_ids=fetched_ids or None)
except Exception as exc:
print(f"[BATCH] history consume mark failed: {exc}", flush=True)
def _get_anthropic_client():
@ -968,11 +995,13 @@ def estimateBatchCost(model=None):
}
def submitTranslationBatches():
def submitTranslationBatches(file_set=None, cost_estimate=None):
"""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."""
fetchTranslationBatches can route results back. Also appends durable
history entries (custom_ids survive later fetch/clear). Returns the batch ids.
"""
flush_batch_queue()
with _batch_file_lock():
queue = _read_batch_file(BATCH_QUEUE_FILE)
@ -983,6 +1012,16 @@ def submitTranslationBatches():
client = _get_anthropic_client()
batches = []
requests, id_map, size = [], {}, 0
models = set()
for entry in queue.values():
m = (entry.get("params") or {}).get("model")
if m:
models.add(m)
model = (
(cost_estimate or {}).get("model")
or next(iter(models), None)
or os.getenv("model", "")
)
def _submit():
nonlocal requests, id_map, size
@ -1003,48 +1042,134 @@ def submitTranslationBatches():
_submit()
_submit()
submitted_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
state_doc = {
"batches": batches,
"submitted_at": submitted_at,
"model": model,
"file_set": list(file_set or []),
"cost_estimate": cost_estimate,
"request_count": sum(len(b.get("custom_ids") or {}) for b in batches),
}
with BATCH_LOCK:
with _batch_file_lock():
_write_batch_file(BATCH_STATE_FILE, {"batches": batches})
_write_batch_file(BATCH_STATE_FILE, state_doc)
try:
from util.batch_history import record_submit
record_submit(
batches,
model=model,
file_set=file_set,
cost_estimate=cost_estimate,
)
except Exception as exc:
print(f"[BATCH] history record_submit failed: {exc}", flush=True)
return [b["id"] for b in batches]
def checkTranslationBatches():
"""Print the processing status of submitted batches. True when all ended."""
"""Print the processing status of submitted batches. True when all ended.
Also returns a structured status list as the second value when called as
``ended, statuses = checkTranslationBatchStatuses()`` - prefer that helper
for UI work. Kept for CLI/print compatibility.
"""
ended, _statuses = checkTranslationBatchStatuses(print_status=True)
return ended
def checkTranslationBatchStatuses(print_status=True):
"""Return (all_ended, statuses) for submitted batches.
Each status dict: id, api_status, counts{processing,succeeded,errored,canceled,expired}.
"""
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
if print_status:
print("[BATCH] No submitted batches - submit the queue first.", flush=True)
return False, []
client = _get_anthropic_client()
all_ended = True
statuses = []
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":
bid = info["id"]
b = client.messages.batches.retrieve(bid)
api_status = getattr(b, "processing_status", "") or ""
counts_obj = getattr(b, "request_counts", None)
counts = {
k: int(getattr(counts_obj, k, 0) or 0)
for k in ("processing", "succeeded", "errored", "canceled", "expired")
} if counts_obj is not None else {
"processing": 0, "succeeded": 0, "errored": 0, "canceled": 0, "expired": 0,
}
statuses.append({
"id": bid,
"api_status": api_status,
"counts": counts,
"request_count": len(info.get("custom_ids") or {}),
})
if print_status:
parts = [f"{k[:4]}={v}" for k, v in counts.items() if v]
suffix = (" " + " ".join(parts)) if parts else ""
print(
f"[BATCH] {time.strftime('%H:%M:%S')} {bid}: {api_status}{suffix}",
flush=True,
)
if api_status != "ended":
all_ended = False
return all_ended
return all_ended, statuses
def fetchTranslationBatches():
def fetchTranslationBatches(batches=None):
"""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."""
during consume. Durable history retains custom_ids for later redownload.
batches: optional list of {id, custom_ids} (defaults to active batch_state).
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)
batch_list = batches if batches is not None else (state.get("batches") or [])
if not batch_list:
print("[BATCH] No submitted batches - nothing to fetch.", flush=True)
return 0, 0
try:
from util.batch_history import download_batch_results, record_fetch, _price_usage
except Exception:
download_batch_results = None
record_fetch = None
_price_usage = None
client = _get_anthropic_client()
results, errored = {}, []
for info in state["batches"]:
usage_totals = {
"input_tokens": 0,
"output_tokens": 0,
"cache_read_input_tokens": 0,
"cache_creation_input_tokens": 0,
"thinking_tokens": 0,
}
batch_ids = []
for info in batch_list:
bid = info.get("id")
if bid:
batch_ids.append(bid)
id_map = info.get("custom_ids", {})
for r in client.messages.batches.results(info["id"]):
if download_batch_results is not None:
part, err_part, usage_part = download_batch_results(bid, id_map, client=client)
results.update(part)
errored.extend(err_part)
for k, v in usage_part.items():
usage_totals[k] = usage_totals.get(k, 0) + (v or 0)
continue
# Fallback if batch_history import failed - preserve prior behaviour.
for r in client.messages.batches.results(bid):
key = id_map.get(r.custom_id)
if key is None:
continue
@ -1066,26 +1191,61 @@ def fetchTranslationBatches():
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,
}
model = state.get("model") or os.getenv("model", "")
actual_cost = None
if _price_usage is not None and model:
try:
actual_cost = _price_usage(usage_totals, model)
except Exception:
actual_cost = None
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
# Drop the queue; keep a lightweight fetched marker (ids for consume→history).
# custom_ids stay in durable history - do not destroy recovery maps.
try:
if BATCH_QUEUE_FILE.exists():
BATCH_QUEUE_FILE.unlink()
except Exception:
pass
_write_batch_file(
BATCH_STATE_FILE,
{
"status": "fetched",
"batch_ids": batch_ids,
"batches": [],
"model": model,
"file_set": state.get("file_set") or [],
"cost_estimate": state.get("cost_estimate"),
"fetched_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
},
)
_batch_results = None
if record_fetch is not None:
try:
record_fetch(
batch_ids,
succeeded=len(results),
errored=len(errored),
usage=usage_totals,
actual_cost=actual_cost,
)
except Exception as exc:
print(f"[BATCH] history record_fetch failed: {exc}", flush=True)
print(f"[BATCH] fetched {len(results)} results ({len(errored)} errored).", flush=True)
if actual_cost is not None:
print(f"[BATCH] actual batch usage cost (est.): ${actual_cost:.4f}", flush=True)
for cid, why in errored[:20]:
print(f"[BATCH] ! {cid}: {why}", flush=True)
if len(errored) > 20:
@ -2626,6 +2786,29 @@ def translateAI(text, history, config, filename=None, pbar=None, lock=None, mism
with lock:
pbar.update(len(tItem) if isinstance(tItem, list) else 1)
# Consume pass: still record what was applied. Cache hits used to
# skip the log entirely, which left the GUI Translation Log empty
# after a resume even though translated/ was written.
if batch_phase == "consume" and not config.estimateMode:
try:
if isinstance(cached_result, list):
out_payload = {
f"Line{i+1}": string for i, string in enumerate(cached_result)
}
formatted_output = json.dumps(out_payload, indent=4, ensure_ascii=False)
else:
formatted_output = json.dumps(
{"Line1": cached_result}, indent=4, ensure_ascii=False
)
Path(config.logFilePath).parent.mkdir(parents=True, exist_ok=True)
with open(config.logFilePath, "a", encoding="utf-8") as logFile:
logFile.write("[CACHE] Applied cached translation (no new API call)\n")
logFile.write(f"Input:\n{subbedT}\n")
logFile.write(f"Output:\n{formatted_output}\n")
logFile.flush()
except Exception:
pass
continue
# Create context — static_system is the stable prompt.txt content;
@ -2701,6 +2884,7 @@ def translateAI(text, history, config, filename=None, pbar=None, lock=None, mism
max_retries = 2 # 1 initial attempt + 2 retries
final_translations = None
last_raw_translation = ""
from_batch = False
numLines = len(clean_tItem) if isinstance(tItem, list) else 1
for attempt in range(max_retries + 1):
@ -2726,7 +2910,7 @@ 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 the consume pass tries the fetched batch result first;
# 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:
@ -2944,6 +3128,8 @@ def translateAI(text, history, config, filename=None, pbar=None, lock=None, mism
# Only open and write to log file when we have something to log
try:
with open(config.logFilePath, "a", encoding="utf-8") as logFile:
if from_batch:
logFile.write("[BATCH] Applied Anthropic batch result\n")
logFile.write(f"Input:\n{subbedT}\n")
logFile.write(f"Output:\n{formatted_output}\n")
logFile.flush() # Ensure data is written to disk immediately