busya
bb21fd3165
M1: ReDoS guard — 500 char max + try/except on re.compile M2: Performance note — O(n×m) regex documented for future optimization M3: Verified test key names correct (metrics.py transforms cumulative_* → short) L1: Removed @staticmethod no-op from module-level function L2: Added aria-label to emoji indicators in CorrectionCell
1045 lines
43 KiB
Python
1045 lines
43 KiB
Python
# #region TranslateJobService [C:4] [TYPE Module] [SEMANTICS sqlalchemy, translate, job, datasource, dialect]
|
|
# @BRIEF Service layer for translation job CRUD with datasource column validation and database dialect detection.
|
|
# @LAYER: Domain
|
|
# @RELATION DEPENDS_ON -> [TranslationJob]
|
|
# @RELATION DEPENDS_ON -> [SupersetClient]
|
|
# @RELATION DEPENDS_ON -> [ConfigManager]
|
|
# @PRE: Database session and config manager are available.
|
|
# @POST: Translation jobs are created/updated/deleted with column validation and dialect caching.
|
|
# @SIDE_EFFECT: Queries Superset for column metadata and database dialect at save time.
|
|
# @RATIONALE: Snapshot isolation — in-progress runs use config snapshot; config edits affect future runs only.
|
|
# @REJECTED: Invalidating in-progress runs on config edit would break scheduled run continuity.
|
|
|
|
import re
|
|
import uuid
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ...core.config_manager import ConfigManager
|
|
from ...core.logger import logger
|
|
from ...core.superset_client import SupersetClient
|
|
from ...models.translate import TranslationJob, TranslationJobDictionary
|
|
from ...schemas.translate import (
|
|
DatasourceColumnResponse,
|
|
DatasourceColumnsResponse,
|
|
TranslateJobCreate,
|
|
TranslateJobResponse,
|
|
TranslateJobUpdate,
|
|
)
|
|
from .dictionary import _validate_bcp47
|
|
|
|
# Supported database dialects for translation
|
|
SUPPORTED_DIALECTS = {
|
|
"postgresql", "mysql", "clickhouse", "sqlite", "mssql",
|
|
"oracle", "snowflake", "bigquery", "redshift", "presto",
|
|
"trino", "druid", "hive", "spark", "databricks",
|
|
}
|
|
|
|
|
|
# #region get_dialect_from_database [TYPE Function]
|
|
# @BRIEF Extract normalized dialect string from a Superset database record.
|
|
# @PRE: database_record is a dict from Superset API.
|
|
# @POST: Returns normalized dialect string or raises ValueError if unsupported.
|
|
def get_dialect_from_database(database_record: dict[str, Any]) -> str:
|
|
"""Extract and validate dialect from Superset database record."""
|
|
backend = (
|
|
database_record.get("backend")
|
|
or database_record.get("engine")
|
|
or ""
|
|
).lower().strip()
|
|
|
|
if not backend:
|
|
raise ValueError("Could not determine database dialect from connection")
|
|
|
|
# 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",
|
|
"snowflake": "snowflake",
|
|
"bigquery": "bigquery",
|
|
"redshift": "redshift",
|
|
"presto": "presto",
|
|
"trino": "trino",
|
|
"druid": "druid",
|
|
"hive": "hive",
|
|
"spark": "spark",
|
|
"databricks": "databricks",
|
|
}
|
|
|
|
normalized = dialect_map.get(backend, backend)
|
|
if normalized not in SUPPORTED_DIALECTS:
|
|
raise ValueError(
|
|
f"Unsupported database dialect: '{backend}'. "
|
|
f"Supported dialects: {', '.join(sorted(SUPPORTED_DIALECTS))}"
|
|
)
|
|
return normalized
|
|
# #endregion get_dialect_from_database
|
|
|
|
|
|
# #region fetch_datasource_metadata [TYPE Function]
|
|
# @BRIEF Fetch datasource columns and database dialect from Superset.
|
|
# @PRE: dataset_id is a valid Superset dataset ID and environment has valid credentials.
|
|
# @POST: Returns (columns_list, dialect_string) or raises on failure.
|
|
def fetch_datasource_metadata(
|
|
dataset_id: int,
|
|
env_id: str,
|
|
config_manager: ConfigManager,
|
|
) -> tuple[list[dict[str, Any]], str]:
|
|
"""Fetch column metadata and database dialect for a datasource from Superset."""
|
|
# Find environment config
|
|
environments = config_manager.get_environments()
|
|
env_config = next((e for e in environments if e.id == env_id), None)
|
|
if not env_config:
|
|
raise ValueError(f"Superset environment '{env_id}' not found in configuration")
|
|
|
|
# Create Superset client and fetch dataset detail
|
|
client = SupersetClient(env_config)
|
|
dataset_detail = client.get_dataset_detail(dataset_id)
|
|
|
|
# Extract columns
|
|
raw_columns = dataset_detail.get("columns", [])
|
|
columns = []
|
|
for col in raw_columns:
|
|
col_name = col.get("name") or col.get("column_name")
|
|
if not col_name:
|
|
continue
|
|
columns.append({
|
|
"name": str(col_name),
|
|
"type": col.get("type"),
|
|
"is_physical": col.get("is_physical", True),
|
|
"is_dttm": col.get("is_dttm", False),
|
|
"description": col.get("description", ""),
|
|
})
|
|
|
|
# Extract database dialect from datasource
|
|
database_info = dataset_detail.get("database", {})
|
|
if isinstance(database_info, dict):
|
|
dialect = get_dialect_from_database(database_info)
|
|
else:
|
|
# Fallback: try to fetch database directly
|
|
try:
|
|
db_id = dataset_detail.get("database_id")
|
|
if db_id:
|
|
db_record = client.get_database(int(db_id))
|
|
db_result = db_record.get("result", db_record) if isinstance(db_record, dict) else db_record
|
|
dialect = get_dialect_from_database(db_result)
|
|
else:
|
|
raise ValueError("No database information available for this datasource")
|
|
except Exception as e:
|
|
logger.warning(f"[translate_service] Could not fetch database dialect: {e}")
|
|
raise ValueError(f"Could not determine database dialect: {e}")
|
|
|
|
return columns, dialect
|
|
# #endregion fetch_datasource_metadata
|
|
|
|
|
|
# #region detect_virtual_columns [TYPE Function]
|
|
# @BRIEF Identify virtual (calculated) columns from column metadata.
|
|
# @PRE: columns is a list of dicts with 'is_physical' key.
|
|
# @POST: Returns list of virtual column names.
|
|
def detect_virtual_columns(columns: list[dict[str, Any]]) -> list[str]:
|
|
"""Return names of columns that are virtual (not physical)."""
|
|
return [col["name"] for col in columns if not col.get("is_physical", True)]
|
|
# #endregion detect_virtual_columns
|
|
|
|
|
|
# #region TranslateJobService [TYPE Class]
|
|
# @BRIEF Service for translation job CRUD with validation and Superset integration.
|
|
class TranslateJobService:
|
|
|
|
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
|
|
|
|
# region list_jobs [TYPE Function]
|
|
# @PURPOSE: List translation jobs with optional status filter and pagination.
|
|
# @POST: Returns tuple of (total_count, list_of_jobs).
|
|
def list_jobs(
|
|
self,
|
|
page: int = 1,
|
|
page_size: int = 20,
|
|
status_filter: str | None = None,
|
|
) -> tuple[int, list[TranslationJob]]:
|
|
query = self.db.query(TranslationJob)
|
|
|
|
if status_filter:
|
|
query = query.filter(TranslationJob.status == status_filter)
|
|
|
|
total = query.count()
|
|
offset = (page - 1) * page_size
|
|
jobs = query.order_by(TranslationJob.created_at.desc()).offset(offset).limit(page_size).all()
|
|
|
|
return total, jobs
|
|
# endregion list_jobs
|
|
|
|
# region get_job [TYPE Function]
|
|
# @PURPOSE: Get a single translation job by ID.
|
|
# @POST: Returns TranslationJob or raises ValueError.
|
|
def get_job(self, job_id: str) -> TranslationJob:
|
|
job = self.db.query(TranslationJob).filter(TranslationJob.id == job_id).first()
|
|
if not job:
|
|
raise ValueError(f"Translation job '{job_id}' not found")
|
|
return job
|
|
# endregion get_job
|
|
|
|
# region create_job [TYPE Function]
|
|
# @PURPOSE: Create a new translation job with column validation.
|
|
# @PRE: payload contains valid job configuration.
|
|
# @POST: Returns the created TranslationJob with database_dialect cached.
|
|
# @SIDE_EFFECT: Validates columns via SupersetClient if source_datasource_id is provided.
|
|
# @SIDE_EFFECT: Caches database_dialect from Superset connection.
|
|
def create_job(self, payload: TranslateJobCreate) -> TranslationJob:
|
|
logger.info(f"[TranslateJobService] Creating job '{payload.name}'")
|
|
|
|
# Validate: must have a translation column if datasource is configured
|
|
if payload.source_datasource_id and not payload.translation_column:
|
|
raise ValueError("A translation column is required when a datasource is selected")
|
|
|
|
# Validate upsert strategy
|
|
valid_strategies = {"MERGE", "INSERT", "UPDATE"}
|
|
if payload.upsert_strategy not in valid_strategies:
|
|
raise ValueError(
|
|
f"Invalid upsert_strategy '{payload.upsert_strategy}'. "
|
|
f"Must be one of: {', '.join(sorted(valid_strategies))}"
|
|
)
|
|
|
|
# Detect database dialect and validate columns if datasource is specified
|
|
dialect = payload.database_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.environment_id or payload.source_dialect
|
|
_, detected_dialect = fetch_datasource_metadata(
|
|
int(payload.source_datasource_id),
|
|
env_id,
|
|
self.config_manager,
|
|
)
|
|
dialect = detected_dialect
|
|
except Exception as e:
|
|
logger.warning(f"[TranslateJobService] Dialect detection failed: {e}")
|
|
dialect = payload.source_dialect
|
|
|
|
# Resolve target_languages: accept new format (list) or legacy single language
|
|
target_languages = payload.target_languages
|
|
if not target_languages and payload.target_language:
|
|
target_languages = [payload.target_language]
|
|
if target_languages:
|
|
if not isinstance(target_languages, list):
|
|
target_languages = [str(target_languages)]
|
|
for lang in target_languages:
|
|
_validate_bcp47(lang, "target_languages")
|
|
|
|
# Build job instance
|
|
job = TranslationJob(
|
|
id=str(uuid.uuid4()),
|
|
name=payload.name,
|
|
description=payload.description,
|
|
source_dialect=payload.source_dialect,
|
|
target_dialect=payload.target_dialect,
|
|
database_dialect=dialect,
|
|
source_datasource_id=payload.source_datasource_id,
|
|
source_table=payload.source_table,
|
|
target_schema=payload.target_schema,
|
|
target_table=payload.target_table,
|
|
source_key_cols=payload.source_key_cols or [],
|
|
target_key_cols=payload.target_key_cols or [],
|
|
translation_column=payload.translation_column,
|
|
target_column=payload.target_column,
|
|
context_columns=payload.context_columns or [],
|
|
target_language=payload.target_language,
|
|
target_languages=target_languages,
|
|
provider_id=payload.provider_id,
|
|
batch_size=payload.batch_size,
|
|
upsert_strategy=payload.upsert_strategy,
|
|
environment_id=payload.environment_id,
|
|
target_database_id=payload.target_database_id,
|
|
status="DRAFT",
|
|
created_by=self.current_user,
|
|
)
|
|
self.db.add(job)
|
|
self.db.flush()
|
|
|
|
# Attach dictionaries
|
|
if payload.dictionary_ids:
|
|
for dict_id in payload.dictionary_ids:
|
|
assoc = TranslationJobDictionary(
|
|
id=str(uuid.uuid4()),
|
|
job_id=job.id,
|
|
dictionary_id=dict_id,
|
|
)
|
|
self.db.add(assoc)
|
|
|
|
self.db.commit()
|
|
self.db.refresh(job)
|
|
logger.info(f"[TranslateJobService] Created job '{job.id}'")
|
|
return job
|
|
# endregion create_job
|
|
|
|
# region update_job [TYPE Function]
|
|
# @PURPOSE: Update an existing translation job.
|
|
# @PRE: payload contains fields to update.
|
|
# @POST: Returns the updated TranslationJob.
|
|
# @SIDE_EFFECT: Re-detects database_dialect if source_datasource_id changed.
|
|
def update_job(self, job_id: str, payload: TranslateJobUpdate) -> TranslationJob:
|
|
logger.info(f"[TranslateJobService] Updating job '{job_id}'")
|
|
job = self.get_job(job_id)
|
|
|
|
update_data = payload.model_dump(exclude_unset=True)
|
|
dict_ids = update_data.pop("dictionary_ids", None)
|
|
|
|
# Backward compat: if only target_language is set (old API), wrap into target_languages
|
|
if "target_language" in update_data and "target_languages" not in update_data:
|
|
target_languages = [update_data["target_language"]] if update_data["target_language"] else []
|
|
update_data["target_languages"] = target_languages
|
|
elif "target_languages" in update_data and "target_language" not in update_data:
|
|
# Keep deprecated field in sync
|
|
if update_data["target_languages"]:
|
|
update_data["target_language"] = update_data["target_languages"][0]
|
|
|
|
# Validate BCP-47 for target_languages
|
|
if update_data.get("target_languages"):
|
|
if not isinstance(update_data["target_languages"], list):
|
|
update_data["target_languages"] = [str(update_data["target_languages"])]
|
|
for lang in update_data["target_languages"]:
|
|
_validate_bcp47(lang, "target_languages")
|
|
|
|
for field, value in update_data.items():
|
|
if hasattr(job, field):
|
|
setattr(job, field, value)
|
|
|
|
# Re-detect dialect if datasource changed
|
|
if payload.source_datasource_id and not payload.database_dialect:
|
|
try:
|
|
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,
|
|
self.config_manager,
|
|
)
|
|
job.database_dialect = detected_dialect
|
|
except Exception as e:
|
|
logger.warning(f"[TranslateJobService] Dialect re-detection failed: {e}")
|
|
|
|
job.updated_at = datetime.now(UTC)
|
|
|
|
# Update dictionary associations if provided
|
|
if dict_ids is not None:
|
|
self.db.query(TranslationJobDictionary).filter(
|
|
TranslationJobDictionary.job_id == job_id
|
|
).delete()
|
|
for dict_id in dict_ids:
|
|
assoc = TranslationJobDictionary(
|
|
id=str(uuid.uuid4()),
|
|
job_id=job_id,
|
|
dictionary_id=dict_id,
|
|
)
|
|
self.db.add(assoc)
|
|
|
|
self.db.commit()
|
|
self.db.refresh(job)
|
|
logger.info(f"[TranslateJobService] Updated job '{job_id}'")
|
|
return job
|
|
# endregion update_job
|
|
|
|
# region delete_job [TYPE Function]
|
|
# @PURPOSE: Delete a translation job and its associations.
|
|
# @PRE: job_id must exist.
|
|
# @POST: Job and all related records are deleted.
|
|
def delete_job(self, job_id: str) -> None:
|
|
logger.info(f"[TranslateJobService] Deleting job '{job_id}'")
|
|
job = self.get_job(job_id)
|
|
|
|
# Delete dictionary associations
|
|
self.db.query(TranslationJobDictionary).filter(
|
|
TranslationJobDictionary.job_id == job_id
|
|
).delete()
|
|
|
|
self.db.delete(job)
|
|
self.db.commit()
|
|
logger.info(f"[TranslateJobService] Deleted job '{job_id}'")
|
|
# endregion delete_job
|
|
|
|
# region duplicate_job [TYPE Function]
|
|
# @PURPOSE: Duplicate a translation job with a new name.
|
|
# @PRE: job_id must exist.
|
|
# @POST: Returns the duplicated TranslationJob with status DRAFT.
|
|
def duplicate_job(self, job_id: str, new_name: str | None = None) -> TranslationJob:
|
|
logger.info(f"[TranslateJobService] Duplicating job '{job_id}'")
|
|
source = self.get_job(job_id)
|
|
|
|
# Copy all fields except id, created_at, updated_at, status
|
|
new_job = TranslationJob(
|
|
id=str(uuid.uuid4()),
|
|
name=new_name or f"{source.name} (Copy)",
|
|
description=source.description,
|
|
source_dialect=source.source_dialect,
|
|
target_dialect=source.target_dialect,
|
|
database_dialect=source.database_dialect,
|
|
source_datasource_id=source.source_datasource_id,
|
|
source_table=source.source_table,
|
|
target_schema=source.target_schema,
|
|
target_table=source.target_table,
|
|
source_key_cols=source.source_key_cols,
|
|
target_key_cols=source.target_key_cols,
|
|
translation_column=source.translation_column,
|
|
target_column=source.target_column,
|
|
context_columns=source.context_columns,
|
|
target_language=source.target_language,
|
|
target_languages=source.target_languages,
|
|
provider_id=source.provider_id,
|
|
batch_size=source.batch_size,
|
|
upsert_strategy=source.upsert_strategy,
|
|
status="DRAFT",
|
|
created_by=self.current_user,
|
|
)
|
|
self.db.add(new_job)
|
|
self.db.flush()
|
|
|
|
# Copy dictionary associations
|
|
old_dicts = self.db.query(TranslationJobDictionary).filter(
|
|
TranslationJobDictionary.job_id == job_id
|
|
).all()
|
|
for assoc in old_dicts:
|
|
new_assoc = TranslationJobDictionary(
|
|
id=str(uuid.uuid4()),
|
|
job_id=new_job.id,
|
|
dictionary_id=assoc.dictionary_id,
|
|
)
|
|
self.db.add(new_assoc)
|
|
|
|
self.db.commit()
|
|
self.db.refresh(new_job)
|
|
logger.info(f"[TranslateJobService] Duplicated job '{job_id}' -> '{new_job.id}'")
|
|
return new_job
|
|
# endregion duplicate_job
|
|
|
|
# region get_job_dictionary_ids [TYPE Function]
|
|
# @PURPOSE: Get dictionary IDs attached to a job.
|
|
# @POST: Returns list of dictionary IDs.
|
|
def get_job_dictionary_ids(self, job_id: str) -> list[str]:
|
|
assocs = self.db.query(TranslationJobDictionary).filter(
|
|
TranslationJobDictionary.job_id == job_id
|
|
).all()
|
|
return [a.dictionary_id for a in assocs]
|
|
# endregion get_job_dictionary_ids
|
|
|
|
# region fetch_available_datasources [TYPE Function]
|
|
# @PURPOSE: List available Superset datasets for translation job creation.
|
|
def fetch_available_datasources(self, env_id: str, search: str | None = None) -> list:
|
|
"""List Superset datasets available for translation."""
|
|
from ...core.superset_client import SupersetClient
|
|
env = self.config_manager.get_environment(env_id)
|
|
if not env:
|
|
raise ValueError(f"Environment '{env_id}' not found")
|
|
client = SupersetClient(env)
|
|
_, datasets = client.get_datasets()
|
|
result = []
|
|
for ds in datasets:
|
|
name = ds.get("table_name", "")
|
|
if search and search.lower() not in name.lower():
|
|
continue
|
|
db_info = ds.get("database", {})
|
|
backend = db_info.get("backend", "")
|
|
dialect = _extract_dialect(backend)
|
|
result.append({
|
|
"id": ds.get("id"),
|
|
"table_name": name,
|
|
"schema": ds.get("schema"),
|
|
"database_name": db_info.get("database_name", "Unknown"),
|
|
"database_dialect": dialect,
|
|
"description": ds.get("description", ""),
|
|
})
|
|
return result
|
|
# endregion fetch_available_datasources
|
|
# #endregion TranslateJobService
|
|
|
|
|
|
# #region _extract_dialect [TYPE Function]
|
|
# @BRIEF Extract database dialect from Superset backend URI.
|
|
def _extract_dialect(backend: str) -> str:
|
|
"""Extract dialect name from a Superset database backend URI (e.g. 'postgresql+psycopg2://...')."""
|
|
if not backend:
|
|
return "unknown"
|
|
try:
|
|
scheme = backend.split("://")[0]
|
|
dialect = scheme.split("+")[0]
|
|
return dialect.lower()
|
|
except Exception:
|
|
return "unknown"
|
|
|
|
|
|
# #region job_to_response [TYPE Function]
|
|
# @BRIEF Convert a TranslationJob ORM model to a TranslateJobResponse schema with dictionary_ids.
|
|
def job_to_response(job: TranslationJob, dict_ids: list[str] | None = None) -> TranslateJobResponse:
|
|
return TranslateJobResponse(
|
|
id=job.id,
|
|
name=job.name,
|
|
description=job.description,
|
|
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 or [],
|
|
target_key_cols=job.target_key_cols or [],
|
|
translation_column=job.translation_column,
|
|
target_column=job.target_column,
|
|
context_columns=job.context_columns or [],
|
|
target_language=job.target_language,
|
|
target_languages=job.target_languages,
|
|
provider_id=job.provider_id,
|
|
batch_size=job.batch_size or 50,
|
|
upsert_strategy=job.upsert_strategy or "MERGE",
|
|
status=job.status,
|
|
created_by=job.created_by,
|
|
created_at=job.created_at,
|
|
updated_at=job.updated_at,
|
|
dictionary_ids=dict_ids or [],
|
|
environment_id=job.environment_id,
|
|
target_database_id=job.target_database_id,
|
|
)
|
|
# #endregion job_to_response
|
|
|
|
|
|
# #region DatasourceColumnsService [TYPE Function]
|
|
# @BRIEF Fetch datasource column metadata from Superset and return structured response.
|
|
# @PRE: datasource_id is a valid Superset dataset ID.
|
|
# @POST: Returns DatasourceColumnsResponse with column metadata and database dialect.
|
|
# @SIDE_EFFECT: Queries Superset API for dataset detail and database info.
|
|
def get_datasource_columns(
|
|
datasource_id: int,
|
|
env_id: str,
|
|
config_manager: ConfigManager,
|
|
) -> DatasourceColumnsResponse:
|
|
"""Fetch and return column metadata for a given Superset datasource."""
|
|
logger.info(f"[get_datasource_columns] Fetching columns for datasource {datasource_id}")
|
|
|
|
# Find environment config
|
|
environments = config_manager.get_environments()
|
|
env_config = next((e for e in environments if e.id == env_id), None)
|
|
if not env_config:
|
|
raise ValueError(f"Superset environment '{env_id}' not found")
|
|
|
|
# Create Superset client
|
|
client = SupersetClient(env_config)
|
|
dataset_detail = client.get_dataset_detail(datasource_id)
|
|
|
|
# Extract database dialect
|
|
database_info = dataset_detail.get("database", {})
|
|
dialect = None
|
|
if isinstance(database_info, dict):
|
|
try:
|
|
dialect = get_dialect_from_database(database_info)
|
|
except (ValueError, KeyError):
|
|
dialect = None
|
|
if dialect is None:
|
|
database_id = dataset_detail.get("database_id")
|
|
if database_id:
|
|
db_record = client.get_database(int(database_id))
|
|
db_result = db_record.get("result", db_record) if isinstance(db_record, dict) else db_record
|
|
dialect = get_dialect_from_database(db_result)
|
|
else:
|
|
raise ValueError("Could not determine database dialect for this datasource")
|
|
|
|
# Extract columns
|
|
raw_columns = dataset_detail.get("columns", [])
|
|
columns = []
|
|
for col in raw_columns:
|
|
col_name = col.get("name") or col.get("column_name")
|
|
if not col_name:
|
|
continue
|
|
columns.append(DatasourceColumnResponse(
|
|
name=str(col_name),
|
|
type=col.get("type"),
|
|
is_physical=col.get("is_physical", True),
|
|
is_dttm=col.get("is_dttm", False),
|
|
description=col.get("description", ""),
|
|
))
|
|
|
|
return DatasourceColumnsResponse(
|
|
datasource_id=datasource_id,
|
|
datasource_name=dataset_detail.get("table_name"),
|
|
schema_name=dataset_detail.get("schema"),
|
|
database_dialect=dialect,
|
|
columns=columns,
|
|
)
|
|
# #endregion DatasourceColumnsService
|
|
|
|
|
|
# #region InlineCorrectionService [C:3] [TYPE Class] [SEMANTICS translate, correction, inline, dictionary]
|
|
# @BRIEF Service for inline editing translated values and submitting corrections to dictionaries.
|
|
class InlineCorrectionService:
|
|
"""Handle inline correction of translated values with optional dictionary submission."""
|
|
|
|
@staticmethod
|
|
def apply_inline_edit(
|
|
db: Session,
|
|
run_id: str,
|
|
record_id: str,
|
|
language_code: str,
|
|
final_value: str,
|
|
submit_to_dictionary: bool = False,
|
|
dictionary_id: str | None = None,
|
|
current_user: str | None = None,
|
|
context_data_override: dict | None = None,
|
|
usage_notes: str | None = None,
|
|
keep_context: bool = True,
|
|
) -> dict:
|
|
"""Apply an inline edit to a TranslationLanguage entry.
|
|
|
|
Updates final_value and user_edit fields. Optionally submits to dictionary.
|
|
Returns the updated TranslationLanguage data as a dict.
|
|
|
|
Args:
|
|
context_data_override: If provided, overrides auto-captured context data.
|
|
usage_notes: Usage notes for the dictionary entry.
|
|
keep_context: If False, clear context on the dictionary entry.
|
|
"""
|
|
from ...models.translate import TranslationLanguage, TranslationRecord
|
|
|
|
# Find the language entry
|
|
lang_entry = (
|
|
db.query(TranslationLanguage)
|
|
.filter(
|
|
TranslationLanguage.record_id == record_id,
|
|
TranslationLanguage.language_code == language_code,
|
|
)
|
|
.first()
|
|
)
|
|
if not lang_entry:
|
|
raise ValueError(f"Language entry '{language_code}' not found for record '{record_id}'")
|
|
|
|
# Verify the record belongs to the run
|
|
record = (
|
|
db.query(TranslationRecord)
|
|
.filter(
|
|
TranslationRecord.id == record_id,
|
|
TranslationRecord.run_id == run_id,
|
|
)
|
|
.first()
|
|
)
|
|
if not record:
|
|
raise ValueError(
|
|
f"Translation record '{record_id}' not found in run '{run_id}'"
|
|
)
|
|
|
|
# Apply the edit
|
|
lang_entry.final_value = final_value
|
|
lang_entry.user_edit = final_value
|
|
if lang_entry.status in ("pending", "translated"):
|
|
lang_entry.status = "edited"
|
|
db.flush()
|
|
|
|
# Optionally submit to dictionary
|
|
dict_result = None
|
|
if submit_to_dictionary and dictionary_id:
|
|
try:
|
|
dict_result = InlineCorrectionService.submit_correction_to_dict(
|
|
db=db,
|
|
record_id=record_id,
|
|
language_code=language_code,
|
|
dictionary_id=dictionary_id,
|
|
corrected_value=final_value,
|
|
current_user=current_user,
|
|
context_data_override=context_data_override,
|
|
usage_notes=usage_notes,
|
|
keep_context=keep_context,
|
|
)
|
|
except Exception as e:
|
|
# Dictionary submission failure should not block the edit
|
|
dict_result = {"error": str(e), "action": "failed"}
|
|
|
|
db.commit()
|
|
db.refresh(lang_entry)
|
|
|
|
from ...schemas.translate import TranslationLanguageResponse
|
|
response = TranslationLanguageResponse.model_validate(lang_entry)
|
|
result = response.model_dump()
|
|
if dict_result:
|
|
result["dictionary_result"] = dict_result
|
|
return result
|
|
|
|
@staticmethod
|
|
def submit_correction_to_dict(
|
|
db: Session,
|
|
record_id: str,
|
|
language_code: str,
|
|
dictionary_id: str,
|
|
corrected_value: str,
|
|
current_user: str | None = None,
|
|
context_data_override: dict | None = None,
|
|
usage_notes: str | None = None,
|
|
keep_context: bool = True,
|
|
) -> dict:
|
|
"""Submit a correction from an inline edit to the dictionary.
|
|
|
|
Fetches the TranslationLanguage entry, auto-captures source text and context,
|
|
and creates/updates a DictionaryEntry with correct language pair.
|
|
Returns conflict info if an entry already exists.
|
|
|
|
Args:
|
|
context_data_override: If provided, overrides auto-captured context_data.
|
|
usage_notes: Optional usage notes to store on the dictionary entry.
|
|
keep_context: If False, sets has_context=False and clears context_data.
|
|
"""
|
|
from ...core.logger import belief_scope, logger
|
|
from ...models.translate import DictionaryEntry, TranslationLanguage, TranslationRecord
|
|
from ._utils import _normalize_term
|
|
|
|
with belief_scope("InlineCorrectionService.submit_correction_to_dict"):
|
|
# Find the language entry
|
|
lang_entry = (
|
|
db.query(TranslationLanguage)
|
|
.filter(
|
|
TranslationLanguage.record_id == record_id,
|
|
TranslationLanguage.language_code == language_code,
|
|
)
|
|
.first()
|
|
)
|
|
if not lang_entry:
|
|
raise ValueError(f"Language entry '{language_code}' not found for record '{record_id}'")
|
|
|
|
# Find the record for source text and context
|
|
record = (
|
|
db.query(TranslationRecord)
|
|
.filter(TranslationRecord.id == record_id)
|
|
.first()
|
|
)
|
|
if not record:
|
|
raise ValueError(f"Translation record '{record_id}' not found")
|
|
|
|
# Get source text from the record
|
|
source_term = record.source_sql or record.source_object_name or ""
|
|
if not source_term:
|
|
raise ValueError("No source text available for this record")
|
|
|
|
# Determine language pair
|
|
source_language = lang_entry.source_language_detected or "und"
|
|
target_language = language_code
|
|
|
|
# Prepare context data — auto-capture from source row
|
|
context_data = {}
|
|
if record.source_data:
|
|
context_data["source_data"] = record.source_data
|
|
if record.source_object_type:
|
|
context_data["source_object_type"] = record.source_object_type
|
|
if record.source_object_name:
|
|
context_data["source_object_name"] = record.source_object_name
|
|
if record.source_object_id:
|
|
context_data["source_object_id"] = record.source_object_id
|
|
|
|
# Apply context_data_override if provided (user edited)
|
|
if context_data_override is not None:
|
|
context_data = context_data_override
|
|
context_source = "manual"
|
|
else:
|
|
context_source = "auto"
|
|
|
|
# Handle keep_context=False (user explicitly removed context)
|
|
if not keep_context:
|
|
context_data = None
|
|
context_source = "manual"
|
|
|
|
# Normalize source term
|
|
normalized = _normalize_term(source_term)
|
|
|
|
# Check for existing entry with same (dictionary_id, source_term_norm, source_language, target_language)
|
|
existing = (
|
|
db.query(DictionaryEntry)
|
|
.filter(
|
|
DictionaryEntry.dictionary_id == dictionary_id,
|
|
DictionaryEntry.source_term_normalized == normalized,
|
|
DictionaryEntry.source_language == source_language,
|
|
DictionaryEntry.target_language == target_language,
|
|
)
|
|
.first()
|
|
)
|
|
|
|
result = {
|
|
"action": "created",
|
|
"entry_id": None,
|
|
"conflict": None,
|
|
"message": None,
|
|
}
|
|
|
|
if existing:
|
|
# Conflict: return conflict info
|
|
result["action"] = "conflict_detected"
|
|
result["conflict"] = {
|
|
"source_term": source_term,
|
|
"existing_target_term": existing.target_term,
|
|
"submitted_target_term": corrected_value,
|
|
"action": "keep_existing",
|
|
}
|
|
result["message"] = (
|
|
f"Existing entry found: '{existing.target_term}' "
|
|
f"for source '{source_term}' ({source_language} → {target_language})"
|
|
)
|
|
logger.reason("Correction conflict detected", result)
|
|
return result
|
|
|
|
# Create new entry
|
|
entry = DictionaryEntry(
|
|
dictionary_id=dictionary_id,
|
|
source_term=source_term.strip(),
|
|
source_term_normalized=normalized,
|
|
target_term=corrected_value.strip(),
|
|
source_language=source_language,
|
|
target_language=target_language,
|
|
context_data=context_data if context_data else None,
|
|
context_notes=None,
|
|
has_context=bool(context_data) and keep_context,
|
|
context_source=context_source,
|
|
usage_notes=usage_notes,
|
|
origin_source_language=source_language,
|
|
origin_run_id=record.run_id,
|
|
origin_row_key=record.id,
|
|
origin_user_id=current_user,
|
|
)
|
|
db.add(entry)
|
|
db.flush()
|
|
result["entry_id"] = entry.id
|
|
result["message"] = (
|
|
f"Entry created for '{source_term}' ({source_language}) → "
|
|
f"'{corrected_value}' ({target_language})"
|
|
)
|
|
logger.reflect("Dictionary entry created from correction", result)
|
|
return result
|
|
# #endregion InlineCorrectionService
|
|
|
|
|
|
# #region BulkFindReplaceService [C:3] [TYPE Class] [SEMANTICS translate, bulk, find, replace, regex]
|
|
# @BRIEF Service for bulk find-and-replace operations on translated values.
|
|
class BulkFindReplaceService:
|
|
"""Handle bulk find-and-replace on TranslationLanguage entries."""
|
|
|
|
MAX_PATTERN_LENGTH = 500
|
|
|
|
@staticmethod
|
|
def _compile_pattern(pattern: str, is_regex: bool) -> re.Pattern:
|
|
"""Compile a search pattern (regex or plain text)."""
|
|
if len(pattern) > BulkFindReplaceService.MAX_PATTERN_LENGTH:
|
|
raise ValueError(
|
|
f"Pattern too long (max {BulkFindReplaceService.MAX_PATTERN_LENGTH} characters)"
|
|
)
|
|
try:
|
|
return re.compile(pattern) if is_regex else re.compile(re.escape(pattern))
|
|
except re.error as e:
|
|
raise ValueError(f"Invalid regex pattern: {e}")
|
|
|
|
@staticmethod
|
|
def _find_matching_entries(
|
|
db: Session,
|
|
run_id: str,
|
|
pattern: str,
|
|
is_regex: bool,
|
|
target_language: str,
|
|
) -> list[Any]:
|
|
"""Find all TranslationLanguage entries matching the pattern."""
|
|
from ...models.translate import TranslationLanguage, TranslationRecord
|
|
|
|
# We scan entries for the target language in this run
|
|
entries = (
|
|
db.query(TranslationLanguage)
|
|
.join(
|
|
TranslationRecord,
|
|
TranslationRecord.id == TranslationLanguage.record_id,
|
|
)
|
|
.filter(
|
|
TranslationRecord.run_id == run_id,
|
|
TranslationLanguage.language_code == target_language,
|
|
)
|
|
.all()
|
|
)
|
|
return entries
|
|
|
|
@staticmethod
|
|
def preview(
|
|
db: Session,
|
|
run_id: str,
|
|
pattern: str,
|
|
is_regex: bool,
|
|
replacement_text: str,
|
|
target_language: str,
|
|
) -> list[dict]:
|
|
"""Scan TranslationLanguage entries and return matching items without applying changes."""
|
|
from ...models.translate import TranslationRecord
|
|
|
|
compiled = BulkFindReplaceService._compile_pattern(pattern, is_regex)
|
|
entries = BulkFindReplaceService._find_matching_entries(
|
|
db, run_id, pattern, is_regex, target_language
|
|
)
|
|
|
|
preview_items = []
|
|
for entry in entries:
|
|
current_value = entry.final_value or entry.translated_value or ""
|
|
if compiled.search(current_value):
|
|
# Get source term for context
|
|
record = (
|
|
db.query(TranslationRecord)
|
|
.filter(TranslationRecord.id == entry.record_id)
|
|
.first()
|
|
)
|
|
source_term = record.source_sql if record else ""
|
|
|
|
new_value = compiled.sub(
|
|
replacement_text,
|
|
current_value,
|
|
)
|
|
if new_value != current_value:
|
|
preview_items.append({
|
|
"record_id": entry.record_id,
|
|
"language_code": entry.language_code,
|
|
"source_term": source_term or "",
|
|
"current_value": current_value,
|
|
"new_value": new_value,
|
|
"source_language_detected": entry.source_language_detected,
|
|
})
|
|
|
|
return preview_items
|
|
|
|
@staticmethod
|
|
def _get_replacement(replacement_text: str, is_regex: bool) -> str:
|
|
"""Return replacement pattern (for regex, use backrefs; for plain, use literal)."""
|
|
return replacement_text
|
|
|
|
@staticmethod
|
|
def apply(
|
|
db: Session,
|
|
run_id: str,
|
|
pattern: str,
|
|
is_regex: bool,
|
|
replacement_text: str,
|
|
target_language: str,
|
|
submit_to_dictionary: bool = False,
|
|
dictionary_id: str | None = None,
|
|
usage_notes: str | None = None,
|
|
current_user: str | None = None,
|
|
submit_to_dictionary_with_context: bool = False,
|
|
) -> dict:
|
|
"""Apply find-and-replace to matching TranslationLanguage entries.
|
|
|
|
Returns dict with rows_affected, corrections_submitted, and preview list.
|
|
|
|
Args:
|
|
submit_to_dictionary_with_context: If True, auto-capture source row context
|
|
when submitting to dictionary.
|
|
"""
|
|
from ...core.logger import belief_scope, logger
|
|
from ...models.translate import DictionaryEntry, TranslationRecord
|
|
|
|
with belief_scope("BulkFindReplaceService.apply"):
|
|
compiled = BulkFindReplaceService._compile_pattern(pattern, is_regex)
|
|
entries = BulkFindReplaceService._find_matching_entries(
|
|
db, run_id, pattern, is_regex, target_language
|
|
)
|
|
|
|
preview_items = []
|
|
rows_affected = 0
|
|
corrections_submitted = 0
|
|
|
|
# Track unique (source_term, replacement) -> first record for context capture
|
|
seen_term_replacements: dict[str, str] = {}
|
|
|
|
for entry in entries:
|
|
current_value = entry.final_value or entry.translated_value or ""
|
|
if compiled.search(current_value):
|
|
new_value = compiled.sub(replacement_text, current_value)
|
|
if new_value != current_value:
|
|
# Apply the replacement
|
|
entry.final_value = new_value
|
|
entry.user_edit = new_value
|
|
if entry.status in ("pending", "translated"):
|
|
entry.status = "edited"
|
|
rows_affected += 1
|
|
|
|
# Get source term for preview
|
|
record = (
|
|
db.query(TranslationRecord)
|
|
.filter(TranslationRecord.id == entry.record_id)
|
|
.first()
|
|
)
|
|
source_term = record.source_sql if record else ""
|
|
|
|
# Optionally submit to dictionary
|
|
if submit_to_dictionary and dictionary_id:
|
|
try:
|
|
dict_result = InlineCorrectionService.submit_correction_to_dict(
|
|
db=db,
|
|
record_id=entry.record_id,
|
|
language_code=entry.language_code,
|
|
dictionary_id=dictionary_id,
|
|
corrected_value=new_value,
|
|
current_user=current_user,
|
|
)
|
|
if dict_result.get("action") in ("created", "updated"):
|
|
corrections_submitted += 1
|
|
|
|
# Track context for unique (source_term, replacement) pairs
|
|
if submit_to_dictionary_with_context and record:
|
|
term_key = f"{source_term}::{new_value}"
|
|
if term_key not in seen_term_replacements:
|
|
seen_term_replacements[term_key] = entry.record_id
|
|
# Auto-capture context from this row
|
|
context_data = {}
|
|
if record.source_data:
|
|
context_data["source_data"] = record.source_data
|
|
if record.source_object_type:
|
|
context_data["source_object_type"] = record.source_object_type
|
|
if record.source_object_name:
|
|
context_data["source_object_name"] = record.source_object_name
|
|
if record.source_object_id:
|
|
context_data["source_object_id"] = record.source_object_id
|
|
# Update the just-created entry's context
|
|
if dict_result.get("action") == "created" and dict_result.get("entry_id"):
|
|
d_entry = (
|
|
db.query(DictionaryEntry)
|
|
.filter(DictionaryEntry.id == dict_result["entry_id"])
|
|
.first()
|
|
)
|
|
if d_entry:
|
|
d_entry.context_data = context_data if context_data else None
|
|
d_entry.has_context = bool(context_data)
|
|
d_entry.context_source = "bulk"
|
|
except Exception as e:
|
|
logger.explore(
|
|
"Bulk replace: dictionary submission failed",
|
|
extra={"record_id": entry.record_id, "error": str(e)},
|
|
)
|
|
|
|
preview_items.append({
|
|
"record_id": entry.record_id,
|
|
"language_code": entry.language_code,
|
|
"source_term": source_term or "",
|
|
"current_value": current_value,
|
|
"new_value": new_value,
|
|
"source_language_detected": entry.source_language_detected,
|
|
})
|
|
|
|
db.commit()
|
|
|
|
result = {
|
|
"rows_affected": rows_affected,
|
|
"corrections_submitted": corrections_submitted,
|
|
"preview": preview_items,
|
|
}
|
|
logger.reason("Bulk replace completed", result)
|
|
return result
|
|
# #endregion BulkFindReplaceService
|
|
|
|
|
|
# #endregion TranslateJobService
|
|
# #endregion TranslateJobService
|