from __future__ import annotations

import hashlib
import json
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Tuple

from sqlalchemy import select
from sqlalchemy.orm import Session

from channel.url_canonical import normalize_plp_path
from db.channel_aliases import channel_aliases_by_canonical
from db.channel_assignments import active_skus_for_channel
from db.channel_capabilities import resolve_product_structure, resolve_product_structure_for_connection
from db.canonical_attributes import canonicalize_attribute_fields, matches_publish_constraints
from db.location_experience import collection_location_context
from db.master_overrides import overrides_for_channel
from db.master_relations import (
    build_parent_relation_fields,
    children_by_parent,
    expand_skus_with_children,
    expand_skus_with_parents,
    parent_by_child,
    parent_skus_with_children,
)
from db.models import (
    CollectionLandingPage,
    MasterLocationRegistry,
    MasterProduct,
    MasterProductAttributeValue,
    MasterProductLocationAvailability,
    MasterProductPrice,
)
from db.source_imports import normalize_column_key

if TYPE_CHECKING:
    from db.shopping_taxonomy_vocabulary import ShoppingTaxonomyVocabulary


CHANNEL_CODES = {"magento", "shopify", "plytix"}


def resolve_pipeline_channel_code(
    channel_code: Optional[str] = None,
    *,
    channel_type: Optional[str] = None,
) -> str:
    """Map a connection display code (e.g. magento-1, shop_code) to a pipeline channel code.

    Master-catalog export, aliases, assignments, and publish state all use the pipeline
    codes magento | shopify | plytix — not per-connection identifiers.
    """
    if channel_type:
        normalized_type = normalize_column_key(channel_type)
        if normalized_type in CHANNEL_CODES:
            return normalized_type
    if channel_code:
        normalized_code = normalize_column_key(channel_code)
        if normalized_code in CHANNEL_CODES:
            return normalized_code
    raise ValueError(
        f"Unsupported channel_code: {channel_code!r} "
        f"(channel_type={channel_type!r}; expected one of {sorted(CHANNEL_CODES)})"
    )


# Per-channel sell price preference, first match wins.
PRICE_TYPE_PRIORITY = {
    "shopify": ("shopify", "base", "rta", "assembled"),
    "magento": ("base", "rta", "assembled"),
    "plytix": ("base",),
}

CORE_CANONICAL_FIELDS = (
    "name",
    "title",
    "brand",
    "manufacturer",
    "product_family",
    "category_l1",
    "category_l2",
    "category_l3",
    "collection",
    "assembly_type",
    "item_size",
    "item_style",
    "width",
    "height",
    "depth",
    "assembled_width",
    "assembled_height",
    "assembled_depth",
    "meta_title",
    "meta_description",
    "meta_keywords",
    "product_url_slug",
    "url_slug",
    "url_key",
    "shopping_collection",
    "shopping_l1",
    "shopping_l2",
    "status",
)


def _first_shopping_value(raw: Any) -> str:
    if isinstance(raw, list):
        for item in raw:
            text = str(item or "").strip()
            if text:
                return text
        return ""
    return str(raw or "").strip()


def _shopping_collection_export_value(raw: Any) -> str:
    from db.collection_landing_pages import _relative_shopping_collection_value

    value = _first_shopping_value(raw)
    if not value:
        return ""
    return str(_relative_shopping_collection_value(value) or value)


def _shopping_path_export_value(raw: Any, resolver) -> str:
    value = _first_shopping_value(raw)
    if not value:
        return ""
    return str(resolver(value) or value)


def _relative_shopping_l1_export_value(path_slug: Any) -> str:
    from db.collection_landing_pages import _relative_shopping_l1_value

    return str(_relative_shopping_l1_value(path_slug) or "")


def _relative_shopping_l2_export_value(path_slug: Any) -> str:
    from db.collection_landing_pages import _relative_shopping_l2_value

    return str(_relative_shopping_l2_value(path_slug) or "")


def resolve_push_skus(
    session: Session,
    channel_code: str,
    *,
    skus: Optional[List[str]] = None,
    only_assigned: bool = True,
    product_structure: Optional[str] = None,
    capabilities: Optional[Dict[str, Any]] = None,
    expand_relations: Optional[bool] = None,
) -> Set[str]:
    """SKUs that should participate in a channel push/export after assignment + structure rules."""
    channel = resolve_pipeline_channel_code(channel_code)
    structure = product_structure or resolve_product_structure_for_connection(
        session,
        channel,
        connection_id=None,
        capabilities=capabilities,
    )

    assigned: Optional[Set[str]] = active_skus_for_channel(session, channel) if only_assigned else None
    if skus:
        wanted = {str(s).strip() for s in skus if str(s).strip()}
        if assigned is not None:
            wanted &= assigned
    elif assigned is not None:
        wanted = set(assigned)
    else:
        wanted = {
            sku
            for (sku,) in session.execute(
                select(MasterProduct.sku).where(MasterProduct.is_active.is_(True))
            ).all()
        }

    should_expand = expand_relations
    if should_expand is None:
        # Curated SKU lists (dashboard selection / limit_skus) must not pull in siblings.
        should_expand = structure == "parent_children" and not skus

    if should_expand and structure == "parent_children":
        wanted = expand_skus_with_children(session, wanted)
        wanted = expand_skus_with_parents(session, wanted)
        raw_parent_children, _ = _variation_builder_raw_children_maps(session, wanted)
        for parent_sku, child_skus in raw_parent_children.items():
            if parent_sku in wanted or any(child_sku in wanted for child_sku in child_skus):
                wanted.add(parent_sku)
                wanted.update(child_skus)
    elif structure != "parent_children":
        # Flat channels can know variation relationships, but they should not publish
        # generated/configurable parent shell SKUs as standalone products.
        wanted -= parent_skus_with_children(session, wanted)
        raw_parent_children, _ = _variation_builder_raw_children_maps(session, wanted)
        wanted -= set(raw_parent_children.keys())
    return wanted


def build_channel_export_bundle(
    session: Session,
    channel_code: str,
    *,
    skus: Optional[List[str]] = None,
    limit: Optional[int] = None,
    only_assigned: bool = True,
    connection_id: Optional[int] = None,
    capabilities: Optional[Dict[str, Any]] = None,
    publish_constraints: Optional[Dict[str, List[Any]]] = None,
    prefer_projection_cache: bool = True,
) -> Dict[str, Any]:
    """Payloads plus relation metadata for channels that use parent/child structure."""
    channel = resolve_pipeline_channel_code(channel_code)
    structure = resolve_product_structure_for_connection(
        session,
        channel,
        connection_id=connection_id,
        capabilities=capabilities,
    )
    wanted = resolve_push_skus(
        session,
        channel,
        skus=skus,
        only_assigned=only_assigned,
        product_structure=structure,
        capabilities=capabilities,
    )
    if limit and len(wanted) > limit:
        wanted = set(sorted(wanted)[:limit])
    payloads = build_channel_product_payloads(
        session,
        channel,
        skus=sorted(wanted),
        only_assigned=False,
        connection_id=connection_id,
        product_structure=structure,
        capabilities=capabilities,
        publish_constraints=publish_constraints,
        prefer_projection_cache=prefer_projection_cache,
    )
    relations = children_by_parent(session, parent_skus_with_children(session, wanted))
    return {
        "channel_code": channel,
        "product_structure": structure,
        "sku_count": len(wanted),
        "payloads": payloads,
        "relations": relations,
    }


