"""Create channel attributes/metafields marked create_new in channel_attribute_alias."""

from __future__ import annotations

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

from sqlalchemy import select
from sqlalchemy.dialects.postgresql import insert as pg_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.models import (
    ChannelAttributeAlias,
    MagentoAttributeOptionRegistry,
    MagentoAttributeRegistry,
    ShopifyAttributeRegistry,
    ShopifyConnection,
)

logger = logging.getLogger(__name__)

CREATE_NEW_ACTION = "create_new"
USE_EXISTING_ACTION = "use_existing"
DEFAULT_SHOPIFY_METAFIELD_NAMESPACE = "custom"


def pending_create_aliases(
    session: Session,
    channel_code: str,
    *,
    canonical_codes: Optional[Iterable[str]] = None,
) -> List[ChannelAttributeAlias]:
    channel = resolve_pipeline_channel_code(channel_code)
    stmt = (
        select(ChannelAttributeAlias)
        .where(ChannelAttributeAlias.channel_code == channel)
        .where(ChannelAttributeAlias.is_active.is_(True))
        .where(ChannelAttributeAlias.action == CREATE_NEW_ACTION)
        .order_by(ChannelAttributeAlias.canonical_code)
    )
    wanted_codes = sorted({str(code).strip().lower() for code in (canonical_codes or []) if str(code).strip()})
    if wanted_codes:
        stmt = stmt.where(ChannelAttributeAlias.canonical_code.in_(wanted_codes))
    return list(session.scalars(stmt).all())


def provision_pending_for_channels(
    session: Session,
    channels: Iterable[str],
    *,
    magento_connection_id: Optional[int] = None,
    shopify_connection_id: Optional[int] = None,
    dry_run: bool = False,
) -> Dict[str, Any]:
    """Provision create_new aliases for each requested channel that has a connection."""
    results: Dict[str, Any] = {}
    for channel in channels:
        resolved = resolve_pipeline_channel_code(channel)
        if resolved == "magento":
            if magento_connection_id is None:
                results["magento"] = {"status": "skipped", "reason": "connection_id required"}
                continue
            results["magento"] = provision_channel_attributes(
                session,
                "magento",
                connection_id=magento_connection_id,
                dry_run=dry_run,
            )
            cfg = provision_magento_configurable_catalog(
                session,
                magento_connection_id,
                dry_run=dry_run,
            )
            results["magento"]["configurable_catalog"] = cfg
        elif resolved == "shopify":
            if shopify_connection_id is None:
                results["shopify"] = {"status": "skipped", "reason": "connection_id required"}
                continue
            results["shopify"] = provision_channel_attributes(
                session,
                "shopify",
                connection_id=shopify_connection_id,
                dry_run=dry_run,
            )
    return results


def provision_channel_attributes(
    session: Session,
    channel_code: str,
    *,
    connection_id: Optional[int] = None,
    shop_code: Optional[str] = None,
    canonical_codes: Optional[Iterable[str]] = None,
    dry_run: bool = False,
) -> Dict[str, Any]:
    """Create remote attributes for aliases with action=create_new; flip to use_existing on success."""
    channel = resolve_pipeline_channel_code(channel_code)
    if channel == "magento":
        native_id = _resolve_magento_connection_id(connection_id)
        if native_id is None:
            return {"status": "failed", "error": "magento connection_id is required"}
        return _provision_magento(
            session,
            native_id,
            canonical_codes=canonical_codes,
            dry_run=dry_run,
        )
    if channel == "shopify":
        return _provision_shopify(
            session,
            connection_id=connection_id,
            shop_code=shop_code,
            canonical_codes=canonical_codes,
            dry_run=dry_run,
        )
    return {"status": "ok", "channel": channel, "created": 0, "linked": 0, "planned": 0}


