"""Remote-first channel attribute mapping and option promotion.

Supports:
- Listing Magento/Shopify registry attributes with master mappings + type + example
- Adding 1→N alias targets without removing existing mappings (dynamic scope)
- Promoting text attributes to select options seeded from distinct master values
- Auto-seeding new option values when master data gains new labels
"""

from __future__ import annotations

import logging
import re
import time
from datetime import datetime, timezone
from typing import Any, Dict, Iterable, List, Optional, Sequence, Set

import sqlalchemy as sa
from sqlalchemy import func, select
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.orm import Session

from db.channel_exports import resolve_pipeline_channel_code
from db.compat_connections import SHOPIFY_COMPAT_ID_OFFSET, decode_compat_connection_id
from db.canonical_attributes import composite_attribute_codes, derive_composite_attribute_value
from db.master_attributes import MASTER_PRODUCT_CORE_MAP, list_master_attributes
from db.master_product_filters import list_master_attribute_values
from db.models import (
    ChannelAttributeAlias,
    MagentoAttributeOptionRegistry,
    MagentoAttributeRegistry,
    MagentoConnection,
    MasterProduct,
    MasterProductAttributeValue,
    ShopifyAttributeRegistry,
    ShopifyConnection,
)
from db.source_imports import normalize_column_key

logger = logging.getLogger(__name__)

SYNC_OPTIONS_FLAG = "sync_options_from_master"
DERIVED_MASTER_OPTION_CODES = frozenset(composite_attribute_codes())
# HTML / free-text composites are never Magento select options — skip expensive derivation.
_NON_OPTION_DERIVED_CODES = frozenset(
    {
        "additional_information",
        "care_instructions",
        "description",
        "materials",
        "warranty",
    }
)
REMOTE_MAPPER_SCOPE = "dynamic"
REMOTE_MAPPER_NOTES = "remote_mapper:add_target"


def magento_alias_target_codes(session: Session) -> Set[str]:
    """Active Magento channel attribute codes from aliases (for normalize allowlist)."""
    rows = session.scalars(
        select(ChannelAttributeAlias.channel_attribute_code)
        .where(ChannelAttributeAlias.channel_code == "magento")
        .where(ChannelAttributeAlias.is_active.is_(True))
    ).all()
    return {normalize_column_key(code) for code in rows if normalize_column_key(code)}


def expand_magento_allowed_attribute_codes(
    session: Session,
    base_codes: Optional[Iterable[str]] = None,
) -> Set[str]:
    """Union JSON allowlist with Magento alias targets so mapped DigiSilk attrs are not dropped."""
    codes = {normalize_column_key(c) for c in (base_codes or []) if normalize_column_key(c)}
    codes |= magento_alias_target_codes(session)
    return codes