def build_channel_product_payloads(
    session: Session,
    channel_code: str,
    *,
    skus: Optional[List[str]] = None,
    limit: Optional[int] = None,
    only_assigned: bool = True,
    connection_id: Optional[int] = None,
    product_structure: Optional[str] = None,
    capabilities: Optional[Dict[str, Any]] = None,
    skip_taxonomy: bool = False,
    publish_constraints: Optional[Dict[str, List[Any]]] = None,
    prefer_projection_cache: bool = True,
) -> List[Dict[str, Any]]:
    """Build one payload per SKU for a channel: master core fields + attribute values,
    with master_product_override applied on top, mapped through channel attribute aliases.
    """
    channel = resolve_pipeline_channel_code(channel_code)
    structure = product_structure or resolve_product_structure_for_connection(
        session,
        channel,
        connection_id=connection_id,
        capabilities=capabilities,
    )

    if only_assigned or product_structure is not None or capabilities is not None:
        wanted = resolve_push_skus(
            session,
            channel,
            skus=skus,
            only_assigned=only_assigned,
            product_structure=structure,
            capabilities=capabilities,
        )
        if limit and len(wanted) > limit:
            wanted = set(sorted(wanted)[:limit])
        skus = sorted(wanted)
        only_assigned = False

    requested_skus = [str(s).strip() for s in (skus or []) if str(s).strip()]
    cached_fresh: List[Dict[str, Any]] = []
    if prefer_projection_cache and requested_skus:
        from db.channel_projection_cache import partition_projection_cache

        partitioned = partition_projection_cache(
            session,
            channel_code=channel,
            connection_id=connection_id,
            skus=requested_skus,
            product_structure=structure,
        )
        cached_fresh = list(partitioned["fresh"])
        rebuild_skus = list(partitioned["rebuild_skus"])
        if not rebuild_skus:
            _apply_canonical_shopping_field_overrides(cached_fresh)
            return cached_fresh
        skus = rebuild_skus
        only_assigned = False

    assigned: Optional[Set[str]] = active_skus_for_channel(session, channel) if only_assigned else None
    products = _products(session, skus=skus, assigned=assigned, limit=limit)
    values = _attribute_values(session, [p.id for p in products])
    prices = _prices(session, [p.id for p in products])
    image_fields = _image_fields(session, [p.sku for p in products])
    collection_availability = _collection_availability_by_product(session, products)
    overrides = overrides_for_channel(session, channel)
    alias_map = _build_alias_map(session, channel)
    taxonomy_index = None
    try:
        from db.master_taxonomy_product_paths import _NodeIndex

        taxonomy_index = _NodeIndex.load(session)
    except Exception:
        taxonomy_index = None

    product_skus = [p.sku for p in products]
    raw_parent_children, raw_child_parent_map = _variation_builder_raw_children_maps(session, set(product_skus))
    parent_set = parent_skus_with_children(session, product_skus) | set(raw_parent_children.keys())
    child_parent_map = dict(raw_child_parent_map)
    child_parent_map.update(parent_by_child(session, product_skus))
    hub_scoped_rules, listing_paths_by_sku, collection_code_by_path = _hub_scope_context(
        session,
        product_skus,
    )

    payloads: List[Dict[str, Any]] = []
    for product in products:
        attrs = enrich_catalog_attributes(
            {normalize_column_key(v.attribute_code): v.value for v in values.get(product.id, [])}
        )
        if taxonomy_index is not None:
            try:
                from db.master_taxonomy_product_paths import canonical_taxonomy_leaf_label

                taxonomy_leaf = canonical_taxonomy_leaf_label(
                    session,
                    product,
                    attrs,
                    index=taxonomy_index,
                )
                if taxonomy_leaf:
                    attrs["taxonomy_leaf_label"] = taxonomy_leaf
            except Exception:
                pass
        attrs.update(image_fields.get(product.sku, {}))
        canonical_fields, override_codes = compose_canonical_fields(
            _core_fields(product),
            attrs,
            overrides.get(product.sku, {}),
        )
        # Collection landing availability is the single source of truth for
        # Magento's per-location availability attributes.
        effective_availability = collection_availability.get(product.id) or {}
        _attach_location_availability_fields(
            canonical_fields,
            effective_availability,
            force_item_availability=bool(effective_availability),
        )
        _attach_collection_location_fields(
            session,
            canonical_fields,
            category_l1=product.category_l1,
            collection=product.collection,
        )
        canonical_fields = canonicalize_attribute_fields(canonical_fields)
        _apply_manual_taxonomy_field_override(session, product.sku, canonical_fields)
        _strip_hub_scoped_canonical_fields(
            canonical_fields,
            product=product,
            hub_scoped_rules=hub_scoped_rules,
            listing_path_slugs=listing_paths_by_sku.get(str(product.sku or "").strip().upper(), []),
            collection_code_by_path=collection_code_by_path,
        )
        if not matches_publish_constraints(canonical_fields, publish_constraints):
            continue
        channel_fields = map_fields_to_channel(canonical_fields, alias_map)
        price = select_price(prices.get(product.id, {}), channel)
        if product.sku in parent_set:
            role = "parent"
        elif product.sku in child_parent_map:
            role = "child"
        else:
            role = "standalone"
        payloads.append(
            {
                "sku": product.sku,
                "channel_code": channel,
                "product_role": role,
                "canonical_fields": canonical_fields,
                "fields": channel_fields,
                "override_codes": sorted(override_codes),
                "price": price,
                "prices": prices.get(product.id, {}),
                "payload_hash": payload_hash({"fields": channel_fields, "price": price}),
                "source_row_hash": product.row_hash,
            }
        )

    if structure == "parent_children":
        parent_fields = build_parent_relation_fields(
            session,
            [p["sku"] for p in payloads if p["product_role"] == "parent"],
            channel_code=channel,
            connection_id=connection_id,
        )
        missing_parent_skus = [
            payload["sku"]
            for payload in payloads
            if payload["product_role"] == "parent" and payload["sku"] not in parent_fields
        ]
        if missing_parent_skus:
            parent_fields.update(
                _variation_builder_parent_relation_fields_fallback(
                    session,
                    missing_parent_skus,
                    channel_code=channel,
                    connection_id=connection_id,
                )
            )

        child_parents = {}
        for parent_sku, relation_fields in parent_fields.items():
            for child_sku in _child_skus_from_relation_fields(relation_fields):
                child_parents.setdefault(child_sku, parent_sku)
        for payload in payloads:
            sku = payload["sku"]
            if sku in parent_fields:
                payload["relation_fields"] = parent_fields[sku]
            elif sku in child_parents:
                payload["product_role"] = "child"
                payload["relation_fields"] = {"variant_of": child_parents[sku], "product_type": "simple"}

    _attach_channel_skus(session, channel, payloads, connection_id=connection_id)
    _attach_association_fields(session, channel, payloads, connection_id=connection_id)
    _attach_url_canonicals(payloads)
    if skip_taxonomy:
        for payload in payloads:
            payload["payload_hash"] = payload_hash(
                {
                    "fields": payload.get("fields") or {},
                    "price": payload.get("price"),
                    "association_fields": payload.get("association_fields") or {},
                }
            )
    else:
        from db.shopping_taxonomy_vocabulary import load_shopping_taxonomy_vocabulary

        shopping_vocabulary = load_shopping_taxonomy_vocabulary(session)
        _attach_taxonomy_targets(session, channel, payloads, connection_id=connection_id)
        manual_locked = _apply_manual_taxonomy_projection_overrides(
            session, channel, payloads, connection_id=connection_id
        )
        _apply_canonical_shopping_field_overrides(
            payloads,
            manual_locked=manual_locked,
            vocabulary=shopping_vocabulary,
        )

    if prefer_projection_cache and payloads:
        from db.channel_projection_cache import store_projection_payloads

        store_projection_payloads(
            session,
            channel_code=channel,
            connection_id=connection_id,
            payloads=payloads,
            product_structure=structure,
            scope_key="write_through",
            notes="Write-through after partial rebuild",
            replace_scope=False,
        )

    if cached_fresh:
        from db.shopping_taxonomy_vocabulary import load_shopping_taxonomy_vocabulary

        # Cached payloads are re-derived too, so manual curation must be re-resolved
        # here as well — otherwise inference silently re-pollutes them on cache hits.
        cached_manual_locked = _manual_locked_shopping_codes(
            session, [payload.get("sku") for payload in cached_fresh]
        )
        _apply_canonical_shopping_field_overrides(
            cached_fresh,
            manual_locked=cached_manual_locked,
            vocabulary=load_shopping_taxonomy_vocabulary(session),
        )
        by_sku = {p["sku"]: p for p in payloads}
        for cached in cached_fresh:
            by_sku.setdefault(cached["sku"], cached)
        payloads = [by_sku[sku] for sku in requested_skus if sku in by_sku] or list(by_sku.values())

    return payloads


