busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix: persist environment selection, support Kilo AI provider, resolve Superset virtual columns for translation preview

Browse Source 
- Add environment_id to TranslationJob model/schema/API + DB migration
- Pass environmentId from page to TranslationPreview and fetchPreview
- Fix _fetch_sample_rows: use result_type='samples' to include virtual cols
- Add 'kilo' and 'openrouter' to supported LLM provider types
- Make response_format conditional (skip for non-OpenAI upstream providers)
- Remove source_table field from form (redundant with datasource)
- Restore datasourceSearch display from saved job on page load
- Add request/response logging for LLM API calls
This commit is contained in:
busya
2026-05-09 23:10:15 +03:00
parent 85121191d4
commit c2bff20937
15 changed files with 292 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
backend/src/api/routes/translate/_preview_routes.py
View File

@@ -52,6 +52,7 @@ async def preview_translation(
job_id=job_id,
sample_size=payload.sample_size,
prompt_template=payload.prompt_template,
env_id=payload.env_id,
)
return result
except ValueError as e:

33
backend/src/core/database.py
View File

@@ -475,6 +475,38 @@ def _ensure_filter_source_enum_values(bind_engine):
# @SIDE_EFFECT: Executes ALTER TABLE statements against dataset review tables in the application database.
# @RELATION: [DEPENDS_ON] ->[DatasetReviewSession]
# @RELATION: [DEPENDS_ON] ->[ImportedFilter]
def _ensure_translation_jobs_columns(bind_engine):
with belief_scope("_ensure_translation_jobs_columns"):
table_name = "translation_jobs"
inspector = inspect(bind_engine)
if table_name not in inspector.get_table_names():
return
existing_columns = {
str(column.get("name") or "").strip()
for column in inspector.get_columns(table_name)
}
if "environment_id" not in existing_columns:
try:
with bind_engine.begin() as connection:
connection.execute(
text(
"ALTER TABLE translation_jobs "
"ADD COLUMN environment_id VARCHAR"
)
)
logger.reflect(
"Added environment_id column to translation_jobs",
)
except Exception as migration_error:
logger.explore(
"Failed to add environment_id to translation_jobs",
extra={"error": str(migration_error)},
)
raise
def _ensure_dataset_review_session_columns(bind_engine):
with belief_scope("_ensure_dataset_review_session_columns"):
inspector = inspect(bind_engine)
@@ -573,6 +605,7 @@ def init_db():
ensure_connection_configs_table(engine)
_ensure_filter_source_enum_values(engine)
_ensure_dataset_review_session_columns(engine)
_ensure_translation_jobs_columns(engine)
# [/DEF:init_db:Function]

3
backend/src/models/translate.py
View File

@@ -53,6 +53,9 @@ class TranslationJob(Base):
batch_size = Column(Integer, nullable=False, default=50, comment="Records per batch")
upsert_strategy = Column(String, nullable=False, default="MERGE", comment="MERGE, INSERT, UPDATE")
# Environment association
environment_id = Column(String, nullable=True, comment="Superset environment ID for datasource access")
created_by = Column(String, nullable=True)
created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = Column(DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))

22
backend/src/plugins/translate/executor.py
View File

@@ -527,12 +527,13 @@ class TranslationExecutor:
model = provider.default_model or "gpt-4o-mini"
provider_type = provider.provider_type.lower() if provider.provider_type else "openai"
if provider_type in ("openai", "openai_compatible"):
if provider_type in ("openai", "openai_compatible", "openrouter", "kilo"):
return self._call_openai_compatible(
base_url=provider.base_url,
api_key=api_key,
model=model,
prompt=prompt,
provider_type=provider_type,
)
else:
raise ValueError(f"Unsupported provider type '{provider_type}'")
@@ -549,6 +550,7 @@ class TranslationExecutor:
api_key: str,
model: str,
prompt: str,
provider_type: str = "openai",
) -> str:
with belief_scope("TranslationExecutor._call_openai_compatible"):
import requests as http_requests
@@ -565,10 +567,26 @@ class TranslationExecutor:
{"role": "user", "content": prompt},
],
"temperature": 0.1,
"response_format": {"type": "json_object"},
"max_tokens": 4096,
}
# Structured output (response_format) only for native OpenAI — upstream providers routed via
# Kilo/OpenRouter may not support it (e.g. StepFun returns "structured_outputs is not supported")
if provider_type in ("openai", "openai_compatible"):
payload["response_format"] = {"type": "json_object"}
logger.reason(
f"LLM request 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)}"
)
response = http_requests.post(url, headers=headers, json=payload, timeout=180)
if not response.ok:
logger.explore(
f"LLM API error status={response.status_code} "
f"model={payload.get('model')} "
f"body={response.text[:2000]}"
)
response.raise_for_status()
data = response.json()