def list_remote_channel_attributes(
    session: Session,
    *,
    channel_type: str,
    connection_id: int,
    unmapped_only: bool = False,
    search: Optional[str] = None,
    limit: int = 2000,
) -> Dict[str, Any]:
    """Remote attributes for a Magento/Shopify connection with master mapping context."""
    channel = resolve_pipeline_channel_code(channel_type)
    native_id = _resolve_native_connection_id(channel, connection_id)
    if native_id is None:
        raise ValueError(f"Invalid connection_id for {channel}: {connection_id}")

    master_by_code = {row["attribute_code"]: row for row in list_master_attributes(session)}
    aliases_by_remote = _aliases_by_remote(session, channel)

    if channel == "magento":
        remote_rows = _list_magento_attributes(session, native_id)
        connection_meta = _magento_connection_meta(session, native_id)
    elif channel == "shopify":
        remote_rows = _list_shopify_attributes(session, native_id)
        connection_meta = _shopify_connection_meta(session, native_id)
    else:
        raise ValueError(f"Unsupported channel_type: {channel_type}")

    needle = (search or "").strip().lower()
    attributes: List[Dict[str, Any]] = []
    for remote in remote_rows:
        code = remote["attribute_code"]
        mapping_rows = aliases_by_remote.get(code, [])
        mapped_masters = []
        for alias in mapping_rows:
            master = master_by_code.get(alias.canonical_code) or {
                "attribute_code": alias.canonical_code,
                "label": alias.canonical_code,
                "data_type": alias.data_type or "text",
            }
            example_values = list_master_attribute_values(session, alias.canonical_code, limit=3)
            mapped_masters.append(
                {
                    "canonical_code": alias.canonical_code,
                    "label": master.get("label") or alias.canonical_code,
                    "master_data_type": master.get("data_type") or alias.data_type or "text",
                    "alias_id": alias.id,
                    "alias_data_type": alias.data_type or remote.get("data_type") or "text",
                    "mapping_scope": alias.mapping_scope,
                    "sync_options_from_master": bool((alias.transform_rule or {}).get(SYNC_OPTIONS_FLAG)),
                    "example_values": example_values,
                    "example": example_values[0] if example_values else "",
                }
            )

        if unmapped_only and mapped_masters:
            continue
        if needle:
            hay = f"{code} {remote.get('label') or ''} {' '.join(m['canonical_code'] for m in mapped_masters)}".lower()
            if needle not in hay:
                continue

        attributes.append(
            {
                **remote,
                "mapped": bool(mapped_masters),
                "mappings": mapped_masters,
                "primary_master_code": mapped_masters[0]["canonical_code"] if mapped_masters else None,
                "primary_master_label": mapped_masters[0]["label"] if mapped_masters else None,
                "example": mapped_masters[0]["example"] if mapped_masters else "",
                "can_convert_to_options": _can_convert_to_options(remote, mapped_masters),
            }
        )
        if len(attributes) >= limit:
            break

    mapped_count = sum(1 for row in attributes if row["mapped"])
    return {
        "channel_type": channel,
        "connection_id": native_id,
        "connection": connection_meta,
        "count": len(attributes),
        "mapped_count": mapped_count,
        "unmapped_count": len(attributes) - mapped_count,
        "attributes": attributes,
        "master_attributes": [
            {"attribute_code": m["attribute_code"], "label": m["label"], "data_type": m["data_type"]}
            for m in master_by_code.values()
        ],
    }


def add_remote_attribute_mapping(
    session: Session,
    *,
    channel_type: str,
    channel_attribute_code: str,
    canonical_code: str,
    data_type: Optional[str] = None,
    sync_options_from_master: bool = False,
) -> Dict[str, Any]:
    """Map a remote attribute to a master attribute without removing other targets (1→N)."""
    channel = resolve_pipeline_channel_code(channel_type)
    remote_code = normalize_column_key(channel_attribute_code)
    master_code = normalize_column_key(canonical_code)
    if not remote_code or not master_code:
        raise ValueError("channel_attribute_code and canonical_code are required")

    transform: Dict[str, Any] = {}
    if sync_options_from_master:
        transform[SYNC_OPTIONS_FLAG] = True

    payload = {
        "channel_code": channel,
        "canonical_code": master_code,
        "channel_attribute_code": remote_code,
        "mapping_scope": REMOTE_MAPPER_SCOPE,
        "action": "use_existing",
        "data_type": normalize_column_key(data_type) if data_type else None,
        "transform_rule": transform or None,
        "is_active": True,
        "notes": REMOTE_MAPPER_NOTES,
    }
    stmt = insert(ChannelAttributeAlias).values(payload)
    stmt = stmt.on_conflict_do_update(
        constraint="uq_channel_attribute_alias_channel_canonical_attr",
        set_={
            "mapping_scope": REMOTE_MAPPER_SCOPE,
            "action": "use_existing",
            "data_type": stmt.excluded.data_type,
            "transform_rule": stmt.excluded.transform_rule,
            "is_active": True,
            "notes": REMOTE_MAPPER_NOTES,
            "updated_at": sa.func.now(),
        },
    )
    session.execute(stmt)
    alias = session.scalar(
        select(ChannelAttributeAlias)
        .where(ChannelAttributeAlias.channel_code == channel)
        .where(ChannelAttributeAlias.canonical_code == master_code)
        .where(ChannelAttributeAlias.channel_attribute_code == remote_code)
    )
    return {
        "status": "ok",
        "channel_code": channel,
        "canonical_code": master_code,
        "channel_attribute_code": remote_code,
        "alias_id": alias.id if alias else None,
        "mapping_scope": REMOTE_MAPPER_SCOPE,
    }