def _variation_builder_raw_children_maps(
    session: Session,
    skus: Optional[Set[str]] = None,
) -> Tuple[Dict[str, List[str]], Dict[str, str]]:
    if not hasattr(session, "scalars"):
        return {}, {}
    wanted = {str(s).strip() for s in (skus or set()) if str(s).strip()}
    parent_to_children: Dict[str, List[str]] = {}
    child_to_parent: Dict[str, str] = {}

    candidates = session.scalars(
        select(MasterProduct)
        .where(MasterProduct.is_active.is_(True))
        .where(MasterProduct.raw_payload.is_not(None))
        .order_by(MasterProduct.sku)
    ).all()
    for product in candidates:
        payload = product.raw_payload if isinstance(product.raw_payload, dict) else {}
        if payload.get("generated_by") != "variation_builder":
            continue
        child_skus = [
            str(child_sku or "").strip()
            for child_sku in (payload.get("children") or [])
            if str(child_sku or "").strip()
        ]
        if not child_skus:
            continue
        if wanted and product.sku not in wanted and not wanted.intersection(child_skus):
            continue
        parent_to_children[product.sku] = child_skus
        for child_sku in child_skus:
            child_to_parent.setdefault(child_sku, product.sku)
    return parent_to_children, child_to_parent


def _variation_builder_parent_relation_fields_fallback(
    session: Session,
    parent_skus: List[str],
    *,
    channel_code: str,
    connection_id: Optional[int],
) -> Dict[str, Dict[str, Any]]:
    if not parent_skus:
        return {}
    from db.variation_builder import export_variation_parent_child_rows

    exported = export_variation_parent_child_rows(
        session,
        parent_skus=parent_skus,
        channel_code=channel_code,
        connection_id=connection_id,
    )
    fields_by_parent: Dict[str, Dict[str, Any]] = {}
    for row in exported.get("rows") or []:
        parent_sku = str(row.get("parent_sku") or "").strip()
        if not parent_sku or parent_sku in fields_by_parent:
            continue
        fields_by_parent[parent_sku] = {
            "product_type": "configurable",
            "variant_list": str(row.get("variant_list") or "").strip(),
            "configurable_attributes": str(row.get("magento_configurable_attributes") or "").strip(),
            "configurable_variation_labels": str(row.get("configurable_variation_labels") or "").strip(),
            "configurable_variations": str(row.get("configurable_variations") or "").strip(),
        }
    return fields_by_parent


def _child_skus_from_relation_fields(relation_fields: Dict[str, Any]) -> List[str]:
    children: List[str] = []
    seen: Set[str] = set()
    variant_list = str((relation_fields or {}).get("variant_list") or "").strip()
    if variant_list:
        for part in variant_list.replace("|", ",").split(","):
            sku = part.strip()
            if sku and sku not in seen:
                seen.add(sku)
                children.append(sku)
    configurable_variations = str((relation_fields or {}).get("configurable_variations") or "").strip()
    if configurable_variations and "sku=" in configurable_variations.lower():
        for segment in configurable_variations.split("|"):
            for pair in segment.split(","):
                key, _, value = pair.partition("=")
                if key.strip().lower() != "sku":
                    continue
                sku = value.strip()
                if sku and sku not in seen:
                    seen.add(sku)
                    children.append(sku)
    return children


def _apply_manual_taxonomy_field_override(
    session: Session,
    master_sku: str,
    canonical_fields: Dict[str, Any],
) -> None:
    from db.manual_taxonomy_assignments import resolve_manual_taxonomy_overrides

    overrides = resolve_manual_taxonomy_overrides(session, master_skus=[master_sku])
    item = overrides.get(str(master_sku or "").strip().upper())
    if not item:
        return
    for field, value in (
        ("category_l1", item.get("category_l1")),
        ("category_l2", item.get("category_l2")),
        ("category_l3", item.get("category_l3")),
        ("collection", item.get("collection")),
        ("shopping_l1", item.get("shopping_l1")),
        ("shopping_l2", item.get("shopping_l2")),
    ):
        if value:
            canonical_fields[field] = value