2
backend/src/plugins/translate/orchestrator.py
View File

@@ -387,7 +387,7 @@ class TranslationOrchestrator:
# Submit to Superset
try:
env_id = job.source_dialect or ""
env_id = job.environment_id or job.source_dialect or ""
executor = SupersetSqlLabExecutor(self.config_manager, env_id)
result = executor.execute_and_poll(
sql=sql,

108
backend/src/plugins/translate/preview.py
View File

@@ -147,6 +147,7 @@ class TranslationPreview:
job_id: str,
sample_size: int = 10,
prompt_template: Optional[str] = None,
env_id: Optional[str] = None,
) -> Dict[str, Any]:
with belief_scope("TranslationPreview.preview_rows"):
logger.reason("Starting preview for job", {"job_id": job_id, "sample_size": sample_size})
@@ -173,12 +174,22 @@ class TranslationPreview:
source_rows = self._fetch_sample_rows(
job=job,
sample_size=sample_size,
env_id=env_id,
)
if not source_rows:
raise ValueError("No rows returned from datasource for preview")
actual_row_count = len(source_rows)
logger.reason("Fetched sample rows", {"actual_count": actual_row_count})
logger.reason(f"Fetched {actual_row_count} sample row(s)")
# Debug: log first row keys and translation column value
if source_rows:
first_row = source_rows[0]
logger.reason(
f"First source row keys={list(first_row.keys())} "
f"translation_col={job.translation_column} "
f"val='{first_row.get(job.translation_column, '')}'"
)
# 4. Build prompt context from rows
all_source_texts = []
@@ -513,17 +524,18 @@ class TranslationPreview:
# @PRE: job has source_datasource_id and translation_column.
# @POST: Returns list of dicts with row data.
# @SIDE_EFFECT: Calls Superset chart data endpoint.
def _fetch_sample_rows(self, job: TranslationJob, sample_size: int = 10) -> List[Dict[str, Any]]:
def _fetch_sample_rows(self, job: TranslationJob, sample_size: int = 10, env_id: Optional[str] = None) -> List[Dict[str, Any]]:
with belief_scope("TranslationPreview._fetch_sample_rows"):
# Find environment config using source_dialect as env_id
# Determine environment: prefer explicit env_id, then job.environment_id, then job.source_dialect (legacy)
environments = self.config_manager.get_environments()
target_env_id = env_id or job.environment_id or job.source_dialect or ""
env_config = next(
(e for e in environments if e.id == job.source_dialect),
(e for e in environments if e.id == target_env_id),
None,
)
if not env_config:
logger.explore("Could not find environment for datasource", {
"env_id": job.source_dialect,
"env_id": target_env_id,
})
# Fallback: try first environment
if environments:
@@ -539,12 +551,12 @@ class TranslationPreview:
# Fetch dataset detail to build proper query context
dataset_detail = client.get_dataset_detail(int(job.source_datasource_id))
# Determine columns to query
query_columns = [job.translation_column]
if job.context_columns:
query_columns.extend(job.context_columns)
# Build query context for chart data endpoint
# Build query context for chart data endpoint.
# Virtual columns (e.g. comment_text_ru) are NOT resolved when:
# - result_type="query" (physical columns only)
# - query_mode="raw" (virtual columns unavailable in raw mode)
# Solution: remove both result_type="query" AND query_mode="raw",
# use aggregate mode with no metrics — this resolves virtual columns.
query_context = client.build_dataset_preview_query_context(
dataset_id=int(job.source_datasource_id),
dataset_record=dataset_detail,
@@ -552,13 +564,18 @@ class TranslationPreview:
effective_filters=[],
)
# Modify to fetch specific columns as raw data (no aggregation)
# Modify: use result_type="samples" which returns sample data
# including all columns (physical + virtual), without needing
# explicit column objects that trigger validation errors.
queries = query_context.get("queries", [])
if queries:
queries[0]["columns"] = query_columns
queries[0]["metrics"] = []
queries[0]["row_limit"] = sample_size
queries[0]["result_type"] = "query"
queries[0].pop("result_type", None)
queries[0].pop("columns", None)
queries[0]["metrics"] = []
query_context["result_type"] = "samples"
form_data = query_context.get("form_data", {})
form_data.pop("query_mode", None)
try:
response = client.network.request(
@@ -568,23 +585,22 @@ class TranslationPreview:
headers={"Content-Type": "application/json"},
)
except Exception as e:
# Try legacy endpoint as fallback
logger.explore("Chart data endpoint failed, trying legacy", {"error": str(e)})
try:
response = client.network.request(
method="POST",
endpoint="/explore_json/form_data",
params={"form_data": json.dumps(query_context.get("form_data", {}))},
headers={"Content-Type": "application/json"},
)
except Exception as e2:
raise ValueError(
f"Failed to fetch sample data from Superset: {e2}"
)
logger.explore("Chart data API failed", {"error": str(e)})
raise ValueError(f"Failed to fetch sample data from Superset: {e}")
# Parse response
rows = self._extract_data_rows(response)
logger.reason("Extracted data rows", {"count": len(rows)})
logger.reason(f"Extracted {len(rows)} data row(s)")
# Debug: log first row keys and translation column value
if rows:
first_row = rows[0]
logger.reason(
f"Row keys={list(first_row.keys())} "
f"target_col={job.translation_column} "
f"val='{first_row.get(job.translation_column, '')}'"
)
return rows
# [/DEF:_fetch_sample_rows:Function]
@@ -646,12 +662,13 @@ class TranslationPreview:
model = provider.default_model or "gpt-4o-mini"
provider_type = provider.provider_type.lower() if provider.provider_type else "openai"
if provider_type in ("openai", "openai_compatible"):
if provider_type in ("openai", "openai_compatible", "openrouter", "kilo"):
response_text = self._call_openai_compatible(
base_url=provider.base_url,
api_key=api_key,
model=model,
prompt=prompt,
provider_type=provider_type,
)
else:
raise ValueError(f"Unsupported provider type '{provider_type}' for preview")
@@ -675,6 +692,7 @@ class TranslationPreview:
api_key: str,
model: str,
prompt: str,
provider_type: str = "openai",
) -> str:
with belief_scope("TranslationPreview._call_openai_compatible"):
import requests as http_requests
@@ -691,10 +709,26 @@ class TranslationPreview:
{"role": "user", "content": prompt},
],
"temperature": 0.1,
"response_format": {"type": "json_object"},
"max_tokens": 4096,
}
# Structured output (response_format) only for native OpenAI — upstream providers routed via
# Kilo/OpenRouter may not support it (e.g. StepFun returns "structured_outputs is not supported")
if provider_type in ("openai", "openai_compatible"):
payload["response_format"] = {"type": "json_object"}
logger.reason(
f"LLM request 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)}"
)
response = http_requests.post(url, headers=headers, json=payload, timeout=120)
if not response.ok:
logger.explore(
f"LLM API error status={response.status_code} "
f"model={payload.get('model')} "
f"body={response.text[:2000]}"
)
response.raise_for_status()
data = response.json()
@@ -716,6 +750,8 @@ class TranslationPreview:
@staticmethod
def _parse_llm_response(response_text: str, expected_count: int) -> Dict[str, str]:
with belief_scope("TranslationPreview._parse_llm_response"):
logger.reason(f"Raw LLM response length={len(response_text)} preview={response_text[:500]}")
try:
data = json.loads(response_text)
except json.JSONDecodeError:
@@ -732,6 +768,7 @@ class TranslationPreview:
rows = data.get("rows", [])
if not isinstance(rows, list):
logger.explore(f"LLM response has no 'rows' array, keys={list(data.keys())} text_preview={response_text[:300]}")
raise ValueError("LLM response missing 'rows' array")
translations: Dict[str, str] = {}
@@ -742,11 +779,10 @@ class TranslationPreview:
translations[row_id] = translation
if len(translations) < expected_count:
logger.explore("LLM returned fewer translations than expected", {
"expected": expected_count,
"received": len(translations),
"missing": [str(i) for i in range(expected_count) if str(i) not in translations],
})
logger.explore(
f"LLM returned fewer translations expected={expected_count} "
f"got={len(translations)} response_preview={response_text[:600]}"
)
return translations
# [/DEF:_parse_llm_response:Function]

