# #region BatchProcessingService [C:4] [TYPE Module] [SEMANTICS translate, batch, process, classify, cache] # @BRIEF Batch processing for translation: classify rows (same-language/cache/preview/LLM), # call LLM service, persist TranslationRecord/TranslationLanguage rows. # Local language detection (lingua) replaces LLM-based detection. # @LAYER Domain # @RELATION DEPENDS_ON -> [TranslationBatch], [TranslationRecord], [TranslationLanguage] # @RELATION DEPENDS_ON -> [DictionaryManager], [LLMTranslationService], [LanguageDetectService] # @RELATION DEPENDS_ON -> [estimate_token_budget], [ConfigManager] # @PRE DB session is available. Job configuration is valid. # @POST TranslationBatch, TranslationRecord, TranslationLanguage rows created and committed. # @SIDE_EFFECT LLM API calls via LLMTranslationService; DB writes. # @RATIONALE Extracted from TranslationExecutor. Batch insert delegated to _batch_insert.py. # @REJECTED Keeping batch processing inside TranslationExecutor — caused class to exceed INV_7. from datetime import UTC, datetime import time from typing import Any import uuid from sqlalchemy.orm import Session from ...core.config_manager import ConfigManager from ...core.logger import belief_scope, logger from ...models.translate import TranslationBatch, TranslationJob, TranslationLanguage, TranslationRecord from ...services.llm_provider import LLMProviderService from ._batch_insert import insert_batch_to_target from ._lang_detect import batch_detect from ._llm_call import LLMTranslationService from ._token_budget import estimate_token_budget from ._utils import _check_translation_cache, _compute_key_hash, _compute_source_hash from .dictionary import DictionaryManager # #region BatchProcessingService [C:4] [TYPE Class] # @BRIEF Create batch records, classify rows, process LLM calls, persist results. class BatchProcessingService: """Process a batch: classify (cache/preview/LLM), persist, and insert to target.""" def __init__(self, db: Session, config_manager: ConfigManager) -> None: self.db = db self.config_manager = config_manager self._llm_service = LLMTranslationService(db) # #region process_batch [C:3] [TYPE Function] # @BRIEF Process a single batch: create record, classify rows, call LLM, persist. # @PRE job and batch_rows are valid. # @POST TranslationBatch and TranslationRecord rows are created. # @SIDE_EFFECT LLM API call; DB writes. def process_batch( self, job: TranslationJob, run_id: str, batch_index: int, batch_rows: list[dict[str, Any]], dict_snapshot_hash: str | None = None, config_hash: str | None = None, preview_edits_cache: dict[str, dict[str, str]] | None = None, ) -> dict[str, int]: """Process a single batch: classify rows, call LLM (if needed), persist records.""" with belief_scope("BatchProcessingService.process_batch"): batch_start = time.monotonic() batch = self._create_batch(run_id, batch_index, batch_rows) bid = batch.id result = {"successful": 0, "failed": 0, "skipped": 0, "retries": 0} tls = job.target_languages or [job.target_dialect or "en"] tls = [str(tls)] if not isinstance(tls, list) else tls # ★ Run local language detection on all rows (heuristic, no LLM) self._detect_languages(batch_rows, tls) source_texts = [r.get("source_text", "") for r in batch_rows if r.get("source_text")] rc = batch_rows[0].get("source_data") if batch_rows else None dict_matches = DictionaryManager.filter_for_batch(self.db, source_texts, job.id, row_context=rc) self._check_cache(job, batch_rows, dict_snapshot_hash, config_hash) llm_rows, pre_rows = self._classify(batch_rows, preview_edits_cache, tls) result["successful"] += self._persist_pre(pre_rows, bid, run_id, tls) if llm_rows: llm_res = self._process_llm(job, run_id, llm_rows, dict_matches, bid, tls) for k in ("successful", "failed", "skipped", "retries"): result[k] += llm_res.get(k, 0) batch.successful_records = result["successful"] batch.failed_records = result["failed"] batch.completed_at = datetime.now(UTC) batch.status = "COMPLETED" if result["failed"] == 0 else "COMPLETED_WITH_ERRORS" self.db.flush() latency = int((time.monotonic() - batch_start) * 1000) logger.reason(f"Batch {batch_index} complete", {"batch_id": bid, "latency_ms": latency, **result}) return {**result, "batch_id": bid} # #endregion process_batch def _create_batch(self, run_id, batch_index, batch_rows): b = TranslationBatch(id=str(uuid.uuid4()), run_id=run_id, batch_index=batch_index, status="RUNNING", total_records=len(batch_rows), started_at=datetime.now(UTC)) self.db.add(b) self.db.flush() return b def _check_cache(self, job, batch_rows, dict_snapshot_hash, config_hash): for row in batch_rows: if row.get("approved_translation"): continue st = row.get("source_text", "") if not st: continue ctx = list(job.context_columns or []) h = _compute_source_hash(st, row.get("source_data"), dict_snapshot_hash, config_hash, ctx) row["_source_hash"] = h cached = _check_translation_cache(self.db, h) if cached: row["_cached_lang_values"] = cached logger.reason("Translation cache hit", {"source_hash": h[:12], "langs": list(cached.keys())}) # ★ Local language detection — replaces LLM-based detection def _detect_languages(self, batch_rows: list[dict], target_languages: list[str]) -> None: """Run local language detection on all batch rows (no LLM). Attaches '_detected_lang' (BCP-47 code or 'und') to each row dict. Uses batch_detect() for efficient multi-text processing. """ texts = [row.get("source_text", "") for row in batch_rows] results = batch_detect(texts, target_languages) for row, lang in zip(batch_rows, results): row["_detected_lang"] = lang def _classify(self, batch_rows, preview_edits_cache, tls): llm_rows, pre_rows = [], [] tls_lower = [str(t).lower() for t in tls] for row in batch_rows: # ★ Same-language pre-filter: only short-circuit when ALL targets # match the detected language. If only SOME match, still process # other targets via cache or LLM. dl = row.get("_detected_lang") if dl and str(dl).lower() not in ("und", "") and str(dl).lower() in tls_lower: non_matching = [t for t in tls if str(t).lower() != str(dl).lower()] if not non_matching: # All targets are the same as detected language — no translation needed row["_same_language"] = True pre_rows.append(row) continue # Partial match: mark same-language for later use, but don't skip row["_same_language"] = True if row.get("approved_translation"): pre_rows.append(row) continue cl = row.get("_cached_lang_values") if cl and all(lc in cl for lc in tls): pre_rows.append(row) continue if preview_edits_cache: sd = row.get("source_data") or {} if sd: kh = _compute_key_hash(sd) pe = preview_edits_cache.get(kh) if pe: fe = next(iter(pe.values()), None) if fe: row["approved_translation"] = fe pre_rows.append(row) continue llm_rows.append(row) return llm_rows, pre_rows def _persist_pre(self, pre_rows, bid, run_id, tls): count = 0 for row in pre_rows: cl = row.get("_cached_lang_values") detected_lang = row.get("_detected_lang", "und") or "und" source_text = row.get("source_text", "") is_same = row.get("_same_language") # target_sql: prefer approved_translation, then first cached value, then source primary_cached = next(iter(cl.values()), "") if cl else "" target_sql = row.get("approved_translation") or primary_cached or source_text rec = TranslationRecord( id=str(uuid.uuid4()), batch_id=bid, run_id=run_id, source_sql=source_text, target_sql=target_sql, source_object_type="table_row", source_object_id=row.get("row_index"), source_object_name=row.get("source_object_name", ""), source_data=row.get("source_data"), source_hash=row.get("_source_hash"), status="SUCCESS", ) self.db.add(rec) for lc in tls: if is_same and str(lc).lower() == str(detected_lang).lower(): # Same language: use source text as-is (no translation needed) fv = source_text elif cl and lc in cl: fv = cl[lc] else: fv = row.get("approved_translation", "") self.db.add(TranslationLanguage( id=str(uuid.uuid4()), record_id=rec.id, language_code=lc, source_language_detected=detected_lang, translated_value=fv, final_value=fv, status="translated", needs_review=False, )) count += 1 return count def _process_llm(self, job, run_id, rows_for_llm, dict_matches, bid, tls): provider_model = None if job.provider_id: try: p = LLMProviderService(self.db).get_provider(job.provider_id) if p: provider_model = p.default_model or "gpt-4o-mini" except Exception: provider_model = None tb = estimate_token_budget( source_rows=rows_for_llm, target_languages=tls, source_column="source_text", context_columns=None, dictionary_entries=dict_matches, batch_size=len(rows_for_llm), provider_info=provider_model, ) if tb["warning"]: logger.explore("Token budget warning", {"batch_id": bid, "warning": tb["warning"]}) return self._llm_service.call_llm_for_batch( job=job, run_id=run_id, batch_rows=rows_for_llm, dict_matches=dict_matches, batch_id=bid, max_tokens=tb["max_output_needed"], ) # -- Batch insert (delegation) -- def insert_batch_to_target(self, job: TranslationJob, batch_id: str, run_id: str) -> None: insert_batch_to_target(self.db, self.config_manager, job, batch_id, run_id) # #endregion BatchProcessingService # #endregion BatchProcessingService