ss-tools/backend/src/models/translate.py

# #region TranslateModels [C:3] [TYPE Module] [SEMANTICS sqlalchemy, translate, model, schema, dashboard, llm]
# @BRIEF SQLAlchemy ORM models for LLM-based SQL/dashboard translation across dialects.
# @LAYER: Domain
# @RELATION INHERITS_FROM -> MappingModels:Base

import uuid
from datetime import UTC, datetime

from sqlalchemy import JSON, Boolean, Column, DateTime, Float, ForeignKey, Index, Integer, String, Text, UniqueConstraint
from sqlalchemy.orm import relationship

from .mapping import Base


def generate_uuid():
    return str(uuid.uuid4())


# #region TranslationJob [TYPE Class]
# @BRIEF A translation job representing a multi-dialect conversion task with column mappings, LLM config, and dictionary attachments.
# @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.
class TranslationJob(Base):
    __tablename__ = "translation_jobs"

    id = Column(String, primary_key=True, default=generate_uuid)
    name = Column(String, nullable=False)
    description = Column(Text, nullable=True)
    source_dialect = Column(String, nullable=False)
    target_dialect = Column(String, nullable=False)
    database_dialect = Column(String, nullable=True, comment="Detected dialect from Superset connection at save time")
    status = Column(String, nullable=False, default="DRAFT")  # DRAFT, READY, RUNNING, COMPLETED, FAILED, CANCELLED

    # Datasource & target table configuration
    source_datasource_id = Column(String, nullable=True, comment="Superset datasource ID")
    source_table = Column(String, nullable=True, comment="Source table name resolved from datasource")
    target_schema = Column(String, nullable=True, comment="Target table schema")
    target_table = Column(String, nullable=True, comment="Target table name")

    # Column mapping
    source_key_cols = Column(JSON, nullable=True, comment="Source key column names for composite key")
    target_key_cols = Column(JSON, nullable=True, comment="Target key column names for composite key")
    translation_column = Column(String, nullable=True, comment="Source column whose values will be translated")
    target_column = Column(String, nullable=True, comment="Target column for translated output (defaults to translation_column)")
    target_language_column = Column(String, nullable=True, comment="Target column for language code (e.g. 'ru', 'en')")
    target_source_column = Column(String, nullable=True, comment="Target column for source/original text")
    target_source_language_column = Column(String, nullable=True, comment="Target column for detected source language (BCP-47)")
    context_columns = Column(JSON, nullable=True, comment="Context column names included in LLM prompt")

    # LLM & processing settings
    # source_language removed — deprecated, auto-detected per row by LLM; use TranslationLanguage.source_language_detected instead
    target_languages = Column(JSON, nullable=True, comment="List of BCP-47 target language codes (multi-language support)")
    provider_id = Column(String, nullable=True, comment="LLM provider ID")
    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")
    target_database_id = Column(String, nullable=True, comment="Superset database ID for SQL Lab insert target")

    created_by = Column(String, nullable=True)
    created_at = Column(DateTime, default=lambda: datetime.now(UTC))
    updated_at = Column(DateTime, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
# #endregion TranslationJob


# #region TranslationRun [TYPE Class]
# @BRIEF Represents a single execution of a translation job, capturing status, timing, and Superset execution metadata.
class TranslationRun(Base):
    __tablename__ = "translation_runs"

    id = Column(String, primary_key=True, default=generate_uuid)
    job_id = Column(String, ForeignKey("translation_jobs.id"), nullable=False, index=True)
    status = Column(String, nullable=False, default="PENDING")  # PENDING, RUNNING, COMPLETED, FAILED, CANCELLED
    trigger_type = Column(String, nullable=True, comment="manual, scheduled, retry, baseline_expired")
    started_at = Column(DateTime, nullable=True)
    completed_at = Column(DateTime, nullable=True)
    error_message = Column(Text, nullable=True)
    total_records = Column(Integer, default=0)
    successful_records = Column(Integer, default=0)
    failed_records = Column(Integer, default=0)
    skipped_records = Column(Integer, default=0)
    insert_status = Column(String, nullable=True, comment="Status of the Superset insert/update operation")
    superset_execution_id = Column(String, nullable=True, comment="Superset execution/task ID")
    superset_execution_log = Column(JSON, nullable=True, comment="Superset execution log output")
    config_snapshot = Column(JSON, nullable=True, comment="Snapshot of job config at run creation time")
    key_hash = Column(String, nullable=True, comment="Hash of source key fields for dedup")
    config_hash = Column(String, nullable=True, comment="Hash of translation configuration state")
    dict_snapshot_hash = Column(String, nullable=True, comment="Hash of dictionary state at run time")
    created_by = Column(String, nullable=True)
    created_at = Column(DateTime, default=lambda: datetime.now(UTC))

    language_stats = relationship("TranslationRunLanguageStats", back_populates="run")
# #endregion TranslationRun


# #region TranslationBatch [TYPE Class]
# @BRIEF Groups translation records within a run into manageable batches with timing and record counts.
class TranslationBatch(Base):
    __tablename__ = "translation_batches"

    id = Column(String, primary_key=True, default=generate_uuid)
    run_id = Column(String, ForeignKey("translation_runs.id"), nullable=False, index=True)
    batch_index = Column(Integer, nullable=False)
    status = Column(String, nullable=False, default="PENDING")
    total_records = Column(Integer, default=0)
    successful_records = Column(Integer, default=0)
    failed_records = Column(Integer, default=0)
    started_at = Column(DateTime, nullable=True)
    completed_at = Column(DateTime, nullable=True)
    created_at = Column(DateTime, default=lambda: datetime.now(UTC))
# #endregion TranslationBatch


# #region TranslationRecord [TYPE Class]
# @BRIEF Individual translation result for a single SQL statement or dashboard element, tracking source, target, and error state.
class TranslationRecord(Base):
    __tablename__ = "translation_records"

    id = Column(String, primary_key=True, default=generate_uuid)
    batch_id = Column(String, ForeignKey("translation_batches.id"), nullable=False, index=True)
    run_id = Column(String, ForeignKey("translation_runs.id"), nullable=False, index=True)
    source_sql = Column(Text, nullable=True)
    target_sql = Column(Text, nullable=True)
    source_object_type = Column(String, nullable=True)  # query, dashboard, chart, dataset
    source_object_id = Column(String, nullable=True)
    source_object_name = Column(String, nullable=True)
    source_data = Column(JSON, nullable=True, comment="Original source row key columns for upsert matching")
    status = Column(String, nullable=False, default="PENDING")  # PENDING, SUCCESS, FAILED, SKIPPED
    error_message = Column(Text, nullable=True)
    token_count_input = Column(Integer, nullable=True)
    token_count_output = Column(Integer, nullable=True)
    translation_duration_ms = Column(Integer, nullable=True)
    created_at = Column(DateTime, default=lambda: datetime.now(UTC))

    languages = relationship("TranslationLanguage", back_populates="record")

    __table_args__ = (
        Index("ix_translation_records_run_status", "run_id", "status"),
    )
# #endregion TranslationRecord


# #region TranslationEvent [TYPE Class]
# @BRIEF Audit/event log for translation operations, with optional run_id for context.
class TranslationEvent(Base):
    __tablename__ = "translation_events"

    id = Column(String, primary_key=True, default=generate_uuid)
    job_id = Column(String, ForeignKey("translation_jobs.id"), nullable=False, index=True)
    run_id = Column(String, ForeignKey("translation_runs.id"), nullable=True, index=True)
    event_type = Column(String, nullable=False)  # JOB_CREATED, JOB_STARTED, JOB_COMPLETED, JOB_FAILED, etc.
    event_data = Column(JSON, nullable=True)
    created_by = Column(String, nullable=True)
    created_at = Column(DateTime, default=lambda: datetime.now(UTC))
# #endregion TranslationEvent


# #region TranslationPreviewSession [TYPE Class]
# @BRIEF A preview session allowing users to review proposed translations before applying them.
class TranslationPreviewSession(Base):
    __tablename__ = "translation_preview_sessions"

    id = Column(String, primary_key=True, default=generate_uuid)
    job_id = Column(String, ForeignKey("translation_jobs.id"), nullable=False, index=True)
    run_id = Column(String, ForeignKey("translation_runs.id"), nullable=True)
    status = Column(String, nullable=False, default="ACTIVE")  # ACTIVE, APPLIED, DISCARDED, EXPIRED
    created_by = Column(String, nullable=True)
    created_at = Column(DateTime, default=lambda: datetime.now(UTC))
    expires_at = Column(DateTime, nullable=True)
# #endregion TranslationPreviewSession


# #region TranslationPreviewRecord [TYPE Class]
# @BRIEF Individual preview entry within a preview session, showing original and translated content side by side.
class TranslationPreviewRecord(Base):
    __tablename__ = "translation_preview_records"

    id = Column(String, primary_key=True, default=generate_uuid)
    session_id = Column(String, ForeignKey("translation_preview_sessions.id"), nullable=False, index=True)
    source_sql = Column(Text, nullable=True)
    target_sql = Column(Text, nullable=True)
    source_object_type = Column(String, nullable=True)
    source_object_id = Column(String, nullable=True)
    source_object_name = Column(String, nullable=True)
    source_data = Column(JSON, nullable=True, comment="Original source row key columns for upsert matching")
    status = Column(String, nullable=False, default="PENDING")  # PENDING, APPROVED, REJECTED
    feedback = Column(Text, nullable=True)
    created_at = Column(DateTime, default=lambda: datetime.now(UTC))

    languages = relationship("TranslationPreviewLanguage", back_populates="preview_record")
# #endregion TranslationPreviewRecord


# #region TerminologyDictionary [TYPE Class]
# @BRIEF A named collection of terminology mappings used during translation to ensure consistent term translation.
class TerminologyDictionary(Base):
    __tablename__ = "terminology_dictionaries"

    id = Column(String, primary_key=True, default=generate_uuid)
    name = Column(String, nullable=False)
    description = Column(Text, nullable=True)
    source_dialect = Column(String, nullable=False)
    target_dialect = Column(String, nullable=False)
    is_active = Column(Boolean, default=True)
    created_by = Column(String, nullable=True)
    created_at = Column(DateTime, default=lambda: datetime.now(UTC))
    updated_at = Column(DateTime, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
# #endregion TerminologyDictionary


# #region DictionaryEntry [TYPE Class]
# @BRIEF A single term mapping entry with case-insensitive unique constraint on source_term_normalized and origin tracking.
class DictionaryEntry(Base):
    __tablename__ = "dictionary_entries"

    id = Column(String, primary_key=True, default=generate_uuid)
    dictionary_id = Column(String, ForeignKey("terminology_dictionaries.id"), nullable=False, index=True)
    source_term = Column(String, nullable=False)
    source_term_normalized = Column(String, nullable=False)
    target_term = Column(String, nullable=False)
    source_language = Column(String, nullable=False, comment="BCP-47 source language code")
    target_language = Column(String, nullable=False, comment="BCP-47 target language code")
    context_notes = Column(Text, nullable=True)
    context_data = Column(JSON, nullable=True, comment="Structured context for term usage")
    usage_notes = Column(Text, nullable=True, comment="Usage guidance for the term mapping")
    has_context = Column(Boolean, default=False, comment="Whether context_data is populated")
    context_source = Column(String, nullable=True, comment="auto|auto_with_edits|manual|bulk")
    origin_source_language = Column(String, nullable=True, comment="Original source language of the term")
    origin_run_id = Column(String, nullable=True, comment="Run ID from which this correction originated")
    origin_row_key = Column(String, nullable=True, comment="Row key within the run that triggered this correction")
    origin_user_id = Column(String, nullable=True, comment="User who submitted the correction")
    created_at = Column(DateTime, default=lambda: datetime.now(UTC))
    updated_at = Column(DateTime, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))

    __table_args__ = (
        UniqueConstraint(
            "dictionary_id", "source_term_normalized", "source_language", "target_language",
            name="uq_dict_source_term_lang"
        ),
        Index("idx_dict_entry_lang", "source_language", "target_language"),
        Index("idx_dict_has_context", "has_context"),
    )
# #endregion DictionaryEntry


# #region TranslationSchedule [TYPE Class]
# @BRIEF Defines a cron-based schedule for recurring translation jobs.
class TranslationSchedule(Base):
    __tablename__ = "translation_schedules"

    id = Column(String, primary_key=True, default=generate_uuid)
    job_id = Column(String, ForeignKey("translation_jobs.id"), nullable=False, index=True)
    cron_expression = Column(String, nullable=False)
    timezone = Column(String, nullable=False, default="UTC")
    is_active = Column(Boolean, default=True)
    last_run_at = Column(DateTime, nullable=True)
    next_run_at = Column(DateTime, nullable=True)
    execution_mode = Column(String, nullable=False, default="full", comment="full, new_key_only")
    created_by = Column(String, nullable=True)
    created_at = Column(DateTime, default=lambda: datetime.now(UTC))
    updated_at = Column(DateTime, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
# #endregion TranslationSchedule


# #region TranslationJobDictionary [TYPE Class]
# @BRIEF Many-to-many association between translation jobs and terminology dictionaries.
class TranslationJobDictionary(Base):
    __tablename__ = "translation_job_dictionaries"

    id = Column(String, primary_key=True, default=generate_uuid)
    job_id = Column(String, ForeignKey("translation_jobs.id"), nullable=False, index=True)
    dictionary_id = Column(String, ForeignKey("terminology_dictionaries.id"), nullable=False, index=True)
    created_at = Column(DateTime, default=lambda: datetime.now(UTC))

    __table_args__ = (
        UniqueConstraint(
            "job_id", "dictionary_id",
            name="uq_job_dictionary"
        ),
    )
# #endregion TranslationJobDictionary


# #region MetricSnapshot [TYPE Class]
# @BRIEF Captures translation performance metrics at a point in time with config/content/dictionary hash keys for aggregation.
class MetricSnapshot(Base):
    __tablename__ = "translation_metric_snapshots"

    id = Column(String, primary_key=True, default=generate_uuid)
    job_id = Column(String, ForeignKey("translation_jobs.id"), nullable=False, index=True)
    run_id = Column(String, ForeignKey("translation_runs.id"), nullable=True, index=True)
    key_hash = Column(String, nullable=False, comment="Hash of dimension key fields for aggregation")
    config_hash = Column(String, nullable=True, comment="Hash of translation configuration state")
    dict_snapshot_hash = Column(String, nullable=True, comment="Hash of dictionary state at capture time")
    covers_events_before = Column(DateTime, nullable=True, comment="Indicates snapshot covers events before this timestamp")
    total_jobs = Column(Integer, default=0)
    total_runs = Column(Integer, default=0)
    total_records = Column(Integer, default=0)
    successful_records = Column(Integer, default=0)
    failed_records = Column(Integer, default=0)
    skipped_records = Column(Integer, default=0)
    avg_duration_ms = Column(Integer, nullable=True)
    p50_duration_ms = Column(Integer, nullable=True)
    p95_duration_ms = Column(Integer, nullable=True)
    p99_duration_ms = Column(Integer, nullable=True)
    per_language_metrics = Column(JSON, nullable=True, comment="Per-language cumulative metrics: {lang: {cumulative_tokens, cumulative_cost, runs}}")
    snapshot_date = Column(DateTime, nullable=False, default=lambda: datetime.now(UTC))
    created_at = Column(DateTime, default=lambda: datetime.now(UTC))

    __table_args__ = (
        Index("ix_metric_snapshots_job_date", "job_id", "snapshot_date"),
    )
# #endregion MetricSnapshot


# #region TranslationLanguage [C:1] [TYPE Class]
# @BRIEF Per-language translation result for a single record, supporting multi-language output.
class TranslationLanguage(Base):
    __tablename__ = "translation_languages"

    id = Column(String, primary_key=True, default=generate_uuid)
    record_id = Column(String, ForeignKey("translation_records.id"), nullable=False)
    language_code = Column(String, nullable=False, comment="BCP-47 language code")
    source_language_detected = Column(String, nullable=True, comment="BCP-47 or 'und' for undetermined")
    translated_value = Column(Text, nullable=True, comment="LLM-generated translation")
    user_edit = Column(Text, nullable=True, comment="User-edited translation")
    final_value = Column(Text, nullable=True, comment="Final resolved value (translated or user edit)")
    status = Column(String, default="pending", comment="pending|translated|approved|edited|rejected|failed|skipped")
    error_message = Column(Text, nullable=True)
    needs_review = Column(Boolean, default=False, comment="Flagged because source language could not be determined")
    language_overridden = Column(Boolean, default=False, comment="Source language was manually overridden by user")
    created_at = Column(DateTime(timezone=True), default=lambda: datetime.now(UTC))

    record = relationship("TranslationRecord", back_populates="languages")

    __table_args__ = (
        UniqueConstraint("record_id", "language_code", name="uq_record_language"),
        Index("idx_tl_language", "language_code"),
        Index("idx_tl_record_lang", "record_id", "language_code"),
    )
# #endregion TranslationLanguage


# #region TranslationPreviewLanguage [C:1] [TYPE Class]
# @BRIEF Per-language preview entry within a preview session.
class TranslationPreviewLanguage(Base):
    __tablename__ = "translation_preview_languages"

    id = Column(String, primary_key=True, default=generate_uuid)
    preview_record_id = Column(String, ForeignKey("translation_preview_records.id"), nullable=False)
    language_code = Column(String, nullable=False, comment="BCP-47 language code")
    source_language_detected = Column(String, nullable=True, comment="BCP-47 or 'und'")
    translated_value = Column(Text, nullable=True)
    user_edit = Column(Text, nullable=True)
    final_value = Column(Text, nullable=True)
    status = Column(String, default="pending", comment="pending|approved|edited|rejected")
    needs_review = Column(Boolean, default=False, comment="Flagged because source language could not be determined")
    created_at = Column(DateTime(timezone=True), default=lambda: datetime.now(UTC))

    preview_record = relationship("TranslationPreviewRecord", back_populates="languages")

    __table_args__ = (
        UniqueConstraint("preview_record_id", "language_code", name="uq_preview_record_language"),
    )
# #endregion TranslationPreviewLanguage


# #region TranslationRunLanguageStats [C:1] [TYPE Class]
# @BRIEF Per-language statistics for a translation run (row counts, tokens, cost).
class TranslationRunLanguageStats(Base):
    __tablename__ = "translation_run_language_stats"

    id = Column(String, primary_key=True, default=generate_uuid)
    run_id = Column(String, ForeignKey("translation_runs.id"), nullable=False)
    language_code = Column(String, nullable=False, comment="BCP-47 language code")
    total_rows = Column(Integer, default=0)
    translated_rows = Column(Integer, default=0)
    failed_rows = Column(Integer, default=0)
    skipped_rows = Column(Integer, default=0)
    token_count = Column(Integer, default=0)
    estimated_cost = Column(Float, default=0.0)

    run = relationship("TranslationRun", back_populates="language_stats")

    __table_args__ = (
        UniqueConstraint("run_id", "language_code", name="uq_run_language"),
        Index("idx_rls_run", "run_id"),
    )
# #endregion TranslationRunLanguageStats


# #endregion TranslateModels