diff --git a/TODO.md b/TODO.md index 85aad13..b54bf83 100644 --- a/TODO.md +++ b/TODO.md @@ -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 [ ...] - 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 — 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. diff --git a/gui/batch_tab.py b/gui/batch_tab.py new file mode 100644 index 0000000..44d0681 --- /dev/null +++ b/gui/batch_tab.py @@ -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) diff --git a/gui/main.py b/gui/main.py index 3989cb9..3cceaeb 100644 --- a/gui/main.py +++ b/gui/main.py @@ -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) diff --git a/gui/translation_tab.py b/gui/translation_tab.py index 5f249b1..d38f614 100644 --- a/gui/translation_tab.py +++ b/gui/translation_tab.py @@ -24,12 +24,13 @@ from tqdm import tqdm from dotenv import load_dotenv from PyQt5.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, QGroupBox, - QTextEdit, QMessageBox, QListWidget, QListWidgetItem, - QSplitter, QFileDialog, QComboBox, QCheckBox, QProgressBar, QFrame, QFormLayout, QStackedWidget + QTextEdit, QMessageBox, QListWidget, QListWidgetItem, + QSplitter, QFileDialog, QComboBox, QCheckBox, QProgressBar, QFrame, QFormLayout, QStackedWidget, + QTableWidget, QTableWidgetItem, QHeaderView, QAbstractItemView, ) from PyQt5.QtWidgets import QSizePolicy from PyQt5.QtCore import Qt, QTimer, pyqtSignal, QThread, QMutex, QProcess, QEvent, QRect, QSettings, QSize -from PyQt5.QtGui import QFont +from PyQt5.QtGui import QFont, QColor, QBrush from gui.log_viewer import LogViewer from gui import qt_icons @@ -96,6 +97,7 @@ class TranslationWorker(QThread): self.batch_resume_state = batch_resume_state self._batch_submit_event = threading.Event() self._batch_submit_approved = False + self._batch_pending_estimate = None self.should_stop = False self.mutex = QMutex() # For thread safety self.executor = None # Store reference to executor for proper shutdown @@ -107,6 +109,7 @@ class TranslationWorker(QThread): self._batch_submit_event.set() def _wait_batch_submit(self, estimate): + self._batch_pending_estimate = estimate self._batch_submit_event.clear() self.batch_phase_signal.emit("submit", estimate) self._batch_submit_event.wait() @@ -129,7 +132,6 @@ class TranslationWorker(QThread): """Submit (if needed), poll until ended, fetch results. None if stopped while polling.""" from util.translation import ( submitTranslationBatches, - checkTranslationBatches, fetchTranslationBatches, _read_batch_file, BATCH_STATE_FILE, @@ -139,30 +141,38 @@ class TranslationWorker(QThread): with _batch_file_lock(): state = _read_batch_file(BATCH_STATE_FILE) if not state.get("batches"): - if not self._emit_batch_output(submitTranslationBatches): + est = self._batch_pending_estimate + file_set = list(self.selected_files or []) + if not self._emit_batch_output( + submitTranslationBatches, + file_set=file_set, + cost_estimate=est, + ): return 0, 0 poll = int(os.getenv("batchPollInterval", "60") or 60) self._emit_batch_phase("polling") self.emit_log( - f"[BATCH] polling every {poll}s (stop is safe — resume later with Batch Translate mode)..." + f"[BATCH] polling every {poll}s (stop is safe - resume later with Batch Translate mode)..." ) + from util.translation import checkTranslationBatchStatuses while True: if self.should_stop: - self.emit_log("[BATCH] Stopped while polling. Batch keeps processing — resume later.") + self.emit_log("[BATCH] Stopped while polling. Batch keeps processing - resume later.") return None buf = io.StringIO() with redirect_stdout(buf): - ended = checkTranslationBatches() + ended, statuses = checkTranslationBatchStatuses(print_status=True) for line in buf.getvalue().splitlines(): if line.strip(): self.emit_log(line) - self._emit_batch_phase("poll_status", line.strip()) + if statuses: + self._emit_batch_phase("poll_status", statuses) if ended: break for _ in range(poll * 10): if self.should_stop: - self.emit_log("[BATCH] Stopped while polling. Batch keeps processing — resume later.") + self.emit_log("[BATCH] Stopped while polling. Batch keeps processing - resume later.") return None time.sleep(0.1) @@ -212,8 +222,26 @@ class TranslationWorker(QThread): self.mutex.unlock() def emit_log(self, message): - """Thread-safe log emission.""" + """Thread-safe log emission. + + Also mirrors into TRANSLATION_RUN_LOG so the LogViewer file-tail shows + batch/status lines (those never go through translateAI's Input/Output writer). + """ self.log_signal.emit(message) + try: + run_log = os.getenv("TRANSLATION_RUN_LOG") + if not run_log or not message: + return + text = _strip_ansi(str(message)).rstrip() + if not text: + return + path = Path(run_log) + path.parent.mkdir(parents=True, exist_ok=True) + with open(path, "a", encoding="utf-8") as f: + f.write(text + "\n") + f.flush() + except Exception: + pass def emit_progress(self, current, total, filename): """Thread-safe progress emission.""" @@ -452,9 +480,8 @@ class TranslationWorker(QThread): break filename = future_to_filename[future] - completed_count += 1 - self.emit_progress(completed_count, total_files, filename) - + # Resolve the future before emitting progress so cost lines from the + # subprocess are queued ahead of the file-complete progress event. try: result = future.result() if isinstance(result, tuple) and len(result) == 2 and result[0] == "SUBPROCESS_ERROR": @@ -462,6 +489,8 @@ class TranslationWorker(QThread): elif result and result not in ("Fail", "Stopped"): total_cost = result elif result == "Stopped": + completed_count += 1 + self.emit_progress(completed_count, total_files, filename) break else: self.emit_log(f"❌ Failed processing {filename}") @@ -471,6 +500,9 @@ class TranslationWorker(QThread): self.emit_log(f"❌ Error processing {filename}: {str(e)} | Line: {tb_line}") self.file_error_signal.emit(filename, str(e)) + completed_count += 1 + self.emit_progress(completed_count, total_files, filename) + if self.executor: try: self.executor.shutdown(wait=False, cancel_futures=True) @@ -587,7 +619,39 @@ class TranslationWorker(QThread): est = dict(est) est["files"] = len(matching_files) if not self._wait_batch_submit(est): - self.emit_log("[BATCH] Not submitted. Queue kept in log/batch_requests.json.") + self.emit_log( + "[BATCH] Not submitted. Queue kept in log/batch_requests.json " + "(resume with Batch Translate to submit without re-collecting)." + ) + self.finished_signal.emit(True, "Batch not submitted") + return + poll_result = self._run_batch_poll_fetch() + if poll_result is None: + self.finished_signal.emit(False, "Batch polling stopped") + return + elif self.batch_resume_state == "queued": + # Resume a declined/interrupted collect: estimate + submit only. + self.emit_log( + "[BATCH] Resuming queued requests (skipping re-collect to avoid " + "duplicate live charges and a second batch submission)..." + ) + n_requests = pendingBatchRequests() + if n_requests == 0: + self.emit_log("[BATCH] Queue is empty - nothing to submit.") + run_consume = False + else: + self._emit_batch_phase("collect_done", { + "files": len(matching_files), + "requests": n_requests, + }) + est = self._emit_batch_output(estimateBatchCost) + if est is not None: + est = dict(est) + est["files"] = len(matching_files) + if not self._wait_batch_submit(est): + self.emit_log( + "[BATCH] Not submitted. Queue kept in log/batch_requests.json." + ) self.finished_signal.emit(True, "Batch not submitted") return poll_result = self._run_batch_poll_fetch() @@ -605,9 +669,21 @@ class TranslationWorker(QThread): self.emit_log("[BATCH] Resuming from fetched results...") if run_consume and not self.should_stop: + try: + from util.batch_history import missing_result_count + present, expected = missing_result_count() + if expected and present < expected: + self.emit_log( + f"[BATCH] WARNING: only {present}/{expected} results present. " + "Missing keys will fall back to the live API (full price)." + ) + except Exception: + pass self._emit_batch_phase("consume") self.emit_log("[BATCH] Pass 2/2: writing translated files...") total_cost = self._run_files(matching_files, False, batch_phase="consume") + if not self.should_stop: + self._emit_batch_phase("done") else: total_cost = self._run_files(matching_files, self.estimate_only) finally: @@ -723,6 +799,7 @@ class TranslationTab(QWidget): self.totals_widget = None self._batch_active = False self._batch_ui_phase = None + self._batch_consume_started = False self._batch_tab_index = -1 self.setup_ui() @@ -904,11 +981,29 @@ class TranslationTab(QWidget): self.batch_phase_title.setStyleSheet("color:#4ec9b0;font-weight:bold;font-size:13px;") batch_pipe_layout.addWidget(self.batch_phase_title) + # Step strip: Collect → Submit → Process → Write + self.batch_steps_row = QHBoxLayout() + self.batch_steps_row.setSpacing(6) + self._batch_step_labels = [] + for i, name in enumerate(("1. Collect", "2. Submit", "3. Process", "4. Write")): + lab = QLabel(name) + lab.setAlignment(Qt.AlignCenter) + lab.setStyleSheet(self._batch_step_style("idle")) + self.batch_steps_row.addWidget(lab, 1) + self._batch_step_labels.append(lab) + if i < 3: + arrow = QLabel("→") + arrow.setStyleSheet("color:#666666;font-size:12px;") + arrow.setAlignment(Qt.AlignCenter) + self.batch_steps_row.addWidget(arrow, 0) + batch_pipe_layout.addLayout(self.batch_steps_row) + self.batch_overall_bar = QProgressBar() self.batch_overall_bar.setRange(0, 100) self.batch_overall_bar.setValue(0) self.batch_overall_bar.setFixedHeight(20) self.batch_overall_bar.setTextVisible(True) + self.batch_overall_bar.setFormat("%p%") self.batch_overall_bar.setStyleSheet(""" QProgressBar { border: none; @@ -943,12 +1038,26 @@ class TranslationTab(QWidget): submit_page = QWidget() submit_layout = QVBoxLayout(submit_page) submit_layout.setContentsMargins(0, 0, 0, 0) - submit_layout.setSpacing(4) + submit_layout.setSpacing(8) self.batch_submit_summary = QLabel("") self.batch_submit_summary.setWordWrap(True) self.batch_submit_summary.setAlignment(Qt.AlignTop) self.batch_submit_summary.setStyleSheet("color:#cccccc;font-size:12px;") - submit_layout.addWidget(self.batch_submit_summary, 1) + submit_layout.addWidget(self.batch_submit_summary) + # Cost chips row + self.batch_cost_row = QHBoxLayout() + self.batch_cost_row.setSpacing(8) + self.batch_cost_cached = QLabel("Batch+cache: —") + self.batch_cost_nocache = QLabel("Batch worst: —") + self.batch_cost_live = QLabel("Live: —") + for lab in (self.batch_cost_cached, self.batch_cost_nocache, self.batch_cost_live): + lab.setStyleSheet( + "color:#cccccc;background:#1e1e1e;border:1px solid #3e3e42;" + "border-radius:4px;padding:8px;font-size:12px;" + ) + lab.setAlignment(Qt.AlignCenter) + self.batch_cost_row.addWidget(lab, 1) + submit_layout.addLayout(self.batch_cost_row) submit_btn_row = QHBoxLayout() submit_btn_row.addStretch() self.batch_submit_yes_btn = QPushButton("Submit Batch") @@ -966,32 +1075,72 @@ class TranslationTab(QWidget): submit_btn_row.addWidget(self.batch_submit_no_btn) submit_btn_row.addWidget(self.batch_submit_yes_btn) submit_layout.addLayout(submit_btn_row) + submit_layout.addStretch(1) self.batch_pipeline_stack.addWidget(submit_page) poll_page = QWidget() poll_layout = QVBoxLayout(poll_page) poll_layout.setContentsMargins(0, 0, 0, 0) + poll_layout.setSpacing(8) + self.batch_poll_id = QLabel("Batch: —") + self.batch_poll_id.setStyleSheet("color:#9cdcfe;font-size:12px;font-family:monospace;") + self.batch_poll_id.setTextInteractionFlags(Qt.TextSelectableByMouse) + poll_layout.addWidget(self.batch_poll_id) self.batch_poll_status = QLabel("Waiting for Anthropic to finish processing the batch…") self.batch_poll_status.setWordWrap(True) self.batch_poll_status.setStyleSheet("color:#cccccc;font-size:12px;") poll_layout.addWidget(self.batch_poll_status) + # Request count chips + self.batch_count_row = QHBoxLayout() + self.batch_count_row.setSpacing(6) + self._batch_count_labels = {} + for key, color in ( + ("succeeded", "#4ec9b0"), + ("processing", "#007acc"), + ("errored", "#f44747"), + ("canceled", "#ce9178"), + ("expired", "#dcdcaa"), + ): + lab = QLabel(f"{key}: 0") + lab.setAlignment(Qt.AlignCenter) + lab.setStyleSheet( + f"color:{color};background:#1e1e1e;border:1px solid #3e3e42;" + f"border-radius:4px;padding:6px 4px;font-size:11px;font-weight:bold;" + ) + self.batch_count_row.addWidget(lab, 1) + self._batch_count_labels[key] = lab + poll_layout.addLayout(self.batch_count_row) self.batch_poll_bar = QProgressBar() - self.batch_poll_bar.setRange(0, 0) # indeterminate + self.batch_poll_bar.setRange(0, 100) + self.batch_poll_bar.setValue(0) self.batch_poll_bar.setFixedHeight(16) + self.batch_poll_bar.setTextVisible(True) + self.batch_poll_bar.setFormat("Requests complete: %v / %m") self.batch_poll_bar.setStyleSheet(""" - QProgressBar { border: none; background-color: #2b2b2b; border-radius: 3px; } - QProgressBar::chunk { background-color: #007acc; border-radius: 3px; } + QProgressBar { border: none; background-color: #2b2b2b; border-radius: 3px; color:#ccc; text-align:center; } + QProgressBar::chunk { background-color: #4ec9b0; border-radius: 3px; } """) poll_layout.addWidget(self.batch_poll_bar) + self.batch_poll_hint = QLabel("Stop is safe - the batch keeps running on Anthropic. Resume later from Batches or Batch Translate.") + self.batch_poll_hint.setWordWrap(True) + self.batch_poll_hint.setStyleSheet("color:#888888;font-size:11px;") + poll_layout.addWidget(self.batch_poll_hint) + poll_layout.addStretch(1) self.batch_pipeline_stack.addWidget(poll_page) consume_page = QWidget() consume_layout = QVBoxLayout(consume_page) consume_layout.setContentsMargins(0, 0, 0, 0) + consume_layout.setSpacing(8) self.batch_consume_status = QLabel("Pass 2/2: writing translated files from batch results…") self.batch_consume_status.setWordWrap(True) self.batch_consume_status.setStyleSheet("color:#cccccc;font-size:12px;") consume_layout.addWidget(self.batch_consume_status) + self.batch_consume_hint = QLabel("Watch the Translation Log for [BATCH]/[CACHE] Input/Output pairs as each chunk is applied.") + self.batch_consume_hint.setWordWrap(True) + self.batch_consume_hint.setStyleSheet("color:#888888;font-size:11px;") + consume_layout.addWidget(self.batch_consume_hint) + consume_layout.addStretch(1) self.batch_pipeline_stack.addWidget(consume_page) batch_pipe_layout.addWidget(self.batch_pipeline_stack, 1) @@ -1035,6 +1184,34 @@ class TranslationTab(QWidget): border-bottom: none; } """) + # Prefer a Batch-history-style table for per-file status; keep the list + # widget as a non-visible compatibility shim for any leftover references. + self.progress_list.setVisible(False) + + self.progress_files_summary = QLabel("No files in this run.") + self.progress_files_summary.setStyleSheet("color:#9d9d9d;font-size:12px;padding:2px 0;") + + self.progress_table = QTableWidget(0, 6) + self.progress_table.setHorizontalHeaderLabels( + ["File", "Status", "Progress", "Tokens", "Cost", "Time"] + ) + self.progress_table.setSelectionBehavior(QAbstractItemView.SelectRows) + self.progress_table.setSelectionMode(QAbstractItemView.SingleSelection) + self.progress_table.setEditTriggers(QAbstractItemView.NoEditTriggers) + self.progress_table.setAlternatingRowColors(True) + self.progress_table.verticalHeader().setVisible(False) + self.progress_table.setShowGrid(True) + self.progress_table.setStyleSheet( + "QTableWidget{background-color:#1e1e1e;color:#cccccc;gridline-color:#3a3a3a;" + "alternate-background-color:#252526;border:1px solid #555555;}" + "QHeaderView::section{background-color:#2d2d30;color:#cccccc;padding:4px;" + "border:1px solid #3a3a3a;font-weight:bold;}" + ) + hdr = self.progress_table.horizontalHeader() + hdr.setSectionResizeMode(0, QHeaderView.Stretch) + for col in range(1, 6): + hdr.setSectionResizeMode(col, QHeaderView.ResizeToContents) + self.progress_table.verticalHeader().setDefaultSectionSize(28) self.progress_overview_page = QWidget() overview_layout = QVBoxLayout(self.progress_overview_page) @@ -1045,7 +1222,9 @@ class TranslationTab(QWidget): self.progress_files_page = QWidget() files_page_layout = QVBoxLayout(self.progress_files_page) files_page_layout.setContentsMargins(0, 0, 0, 0) - files_page_layout.addWidget(self.progress_list) + files_page_layout.setSpacing(8) + files_page_layout.addWidget(self.progress_files_summary) + files_page_layout.addWidget(self.progress_table, 1) progress_tab_btn_style = """ QPushButton { @@ -1624,13 +1803,85 @@ class TranslationTab(QWidget): except Exception: pass + def _batch_step_style(self, state: str) -> str: + """CSS for a pipeline step chip: idle | active | done.""" + if state == "active": + return ( + "color:#ffffff;background:#007acc;border:1px solid #007acc;" + "border-radius:4px;padding:4px 6px;font-size:11px;font-weight:bold;" + ) + if state == "done": + return ( + "color:#4ec9b0;background:#1e1e1e;border:1px solid #4ec9b0;" + "border-radius:4px;padding:4px 6px;font-size:11px;" + ) + return ( + "color:#888888;background:#1e1e1e;border:1px solid #3e3e42;" + "border-radius:4px;padding:4px 6px;font-size:11px;" + ) + + def _set_batch_steps(self, active_index: int): + """Highlight pipeline steps. active_index 0..3, or 4 when fully done.""" + labels = getattr(self, "_batch_step_labels", None) or [] + for i, lab in enumerate(labels): + if i < active_index: + lab.setStyleSheet(self._batch_step_style("done")) + elif i == active_index and active_index < 4: + lab.setStyleSheet(self._batch_step_style("active")) + elif active_index >= 4: + lab.setStyleSheet(self._batch_step_style("done")) + else: + lab.setStyleSheet(self._batch_step_style("idle")) + + def _update_batch_poll_dashboard(self, statuses): + """Render structured Anthropic batch statuses into the poll panel.""" + if not isinstance(statuses, list) or not statuses: + return + totals = {"processing": 0, "succeeded": 0, "errored": 0, "canceled": 0, "expired": 0} + ids = [] + api_states = [] + expected = 0 + for st in statuses: + ids.append(st.get("id") or "") + api_states.append(st.get("api_status") or "") + expected += int(st.get("request_count") or 0) + for k, v in (st.get("counts") or {}).items(): + if k in totals: + totals[k] += int(v or 0) + done = totals["succeeded"] + totals["errored"] + totals["canceled"] + totals["expired"] + total = max(expected, done + totals["processing"], 1) + if hasattr(self, "batch_poll_id"): + self.batch_poll_id.setText("Batch: " + (", ".join(i for i in ids if i) or "—")) + for key, lab in (getattr(self, "_batch_count_labels", {}) or {}).items(): + lab.setText(f"{key}: {totals.get(key, 0)}") + if hasattr(self, "batch_poll_bar"): + self.batch_poll_bar.setRange(0, total) + self.batch_poll_bar.setValue(min(done, total)) + self.batch_poll_bar.setFormat(f"Requests finished: {done} / {total}") + # Overall bar: process phase spans ~55-80% + frac = done / total if total else 0 + self.batch_overall_bar.setValue(55 + int(20 * frac)) + state_txt = ", ".join(sorted(set(api_states))) or "unknown" + self.batch_poll_status.setText( + f"Anthropic status: {state_txt}\n" + f"{totals['succeeded']} succeeded, {totals['processing']} still processing" + + (f", {totals['errored']} errored" if totals["errored"] else "") + + (f", {totals['canceled']} canceled" if totals["canceled"] else "") + + (f", {totals['expired']} expired" if totals["expired"] else "") + + "." + ) + self.batch_live_status.setText( + f"Polling… {done}/{total} requests finished. Details stay in the Translation Log." + ) + def _on_batch_phase(self, phase, payload): """Update the inline batch pipeline panel for the current phase.""" self._batch_ui_phase = phase self.batch_pipeline_widget.setVisible(True) if phase == "collect": - self.batch_phase_title.setText("Batch Translate — Pass 1/2: Collect") + self._set_batch_steps(0) + self.batch_phase_title.setText("Batch Translate - Pass 1/2: Collect") self.batch_overall_bar.setRange(0, 100) self.batch_overall_bar.setValue(15) self.batch_pipeline_stack.setCurrentIndex(0) @@ -1642,59 +1893,83 @@ class TranslationTab(QWidget): info = payload or {} n_files = info.get("files", "?") n_req = info.get("requests", "?") - self.batch_phase_title.setText("Batch Translate — Collect complete") + self._set_batch_steps(1) + self.batch_phase_title.setText("Batch Translate - Collect complete") self.batch_overall_bar.setValue(30) self.batch_pipeline_stack.setCurrentIndex(0) self.batch_collect_status.setText( - f"Finished scanning {n_files} file(s) — {n_req} API request(s) queued.\n" - "Review the total cost below, then submit once for the whole batch." + f"Finished scanning {n_files} file(s) - {n_req} API request(s) queued.\n" + "Review the cost estimate, then submit once for the whole batch." ) self.batch_live_status.setText( - f"All {n_files} files collected. Switch to the Files tab to inspect individual rows." + f"All {n_files} files collected. Open the Files tab to inspect per-file rows." ) elif phase == "submit": est = payload or {} n_files = est.get("files", "?") n_req = est.get("requests", "?") - self.batch_phase_title.setText("Batch Translate — Review & Submit (once)") + self._set_batch_steps(1) + self.batch_phase_title.setText("Batch Translate - Review & Submit") self.batch_overall_bar.setValue(35) self.batch_pipeline_stack.setCurrentIndex(1) self.batch_submit_summary.setText( - f"All {n_files} file(s) scanned — {n_req} request(s) queued total.\n\n" - f"Submit everything in one Anthropic batch?\n\n" - f"Estimated cost: ${est.get('batch_cached_cost', 0):.2f} (batch + prompt cache)\n" - f"Worst case: ${est.get('batch_nocache_cost', 0):.2f} (batch, no cache hits)\n" - f"Live API would be: ${est.get('live_cost', 0):.2f}\n\n" - "One submission covers all files. The batch usually finishes within an hour. " - "You can stop safely and resume later." + f"{n_files} file(s) scanned - {n_req} request(s) queued.\n" + f"Model: {est.get('model') or os.getenv('model', '') or '—'}\n\n" + "One Anthropic submission covers every queued request (50% batch discount). " + "You can stop safely after submit and resume later." ) + if hasattr(self, "batch_cost_cached"): + self.batch_cost_cached.setText( + f"Batch + cache\n${float(est.get('batch_cached_cost') or 0):.2f}" + ) + self.batch_cost_nocache.setText( + f"Batch worst-case\n${float(est.get('batch_nocache_cost') or 0):.2f}" + ) + self.batch_cost_live.setText( + f"Live API\n${float(est.get('live_cost') or 0):.2f}" + ) self.batch_submit_yes_btn.setText(f"Submit All ({n_req} requests)") elif phase == "polling": - self.batch_phase_title.setText("Batch Translate — Processing") + self._set_batch_steps(2) + self.batch_phase_title.setText("Batch Translate - Processing") self.batch_overall_bar.setRange(0, 100) self.batch_overall_bar.setValue(55) self.batch_pipeline_stack.setCurrentIndex(2) - self.batch_poll_status.setText("Submitted — waiting for Anthropic to process the batch…") + self.batch_poll_status.setText("Submitted - waiting for Anthropic to process the batch…") + if hasattr(self, "batch_poll_bar"): + self.batch_poll_bar.setRange(0, 100) + self.batch_poll_bar.setValue(0) + self.batch_poll_bar.setFormat("Waiting for first status…") elif phase == "poll_status": - if isinstance(payload, str) and payload.strip(): - self.batch_poll_status.setText(payload.strip()) - self.batch_overall_bar.setValue(min(75, self.batch_overall_bar.value() + 2)) + self._set_batch_steps(2) + self.batch_pipeline_stack.setCurrentIndex(2) + self._update_batch_poll_dashboard(payload) elif phase == "consume": - self.batch_phase_title.setText("Batch Translate — Pass 2/2: Write") + self._batch_consume_started = True + self._set_batch_steps(3) + self.batch_phase_title.setText("Batch Translate - Pass 2/2: Write") self.batch_overall_bar.setValue(80) self.batch_pipeline_stack.setCurrentIndex(3) + self.batch_consume_status.setText("Pass 2/2: writing translated files from batch results…") self._reset_files_for_consume() if self.progress_tab_row.isVisible(): self._switch_progress_tab(0) elif phase == "done": + self._set_batch_steps(4) self.batch_overall_bar.setValue(100) - self.batch_phase_title.setText("Batch Translate — Complete") + self.batch_phase_title.setText("Batch Translate - Complete") + self.batch_pipeline_stack.setCurrentIndex(3) + self.batch_consume_status.setText( + "Pass 2/2 finished. Translations written - use the back arrow to return to the file list." + ) + self.batch_live_status.setText("Batch complete") self._update_batch_stop_button() def _reset_batch_pipeline_ui(self): self._batch_active = False self._batch_ui_phase = None + self._batch_consume_started = False if hasattr(self, "batch_pipeline_widget"): self.batch_pipeline_widget.setVisible(False) if hasattr(self, "batch_live_status"): @@ -1713,35 +1988,24 @@ class TranslationTab(QWidget): try: item["checkbox"].setChecked(False) item["label"].setText("Waiting...") - item["label"].setStyleSheet("color: #888888; font-size: 10px;") - item["progress_bar"].setVisible(True) item["progress_bar"].setValue(0) item["progress_bar"].setMaximum(100) - item["progress_bar"].setTextVisible(True) - item["progress_bar"].setStyleSheet(""" - QProgressBar { - border: 1px solid #555555; - border-radius: 2px; - text-align: center; - background-color: #2b2b2b; - color: white; - } - QProgressBar::chunk { - background-color: #007acc; - border-radius: 1px; - } - """) - item["tokens_label"].setVisible(False) item["tokens_label"].setText("") - item["cost_label"].setVisible(False) item["cost_label"].setText("") - item["time_label"].setVisible(False) item["time_label"].setText("") - item["status_label"].setVisible(False) item["status_label"].setText("") item.pop("_skip_reason", None) except Exception: pass + self._set_progress_row( + filename, + status="Waiting", + status_color="#888888", + progress="-", + tokens="-", + cost="-", + time_s="-", + ) if hasattr(self, "_applied_file_totals"): self._applied_file_totals.clear() self.totals_input_tokens = 0 @@ -1759,35 +2023,23 @@ class TranslationTab(QWidget): pass def mark_file_queued(self, filename): - """Collect pass finished for a file — queued for batch, not translated yet.""" + """Collect pass finished for a file - queued for batch, not translated yet.""" if filename not in self.file_progress_items: return item = self.file_progress_items[filename] try: item["label"].setText("Collected") - item["label"].setStyleSheet("color: #d4a017; font-weight: bold; font-size: 10px;") - item["progress_bar"].setVisible(True) item["progress_bar"].setMaximum(100) item["progress_bar"].setValue(100) - item["progress_bar"].setTextVisible(False) - item["progress_bar"].setStyleSheet(""" - QProgressBar { - border: 1px solid #555555; - border-radius: 2px; - background-color: #2b2b2b; - } - QProgressBar::chunk { - background-color: #6a5a20; - border-radius: 1px; - } - """) item["checkbox"].setChecked(False) - item["status_label"].setVisible(False) - item["tokens_label"].setVisible(False) - item["cost_label"].setVisible(False) - item["time_label"].setVisible(False) except Exception: pass + self._set_progress_row( + filename, + status="Collected", + status_color="#d4a017", + progress="queued", + ) def _check_model_pricing(self): """Fetch live pricing for the current model and print it to the log.""" @@ -1925,6 +2177,19 @@ class TranslationTab(QWidget): except Exception: pass return selected + + def select_files_by_name(self, file_names): + """Check the given relative file names (and uncheck others).""" + wanted = set(file_names or []) + if not wanted: + return + self.refresh_file_lists() + for i in range(self.file_list.count()): + item = self.file_list.item(i) + try: + item.setCheckState(Qt.Checked if item.text() in wanted else Qt.Unchecked) + except Exception: + pass def add_input_files(self): """Add files to the input directory, remembering last used directory.""" @@ -2135,111 +2400,141 @@ class TranslationTab(QWidget): except Exception as e: QMessageBox.warning(self, "Error", f"Failed to open folder:\n{str(e)}") + def _refresh_files_summary(self): + """Update the Files-tab summary line from current row states.""" + if not hasattr(self, "progress_files_summary"): + return + items = getattr(self, "file_progress_items", {}) or {} + total = len(items) + if total == 0: + self.progress_files_summary.setText("No files in this run.") + return + counts = {} + for meta in items.values(): + st = (meta.get("status_text") or "Waiting").split(" ")[0] + counts[st] = counts.get(st, 0) + 1 + parts = [f"{total} file(s)"] + for key in ( + "Done", "Collected", "Writing", "Scanning", "Translating", + "Failed", "Skipped", "Unsupported", "Waiting", + ): + if counts.get(key): + parts.append(f"{counts[key]} {key.lower()}") + self.progress_files_summary.setText(" · ".join(parts)) + + def _set_progress_row(self, filename, *, status=None, status_color=None, + progress=None, tokens=None, cost=None, time_s=None): + """Update one Files-tab table row.""" + meta = (getattr(self, "file_progress_items", {}) or {}).get(filename) + if not meta: + return + row = meta.get("row") + table = getattr(self, "progress_table", None) + if table is None or row is None or row >= table.rowCount(): + return + + def _cell(col, text, color=None): + item = table.item(row, col) + if item is None: + item = QTableWidgetItem("") + table.setItem(row, col, item) + item.setText(str(text)) + if color: + item.setForeground(QBrush(QColor(color))) + + if status is not None: + meta["status_text"] = status + _cell(1, status, status_color or "#cccccc") + if progress is not None: + _cell(2, progress, "#9cdcfe") + if tokens is not None: + _cell(3, tokens, "#f1c40f") + if cost is not None: + _cell(4, cost, "#4ec9b0") + if time_s is not None: + _cell(5, time_s, "#4da6ff") + self._refresh_files_summary() + def create_progress_item(self, filename): - """Create a progress item widget for a file.""" - widget = QWidget() - widget.setFixedHeight(36) # Fixed height instead of minimum - layout = QHBoxLayout() - layout.setContentsMargins(6, 4, 6, 4) # Reduced margins - layout.setSpacing(10) - - # Checkbox (initially unchecked, will check when done) + """Add a Files-tab table row for a file and return a placeholder widget.""" + table = getattr(self, "progress_table", None) + if table is None: + # Fallback empty widget if table is unavailable + return QWidget() + + row = table.rowCount() + table.insertRow(row) + values = [filename, "Waiting", "-", "-", "-", "-"] + colors = ["#ffffff", "#888888", "#666666", "#666666", "#666666", "#666666"] + for col, (val, color) in enumerate(zip(values, colors)): + item = QTableWidgetItem(val) + item.setForeground(QBrush(QColor(color))) + if col == 0: + item.setData(Qt.UserRole, filename) + table.setItem(row, col, item) + + # Keep a small compatibility widget so older helpers that expect + # checkbox/label/progress_bar keys do not crash. checkbox = QCheckBox() - checkbox.setEnabled(False) # Not interactive - checkbox.setFixedSize(18, 18) # Slightly smaller - layout.addWidget(checkbox) - - # Filename label - filename_label = QLabel(filename) - filename_label.setStyleSheet("font-weight: bold; color: white; font-size: 13px;") - filename_label.setFixedWidth(250) # Fixed width for consistent alignment - layout.addWidget(filename_label) - - # Progress label (small) shown next to filename + checkbox.setEnabled(False) + checkbox.setVisible(False) progress_label = QLabel("Waiting...") - progress_label.setStyleSheet("color: #888888; font-size: 10px;") - progress_label.setFixedWidth(110) - layout.addWidget(progress_label) - - # Progress bar (stretch to fill remaining space) + progress_label.setVisible(False) progress_bar = QProgressBar() - progress_bar.setMaximum(100) - progress_bar.setValue(0) - progress_bar.setFixedHeight(18) # Fixed height instead of minimum - progress_bar.setStyleSheet(""" - QProgressBar { - border: 1px solid #555555; - border-radius: 2px; - text-align: center; - background-color: #2b2b2b; - color: white; - } - QProgressBar::chunk { - background-color: #007acc; - border-radius: 1px; - } - """) - layout.addWidget(progress_bar, 1) # Stretch factor of 1 to fill remaining space - - # Inline result labels (hidden until completion) - anchored to the right - # Order (visual): tokens | time | cost | status(check) + progress_bar.setVisible(False) tokens_label = QLabel("") - tokens_label.setStyleSheet("color: #f1c40f; font-weight: bold; font-size: 9px;") - tokens_label.setFixedWidth(100) tokens_label.setVisible(False) - tokens_label.setAlignment(Qt.AlignRight | Qt.AlignVCenter) - layout.addWidget(tokens_label) - - time_label = QLabel("") - time_label.setStyleSheet("color: #4da6ff; font-weight: bold; font-size: 9px;") - time_label.setFixedWidth(90) - time_label.setVisible(False) - time_label.setAlignment(Qt.AlignRight | Qt.AlignVCenter) - layout.addWidget(time_label) - - # Cost label (to the left of the status/checkmark) cost_label = QLabel("") - cost_label.setStyleSheet("color: #4ec9b0; font-weight: bold; font-size: 9px;") - cost_label.setFixedWidth(100) cost_label.setVisible(False) - cost_label.setAlignment(Qt.AlignRight | Qt.AlignVCenter) - layout.addWidget(cost_label) - - # Small status label (checkmark or X) placed to the right of cost + time_label = QLabel("") + time_label.setVisible(False) status_label = QLabel("") - status_label.setStyleSheet("font-weight: bold; font-size: 11px;") - status_label.setFixedWidth(100) status_label.setVisible(False) - status_label.setAlignment(Qt.AlignCenter) - layout.addWidget(status_label) - - widget.setLayout(layout) - - # Store references + shim = QWidget() + shim.setVisible(False) + self.file_progress_items[filename] = { - 'widget': widget, - 'checkbox': checkbox, - 'label': progress_label, - 'progress_bar': progress_bar, - 'tokens_label': tokens_label, - 'cost_label': cost_label, - 'time_label': time_label, - 'status_label': status_label + "row": row, + "status_text": "Waiting", + "widget": shim, + "checkbox": checkbox, + "label": progress_label, + "progress_bar": progress_bar, + "tokens_label": tokens_label, + "cost_label": cost_label, + "time_label": time_label, + "status_label": status_label, } - - return widget - + self._refresh_files_summary() + return shim + def update_file_item_progress(self, filename, current, total): """Update progress for a specific file.""" - if filename in self.file_progress_items: - item = self.file_progress_items[filename] - item['progress_bar'].setMaximum(total if total > 0 else 100) - item['progress_bar'].setValue(current) - item['label'].setText(f"{current}/{total}") - item['label'].setStyleSheet("color: #007acc; font-weight: bold;") - + if filename not in self.file_progress_items: + return + item = self.file_progress_items[filename] + try: + item["progress_bar"].setMaximum(total if total > 0 else 100) + item["progress_bar"].setValue(current) + item["label"].setText(f"{current}/{total}") + except Exception: + pass + phase = getattr(self, "_batch_ui_phase", None) + if getattr(self, "_batch_active", False) and phase == "collect": + status, color = "Scanning", "#007acc" + elif getattr(self, "_batch_active", False) and phase == "consume": + status, color = "Writing", "#007acc" + else: + status, color = "Translating", "#007acc" + self._set_progress_row( + filename, + status=status, + status_color=color, + progress=f"{current}/{total}" if total else "-", + ) + def _apply_success_status_icon(self, item, completion_kind="normal"): - """Green checkmark for every successful row; tooltips explain skip / idle when relevant.""" + """Mark a successful row; tooltips explain skip / idle when relevant.""" try: item["status_label"].setText("✓") item["status_label"].setStyleSheet( @@ -2248,12 +2543,20 @@ class TranslationTab(QWidget): if completion_kind == "skip": reason = (item.get("_skip_reason") or "").strip() tip = f"Skipped: {reason}" if reason else "Whole file skipped (paths/fonts only)." + status, color = "Skipped", "#dcdcaa" elif completion_kind == "idle": tip = "No translatable lines (non-dialogue content only)." + status, color = "Done", "#4ec9b0" else: tip = "" + status, color = "Done", "#4ec9b0" item["status_label"].setToolTip(tip) item["status_label"].setVisible(True) + # Mirror into the table when we know the filename + for fname, meta in (getattr(self, "file_progress_items", {}) or {}).items(): + if meta is item: + self._set_progress_row(fname, status=status, status_color=color, progress="✓") + break except Exception: pass @@ -2311,30 +2614,44 @@ class TranslationTab(QWidget): if is_unsupported: item['status_label'].setText("⚠") item['status_label'].setStyleSheet("color: #f1c40f; font-weight: bold; font-size: 13px;") + item['label'].setText("Not Supported") + self._set_progress_row( + filename, + status="Unsupported", + status_color="#f1c40f", + progress="-", + ) else: item['status_label'].setText("✗") item['status_label'].setStyleSheet("color: #f48771; font-weight: bold; font-size: 11px;") + item['label'].setText("Failed") + self._set_progress_row( + filename, + status="Failed", + status_color="#f48771", + progress="-", + ) item['status_label'].setVisible(True) - # Set tooltip with error message if provided if error_message: item['status_label'].setToolTip(f"Error: {error_message}") item['widget'].setToolTip(f"Error: {error_message}") except Exception: pass try: - if is_unsupported: - item['label'].setText("Not Supported") - item['label'].setStyleSheet("color: #f1c40f; font-weight: bold; font-size: 10px;") - else: - item['label'].setText("Failed") - item['label'].setStyleSheet("color: #f48771; font-weight: bold; font-size: 10px;") - except Exception: - pass - try: - # Hide progress bar and show error state item['progress_bar'].setVisible(False) except Exception: pass + # Always refresh table status for success paths that used the icon helper + if success: + tokens = item.get("tokens_label").text() if item.get("tokens_label") else "" + cost = item.get("cost_label").text() if item.get("cost_label") else "" + time_s = item.get("time_label").text() if item.get("time_label") else "" + self._set_progress_row( + filename, + tokens=tokens or None, + cost=cost or None, + time_s=time_s or None, + ) def reset_to_file_view(self): """Reset back to file selection view.""" @@ -2364,12 +2681,14 @@ class TranslationTab(QWidget): self.stop_button.setVisible(False) self.refresh_file_lists() - def start_translation(self, skip_confirm: bool = False): + def start_translation(self, skip_confirm: bool = False, forced_resume_state: str | None = None): """Start the translation process. skip_confirm: when True the confirmation dialog is bypassed (used when called programmatically from the Workflow tab so the user doesn't need an extra click to confirm what they just explicitly requested). + forced_resume_state: when set (from the Batch history tab), force Batch + Translate mode and resume with this state without the Resume? dialog. """ # Get checked files selected_files = self.get_selected_files() @@ -2388,6 +2707,12 @@ class TranslationTab(QWidget): # Get mode from dropdown mode = self.mode_combo.currentText() + if forced_resume_state: + # Batch history Resume always runs Batch Translate. + idx = self.mode_combo.findText(BATCH_MODE_LABEL) + if idx >= 0: + self.mode_combo.setCurrentIndex(idx) + mode = BATCH_MODE_LABEL estimate_only = (mode == "Estimate") parse_speakers = (mode == "Parse Speakers") batch_mode = (mode == BATCH_MODE_LABEL) @@ -2410,28 +2735,51 @@ class TranslationTab(QWidget): "pointing at anthropic.com.\n\nChange your model/API settings and try again.", ) return - if skip_confirm: + if forced_resume_state: + batch_resume_state = forced_resume_state + elif skip_confirm: # Workflow auto-start: each phase is an independent batch run. # Never resume stale queue/results left over from a prior phase. - from util.translation import clearBatchFiles + from util.translation import clearBatchFiles, batchRunState as _brs + prior = _brs() + if prior: + # Log discard so operators notice unpaid queue / in-flight work. + print( + f"[BATCH] Workflow start discarding prior batch state ({prior}).", + flush=True, + ) clearBatchFiles() batch_resume_state = None else: batch_resume_state = batchRunState() if batch_resume_state: - reply = QMessageBox.question( - self, - "Resume Batch?", - f"A previous batch run was interrupted ({batch_resume_state}).\n\n" - "Resume it instead of re-collecting?\n" - "(Re-collecting would queue new requests and bill again.)", - QMessageBox.Yes | QMessageBox.No, - ) + if batch_resume_state == "queued": + reply = QMessageBox.question( + self, + "Resume Queued Batch?", + "A previous collect finished but the batch was not submitted.\n" + "The queue is still in log/batch_requests.json.\n\n" + "Resume and submit that queue?\n\n" + "Choosing No discards the queue and re-collects " + "(speaker/name strings bill at live rates again).", + QMessageBox.Yes | QMessageBox.No, + ) + else: + reply = QMessageBox.question( + self, + "Resume Batch?", + f"A previous batch run was interrupted ({batch_resume_state}).\n\n" + "Resume it instead of re-collecting?\n\n" + "Re-collecting discards the current run and can bill again " + "(live collect charges + a new batch submission).", + QMessageBox.Yes | QMessageBox.No, + ) if reply != QMessageBox.Yes: batch_resume_state = None - # Confirm start (skipped when called programmatically from the Workflow tab) - if not skip_confirm: + # Confirm start (skipped when called programmatically from the Workflow tab + # or when Batch history already confirmed Resume). + if not skip_confirm and not forced_resume_state: if batch_mode and not batch_resume_state: reply = QMessageBox.question( self, @@ -2457,18 +2805,16 @@ class TranslationTab(QWidget): # Switch to progress view self.file_stack.setCurrentIndex(1) - # Initialize progress list with all files + # Initialize Files-tab table with all selected files self.progress_list.clear() self.file_progress_items.clear() - + if hasattr(self, "progress_table"): + self.progress_table.setRowCount(0) + if hasattr(self, "progress_files_summary"): + self.progress_files_summary.setText("No files in this run.") + for filename in selected_files: - item_widget = self.create_progress_item(filename) - list_item = QListWidgetItem(self.progress_list) - # Set fixed size hint to match widget height - from PyQt5.QtCore import QSize - list_item.setSizeHint(QSize(0, 36)) # Width 0 = auto, height 36px - self.progress_list.addItem(list_item) - self.progress_list.setItemWidget(list_item, item_widget) + self.create_progress_item(filename) # Toggle button visibility self.translate_button.setVisible(False) @@ -2519,15 +2865,22 @@ class TranslationTab(QWidget): self.files_total = len(selected_files) self._batch_active = batch_mode self._batch_ui_phase = None + self._batch_consume_started = False + self._finish_pending = None if batch_mode: self._set_progress_view_mode(True, len(selected_files)) self.batch_overall_bar.setValue(0) self.batch_pipeline_stack.setCurrentIndex(0) self.batch_live_status.setText("") + # Resume-from-fetched jumps straight to write; arm the UI before the + # worker starts so early cost lines are not dropped as "pre-consume". + if batch_resume_state == "fetched": + self._on_batch_phase("consume", None) else: self._set_progress_view_mode(False, len(selected_files)) self._batch_active = False self._batch_ui_phase = None + self._batch_consume_started = False self.files_translated_label.setText(f"0/{self.files_total}") self.translating_label.setText("Starting...") self.item_progress_label.setText("0/0") @@ -2579,6 +2932,22 @@ class TranslationTab(QWidget): run_log_path = history_dir / fname # Don't create the file yet - it will be created when first log is written + # Create the per-run log immediately with a header so the Translation + # Log panel is never blank while a batch resume/consume runs. + try: + mode = BATCH_MODE_LABEL if batch_mode else mode + with open(run_log_path, "a", encoding="utf-8") as f: + f.write( + f"[RUN] {mode}" + + (f" resume={batch_resume_state}" if batch_resume_state else "") + + f" files={len(selected_files)}\n" + ) + for name in selected_files: + f.write(f"[RUN] - {name}\n") + f.flush() + except Exception: + pass + # Export env var so subprocess workers inherit the path try: os.environ['TRANSLATION_RUN_LOG'] = str(run_log_path) @@ -2628,8 +2997,14 @@ class TranslationTab(QWidget): except Exception: pass # During batch collect/poll/submit, per-file cost lines are not final translations. - if getattr(self, "_batch_active", False) and getattr(self, "_batch_ui_phase", None) != "consume": - return + # Accept costs once consume has started (and after done) - resume→consume is fast + # enough that log lines from the pool thread can arrive before/after the phase + # signal is processed on the UI thread. + if getattr(self, "_batch_active", False): + phase = getattr(self, "_batch_ui_phase", None) + consume_started = getattr(self, "_batch_consume_started", False) + if not consume_started and phase not in ("consume", "done"): + return try: stripped = _strip_ansi(message) pattern = ( @@ -2689,11 +3064,14 @@ class TranslationTab(QWidget): item["_skip_reason"] = skip_reason # Distinguish "no API usage" from real token counts in the list UI if completion_kind in ("skip", "idle"): - item["tokens_label"].setText("—") + tokens_text = "-" else: - item["tokens_label"].setText(f"{input_tokens}/{output_tokens}") - item['cost_label'].setText(f"${cost:.4f}") - item['time_label'].setText(f"{time_s:.1f}s") + tokens_text = f"{input_tokens}/{output_tokens}" + cost_text = f"${cost:.4f}" + time_text = f"{time_s:.1f}s" + item["tokens_label"].setText(tokens_text) + item['cost_label'].setText(cost_text) + item['time_label'].setText(time_text) item['tokens_label'].setVisible(True) item['cost_label'].setVisible(True) item['time_label'].setVisible(True) @@ -2701,6 +3079,12 @@ class TranslationTab(QWidget): item['progress_bar'].setVisible(False) except Exception: pass + self._set_progress_row( + filename, + tokens=tokens_text, + cost=cost_text, + time_s=time_text, + ) except Exception: pass diff --git a/modules/main.py b/modules/main.py index 8033c61..40b23b8 100644 --- a/modules/main.py +++ b/modules/main.py @@ -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) diff --git a/tests/test_batch_history.py b/tests/test_batch_history.py new file mode 100644 index 0000000..560b2d0 --- /dev/null +++ b/tests/test_batch_history.py @@ -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() diff --git a/util/batch_history.py b/util/batch_history.py new file mode 100644 index 0000000..53bf837 --- /dev/null +++ b/util/batch_history.py @@ -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) diff --git a/util/translation.py b/util/translation.py index ce667cd..5ccd1fa 100644 --- a/util/translation.py +++ b/util/translation.py @@ -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