def _apply_manual_taxonomy_projection_overrides(
    session: Session,
    channel: str,
    payloads: List[Dict[str, Any]],
    *,
    connection_id: Optional[int] = None,
) -> Dict[str, Set[str]]:
    """Project manual taxonomy onto payloads.

    Returns the shopping_* codes each SKU has curated manually (blank included), so
    downstream path inference cannot overwrite or backfill an operator decision.
    """
    from db.collection_landing_pages import (
        _relative_shopping_collection_value,
        _relative_shopping_l1_value,
        _relative_shopping_l2_value,
    )
    from db.manual_taxonomy_assignments import resolve_manual_taxonomy_overrides
    from db.models import MasterTaxonomyChannelLink, MasterTaxonomyNode

    if not payloads:
        return {}
    overrides = resolve_manual_taxonomy_overrides(session, master_skus=[payload.get("sku") for payload in payloads])
    if not overrides:
        return {}
    manual_locked: Dict[str, Set[str]] = {}
    path_slugs = sorted(
        {
            str(path).strip()
            for item in overrides.values()
            for path in item.get("membership_path_slugs") or []
            if str(path).strip()
        }
    )
    link_rows: List[Any] = []
    if path_slugs:
        stmt = (
            select(MasterTaxonomyNode.path_slug, MasterTaxonomyChannelLink)
            .join(
                MasterTaxonomyChannelLink,
                MasterTaxonomyChannelLink.taxonomy_node_id == MasterTaxonomyNode.id,
            )
            .where(MasterTaxonomyNode.path_slug.in_(path_slugs))
            .where(MasterTaxonomyNode.is_active.is_(True))
            .where(MasterTaxonomyChannelLink.channel_code == channel)
            .where(MasterTaxonomyChannelLink.is_active.is_(True))
        )
        if connection_id is None:
            stmt = stmt.where(MasterTaxonomyChannelLink.connection_id.is_(None))
        else:
            stmt = stmt.where(
                (MasterTaxonomyChannelLink.connection_id == connection_id)
                | (MasterTaxonomyChannelLink.connection_id.is_(None))
            )
        link_rows = session.execute(stmt).all()
    links_by_path: Dict[str, Dict[str, List[Any]]] = {}
    for path_slug, link in link_rows:
        links_by_path.setdefault(str(path_slug), {}).setdefault(str(link.taxonomy_kind or ""), []).append(link)

    for payload in payloads:
        sku = str(payload.get("sku") or "").strip().upper()
        override = overrides.get(sku)
        if not override:
            continue
        canonical_fields = dict(payload.get("canonical_fields") or {})
        channel_fields = dict(payload.get("fields") or {})
        taxonomy_targets = dict(payload.get("taxonomy_targets") or {})
        plp_path_slugs = [
            normalized
            for normalized in (
                normalize_plp_path(item)
                for item in list(taxonomy_targets.get("plp_path_slugs") or []) + list(override.get("plp_path_slugs") or [])
            )
            if normalized
        ]
        seen_paths = set()
        plp_path_slugs = [item for item in plp_path_slugs if not (item in seen_paths or seen_paths.add(item))]
        taxonomy_targets["plp_path_slugs"] = plp_path_slugs

        for field, value in (
            ("category_l1", override.get("category_l1")),
            ("category_l2", override.get("category_l2")),
            ("category_l3", override.get("category_l3")),
            ("collection", override.get("collection")),
        ):
            if value:
                canonical_fields[field] = value

        collection_value = _relative_shopping_collection_value(override.get("collection_path_slug") or "")
        # Curated only: a blank manual value means "do not publish", never "derive one".
        # Deriving shopping_l2 from a hub/L1 path is what created uncurated Magento
        # options such as kitchen-cabinets/mouldings.
        shopping_values = {
            "shopping_collection": str(
                collection_value
                or _shopping_collection_export_value(override.get("shopping_collection"))
                or ""
            ).strip(),
            "shopping_l1": str(
                _shopping_path_export_value(override.get("shopping_l1"), _relative_shopping_l1_export_value)
                or ""
            ).strip(),
            "shopping_l2": str(
                _shopping_path_export_value(override.get("shopping_l2"), _relative_shopping_l2_export_value)
                or ""
            ).strip(),
        }
        locked_codes = manual_locked.setdefault(sku, set())
        for code, value in shopping_values.items():
            locked_codes.add(code)
            if value:
                canonical_fields[code] = value
                channel_fields[code] = value
            else:
                canonical_fields.pop(code, None)
                channel_fields.pop(code, None)

        membership_paths = [normalize_plp_path(item) for item in (override.get("membership_path_slugs") or []) if normalize_plp_path(item)]
        if channel == "magento":
            existing_ids = [int(item) for item in (taxonomy_targets.get("magento_category_ids") or []) if str(item).isdigit()]
            existing_paths = [str(item).strip() for item in (taxonomy_targets.get("magento_category_paths") or []) if str(item).strip()]
            seen_ids = set(existing_ids)
            seen_category_paths = set(existing_paths)
            for path_slug in membership_paths:
                for kind, links in (links_by_path.get(path_slug) or {}).items():
                    if kind not in {"category", "collection"}:
                        continue
                    for link in links:
                        if str(link.remote_id or "").isdigit():
                            category_id = int(str(link.remote_id))
                            if category_id not in seen_ids:
                                seen_ids.add(category_id)
                                existing_ids.append(category_id)
                        remote_path = str(link.remote_path or "").strip() or path_slug
                        if remote_path and remote_path not in seen_category_paths:
                            seen_category_paths.add(remote_path)
                            existing_paths.append(remote_path)
            taxonomy_targets["magento_category_ids"] = existing_ids
            taxonomy_targets["magento_category_paths"] = existing_paths
        elif channel == "shopify":
            existing_product_category = str(taxonomy_targets.get("shopify_product_category") or "").strip()
            existing_collection_ids = [str(item).strip() for item in (taxonomy_targets.get("shopify_collection_ids") or []) if str(item).strip()]
            seen_collection_ids = set(existing_collection_ids)
            for path_slug in membership_paths:
                for kind, links in (links_by_path.get(path_slug) or {}).items():
                    for link in links:
                        remote_id = str(link.remote_id or "").strip()
                        if not remote_id:
                            continue
                        if kind == "product_category" and not existing_product_category:
                            existing_product_category = remote_id
                        if kind == "collection" and remote_id not in seen_collection_ids:
                            seen_collection_ids.add(remote_id)
                            existing_collection_ids.append(remote_id)
            if existing_product_category:
                taxonomy_targets["shopify_product_category"] = existing_product_category
            taxonomy_targets["shopify_collection_ids"] = existing_collection_ids

        payload["canonical_fields"] = canonical_fields
        payload["fields"] = channel_fields
        payload["taxonomy_targets"] = taxonomy_targets
        payload["payload_hash"] = payload_hash(
            {
                "fields": channel_fields,
                "price": payload.get("price"),
                "taxonomy_targets": taxonomy_targets,
                "association_fields": payload.get("association_fields") or {},
            }
        )

    return manual_locked


def _attach_association_fields(
    session: Session,
    channel: str,
    payloads: List[Dict[str, Any]],
    *,
    connection_id: Optional[int] = None,
) -> None:
    from db.association_rules import LINK_TYPES
    from db.channel_sku_mapping import mapping_by_master
    from db.master_associations import association_fields_by_sku

    if not payloads:
        return
    master_skus = [p["sku"] for p in payloads]
    associations = association_fields_by_sku(session, master_skus)
    sku_map = mapping_by_master(session, channel, master_skus, connection_id=connection_id)
    # Also map association targets that may not be in this push batch.
    target_skus = sorted(
        {
            target
            for by_type in associations.values()
            for targets in by_type.values()
            for target in targets
        }
    )
    if target_skus:
        sku_map.update(
            mapping_by_master(session, channel, target_skus, connection_id=connection_id)
        )

    for payload in payloads:
        master_assoc = associations.get(payload["sku"]) or {}
        payload["association_fields_master"] = {
            link_type: list(master_assoc.get(link_type) or []) for link_type in LINK_TYPES
        }
        payload["association_fields"] = {
            link_type: [sku_map.get(sku, sku) for sku in (master_assoc.get(link_type) or [])]
            for link_type in LINK_TYPES
        }


def _attach_channel_skus(
    session: Session,
    channel: str,
    payloads: List[Dict[str, Any]],
    *,
    connection_id: Optional[int] = None,
) -> None:
    from db.channel_sku_mapping import mapping_by_master

    if not payloads:
        return
    sku_map = mapping_by_master(session, channel, [p["sku"] for p in payloads], connection_id=connection_id)
    for payload in payloads:
        master_sku = payload["sku"]
        channel_sku = sku_map.get(master_sku, master_sku)
        payload["channel_sku"] = channel_sku
        if payload.get("relation_fields"):
            from db.channel_sku_mapping import apply_sku_map_to_relation_fields

            payload["relation_fields_master"] = dict(payload["relation_fields"])
            payload["relation_fields"] = apply_sku_map_to_relation_fields(payload["relation_fields"], sku_map)


def _attach_url_canonicals(payloads: List[Dict[str, Any]]) -> None:
    from channel.url_canonical import build_pdp_slug
    from magento.payload_mapper import is_polluted_product_title

    for payload in payloads:
        canonical = dict(payload.get("canonical_fields") or {})
        seo_title = canonical.get("seo_title")
        title = canonical.get("title") or canonical.get("name") or seo_title
        slug_base = seo_title or title
        if slug_base and is_polluted_product_title(str(slug_base)):
            slug_base = title or slug_base
        slug = build_pdp_slug(slug_base, payload.get("channel_sku") or payload.get("sku") or "")
        canonical["product_url_slug"] = slug
        payload["canonical_fields"] = canonical
        fields = dict(payload.get("fields") or {})
        fields["product_url_slug"] = slug
        if payload.get("channel_code") == "shopify":
            fields["handle"] = slug
        if payload.get("channel_code") == "magento":
            fields["url_key"] = slug
        payload["fields"] = fields


def _attach_taxonomy_targets(
    session: Session,
    channel: str,
    payloads: List[Dict[str, Any]],
    *,
    connection_id: Optional[int] = None,
) -> None:
    from db.compat_connections import decode_compat_connection_id
    from db.product_channel_taxonomy import resolve_taxonomy_targets_for_push

    native_connection_id = connection_id
    if connection_id is not None:
        channel_type, native_id = decode_compat_connection_id(connection_id)
        if channel_type in {"magento", "shopify"}:
            native_connection_id = native_id

    for payload in payloads:
        taxonomy_targets = resolve_taxonomy_targets_for_push(
            session,
            master_sku=payload["sku"],
            channel_code=channel,
            canonical_fields=payload.get("canonical_fields") or {},
            connection_id=native_connection_id,
        )
        payload["taxonomy_targets"] = taxonomy_targets
        payload["payload_hash"] = payload_hash(
            {
                "fields": payload.get("fields") or {},
                "price": payload.get("price"),
                "taxonomy_targets": taxonomy_targets,
                "association_fields": payload.get("association_fields") or {},
            }
        )