def remove_remote_attribute_mapping(
    session: Session,
    *,
    channel_type: str,
    channel_attribute_code: str,
    canonical_code: str,
) -> Dict[str, Any]:
    """Deactivate a remote↔master alias (does not touch other targets)."""
    channel = resolve_pipeline_channel_code(channel_type)
    remote_code = normalize_column_key(channel_attribute_code)
    master_code = normalize_column_key(canonical_code)
    alias = session.scalar(
        select(ChannelAttributeAlias)
        .where(ChannelAttributeAlias.channel_code == channel)
        .where(ChannelAttributeAlias.canonical_code == master_code)
        .where(ChannelAttributeAlias.channel_attribute_code == remote_code)
    )
    if alias is None:
        return {"status": "not_found", "removed": False}
    alias.is_active = False
    alias.updated_at = datetime.utcnow()
    return {
        "status": "ok",
        "removed": True,
        "channel_code": channel,
        "canonical_code": master_code,
        "channel_attribute_code": remote_code,
        "alias_id": alias.id,
    }


def promote_attribute_to_options(
    session: Session,
    *,
    channel_type: str,
    connection_id: int,
    channel_attribute_code: str,
    canonical_code: Optional[str] = None,
    dry_run: bool = False,
) -> Dict[str, Any]:
    """Scan master values and create/seed remote select options; enable future auto-sync."""
    channel = resolve_pipeline_channel_code(channel_type)
    native_id = _resolve_native_connection_id(channel, connection_id)
    if native_id is None:
        raise ValueError(f"Invalid connection_id for {channel}: {connection_id}")

    remote_code = normalize_column_key(channel_attribute_code)
    master_code = normalize_column_key(canonical_code) if canonical_code else None
    if not master_code:
        aliases = _aliases_by_remote(session, channel).get(remote_code) or []
        if not aliases:
            raise ValueError(
                f"No master mapping for {remote_code}; pass canonical_code or map the attribute first"
            )
        master_code = aliases[0].canonical_code

    values = list_all_master_attribute_values(session, master_code)

    add_remote_attribute_mapping(
        session,
        channel_type=channel,
        channel_attribute_code=remote_code,
        canonical_code=master_code,
        data_type="select",
        sync_options_from_master=True,
    )

    result: Dict[str, Any] = {
        "status": "ok",
        "channel_type": channel,
        "connection_id": native_id,
        "channel_attribute_code": remote_code,
        "canonical_code": master_code,
        "distinct_values": len(values),
        "values_sample": values[:20],
        "dry_run": dry_run,
        "created_options": [],
        "existing_options": [],
        "errors": [],
        "sync_options_from_master": True,
    }

    if channel == "magento":
        seed = _seed_magento_options(
            session,
            connection_id=native_id,
            attribute_code=remote_code,
            labels=values,
            dry_run=dry_run,
        )
        result.update(seed)
    elif channel == "shopify":
        # Shopify metafields with choice lists are provisioned via registry; store flag for push-time sync.
        result["message"] = (
            "Shopify option sync flagged on alias; values are enforced as metafield choice "
            "labels when the metafield definition supports validations."
        )
        if not dry_run:
            _ensure_shopify_registry_type(session, native_id, remote_code)
    else:
        raise ValueError(f"Unsupported channel for option promotion: {channel}")

    if result.get("errors"):
        result["status"] = "partial"

    return result


