"""
Helpers for loading CDP execution results and model metadata.
"""

from __future__ import annotations

import sqlite3
from typing import Any, Dict, List, Optional

import pandas as pd

from .models import WorkflowStep
from .results_manager import ExecutionStatus, ResultsManager
from .bridge_loader import CDPBridge
from . import debug_logger


class CDPResultsService:
    """Loads CDP results and the associated model context."""

    _DISPLAY_FIELD_CANDIDATES = (
        "DisplayName",
        "display_name",
        "name",
        "Name",
        "label",
        "Label",
    )
    _PRIMARY_KEY_CANDIDATES = (
        "ogc_fid",
        "RowID",
        "rowid",
        "OBJECTID",
        "objectid",
        "FID",
        "fid",
        "id",
        "ID",
        "feature_id",
    )
    _PRIMARY_SCORE_CANDIDATES = (
        "cdp_overall_score",
        "overall_score",
        "priority_score",
        "score",
    )
    _RANK_COLUMN_CANDIDATES = (
        "cdp_rank",
        "rank",
        "priority_rank",
    )

    def __init__(self, cdp_engine_path: str, results_manager: Optional[ResultsManager] = None):
        self.cdp_engine_path = cdp_engine_path
        self.results_manager = results_manager or ResultsManager()

    def get_latest_completed_execution(self, workflow_step_id: str) -> Optional[str]:
        """Return the latest completed execution ID for a workflow step."""
        executions = self.results_manager.get_workflow_executions(workflow_step_id)
        for execution in executions:
            if execution.status == ExecutionStatus.COMPLETED:
                return execution.execution_id
        return None

    def load_viewer_payload(self, workflow_step: WorkflowStep) -> Dict[str, Any]:
        """Build a payload for the CDP results dialog."""
        execution_id = self.get_latest_completed_execution(workflow_step.id)
        if not execution_id:
            raise ValueError("No completed CDP execution found for this workflow step")

        parameters = workflow_step.parameters or {}
        workflow_metadata = getattr(workflow_step.results, "metadata", {}) or {}
        database_path = str(
            parameters.get("database_path")
            or workflow_metadata.get("database_path")
            or ""
        ).strip()
        output_table_name = str(
            parameters.get("output_table_name")
            or workflow_metadata.get("table_name")
            or workflow_metadata.get("output_table_name")
            or ""
        ).strip()

        results_df = self._load_preferred_results_dataframe(
            execution_id=execution_id,
            database_path=database_path,
            output_table_name=output_table_name,
        )
        if results_df is None or results_df.empty:
            raise ValueError("Execution results are empty")

        model_path = (
            parameters.get("model_file")
            or parameters.get("config")
            or ""
        )
        field_mappings = list(parameters.get("field_mappings", []) or [])
        model_info = self._try_get_model_info(model_path)
        model_summary = self.build_model_summary(model_info)
        display_field = self._build_column_resolution(
            results_df=results_df,
            requested_column=str(parameters.get("display_field", "")).strip(),
            fallback_candidates=self._DISPLAY_FIELD_CANDIDATES,
        )
        persisted_display_name = self._find_matching_column(results_df, "DisplayName")
        if persisted_display_name:
            display_field = self._build_column_resolution(
                results_df=results_df,
                requested_column=persisted_display_name,
                fallback_candidates=self._DISPLAY_FIELD_CANDIDATES,
                preferred_columns=[persisted_display_name],
            )
        primary_key = self._build_column_resolution(
            results_df=results_df,
            requested_column=str(parameters.get("primary_key", "")).strip(),
            fallback_candidates=self._PRIMARY_KEY_CANDIDATES,
            require_unique_preference=True,
        )
        score_columns = self._detect_score_columns(results_df)
        rank_columns = self._detect_rank_columns(results_df)
        prioritized_score = self._build_column_resolution(
            results_df=results_df,
            requested_column="",
            fallback_candidates=self._PRIMARY_SCORE_CANDIDATES,
            preferred_columns=score_columns,
        )
        field_mapping_summary = self._build_field_mapping_summary(results_df, field_mappings)
        key_result_columns = self._build_key_result_columns(
            results_df=results_df,
            display_field=display_field.get("resolved_name", ""),
            primary_key=primary_key.get("resolved_name", ""),
            prioritized_score=prioritized_score.get("resolved_name", ""),
            score_columns=score_columns,
            rank_columns=rank_columns,
            field_mappings=field_mappings,
        )
        rows_preview = self._build_rows_preview(
            results_df=results_df,
            display_field=display_field.get("resolved_name", ""),
            primary_key=primary_key.get("resolved_name", ""),
            prioritized_score=prioritized_score.get("resolved_name", ""),
            key_result_columns=key_result_columns,
        )
        summary = self.build_results_summary(
            results_df,
            primary_key=primary_key.get("resolved_name", ""),
            prioritized_score=prioritized_score.get("resolved_name", ""),
            score_columns=score_columns,
            rank_columns=rank_columns,
        )
        row_key_field = self._resolve_row_key_field(
            results_df=results_df,
            primary_key=primary_key.get("resolved_name", ""),
        )
        viewer_datasets = self._load_engine_viewer_datasets(
            results_df=results_df,
            model_path=model_path,
            model_info=model_info,
            parameters=parameters,
            display_field=display_field.get("resolved_name", ""),
            primary_key=primary_key.get("resolved_name", ""),
            input_table_name=str(parameters.get("input_table_name", "") or output_table_name).strip(),
        )

        debug_logger.debug("Loaded CDP viewer payload", {
            "workflow_step_id": workflow_step.id,
            "execution_id": execution_id,
            "rows": len(results_df.index),
            "columns": list(results_df.columns),
            "has_model_info": bool(model_info),
            "database_path": database_path,
            "output_table_name": output_table_name,
            "display_field_requested": parameters.get("display_field", ""),
            "display_field_resolved": display_field.get("resolved_name", ""),
            "display_field_prefers_persisted_display_name": bool(persisted_display_name),
            "primary_key_requested": parameters.get("primary_key", ""),
            "primary_key_resolved": primary_key.get("resolved_name", ""),
            "row_key_field": row_key_field,
            "prioritized_score_column": prioritized_score.get("resolved_name", ""),
            "field_mapping_count": len(field_mappings),
        })

        return {
            "workflow_step_id": workflow_step.id,
            "workflow_name": workflow_step.name,
            "execution_id": execution_id,
            "results_df": results_df,
            "summary": summary,
            "database_path": database_path,
            "output_table_name": output_table_name,
            "row_key_field": row_key_field,
            "model_path": model_path,
            "model_info": model_info,
            "model_summary": model_summary,
            "cdp_engine_path": self.cdp_engine_path,
            "display_field": str(parameters.get("display_field", "")).strip(),
            "primary_key": str(parameters.get("primary_key", "")).strip(),
            "field_mappings": field_mappings,
            "mapping_rows": list(parameters.get("mapping_rows", []) or []),
            "input_table_name": str(parameters.get("input_table_name", "") or output_table_name).strip(),
            "validation_settings": dict(parameters.get("validation_settings", {}) or {}),
            "resolved_display_field": display_field,
            "resolved_primary_key": primary_key,
            "prioritized_score_column": prioritized_score,
            "score_columns": score_columns,
            "rank_columns": rank_columns,
            "field_mapping_summary": field_mapping_summary,
            "key_result_columns": key_result_columns,
            "rows_preview": rows_preview,
            **viewer_datasets,
        }

    def _load_preferred_results_dataframe(
        self,
        *,
        execution_id: str,
        database_path: str = "",
        output_table_name: str = "",
    ) -> pd.DataFrame:
        """Prefer the persisted project table over the execution-results cache."""
        persisted_df = self._load_persisted_results_dataframe(
            database_path=database_path,
            output_table_name=output_table_name,
        )
        if persisted_df is not None and not persisted_df.empty:
            return persisted_df
        return self.results_manager.get_execution_results_dataframe(execution_id)

    def _load_persisted_results_dataframe(
        self,
        *,
        database_path: str,
        output_table_name: str,
    ) -> Optional[pd.DataFrame]:
        """Load the persisted CDP output table from the project GeoPackage when available."""
        if not database_path or not output_table_name:
            return None

        try:
            conn = sqlite3.connect(database_path)
            query = f'SELECT * FROM "{output_table_name.replace("\"", "\"\"")}"'
            data_frame = pd.read_sql_query(query, conn)
            conn.close()
        except Exception as exc:
            debug_logger.warning("Could not load persisted CDP output table", {
                "database_path": database_path,
                "output_table_name": output_table_name,
                "exception": str(exc),
            })
            return None

        geometry_columns = [
            column_name
            for column_name in data_frame.columns
            if str(column_name).strip().lower() in {"geom", "geometry", "wkb_geometry"}
        ]
        if geometry_columns:
            data_frame = data_frame.drop(columns=geometry_columns, errors="ignore")

        return data_frame

    def _try_get_model_info(self, model_path: str) -> Dict[str, Any]:
        """Load CDP model info if the model path is available."""
        if not model_path:
            return {}

        try:
            bridge = CDPBridge(self.cdp_engine_path)
            return bridge.get_model_info(model_path)
        except Exception as exc:
            debug_logger.warning("Could not load CDP model info", {
                "model_path": model_path,
                "engine_path": self.cdp_engine_path,
                "exception": str(exc),
            })
            return {}

    @classmethod
    def _resolve_row_key_field(cls, *, results_df: pd.DataFrame, primary_key: str = "") -> str:
        for candidate in (
            primary_key,
            "ogc_fid",
            "fid",
            "FID",
            "RowID",
            "OBJECTID",
            "id",
            "ID",
        ):
            if candidate and candidate in results_df.columns:
                return str(candidate)
        return ""

    def _load_engine_viewer_datasets(
        self,
        *,
        results_df: pd.DataFrame,
        model_path: str,
        model_info: Dict[str, Any],
        parameters: Dict[str, Any],
        display_field: str,
        primary_key: str,
        input_table_name: str,
    ) -> Dict[str, Any]:
        if not model_path or not self.cdp_engine_path or results_df.empty:
            return self._empty_viewer_dataset_payload()

        payload = self._empty_viewer_dataset_payload()
        payload["model_graph"] = self._build_model_graph_payload_from_model_info(model_info)
        payload["model_diagram"] = self._build_model_diagram_payload_from_graph_payload(payload["model_graph"])

        try:
            bridge = CDPBridge(self.cdp_engine_path)
            field_mappings = list(parameters.get("field_mappings", []) or [])
            mapping_rows = list(parameters.get("mapping_rows", []) or [])
            validation_settings = dict(parameters.get("validation_settings", {}) or {})
            try:
                model_graph = bridge.get_model_graph(model_path)
                if isinstance(model_graph, dict) and model_graph:
                    payload["model_graph"] = model_graph
            except Exception as exc:
                debug_logger.warning("Could not load CDP model graph payload", {
                    "model_path": model_path,
                    "engine_path": self.cdp_engine_path,
                    "exception": str(exc),
                })

            try:
                model_diagram = bridge.get_model_diagram(model_path)
                if isinstance(model_diagram, dict) and model_diagram:
                    payload["model_diagram"] = model_diagram
            except Exception as exc:
                debug_logger.warning("Could not load CDP model diagram payload", {
                    "model_path": model_path,
                    "engine_path": self.cdp_engine_path,
                    "exception": str(exc),
                })

            lowest = bridge.get_lowest_criteria(model_path)
            lowest_table = lowest.get("table") if isinstance(lowest, dict) else pd.DataFrame()
            lowest_info = lowest.get("model_info", {}) if isinstance(lowest, dict) else {}

            lowest_rows = (
                lowest_table.to_dict("records")
                if isinstance(lowest_table, pd.DataFrame) and not lowest_table.empty
                else list(lowest_info.get("LowestCriteriaRows", []) or [])
            )
            payload["lowest_criteria_rows"] = lowest_rows

            lowest_default_hblock = self._extract_first_int(lowest_rows, "HBlockID", 0)

            contributions = bridge.get_contributions(
                results_df,
                model_path,
                field_mappings=field_mappings,
                mapping_rows=mapping_rows,
                display_field=display_field,
                primary_key=primary_key,
                input_table_name=input_table_name,
                validation_settings=validation_settings,
                source_level_index=1,
                target_hblock_id=lowest_default_hblock,
            )
            sensitivity = bridge.get_sensitivity(
                results_df,
                model_path,
                field_mappings=field_mappings,
                mapping_rows=mapping_rows,
                display_field=display_field,
                primary_key=primary_key,
                input_table_name=input_table_name,
                validation_settings=validation_settings,
            )
            tradeoffs = bridge.get_tradeoffs(
                results_df,
                model_path,
                field_mappings=field_mappings,
                mapping_rows=mapping_rows,
                display_field=display_field,
                primary_key=primary_key,
                input_table_name=input_table_name,
                validation_settings=validation_settings,
                reference_hblock_id=lowest_default_hblock,
            )

            payload.update({
                "contributions_table": contributions.get("table", pd.DataFrame()),
                "source_level_options": list(
                    (contributions.get("model_info") or {}).get("SourceLevelOptions", []) or []
                ),
                "target_level_options": list(
                    (contributions.get("model_info") or {}).get("TargetLevelOptions", []) or []
                ),
                "sensitivity_table": sensitivity.get("table", pd.DataFrame()),
                "sensitivity_criterion_options": list(
                    (sensitivity.get("model_info") or {}).get("CriterionOptions", []) or []
                ),
                "tradeoffs_table": tradeoffs.get("table", pd.DataFrame()),
            })
            return payload
        except Exception as exc:
            debug_logger.warning("Could not load engine-backed CDP viewer datasets", {
                "model_path": model_path,
                "engine_path": self.cdp_engine_path,
                "exception": str(exc),
            })
            return payload

    @staticmethod
    def _extract_first_int(rows: List[Dict[str, Any]], column_name: str, default_value: int = 0) -> int:
        for row in rows:
            try:
                return int(row.get(column_name))
            except Exception:
                continue
        return default_value

    @staticmethod
    def _empty_viewer_dataset_payload() -> Dict[str, Any]:
        return {
            "model_graph": {},
            "model_diagram": {},
            "lowest_criteria_rows": [],
            "contributions_table": pd.DataFrame(),
            "source_level_options": [],
            "target_level_options": [],
            "sensitivity_table": pd.DataFrame(),
            "sensitivity_criterion_options": [],
            "tradeoffs_table": pd.DataFrame(),
        }

    @staticmethod
    def _build_model_graph_payload_from_model_info(model_info: Dict[str, Any]) -> Dict[str, Any]:
        if not isinstance(model_info, dict) or not model_info:
            return {}

        levels = list(model_info.get("Levels", []) or [])
        nodes = list(model_info.get("AllCriteria", []) or model_info.get("Criteria", []) or [])
        edges = list(model_info.get("Weights", []) or [])
        lowest_rows = list(model_info.get("LowestCriteriaRows", []) or model_info.get("Criteria", []) or [])

        parent_ids = set()
        child_ids = set()
        for edge in edges:
            try:
                parent_ids.add(int(edge.get("CriterionID")))
            except Exception:
                pass
            try:
                child_ids.add(int(edge.get("SubCriterionID")))
            except Exception:
                pass

        root_node_ids = sorted(parent_ids - child_ids)
        if not root_node_ids and nodes:
            try:
                first_node_id = int(nodes[0].get("HBlockID"))
                root_node_ids = [first_node_id]
            except Exception:
                root_node_ids = []

        return {
            "Name": model_info.get("Name", ""),
            "Description": model_info.get("Description", ""),
            "Levels": levels,
            "Nodes": nodes,
            "Edges": edges,
            "LowestCriteriaRows": lowest_rows,
            "RootNodeIds": root_node_ids,
            "NodeCount": len(nodes),
            "EdgeCount": len(edges),
        }

    @classmethod
    def _build_model_diagram_payload_from_graph_payload(cls, model_graph: Dict[str, Any]) -> Dict[str, Any]:
        if not isinstance(model_graph, dict) or not model_graph:
            return {}

        nodes = [dict(node) for node in list(model_graph.get("Nodes", []) or [])]
        levels = [dict(level) for level in list(model_graph.get("Levels", []) or [])]
        lowest_rows = list(model_graph.get("LowestCriteriaRows", []) or [])
        lowest_ids = {
            str(row.get("HBlockID")).strip()
            for row in lowest_rows
            if row.get("HBlockID") is not None
        }

        ordered_level_values = []
        for node in nodes:
            try:
                ordered_level_values.append(int(node.get("Level")))
            except Exception:
                continue
        ordered_level_values = sorted(set(ordered_level_values))

        level_payloads = []
        for level_index, level_value in enumerate(ordered_level_values):
            matching_level = next(
                (
                    level
                    for level in levels
                    if cls._safe_int(level.get("LevelID"), -1) == level_value
                    or cls._safe_int(level.get("HLevelsCounter"), -1) == level_value
                ),
                {},
            )
            level_payload = dict(matching_level)
            level_payload.update({
                "DiagramLevelIndex": level_value,
                "DiagramLeft": float(level_index * 240),
                "DiagramTop": 0.0,
                "DiagramRight": float((level_index * 240) + 220),
                "DiagramBottom": 24.0,
            })
            level_payloads.append(level_payload)

        max_level_value = max(ordered_level_values) if ordered_level_values else -1
        positioned_nodes = []
        for node in nodes:
            positioned_node = dict(node)
            level_value = cls._safe_int(node.get("Level"), 0)
            positioned_node["DiagramLevel"] = level_value
            positioned_node["DiagramVerticalPos"] = cls._safe_int(
                node.get("VerticalPosition"),
                cls._safe_int(node.get("VerticalPos"), 0),
            )
            positioned_node["DiagramMatched"] = False
            positioned_node["IsAlternative"] = bool(level_value == max_level_value)
            positioned_node["IsLowestCriterion"] = str(node.get("HBlockID", "")).strip() in lowest_ids
            positioned_nodes.append(positioned_node)

        return {
            "Name": model_graph.get("Name", ""),
            "Description": model_graph.get("Description", ""),
            "Engine": model_graph.get("Engine", "CDP (Criterium Decision Plus)"),
            "Source": "derived_graph",
            "Levels": level_payloads,
            "Nodes": positioned_nodes,
            "Edges": list(model_graph.get("Edges", []) or []),
            "LowestCriteriaRows": lowest_rows,
            "RootNodeIds": list(model_graph.get("RootNodeIds", []) or []),
            "NodeCount": len(positioned_nodes),
            "EdgeCount": len(list(model_graph.get("Edges", []) or [])),
        }

    @staticmethod
    def _safe_int(value: Any, default_value: int = 0) -> int:
        try:
            return int(value)
        except Exception:
            return default_value

    @classmethod
    def build_results_summary(
        cls,
        results_df: pd.DataFrame,
        primary_key: str = "",
        prioritized_score: str = "",
        score_columns: Optional[List[str]] = None,
        rank_columns: Optional[List[str]] = None,
    ) -> Dict[str, Any]:
        """Compute a small summary block for the results tab."""
        summary: Dict[str, Any] = {
            "row_count": len(results_df.index),
            "column_count": len(results_df.columns),
        }

        score_columns = list(score_columns or [])
        rank_columns = list(rank_columns or [])
        score_column = prioritized_score or (score_columns[0] if score_columns else "")
        rank_column = rank_columns[0] if rank_columns else ""

        summary["score_columns"] = score_columns
        summary["rank_columns"] = rank_columns
        summary["primary_score_column"] = score_column
        summary["primary_rank_column"] = rank_column

        if score_column and score_column in results_df.columns:
            scores = pd.to_numeric(results_df[score_column], errors="coerce")
            summary["score_min"] = None if scores.dropna().empty else float(scores.min())
            summary["score_max"] = None if scores.dropna().empty else float(scores.max())
            summary["score_mean"] = None if scores.dropna().empty else float(scores.mean())
            summary["score_non_null_count"] = int(scores.notna().sum())

        if primary_key and primary_key in results_df.columns:
            key_series = results_df[primary_key]
            non_null = key_series.dropna()
            summary["primary_key"] = primary_key
            summary["primary_key_unique_count"] = int(non_null.nunique())
            summary["primary_key_duplicate_count"] = int(non_null.size - non_null.nunique())

        cdp_columns = [column for column in results_df.columns if str(column).lower().startswith("cdp_")]
        summary["cdp_column_count"] = len(cdp_columns)
        summary["cdp_columns_preview"] = cdp_columns[:8]

        return summary

    @staticmethod
    def build_model_summary(model_info: Dict[str, Any]) -> Dict[str, Any]:
        """Return compact, UI-friendly model metadata."""
        if not isinstance(model_info, dict) or not model_info:
            return {
                "has_model_info": False,
                "top_level_keys": [],
                "collection_sizes": {},
            }

        collection_sizes: Dict[str, int] = {}
        scalar_preview: Dict[str, Any] = {}
        for key, value in model_info.items():
            if isinstance(value, (list, tuple, set)):
                collection_sizes[key] = len(value)
            elif isinstance(value, dict):
                collection_sizes[key] = len(value.keys())
            elif len(scalar_preview) < 8:
                scalar_preview[key] = value

        return {
            "has_model_info": True,
            "top_level_keys": list(model_info.keys()),
            "collection_sizes": collection_sizes,
            "scalar_preview": scalar_preview,
        }

    @staticmethod
    def _normalize_name(value: Any) -> str:
        return str(value or "").strip().lower()

    @classmethod
    def _find_matching_column(cls, results_df: pd.DataFrame, requested_column: str) -> str:
        """Resolve a requested column name case-insensitively."""
        if not requested_column:
            return ""
        requested_normalized = cls._normalize_name(requested_column)
        for column_name in results_df.columns:
            if cls._normalize_name(column_name) == requested_normalized:
                return str(column_name)
        return ""

    @classmethod
    def _build_column_resolution(
        cls,
        *,
        results_df: pd.DataFrame,
        requested_column: str,
        fallback_candidates: List[str] | tuple,
        preferred_columns: Optional[List[str]] = None,
        require_unique_preference: bool = False,
    ) -> Dict[str, Any]:
        """Resolve a requested or inferred column and provide preview metadata."""
        preferred_columns = list(preferred_columns or [])
        available_columns = [str(column) for column in results_df.columns]
        resolved_name = cls._find_matching_column(results_df, requested_column)
        resolution_source = "requested" if resolved_name else ""

        if not resolved_name:
            for candidate in preferred_columns:
                resolved_name = cls._find_matching_column(results_df, candidate)
                if resolved_name:
                    resolution_source = "preferred"
                    break

        if not resolved_name and require_unique_preference:
            for candidate in available_columns:
                series = results_df[candidate]
                non_null = series.dropna()
                if non_null.empty:
                    continue
                if non_null.nunique(dropna=True) == len(non_null.index):
                    resolved_name = candidate
                    resolution_source = "inferred_unique"
                    break

        if not resolved_name:
            for candidate in fallback_candidates:
                resolved_name = cls._find_matching_column(results_df, candidate)
                if resolved_name:
                    resolution_source = "candidate"
                    break

        preview_values: List[Any] = []
        unique_count = 0
        duplicate_count = 0
        non_null_count = 0
        if resolved_name:
            series = results_df[resolved_name]
            non_null = series.dropna()
            non_null_count = int(non_null.size)
            unique_count = int(non_null.nunique())
            duplicate_count = int(non_null.size - unique_count)
            preview_values = [
                value for value in non_null.astype(str).drop_duplicates().head(5).tolist()
            ]

        return {
            "requested_name": requested_column,
            "resolved_name": resolved_name,
            "exists": bool(resolved_name),
            "resolution_source": resolution_source or ("missing" if requested_column else "unresolved"),
            "preview_values": preview_values,
            "unique_count": unique_count,
            "duplicate_count": duplicate_count,
            "non_null_count": non_null_count,
        }

    @classmethod
    def _detect_score_columns(cls, results_df: pd.DataFrame) -> List[str]:
        """Return likely score columns in useful display order."""
        columns = [str(column) for column in results_df.columns]
        preferred = []
        seen = set()

        for candidate in cls._PRIMARY_SCORE_CANDIDATES:
            resolved = cls._find_matching_column(results_df, candidate)
            if resolved and resolved not in seen:
                preferred.append(resolved)
                seen.add(resolved)

        for column_name in columns:
            normalized = cls._normalize_name(column_name)
            if column_name in seen:
                continue
            if "score" in normalized or "priority" in normalized:
                preferred.append(column_name)
                seen.add(column_name)

        return preferred

    @classmethod
    def _detect_rank_columns(cls, results_df: pd.DataFrame) -> List[str]:
        """Return likely ranking columns."""
        columns = [str(column) for column in results_df.columns]
        preferred = []
        seen = set()

        for candidate in cls._RANK_COLUMN_CANDIDATES:
            resolved = cls._find_matching_column(results_df, candidate)
            if resolved and resolved not in seen:
                preferred.append(resolved)
                seen.add(resolved)

        for column_name in columns:
            normalized = cls._normalize_name(column_name)
            if column_name in seen:
                continue
            if "rank" in normalized or normalized.endswith("_order"):
                preferred.append(column_name)
                seen.add(column_name)

        return preferred

    @classmethod
    def _build_field_mapping_summary(
        cls,
        results_df: pd.DataFrame,
        field_mappings: List[Dict[str, Any]],
    ) -> Dict[str, Any]:
        """Summarize wizard field mappings against the actual results columns."""
        mapping_pairs = []
        matched_source_columns = []
        missing_source_columns = []
        target_variables = []

        for mapping in field_mappings:
            source_column = str(mapping.get("source_column", "")).strip()
            target_variable = str(mapping.get("target_variable", "")).strip()
            resolved_source = cls._find_matching_column(results_df, source_column)
            mapping_pairs.append(
                {
                    "target_variable": target_variable,
                    "source_column": source_column,
                    "resolved_source_column": resolved_source,
                    "matched": bool(resolved_source),
                }
            )
            if target_variable:
                target_variables.append(target_variable)
            if resolved_source:
                matched_source_columns.append(resolved_source)
            elif source_column:
                missing_source_columns.append(source_column)

        return {
            "mapping_count": len(field_mappings),
            "matched_mapping_count": len(matched_source_columns),
            "missing_mapping_count": len(missing_source_columns),
            "target_variables": target_variables,
            "mapped_source_columns": matched_source_columns,
            "missing_source_columns": missing_source_columns,
            "mapping_pairs_preview": mapping_pairs[:12],
        }

    @classmethod
    def _build_key_result_columns(
        cls,
        *,
        results_df: pd.DataFrame,
        display_field: str,
        primary_key: str,
        prioritized_score: str,
        score_columns: List[str],
        rank_columns: List[str],
        field_mappings: List[Dict[str, Any]],
    ) -> Dict[str, List[str]]:
        """Group columns into stable buckets for future viewer tabs."""
        columns = [str(column) for column in results_df.columns]
        mapped_columns = []
        for mapping in field_mappings:
            resolved = cls._find_matching_column(results_df, str(mapping.get("source_column", "")).strip())
            if resolved and resolved not in mapped_columns:
                mapped_columns.append(resolved)

        identifier_columns = [
            column for column in [primary_key, display_field]
            if column and column in columns
        ]
        cdp_columns = [column for column in columns if cls._normalize_name(column).startswith("cdp_")]
        priority_columns = []
        for column_name in columns:
            normalized = cls._normalize_name(column_name)
            if column_name == prioritized_score:
                continue
            if "priority" in normalized:
                priority_columns.append(column_name)

        preview_columns = []
        for column_name in identifier_columns + [prioritized_score] + rank_columns + mapped_columns:
            if column_name and column_name in columns and column_name not in preview_columns:
                preview_columns.append(column_name)
        for column_name in cdp_columns:
            if len(preview_columns) >= 8:
                break
            if column_name not in preview_columns:
                preview_columns.append(column_name)

        return {
            "identifier_columns": identifier_columns,
            "display_columns": [display_field] if display_field and display_field in columns else [],
            "primary_key_columns": [primary_key] if primary_key and primary_key in columns else [],
            "score_columns": [column for column in score_columns if column in columns],
            "rank_columns": [column for column in rank_columns if column in columns],
            "priority_columns": priority_columns,
            "cdp_columns": cdp_columns,
            "mapped_source_columns": mapped_columns,
            "preview_columns": preview_columns[:8],
        }

    @classmethod
    def _build_rows_preview(
        cls,
        *,
        results_df: pd.DataFrame,
        display_field: str,
        primary_key: str,
        prioritized_score: str,
        key_result_columns: Dict[str, List[str]],
    ) -> Dict[str, Any]:
        """Build compact row previews for future viewer tabs."""
        preview_columns = [
            column for column in key_result_columns.get("preview_columns", [])
            if column in results_df.columns
        ]
        if not preview_columns:
            preview_columns = [str(column) for column in results_df.columns[:6]]

        def row_to_preview(row_series: pd.Series) -> Dict[str, Any]:
            preview = {
                "row_index": int(row_series.name) if isinstance(row_series.name, int) else row_series.name,
            }
            if primary_key and primary_key in row_series.index:
                preview["primary_key"] = row_series.get(primary_key)
            if display_field and display_field in row_series.index:
                preview["display_value"] = row_series.get(display_field)
            if prioritized_score and prioritized_score in row_series.index:
                preview["score"] = row_series.get(prioritized_score)
            preview["values"] = {
                column_name: row_series.get(column_name)
                for column_name in preview_columns
            }
            return preview

        first_rows = [
            row_to_preview(row_series)
            for _, row_series in results_df.head(5).iterrows()
        ]

        top_score_rows: List[Dict[str, Any]] = []
        bottom_score_rows: List[Dict[str, Any]] = []
        if prioritized_score and prioritized_score in results_df.columns:
            scored = results_df.copy()
            scored["__cdp_score__"] = pd.to_numeric(scored[prioritized_score], errors="coerce")
            scored = scored[scored["__cdp_score__"].notna()]
            if not scored.empty:
                top_score_rows = [
                    row_to_preview(row_series)
                    for _, row_series in scored.sort_values("__cdp_score__", ascending=False).head(5).iterrows()
                ]
                bottom_score_rows = [
                    row_to_preview(row_series)
                    for _, row_series in scored.sort_values("__cdp_score__", ascending=True).head(5).iterrows()
                ]

        return {
            "preview_columns": preview_columns,
            "first_rows": first_rows,
            "top_score_rows": top_score_rows,
            "bottom_score_rows": bottom_score_rows,
        }
