# #region RunExecutionService [C:4] [TYPE Module] [SEMANTICS translate, run, execution, orchestration]
# @BRIEF Full run lifecycle: prepare run, fetch source rows, filter new keys, orchestrate batches,
#         handle cancellation, update per-language stats, finalize run status.
# @LAYER Domain
# @RELATION DEPENDS_ON -> [TranslationRun], [TranslationJob], [TranslationBatch], [TranslationRecord]
# @RELATION DEPENDS_ON -> [TranslationPreviewSession], [BatchProcessingService], [AdaptiveBatchSizer]
# @RELATION DEPENDS_ON -> [RunSourceFetcher]
# @PRE DB session available. Run has valid job config.
# @POST Run status updated to COMPLETED/FAILED/CANCELLED. Batches and records created.
# @SIDE_EFFECT Fetches data from Superset API; calls BatchProcessingService; creates DB rows.
# @RATIONALE Extracted from TranslationExecutor. Source fetching delegated to _run_source.py.
# @REJECTED Keeping all run orchestration inside TranslationExecutor — caused class to exceed INV_7.

from collections.abc import Callable
from typing import Any

from sqlalchemy.orm import Session

from ...core.config_manager import ConfigManager
from ...core.logger import belief_scope, logger
from ...models.translate import TranslationRecord, TranslationRun, TranslationRunLanguageStats
from ._run_source import _extract_chart_data_rows, fetch_source_rows


