"""
Master catalog → Magento row normalization (active integration).

Snapshot rows come from magento.master_catalog_rows.build_magento_snapshot_rows_from_master
(channel exports + variation relations). normalize_plytix_df in normalize.py is legacy
reference only (old Plytix feed ingest).
"""

from __future__ import annotations

from typing import Dict, List, Optional, Tuple

import pandas as pd

from normalize import (
    _RELATION_FIELD_COLS,
    _axis_codes_for_row,
    _case_insensitive_col,
    _extract_axis_codes,
    _filter_value_by_options,
    _format_kv_blob,
    _get_child_value,
    _has_built_configurable_variations,
    _infer_axis_codes_from_variations,
    _is_empty,
    _normalize_axis_code,
    _normalize_width_value,
    _parse_kv_blob,
    _seed_children_for_row,
)
from settings import MagentoNormalizeConfig


def normalize_master_catalog_df(
    df: pd.DataFrame,
    cfg: MagentoNormalizeConfig,
    allowed_attribute_codes: Optional[set[str]] = None,
    allowed_attribute_options: Optional[Dict[str, List[str]]] = None,
    enforce_required_name: bool = False,
    enforce_required_price: bool = False,
    prefer_seo_title: bool = True,
) -> Tuple[pd.DataFrame, pd.DataFrame]:
    """
    Normalize master-catalog snapshot rows for Magento sync_planner / sync_service.

    Relation fields (variant_list, configurable_variations, …) are usually pre-built
    by db.master_relations; this step validates them, packs custom attributes, and
    quarantines incomplete configurable parents.
    """
    out = df.copy()

    if cfg.drop_before_pack:
        drop_actual = []
        for c in cfg.drop_before_pack:
            actual = _case_insensitive_col(out, c) or c
            if actual in out.columns:
                drop_actual.append(actual)
        if drop_actual:
            out = out.drop(columns=drop_actual)

    for c in cfg.keep_cols:
        if c not in out.columns:
            out[c] = ""

    if cfg.empty_to_default_attribute_set:
        if "attribute_set_code" not in out.columns:
            out["attribute_set_code"] = cfg.default_attribute_set
        else:
            out["attribute_set_code"] = out["attribute_set_code"].fillna("").astype(str).str.strip()
            out.loc[out["attribute_set_code"].eq(""), "attribute_set_code"] = cfg.default_attribute_set

    if "product_online" in out.columns:
        out["product_online"] = out["product_online"].fillna("").astype(str).str.strip().map(
            lambda v: "1" if v.lower() in {"1", "true", "yes", "y", "on"} else "0"
        )

    if "product_type" in out.columns and "price" in out.columns:
        configurable_mask = out["product_type"].fillna("").astype(str).str.strip().str.lower().eq("configurable")
        out["price"] = out["price"].astype(str)
        out.loc[configurable_mask, "price"] = "0"

    if prefer_seo_title:
        # Only fill empty names from SEO overlays. Never overwrite a real master
        # catalog name with a possibly-stale meta_title/seo_title attribute.
        seo_col = _case_insensitive_col(out, "seo_title")
        meta_col = _case_insensitive_col(out, "meta_title")
        title_col = seo_col or meta_col
        name_col = _case_insensitive_col(out, "name") or "name"
        if title_col:
            if name_col not in out.columns:
                out[name_col] = ""
            title_series = out[title_col].fillna("").astype(str).str.strip()
            name_series = out[name_col].fillna("").astype(str).str.strip()
            fill_mask = name_series.eq("") & title_series.ne("")
            out.loc[fill_mask, name_col] = title_series[fill_mask]

    sku_col = _case_insensitive_col(out, "sku") or "sku"
    seed_col = _case_insensitive_col(out, cfg.seed_children_col) or cfg.seed_children_col
    axis_col = _case_insensitive_col(out, cfg.axis_col) or cfg.axis_col
    labels_col = _case_insensitive_col(out, cfg.labels_col) or cfg.labels_col
    add_col = _case_insensitive_col(out, cfg.additional_col) or cfg.additional_col

    if "product_type" in out.columns and "name" in out.columns:
        configurable_mask = (
            out["product_type"].fillna("").astype(str).str.strip().str.lower().eq("configurable")
        )
        seo_title_col = _case_insensitive_col(out, "seo_title") or _case_insensitive_col(out, "meta_title")
        if seo_title_col:
            has_seo_mask = out[seo_title_col].fillna("").astype(str).str.strip().ne("")
        else:
            has_seo_mask = pd.Series(False, index=out.index)
        needs_sku_suffix = configurable_mask & ~has_seo_mask
        sku_series = out[sku_col].fillna("").astype(str).str.strip()
        out.loc[needs_sku_suffix, "name"] = (
            out.loc[needs_sku_suffix, "name"].fillna("").astype(str).str.strip()
            + " - "
            + sku_series[needs_sku_suffix]
        )

    seed_series = out[seed_col].copy() if seed_col in out.columns else pd.Series("", index=out.index)

    keep_set = {c.lower() for c in cfg.keep_cols}
    allowed_set = (
        {c.lower() for c in allowed_attribute_codes}
        if allowed_attribute_codes
        else None
    )
    options_map = (
        {k.lower(): v for k, v in (allowed_attribute_options or {}).items()}
        if allowed_attribute_options
        else None
    )

    axis_codes_all = set()
    for _, row in out.iterrows():
        axes = _axis_codes_for_row(row, axis_col=axis_col, labels_col=labels_col)
        for axis in axes:
            norm_axis = _normalize_axis_code(axis, list(out.columns), allowed_attribute_codes)
            axis_codes_all.add(norm_axis)

    allowed_for_extra = (
        (allowed_set or set()) | axis_codes_all
        if allowed_set is not None
        else None
    )

    extra_cols = [
        c
        for c in out.columns
        if str(c).strip().lower() not in keep_set
        and str(c).strip().lower() not in _RELATION_FIELD_COLS
        and (allowed_for_extra is None or str(c).strip().lower() in allowed_for_extra)
    ]
    extra_cols_lower = {str(c).strip().lower() for c in extra_cols}
    if "width" in extra_cols_lower and "width_in" in extra_cols_lower:
        extra_cols = [c for c in extra_cols if str(c).strip().lower() != "width_in"]

    quarantine_rows: List[Dict[str, object]] = []

    if "configurable_attributes" not in out.columns:
        out["configurable_attributes"] = ""

    preserved_cv = (
        out["configurable_variations"].copy()
        if "configurable_variations" in out.columns
        else pd.Series("", index=out.index)
    )
    out["configurable_variations"] = ""

    for idx, row in out.iterrows():
        seed_children = _seed_children_for_row(row, seed_series.loc[idx])
        if not seed_children:
            continue

        existing_cv = str(preserved_cv.loc[idx] if idx in preserved_cv.index else "").strip()
        if _has_built_configurable_variations(existing_cv):
            out.at[idx, "configurable_variations"] = existing_cv
            existing_attrs = str(row.get("configurable_attributes", "") or "").strip()
            if not existing_attrs:
                from magento.configurable_requirements import sanitize_attribute_code_list

                existing_attrs = ",".join(
                    sanitize_attribute_code_list(_infer_axis_codes_from_variations(existing_cv))
                )
            else:
                from magento.configurable_requirements import sanitize_attribute_code_list

                existing_attrs = ",".join(sanitize_attribute_code_list(_extract_axis_codes(existing_attrs)))
            if existing_attrs:
                out.at[idx, "configurable_attributes"] = existing_attrs
            if "variant_list" in out.columns and _is_empty(row.get("variant_list", "")):
                out.at[idx, "variant_list"] = ",".join(seed_children)
            continue

        axes = _axis_codes_for_row(row, axis_col=axis_col, labels_col=labels_col)
        axes = [_normalize_axis_code(a, list(out.columns), allowed_attribute_codes) for a in axes]

        if not axes:
            quarantine_rows.append({
                "row_index": idx,
                "sku": row.get(sku_col, ""),
                "reason": "Missing axis codes (configurable_variation_axis or configurable_variation_labels)",
                "seed_children": str(seed_series.loc[idx]),
            })
            continue

        parts: List[str] = []
        missing: List[str] = []

        for child_sku in seed_children:
            axis_pairs: List[str] = []
            ok = True
            for a in axes:
                val = _get_child_value(out, child_sku, a)
                if _is_empty(val):
                    missing.append(f"{child_sku}:{a}")
                    ok = False
                else:
                    axis_pairs.append(f"{a}={val}")
            if not ok:
                continue
            parts.append(f"sku={child_sku}," + ",".join(axis_pairs))

        if not parts:
            quarantine_rows.append({
                "row_index": idx,
                "sku": row.get(sku_col, ""),
                "reason": "No child had complete axis values",
                "seed_children": str(seed_series.loc[idx]),
                "missing_axis_values": ";".join(missing)[:5000],
            })
            continue

        out.at[idx, "configurable_variations"] = "|".join(parts)
        from magento.configurable_requirements import sanitize_attribute_code_list

        out.at[idx, "configurable_attributes"] = ",".join(sanitize_attribute_code_list(axes))

        if "variant_list" in out.columns and _is_empty(row.get("variant_list", "")):
            out.at[idx, "variant_list"] = ",".join(seed_children)

    child_to_parent: Dict[str, str] = {}
    for idx, row in out.iterrows():
        parent_sku = str(row.get(sku_col, "")).strip()
        seed_children = _seed_children_for_row(row, seed_series.loc[idx])
        if not parent_sku or not seed_children:
            continue
        for child in seed_children:
            if child not in child_to_parent:
                child_to_parent[child] = parent_sku

    if "variant_of" in out.columns:
        out["variant_of"] = out["variant_of"].apply(lambda v: "" if _is_empty(v) else str(v).strip())
        for idx, row in out.iterrows():
            sku_val = str(row.get(sku_col, "")).strip()
            if sku_val and _is_empty(row.get("variant_of", "")) and sku_val in child_to_parent:
                out.at[idx, "variant_of"] = child_to_parent[sku_val]

    invalid_indices = set()
    for idx, row in out.iterrows():
        sku_val = row.get(sku_col, "")
        name_val = row.get("name", "")
        if _is_empty(sku_val):
            quarantine_rows.append({"row_index": idx, "sku": "", "reason": "Missing sku"})
            invalid_indices.add(idx)
            continue
        if enforce_required_name and _is_empty(name_val):
            quarantine_rows.append({"row_index": idx, "sku": sku_val, "reason": "Missing name"})
            invalid_indices.add(idx)
            continue
        price_val = row.get("price", "")
        if enforce_required_price and _is_empty(price_val):
            quarantine_rows.append({"row_index": idx, "sku": sku_val, "reason": "Missing price"})
            invalid_indices.add(idx)
            continue

        product_type = str(row.get("product_type", "")).strip().lower()
        if product_type == "configurable" and _is_empty(row.get("configurable_variations", "")):
            quarantine_rows.append({
                "row_index": idx,
                "sku": sku_val,
                "reason": "Missing configurable variations",
            })
            invalid_indices.add(idx)

    additional_out: List[str] = []

    for _, row in out.iterrows():
        merged: Dict[str, str] = {}

        existing = _parse_kv_blob(row.get(add_col, ""))
        for k, v in existing.items():
            if _is_empty(v):
                continue
            filtered = _filter_value_by_options(k, str(v).strip(), options_map)
            if filtered is None:
                continue
            merged[k] = filtered

        for c in extra_cols:
            v = row.get(c, "")
            if _is_empty(v):
                continue
            key = str(c).strip()
            val_text = str(v).strip()
            if key.strip().lower() == "width":
                val_text = _normalize_width_value(val_text)
            filtered = _filter_value_by_options(key, val_text, options_map)
            if filtered is None:
                continue
            merged[str(c).strip()] = filtered

        for c in (cfg.also_pack_cols or []):
            if allowed_set is not None and str(c).strip().lower() not in allowed_set:
                continue
            actual = _case_insensitive_col(out, c) or c
            if actual not in out.columns:
                continue
            v = row.get(actual, "")
            if _is_empty(v):
                continue
            filtered = _filter_value_by_options(str(c).strip(), str(v).strip(), options_map)
            if filtered is None:
                continue
            merged[str(c).strip()] = filtered

        additional_out.append(_format_kv_blob(merged))

    out["additional_attributes"] = additional_out

    if cfg.drop_after_compute:
        drop_actual = []
        for c in cfg.drop_after_compute:
            actual = _case_insensitive_col(out, c) or c
            lowered = str(actual).strip().lower()
            if actual in out.columns and lowered not in keep_set and lowered not in _RELATION_FIELD_COLS:
                drop_actual.append(actual)
        if drop_actual:
            out = out.drop(columns=drop_actual)

    if invalid_indices:
        out = out.drop(index=list(invalid_indices))

    output_cols = list(cfg.keep_cols)
    for col in _RELATION_FIELD_COLS:
        actual = _case_insensitive_col(out, col) or col
        if actual in out.columns and actual not in output_cols:
            output_cols.append(actual)
    normalized = out[output_cols].copy()
    quarantine_df = (
        pd.DataFrame(quarantine_rows)
        if quarantine_rows
        else pd.DataFrame(columns=["row_index", "sku", "reason", "seed_children", "missing_axis_values"])
    )
    return normalized, quarantine_df