def _resolve_magento_connection_id(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 == "magento" else None
    return connection_id


def _provision_magento(
    session: Session,
    connection_id: int,
    *,
    canonical_codes: Optional[Iterable[str]] = None,
    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

    aliases = pending_create_aliases(session, "magento", canonical_codes=canonical_codes)
    if not aliases:
        return {"status": "ok", "channel": "magento", "created": 0, "linked": 0, "planned": 0}

    existing_codes = _magento_registry_codes(session, connection_id)
    repo = SqlAlchemyMagentoConnectionRepository(session)
    conn = repo.get_for_sync(connection_id)
    if not conn:
        return {"status": "failed", "error": f"Magento connection {connection_id} not found"}

    planned: List[str] = []
    created: List[str] = []
    linked: List[str] = []
    errors: List[str] = []

    api = MagentoRestClient(
        MagentoOAuthClient(**build_magento_oauth_kwargs(conn))
    )
    default_set_id = _resolve_default_magento_attribute_set_id(session, connection_id)

    for alias in aliases:
        code = str(alias.channel_attribute_code or "").strip()
        if not code or code in existing_codes:
            if code and not dry_run:
                _mark_alias_provisioned(session, alias)
                linked.append(code)
            continue
        if not re.match(r"^[a-z0-9_]{1,64}$", code):
            errors.append(f"Invalid Magento attribute code '{code}'")
            continue
        planned.append(code)
        if dry_run:
            continue
        frontend_input, backend_type = _magento_input_types(alias.data_type)
        label = alias.canonical_code.replace("_", " ").title()
        try:
            remote_status, remote_attr = api.get_attribute(code)
            if remote_status == 200 and remote_attr:
                if not _assign_magento_attribute_to_default_set(api, default_set_id, code, errors):
                    continue
                _record_magento_attribute_registry(
                    session,
                    connection_id=connection_id,
                    code=code,
                    label=remote_attr.get("frontend_label") or remote_attr.get("default_frontend_label") or label,
                    frontend_input=remote_attr.get("frontend_input") or frontend_input,
                    backend_type=remote_attr.get("backend_type") or backend_type,
                    attribute_id=remote_attr.get("attribute_id"),
                )
                _mark_alias_provisioned(session, alias)
                existing_codes.add(code)
                linked.append(code)
                continue
            create_options = _magento_create_options(frontend_input)
            status, body = api.create_attribute(
                code,
                frontend_label=label,
                frontend_input=frontend_input,
                backend_type=backend_type,
                is_configurable=frontend_input == "select",
                **create_options,
            )
            if status in (200, 201) and body:
                add_status, add_err = api.add_attribute_to_attribute_set(default_set_id, code)
                if add_status not in (200, 201):
                    errors.append(
                        f"Create {code}: created but failed to assign to default attribute set "
                        f"{default_set_id} (HTTP {add_status}{f' - {add_err}' if add_err else ''})"
                    )
                    continue
                created.append(code)
                existing_codes.add(code)
                _mark_alias_provisioned(session, alias)
                _record_magento_attribute_registry(
                    session,
                    connection_id=connection_id,
                    code=code,
                    label=label,
                    frontend_input=frontend_input,
                    backend_type=backend_type,
                    attribute_id=body.get("attribute_id"),
                )
            else:
                retry_status, retry_attr = api.get_attribute(code)
                if retry_status == 200 and retry_attr:
                    if not _assign_magento_attribute_to_default_set(api, default_set_id, code, errors):
                        continue
                    _record_magento_attribute_registry(
                        session,
                        connection_id=connection_id,
                        code=code,
                        label=retry_attr.get("frontend_label") or retry_attr.get("default_frontend_label") or label,
                        frontend_input=retry_attr.get("frontend_input") or frontend_input,
                        backend_type=retry_attr.get("backend_type") or backend_type,
                        attribute_id=retry_attr.get("attribute_id"),
                    )
                    _mark_alias_provisioned(session, alias)
                    existing_codes.add(code)
                    linked.append(code)
                else:
                    errors.append(f"Create {code}: HTTP {status}{_remote_error_suffix(body)}")
        except Exception as exc:
            errors.append(f"Create {code}: {exc}")

    return {
        "status": "ok" if not errors else "partial",
        "channel": "magento",
        "connection_id": connection_id,
        "planned": planned,
        "created": created,
        "linked": linked,
        "errors": errors,
        "default_attribute_set_id": default_set_id,
    }


def _assign_magento_attribute_to_default_set(
    api: Any,
    default_set_id: Optional[int],
    code: str,
    errors: List[str],
) -> bool:
    if default_set_id is None:
        errors.append(f"Assign {code}: default Magento attribute set not found")
        return False
    add_status, add_err = api.add_attribute_to_attribute_set(default_set_id, code)
    if add_status in (200, 201):
        return True
    err_text = str(add_err or "")
    if "already" in err_text.lower() and "exist" in err_text.lower():
        return True
    errors.append(
        f"Assign {code}: failed to assign to default attribute set "
        f"{default_set_id} (HTTP {add_status}{f' - {add_err}' if add_err else ''})"
    )
    return False


def provision_magento_configurable_catalog(
    session: Session,
    connection_id: int,
    *,
    snapshot_rows: Optional[List[Dict[str, Any]]] = None,
    dry_run: bool = False,
) -> Dict[str, Any]:
    """Create Magento configurable axis attributes/options needed by snapshot rows."""
    native_id = _resolve_magento_connection_id(connection_id)
    if native_id is None:
        return {"status": "failed", "error": "magento connection_id is required"}

    rows = list(snapshot_rows or [])
    if not rows:
        from magento.master_catalog_rows import build_magento_snapshot_rows_from_master

        rows = build_magento_snapshot_rows_from_master(
            session,
            only_assigned=True,
            connection_id=connection_id,
        )

    from magento.configurable_requirements import extract_configurable_requirements

    codes, _, option_labels = extract_configurable_requirements(rows)
    if dry_run:
        return {
            "status": "ok",
            "channel": "magento",
            "dry_run": True,
            "attribute_codes": sorted(codes),
            "option_labels": {k: sorted(v) for k, v in option_labels.items() if v},
        }

    from db.magento_repositories import (
        SqlAlchemyAttributeRegistryRepository,
        SqlAlchemyBaselineSyncRunRepository,
        SqlAlchemyMagentoConnectionRepository,
    )
    from magento.magento_api import MagentoOAuthClient, MagentoRestClient
    from magento.oauth_client import build_magento_oauth_kwargs
    from magento.sync_service import MagentoSyncService

    conn_repo = SqlAlchemyMagentoConnectionRepository(session)
    conn = conn_repo.get_for_sync(native_id)
    if not conn:
        return {"status": "failed", "error": f"Magento connection {native_id} not found"}

    api = MagentoRestClient(
        MagentoOAuthClient(**build_magento_oauth_kwargs(conn))
    )
    attr_repo = SqlAlchemyAttributeRegistryRepository(session)
    sync_svc = MagentoSyncService(
        api,
        conn,
        attr_cache_repo=attr_repo,
        baseline_repo=SqlAlchemyBaselineSyncRunRepository(session),
    )
    result = sync_svc.ensure_attributes_and_options_exist(native_id, rows, provision=True)
    return {
        "status": "ok" if not result.get("errors") else "partial",
        "channel": "magento",
        "connection_id": native_id,
        **result,
    }


def repair_magento_select_attributes(
    session: Session,
    connection_id: int,
    *,
    canonical_codes: Optional[Iterable[str]] = None,
    dry_run: bool = False,
) -> Dict[str, Any]:
    """Recreate legacy Magento text attributes as select attributes."""
    from db.magento_repositories import SqlAlchemyMagentoConnectionRepository
    from magento.magento_api import MagentoOAuthClient, MagentoRestClient
    from magento.oauth_client import build_magento_oauth_kwargs

    stmt = (
        select(ChannelAttributeAlias)
        .where(ChannelAttributeAlias.channel_code == "magento")
        .where(ChannelAttributeAlias.is_active.is_(True))
    )
    wanted = sorted({str(code).strip().lower() for code in (canonical_codes or []) if str(code).strip()})
    if wanted:
        stmt = stmt.where(ChannelAttributeAlias.canonical_code.in_(wanted))
    aliases = list(session.scalars(stmt).all())
    codes = sorted(
        {
            normalize_channel_attribute_code(alias.channel_attribute_code)
            for alias in aliases
            if normalize_channel_attribute_code(alias.channel_attribute_code)
        }
    )
    if not codes:
        return {
            "status": "ok",
            "channel": "magento",
            "connection_id": connection_id,
            "checked": [],
            "recreated": [],
            "skipped": [],
            "errors": [],
            "dry_run": dry_run,
        }

    repo = SqlAlchemyMagentoConnectionRepository(session)
    conn = repo.get_for_sync(connection_id)
    if not conn:
        return {"status": "failed", "error": f"Magento connection {connection_id} not found"}

    api = MagentoRestClient(MagentoOAuthClient(**build_magento_oauth_kwargs(conn)))
    default_set_id = _resolve_default_magento_attribute_set_id(session, connection_id)
    checked: List[Dict[str, Any]] = []
    recreated: List[str] = []
    skipped: List[str] = []
    errors: List[str] = []

    for code in codes:
        status, attr = api.get_attribute(code)
        if status != 200 or not isinstance(attr, dict):
            checked.append({"attribute_code": code, "status": status, "action": "missing"})
            skipped.append(code)
            continue

        frontend_input = str(attr.get("frontend_input") or "").strip().lower()
        backend_type = str(attr.get("backend_type") or "").strip().lower()
        checked.append(
            {
                "attribute_code": code,
                "frontend_input": frontend_input,
                "backend_type": backend_type,
                "attribute_id": attr.get("attribute_id"),
            }
        )
        if frontend_input == "select" and backend_type == "int":
            skipped.append(code)
            continue

        if dry_run:
            recreated.append(code)
            continue

        delete_status, _delete_body, delete_err = api.delete_attribute(code)
        if delete_status not in (200, 201, 204):
            errors.append(
                f"Delete {code}: HTTP {delete_status}{f' - {delete_err}' if delete_err else ''}"
            )
            continue

        _purge_magento_attribute_cache(session, connection_id=connection_id, code=code)

        create_status, create_body = api.create_attribute(
            code,
            frontend_label=str(
                attr.get("frontend_label")
                or attr.get("default_frontend_label")
                or code.replace("_", " ").title()
            ),
            frontend_input="select",
            backend_type="int",
            is_searchable=True,
            is_visible_in_advanced_search=True,
            is_filterable=True,
            is_filterable_in_search=True,
            used_in_product_listing=True,
        )
        if create_status not in (200, 201) or not isinstance(create_body, dict):
            errors.append(f"Create {code}: HTTP {create_status}{_remote_error_suffix(create_body)}")
            continue

        if not _assign_magento_attribute_to_default_set(api, default_set_id, code, errors):
            continue

        _record_magento_attribute_registry(
            session,
            connection_id=connection_id,
            code=code,
            label=str(
                create_body.get("frontend_label")
                or create_body.get("default_frontend_label")
                or code.replace("_", " ").title()
            ),
            frontend_input="select",
            backend_type="int",
            attribute_id=create_body.get("attribute_id"),
        )
        recreated.append(code)

    return {
        "status": "ok" if not errors else ("partial" if recreated else "failed"),
        "channel": "magento",
        "connection_id": connection_id,
        "checked": checked,
        "recreated": recreated,
        "skipped": skipped,
        "errors": errors,
        "dry_run": dry_run,
        "default_attribute_set_id": default_set_id,
    }


def _resolve_default_magento_attribute_set_id(session: Session, connection_id: int) -> int:
    """Resolve default Magento product attribute set id from registry/baseline, fallback 4."""
    from db.magento_repositories import SqlAlchemyAttributeSetRegistryRepository, SqlAlchemyBaselineSyncRunRepository

    try:
        reg_repo = SqlAlchemyAttributeSetRegistryRepository(session)
        default_id = reg_repo.get_by_name(connection_id, "Default")
        if default_id:
            return int(default_id)
    except Exception:
        logger.debug("Could not resolve default set id from attribute set registry", exc_info=True)
    try:
        baseline_repo = SqlAlchemyBaselineSyncRunRepository(session)
        baseline_id = baseline_repo.get_default_attribute_set_id(connection_id)
        if baseline_id:
            return int(baseline_id)
    except Exception:
        logger.debug("Could not resolve default set id from baseline run", exc_info=True)
    return 4


def _provision_shopify(
    session: Session,
    *,
    connection_id: Optional[int],
    shop_code: Optional[str],
    canonical_codes: Optional[Iterable[str]] = None,
    dry_run: bool,
) -> Dict[str, Any]:
    from db.shopify_registry import resolve_registry_shop_code
    from shopify.connections import build_client, get_active_connection

    aliases = pending_create_aliases(session, "shopify", canonical_codes=canonical_codes)
    if not aliases:
        return {"status": "ok", "channel": "shopify", "created": 0, "linked": 0, "planned": 0}

    connection: Optional[ShopifyConnection] = None
    if connection_id:
        native_id = connection_id
        if connection_id >= SHOPIFY_COMPAT_ID_OFFSET:
            channel_type, native_id = decode_compat_connection_id(connection_id)
            if channel_type != "shopify":
                return {"status": "failed", "error": f"Connection {connection_id} is not Shopify"}
        connection = session.get(ShopifyConnection, native_id)
    elif shop_code:
        connection = get_active_connection(session, shop_code=shop_code)
    else:
        connection = get_active_connection(session)

    effective_shop_code = resolve_registry_shop_code(session, connection=connection, shop_code=shop_code)
    existing_codes = _shopify_registry_codes(session, effective_shop_code)
    client = build_client(connection) if connection or not dry_run else None

    planned: List[str] = []
    created: List[str] = []
    linked: List[str] = []
    errors: List[str] = []

    for alias in aliases:
        code = str(alias.channel_attribute_code or "").strip()
        if not code or code in existing_codes:
            if code and not dry_run:
                _mark_alias_provisioned(session, alias)
                linked.append(code)
            continue
        namespace, key = _shopify_namespace_key(code)
        planned.append(code)
        if dry_run:
            continue
        if client is None:
            errors.append("No Shopify connection available for metafield creation")
            break
        shopify_type = _shopify_metafield_type(alias.data_type)
        try:
            data = client.graphql(
                METAFIELD_DEFINITION_CREATE,
                {
                    "definition": {
                        "name": alias.canonical_code.replace("_", " ").title(),
                        "namespace": namespace,
                        "key": key,
                        "ownerType": "PRODUCT",
                        "type": shopify_type,
                    }
                },
            )
            result = data.get("metafieldDefinitionCreate") or {}
            user_errors = result.get("userErrors") or []
            if user_errors:
                errors.append(f"Create {code}: {user_errors}")
                continue
            created.append(code)
            existing_codes.add(code)
            _mark_alias_provisioned(session, alias)
            session.add(
                ShopifyAttributeRegistry(
                    shop_code=effective_shop_code or "default",
                    owner_type="product_metafield",
                    attribute_code=code,
                    name=alias.canonical_code.replace("_", " ").title(),
                    namespace=namespace,
                    key=key,
                    data_type=shopify_type,
                    is_standard=False,
                    raw_json={"source": "create_new_provision"},
                    fetched_at=datetime.now(timezone.utc),
                )
            )
        except Exception as exc:
            errors.append(f"Create {code}: {exc}")

    return {
        "status": "ok" if not errors else "partial",
        "channel": "shopify",
        "shop_code": effective_shop_code,
        "planned": planned,
        "created": created,
        "linked": linked,
        "errors": errors,
    }


METAFIELD_DEFINITION_CREATE = """
mutation metafieldDefinitionCreate($definition: MetafieldDefinitionInput!) {
  metafieldDefinitionCreate(definition: $definition) {
    createdDefinition { id namespace key }
    userErrors { field message }
  }
}
"""


def _mark_alias_provisioned(session: Session, alias: ChannelAttributeAlias) -> None:
    alias.action = USE_EXISTING_ACTION
    alias.notes = (alias.notes or "").strip()
    if "provisioned" not in (alias.notes or ""):
        alias.notes = f"{alias.notes};provisioned".strip(";")


def _magento_registry_codes(session: Session, connection_id: int) -> Set[str]:
    return {
        str(code).strip()
        for (code,) in session.execute(
            select(MagentoAttributeRegistry.attribute_code).where(
                MagentoAttributeRegistry.connection_id == connection_id
            )
        ).all()
        if code
    }


def _shopify_registry_codes(session: Session, shop_code: Optional[str]) -> Set[str]:
    stmt = select(ShopifyAttributeRegistry.attribute_code)
    if shop_code:
        stmt = stmt.where(ShopifyAttributeRegistry.shop_code == shop_code)
    return {str(code).strip() for (code,) in session.execute(stmt).all() if code}


def _magento_input_types(data_type: Optional[str]) -> tuple[str, str]:
    text = str(data_type or "text").strip().lower()
    if text in {"number", "integer", "int", "decimal"}:
        return "text", "varchar"
    if text in {"html", "textarea"}:
        return "textarea", "text"
    if text in {"boolean", "bool"}:
        return "boolean", "int"
    if text in {"select", "option"}:
        return "select", "int"
    return "text", "varchar"


def _magento_create_options(frontend_input: str) -> Dict[str, Any]:
    """Use conservative flags; Magento rejects filter/search flags on many text attributes."""
    if frontend_input == "select":
        return {
            "is_searchable": True,
            "is_visible_in_advanced_search": True,
            "is_filterable": True,
            "is_filterable_in_search": True,
            "used_in_product_listing": True,
        }
    return {
        "is_searchable": False,
        "is_visible_in_advanced_search": False,
        "is_filterable": False,
        "is_filterable_in_search": False,
        "used_in_product_listing": True,
    }


def _remote_error_suffix(body: Any) -> str:
    if body is None:
        return ""
    if isinstance(body, dict):
        message = body.get("message") or body.get("error") or body
        parameters = body.get("parameters")
        if isinstance(parameters, dict) and parameters.get("fieldName"):
            return f" - {message} (field={parameters['fieldName']})"
        return f" - {message}"
    return f" - {body}"


def _record_magento_attribute_registry(
    session: Session,
    *,
    connection_id: int,
    code: str,
    label: str,
    frontend_input: str,
    backend_type: str,
    attribute_id: Any,
) -> None:
    if not attribute_id:
        return
    payload = {
        "connection_id": connection_id,
        "attribute_code": code,
        "attribute_id": int(attribute_id),
        "frontend_label": label,
        "frontend_input": frontend_input,
        "backend_type": backend_type,
        "is_user_defined": True,
        "fetched_at": datetime.now(timezone.utc),
    }
    bind = session.get_bind()
    if bind is not None and bind.dialect.name == "postgresql":
        stmt = pg_insert(MagentoAttributeRegistry).values(payload)
        stmt = stmt.on_conflict_do_update(
            constraint="uq_attribute_registry_conn_code",
            set_={
                "attribute_id": stmt.excluded.attribute_id,
                "frontend_label": stmt.excluded.frontend_label,
                "frontend_input": stmt.excluded.frontend_input,
                "backend_type": stmt.excluded.backend_type,
                "is_user_defined": stmt.excluded.is_user_defined,
                "fetched_at": stmt.excluded.fetched_at,
            },
        )
        session.execute(stmt)
        return
    existing = session.scalar(
        select(MagentoAttributeRegistry)
        .where(MagentoAttributeRegistry.connection_id == connection_id)
        .where(MagentoAttributeRegistry.attribute_code == code)
    )
    if existing is None:
        session.add(
            MagentoAttributeRegistry(
                **payload,
            )
        )
        return
    for key, value in payload.items():
        if key in {"connection_id", "attribute_code"}:
            continue
        setattr(existing, key, value)


def _purge_magento_attribute_cache(
    session: Session,
    *,
    connection_id: int,
    code: str,
) -> None:
    session.query(MagentoAttributeOptionRegistry).filter(
        MagentoAttributeOptionRegistry.connection_id == connection_id,
        MagentoAttributeOptionRegistry.attribute_code == code,
    ).delete(synchronize_session=False)
    session.query(MagentoAttributeRegistry).filter(
        MagentoAttributeRegistry.connection_id == connection_id,
        MagentoAttributeRegistry.attribute_code == code,
    ).delete(synchronize_session=False)


def normalize_channel_attribute_code(value: Optional[str]) -> str:
    return str(value or "").strip().lower()


def _shopify_metafield_type(data_type: Optional[str]) -> str:
    text = str(data_type or "text").strip().lower()
    if text in {"number", "integer", "int"}:
        return "number_integer"
    if text in {"decimal", "float"}:
        return "number_decimal"
    if text in {"html", "textarea"}:
        return "multi_line_text_field"
    if text in {"boolean", "bool"}:
        return "boolean"
    return "single_line_text_field"


def _shopify_namespace_key(code: str) -> tuple[str, str]:
    key = code[len("custom_") :] if code.startswith("custom_") else code
    return DEFAULT_SHOPIFY_METAFIELD_NAMESPACE, key