def sync_option_values_from_master(
    session: Session,
    *,
    channel_type: str = "magento",
    connection_id: Optional[int] = None,
    dry_run: bool = False,
    attribute_codes: Optional[Iterable[str]] = None,
) -> Dict[str, Any]:
    """Create missing remote option values for aliases flagged sync_options_from_master."""
    started = time.perf_counter()
    channel = resolve_pipeline_channel_code(channel_type)
    wanted_codes = {
        normalize_column_key(code)
        for code in (attribute_codes or [])
        if normalize_column_key(code)
    }
    aliases = [
        alias
        for alias in session.scalars(
            select(ChannelAttributeAlias)
            .where(ChannelAttributeAlias.channel_code == channel)
            .where(ChannelAttributeAlias.is_active.is_(True))
        ).all()
        if (alias.transform_rule or {}).get(SYNC_OPTIONS_FLAG)
        and (
            not wanted_codes
            or normalize_column_key(alias.channel_attribute_code) in wanted_codes
            or normalize_column_key(alias.canonical_code) in wanted_codes
        )
    ]
    if not aliases:
        return {"status": "ok", "channel": channel, "aliases": 0, "results": [], "elapsed_ms": 0}

    native_id = _resolve_native_connection_id(channel, connection_id) if connection_id else None
    if channel == "magento" and native_id is None:
        native_id = session.scalar(select(MagentoConnection.id).order_by(MagentoConnection.id).limit(1))

    # Load lightweight field maps once for all composite/derived option codes.
    needs_derived = any(
        normalize_column_key(alias.canonical_code) in DERIVED_MASTER_OPTION_CODES
        and normalize_column_key(alias.canonical_code) not in _NON_OPTION_DERIVED_CODES
        for alias in aliases
    )
    field_maps: Optional[List[Dict[str, Any]]] = (
        _load_active_master_field_maps(session) if needs_derived else None
    )
    if field_maps is not None:
        logger.info(
            "sync_option_values_from_master: loaded %d product field maps for derived options",
            len(field_maps),
        )

    results = []
    for alias in aliases:
        alias_started = time.perf_counter()
        values = list_all_master_attribute_values(
            session,
            alias.canonical_code,
            field_maps=field_maps,
        )
        if channel == "magento" and native_id is not None:
            seed = _seed_magento_options(
                session,
                connection_id=native_id,
                attribute_code=alias.channel_attribute_code,
                labels=values,
                dry_run=dry_run,
            )
            results.append(
                {
                    "canonical_code": alias.canonical_code,
                    "channel_attribute_code": alias.channel_attribute_code,
                    "distinct_values": len(values),
                    "elapsed_ms": int((time.perf_counter() - alias_started) * 1000),
                    **seed,
                }
            )
        else:
            results.append(
                {
                    "canonical_code": alias.canonical_code,
                    "channel_attribute_code": alias.channel_attribute_code,
                    "distinct_values": len(values),
                    "elapsed_ms": int((time.perf_counter() - alias_started) * 1000),
                    "skipped": True,
                }
            )
    elapsed_ms = int((time.perf_counter() - started) * 1000)
    logger.info(
        "sync_option_values_from_master: channel=%s aliases=%d elapsed_ms=%d",
        channel,
        len(aliases),
        elapsed_ms,
    )
    return {
        "status": "ok",
        "channel": channel,
        "aliases": len(aliases),
        "results": results,
        "elapsed_ms": elapsed_ms,
    }


# ---------------------------------------------------------------------------
# Internals
# ---------------------------------------------------------------------------