def _build_alias_map(session: Session, channel: str) -> Dict[str, List[str]]:
    """Merge static wide-table mappings with channel_attribute_alias (supports 1→N fan-out).

    Static mappings seed the first target. Additional alias rows (typically
    mapping_scope=dynamic) append extra remote attribute codes without removing
    the static target — e.g. collection → collection_style + family.
    """
    from db.master_static_field_mapping import static_alias_map_for_channel

    alias_map: Dict[str, List[str]] = {
        canonical: [channel_attr]
        for canonical, channel_attr in static_alias_map_for_channel(session, channel).items()
    }
    aliases = channel_aliases_by_canonical(session, channel)
    for canonical, alias_rows in aliases.items():
        targets = alias_map.setdefault(canonical, [])
        for alias in alias_rows:
            code = normalize_column_key(alias.channel_attribute_code)
            if code and code not in targets:
                targets.append(code)
    return alias_map


def enrich_catalog_attributes(attrs: Dict[str, Any]) -> Dict[str, Any]:
    """Derive shopping-facet attributes used for layered nav and category filters."""
    from db.catalog_intent_planner import _normalize_customer_category_label
    from db.shopping_facet_catalog import magento_option_label

    enriched = dict(attrs)
    sub1 = str(enriched.get("cabinet_type_sub_category_1") or "").strip()
    cabinet_type = str(enriched.get("cabinet_type") or "").strip()

    if sub1:
        normalized_sub = _normalize_customer_category_label(sub1)
        if normalized_sub:
            enriched["cabinet_type_sub_category_1"] = normalized_sub
            if not cabinet_type:
                enriched["cabinet_type"] = magento_option_label(normalized_sub)
    if cabinet_type:
        normalized_type = _normalize_customer_category_label(cabinet_type) or cabinet_type
        enriched["cabinet_type"] = magento_option_label(normalized_type)
        if not sub1 and normalized_type:
            enriched["cabinet_type_sub_category_1"] = normalized_type

    return canonicalize_attribute_fields(enriched)


def _hub_scope_context(
    session: Session,
    product_skus: List[str],
) -> tuple[List[Dict[str, Any]], Dict[str, List[str]], Dict[str, str]]:
    from db.manual_attributes import list_hub_scoped_attribute_rules
    from db.models import ChannelListingPath, MasterCollectionRegistry, ProductListingPathAssignment

    rules = list_hub_scoped_attribute_rules(session)
    listing_paths_by_sku: Dict[str, List[str]] = {}
    collection_code_by_path: Dict[str, str] = {
        normalize_plp_path(row.path_slug): str(row.code or "").strip().upper()
        for row in session.scalars(
            select(MasterCollectionRegistry)
            .where(MasterCollectionRegistry.is_active.is_(True))
            .where(MasterCollectionRegistry.path_slug.is_not(None))
        ).all()
        if normalize_plp_path(row.path_slug)
    }
    if not product_skus:
        return rules, listing_paths_by_sku, collection_code_by_path

    sku_keys = [str(sku or "").strip().upper() for sku in product_skus if str(sku or "").strip()]
    if not sku_keys:
        return rules, listing_paths_by_sku, collection_code_by_path

    path_by_id = {
        row.id: normalize_plp_path(row.path_slug or "")
        for row in session.scalars(select(ChannelListingPath).where(ChannelListingPath.is_active.is_(True))).all()
        if normalize_plp_path(row.path_slug or "")
    }
    for assignment in session.scalars(
        select(ProductListingPathAssignment)
        .where(ProductListingPathAssignment.master_sku.in_(sku_keys))
        .where(ProductListingPathAssignment.assignment_status == "active")
    ).all():
        slug = path_by_id.get(int(assignment.listing_path_id))
        if not slug:
            continue
        key = str(assignment.master_sku or "").strip().upper()
        bucket = listing_paths_by_sku.setdefault(key, [])
        if slug not in bucket:
            bucket.append(slug)
    return rules, listing_paths_by_sku, collection_code_by_path


def _strip_hub_scoped_canonical_fields(
    canonical_fields: Dict[str, Any],
    *,
    product: MasterProduct,
    hub_scoped_rules: List[Dict[str, Any]],
    listing_path_slugs: List[str],
    collection_code_by_path: Dict[str, str],
) -> None:
    if not hub_scoped_rules:
        return
    from db.manual_attributes import product_scope_path_slugs, strip_hub_scoped_fields

    category_l1 = str(canonical_fields.get("category_l1") or product.category_l1 or "").strip() or None
    collection = str(canonical_fields.get("collection") or product.collection or "").strip() or None
    path_slugs = product_scope_path_slugs(
        category_l1=category_l1,
        collection=collection,
        listing_path_slugs=listing_path_slugs,
    )
    collection_code = None
    for slug in path_slugs:
        code = collection_code_by_path.get(normalize_plp_path(slug))
        if code:
            collection_code = code
            break
    strip_hub_scoped_fields(
        canonical_fields,
        path_slugs=path_slugs,
        collection_code=collection_code,
        scoped_rules=hub_scoped_rules,
    )


def compose_canonical_fields(
    core_fields: Dict[str, Any],
    attribute_values: Dict[str, Any],
    overrides: Dict[str, Any],
) -> tuple[Dict[str, Any], Set[str]]:
    """Pure merge: core master fields, then discovered attribute values, then overrides win.

    Overrides may also introduce canonical codes that do not exist in master data (e.g. meta_title).
    Returns (fields, override_codes_applied).
    """
    fields: Dict[str, Any] = {}
    for code, value in {**core_fields, **attribute_values}.items():
        if value is None or value == "":
            continue
        fields[normalize_column_key(code)] = value
    applied: Set[str] = set()
    for code, value in overrides.items():
        key = normalize_column_key(code)
        if value is None or value == "":
            continue
        fields[key] = value
        applied.add(key)
    return canonicalize_attribute_fields(fields), applied


def map_fields_to_channel(
    canonical_fields: Dict[str, Any],
    alias_map: Dict[str, List[str]],
) -> Dict[str, Any]:
    """Pure projection: canonical code → channel attribute code(s) via aliases; unaliased codes pass through."""
    mapped: Dict[str, Any] = {}
    for canonical_code, value in canonical_fields.items():
        for channel_code in alias_map.get(canonical_code, [canonical_code]):
            mapped[channel_code] = value
    return mapped


def _manual_locked_shopping_codes(
    session: Session,
    skus: List[Optional[str]],
) -> Dict[str, Set[str]]:
    """shopping_* codes owned by a manual taxonomy assignment, per SKU."""
    from db.manual_taxonomy_assignments import resolve_manual_taxonomy_overrides

    overrides = resolve_manual_taxonomy_overrides(session, master_skus=[sku for sku in skus if sku])
    return {
        sku: {"shopping_collection", "shopping_l1", "shopping_l2"}
        for sku in overrides
    }


