fix(translation): skip ThinkingBlock when reading Claude replies

- Join only content blocks that expose .text so live Anthropic calls
  no longer crash when thinking precedes the TextBlock
- Add unit coverage for thinking-then-text response shapes
This commit is contained in:
DazedAnon 2026-07-25 13:22:07 -05:00
parent fca1f8c1c6
commit ba8b0e3bf6
2 changed files with 49 additions and 2 deletions

View file

@ -0,0 +1,36 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Tests for Anthropic content-block text extraction."""
import unittest
from types import SimpleNamespace
from util.translation import _anthropic_content_text
class AnthropicContentTextTests(unittest.TestCase):
def test_empty_content(self):
self.assertEqual(_anthropic_content_text(None), "")
self.assertEqual(_anthropic_content_text([]), "")
def test_text_block_only(self):
content = [SimpleNamespace(type="text", text='{"0":"Hello"}')]
self.assertEqual(_anthropic_content_text(content), '{"0":"Hello"}')
def test_skips_thinking_block_without_text(self):
# Mirrors Anthropic ThinkingBlock: has .thinking, no .text.
thinking = SimpleNamespace(type="thinking", thinking="internal notes")
text = SimpleNamespace(type="text", text='{"0":"Alice"}')
self.assertEqual(_anthropic_content_text([thinking, text]), '{"0":"Alice"}')
def test_accessing_first_block_text_would_fail(self):
thinking = SimpleNamespace(type="thinking", thinking="notes")
text = SimpleNamespace(type="text", text="ok")
content = [thinking, text]
with self.assertRaises(AttributeError):
_ = content[0].text
self.assertEqual(_anthropic_content_text(content), "ok")
if __name__ == "__main__":
unittest.main()

View file

@ -1187,7 +1187,7 @@ def fetchTranslationBatches(batches=None):
errored.append((r.custom_id, detail)) errored.append((r.custom_id, detail))
continue continue
msg = res.message msg = res.message
text = "".join(getattr(b, "text", "") or "" for b in msg.content) text = _anthropic_content_text(msg.content)
u = msg.usage u = msg.usage
cr = getattr(u, "cache_read_input_tokens", 0) or 0 cr = getattr(u, "cache_read_input_tokens", 0) or 0
cw = getattr(u, "cache_creation_input_tokens", 0) or 0 cw = getattr(u, "cache_creation_input_tokens", 0) or 0
@ -1998,6 +1998,17 @@ def createTranslationSchema(numLines):
} }
def _anthropic_content_text(content) -> str:
"""Join text from Anthropic message content blocks.
Newer Claude models often return a ThinkingBlock (no ``.text``) before the
TextBlock. Only blocks that expose ``.text`` contribute to the result.
"""
if not content:
return ""
return "".join(getattr(b, "text", "") or "" for b in content)
class _AnthropicCompat: class _AnthropicCompat:
"""OpenAI-shaped wrapper around an Anthropic response (text + usage).""" """OpenAI-shaped wrapper around an Anthropic response (text + usage)."""
class _Usage: class _Usage:
@ -2232,7 +2243,7 @@ def translateText(system, user, history, penalty, formatType, model, numLines=No
except Exception as e: except Exception as e:
raise Exception(f"Anthropic API error: {e}") raise Exception(f"Anthropic API error: {e}")
_ant_text = ant_resp.content[0].text if ant_resp.content else "" _ant_text = _anthropic_content_text(ant_resp.content)
_u = ant_resp.usage _u = ant_resp.usage
_cr = getattr(_u, "cache_read_input_tokens", 0) or 0 _cr = getattr(_u, "cache_read_input_tokens", 0) or 0
_cw = getattr(_u, "cache_creation_input_tokens", 0) or 0 _cw = getattr(_u, "cache_creation_input_tokens", 0) or 0