def list_all_master_attribute_values(
    session: Session,
    attribute_code: str,
    *,
    limit: int = 5000,
    field_maps: Optional[Sequence[Dict[str, Any]]] = None,
) -> List[str]:
    """Distinct master values for option seeding (higher cap than autocomplete API).

    ``collection`` uses identified canonical collection names (landing pages /
    registry), not raw product.collection strings.

    ``depth`` is derived from assembled_depth / variation_depth_in (etc.); master
    rarely stores attribute_code ``depth``.

    ``field_maps`` is an optional preloaded product field cache used for composite
    attribute derivation so callers can avoid rebuilding it per attribute.
    """
    code = normalize_column_key(attribute_code)
    if not code:
        return []
    limit = max(1, min(int(limit or 5000), 20000))

    if code == "collection":
        from db.collection_path_guard import canonical_collection_option_labels

        return canonical_collection_option_labels(session, limit=limit)

    if code == "depth":
        return _list_canonical_depth_option_labels(session, limit=limit)

    if code in _NON_OPTION_DERIVED_CODES:
        return []

    core_by_canonical = {canonical: column for column, canonical in MASTER_PRODUCT_CORE_MAP.items()}
    core_column = core_by_canonical.get(code)
    if core_column and hasattr(MasterProduct, core_column):
        column = getattr(MasterProduct, core_column)
        return [
            str(value).strip()
            for (value,) in session.execute(
                select(func.distinct(column))
                .select_from(MasterProduct)
                .where(MasterProduct.is_active.is_(True))
                .where(column.is_not(None))
                .where(column != "")
                .order_by(column)
                .limit(limit)
            ).all()
            if str(value).strip()
        ]

    values = [
        str(value).strip()
        for (value,) in session.execute(
            select(func.distinct(MasterProductAttributeValue.value))
            .join(MasterProduct, MasterProduct.id == MasterProductAttributeValue.product_id)
            .where(MasterProduct.is_active.is_(True))
            .where(MasterProductAttributeValue.attribute_code == code)
            .where(MasterProductAttributeValue.value.is_not(None))
            .where(MasterProductAttributeValue.value != "")
            .order_by(MasterProductAttributeValue.value)
            .limit(limit)
        ).all()
        if str(value).strip()
    ]
    if code in DERIVED_MASTER_OPTION_CODES:
        derived = _list_derived_master_attribute_values(
            session,
            code,
            limit=limit,
            field_maps=field_maps,
        )
        if derived:
            seen = {str(value).strip() for value in values if str(value).strip()}
            for value in derived:
                text = str(value).strip()
                if text and text not in seen:
                    values.append(text)
                    seen.add(text)
    return values


def _list_canonical_depth_option_labels(session: Session, *, limit: int) -> List[str]:
    """Distinct Magento ``depth`` option labels derived from assembled/variation sources."""
    from db.canonical_attributes import DEPTH_OPTION_SOURCE_CODES, normalize_depth_option_label

    raw_values = [
        str(value).strip()
        for (value,) in session.execute(
            select(func.distinct(MasterProductAttributeValue.value))
            .join(MasterProduct, MasterProduct.id == MasterProductAttributeValue.product_id)
            .where(MasterProduct.is_active.is_(True))
            .where(MasterProductAttributeValue.attribute_code.in_(DEPTH_OPTION_SOURCE_CODES))
            .where(MasterProductAttributeValue.value.is_not(None))
            .where(MasterProductAttributeValue.value != "")
        ).all()
        if str(value).strip()
    ]
    seen: Set[str] = set()
    labels: List[str] = []
    for raw in raw_values:
        normalized = normalize_depth_option_label(raw)
        if not normalized or normalized in seen:
            continue
        seen.add(normalized)
        labels.append(normalized)

    def _sort_key(label: str) -> tuple:
        try:
            return (0, float(label))
        except (TypeError, ValueError):
            return (1, 0.0)

    labels.sort(key=_sort_key)
    return labels[:limit]


