"""Translator module using Claude CLI.""" from __future__ import annotations import asyncio import json import logging import os import re import subprocess import time from collections.abc import Callable from typing import TYPE_CHECKING, Any from src.context.builder import ContextBuilder from src.data.storage import Storage from src.exceptions import ( ChunkParsingError, ClaudeAPIError, EvaluationError, QualityThresholdError, RateLimitError, classify_claude_error, ) from src.models.pericope import ChunkTranslationContext, ChunkTranslationOutput, Pericope from src.models.verse import TranslationOutput, VerseReference from src.pipeline.prompt import PromptBuilder from src.utils.furigana import parse_furigana from src.utils.retry import RetryConfig, RetryState if TYPE_CHECKING: from src.context.translation_memory import TranslationMemoryManager from src.evaluation.evaluator import TranslationEvaluator from src.evaluation.storage import EvaluationStorage from src.models.evaluation import VerseEvaluation from src.models.verse import TranslationContext from src.orchestration.provenance import TranslationProvenance logger = logging.getLogger(__name__) # Find Claude binary CLAUDE_BINARY = os.environ.get("CLAUDE_BINARY", "claude") def _clean_env() -> dict[str, str]: """Return a copy of the environment without CLAUDECODE to avoid nested session errors.""" env = os.environ.copy() env.pop("CLAUDECODE", None) return env DEFAULT_MODEL = os.environ.get("TRANSLATE_BIBLE_MODEL", "claude-opus-4-8") def _sanitize_translation_output(raw: str) -> str: """Strip meta-commentary from a verse translation response. Models occasionally emit self-corrections ("Wait—let me provide the correct translation.") or markdown fences around the answer. A verse translation is a single Japanese block, so: drop code fences, drop predominantly-Latin blocks, and if several Japanese blocks remain (a self-correction), keep the last one. """ text = raw.strip() if not text: return text lines = [ln for ln in text.split("\n") if not ln.strip().startswith("```")] blocks: list[list[str]] = [[]] for ln in lines: if ln.strip(): blocks[-1].append(ln) elif blocks[-1]: blocks.append([]) block_texts = ["\n".join(b).strip() for b in blocks if b] def is_japanese(block: str) -> bool: cjk = sum(1 for ch in block if "぀" <= ch <= "鿿" or "！" <= ch <= "｠") latin = sum(1 for ch in block if ch.isascii() and ch.isalpha()) return cjk > latin japanese_blocks = [b for b in block_texts if is_japanese(b)] if not japanese_blocks: return text result = japanese_blocks[-1] result = re.sub(r"^(翻訳|訳文|Translation)\s*[:：]\s*", "", result) if result != text: logger.warning("Sanitized translation output (removed meta-commentary)") return result # Default retry configuration DEFAULT_RETRY_CONFIG = RetryConfig( max_attempts=3, base_delay=2.0, max_delay=60.0, exponential_base=2.0, retryable_exceptions=(RateLimitError, ConnectionError, TimeoutError), ) # Quality thresholds for evaluation gating DEFAULT_QUALITY_THRESHOLD = 70.0 THEOLOGICAL_ACCURACY_THRESHOLD = 75.0 class Translator: """Translates Bible verses using Claude CLI.""" def __init__( self, context_builder: ContextBuilder, prompt_builder: PromptBuilder, storage: Storage, model: str = DEFAULT_MODEL, rate_limit_delay: float = 1.2, translation_memory: TranslationMemoryManager | None = None, auto_evaluate: bool = True, evaluator: TranslationEvaluator | None = None, evaluation_storage: EvaluationStorage | None = None, retry_config: RetryConfig | None = None, quality_threshold: float = DEFAULT_QUALITY_THRESHOLD, enforce_quality_gate: bool = False, subprocess_timeout: float | None = None, enable_proto_archive: bool = False, ) -> None: """Initialize the translator. Args: context_builder: Context builder for creating translation context. prompt_builder: Prompt builder for formatting prompts. storage: Storage for saving translations. model: Claude model to use. rate_limit_delay: Delay between API calls in seconds. translation_memory: Optional TM manager for reuse. auto_evaluate: Whether to automatically evaluate translations. evaluator: Evaluator to use for auto-evaluation. evaluation_storage: Storage for evaluation results. retry_config: Configuration for retry with backoff. quality_threshold: Minimum score for evaluation gating. enforce_quality_gate: If True, reject low-quality translations. subprocess_timeout: Timeout for Claude CLI calls in seconds. Defaults to the TRANSLATE_BIBLE_TIMEOUT env var, or 120. Long pericope chunks (10+ verses) can need 300s+. enable_proto_archive: Whether to archive translation inputs/outputs. """ self.context_builder = context_builder self.prompt_builder = prompt_builder self.storage = storage self.model = model self.rate_limit_delay = rate_limit_delay self._translation_memory = translation_memory self._auto_evaluate = auto_evaluate self._evaluator = evaluator self._evaluation_storage = evaluation_storage self._retry_config = retry_config or DEFAULT_RETRY_CONFIG self._quality_threshold = quality_threshold self._enforce_quality_gate = enforce_quality_gate if subprocess_timeout is None: subprocess_timeout = float(os.environ.get("TRANSLATE_BIBLE_TIMEOUT", "120")) self._subprocess_timeout = subprocess_timeout self._enable_archive = enable_proto_archive # Enable archive on storage if requested if enable_proto_archive: self.storage.enable_archive = True @property def enable_proto_archive(self) -> bool: """Get proto archive enabled status.""" return self._enable_archive @enable_proto_archive.setter def enable_proto_archive(self, value: bool) -> None: """Set proto archive enabled status.""" self._enable_archive = value self.storage.enable_archive = value @property def translation_memory(self) -> TranslationMemoryManager | None: """Get the translation memory manager.""" return self._translation_memory @translation_memory.setter def translation_memory(self, value: TranslationMemoryManager | None) -> None: """Set the translation memory manager.""" self._translation_memory = value @property def auto_evaluate(self) -> bool: """Get auto-evaluation setting.""" return self._auto_evaluate @auto_evaluate.setter def auto_evaluate(self, value: bool) -> None: """Set auto-evaluation setting.""" self._auto_evaluate = value def _call_claude(self, system_prompt: str, user_prompt: str) -> str: """Call Claude CLI with the given prompts. Includes retry logic with exponential backoff for transient errors. Args: system_prompt: System prompt for Claude. user_prompt: User prompt for Claude. Returns: The response text from Claude. Raises: ClaudeAPIError: If the Claude CLI call fails after all retries. """ retry_state = RetryState(self._retry_config) while True: try: result = subprocess.run( [ CLAUDE_BINARY, "--print", "-p", user_prompt, "--system-prompt", system_prompt, "--model", self.model, ], capture_output=True, text=True, check=False, timeout=self._subprocess_timeout, env=_clean_env(), ) if result.returncode != 0: error = classify_claude_error(result.stderr, result.returncode) raise error return result.stdout.strip() except subprocess.TimeoutExpired as e: error = ClaudeAPIError( f"Claude CLI timed out after {self._subprocess_timeout}s", stderr=str(e), return_code=-1, ) if retry_state.should_retry(error): logger.warning( "Claude CLI timeout, attempt %d/%d. Retrying...", retry_state.attempt + 1, self._retry_config.max_attempts, ) retry_state.wait() else: raise error from e except FileNotFoundError as e: # Claude CLI not installed from src.exceptions import CLINotFoundError raise CLINotFoundError() from e except ClaudeAPIError as e: if retry_state.should_retry(e): logger.warning( "Claude API error, attempt %d/%d: %s. Retrying...", retry_state.attempt + 1, self._retry_config.max_attempts, str(e), ) retry_state.wait() else: raise async def _call_claude_async(self, system_prompt: str, user_prompt: str) -> str: """Call Claude CLI asynchronously with the given prompts. Includes retry logic with exponential backoff for transient errors. Args: system_prompt: System prompt for Claude. user_prompt: User prompt for Claude. Returns: The response text from Claude. Raises: ClaudeAPIError: If the Claude CLI call fails after all retries. """ retry_state = RetryState(self._retry_config) while True: try: process = await asyncio.wait_for( asyncio.create_subprocess_exec( CLAUDE_BINARY, "--print", "-p", user_prompt, "--system-prompt", system_prompt, "--model", self.model, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, env=_clean_env(), ), timeout=self._subprocess_timeout, ) stdout, stderr = await asyncio.wait_for( process.communicate(), timeout=self._subprocess_timeout, ) if process.returncode != 0: error = classify_claude_error(stderr.decode(), process.returncode or 1) raise error return stdout.decode().strip() except asyncio.TimeoutError as e: error = ClaudeAPIError( f"Claude CLI timed out after {self._subprocess_timeout}s", stderr="Async timeout", return_code=-1, ) if retry_state.should_retry(error): logger.warning( "Claude CLI async timeout, attempt %d/%d. Retrying...", retry_state.attempt + 1, self._retry_config.max_attempts, ) await retry_state.async_wait() else: raise error from e except FileNotFoundError as e: from src.exceptions import CLINotFoundError raise CLINotFoundError() from e except ClaudeAPIError as e: if retry_state.should_retry(e): logger.warning( "Claude API error (async), attempt %d/%d: %s. Retrying...", retry_state.attempt + 1, self._retry_config.max_attempts, str(e), ) await retry_state.async_wait() else: raise def _get_tm_matches(self, context: TranslationContext) -> list | None: """Get TM matches for the context if TM is enabled. Args: context: Translation context. Returns: List of TM matches or None if TM not available. """ if self._translation_memory is None: return None # Get source text for matching source_text = ( context.target_verse.greek_text or context.target_verse.hebrew_text or "" ) if not source_text: return None matches = self._translation_memory.find_matches( source_text, threshold=0.7, max_results=3 ) return matches if matches else None def _add_to_tm( self, context: TranslationContext, translation_text: str ) -> None: """Add completed translation to translation memory. Args: context: Translation context. translation_text: The translated text. """ if self._translation_memory is None: return source_text = ( context.target_verse.greek_text or context.target_verse.hebrew_text ) if not source_text: return self._translation_memory.add_from_translation( source_text=source_text, target_text=translation_text, reference=context.target_verse.reference, metadata={"model": self.model}, ) def _evaluate_translation( self, output: TranslationOutput, context: TranslationContext, ) -> VerseEvaluation | None: """Evaluate a translation if auto-evaluate is enabled. Args: output: The translation output. context: The translation context. Returns: VerseEvaluation or None if evaluation failed/disabled. Raises: EvaluationError: If enforce_quality_gate is True and evaluation fails. """ if not self._auto_evaluate or not self._evaluator: return None ref = output.reference ref_str = f"{ref.book} {ref.chapter}:{ref.verse}" try: from src.evaluation.evaluator import ( EVALUATION_SYSTEM_PROMPT, parse_evaluation_response, ) # Build evaluation prompt eval_prompt = self._evaluator.build_verse_prompt( output.text, context ) # Call Claude for evaluation response = self._call_claude(EVALUATION_SYSTEM_PROMPT, eval_prompt) # Parse response scores = parse_evaluation_response(response) # Create evaluation evaluation = self._evaluator.create_verse_evaluation( verse_number=output.reference.verse, scores=scores, ) # Log evaluation result logger.info( "Evaluated %s: overall=%.1f%%, flags=%s", ref_str, evaluation.overall_score, evaluation.flags or "none", ) return evaluation except Exception as e: # Log the error with details logger.error( "Evaluation failed for %s: %s", ref_str, str(e), ) # If quality gate is enforced, raise the error if self._enforce_quality_gate: raise EvaluationError( f"Evaluation failed for {ref_str}: {e}", verse_ref=ref_str, ) from e # Otherwise, return None (don't block translation) return None def _check_quality_gate( self, evaluation: VerseEvaluation, reference: VerseReference, ) -> None: """Check if translation passes quality threshold. Args: evaluation: The verse evaluation. reference: The verse reference. Raises: QualityThresholdError: If translation fails quality gate. """ if not self._enforce_quality_gate: return ref_str = f"{reference.book} {reference.chapter}:{reference.verse}" # Check overall score if evaluation.overall_score < self._quality_threshold: raise QualityThresholdError( f"Translation for {ref_str} failed quality gate: " f"{evaluation.overall_score:.1f}% < {self._quality_threshold}%", score=evaluation.overall_score, threshold=self._quality_threshold, ) # Check theological accuracy specifically if "theological_accuracy" in evaluation.scores: ta_score = evaluation.scores["theological_accuracy"].score if ta_score < THEOLOGICAL_ACCURACY_THRESHOLD: raise QualityThresholdError( f"Translation for {ref_str} failed theological accuracy: " f"{ta_score:.1f}% < {THEOLOGICAL_ACCURACY_THRESHOLD}%", score=ta_score, threshold=THEOLOGICAL_ACCURACY_THRESHOLD, dimension="theological_accuracy", ) def _save_evaluation( self, evaluation: VerseEvaluation, reference: VerseReference, ) -> None: """Save evaluation result if storage is configured. Args: evaluation: The evaluation to save. reference: The verse reference. """ if not self._evaluation_storage: return self._evaluation_storage.save_verse_evaluation( reference.book, reference.chapter, evaluation, ) def translate_verse( self, reference: VerseReference, skip_existing: bool = True, evaluate: bool | None = None, ) -> TranslationOutput: """Translate a single verse. Args: reference: Verse reference to translate. skip_existing: If True, return existing translation if available. evaluate: Override auto_evaluate setting for this call. Returns: TranslationOutput with the translation. """ # Check if already translated if skip_existing and self.storage.is_verse_translated(reference): existing = self.storage.load_translation(reference) if existing: return existing # Build context and prompt context = self.context_builder.build_context(reference) # Get TM matches tm_matches = self._get_tm_matches(context) # Build prompt (with TM matches if available) prompt = self.prompt_builder.build_prompt(context, tm_matches=tm_matches) system_prompt = self.prompt_builder.get_system_prompt() # Call Claude CLI translation_text = _sanitize_translation_output( self._call_claude(system_prompt, prompt) ) # Parse furigana from Claude's response plain_text, annotations = parse_furigana(translation_text) # Create output output = TranslationOutput( reference=reference, text=plain_text, text_with_furigana=translation_text, furigana_annotations=[ { "start_offset": a.start_offset, "end_offset": a.end_offset, "base_text": a.base_text, "reading": a.reading, } for a in annotations ], model=self.model, sources_used=self._get_sources_used(context), glossary_terms_applied=[t.original for t in context.glossary_terms], ) # Save translation self.storage.save_translation(output) # Add to TM self._add_to_tm(context, plain_text) # Evaluate if enabled should_evaluate = evaluate if evaluate is not None else self._auto_evaluate if should_evaluate: evaluation = self._evaluate_translation(output, context) if evaluation: self._save_evaluation(evaluation, reference) # Check quality gate self._check_quality_gate(evaluation, reference) return output def translate_verse_with_provenance( # pragma: no cover self, reference: VerseReference, skip_existing: bool = True, ) -> tuple[TranslationOutput, "TranslationProvenance"]: """Translate a verse and capture full provenance. Args: reference: Verse reference to translate. skip_existing: If True, return existing translation if available. Returns: Tuple of (TranslationOutput, TranslationProvenance). """ from datetime import datetime from src.orchestration.provenance import ( TranslationProvenance, compute_glossary_hash, compute_style_hash, ) # Check if already translated if skip_existing and self.storage.is_verse_translated(reference): existing = self.storage.load_translation(reference) if existing: # Try to load existing provenance provenance = self.storage.load_provenance(reference) if provenance: return existing, provenance # Create minimal provenance for existing translation provenance = TranslationProvenance(model=existing.model) return existing, provenance # Track timing started_at = datetime.utcnow() # Build context context = self.context_builder.build_context(reference) # Get TM matches tm_matches = self._get_tm_matches(context) or [] # Build prompts prompt = self.prompt_builder.build_prompt(context, tm_matches=tm_matches) system_prompt = self.prompt_builder.get_system_prompt() # Build provenance provenance = TranslationProvenance( model=self.model, prompt_version="v1.0", system_prompt=system_prompt, user_prompt=prompt, started_at=started_at, attempt_number=1, ) # Capture source texts verse = context.target_verse if verse.greek_text: provenance.source_texts["greek"] = verse.greek_text if verse.hebrew_text: provenance.source_texts["hebrew"] = verse.hebrew_text if verse.english_kjv: provenance.source_texts["kjv"] = verse.english_kjv if verse.english_asv: provenance.source_texts["asv"] = verse.english_asv if verse.japanese_meiji: provenance.source_texts["meiji"] = verse.japanese_meiji if verse.japanese_taisho: provenance.source_texts["taisho"] = verse.japanese_taisho # Compute source hashes provenance.compute_source_hashes() # Capture context verses provenance.preceding_verses = [ str(v.reference) for v in context.preceding_verses ] provenance.following_verses = [ str(v.reference) for v in context.following_verses ] # Capture chapter summary if context.chapter_summary: provenance.chapter_summary = context.chapter_summary.summary # Capture book metadata if context.book_metadata: provenance.book_metadata = { "name": context.book_metadata.name_english, "testament": context.book_metadata.testament, "themes": context.book_metadata.themes, } # Capture glossary hash provenance.glossary_hash = compute_glossary_hash(context.glossary_terms) # Capture style hash if self.prompt_builder._style: style_dict = self.prompt_builder._style.to_dict() provenance.style_config_hash = compute_style_hash(style_dict) # Capture TM matches if tm_matches: provenance.tm_matches_used = [ { "source": m.segment.source_text, "target": m.segment.target_text, "similarity": m.similarity, } for m in tm_matches ] # Create input proto for archive (before translation) input_proto = None if self._enable_archive: from src.protos.converters import ( translation_context_to_prompt_input, tm_match_to_proto, ) from src.protos.serialization import compute_proto_hash input_proto = translation_context_to_prompt_input( context, system_prompt, prompt, self.model, ) # Add TM matches to input proto if tm_matches and hasattr(tm_matches[0], "segment"): input_proto.tm_matches = [tm_match_to_proto(m) for m in tm_matches] # Call Claude translation_text = self._call_claude(system_prompt, prompt) # Complete timing completed_at = datetime.utcnow() provenance.completed_at = completed_at # Parse furigana from Claude's response plain_text, annotations = parse_furigana(translation_text) # Create output output = TranslationOutput( reference=reference, text=plain_text, text_with_furigana=translation_text, furigana_annotations=[ { "start_offset": a.start_offset, "end_offset": a.end_offset, "base_text": a.base_text, "reading": a.reading, } for a in annotations ], model=self.model, sources_used=list(provenance.source_texts.keys()), glossary_terms_applied=[t.original for t in context.glossary_terms], ) # Save translation with provenance self.storage.save_translation_with_provenance(output, provenance) # Save proto archive if enabled if self._enable_archive and input_proto is not None: from src.protos.converters import ( provenance_to_proto, translation_result_to_proto, ) from src.protos.serialization import compute_proto_hash output_proto = translation_result_to_proto( output, translation_text, annotations, input_hash=compute_proto_hash(input_proto) ) output_proto.provenance = provenance_to_proto(provenance) self.storage.save_translation_archive(reference, input_proto, output_proto) # Add to TM self._add_to_tm(context, plain_text) return output, provenance async def translate_verse_async( self, reference: VerseReference, skip_existing: bool = True ) -> TranslationOutput: """Translate a single verse asynchronously. Args: reference: Verse reference to translate. skip_existing: If True, return existing translation if available. Returns: TranslationOutput with the translation. """ # Check if already translated if skip_existing and self.storage.is_verse_translated(reference): existing = self.storage.load_translation(reference) if existing: return existing # Build context and prompt context = self.context_builder.build_context(reference) # Get TM matches tm_matches = self._get_tm_matches(context) # Build prompt prompt = self.prompt_builder.build_prompt(context, tm_matches=tm_matches) system_prompt = self.prompt_builder.get_system_prompt() # Call Claude CLI translation_text = await self._call_claude_async(system_prompt, prompt) # Parse furigana from Claude's response plain_text, annotations = parse_furigana(translation_text) # Create output output = TranslationOutput( reference=reference, text=plain_text, text_with_furigana=translation_text, furigana_annotations=[ { "start_offset": a.start_offset, "end_offset": a.end_offset, "base_text": a.base_text, "reading": a.reading, } for a in annotations ], model=self.model, sources_used=self._get_sources_used(context), glossary_terms_applied=[t.original for t in context.glossary_terms], ) # Save translation self.storage.save_translation(output) # Add to TM self._add_to_tm(context, plain_text) return output def translate_chapter( self, book: str, chapter: int, skip_existing: bool = True, progress_callback: Callable[..., Any] | None = None, evaluate: bool | None = None, ) -> list[TranslationOutput]: """Translate all verses in a chapter. Args: book: Book name. chapter: Chapter number. skip_existing: If True, skip already translated verses. progress_callback: Optional callback for progress updates. evaluate: Override auto_evaluate setting for this call. Returns: List of TranslationOutput for each verse. """ # Get all verses in the chapter verses = self.context_builder.loader.load_chapter_verses(book, chapter) outputs = [] for i, verse in enumerate(verses): if progress_callback: progress_callback(i + 1, len(verses), verse.reference) output = self.translate_verse( verse.reference, skip_existing, evaluate=evaluate ) outputs.append(output) # Rate limiting if i < len(verses) - 1: time.sleep(self.rate_limit_delay) # Save TM after chapter if self._translation_memory is not None: self._translation_memory.save() return outputs async def translate_chapter_async( self, book: str, chapter: int, skip_existing: bool = True, max_concurrent: int = 5, ) -> list[TranslationOutput]: """Translate all verses in a chapter asynchronously. Args: book: Book name. chapter: Chapter number. skip_existing: If True, skip already translated verses. max_concurrent: Maximum concurrent CLI calls. Returns: List of TranslationOutput for each verse. """ # Get all verses in the chapter verses = self.context_builder.loader.load_chapter_verses(book, chapter) # Create semaphore for rate limiting semaphore = asyncio.Semaphore(max_concurrent) async def translate_with_limit(ref: VerseReference) -> TranslationOutput: async with semaphore: result = await self.translate_verse_async(ref, skip_existing) await asyncio.sleep(self.rate_limit_delay) return result tasks = [translate_with_limit(v.reference) for v in verses] outputs = await asyncio.gather(*tasks) # Save TM after chapter if self._translation_memory is not None: self._translation_memory.save() return list(outputs) def _get_sources_used(self, context: TranslationContext) -> list[str]: """Get list of source text types used.""" sources = [] verse = context.target_verse if verse.greek_text: sources.append("greek") if verse.hebrew_text: sources.append("hebrew") if verse.japanese_meiji: sources.append("meiji") if verse.japanese_taisho: sources.append("taisho") if verse.japanese_kougo: sources.append("kougo") if verse.english_kjv: sources.append("kjv") if verse.english_asv: sources.append("asv") return sources def translate_chunk( self, book: str, pericope: Pericope, skip_existing: bool = True, ) -> ChunkTranslationOutput: """Translate an entire pericope/chunk as a cohesive unit. Args: book: Book name. pericope: The pericope to translate. skip_existing: If True, skip verses already translated. Returns: ChunkTranslationOutput with verse-aligned translations. """ # Build chunk context context = self.context_builder.build_chunk_context(book, pericope) # Check if all verses already translated if skip_existing: existing = self._get_existing_chunk_translations(book, pericope) if len(existing) == len(pericope.verses): return ChunkTranslationOutput( pericope_id=pericope.id, book=book, translations=existing, model=self.model, coherence_note="Loaded from existing translations", ) # Get TM matches for the combined source text tm_matches = self._get_chunk_tm_matches(context) # Seam continuity: surface our own translations of the immediately # preceding verses so the new chunk continues them seamlessly. preceding_translations = [] for v in context.preceding_verses: existing_t = self.storage.load_translation(v.reference) if existing_t and existing_t.text: preceding_translations.append((v.reference, existing_t.text)) # Build chunk prompt prompt = self.prompt_builder.build_chunk_prompt( context, tm_matches=tm_matches, preceding_translations=preceding_translations or None, ) system_prompt = self.prompt_builder.get_chunk_system_prompt() # Call Claude CLI response = self._call_claude(system_prompt, prompt) # Parse JSON response translations = self._parse_chunk_response(response, pericope.verses) # Create output output = ChunkTranslationOutput( pericope_id=pericope.id, book=book, translations=translations, model=self.model, ) # Save individual verse translations self._save_chunk_translations(book, translations) # Add to translation memory self._add_chunk_to_tm(context, translations) return output def translate_chunk_by_id( self, pericope_id: str, skip_existing: bool = True, ) -> ChunkTranslationOutput | None: """Translate a pericope by its ID. Args: pericope_id: The pericope identifier. skip_existing: If True, skip verses already translated. Returns: ChunkTranslationOutput or None if pericope not found. """ result = self.context_builder.pericope_manager.get_pericope_by_id(pericope_id) if result is None: return None book, pericope = result return self.translate_chunk(book, pericope, skip_existing) def translate_chapter_by_pericopes( self, book: str, chapter: int, skip_existing: bool = True, progress_callback: Callable[..., Any] | None = None, ) -> list[ChunkTranslationOutput]: """Translate a chapter using pericope-based chunking. Args: book: Book name. chapter: Chapter number. skip_existing: If True, skip already translated pericopes. progress_callback: Optional callback for progress updates. Returns: List of ChunkTranslationOutput for each pericope. """ pericopes = self.context_builder.pericope_manager.get_chapter_pericopes( book, chapter ) if not pericopes: # Fall back to verse-by-verse translation outputs = self.translate_chapter( book, chapter, skip_existing, progress_callback ) # Convert to chunk format return [ ChunkTranslationOutput( pericope_id=f"{book}-{chapter}-fallback", book=book, translations={ f"{o.reference.chapter}:{o.reference.verse}": o.text for o in outputs }, model=self.model, coherence_note="Verse-by-verse fallback (no pericope data)", ) ] outputs = [] for i, pericope in enumerate(pericopes): if progress_callback: progress_callback(i + 1, len(pericopes), pericope) output = self.translate_chunk(book, pericope, skip_existing) outputs.append(output) # Rate limiting between pericopes if i < len(pericopes) - 1: time.sleep(self.rate_limit_delay) # Save TM after chapter if self._translation_memory is not None: self._translation_memory.save() return outputs def _get_existing_chunk_translations( self, book: str, pericope: Pericope ) -> dict[str, str]: """Get existing translations for verses in a pericope. Args: book: Book name. pericope: The pericope to check. Returns: Dict mapping verse ref to translation text. """ existing = {} for verse_ref in pericope.verses: chapter, verse = map(int, verse_ref.split(":")) ref = VerseReference(book=book, chapter=chapter, verse=verse) if self.storage.is_verse_translated(ref): translation = self.storage.load_translation(ref) if translation: existing[verse_ref] = translation.text return existing def _get_chunk_tm_matches( self, context: ChunkTranslationContext ) -> list | None: """Get TM matches for chunk context. Args: context: Chunk translation context. Returns: List of TM matches or None. """ if self._translation_memory is None: return None # Use combined source text for matching combined = context.get_combined_source_text() if not combined: return None # Get matches for the first verse as representative first_verse = context.verses[0] if context.verses else None if not first_verse: # pragma: no cover - combined text implies a verse return None source = first_verse.greek_text or first_verse.hebrew_text or "" if not source: return None matches = self._translation_memory.find_matches( source, threshold=0.7, max_results=3 ) return matches if matches else None def _parse_chunk_response( self, response: str, expected_verses: list[str] ) -> dict[str, str]: """Parse JSON response from chunk translation. Uses robust JSON extraction that handles: - Markdown code blocks (```json ... ```) - Nested JSON structures - Text before/after the JSON - Unicode characters in values Args: response: Raw response from Claude. expected_verses: List of expected verse refs. Returns: Dict mapping verse ref to translation. Raises: ChunkParsingError: If response cannot be parsed. """ original_response = response response = response.strip() # Strategy 1: Remove markdown code blocks if present if "```" in response: response = self._extract_from_code_block(response) # Strategy 2: Try direct JSON parse first (most common case) translations = self._try_parse_json(response) if translations is not None: return self._validate_translations(translations, expected_verses, original_response) # Strategy 3: Find JSON object using bracket matching json_str = self._extract_json_by_brackets(response) if json_str: translations = self._try_parse_json(json_str) if translations is not None: return self._validate_translations(translations, expected_verses, original_response) # Strategy 4: Try to find JSON anywhere in the response for match in re.finditer(r'\{', response): start = match.start() json_str = self._extract_json_from_position(response, start) if json_str: translations = self._try_parse_json(json_str) if translations is not None: return self._validate_translations(translations, expected_verses, original_response) # All strategies failed raise ChunkParsingError( "Failed to parse chunk translation response as JSON", response=original_response, expected_verses=expected_verses, ) def _extract_from_code_block(self, response: str) -> str: """Extract content from markdown code block. Args: response: Response potentially containing code blocks. Returns: Content inside code block, or original if not found. """ lines = response.split("\n") json_lines = [] in_block = False for line in lines: stripped = line.strip() if stripped.startswith("```") and not in_block: in_block = True continue elif stripped.startswith("```") and in_block: break elif in_block: json_lines.append(line) if json_lines: return "\n".join(json_lines) return response def _try_parse_json(self, text: str) -> dict | None: """Try to parse text as JSON dict. Args: text: Text to parse. Returns: Parsed dict or None if parsing failed. """ try: result = json.loads(text) if isinstance(result, dict): return result except json.JSONDecodeError: pass return None def _extract_json_by_brackets(self, text: str) -> str | None: """Extract JSON object by matching brackets. Handles nested objects correctly. Args: text: Text containing JSON. Returns: Extracted JSON string or None if not found. """ start = text.find("{") if start == -1: return None depth = 0 in_string = False escape_next = False for i, char in enumerate(text[start:], start): if escape_next: escape_next = False continue if char == "\\": escape_next = True continue if char == '"' and not escape_next: in_string = not in_string continue if in_string: continue if char == "{": depth += 1 elif char == "}": depth -= 1 if depth == 0: return text[start : i + 1] return None def _extract_json_from_position(self, text: str, start: int) -> str | None: """Extract JSON object starting at given position. Args: text: Full text. start: Starting position of '{'. Returns: Extracted JSON string or None if invalid. """ if start >= len(text) or text[start] != "{": return None depth = 0 in_string = False escape_next = False for i, char in enumerate(text[start:], start): if escape_next: escape_next = False continue if char == "\\": escape_next = True continue if char == '"' and not escape_next: in_string = not in_string continue if in_string: continue if char == "{": depth += 1 elif char == "}": depth -= 1 if depth == 0: return text[start : i + 1] return None def _validate_translations( self, translations: dict, expected_verses: list[str], original_response: str, ) -> dict[str, str]: """Validate parsed translations have all expected verses. Args: translations: Parsed translation dict. expected_verses: List of expected verse refs. original_response: Original response for error context. Returns: Validated translations dict. Raises: ChunkParsingError: If validation fails. """ if not isinstance(translations, dict): raise ChunkParsingError( f"Expected dict, got {type(translations).__name__}", response=original_response, expected_verses=expected_verses, ) # Validate all expected verses are present missing = set(expected_verses) - set(translations.keys()) if missing: raise ChunkParsingError( f"Missing translations for verses: {sorted(missing)}", response=original_response, expected_verses=expected_verses, ) # Keep only the verses this pericope asked for. The prompt shows the # preceding and following verses with their references, and a model # that "helpfully" translates one of them would otherwise overwrite a # verse in the neighbouring pericope; a stray non-verse key ("note") # would crash the save loop after half the chunk was written. extra = set(translations) - set(expected_verses) if extra: logger.warning( "Discarding %d unrequested key(s) from chunk response: %s", len(extra), sorted(map(str, extra))[:5], ) # Validate all values are strings for ref in expected_verses: text = translations[ref] if not isinstance(text, str): raise ChunkParsingError( f"Translation for {ref} is not a string: {type(text).__name__}", response=original_response, expected_verses=expected_verses, ) return {ref: translations[ref] for ref in expected_verses} def _save_chunk_translations( self, book: str, translations: dict[str, str] ) -> None: """Save individual verse translations from chunk output. Args: book: Book name. translations: Dict mapping verse ref to translation. """ for verse_ref, text in translations.items(): chapter, verse = map(int, verse_ref.split(":")) ref = VerseReference(book=book, chapter=chapter, verse=verse) # Parse furigana from translation text plain_text, annotations = parse_furigana(text) output = TranslationOutput( reference=ref, text=plain_text, text_with_furigana=text, furigana_annotations=[ { "start_offset": a.start_offset, "end_offset": a.end_offset, "base_text": a.base_text, "reading": a.reading, } for a in annotations ], model=self.model, sources_used=["chunk_translation"], ) self.storage.save_translation(output) def _add_chunk_to_tm( self, context: ChunkTranslationContext, translations: dict[str, str], ) -> None: """Add chunk translations to translation memory. Args: context: Chunk translation context. translations: Dict mapping verse ref to translation. """ if self._translation_memory is None: return for verse in context.verses: ref = f"{verse.reference.chapter}:{verse.reference.verse}" source = verse.greek_text or verse.hebrew_text target = translations.get(ref) if source and target: self._translation_memory.add_from_translation( source_text=source, target_text=target, reference=verse.reference, metadata={"model": self.model, "pericope_id": context.pericope.id}, )