def _apply_canonical_shopping_field_overrides(
    payloads: List[Dict[str, Any]],
    *,
    manual_locked: Optional[Dict[str, Set[str]]] = None,
    vocabulary: Optional["ShoppingTaxonomyVocabulary"] = None,
) -> None:
    """Refresh shopping_* from taxonomy paths, subject to manual curation and vocabulary.

    Manually curated codes are left untouched, and an inferred value is only accepted
    when the curated Start Shopping vocabulary contains it.
    """
    locked_by_sku = manual_locked or {}
    for payload in payloads:
        canonical_fields = dict(payload.get("canonical_fields") or {})
        channel_fields = dict(payload.get("fields") or {})
        locked_codes = locked_by_sku.get(str(payload.get("sku") or "").strip().upper()) or set()
        shopping_values = _derive_shopping_fields_from_taxonomy_targets(payload)
        changed = False
        for code in ("shopping_collection", "shopping_l1", "shopping_l2"):
            if code in locked_codes:
                # Operator owns this code; only scrub a value the vocabulary rejects
                # (stale projection cache written before curation was enforced).
                value = canonical_fields.get(code) or channel_fields.get(code)
                if not value or vocabulary is None or vocabulary.allows(code, value):
                    continue
                value = None
            else:
                value = shopping_values.get(code)
                if value and vocabulary is not None and not vocabulary.allows(code, value):
                    value = None
            if value:
                if canonical_fields.get(code) != value:
                    canonical_fields[code] = value
                    changed = True
                if channel_fields.get(code) != value:
                    channel_fields[code] = value
                    changed = True
            else:
                if code in canonical_fields:
                    canonical_fields.pop(code, None)
                    changed = True
                if code in channel_fields:
                    channel_fields.pop(code, None)
                    changed = True
        if not changed:
            continue
        payload["canonical_fields"] = canonical_fields
        payload["fields"] = channel_fields
        payload["payload_hash"] = payload_hash(
            {
                "fields": channel_fields,
                "price": payload.get("price"),
                "association_fields": payload.get("association_fields") or {},
            }
        )


def _derive_shopping_fields_from_taxonomy_targets(payload: Dict[str, Any]) -> Dict[str, str]:
    from db.collection_landing_pages import (
        DEFAULT_CATEGORY_L1,
        _relative_shopping_collection_value,
        _relative_shopping_l1_value,
        _relative_shopping_l2_value,
        collection_path_slug,
    )

    canonical_fields = dict(payload.get("canonical_fields") or {})
    taxonomy_targets = payload.get("taxonomy_targets") or {}
    raw_path_slugs = taxonomy_targets.get("plp_path_slugs") or []
    path_slugs = [
        normalized
        for normalized in (normalize_plp_path(str(item or "")) for item in raw_path_slugs)
        if normalized
    ]
    path_set = set(path_slugs)
    if not path_set:
        return {}

    collection_slug = normalize_plp_path(
        collection_path_slug(
            canonical_fields.get("category_l1") or DEFAULT_CATEGORY_L1,
            canonical_fields.get("collection"),
        )
    )
    collection_path = collection_slug if collection_slug in path_set else ""

    # Older deploys may not expose START_SHOPPING_L1_KEYS yet; infer L1 from
    # taxonomy depth so product pushes do not fail during partial rollouts.
    l1_candidates = [
        slug
        for slug in path_set
        if len([part for part in slug.split("/") if part]) == 2 and slug != collection_path
    ]
    l1_path = min(l1_candidates, key=lambda slug: (len(slug.split("/")), slug)) if l1_candidates else ""
    l2_candidates = [
        slug
        for slug in path_set
        if l1_path and slug.startswith(f"{l1_path}/")
    ]
    l2_path = min(l2_candidates, key=lambda slug: (len(slug.split("/")), slug)) if l2_candidates else ""

    out: Dict[str, str] = {}
    collection_value = _relative_shopping_collection_value(collection_path)
    l1_value = _relative_shopping_l1_value(l1_path)
    l2_value = _relative_shopping_l2_value(l2_path)
    if collection_value:
        out["shopping_collection"] = str(collection_value)
    if l1_value:
        out["shopping_l1"] = str(l1_value)
    if l2_value:
        out["shopping_l2"] = str(l2_value)
    return out


def select_price(prices: Dict[str, Any], channel: str) -> Optional[float]:
    for price_type in PRICE_TYPE_PRIORITY.get(channel, ("base",)):
        amount = prices.get(price_type)
        if amount is not None:
            return float(amount)
    return None


def payload_hash(payload: Dict[str, Any]) -> str:
    blob = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str)
    return hashlib.sha256(blob.encode("utf-8")).hexdigest()


RELATION_FIELD_COLUMNS = (
    "product_type",
    "variant_of",
    "variant_list",
    "configurable_attributes",
    "configurable_variation_labels",
    "configurable_variations",
)

ASSOCIATION_FIELD_COLUMNS = (
    "related",
    "upsell",
    "crosssell",
)


def flatten_export_bundle(bundle: Dict[str, Any]) -> List[Dict[str, Any]]:
    """One CSV-ready row per SKU: identity + price + relation fields + every channel field.

    This is exactly what a push would send, so the dashboard can review it before going live.
    """
    payloads = bundle.get("payloads") or []
    field_columns: List[str] = []
    seen: Set[str] = set()
    for payload in payloads:
        for code in payload.get("fields") or {}:
            if code not in seen:
                seen.add(code)
                field_columns.append(code)
    field_columns.sort()

    rows: List[Dict[str, Any]] = []
    for payload in payloads:
        relation_fields = payload.get("relation_fields") or {}
        association_fields = payload.get("association_fields") or {}
        row: Dict[str, Any] = {
            "master_sku": payload.get("sku"),
            "channel_sku": payload.get("channel_sku") or payload.get("sku"),
            "channel_code": payload.get("channel_code"),
            "product_role": payload.get("product_role") or "standalone",
            "price": payload.get("price"),
            "override_codes": ",".join(payload.get("override_codes") or []),
            "payload_hash": payload.get("payload_hash"),
        }
        for column in RELATION_FIELD_COLUMNS:
            row[f"relation_{column}"] = relation_fields.get(column, "")
        for column in ASSOCIATION_FIELD_COLUMNS:
            targets = association_fields.get(column) or []
            row[f"association_{column}"] = ",".join(targets)
        fields = payload.get("fields") or {}
        for column in field_columns:
            row[column] = fields.get(column, "")
        rows.append(row)
    return rows


def build_channel_attribute_preview(
    session: Session,
    channel_code: str,
    *,
    limit: Optional[int] = None,
    only_assigned: bool = True,
    connection_id: Optional[int] = None,
) -> List[Dict[str, Any]]:
    """Flat per-attribute preview rows derived from the same payloads the push uses."""
    payloads = build_channel_product_payloads(
        session,
        channel_code,
        limit=limit,
        only_assigned=only_assigned,
        connection_id=connection_id,
    )
    channel = normalize_column_key(channel_code)
    alias_map = _build_alias_map(session, channel)

    rows: List[Dict[str, Any]] = []
    for payload in payloads:
        override_codes = set(payload["override_codes"])
        for canonical_code, value in payload["canonical_fields"].items():
            for channel_attribute_code in alias_map.get(canonical_code, [canonical_code]):
                rows.append(
                    {
                        "channel_code": channel,
                        "sku": payload["sku"],
                        "canonical_code": canonical_code,
                        "channel_attribute_code": channel_attribute_code,
                        "value": value,
                        "source": "override" if canonical_code in override_codes else "master_product",
                        "action": "upsert_attribute",
                    }
                )
        if payload["price"] is not None:
            rows.append(
                {
                    "channel_code": channel,
                    "sku": payload["sku"],
                    "canonical_code": "price",
                    "channel_attribute_code": "price",
                    "value": payload["price"],
                    "source": "master_product_price",
                    "action": "upsert_price",
                }
            )
    return rows