def _load_active_master_field_maps(session: Session) -> List[Dict[str, Any]]:
    """Lightweight per-product field maps for composite option derivation.

    Avoids ``build_channel_product_payloads`` (taxonomy/aliases/images/overrides),
    which previously made sync_options_from_master take many minutes per alias.
    """
    products = session.scalars(
        select(MasterProduct).where(MasterProduct.is_active.is_(True))
    ).all()
    if not products:
        return []

    by_id: Dict[int, Dict[str, Any]] = {}
    for product in products:
        fields: Dict[str, Any] = {"sku": product.sku}
        for column, canonical in MASTER_PRODUCT_CORE_MAP.items():
            value = getattr(product, column, None)
            if value is None:
                continue
            text = str(value).strip()
            if text:
                fields[canonical] = text
        by_id[product.id] = fields

    product_ids = list(by_id.keys())
    # Chunk IN clauses for large catalogs.
    chunk_size = 2000
    for offset in range(0, len(product_ids), chunk_size):
        chunk = product_ids[offset : offset + chunk_size]
        rows = session.execute(
            select(
                MasterProductAttributeValue.product_id,
                MasterProductAttributeValue.attribute_code,
                MasterProductAttributeValue.value,
            ).where(MasterProductAttributeValue.product_id.in_(chunk))
        ).all()
        for product_id, attr_code, value in rows:
            fields = by_id.get(int(product_id))
            if fields is None:
                continue
            code = normalize_column_key(attr_code)
            text = str(value or "").strip()
            if code and text:
                fields[code] = text
    return list(by_id.values())


def _list_derived_master_attribute_values(
    session: Session,
    attribute_code: str,
    *,
    limit: int,
    field_maps: Optional[Sequence[Dict[str, Any]]] = None,
) -> List[str]:
    code = normalize_column_key(attribute_code)
    if not code or code in _NON_OPTION_DERIVED_CODES:
        return []

    maps = list(field_maps) if field_maps is not None else _load_active_master_field_maps(session)
    values: List[str] = []
    seen: set[str] = set()
    for fields in maps:
        text = str(derive_composite_attribute_value(code, fields) or "").strip()
        if not text or text in seen:
            continue
        # Skip obvious HTML blobs — not Magento select option labels.
        if "<" in text and ">" in text:
            continue
        values.append(text)
        seen.add(text)
        if len(values) >= limit:
            break
    return values


def _aliases_by_remote(session: Session, channel: str) -> Dict[str, List[ChannelAttributeAlias]]:
    rows: Dict[str, List[ChannelAttributeAlias]] = {}
    for alias in session.scalars(
        select(ChannelAttributeAlias)
        .where(ChannelAttributeAlias.channel_code == channel)
        .where(ChannelAttributeAlias.is_active.is_(True))
        .order_by(ChannelAttributeAlias.canonical_code)
    ).all():
        code = normalize_column_key(alias.channel_attribute_code)
        if code:
            rows.setdefault(code, []).append(alias)
    return rows


def _list_magento_attributes(session: Session, connection_id: int) -> List[Dict[str, Any]]:
    rows = session.scalars(
        select(MagentoAttributeRegistry)
        .where(MagentoAttributeRegistry.connection_id == connection_id)
        .order_by(MagentoAttributeRegistry.attribute_code)
    ).all()
    return [
        {
            "attribute_code": normalize_column_key(row.attribute_code),
            "label": row.frontend_label or row.attribute_code,
            "data_type": row.frontend_input or row.backend_type or "text",
            "frontend_input": row.frontend_input,
            "backend_type": row.backend_type,
            "is_user_defined": row.is_user_defined,
            "attribute_id": row.attribute_id,
        }
        for row in rows
        if normalize_column_key(row.attribute_code)
    ]


def _list_shopify_attributes(session: Session, connection_id: int) -> List[Dict[str, Any]]:
    conn = session.get(ShopifyConnection, connection_id)
    shop_code = (conn.shop_code if conn else None) or ""
    stmt = select(ShopifyAttributeRegistry).order_by(ShopifyAttributeRegistry.attribute_code)
    if shop_code:
        stmt = stmt.where(ShopifyAttributeRegistry.shop_code == shop_code)
    rows = session.scalars(stmt).all()
    return [
        {
            "attribute_code": normalize_column_key(row.attribute_code),
            "label": row.name or row.attribute_code,
            "data_type": row.data_type or "single_line_text_field",
            "namespace": row.namespace,
            "key": row.key,
            "owner_type": row.owner_type,
            "is_standard": row.is_standard,
        }
        for row in rows
        if normalize_column_key(row.attribute_code)
    ]


