ss-tools/backend/src/plugins/translate/preview_prompt_builder.py

# #region PreviewPromptBuilder [C:3] [TYPE Class] [SEMANTICS prompt, dictionary, preview]
# @defgroup Translate Module group.
# @BRIEF Build LLM prompts for preview sessions including dictionary glossary and row context.
# @RELATION DEPENDS_ON -> [DictionaryManager]
# @RELATION DEPENDS_ON -> [render_prompt]
# @RELATION DEPENDS_ON -> [preview_prompt_helpers]
# @RATIONALE Token estimation and budget helpers extracted to preview_prompt_helpers module for INV_7 compliance.

import json
from typing import Any

from sqlalchemy.orm import Session

from ...core.logger import belief_scope
from ...models.translate import TranslationJob
from ...services.llm_prompt_templates import render_prompt
from .dictionary import DictionaryManager
from .preview_constants import DEFAULT_PREVIEW_PROMPT_TEMPLATE
from .preview_prompt_helpers import (
    compute_build_token_metadata,
    estimate_token_budget_for_rows,
)


class PreviewPromptBuilder:
    """Build prompts for preview translation including dictionary glossary and context."""

    def __init__(self, db: Session):
        self.db = db

    # region build_prompt_from_rows [TYPE Function]
    # @PURPOSE: Build the complete LLM prompt from source rows, dictionary, and job config.
    # @PRE  job has valid configuration. source_rows is non-empty.
    # @POST  Returns prompt string and token budget metadata.
    def build_prompt_from_rows(
        self,
        job: TranslationJob,
        source_rows: list[dict[str, Any]],
        sample_size: int,
        prompt_template: str | None = None,
    ) -> dict[str, Any]:
        with belief_scope("PreviewPromptBuilder.build_prompt_from_rows"):
            actual_row_count = len(source_rows)
            target_languages = job.target_languages or [job.target_dialect or "en"]
            if not isinstance(target_languages, list):
                target_languages = [str(target_languages)]
            num_languages = len(target_languages)

            # Build row metadata
            all_source_texts = []
            row_meta: list[dict[str, Any]] = []
            for idx, row in enumerate(source_rows):
                translation_value = str(row.get(job.translation_column, "") or "")
                context_values = {}
                if job.context_columns:
                    for col in job.context_columns:
                        context_values[col] = str(row.get(col, "") or "")
                all_source_texts.append(translation_value)
                row_meta.append({
                    "row_index": idx,
                    "source_text": translation_value,
                    "context_data": context_values,
                    "source_row": row,
                })

            # Filter dictionary entries with row context
            row_context = row_meta[0].get("context_data") if row_meta else None
            dict_matches = DictionaryManager.filter_for_batch(
                self.db, all_source_texts, job.id,
                row_context=row_context,
            )

            dictionary_section = ""
            if dict_matches:
                glossary_lines = []
                for match in dict_matches:
                    glossary_lines.append(
                        f"- '{match['source_term']}' -> '{match['target_term']}'"
                        f"{' (' + match['context_notes'] + ')' if match.get('context_notes') else ''}"
                    )
                dictionary_section = (
                    "Terminology dictionary (use these translations when applicable):\n"
                    + "\n".join(glossary_lines)
                    + "\n\n"
                )

            rows_json = json.dumps([
                {"row_id": str(m["row_index"]), "text": m["source_text"], "context": m["context_data"]}
                for m in row_meta
            ], indent=2)

            target_languages_str = ", ".join(target_languages)
            template = prompt_template or DEFAULT_PREVIEW_PROMPT_TEMPLATE
            prompt = render_prompt(template, {
                "source_language": job.source_dialect or "SQL",
                "target_language": target_languages_str,
                "source_dialect": job.source_dialect or "",
                "target_languages": target_languages_str,
                "translation_column": job.translation_column or "",
                "dictionary_section": dictionary_section,
                "rows_json": rows_json,
                "row_count": str(actual_row_count),
            })

            return compute_build_token_metadata(
                prompt=prompt,
                row_meta=row_meta,
                target_languages=target_languages,
                num_languages=num_languages,
                dictionary_section=dictionary_section,
                actual_row_count=actual_row_count,
                sample_size=sample_size,
            )
    # endregion build_prompt_from_rows

    # region estimate_token_budget_for_rows [TYPE Function]
    # @PURPOSE: Check token budget and optionally truncate rows.
    def estimate_token_budget_for_rows(
        self,
        source_rows: list[dict[str, Any]],
        target_languages: list[str],
        job: TranslationJob,
        provider_model: str | None,
    ) -> dict[str, Any]:
        with belief_scope("PreviewPromptBuilder.estimate_token_budget_for_rows"):
            return estimate_token_budget_for_rows(
                source_rows=source_rows,
                target_languages=target_languages,
                job=job,
                provider_model=provider_model,
            )
    # endregion estimate_token_budget_for_rows

# #endregion PreviewPromptBuilder