def _products(
    session: Session,
    *,
    skus: Optional[List[str]],
    assigned: Optional[Set[str]],
    limit: Optional[int],
) -> List[MasterProduct]:
    stmt = select(MasterProduct).where(MasterProduct.is_active.is_(True)).order_by(MasterProduct.sku)
    wanted = None
    if skus:
        wanted = {str(s).strip() for s in skus if str(s).strip()}
    if assigned is not None:
        wanted = wanted & assigned if wanted is not None else assigned
        if not wanted:
            return []
    if wanted is not None:
        stmt = stmt.where(MasterProduct.sku.in_(sorted(wanted)))
    if limit:
        stmt = stmt.limit(limit)
    return list(session.scalars(stmt).all())


def _image_fields(session: Session, skus: List[str]) -> Dict[str, Dict[str, str]]:
    from db.master_product_images import image_fields_by_sku

    return image_fields_by_sku(session, skus)


def _attribute_values(session: Session, product_ids: List[int]) -> Dict[int, List[MasterProductAttributeValue]]:
    if not product_ids:
        return {}
    rows: Dict[int, List[MasterProductAttributeValue]] = {}
    for value in session.scalars(
        select(MasterProductAttributeValue)
        .where(MasterProductAttributeValue.product_id.in_(product_ids))
        .order_by(MasterProductAttributeValue.product_id, MasterProductAttributeValue.attribute_code)
    ).all():
        rows.setdefault(value.product_id, []).append(value)
    return rows


def _prices(session: Session, product_ids: List[int]) -> Dict[int, Dict[str, float]]:
    if not product_ids:
        return {}
    rows: Dict[int, Dict[str, float]] = {}
    for price in session.scalars(
        select(MasterProductPrice).where(MasterProductPrice.product_id.in_(product_ids))
    ).all():
        if price.amount is None:
            continue
        rows.setdefault(price.product_id, {})[price.price_type] = float(price.amount)
    return rows


def _location_availability(session: Session, product_ids: List[int]) -> Dict[int, Dict[str, Any]]:
    if not product_ids:
        return {}
    location_rows = session.scalars(
        select(MasterProductLocationAvailability).where(MasterProductLocationAvailability.product_id.in_(product_ids))
    ).all()
    location_ids = sorted({int(row.location_id) for row in location_rows if row.location_id is not None})
    locations_by_id = {
        row.id: row
        for row in session.scalars(select(MasterLocationRegistry).where(MasterLocationRegistry.id.in_(location_ids))).all()
    } if location_ids else {}
    location_code_lookup = _location_code_lookup(session)

    grouped: Dict[int, Dict[str, Any]] = {}
    for row in location_rows:
        payload = grouped.setdefault(
            int(row.product_id),
            {
                "availability_by_location": {},
                "available_location_codes": [],
                "available_location_labels": [],
            },
        )
        code = _canonical_location_code(str(row.location_code or "").strip(), location_code_lookup)
        if not code:
            continue
        location = locations_by_id.get(row.location_id) if row.location_id is not None else None
        if location is None:
            location = location_code_lookup.get(code)
        label = str(location.name).strip() if location is not None and str(location.name or "").strip() else code
        payload.setdefault("_source_values_by_location", {})[code] = row.source_value
        mode = _normalize_location_availability_mode(row.is_available, row.source_value)
        payload["availability_by_location"][code] = mode
        if row.is_available:
            payload["available_location_codes"].append(code)
            payload["available_location_labels"].append(label)
        if mode == "grab_and_go":
            payload.setdefault("grab_and_go_location_codes", []).append(code)
            payload.setdefault("grab_and_go_location_labels", []).append(label)
        elif mode == "available":
            payload.setdefault("available_non_grab_location_codes", []).append(code)
            payload.setdefault("available_non_grab_location_labels", []).append(label)
        else:
            payload.setdefault("special_order_location_codes", []).append(code)
            payload.setdefault("special_order_location_labels", []).append(label)

    for payload in grouped.values():
        for key in list(payload.keys()):
            if key.endswith("_codes") or key.endswith("_labels"):
                payload[key] = sorted(dict.fromkeys(payload[key]))
        payload["is_grab_and_go_any"] = bool(payload.get("grab_and_go_location_codes"))
        payload["is_available_any"] = bool(payload.get("available_location_codes"))
        payload["is_special_order_only"] = not bool(payload.get("available_location_codes"))
        payload["default_location_availability_mode"] = _default_location_availability_mode(payload)
    return grouped


def _collection_availability_by_product(
    session: Session,
    products: List[MasterProduct],
) -> Dict[int, Dict[str, Any]]:
    """Build inherited availability for products without SKU-level location rows."""
    collections = {str(product.collection or "").strip() for product in products if str(product.collection or "").strip()}
    if not collections:
        return {}

    from db.collection_landing_schema import normalize_collection_badges

    pages = session.scalars(
        select(CollectionLandingPage)
        .where(CollectionLandingPage.is_active.is_(True))
        .where(CollectionLandingPage.collection.in_(collections))
        .order_by(CollectionLandingPage.updated_at.desc(), CollectionLandingPage.id.desc())
    ).all()
    by_scope: Dict[tuple[str, str], Dict[str, Any]] = {}
    for page in pages:
        collection = str(page.collection or "").strip().lower()
        if not collection:
            continue
        key = (str(page.category_l1 or "").strip().lower(), collection)
        by_scope.setdefault(key, normalize_collection_badges(page.badges) or {})

    location_lookup = _location_code_lookup(session)
    out: Dict[int, Dict[str, Any]] = {}
    for product in products:
        collection = str(product.collection or "").strip().lower()
        category = str(product.category_l1 or "").strip().lower()
        badges = by_scope.get((category, collection)) or by_scope.get(("", collection))
        inherited = _availability_from_collection_badges(badges, location_lookup)
        if inherited:
            out[product.id] = inherited
    return out


def _availability_from_collection_badges(
    badges: Any,
    location_lookup: Dict[str, MasterLocationRegistry],
) -> Dict[str, Any]:
    """Translate collection badge locations into Magento's location-code fields."""
    if not isinstance(badges, dict):
        return {}
    entries = badges.get("availability_locations")
    if not isinstance(entries, list):
        return {}

    payload: Dict[str, Any] = {
        "availability_by_location": {},
        "available_location_codes": [],
        "available_location_labels": [],
    }
    declared_modes: Set[str] = set()
    for entry in entries:
        if not isinstance(entry, dict):
            continue
        badge = str(entry.get("badge") or "").strip()
        mode = _collection_badge_availability_mode(badge)
        if not mode:
            continue
        declared_modes.add(mode)
        locations = entry.get("locations") if isinstance(entry.get("locations"), list) else []
        for raw_location in locations:
            code = _canonical_location_code(str(raw_location or "").strip(), location_lookup)
            if not code:
                continue
            location = location_lookup.get(code) or location_lookup.get(_normalize_location_token(raw_location))
            label = str(location.name or "").strip() if location is not None else str(raw_location).strip()
            payload["availability_by_location"][code] = mode
            if mode != "special_order":
                payload["available_location_codes"].append(code)
                payload["available_location_labels"].append(label)
            if mode == "grab_and_go":
                payload.setdefault("grab_and_go_location_codes", []).append(code)
                payload.setdefault("grab_and_go_location_labels", []).append(label)
            elif mode == "available":
                payload.setdefault("available_non_grab_location_codes", []).append(code)
                payload.setdefault("available_non_grab_location_labels", []).append(label)
            else:
                payload.setdefault("special_order_location_codes", []).append(code)
                payload.setdefault("special_order_location_labels", []).append(label)

    if not declared_modes:
        return {}
    for key in list(payload):
        if key.endswith("_codes") or key.endswith("_labels"):
            payload[key] = sorted(dict.fromkeys(payload[key]))
    payload["is_grab_and_go_any"] = bool(payload.get("grab_and_go_location_codes"))
    payload["is_available_any"] = bool(payload.get("available_location_codes"))
    payload["is_special_order_only"] = not payload["is_available_any"]
    if payload["is_grab_and_go_any"]:
        payload["default_location_availability_mode"] = "grab_and_go"
    elif payload["is_available_any"]:
        payload["default_location_availability_mode"] = "available"
    elif "special_order" in declared_modes:
        payload["default_location_availability_mode"] = "special_order"
    return payload