def _magento_connection_meta(session: Session, connection_id: int) -> Dict[str, Any]:
    conn = session.get(MagentoConnection, connection_id)
    if not conn:
        return {"id": connection_id}
    return {
        "id": conn.id,
        "display_name": f"Magento #{conn.id} ({conn.environment})",
        "store_base_url": conn.store_base_url,
        "environment": conn.environment,
        "status": conn.status,
    }


def _shopify_connection_meta(session: Session, connection_id: int) -> Dict[str, Any]:
    conn = session.get(ShopifyConnection, connection_id)
    if not conn:
        return {"id": connection_id}
    display_name = (
        str(getattr(conn, "display_name", "") or "").strip()
        or str(getattr(conn, "shop_code", "") or "").strip()
        or str(getattr(conn, "shop_domain", "") or "").strip()
        or f"Shopify #{conn.id}"
    )
    return {
        "id": conn.id,
        "display_name": display_name,
        "shop_code": conn.shop_code,
        "shop_domain": conn.shop_domain,
        "environment": conn.environment,
        "status": conn.status,
    }


def _resolve_native_connection_id(channel: str, connection_id: Optional[int]) -> Optional[int]:
    if connection_id is None:
        return None
    if connection_id >= SHOPIFY_COMPAT_ID_OFFSET:
        channel_type, native_id = decode_compat_connection_id(connection_id)
        return native_id if channel_type == channel else None
    return connection_id


def _can_convert_to_options(remote: Dict[str, Any], mappings: List[Dict[str, Any]]) -> bool:
    if not mappings:
        return False
    data_type = str(remote.get("data_type") or remote.get("frontend_input") or "").lower()
    if data_type in {"select", "multiselect", "dropdown", "option"}:
        return True
    # Text-like remotes can still be promoted when we create/update as select.
    return data_type in {"", "text", "varchar", "string", "input field", "single_line_text_field"}


def _seed_magento_options(
    session: Session,
    *,
    connection_id: int,
    attribute_code: str,
    labels: List[str],
    dry_run: bool,
) -> Dict[str, Any]:
    from db.magento_repositories import SqlAlchemyMagentoConnectionRepository
    from magento.magento_api import MagentoOAuthClient, MagentoRestClient
    from magento.oauth_client import build_magento_oauth_kwargs

    code = normalize_column_key(attribute_code)
    existing = {
        normalize_column_key(row.option_label_norm) or normalize_column_key(row.option_label): row
        for row in session.scalars(
            select(MagentoAttributeOptionRegistry)
            .where(MagentoAttributeOptionRegistry.connection_id == connection_id)
            .where(MagentoAttributeOptionRegistry.attribute_code == code)
        ).all()
    }
    # Also key by raw label lower for matching.
    existing_labels = {
        str(row.option_label or "").strip().lower(): row for row in existing.values() if row.option_label
    }

    cleaned_labels: List[str] = []
    seen: Set[str] = set()
    for label in labels:
        text = str(label or "").strip()
        if not text:
            continue
        key = text.lower()
        if key in seen:
            continue
        seen.add(key)
        cleaned_labels.append(text)

    created: List[str] = []
    already: List[str] = []
    errors: List[str] = []

    missing = [label for label in cleaned_labels if label.lower() not in existing_labels]
    for label in cleaned_labels:
        if label.lower() in existing_labels:
            already.append(label)

    if dry_run:
        return {
            "created_options": [],
            "existing_options": already,
            "would_create_options": missing,
            "errors": [],
        }

    if not missing:
        return {
            "created_options": [],
            "existing_options": already,
            "would_create_options": [],
            "errors": [],
        }

    repo = SqlAlchemyMagentoConnectionRepository(session)
    conn = repo.get_for_sync(connection_id)
    if not conn:
        return {
            "created_options": [],
            "existing_options": already,
            "would_create_options": missing,
            "errors": [f"Magento connection {connection_id} not found"],
        }

    api = MagentoRestClient(
        MagentoOAuthClient(**build_magento_oauth_kwargs(conn))
    )

    # Ensure attribute exists and is select when possible.
    status, attr = api.get_attribute(code)
    if status == 200 and isinstance(attr, dict):
        frontend_input = str(attr.get("frontend_input") or "").lower()
        if frontend_input not in {"select", "multiselect", "swatch_visual", "swatch_text"}:
            # Magento often rejects frontend_input changes; attempt and report.
            upd_status, upd_body = api.update_attribute(
                code,
                {
                    "frontend_input": "select",
                    "backend_type": "int",
                    "is_filterable": True,
                    "is_filterable_in_search": True,
                    "used_in_product_listing": True,
                },
            )
            if upd_status not in (200, 201):
                errors.append(
                    f"Attribute '{code}' is '{frontend_input}' and could not be converted to select "
                    f"(HTTP {upd_status}). Map to an existing DigiSilk select attribute instead."
                )
                return {
                    "created_options": [],
                    "existing_options": already,
                    "would_create_options": missing,
                    "errors": errors,
                }

    for label in missing:
        opt_status, body, err = api.create_attribute_option(code, label)
        if opt_status in (200, 201):
            created.append(label)
            value_index = None
            if isinstance(body, dict):
                value_index = body.get("value") or body.get("option_id")
            elif isinstance(body, (int, float, str)) and str(body).isdigit():
                value_index = int(body)
            if value_index is not None:
                _record_magento_option(
                    session,
                    connection_id=connection_id,
                    attribute_code=code,
                    option_label=label,
                    value_index=int(value_index),
                )
        else:
            # Treat "already exists" style errors as existing.
            msg = (err or str(body) or "").lower()
            if "already" in msg or "exist" in msg:
                already.append(label)
            else:
                errors.append(f"{label}: {err or body}")

    return {
        "created_options": created,
        "existing_options": already,
        "would_create_options": [],
        "errors": errors,
    }


