# #region TranslationOrchestrator [C:5] [TYPE Module] [SEMANTICS sqlalchemy, tenacity, translate, orchestration, batch] # @BRIEF Run lifecycle coordination: validate preconditions, dispatch executor, generate SQL, submit to Superset, record events. # @LAYER: Domain # @RELATION DEPENDS_ON -> [TranslationRun] # @RELATION DEPENDS_ON -> [TranslationJob] # @RELATION DEPENDS_ON -> [TranslationExecutor] # @RELATION DEPENDS_ON -> [SQLGenerator] # @RELATION DEPENDS_ON -> [SupersetSqlLabExecutor] # @RELATION DEPENDS_ON -> [TranslationEventLog] # @RELATION DEPENDS_ON -> [TranslationPreviewSession] # @PRE: Valid job and accepted preview (for manual runs). Superset and LLM are reachable. # @POST: Translation run is executed, SQL generated and submitted, events recorded. # @SIDE_EFFECT: Creates TranslationRun; creates batches and records; generates SQL; submits to Superset; records events. # @DATA_CONTRACT: Input[db, config_manager, current_user] -> Output[TranslationRun] # @INVARIANT: State transitions: PENDING -> RUNNING -> COMPLETED|FAILED|CANCELLED. Exactly one terminal event per run. # @RATIONALE: C5 orchestrator because preview, execution, event logging, and retry share state within single transaction boundary. # @REJECTED: Distributed actor model (Celery) — eventual-consistency challenges at current scale. # @REJECTED: stdout-only logging — lacks structured payload integrity; cannot enforce terminal-event invariant. import json import uuid from collections.abc import Callable from datetime import UTC, datetime from typing import Any from sqlalchemy import text from sqlalchemy.orm import Session, joinedload, selectinload from ...core.config_manager import ConfigManager from ...core.logger import belief_scope, logger from ...models.translate import ( TranslationBatch, TranslationJob, TranslationLanguage, TranslationPreviewSession, TranslationRecord, TranslationRun, TranslationRunLanguageStats, ) from .events import TranslationEventLog from .executor import TranslationExecutor from .sql_generator import SQLGenerator, _normalize_timestamp_value from .superset_executor import SupersetSqlLabExecutor # #region TranslationOrchestrator [C:5] [TYPE Class] # @BRIEF Coordinates full translation run lifecycle: validation, execution, SQL generation, Superset submission, event logging. # @PRE: DB session and config manager are available. # @POST: Runs are created, executed, and finalized with event records. # @SIDE_EFFECT: Creates/modifies TranslationRun, TranslationBatch, TranslationRecord, TranslationEvent rows. # @INVARIANT: State transitions: PENDING -> RUNNING -> COMPLETED|FAILED|CANCELLED. class TranslationOrchestrator: def __init__( self, db: Session, config_manager: ConfigManager, current_user: str | None = None, ): self.db = db self.config_manager = config_manager self.current_user = current_user self.event_log = TranslationEventLog(db) self._job: TranslationJob | None = None # region start_run [TYPE Function] # @PURPOSE: Start a new translation run for a job. # @PRE: job_id exists. For manual runs, there must be an accepted preview session. # @POST: TranslationRun is created in PENDING status with hash fields and config snapshot. Events are recorded. # @SIDE_EFFECT: DB writes. def start_run( self, job_id: str, is_scheduled: bool = False, trigger_type: str | None = None, full_translation: bool = False, ) -> TranslationRun: with belief_scope("TranslationOrchestrator.start_run"): # Load and validate job job = self.db.query(TranslationJob).filter(TranslationJob.id == job_id).first() if not job: raise ValueError(f"Translation job '{job_id}' not found") self._job = job logger.reason("Starting translation run", { "job_id": job_id, "is_scheduled": is_scheduled, }) # Validate preconditions self._validate_preconditions(job, is_scheduled=is_scheduled) # Compute hashes config_hash = self._compute_config_hash(job) dict_snapshot_hash = self._compute_dict_snapshot_hash(job_id) # Build config snapshot config_snapshot = { "source_dialect": job.source_dialect, "target_dialect": job.target_dialect, "database_dialect": job.database_dialect, "source_datasource_id": job.source_datasource_id, "source_table": job.source_table, "target_schema": job.target_schema, "target_table": job.target_table, "source_key_cols": job.source_key_cols, "target_key_cols": job.target_key_cols, "translation_column": job.translation_column, "target_column": job.target_column, "context_columns": job.context_columns, "provider_id": job.provider_id, "batch_size": job.batch_size, "upsert_strategy": job.upsert_strategy, "dictionary_ids": self._compute_dict_snapshot_hash(job_id), "full_translation": full_translation, } # Compute key_hash from source_key_cols import hashlib key_hash_input = json.dumps({ "source_key_cols": job.source_key_cols, "source_datasource_id": job.source_datasource_id, "source_table": job.source_table, }, sort_keys=True) key_hash = hashlib.sha256(key_hash_input.encode()).hexdigest()[:16] # Create run record with all hash/snapshot fields run = TranslationRun( id=str(uuid.uuid4()), job_id=job_id, status="PENDING", trigger_type=trigger_type or ("scheduled" if is_scheduled else "manual"), config_snapshot=config_snapshot, key_hash=key_hash, config_hash=config_hash, dict_snapshot_hash=dict_snapshot_hash, created_by=self.current_user, created_at=datetime.now(UTC), ) self.db.add(run) self.db.flush() # Record event self.event_log.log_event( job_id=job_id, run_id=run.id, event_type="RUN_STARTED", payload={ "is_scheduled": is_scheduled, "trigger_type": run.trigger_type, "config_hash": config_hash, "dict_snapshot_hash": dict_snapshot_hash, "key_hash": key_hash, }, created_by=self.current_user, ) self.db.commit() self.db.refresh(run) logger.reflect("Run created", { "run_id": run.id, "job_id": job_id, "status": run.status, "trigger_type": run.trigger_type, }) return run # endregion start_run # region execute_run [TYPE Function] # @PURPOSE: Execute a translation run: dispatch executor, generate SQL, submit to Superset. # @PRE: run is in PENDING status. # @POST: Run is executed, SQL generated, Superset submission attempted. # @SIDE_EFFECT: LLM calls, DB writes, Superset API calls. def execute_run( self, run: TranslationRun, on_batch_progress: Callable[[str, int, int, int, int], None] | None = None, skip_insert: bool = False, ) -> TranslationRun: with belief_scope("TranslationOrchestrator.execute_run"): job = self.db.query(TranslationJob).filter(TranslationJob.id == run.job_id).first() if not job: raise ValueError(f"Job '{run.job_id}' not found") self._job = job if run.status != "PENDING": raise ValueError( f"Cannot execute run in status '{run.status}'. " f"Run must be in PENDING status." ) logger.reason("Executing run", { "run_id": run.id, "job_id": job.id, "skip_insert": skip_insert, }) # Record translation phase start self.event_log.log_event( job_id=job.id, run_id=run.id, event_type="TRANSLATION_PHASE_STARTED", payload={}, created_by=self.current_user, ) # Initialize per-language stats target_languages = job.target_languages or [job.target_dialect or "en"] if not isinstance(target_languages, list): target_languages = [str(target_languages)] language_stats_map: dict[str, TranslationRunLanguageStats] = {} for lang_code in target_languages: lang_stat = TranslationRunLanguageStats( id=str(uuid.uuid4()), run_id=run.id, language_code=lang_code, total_rows=0, translated_rows=0, failed_rows=0, skipped_rows=0, token_count=0, estimated_cost=0.0, ) self.db.add(lang_stat) language_stats_map[lang_code] = lang_stat self.db.flush() # Dispatch executor executor = TranslationExecutor( self.db, self.config_manager, self.current_user, on_batch_progress=on_batch_progress, ) try: run = executor.execute_run(run, llm_progress_callback=None, language_stats_map=language_stats_map) except Exception as e: logger.explore("Translation execution failed", { "run_id": run.id, "error": str(e), }) # Rollback the session if it's in a broken state (e.g. after a # failed flush from a previous exception in executor.execute_run). # Without this, subsequent flush()/commit() calls raise # "This Session's transaction has been rolled back". try: self.db.rollback() except Exception: pass run = self.db.merge(run) run.status = "FAILED" run.error_message = f"Translation execution failed: {e}" run.completed_at = datetime.now(UTC) self.db.flush() self.event_log.log_event( job_id=job.id, run_id=run.id, event_type="RUN_FAILED", payload={"error": str(e), "phase": "translation"}, created_by=self.current_user, ) self.db.commit() return run # Check if executor cancelled itself due to cancellation flag if run.status == "CANCELLED": self._update_language_stats(run.id, language_stats_map) self.event_log.log_event( job_id=job.id if job else (self._job.id if self._job else run.job_id), run_id=run.id, event_type="RUN_CANCELLED", payload={"reason": "cancellation_flag"}, created_by=self.current_user, ) self.db.commit() logger.reflect("Run cancelled via cancellation flag", { "run_id": run.id, }) return run # Aggregate per-language statistics after executor completes self._update_language_stats(run.id, language_stats_map) # Record translation phase complete self.event_log.log_event( job_id=job.id, run_id=run.id, event_type="TRANSLATION_PHASE_COMPLETED", payload={ "total": run.total_records, "successful": run.successful_records, "failed": run.failed_records, "skipped": run.skipped_records, }, created_by=self.current_user, ) # Skip insert phase if requested (e.g., for preview-only execution) if skip_insert: run.status = "COMPLETED" run.completed_at = datetime.now(UTC) self.db.flush() self.event_log.log_event( job_id=job.id, run_id=run.id, event_type="RUN_COMPLETED", payload={"skip_insert": True}, created_by=self.current_user, ) self.db.commit() return run # Generate SQL and submit to Superset insert_result = self._generate_and_insert_sql(job, run) run.insert_status = insert_result.get("status") run.superset_execution_id = str(insert_result.get("query_id") or "") run.superset_execution_log = insert_result # Preserve the translation-phase status. If the executor already # marked the run FAILED (all LLM calls failed) we do NOT upgrade it # to COMPLETED just because the insert phase ran. if run.status != "FAILED": run.status = "COMPLETED" if insert_result.get("error_message"): run.error_message = insert_result["error_message"] run.completed_at = datetime.now(UTC) self.db.flush() # Record terminal event terminal_event = "RUN_COMPLETED" self.event_log.log_event( job_id=job.id, run_id=run.id, event_type=terminal_event, payload={ "insert_status": insert_result.get("status"), "query_id": insert_result.get("query_id"), "rows_affected": insert_result.get("rows_affected"), "total_records": run.total_records, "successful": run.successful_records, "failed": run.failed_records, "skipped": run.skipped_records, }, created_by=self.current_user, ) self.db.commit() # Re-query run after commit — refresh may fail if the object was # created in a different session or became detached during commit. run = self.db.query(TranslationRun).filter(TranslationRun.id == run.id).first() logger.reflect("Run execution complete", { "run_id": run.id, "status": run.status, "insert_status": run.insert_status, }) return run # endregion execute_run # region _generate_and_insert_sql [TYPE Function] # @PURPOSE: Generate INSERT SQL from successful records and submit to Superset SQL Lab. # @PRE: job has target table configured. run has successful records. # @POST: SQL is generated and submitted. Returns execution result. # @SIDE_EFFECT: Superset API call. def _generate_and_insert_sql( self, job: TranslationJob, run: TranslationRun, ) -> dict[str, Any]: with belief_scope("TranslationOrchestrator._generate_and_insert_sql"): # Fetch successful records with eager-loaded language data records = ( self.db.query(TranslationRecord) .options(joinedload(TranslationRecord.languages)) .filter( TranslationRecord.run_id == run.id, TranslationRecord.status == "SUCCESS", TranslationRecord.target_sql.isnot(None), ) .all() ) if not records: logger.reason("No successful records to insert", {"run_id": run.id}) return {"status": "skipped", "reason": "no_records", "query_id": None} logger.reason(f"Generating SQL for {len(records)} records", { "run_id": run.id, "dialect": job.database_dialect or job.target_dialect, }) effective_target = job.target_column or job.translation_column primary_language = (job.target_languages or ["en"])[0] # Columns that exist in the target ClickHouse table columns = [] if job.target_key_cols: columns.extend(job.target_key_cols) if effective_target: columns.append(effective_target) if job.target_language_column: columns.append(job.target_language_column) if job.target_source_column: columns.append(job.target_source_column) if job.target_source_language_column: columns.append(job.target_source_language_column) columns.append("context") columns.append("is_original") # Deduplicate while preserving order seen: set[str] = set() deduped: list[str] = [] for c in columns: if c and c not in seen: deduped.append(c) seen.add(c) columns = deduped # Keys for the context JSON: context_columns + original translation_column context_keys = list(job.context_columns or []) if job.translation_column and job.translation_column != effective_target and job.translation_column not in context_keys: context_keys.append(job.translation_column) rows_for_sql: list[dict[str, object]] = [] for rec in records: source_data = rec.source_data or {} # Detect source language from first TranslationLanguage entry detected_src_lang = "und" if rec.languages and len(rec.languages) > 0: detected_src_lang = rec.languages[0].source_language_detected or "und" # Build context JSON: all extra columns the user configured context_data: dict[str, str] = {} for key in context_keys: val = source_data.get(key) context_data[key] = str(val) if val is not None else "" # ── Shared base row ── base_row: dict[str, object] = {} if job.target_key_cols: for k in job.target_key_cols: raw = source_data.get(k) if raw is not None: normalized = _normalize_timestamp_value(raw) base_row[k] = normalized if normalized else raw else: base_row[k] = None if job.target_source_column: base_row[job.target_source_column] = rec.source_sql or "" if job.target_source_language_column: base_row[job.target_source_language_column] = detected_src_lang base_row["context"] = json.dumps(context_data, ensure_ascii=False) # ── 1. ORIGINAL row (is_original = 1) ── original_row = dict(base_row) if effective_target: original_row[effective_target] = rec.source_sql or "" if job.target_language_column: original_row[job.target_language_column] = detected_src_lang original_row["is_original"] = 1 rows_for_sql.append(original_row) # ── 2. TRANSLATION rows (is_original = 0) ── # Skip language that matches the source — the original row already covers it if rec.languages and len(rec.languages) > 0: for lang in rec.languages: if lang.language_code == detected_src_lang: continue trans_row = dict(base_row) trans_value = lang.final_value or lang.translated_value or "" if effective_target: trans_row[effective_target] = trans_value if job.target_language_column: trans_row[job.target_language_column] = lang.language_code trans_row["is_original"] = 0 rows_for_sql.append(trans_row) else: # Fallback: no per-language data fallback_row = dict(base_row) if effective_target: fallback_row[effective_target] = rec.target_sql or "" if job.target_language_column: fallback_row[job.target_language_column] = primary_language fallback_row["is_original"] = 0 rows_for_sql.append(fallback_row) if not columns: columns = [effective_target or "translated_text"] rows_for_sql = [{columns[0]: rec.target_sql or ""} for rec in records] # Resolve the real database backend engine from Superset try: env_id = job.environment_id or job.source_dialect or "" executor = SupersetSqlLabExecutor(self.config_manager, env_id) executor.resolve_database_id( target_database_id=job.target_database_id, ) real_backend = executor.get_database_backend() except Exception as e: logger.explore("Failed to resolve database backend, falling back to job dialet", { "error": str(e), }) real_backend = None dialect = real_backend or job.database_dialect or job.target_dialect or "postgresql" # Generate SQL try: sql, row_count = SQLGenerator.generate( dialect=dialect, target_schema=job.target_schema, target_table=job.target_table or "translated_data", columns=columns, rows=rows_for_sql, key_columns=job.target_key_cols, upsert_strategy=job.upsert_strategy or "MERGE", ) except ValueError as e: logger.explore("SQL generation failed", {"error": str(e)}) return {"status": "failed", "error_message": str(e), "query_id": None} logger.reason("SQL generated with dialect", { "dialect": dialect, "real_backend": real_backend, "job_database_dialect": job.database_dialect, "job_target_dialect": job.target_dialect, }) # Log insert phase start self.event_log.log_event( job_id=job.id, run_id=run.id, event_type="INSERT_PHASE_STARTED", payload={"sql_length": len(sql), "row_count": row_count, "dialect": dialect}, created_by=self.current_user, ) # Submit to Superset try: result = executor.execute_and_poll( sql=sql, max_polls=30, poll_interval_seconds=2.0, ) except Exception as e: logger.explore("Superset SQL submission failed", { "run_id": run.id, "error": str(e), }) result = {"status": "failed", "error_message": str(e), "query_id": None} # Log insert phase complete self.event_log.log_event( job_id=job.id, run_id=run.id, event_type="INSERT_PHASE_COMPLETED", payload=result, created_by=self.current_user, ) return result # endregion _generate_and_insert_sql # region _validate_preconditions [TYPE Function] # @PURPOSE: Validate preconditions before starting a run. # @PRE: None. # @POST: Raises ValueError if preconditions are not met. # @SIDE_EFFECT: None. def _validate_preconditions( self, job: TranslationJob, is_scheduled: bool = False, ) -> None: with belief_scope("TranslationOrchestrator._validate_preconditions"): # Job must be in valid status if job.status in ("DRAFT",): raise ValueError( f"Cannot run job '{job.id}' in status '{job.status}'. " f"Job must be READY, ACTIVE, or COMPLETED." ) # Must have target table configured if not job.target_table: raise ValueError( f"Job '{job.id}' has no target table configured. " "Configure a target table before running." ) # Must have LLM provider configured if not job.provider_id: raise ValueError( f"Job '{job.id}' has no LLM provider configured. " "Select an LLM provider before running." ) # Must have a translation column if not job.translation_column: raise ValueError( f"Job '{job.id}' has no translation column configured. " "Select a translation column before running." ) # For manual runs, must have accepted preview if not is_scheduled: accepted_session = ( self.db.query(TranslationPreviewSession) .filter( TranslationPreviewSession.job_id == job.id, TranslationPreviewSession.status == "APPLIED", ) .order_by(TranslationPreviewSession.created_at.desc()) .first() ) if not accepted_session: raise ValueError( f"Job '{job.id}' has no accepted preview session. " "Run and accept a preview before executing a manual translation run." ) logger.reason("Preconditions validated", { "job_id": job.id, "is_scheduled": is_scheduled, }) # endregion _validate_preconditions # region retry_failed_batches [TYPE Function] # @PURPOSE: Retry failed batches in a run. # @PRE: run exists and has failed batches. # @POST: Failed batches are re-processed. # @SIDE_EFFECT: LLM calls, DB writes. def retry_failed_batches( self, run_id: str, ) -> TranslationRun: with belief_scope("TranslationOrchestrator.retry_failed_batches"): run = self.db.query(TranslationRun).filter(TranslationRun.id == run_id).first() if not run: raise ValueError(f"Run '{run_id}' not found") job = self.db.query(TranslationJob).filter(TranslationJob.id == run.job_id).first() if not job: raise ValueError(f"Job '{run.job_id}' not found") self._job = job # Find failed batches failed_batches = ( self.db.query(TranslationBatch) .filter( TranslationBatch.run_id == run_id, TranslationBatch.status.in_(["FAILED", "COMPLETED_WITH_ERRORS"]), ) .all() ) if not failed_batches: raise ValueError(f"No failed batches found for run '{run_id}'") logger.reason("Retrying failed batches", { "run_id": run_id, "batch_count": len(failed_batches), }) self.event_log.log_event( job_id=job.id, run_id=run.id, event_type="RUN_RETRYING", payload={"batch_count": len(failed_batches)}, created_by=self.current_user, ) # Re-process each failed batch executor = TranslationExecutor(self.db, self.config_manager, self.current_user) run.status = "RUNNING" self.db.flush() for batch in failed_batches: # Fetch failed records for this batch failed_records = ( self.db.query(TranslationRecord) .filter( TranslationRecord.batch_id == batch.id, TranslationRecord.status == "FAILED", ) .all() ) if not failed_records: continue # Build rows from failed records retry_rows = [] for rec in failed_records: retry_rows.append({ "row_index": rec.source_object_id or "0", "source_text": rec.source_sql or "", "approved_translation": None, "source_object_name": rec.source_object_name or "", }) # Process retry batch result = executor._process_batch( job=job, run_id=run_id, batch_index=batch.batch_index, batch_rows=retry_rows, ) # Update run stats run.successful_records = (run.successful_records or 0) + result["successful"] run.failed_records = (run.failed_records or 0) + result["failed"] run.skipped_records = (run.skipped_records or 0) + result["skipped"] self.db.flush() # Update run status if run.failed_records == 0: run.status = "COMPLETED" elif run.successful_records == 0: run.status = "FAILED" else: run.status = "COMPLETED" run.completed_at = datetime.now(UTC) self.db.flush() self.event_log.log_event( job_id=job.id, run_id=run.id, event_type="RUN_COMPLETED", payload={"retry": True}, created_by=self.current_user, ) self.db.commit() logger.reflect("Retry complete", { "run_id": run_id, "status": run.status, }) return run # endregion retry_failed_batches # region retry_insert [TYPE Function] # @PURPOSE: Retry the SQL insert phase for a completed run. # @PRE: run exists and has successful records. # @POST: SQL is regenerated and re-submitted to Superset. # @SIDE_EFFECT: Superset API call. def retry_insert( self, run_id: str, ) -> TranslationRun: with belief_scope("TranslationOrchestrator.retry_insert"): run = self.db.query(TranslationRun).filter(TranslationRun.id == run_id).first() if not run: raise ValueError(f"Run '{run_id}' not found") job = self.db.query(TranslationJob).filter(TranslationJob.id == run.job_id).first() if not job: raise ValueError(f"Job '{run.job_id}' not found") self._job = job logger.reason("Retrying insert phase", { "run_id": run_id, }) self.event_log.log_event( job_id=job.id, run_id=run.id, event_type="RUN_RETRY_INSERT", payload={}, created_by=self.current_user, ) # Regenerate SQL and submit insert_result = self._generate_and_insert_sql(job, run) # Update run run.insert_status = insert_result.get("status") run.superset_execution_id = str(insert_result.get("query_id") or "") if insert_result.get("error_message"): run.error_message = insert_result["error_message"] self.db.flush() self.db.commit() self.db.refresh(run) logger.reflect("Insert retry complete", { "run_id": run_id, "insert_status": run.insert_status, }) return run # endregion retry_insert # region cancel_run [TYPE Function] # @PURPOSE: Cancel a running translation. # @PRE: run is in PENDING or RUNNING status. # @POST: Run status is set to CANCELLED. # @SIDE_EFFECT: DB write; records event. def cancel_run(self, run_id: str) -> TranslationRun: with belief_scope("TranslationOrchestrator.cancel_run"): # Set short lock timeout to avoid blocking on row lock held by # background executor thread (which holds RowExclusiveLock during # batch processing). If the row is locked, we fall back to setting # a cancellation flag via direct SQL UPDATE, which the executor # checks after each batch commit. try: self.db.execute(text("SET LOCAL lock_timeout = '3s'")) run = self.db.query(TranslationRun).filter(TranslationRun.id == run_id).first() except Exception: # Row is locked — set cancellation flag via direct SQL logger.explore("Row lock timeout — setting cancellation flag via direct SQL", { "run_id": run_id, }) self.db.execute( text("UPDATE translation_runs SET error_message = 'CANCEL_REQUESTED' WHERE id = :run_id"), {"run_id": run_id}, ) self.db.commit() run = self.db.query(TranslationRun).filter(TranslationRun.id == run_id).first() if not run: raise ValueError(f"Run '{run_id}' not found") logger.reflect("Cancellation flag set for locked run", { "run_id": run_id, }) return run if not run: raise ValueError(f"Run '{run_id}' not found") if run.status not in ("PENDING", "RUNNING"): raise ValueError( f"Cannot cancel run in status '{run.status}'. " f"Only PENDING or RUNNING runs can be cancelled." ) run.status = "CANCELLED" run.completed_at = datetime.now(UTC) self.db.flush() self.event_log.log_event( job_id=run.job_id, run_id=run.id, event_type="RUN_CANCELLED", payload={}, created_by=self.current_user, ) self.db.commit() self.db.refresh(run) logger.reflect("Run cancelled", { "run_id": run_id, }) return run # endregion cancel_run # region get_run_status [TYPE Function] # @PURPOSE: Get run status with statistics. # @PRE: run_id exists. # @POST: Returns dict with run details. def get_run_status(self, run_id: str) -> dict[str, Any]: with belief_scope("TranslationOrchestrator.get_run_status"): run = self.db.query(TranslationRun).filter(TranslationRun.id == run_id).first() if not run: raise ValueError(f"Run '{run_id}' not found") # Count batches batch_count = ( self.db.query(TranslationBatch) .filter(TranslationBatch.run_id == run_id) .count() ) # Get event summary event_summary = self.event_log.get_run_event_summary(run_id) # Get language stats language_stats_entries = ( self.db.query(TranslationRunLanguageStats) .filter(TranslationRunLanguageStats.run_id == run_id) .all() ) language_stats = [ { "language_code": ls.language_code, "total_rows": ls.total_rows or 0, "translated_rows": ls.translated_rows or 0, "failed_rows": ls.failed_rows or 0, "skipped_rows": ls.skipped_rows or 0, "token_count": ls.token_count or 0, "estimated_cost": ls.estimated_cost or 0.0, } for ls in language_stats_entries ] return { "id": run.id, "job_id": run.job_id, "status": run.status, "trigger_type": run.trigger_type, "started_at": run.started_at.isoformat() if run.started_at else None, "completed_at": run.completed_at.isoformat() if run.completed_at else None, "error_message": run.error_message, "total_records": run.total_records or 0, "successful_records": run.successful_records or 0, "failed_records": run.failed_records or 0, "skipped_records": run.skipped_records or 0, "insert_status": run.insert_status, "superset_execution_id": run.superset_execution_id, "batch_count": batch_count, "language_stats": language_stats, "event_invariants": { "has_run_started": event_summary["has_run_started"], "terminal_event_count": event_summary["terminal_event_count"], "invariant_valid": event_summary["invariant_valid"], }, "created_by": run.created_by, "created_at": run.created_at.isoformat() if run.created_at else None, } # endregion get_run_status # region get_run_records [TYPE Function] # @PURPOSE: Get paginated records for a run. # @PRE: run_id exists. # @POST: Returns dict with records and pagination info. def get_run_records( self, run_id: str, page: int = 1, page_size: int = 50, status_filter: str | None = None, ) -> dict[str, Any]: with belief_scope("TranslationOrchestrator.get_run_records"): query = ( self.db.query(TranslationRecord) .options(selectinload(TranslationRecord.languages)) .filter(TranslationRecord.run_id == run_id) ) if status_filter: query = query.filter(TranslationRecord.status == status_filter) total = query.count() offset = (page - 1) * page_size records = ( query.order_by(TranslationRecord.created_at.desc()) .offset(offset) .limit(page_size) .all() ) return { "items": [ { "id": r.id, "batch_id": r.batch_id, "source_sql": r.source_sql, "target_sql": r.target_sql, "source_object_type": r.source_object_type, "source_object_id": r.source_object_id, "source_object_name": r.source_object_name, "status": r.status, "error_message": r.error_message, "created_at": r.created_at.isoformat() if r.created_at else None, "languages": [ { "language_code": tl.language_code, "translated_value": tl.translated_value, "final_value": tl.final_value or tl.translated_value, "source_language_detected": tl.source_language_detected, "status": tl.status, "needs_review": tl.needs_review or False, } for tl in (r.languages or []) ], } for r in records ], "total": total, "page": page, "page_size": page_size, "status_filter": status_filter, } # endregion get_run_records # region get_run_history [TYPE Function] # @PURPOSE: Get run history for a job. # @PRE: job_id exists. # @POST: Returns list of runs. def get_run_history( self, job_id: str, page: int = 1, page_size: int = 20, ) -> tuple[int, list[dict[str, Any]]]: with belief_scope("TranslationOrchestrator.get_run_history"): query = self.db.query(TranslationRun).filter( TranslationRun.job_id == job_id ) total = query.count() runs = ( query.order_by(TranslationRun.created_at.desc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) return total, [ { "id": r.id, "job_id": r.job_id, "status": r.status, "started_at": r.started_at.isoformat() if r.started_at else None, "completed_at": r.completed_at.isoformat() if r.completed_at else None, "error_message": r.error_message, "total_records": r.total_records or 0, "successful_records": r.successful_records or 0, "failed_records": r.failed_records or 0, "skipped_records": r.skipped_records or 0, "insert_status": r.insert_status, "superset_execution_id": r.superset_execution_id, "created_by": r.created_by, "created_at": r.created_at.isoformat() if r.created_at else None, } for r in runs ] # endregion get_run_history # region _update_language_stats [TYPE Function] # @PURPOSE: Aggregate TranslationLanguage entries by language_code and update TranslationRunLanguageStats. # @PRE: run_id and language_stats_map are valid. DB session is available. # @POST: Language stats are updated with row counts and estimated tokens/cost. # @SIDE_EFFECT: DB writes on language_stats objects. def _update_language_stats( self, run_id: str, language_stats_map: dict[str, TranslationRunLanguageStats], ) -> None: with belief_scope("TranslationOrchestrator._update_language_stats"): # Get all records for this run to join with TranslationLanguage records = ( self.db.query(TranslationRecord) .filter(TranslationRecord.run_id == run_id) .all() ) record_ids = [r.id for r in records] if not record_ids: logger.reason("No records for language stats aggregation", {"run_id": run_id}) return # Get all language entries for this run's records lang_entries = ( self.db.query(TranslationLanguage) .filter(TranslationLanguage.record_id.in_(record_ids)) .all() ) # Aggregate by language_code from collections import defaultdict agg: dict[str, dict[str, int]] = defaultdict(lambda: {"total": 0, "translated": 0, "failed": 0, "skipped": 0}) for le in lang_entries: code = le.language_code agg[code]["total"] += 1 if le.status in ("translated", "approved", "edited"): agg[code]["translated"] += 1 elif le.status == "failed": agg[code]["failed"] += 1 elif le.status == "skipped": agg[code]["skipped"] += 1 # Estimate tokens: heuristic based on character count of translated values total_chars = sum( len(le.translated_value or "") for le in lang_entries if le.translated_value ) total_tokens = max(1, total_chars // 4) # ~4 chars per token cost_per_token = 0.002 / 1000 # $0.002 per 1K tokens # Update each language stat entry for lang_code, lang_stat in language_stats_map.items(): data = agg.get(lang_code, {"total": 0, "translated": 0, "failed": 0, "skipped": 0}) lang_stat.total_rows = data["total"] lang_stat.translated_rows = data["translated"] lang_stat.failed_rows = data["failed"] lang_stat.skipped_rows = data["skipped"] # Proportional token split: share tokens across languages num_langs = len(language_stats_map) if num_langs > 0: lang_stat.token_count = total_tokens // num_langs lang_stat.estimated_cost = round((lang_stat.token_count / 1000) * cost_per_token, 6) self.db.flush() logger.reason("Language stats updated", { "run_id": run_id, "languages": list(language_stats_map.keys()), "total_tokens_est": total_tokens, }) # endregion _update_language_stats # region _compute_config_hash [TYPE Function] # @PURPOSE: Compute a hash of the job's current configuration for snapshot comparison. @staticmethod def _compute_config_hash(job: TranslationJob) -> str: import hashlib config_str = json.dumps({ "source_dialect": job.source_dialect, "target_dialect": job.target_dialect, "source_datasource_id": job.source_datasource_id, "translation_column": job.translation_column, "context_columns": job.context_columns, "target_languages": sorted(job.target_languages) if job.target_languages else [], "provider_id": job.provider_id, "batch_size": job.batch_size, "upsert_strategy": job.upsert_strategy, }, sort_keys=True) return hashlib.sha256(config_str.encode()).hexdigest()[:16] # endregion _compute_config_hash # region _compute_dict_snapshot_hash [TYPE Function] # @PURPOSE: Compute a hash of dictionary state for snapshot comparison. def _compute_dict_snapshot_hash(self, job_id: str) -> str: import hashlib from ...models.translate import TranslationJobDictionary dict_links = ( self.db.query(TranslationJobDictionary) .filter(TranslationJobDictionary.job_id == job_id) .all() ) dict_ids = sorted([dl.dictionary_id for dl in dict_links]) hash_input = ",".join(dict_ids) return hashlib.sha256(hash_input.encode()).hexdigest()[:16] # endregion _compute_dict_snapshot_hash # #endregion TranslationOrchestrator