def _collection_badge_availability_mode(badge: str) -> str:
    label = str(badge or "").strip().lower()
    if "grab" in label:
        return "grab_and_go"
    if "special" in label:
        return "special_order"
    if "available" in label:
        return "available"
    return ""


def _attach_location_availability_fields(
    canonical_fields: Dict[str, Any],
    availability: Dict[str, Any],
    *,
    force_item_availability: bool = False,
) -> None:
    if not availability:
        return
    if _prefers_grab_and_go(canonical_fields) and not availability.get("grab_and_go_location_codes"):
        promoted_codes = list(availability.get("available_location_codes") or [])
        promoted_labels = list(availability.get("available_location_labels") or [])
        if promoted_codes:
            availability["grab_and_go_location_codes"] = promoted_codes
            availability["grab_and_go_location_labels"] = promoted_labels
            availability["available_non_grab_location_codes"] = []
            availability["available_non_grab_location_labels"] = []
            availability["availability_by_location"] = {
                code: ("grab_and_go" if code in promoted_codes else value)
                for code, value in (availability.get("availability_by_location") or {}).items()
            }
            availability["is_grab_and_go_any"] = True
            availability["default_location_availability_mode"] = "grab_and_go"
    for key, value in availability.items():
        if str(key).startswith("_"):
            continue
        canonical_fields[key] = value
    # Magento multiselect attributes receive labels here; sync_service resolves
    # them to Magento's stored option IDs before the PUT request.
    grab_codes = [str(c).strip() for c in (availability.get("grab_and_go_location_codes") or []) if str(c).strip()]
    if availability.get("available_non_grab_location_codes") is not None:
        available_codes = [
            str(c).strip()
            for c in (availability.get("available_non_grab_location_codes") or [])
            if str(c).strip()
        ]
    else:
        available_codes = [
            str(c).strip()
            for c in (availability.get("available_location_codes") or [])
            if str(c).strip() and str(c).strip() not in set(grab_codes)
        ]
    special_codes = [str(c).strip() for c in (availability.get("special_order_location_codes") or []) if str(c).strip()]
    grab_labels = [str(label).strip() for label in (availability.get("grab_and_go_location_labels") or []) if str(label).strip()]
    available_labels = [
        str(label).strip()
        for label in (availability.get("available_non_grab_location_labels") or [])
        if str(label).strip()
    ]
    special_labels = [str(label).strip() for label in (availability.get("special_order_location_labels") or []) if str(label).strip()]
    canonical_fields["pim_avail_grab_go"] = ",".join(grab_labels)
    canonical_fields["pim_avail_available"] = ",".join(available_labels)
    canonical_fields["pim_avail_special"] = ",".join(special_labels)
    if force_item_availability or not canonical_fields.get("item_availability"):
        mode = availability.get("default_location_availability_mode")
        if mode == "grab_and_go":
            canonical_fields["item_availability"] = "Grab & Go"
            canonical_fields.setdefault("collection_availability", "Grab & Go")
        elif mode == "available":
            canonical_fields["item_availability"] = "Available"
            canonical_fields.setdefault("collection_availability", "Available")
        elif mode == "special_order":
            canonical_fields["item_availability"] = "Special Order"
            canonical_fields.setdefault("collection_availability", "Special Order")


def _attach_collection_location_fields(
    session: Session,
    canonical_fields: Dict[str, Any],
    *,
    category_l1: Optional[str],
    collection: Optional[str],
) -> None:
    context = collection_location_context(
        session,
        category_l1=category_l1,
        collection_name=collection,
    )
    settings = context.get("location_settings") if isinstance(context.get("location_settings"), dict) else {}
    resolved = context.get("resolved_location") if isinstance(context.get("resolved_location"), dict) else {}
    directory = context.get("location_directory") if isinstance(context.get("location_directory"), list) else []
    if settings:
        canonical_fields["location_settings"] = settings
        canonical_fields["default_location_code"] = settings.get("default_location_code")
        canonical_fields["default_location_name"] = settings.get("default_location_name")
        canonical_fields["allow_geolocation"] = bool(settings.get("allow_geolocation"))
        canonical_fields["allow_manual_location_selection"] = bool(settings.get("allow_manual_selection"))
        canonical_fields["location_selection_mode"] = settings.get("selection_mode")
        canonical_fields["location_selector_label"] = settings.get("location_label")
    if resolved:
        canonical_fields["resolved_default_location_code"] = resolved.get("selected_location_code")
        canonical_fields["resolved_default_location_name"] = resolved.get("selected_location_name")
        canonical_fields["resolved_location_selection_strategy"] = resolved.get("selection_strategy")
    if directory:
        canonical_fields["location_directory"] = directory


def _normalize_location_availability_mode(is_available: bool, source_value: Any) -> str:
    text = str(source_value or "").strip().lower()
    if "grab" in text:
        return "grab_and_go"
    return "available" if bool(is_available) else "special_order"


def _prefers_grab_and_go(fields: Dict[str, Any]) -> bool:
    for key in ("grab_n_go", "grab_ngo", "grab_go"):
        if _truthy_token(fields.get(key)):
            return True
    text = str(fields.get("item_availability") or fields.get("collection_availability") or "").strip().lower()
    return "grab" in text


def _default_location_availability_mode(payload: Dict[str, Any]) -> str:
    if payload.get("grab_and_go_location_codes"):
        return "grab_and_go"
    if payload.get("available_location_codes"):
        return "available"
    return "special_order"


def _location_code_lookup(session: Session) -> Dict[str, MasterLocationRegistry]:
    rows = session.scalars(select(MasterLocationRegistry).where(MasterLocationRegistry.is_active.is_(True))).all()
    lookup: Dict[str, MasterLocationRegistry] = {}
    for row in rows:
        if row.code:
            lookup[str(row.code).strip()] = row
        labels = []
        aliases = row.aliases if isinstance(row.aliases, dict) else {}
        labels.extend(aliases.get("labels", []) if isinstance(aliases.get("labels"), list) else [])
        labels.append(row.name)
        for label in labels:
            key = _normalize_location_token(label)
            if key and key not in lookup:
                lookup[key] = row
    return lookup


def _canonical_location_code(raw_code: str, lookup: Dict[str, MasterLocationRegistry]) -> str:
    text = str(raw_code or "").strip()
    if not text:
        return ""
    if text in lookup:
        return str(lookup[text].code)
    normalized = _normalize_location_token(text)
    row = lookup.get(normalized)
    if row is not None:
        return str(row.code)
    return text


def _normalize_location_token(value: Any) -> str:
    text = normalize_column_key(value)
    if text.startswith("location_"):
        text = text[len("location_") :]
    for suffix in ("_warehouse", "_store", "_pickup_location", "_location"):
        if text.endswith(suffix):
            text = text[: -len(suffix)]
    return text


def _truthy_token(value: Any) -> bool:
    return str(value or "").strip().lower() in {"1", "true", "yes", "y", "available", "in stock", "grab go", "grab and go"}


def _core_fields(product: MasterProduct) -> Dict[str, Any]:
    return {
        "title": product.name,
        "brand": product.brand,
        "manufacturer": product.manufacturer,
        "product_family": product.product_family,
        "category_l1": product.category_l1,
        "category_l2": product.category_l2,
        "category_l3": product.category_l3,
        "collection": product.collection,
        "assembly_type": product.assembly_type,
        "item_size": product.item_size,
        "status": product.status,
    }