diff --git a/backend/src/plugins/llm_analysis/models.py b/backend/src/plugins/llm_analysis/models.py index 31c31b5b..400941c4 100644 --- a/backend/src/plugins/llm_analysis/models.py +++ b/backend/src/plugins/llm_analysis/models.py @@ -17,6 +17,7 @@ class LLMProviderType(str, Enum): OPENAI = "openai" OPENROUTER = "openrouter" KILO = "kilo" + LITELLM = "litellm" # #endregion LLMProviderType # #region LLMProviderConfig [TYPE Class] diff --git a/backend/src/plugins/translate/__tests__/test_clickhouse_insert_integration.py b/backend/src/plugins/translate/__tests__/test_clickhouse_insert_integration.py index 6ce0d255..d029a554 100644 --- a/backend/src/plugins/translate/__tests__/test_clickhouse_insert_integration.py +++ b/backend/src/plugins/translate/__tests__/test_clickhouse_insert_integration.py @@ -36,6 +36,10 @@ def _make_mock_job(**overrides): job.environment_id = "test-env" job.context_columns = [] job.target_column = None + job.target_language_column = None + job.target_source_column = None + job.target_source_language_column = None + job.target_languages = ["en"] for k, v in overrides.items(): setattr(job, k, v) return job @@ -63,6 +67,7 @@ def _make_mock_record( rec.status = "SUCCESS" rec.error_message = None rec.created_at = None + rec.languages = [] rec.source_data = { "report_date": report_date, "document_number": document_number, @@ -347,8 +352,9 @@ class TestOrchestratorInsertFlow: ) # Mock the DB query to return our records + # Note: orchestrator uses .options(joinedload()) before .filter() mock_query = MagicMock() - mock_query.filter.return_value.all.return_value = [rec1, rec2] + mock_query.options.return_value.filter.return_value.all.return_value = [rec1, rec2] db.query.return_value = mock_query # Create mock job diff --git a/backend/src/plugins/translate/executor.py b/backend/src/plugins/translate/executor.py index 08bc7fcf..0127090a 100644 --- a/backend/src/plugins/translate/executor.py +++ b/backend/src/plugins/translate/executor.py @@ -1,4 +1,4 @@ -# #region TranslationExecutor [C:4] [TYPE Module] [SEMANTICS sqlalchemy, tenacity, translate, insert, llm-retry] +# #region TranslationExecutor [C:5] [TYPE Module] [SEMANTICS sqlalchemy, tenacity, translate, insert, llm-retry] # @BRIEF Process translation in batches: fetch source rows, call LLM, persist TranslationBatch and TranslationRecord rows. # @LAYER: Domain # @RELATION DEPENDS_ON -> [TranslationBatch] @@ -10,6 +10,8 @@ # @PRE: Valid TranslationRun with job configuration. DB session is available. # @POST: TranslationBatch and TranslationRecord rows are created. Run status is updated. # @SIDE_EFFECT: Calls LLM provider; creates DB rows; updates run statistics. +# @DATA_CONTRACT Input: TranslationRun + Job -> Output: updated Run with batch/record rows +# @INVARIANT Batch processing is independent — one batch failure does not affect others. # @RATIONALE: Batch processing with retry — independent batches allow partial recovery. # @REJECTED: Single monolithic LLM call — would lose all progress on any failure. @@ -20,7 +22,7 @@ from collections.abc import Callable from datetime import UTC, datetime from typing import Any -from sqlalchemy.orm import Session +from sqlalchemy.orm import Session, joinedload from ...core.config_manager import ConfigManager from ...core.logger import belief_scope, logger @@ -32,6 +34,7 @@ from ...models.translate import ( TranslationPreviewSession, TranslationRecord, TranslationRun, + TranslationRunLanguageStats, ) from ...services.llm_prompt_templates import render_prompt from ...services.llm_provider import LLMProviderService @@ -39,16 +42,17 @@ from ._token_budget import DEFAULT_CONTEXT_WINDOW, DEFAULT_MAX_OUTPUT_TOKENS, es from .dictionary import DictionaryManager from .preview import DEFAULT_EXECUTION_PROMPT_TEMPLATE from .prompt_builder import ContextAwarePromptBuilder +from .sql_generator import SQLGenerator, _normalize_timestamp_value +from .superset_executor import SupersetSqlLabExecutor # #region MAX_RETRIES_PER_BATCH [TYPE Constant] -# @BRIEF Maximum number of retries for a single batch before marking it failed. MAX_RETRIES_PER_BATCH = 3 # #endregion MAX_RETRIES_PER_BATCH # #region MAX_ROWS_PER_RUN [TYPE Constant] -# @BRIEF Safety cap on rows fetched from datasource per run to prevent unbounded LLM processing. -# @RATIONALE Without a cap, a datasource with thousands of rows blocks the single uvicorn worker -# for minutes/hours. Preview uses sample_size (5-10). Full run should stay within reasonable bounds. +# Safety cap: without it, a datasource with thousands of rows blocks the single +# uvicorn worker for minutes/hours. Preview uses sample_size (5-10). Full run +# should stay within reasonable bounds. MAX_ROWS_PER_RUN = 10000 # #endregion MAX_ROWS_PER_RUN @@ -82,6 +86,7 @@ class TranslationExecutor: self, run: TranslationRun, llm_progress_callback: Callable[[str, int, int, int], None] | None = None, + language_stats_map: dict[str, TranslationRunLanguageStats] | None = None, ) -> TranslationRun: with belief_scope("TranslationExecutor.execute_run"): job = self.db.query(TranslationJob).filter(TranslationJob.id == run.job_id).first() @@ -170,6 +175,19 @@ class TranslationExecutor: # status + batch progress visible to other DB sessions (frontend polling). self.db.commit() + # Incremental INSERT into target table after each batch, + # so data appears incrementally in the target view/table + # without waiting for the entire run to complete. + batch_id = batch_result.get("batch_id") + if batch_id and batch_result["successful"] > 0: + try: + self._insert_batch_to_target(job, batch_id, run.id) + except Exception as e: + logger.explore("Batch INSERT failed (non-fatal, continuing)", { + "batch_id": batch_id, + "error": str(e), + }) + # Re-fetch run after commit to check for cancellation flag set by # cancel_run() fallback (direct SQL UPDATE of error_message). self.db.refresh(run) @@ -184,6 +202,17 @@ class TranslationExecutor: self.db.commit() return run + # Update per-language statistics incrementally after each batch + # so the frontend shows real-time per-language counts for RUNNING runs. + if language_stats_map and batch_result["successful"] > 0: + try: + self._update_language_stats_incremental(run.id, language_stats_map) + except Exception as e: + logger.explore("Language stats update failed (non-fatal)", { + "batch_id": batch_id, + "error": str(e), + }) + if self.on_batch_progress: self.on_batch_progress( run.id, batch_idx + 1, len(batches), @@ -665,9 +694,251 @@ class TranslationExecutor: **result, }) - return result + return {**result, "batch_id": batch_id} # endregion _process_batch + # region _insert_batch_to_target [TYPE Function] + # @PURPOSE: Insert successful records from a single batch into the target table via Superset SQL Lab. + # @PRE: batch has committed TranslationRecords with status SUCCESS. job has target_table configured. + # @POST: Per record, N+1 rows are INSERTED: 1 original + N translations. + # Context columns are bundled into JSON in the `context` column. + # `is_original=1` marks the source-language row. + # @SIDE_EFFECT: HTTP call to Superset SQL Lab API. Writes to target database. + def _insert_batch_to_target( + self, + job: TranslationJob, + batch_id: str, + run_id: str, + ) -> None: + with belief_scope("TranslationExecutor._insert_batch_to_target"): + records = ( + self.db.query(TranslationRecord) + .options(joinedload(TranslationRecord.languages)) + .filter( + TranslationRecord.batch_id == batch_id, + TranslationRecord.status == "SUCCESS", + TranslationRecord.target_sql.isnot(None), + ) + .all() + ) + + if not records: + return + + 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 (columns common to original and all translations) ── + base_row: dict[str, object] = {} + + # Key columns (report_date, document_number) — normalize timestamps to YYYY-MM-DD + 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 + + # Source text: original + if job.target_source_column: + base_row[job.target_source_column] = rec.source_sql or "" + + # Source language (same for all rows of this record) + if job.target_source_language_column: + base_row[job.target_source_language_column] = detected_src_lang + + # Context JSON string + 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 → single translation row with primary language + 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] + + 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 for batch insert", { + "batch_id": batch_id, + "error": str(e), + }) + real_backend = None + + dialect = real_backend or job.database_dialect or job.target_dialect or "postgresql" + + 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 for batch", { + "batch_id": batch_id, + "error": str(e), + }) + return + + 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 for batch", { + "batch_id": batch_id, + "error": str(e), + }) + return + + logger.reason(f"Batch {batch_id[:12]} inserted {row_count} rows", { + "batch_id": batch_id, + "rows": row_count, + "status": result.get("status"), + }) + # endregion _insert_batch_to_target + + # region _update_language_stats_incremental [TYPE Function] + # @PURPOSE: Update per-language TranslationRunLanguageStats incrementally after each batch. + # @PRE: language_stats_map has entries for all target languages. + # @POST: Language stat objects updated with counts from committed TranslationLanguage rows. + # @SIDE_EFFECT: Mutates ORM objects; caller must commit. + def _update_language_stats_incremental( + self, + run_id: str, + language_stats_map: dict[str, TranslationRunLanguageStats], + ) -> None: + with belief_scope("TranslationExecutor._update_language_stats_incremental"): + records = ( + self.db.query(TranslationRecord) + .filter(TranslationRecord.run_id == run_id) + .all() + ) + record_ids = [r.id for r in records] + if not record_ids: + return + + lang_entries = ( + self.db.query(TranslationLanguage) + .filter(TranslationLanguage.record_id.in_(record_ids)) + .all() + ) + + 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 + + total_tokens_est = max(1, sum(len(le.translated_value or "") for le in lang_entries if le.translated_value) // 4) + num_langs = len(language_stats_map) or 1 + cost_per_token = 0.002 / 1000 + + 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"] + lang_stat.token_count = total_tokens_est // num_langs + lang_stat.estimated_cost = round((lang_stat.token_count / 1000) * cost_per_token, 6) + + self.db.flush() + # endregion _update_language_stats_incremental + # region _call_llm_for_batch [TYPE Function] # @PURPOSE: Call LLM for a batch of rows requiring translation. Parse structured JSON response. # @PRE: job has valid provider_id. batch_rows is non-empty. @@ -959,7 +1230,7 @@ class TranslationExecutor: disable_reasoning = getattr(job, 'disable_reasoning', False) - if provider_type in ("openai", "openai_compatible", "openrouter", "kilo"): + if provider_type in ("openai", "openai_compatible", "openrouter", "kilo", "litellm"): response_text = self._call_openai_compatible( base_url=provider.base_url, api_key=api_key, @@ -1006,23 +1277,23 @@ class TranslationExecutor: "temperature": 0.1, "max_tokens": max_tokens, } - # Structured output — native OpenAI and compatible providers (e.g. Ollama, vLLM). - # Kilo gateway API docs show response_format support, but upstream providers (e.g. StepFun) - # reject it with "structured_outputs is not supported". Skip for Kilo/OpenRouter to avoid 400. - if provider_type in ("openai", "openai_compatible"): + # Structured output — OpenRouter and Kilo also support response_format, but some + # upstream providers (e.g. StepFun) reject it. We try with response_format and + # fall back on 400 "structured_outputs is not supported". + if provider_type in ("openai", "openai_compatible", "kilo", "openrouter", "litellm"): payload["response_format"] = {"type": "json_object"} # Suppress Chain of Thought reasoning to save output tokens - # NOTE: Kilo/OpenRouter do NOT support disabling reasoning (returns 400) + # NOTE: Kilo/OpenRouter/LiteLLM do NOT support disabling reasoning (returns 400) if disable_reasoning: - if provider_type not in ("kilo", "openrouter"): + if provider_type not in ("kilo", "openrouter", "litellm"): payload["reasoning_effort"] = "none" payload["extra_body"] = {"reasoning_effort": "none"} payload.pop("response_format", None) payload["messages"][0] = {"role": "system", "content": "You are a database content translation assistant. Translate the provided text accurately, preserving data semantics. Respond directly with ONLY the JSON result. Do NOT include any reasoning, thinking, chain-of-thought, analysis, or explanation. Output ONLY valid JSON."} logger.reason( - f"LLM request model={payload.get('model')} " + f"LLM request url={base_url} model={payload.get('model')} " f"provider_type={provider_type} " f"response_format={'yes' if 'response_format' in payload else 'no'} " f"prompt_len={len(prompt)}" diff --git a/backend/src/plugins/translate/orchestrator.py b/backend/src/plugins/translate/orchestrator.py index 8d296b69..2bbf96b1 100644 --- a/backend/src/plugins/translate/orchestrator.py +++ b/backend/src/plugins/translate/orchestrator.py @@ -39,7 +39,7 @@ from ...models.translate import ( ) from .events import TranslationEventLog from .executor import TranslationExecutor -from .sql_generator import SQLGenerator +from .sql_generator import SQLGenerator, _normalize_timestamp_value from .superset_executor import SupersetSqlLabExecutor @@ -232,7 +232,7 @@ class TranslationOrchestrator: on_batch_progress=on_batch_progress, ) try: - run = executor.execute_run(run, llm_progress_callback=None) + 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, @@ -371,83 +371,106 @@ class TranslationOrchestrator: "dialect": job.database_dialect or job.target_dialect, }) - # Determine effective target column for INSERT (defaults to translation_column) effective_target = job.target_column or job.translation_column - - # Build columns for SQL generation - columns = job.context_columns or [] - - # Always include translation_column (original text) if it's different from target - if job.translation_column and job.translation_column not in columns: - columns.append(job.translation_column) - - # Add target_column separately if it differs from translation_column - if effective_target and effective_target != job.translation_column and effective_target not in columns: - columns.append(effective_target) - - # Also include key columns if used for upsert - if job.target_key_cols: - for k in job.target_key_cols: - if k not in columns: - columns.append(k) - - # Add target metadata columns for enhanced table mapping - if job.target_source_column and job.target_source_column not in columns: - columns.append(job.target_source_column) - if job.target_language_column and job.target_language_column not in columns: - columns.append(job.target_language_column) - if job.target_source_language_column and job.target_source_language_column not in columns: - columns.append(job.target_source_language_column) - - # Resolve the primary target language for INSERT primary_language = (job.target_languages or ["en"])[0] - rows_for_sql = [] + # 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: - row_data = {} source_data = rec.source_data or {} - # Context columns from source data - if job.context_columns: - for col in job.context_columns: - row_data[col] = source_data.get(col, "") + # 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" - # Original text column (translation_column) - if job.translation_column and job.translation_column not in (job.target_key_cols or []): - # If target_column differs, keep the original value from source_data; - # otherwise use the translated value - if effective_target and effective_target != job.translation_column: - row_data[job.translation_column] = source_data.get(job.translation_column, "") - else: - row_data[job.translation_column] = rec.target_sql or "" + # 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 "" - # Translated text goes into target_column (may be same as translation_column) - if effective_target: - row_data[effective_target] = rec.target_sql or "" + # ── Shared base row ── + base_row: dict[str, object] = {} - # Source text column: INSERT original source text - if job.target_source_column: - row_data[job.target_source_column] = rec.source_sql or "" - - # Language column: INSERT language code (e.g. 'ru', 'en') - if job.target_language_column: - row_data[job.target_language_column] = primary_language - - # Source language column: INSERT detected source language (BCP-47) - if job.target_source_language_column: - detected = "und" - if rec.languages and len(rec.languages) > 0: - detected = rec.languages[0].source_language_detected or "und" - row_data[job.target_source_language_column] = detected - - # Key columns from source data if job.target_key_cols: for k in job.target_key_cols: - row_data[k] = source_data.get(k, "") - rows_for_sql.append(row_data) + 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: - # Use target_sql as the sole column columns = [effective_target or "translated_text"] rows_for_sql = [{columns[0]: rec.target_sql or ""} for rec in records] @@ -1094,6 +1117,4 @@ class TranslationOrchestrator: return hashlib.sha256(hash_input.encode()).hexdigest()[:16] # endregion _compute_dict_snapshot_hash - -# #endregion TranslationOrchestrator # #endregion TranslationOrchestrator diff --git a/backend/src/plugins/translate/preview.py b/backend/src/plugins/translate/preview.py index 8fd531f7..5f86350a 100644 --- a/backend/src/plugins/translate/preview.py +++ b/backend/src/plugins/translate/preview.py @@ -964,7 +964,7 @@ class TranslationPreview: provider_type = provider.provider_type.lower() if provider.provider_type else "openai" disable_reasoning = getattr(job, 'disable_reasoning', False) - if provider_type in ("openai", "openai_compatible", "openrouter", "kilo"): + if provider_type in ("openai", "openai_compatible", "openrouter", "kilo", "litellm"): max_attempts = 2 last_error = None for attempt in range(max_attempts): @@ -1030,14 +1030,14 @@ class TranslationPreview: } # Structured output — Kilo gateway supports response_format, but upstream providers # (e.g. StepFun) may reject it. We try with response_format and fall back on 400. - if provider_type in ("openai", "openai_compatible", "kilo", "openrouter"): + if provider_type in ("openai", "openai_compatible", "kilo", "openrouter", "litellm"): payload["response_format"] = {"type": "json_object"} # Suppress Chain of Thought reasoning to save output tokens - # NOTE: Kilo/OpenRouter do NOT support disabling reasoning (returns 400) + # NOTE: Kilo/OpenRouter/LiteLLM reject reasoning_effort — only use for native OpenAI-compatible if disable_reasoning: - # Kilo/OpenRouter reject reasoning_effort — only use for native OpenAI-compatible - if provider_type not in ("kilo", "openrouter"): + # Kilo/OpenRouter/LiteLLM reject reasoning_effort — only use for native OpenAI-compatible + if provider_type not in ("kilo", "openrouter", "litellm"): payload["reasoning_effort"] = "none" payload["extra_body"] = {"reasoning_effort": "none"} payload.pop("response_format", None) # JSON mode triggers reasoning on some models @@ -1048,7 +1048,7 @@ class TranslationPreview: payload["messages"][0] = {"role": "system", "content": system_content} logger.reason( - f"LLM request model={payload.get('model')} " + f"LLM request url={base_url} model={payload.get('model')} " f"provider_type={provider_type} " f"response_format={'yes' if 'response_format' in payload else 'no'} " f"reasoning={'no' if disable_reasoning else 'yes'} " diff --git a/frontend/src/components/llm/ProviderConfig.svelte b/frontend/src/components/llm/ProviderConfig.svelte index 44e17deb..88afc6d7 100644 --- a/frontend/src/components/llm/ProviderConfig.svelte +++ b/frontend/src/components/llm/ProviderConfig.svelte @@ -22,6 +22,13 @@ let editingProvider = $state(null); let showForm = $state(false); + const DEFAULT_BASE_URLS = { + openai: "https://api.openai.com/v1", + openrouter: "https://openrouter.ai/api/v1", + kilo: "https://api.kilo.chat/v1", + litellm: "http://localhost:4000/v1", + }; + let formData = $state({ name: "", provider_type: "openai", @@ -57,6 +64,16 @@ ); } + function updateBaseUrlForType(providerType) { + // Only auto-update base_url if user hasn't changed it from the default + // or the base_url matches a previous default for a different type + const currentUrl = formData.base_url; + const isCurrentlyDefault = Object.values(DEFAULT_BASE_URLS).includes(currentUrl); + if (isCurrentlyDefault || !currentUrl) { + formData.base_url = DEFAULT_BASE_URLS[providerType] || currentUrl; + } + } + function resetForm() { formData = { name: "", @@ -322,11 +339,13 @@ diff --git a/frontend/src/lib/components/translate/TranslationRunGlobalIndicator.svelte b/frontend/src/lib/components/translate/TranslationRunGlobalIndicator.svelte new file mode 100644 index 00000000..70871649 --- /dev/null +++ b/frontend/src/lib/components/translate/TranslationRunGlobalIndicator.svelte @@ -0,0 +1,139 @@ + + + + + + + + + + + +{#if show} +
+ +
+
+
+ +
+ {label} + + {#if uxState === 'running' || uxState === 'inserting'} + + {runState.current?.successfulRecords || 0}/{runState.current?.totalRecords || 0} + + {#if (runState.current?.failedRecords || 0) > 0} + + {$t.translate?.run?.failed || 'Failed'}: {runState.current?.failedRecords} + + {/if} + + ▶ {$t.common?.open || 'Open'} + + {:else if uxState === 'partial'} + + {runState.current?.successfulRecords || 0}/{runState.current?.totalRecords || 0} + + + ▶ {$t.common?.open || 'Open'} + + {:else} + + ▶ {$t.common?.open || 'Open'} + + {/if} +
+
+{/if} + diff --git a/frontend/src/lib/stores/translationRun.js b/frontend/src/lib/stores/translationRun.js new file mode 100644 index 00000000..6eea7284 --- /dev/null +++ b/frontend/src/lib/stores/translationRun.js @@ -0,0 +1,209 @@ +// #region TranslationRunStore [C:3] [TYPE Store] [SEMANTICS translate, run, progress, polling, store] +// @BRIEF Global store for active translation run progress — survives page navigation. +// @RELATION DEPENDS_ON -> [TranslateApi] +// @UX_STATE idle -> No active run +// @UX_STATE running -> Translation phase in progress +// @UX_STATE inserting -> Insert phase in progress +// @UX_STATE completed -> Run completed +// @UX_STATE partial -> Completed with errors +// @UX_STATE failed -> Run failed +// @UX_STATE cancelled -> Run cancelled +// @INVARIANT Polling stops when run reaches a terminal state (completed/failed/cancelled). +// @INVARIANT Store is writable so page components can also set initial state. + +import { writable, derived } from 'svelte/store'; +import { fetchRunStatus } from '$lib/api/translate.js'; + +/** + * @typedef {'idle'|'running'|'inserting'|'completed'|'partial'|'failed'|'cancelled'} UxState + */ + +/** + * @typedef {Object} TranslationRunState + * @property {string|null} runId - Active translation run ID + * @property {UxState} uxState - Current UX state + * @property {Object|null} status - Raw status data from API + * @property {number} totalRecords + * @property {number} successfulRecords + * @property {number} failedRecords + * @property {number} skippedRecords + * @property {number} progressPct - 0-100 + * @property {string|null} insertStatus + * @property {number} batchCount + * @property {string|null} jobId - The job this run belongs to (for navigation) + * @property {boolean} isFullRun + * @property {boolean} isActive - Derived: true when polling is active + */ + +const initialState = { + runId: null, + uxState: 'idle', + status: null, + totalRecords: 0, + successfulRecords: 0, + failedRecords: 0, + skippedRecords: 0, + progressPct: 0, + insertStatus: null, + batchCount: 0, + jobId: null, + isFullRun: false, +}; + +/** @type {import('svelte/store').Writable} */ +export const translationRunStore = writable(initialState); + +/** Derived: true when a run is actively being polled */ +export const isTranslationActive = derived( + translationRunStore, + ($store) => $store.uxState === 'running' || $store.uxState === 'inserting' +); + +/** Derived: true when run has finished */ +export const isTranslationFinished = derived( + translationRunStore, + ($store) => + $store.uxState === 'completed' || + $store.uxState === 'partial' || + $store.uxState === 'failed' || + $store.uxState === 'cancelled' +); + +// Internal polling state (not reactive) +let _pollingInterval = null; +let _pollCount = 0; +const MAX_POLLS = 300; +let _onCompleteCallback = null; + +/** + * Clear the onComplete callback without stopping polling. + * Used by page components when they unmount — the store keeps polling + * so the global indicator still works, but the page callback is detached. + */ +export function clearOnCompleteCallback() { + _onCompleteCallback = null; +} + +/** + * Start polling for a translation run. + * @param {string} runId + * @param {Object} [options] + * @param {string} [options.jobId] - Job ID for navigation + * @param {boolean} [options.isFullRun] + * @param {Function} [options.onComplete] - Called when run reaches terminal state + */ +export function startTranslationRun(runId, options = {}) { + if (!runId) return; + + stopTranslationRun(); + + translationRunStore.set({ + ...initialState, + runId, + uxState: 'running', + jobId: options.jobId || null, + isFullRun: options.isFullRun || false, + }); + + _onCompleteCallback = options.onComplete || null; + _pollCount = 0; + _pollingInterval = setInterval(pollStatus, 2000); + pollStatus(); +} + +/** + * Stop polling without clearing state (run may persist). + */ +export function stopTranslationRun() { + if (_pollingInterval) { + clearInterval(_pollingInterval); + _pollingInterval = null; + } +} + +/** + * Reset store to idle. + */ +export function resetTranslationRun() { + stopTranslationRun(); + translationRunStore.set(initialState); + _onCompleteCallback = null; +} + +/** + * Manually set the run state (used by page component for lifecycle hooks). + * @param {Partial} state + */ +export function updateTranslationRunState(state) { + translationRunStore.update(prev => ({ ...prev, ...state })); +} + +/** @returns {Promise} */ +async function pollStatus() { + _pollCount++; + if (_pollCount > MAX_POLLS) { + console.warn('[translationRunStore] Max polls reached, stopping'); + stopTranslationRun(); + translationRunStore.update(s => ({ ...s, uxState: 'failed' })); + if (_onCompleteCallback) _onCompleteCallback({ status: 'TIMEOUT', error_message: 'Translation run timed out' }); + return; + } + + let currentState; + translationRunStore.subscribe(s => { currentState = s; })(); + if (!currentState || !currentState.runId) { + stopTranslationRun(); + return; + } + + try { + const data = await fetchRunStatus(currentState.runId); + if (!data) return; + + const s = data?.status; + const insertS = data?.insert_status; + + let newUxState = currentState.uxState; + + if (s === 'PENDING' || s === 'RUNNING') { + newUxState = (insertS === 'started' || insertS === 'pending' || insertS === 'running') + ? 'inserting' : 'running'; + } else if (s === 'COMPLETED') { + newUxState = (insertS === 'failed' || insertS === 'timeout') ? 'insert_failed' + : (data.failed_records > 0) ? 'partial' : 'completed'; + stopTranslationRun(); + } else if (s === 'FAILED') { + newUxState = 'failed'; + stopTranslationRun(); + } else if (s === 'CANCELLED') { + newUxState = 'cancelled'; + stopTranslationRun(); + } + + const total = data?.total_records || 0; + const pct = total > 0 + ? Math.round(((data.successful_records + data.failed_records + data.skipped_records) / total) * 100) + : 0; + + translationRunStore.update(prev => ({ + ...prev, + uxState: newUxState, + status: data, + totalRecords: total, + successfulRecords: data?.successful_records || 0, + failedRecords: data?.failed_records || 0, + skippedRecords: data?.skipped_records || 0, + progressPct: pct, + insertStatus: data?.insert_status || null, + batchCount: data?.batch_count || 0, + })); + + // Fire completion callback if terminal + if (newUxState !== 'running' && newUxState !== 'inserting') { + if (_onCompleteCallback) _onCompleteCallback(data); + } + } catch (err) { + console.warn('[translationRunStore] Poll error:', err); + } +} +// #endregion TranslationRunStore diff --git a/frontend/src/routes/+layout.svelte b/frontend/src/routes/+layout.svelte index 28fa914d..c9507c15 100644 --- a/frontend/src/routes/+layout.svelte +++ b/frontend/src/routes/+layout.svelte @@ -30,6 +30,7 @@ import TopNavbar from '$lib/components/layout/TopNavbar.svelte'; import TaskDrawer from '$lib/components/layout/TaskDrawer.svelte'; import AssistantChatPanel from '$lib/components/assistant/AssistantChatPanel.svelte'; + import TranslationRunGlobalIndicator from '$lib/components/translate/TranslationRunGlobalIndicator.svelte'; import { t } from '$lib/i18n'; import { isProductionContextStore, @@ -53,6 +54,9 @@ + + +
{#if isLoginPage}
diff --git a/frontend/src/routes/translate/[id]/+page.svelte b/frontend/src/routes/translate/[id]/+page.svelte index 05dbf5a4..0fbf0905 100644 --- a/frontend/src/routes/translate/[id]/+page.svelte +++ b/frontend/src/routes/translate/[id]/+page.svelte @@ -30,7 +30,7 @@ * @UX_REACTIVITY: columnList is $derived from selected datasource */ - import { onMount } from 'svelte'; + import { onMount, onDestroy } from 'svelte'; import { goto } from '$app/navigation'; import { page } from '$app/state'; import { t, _ } from '$lib/i18n'; @@ -50,6 +50,13 @@ import ScheduleConfig from '$lib/components/translate/ScheduleConfig.svelte'; import { triggerRun, fetchRunHistory, cancelRun } from '$lib/api/translate.js'; import BulkReplaceModal from '$lib/components/translate/BulkReplaceModal.svelte'; + import { fromStore } from 'svelte/store'; + import { + translationRunStore, + startTranslationRun, + resetTranslationRun, + clearOnCompleteCallback, + } from '$lib/stores/translationRun.js'; const LANGUAGES = [ { code: 'ru', name: 'Russian' }, @@ -154,8 +161,9 @@ let validationErrors = $state({}); let warnings = $state([]); - // Run state - let currentRunId = $state(null); + // Run state — using global store to survive page navigation + let runState = fromStore(translationRunStore); + let currentRunId = $derived(runState.current?.runId || null); let completedRuns = $state([]); let isRunning = $state(false); let isFullRun = $state(false); @@ -168,7 +176,7 @@ runError = ''; try { const run = await triggerRun(jobId, full); - currentRunId = run.id; + startTranslationRun(run.id, { jobId, isFullRun: full, onComplete: handleRunComplete }); addToast(full ? 'Полный перевод запущен (все строки)' : _('translate.config.run_started'), 'success'); } catch (err) { runError = err?.message || _('translate.config.run_failed'); @@ -177,8 +185,9 @@ } function handleRunComplete(statusData) { + if (!isRunning) return; // Idempotent: already handled isRunning = false; - currentRunId = null; + resetTranslationRun(); loadRunHistory(); const statusLabel = statusData?.status || _('translate.run.completed'); addToast(`${_('translate.run.run_id')} ${statusLabel.toLowerCase()}`, 'info'); @@ -199,7 +208,7 @@ await cancelRun(currentRunId); } catch (_e) { /* ignore */ } } - currentRunId = null; + resetTranslationRun(); await handleTriggerRun(); } @@ -235,6 +244,12 @@ } }); + // Clean up store callback when page unmounts — the store keeps + // polling so the global indicator in the layout still works. + onDestroy(() => { + clearOnCompleteCallback(); + }); + /** @returns {Promise} */ async function loadInitialData() { try {