10
backend/src/plugins/translate/service.py
View File

@@ -55,8 +55,10 @@ def get_dialect_from_database(database_record: Dict[str, Any]) -> str:
# Map Superset backend names to normalized dialect
dialect_map = {
"postgresql": "postgresql",
"greenplum": "postgresql",
"mysql": "mysql",
"clickhouse": "clickhouse",
"clickhousedb": "clickhouse",
"sqlite": "sqlite",
"mssql": "mssql",
"oracle": "oracle",
@@ -211,11 +213,11 @@ class TranslateJobService:
# Detect database dialect and validate columns if datasource is specified
dialect = payload.database_dialect
if payload.source_datasource_id and payload.source_dialect:
if payload.source_datasource_id and (payload.environment_id or payload.source_dialect):
# If no explicit dialect, try to detect it
if not dialect:
try:
env_id = payload.source_dialect
env_id = payload.environment_id or payload.source_dialect
_, detected_dialect = fetch_datasource_metadata(
int(payload.source_datasource_id),
env_id,
@@ -246,6 +248,7 @@ class TranslateJobService:
provider_id=payload.provider_id,
batch_size=payload.batch_size,
upsert_strategy=payload.upsert_strategy,
environment_id=payload.environment_id,
status="DRAFT",
created_by=self.current_user,
)
@@ -287,7 +290,7 @@ class TranslateJobService:
# Re-detect dialect if datasource changed
if payload.source_datasource_id and not payload.database_dialect:
try:
env_id = (payload.source_dialect or job.source_dialect)
env_id = (payload.environment_id or payload.source_dialect or job.environment_id or job.source_dialect)
_, detected_dialect = fetch_datasource_metadata(
int(payload.source_datasource_id),
env_id,
@@ -470,6 +473,7 @@ def job_to_response(job: TranslationJob, dict_ids: Optional[List[str]] = None) -
created_at=job.created_at,
updated_at=job.updated_at,
dictionary_ids=dict_ids or [],
environment_id=job.environment_id,
)
# [/DEF:job_to_response:Function]

4
backend/src/schemas/translate.py
View File

@@ -32,6 +32,7 @@ class TranslateJobCreate(BaseModel):
batch_size: int = Field(50, description="Records per batch")
upsert_strategy: str = Field("MERGE", description="UPSERT strategy: MERGE, INSERT, UPDATE")
dictionary_ids: Optional[List[str]] = Field(default_factory=list, description="Associated terminology dictionary IDs")
environment_id: Optional[str] = Field(None, description="Superset environment ID")
# [/DEF:TranslateJobCreate:Class]
@@ -57,6 +58,7 @@ class TranslateJobUpdate(BaseModel):
upsert_strategy: Optional[str] = None
status: Optional[str] = None
dictionary_ids: Optional[List[str]] = None
environment_id: Optional[str] = None
# [/DEF:TranslateJobUpdate:Class]
@@ -86,6 +88,7 @@ class TranslateJobResponse(BaseModel):
created_at: datetime
updated_at: Optional[datetime] = None
dictionary_ids: Optional[List[str]] = None
environment_id: Optional[str] = None
class Config:
from_attributes = True
@@ -214,6 +217,7 @@ class DictionaryImportResult(BaseModel):
class PreviewRequest(BaseModel):
sample_size: int = Field(10, ge=1, le=100, description="Number of sample rows to preview")
prompt_template: Optional[str] = Field(None, description="Optional custom prompt template")
env_id: Optional[str] = Field(None, description="Superset environment ID for preview data fetch")
# [DEF:PreviewRowUpdate:Class]