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

# #region preview_response_parser [C:3] [TYPE Module] [SEMANTICS llm, parse, json, superset, hash]
# @BRIEF Parse LLM JSON responses and Superset data; compute config/dict hashes for preview.
# @RELATION DEPENDS_ON -> [TranslationJob]
# @RELATION DEPENDS_ON -> [TranslationJobDictionary]

import hashlib
import json
import re
from typing import Any

from sqlalchemy.orm import Session

from ...models.translate import TranslationJob, TranslationJobDictionary


# #region parse_llm_response [C:3] [TYPE Function] [SEMANTICS llm, parse, json, translate]
# @BRIEF Parse LLM JSON response into structured translations dict with per-language support.
# @PRE response_text is a valid JSON string (possibly wrapped in markdown code block).
# @POST Returns dict mapping row_id -> {detected_source_language, lang_code: translation, ...}.
def parse_llm_response(
    response_text: str,
    expected_count: int,
    target_languages: list[str] | None = None,
    finish_reason: str | None = None,
) -> dict[str, dict[str, str]]:
    """Parse LLM JSON response into structured translations dict."""
    try:
        data = json.loads(response_text)
    except json.JSONDecodeError:
        data = None
        match = re.search(r'```(?:json)?\s*\n?(.*?)\n?```', response_text, re.DOTALL)
        if match:
            try:
                data = json.loads(match.group(1))
            except json.JSONDecodeError:
                pass

        if data is None:
            rows_match = re.findall(r'\{\s*"row_id"\s*:\s*(?:\d+|"\d+").*?\}\s*', response_text, re.DOTALL)
            if rows_match:
                partial_rows = []
                for row_text in rows_match:
                    try:
                        partial_rows.append(json.loads(row_text))
                    except json.JSONDecodeError:
                        continue
                if partial_rows:
                    data = {"rows": partial_rows}
            if data is None:
                raise ValueError("LLM response was not valid JSON")

    rows = data.get("rows", [])
    if not isinstance(rows, list):
        raise ValueError("LLM response missing 'rows' array")

    translations: dict[str, dict[str, str]] = {}
    for item in rows:
        row_id = str(item.get("row_id", ""))
        if not row_id:
            continue
        detected_lang = str(item.get("detected_source_language", "und")) if item.get("detected_source_language") else "und"
        result: dict[str, str] = {"detected_source_language": detected_lang}
        has_language_data = False
        if target_languages:
            for lang_code in target_languages:
                lang_val = item.get(lang_code)
                if lang_val is not None and str(lang_val).strip():
                    result[lang_code] = str(lang_val)
                    has_language_data = True
        if not has_language_data:
            translation = item.get("translation")
            if translation is not None:
                result["translation"] = str(translation)
            else:
                continue
        translations[row_id] = result

    return translations
# #endregion parse_llm_response


# #region extract_data_rows [C:1] [TYPE Function] [SEMANTICS superset, data, extraction]
# @BRIEF Extract data rows from Superset chart data API response.
def extract_data_rows(response: dict[str, Any]) -> list[dict[str, Any]]:
    """Extract data rows from Superset chart data API response."""
    result = response.get("result")
    if isinstance(result, list):
        for item in result:
            if isinstance(item, dict):
                data = item.get("data")
                if isinstance(data, list) and data:
                    return data
    if isinstance(result, dict):
        data = result.get("data")
        if isinstance(data, list) and data:
            return data
    data = response.get("data")
    if isinstance(data, list) and data:
        return data
    if isinstance(result, list):
        return result
    return []
# #endregion extract_data_rows


# #region compute_config_hash [C:1] [TYPE Function] [SEMANTICS config, hash, deterministic]
# @BRIEF Compute a deterministic hash of job configuration for snapshot comparison.
def compute_config_hash(job: TranslationJob) -> str:
    """Compute a deterministic hash of job configuration for snapshot comparison."""
    config_str = json.dumps({
        "source_dialect": job.source_dialect,
        "target_dialect": job.target_dialect,
        "source_datasource_id": job.source_datasource_id,
        "translation_column": job.translation_column,
        "context_columns": job.context_columns,
        "provider_id": job.provider_id,
        "batch_size": job.batch_size,
        "upsert_strategy": job.upsert_strategy,
    }, sort_keys=True)
    return hashlib.sha256(config_str.encode()).hexdigest()[:16]
# #endregion compute_config_hash


# #region compute_dict_snapshot_hash [C:2] [TYPE Function] [SEMANTICS dictionary, hash, snapshot]
# @BRIEF Compute a hash of dictionary state for snapshot comparison.
def compute_dict_snapshot_hash(db: Session, job_id: str) -> str:
    """Compute a hash of dictionary state for snapshot comparison."""
    dict_links = (
        db.query(TranslationJobDictionary)
        .filter(TranslationJobDictionary.job_id == job_id)
        .all()
    )
    dict_ids = sorted([dl.dictionary_id for dl in dict_links])
    hash_input = ",".join(dict_ids)
    return hashlib.sha256(hash_input.encode()).hexdigest()[:16]
# #endregion compute_dict_snapshot_hash
# #endregion preview_response_parser