busya
ce0bdd31ef
1. dict_snapshot_hash now includes entry count + max(updated_at) per dictionary. Previously only hashed dictionary IDs, meaning edits to dictionary entries did NOT invalidate the translation cache. Stale cached translations could be served after editing dictionary entries. (HIGH severity) 2. _batch_insert.py now uses SQLGenerator.generate_batch() with 500-row chunking instead of a single massive INSERT statement. Prevents potential SQL size limit issues with large batches or many target languages. (LOW severity) 3. Fixed same bug in preview_response_parser.py — compute_dict_snapshot_hash had identical ID-only hash flaw. Tests: 69/69 translate tests pass.
247 lines
10 KiB
Python
247 lines
10 KiB
Python
# #region BatchInsertService [C:3] [TYPE Module] [SEMANTICS translate, insert, sqllab, upsert, target]
|
|
# @BRIEF Insert successful translation records into target table via Superset SQL Lab.
|
|
# Builds INSERT SQL (original + per-language rows), resolves backend dialect,
|
|
# and executes via SupersetSqlLabExecutor.
|
|
# @LAYER Infrastructure
|
|
# @RELATION DEPENDS_ON -> [TranslationRecord], [TranslationLanguage]
|
|
# @RELATION DEPENDS_ON -> [SQLGenerator], [SupersetSqlLabExecutor]
|
|
# @RELATION DEPENDS_ON -> [ConfigManager]
|
|
# @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.
|
|
# @SIDE_EFFECT HTTP call to Superset SQL Lab API. Writes to target database.
|
|
# @RATIONALE Extracted from BatchProcessingService (583 lines) to comply with INV_7.
|
|
# @REJECTED Inline insert logic in process_batch — made the function 583 lines.
|
|
|
|
import json
|
|
|
|
from sqlalchemy.orm import Session, joinedload
|
|
|
|
from ...core.config_manager import ConfigManager
|
|
from ...core.logger import belief_scope, logger
|
|
from ...models.translate import TranslationJob, TranslationRecord
|
|
from .sql_generator import SQLGenerator, _normalize_timestamp_value
|
|
from .superset_executor import SupersetSqlLabExecutor
|
|
|
|
|
|
# #region insert_batch_to_target [C:3] [TYPE Function] [SEMANTICS translate, insert, orchestrate]
|
|
# @BRIEF Insert successful records from a single batch into the target table.
|
|
def insert_batch_to_target(
|
|
db: Session, config_manager: ConfigManager,
|
|
job: TranslationJob, batch_id: str, run_id: str,
|
|
) -> None:
|
|
"""Insert successful batch records into the target table via Superset SQL Lab."""
|
|
with belief_scope("BatchInsertService.insert_batch_to_target"):
|
|
records = _fetch_batch_records(db, batch_id)
|
|
if not records:
|
|
return
|
|
|
|
effective_target = job.target_column or job.translation_column
|
|
primary_language = (job.target_languages or ["en"])[0]
|
|
columns = _build_target_columns(job, effective_target)
|
|
context_keys = _build_context_keys(job, effective_target)
|
|
rows_for_sql = _build_insert_rows(records, job, effective_target, primary_language, context_keys)
|
|
|
|
if not columns:
|
|
columns = [effective_target or "translated_text"]
|
|
rows_for_sql = [{columns[0]: rec.target_sql or ""} for rec in records]
|
|
|
|
dialect, executor = _resolve_insert_backend(config_manager, job, batch_id)
|
|
if dialect is None:
|
|
return
|
|
|
|
# Use chunked INSERT to avoid SQL size limits (max 500 VALUES rows per statement)
|
|
statements = SQLGenerator.generate_batch(
|
|
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",
|
|
max_rows_per_statement=500,
|
|
)
|
|
total_inserted = 0
|
|
for sql, chunk_count in statements:
|
|
if sql is None:
|
|
continue
|
|
_execute_insert_sql(executor, sql, batch_id, chunk_count)
|
|
total_inserted += chunk_count
|
|
logger.reason(f"Batch {batch_id[:12]} inserted {total_inserted} rows",
|
|
{"batch_id": batch_id, "rows": total_inserted, "chunks": len(statements)})
|
|
# #endregion insert_batch_to_target
|
|
|
|
|
|
# #region _fetch_batch_records [C:1] [TYPE Function]
|
|
def _fetch_batch_records(db: Session, batch_id: str) -> list[TranslationRecord]:
|
|
"""Fetch successful TranslationRecords for a batch with languages."""
|
|
return (
|
|
db.query(TranslationRecord)
|
|
.options(joinedload(TranslationRecord.languages))
|
|
.filter(
|
|
TranslationRecord.batch_id == batch_id,
|
|
TranslationRecord.status == "SUCCESS",
|
|
TranslationRecord.target_sql.isnot(None),
|
|
)
|
|
.all()
|
|
)
|
|
# #endregion _fetch_batch_records
|
|
|
|
|
|
# #region _build_target_columns [C:1] [TYPE Function]
|
|
def _build_target_columns(job: TranslationJob, effective_target: str | None) -> list[str]:
|
|
"""Build the list of column names for the target INSERT."""
|
|
columns: list[str] = []
|
|
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")
|
|
seen: set[str] = set()
|
|
deduped: list[str] = []
|
|
for c in columns:
|
|
if c and c not in seen:
|
|
deduped.append(c)
|
|
seen.add(c)
|
|
return deduped
|
|
# #endregion _build_target_columns
|
|
|
|
|
|
# #region _build_context_keys [C:1] [TYPE Function]
|
|
def _build_context_keys(job: TranslationJob, effective_target: str | None) -> list[str]:
|
|
"""Build context keys for JSON bundled 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)
|
|
return context_keys
|
|
# #endregion _build_context_keys
|
|
|
|
|
|
# #region _build_insert_rows [C:3] [TYPE Function]
|
|
def _build_insert_rows(
|
|
records: list[TranslationRecord], job: TranslationJob,
|
|
effective_target: str | None, primary_language: str, context_keys: list[str],
|
|
) -> list[dict[str, object]]:
|
|
"""Build row dicts for the INSERT SQL statement."""
|
|
rows_for_sql: list[dict[str, object]] = []
|
|
for rec in records:
|
|
source_data = rec.source_data or {}
|
|
detected_src_lang = "und"
|
|
if rec.languages and len(rec.languages) > 0:
|
|
detected_src_lang = rec.languages[0].source_language_detected or "und"
|
|
|
|
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 ""
|
|
|
|
base_row: dict[str, object] = {}
|
|
if job.target_key_cols:
|
|
for k in job.target_key_cols:
|
|
raw = source_data.get(k)
|
|
if raw is not None:
|
|
normalized = _normalize_timestamp_value(raw)
|
|
base_row[k] = normalized if normalized else raw
|
|
else:
|
|
base_row[k] = None
|
|
if job.target_source_column:
|
|
base_row[job.target_source_column] = rec.source_sql or ""
|
|
if job.target_source_language_column:
|
|
base_row[job.target_source_language_column] = detected_src_lang
|
|
base_row["context"] = json.dumps(context_data, ensure_ascii=False)
|
|
|
|
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)
|
|
|
|
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 = dict(base_row)
|
|
if effective_target:
|
|
fallback[effective_target] = rec.target_sql or ""
|
|
if job.target_language_column:
|
|
fallback[job.target_language_column] = primary_language
|
|
fallback["is_original"] = 0
|
|
rows_for_sql.append(fallback)
|
|
return rows_for_sql
|
|
# #endregion _build_insert_rows
|
|
|
|
|
|
# #region _resolve_insert_backend [C:1] [TYPE Function]
|
|
def _resolve_insert_backend(
|
|
config_manager: ConfigManager, job: TranslationJob, batch_id: str,
|
|
) -> tuple[str | None, SupersetSqlLabExecutor | None]:
|
|
"""Resolve the database backend and SQL executor for batch insert."""
|
|
try:
|
|
env_id = job.environment_id or job.source_dialect or ""
|
|
executor = SupersetSqlLabExecutor(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)})
|
|
return None, None
|
|
dialect = real_backend or job.database_dialect or job.target_dialect or "postgresql"
|
|
return dialect, executor
|
|
# #endregion _resolve_insert_backend
|
|
|
|
|
|
# #region _generate_insert_sql [C:1] [TYPE Function]
|
|
def _generate_insert_sql(
|
|
dialect: str, job: TranslationJob, columns: list[str],
|
|
rows_for_sql: list[dict[str, object]], batch_id: str = "",
|
|
) -> tuple[str | None, int]:
|
|
"""Generate INSERT SQL using SQLGenerator."""
|
|
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",
|
|
)
|
|
return sql, row_count
|
|
except ValueError as e:
|
|
logger.explore("SQL generation failed for batch", {"batch_id": batch_id, "error": str(e)})
|
|
return None, 0
|
|
# #endregion _generate_insert_sql
|
|
|
|
|
|
# #region _execute_insert_sql [C:1] [TYPE Function]
|
|
def _execute_insert_sql(executor: SupersetSqlLabExecutor, sql: str, batch_id: str, row_count: int) -> None:
|
|
"""Execute the INSERT SQL via Superset SQL Lab."""
|
|
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 _execute_insert_sql
|
|
# #endregion BatchInsertService
|