# #region RunExecutionService [C:4] [TYPE Class]
# @BRIEF Orchestrate full translation run: fetch data, process batches, finalize.
class RunExecutionService:
    """Orchestrate a full translation run: prepare, process batches, finalize."""

    def __init__(
        self, db: Session, config_manager: ConfigManager,
        current_user: str | None = None,
        on_batch_progress: Callable[[str, int, int, int, int], None] | None = None,
    ) -> None:
        self.db = db
        self.config_manager = config_manager
        self.current_user = current_user
        self.on_batch_progress = on_batch_progress
        self._preview_edits_cache: dict[str, dict[str, str]] | None = None

    # -- Source fetching (thin wrappers) --
    def _fetch_source_rows(self, job_id: str, run_id: str) -> list[dict[str, Any]]:
        return fetch_source_rows(self.db, self.config_manager, job_id, run_id)

    @staticmethod
    def _extract_chart_data_rows(response):
        return _extract_chart_data_rows(response)

    # -- Key/filter helpers --
    def _filter_new_keys(self, job, run_id: str, source_rows: list) -> list:
        """Filter source rows to only new keys not yet translated."""
        with belief_scope("RunExecutionService._filter_new_keys"):
            prev_run = (
                self.db.query(TranslationRun)
                .filter(TranslationRun.job_id == job.id, TranslationRun.status == "COMPLETED",
                        TranslationRun.insert_status == "succeeded", TranslationRun.id != run_id)
                .order_by(TranslationRun.created_at.desc()).first()
            )
            if not prev_run:
                logger.reason("No prior successful run — all keys treated as new", {"job_id": job.id})
                return source_rows

            prev_records = (
                self.db.query(TranslationRecord)
                .filter(TranslationRecord.run_id == prev_run.id, TranslationRecord.status == "SUCCESS")
                .all()
            )
            if not prev_records:
                return source_rows

            key_cols = job.target_key_cols or job.source_key_cols or []
            if not key_cols:
                logger.explore("No key columns configured — skipping new-key-only filter", {"job_id": job.id})
                return source_rows

            existing_keys = set()
            for rec in prev_records:
                sd = rec.source_data or {}
                key_tuple = tuple(str(sd.get(k, "")) for k in key_cols)
                existing_keys.add(key_tuple)

            filtered = []
            skipped = 0
            for row in source_rows:
                sd = row.get("source_data", {}) or {}
                key_tuple = tuple(str(sd.get(k, "")) for k in key_cols)
                if key_tuple not in existing_keys:
                    filtered.append(row)
                else:
                    skipped += 1

            logger.reason(f"New-key-only filter: {len(source_rows)} total -> {len(filtered)} new, {skipped} skipped",
                          {"job_id": job.id, "prev_run_id": prev_run.id, "key_cols": key_cols})
            return filtered

    def _load_preview_edits(self, job_id: str) -> None:
        """Load preview edits for carry-forward during execution."""
        from ...models.translate import TranslationPreviewRecord, TranslationPreviewSession

        with belief_scope("RunExecutionService._load_preview_edits"):
            session = (
                self.db.query(TranslationPreviewSession)
                .filter(TranslationPreviewSession.job_id == job_id, TranslationPreviewSession.status == "APPLIED")
                .order_by(TranslationPreviewSession.created_at.desc()).first()
            )
            if not session:
                logger.reason("No applied preview session found — no edits to carry forward", {"job_id": job_id})
                self._preview_edits_cache = {}
                return

            records = (
                self.db.query(TranslationPreviewRecord)
                .filter(TranslationPreviewRecord.session_id == session.id).all()
            )
            edits: dict[str, dict[str, str]] = {}
            for rec in records:
                if not rec.source_data:
                    continue
                key_hash = self._compute_key_hash(rec.source_data)
                edited_langs: dict[str, str] = {}
                for lang_entry in (rec.languages or []):
                    if lang_entry.status in ("edited", "approved") and lang_entry.user_edit:
                        edited_langs[lang_entry.language_code] = lang_entry.final_value or lang_entry.user_edit
                        logger.reason("Carrying forward preview edit", {"key_hash": key_hash, "language_code": lang_entry.language_code})
                if edited_langs:
                    edits[key_hash] = edited_langs
            self._preview_edits_cache = edits
            logger.reason(f"Loaded {len(edits)} preview edits for carry-forward", {"job_id": job_id})

    @staticmethod
    def _compute_key_hash(source_data: dict) -> str:
        import hashlib
        import json
        return hashlib.sha256(json.dumps(source_data, sort_keys=True).encode()).hexdigest()[:16]

    # -- Language stats --
    def _update_language_stats_incremental(self, run_id: str, language_stats_map: dict[str, TranslationRunLanguageStats]) -> None:
        """Update per-language statistics incrementally after each batch."""
        with belief_scope("RunExecutionService._update_language_stats_incremental"):
            records = self.db.query(TranslationRecord).filter(TranslationRecord.run_id == run_id).all()
            record_ids = [r.id for r in records]
            if not record_ids:
                return
            from ...models.translate import TranslationLanguage
            lang_entries = (
                self.db.query(TranslationLanguage)
                .filter(TranslationLanguage.record_id.in_(record_ids)).all()
            )
            from collections import defaultdict
            agg: dict[str, dict[str, int]] = defaultdict(lambda: {"total": 0, "translated": 0, "failed": 0, "skipped": 0})
            for le in lang_entries:
                code = le.language_code
                agg[code]["total"] += 1
                if le.status in ("translated", "approved", "edited"):
                    agg[code]["translated"] += 1
                elif le.status == "failed":
                    agg[code]["failed"] += 1
                elif le.status == "skipped":
                    agg[code]["skipped"] += 1

            total_tokens = max(1, sum(len(le.translated_value or "") for le in lang_entries if le.translated_value) // 4)
            num_langs = len(language_stats_map) or 1
            cost_per_token = 0.002 / 1000

            for lang_code, lang_stat in language_stats_map.items():
                data = agg.get(lang_code, {"total": 0, "translated": 0, "failed": 0, "skipped": 0})
                lang_stat.total_rows = data["total"]
                lang_stat.translated_rows = data["translated"]
                lang_stat.failed_rows = data["failed"]
                lang_stat.skipped_rows = data["skipped"]
                lang_stat.token_count = total_tokens // num_langs
                lang_stat.estimated_cost = round((lang_stat.token_count / 1000) * cost_per_token, 6)
            self.db.flush()
# #endregion RunExecutionService
# #endregion RunExecutionService