def _record_magento_option(
    session: Session,
    *,
    connection_id: int,
    attribute_code: str,
    option_label: str,
    value_index: int,
) -> None:
    norm = re.sub(r"\s+", " ", option_label.strip().lower())
    payload = {
        "connection_id": connection_id,
        "attribute_code": attribute_code,
        "option_label": option_label,
        "option_label_norm": norm,
        "value_index": value_index,
        "fetched_at": datetime.now(timezone.utc),
    }
    bind = session.get_bind()
    if bind is not None and bind.dialect.name == "postgresql":
        stmt = insert(MagentoAttributeOptionRegistry).values(payload)
        stmt = stmt.on_conflict_do_update(
            constraint="uq_attr_opt_registry_conn_code_norm",
            set_={
                "option_label": stmt.excluded.option_label,
                "value_index": stmt.excluded.value_index,
                "fetched_at": stmt.excluded.fetched_at,
            },
        )
        session.execute(stmt)
        return
    existing = session.scalar(
        select(MagentoAttributeOptionRegistry)
        .where(MagentoAttributeOptionRegistry.connection_id == connection_id)
        .where(MagentoAttributeOptionRegistry.attribute_code == attribute_code)
        .where(MagentoAttributeOptionRegistry.option_label_norm == norm)
    )
    if existing is None:
        session.add(MagentoAttributeOptionRegistry(**payload))
    else:
        existing.option_label = option_label
        existing.value_index = value_index
        existing.fetched_at = payload["fetched_at"]


def _ensure_shopify_registry_type(session: Session, connection_id: int, attribute_code: str) -> None:
    conn = session.get(ShopifyConnection, connection_id)
    if not conn:
        return
    row = session.scalar(
        select(ShopifyAttributeRegistry)
        .where(ShopifyAttributeRegistry.shop_code == conn.shop_code)
        .where(ShopifyAttributeRegistry.attribute_code == attribute_code)
    )
    if row and not (row.data_type or "").startswith("list."):
        # Keep existing type; promotion mainly flags alias for future choice sync.
        pass