"""
Materialize calculated fields into a GeoPackage-backed QgsVectorLayer.

The dialog in ``dialogs/calculated_field_dialog.py`` collects a spec; this
module commits it: it adds the new column via the layer's data provider
(same pattern used for NetWeaver outputs in ``layer_creator.py``),
evaluates the user expression per-feature with QgsExpression, writes the
values, and commits the edit session.

Returns a pandas DataFrame snapshot of the layer after the change so
callers can refresh their grid view.
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Tuple

import pandas as pd
from qgis.core import (
    QgsExpression,
    QgsExpressionContext,
    QgsExpressionContextUtils,
    QgsField,
    QgsVectorLayer,
)


class CalculatedFieldError(RuntimeError):
    """Raised when a calculated field cannot be materialized."""


@dataclass
class CalculatedFieldOutcome:
    """Summary of a successful materialization."""

    field_name: str
    rows_written: int
    skipped_rows: int
    dataframe: pd.DataFrame


def materialize_calculated_field(
    *,
    layer: QgsVectorLayer,
    field_name: str,
    qvariant_type: int,
    expression_text: str,
) -> CalculatedFieldOutcome:
    """Add *field_name* to *layer* and populate it from *expression_text*.

    Raises ``CalculatedFieldError`` for any failure that prevents the new
    column from being written. The error message is suitable for showing
    directly to the user.
    """
    if not isinstance(layer, QgsVectorLayer) or not layer.isValid():
        raise CalculatedFieldError("The result layer could not be opened for editing.")

    if not field_name:
        raise CalculatedFieldError("Field name is required.")

    existing_names = {field.name().lower() for field in layer.fields()}
    if field_name.lower() in existing_names:
        raise CalculatedFieldError(
            f"A field named '{field_name}' already exists in this table."
        )

    expression = QgsExpression(expression_text)
    if expression.hasParserError():
        raise CalculatedFieldError(
            f"Expression has a parser error: {expression.parserErrorString()}"
        )

    context = QgsExpressionContext()
    context.appendScopes(QgsExpressionContextUtils.globalProjectLayerScopes(layer))
    if not expression.prepare(context):
        raise CalculatedFieldError(
            f"Could not prepare expression: {expression.evalErrorString()}"
        )

    provider = layer.dataProvider()
    if provider is None:
        raise CalculatedFieldError("This layer's data provider does not allow editing.")

    new_field = QgsField(field_name, qvariant_type)
    if not provider.addAttributes([new_field]):
        raise CalculatedFieldError(
            "The data provider rejected the new column. The underlying file may be read-only."
        )
    layer.updateFields()

    new_field_index = layer.fields().indexFromName(field_name)
    if new_field_index < 0:
        raise CalculatedFieldError(
            f"Field '{field_name}' was created but could not be found on the layer."
        )

    rows_written = 0
    skipped_rows = 0
    pending_changes: Dict[int, Dict[int, Any]] = {}

    for feature in layer.getFeatures():
        context.setFeature(feature)
        value = expression.evaluate(context)
        if expression.hasEvalError():
            skipped_rows += 1
            continue
        pending_changes[feature.id()] = {new_field_index: value}
        rows_written += 1

    if pending_changes:
        if not provider.changeAttributeValues(pending_changes):
            raise CalculatedFieldError(
                "The data provider rejected the calculated values."
            )

    layer.updateExtents()
    layer.triggerRepaint()

    refreshed_df = _layer_to_dataframe(layer)
    return CalculatedFieldOutcome(
        field_name=field_name,
        rows_written=rows_written,
        skipped_rows=skipped_rows,
        dataframe=refreshed_df,
    )


def remove_calculated_field(layer: QgsVectorLayer, field_name: str) -> pd.DataFrame:
    """Drop *field_name* from *layer*. Returns the refreshed DataFrame."""
    if not isinstance(layer, QgsVectorLayer) or not layer.isValid():
        raise CalculatedFieldError("The result layer could not be opened for editing.")
    field_index = layer.fields().indexFromName(field_name)
    if field_index < 0:
        raise CalculatedFieldError(f"Field '{field_name}' is not on this layer.")

    provider = layer.dataProvider()
    if provider is None or not provider.deleteAttributes([field_index]):
        raise CalculatedFieldError("The data provider could not remove the column.")
    layer.updateFields()
    return _layer_to_dataframe(layer)


def _layer_to_dataframe(layer: QgsVectorLayer) -> pd.DataFrame:
    """Pull the layer's current attribute table into a pandas DataFrame."""
    columns = [field.name() for field in layer.fields()]
    rows: List[List[Any]] = []
    for feature in layer.getFeatures():
        rows.append([feature.attribute(name) for name in columns])
    if not columns:
        return pd.DataFrame()
    return pd.DataFrame(rows, columns=columns)
