"""Locked master-attribute vocabularies (allowed values + aliases).

Dashboard: GET/PUT /api/manual-attributes
Import: canonicalize/drop polluted CSV values before attribute write.
Values API: when locked, return vocabulary values only.
"""

from __future__ import annotations

from typing import Any, Dict, List, Optional, Set, Tuple

from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy import delete, select
from sqlalchemy.orm import Session

from channel.url_canonical import normalize_plp_path
from db.channel_exports import resolve_pipeline_channel_code
from db.channel_listing_path import _clean_filter_keys, _merged_enabled_filters
from db.models import (
    ChannelListingPath,
    ManualAttributeVocabulary,
    ManualAttributeVocabularyValue,
    MasterAttributeDefinition,
    MasterCollectionRegistry,
)
from db.source_imports import normalize_column_key

# Built-in seed used by ensure_default_manual_attributes (tests / fresh DBs).
DEFAULT_MANUAL_ATTRIBUTES: List[Dict[str, Any]] = [
    {
        "attribute_code": "assembled_or_rta",
        "label": "Assembled or RTA",
        "is_locked": True,
        "unknown_policy": "drop",
        "covered_codes": ["assembled_or_rta", "assembly_type"],
        "sort_order": 10,
        "notes": "Locked vocabulary: Assembled or Unassembled/RTA only",
        "values": [
            {
                "value": "Assembled",
                "aliases": ["Assembled", "assembled"],
                "sort_order": 10,
            },
            {
                "value": "Unassembled/RTA",
                "aliases": [
                    "Unassembled/RTA",
                    "RTA",
                    "Unassembled",
                    "unassembled",
                    "Ready-to-Assemble",
                    "ready to assemble",
                    "ready-to-assemble",
                ],
                "sort_order": 20,
            },
        ],
    }
]

UNRESTRICTED_MANUAL_ATTRIBUTE_CODES: Set[str] = {
    "name",
    "title",
    "category_l1",
    "category_l2",
    "category_l3",
    "collection",
    "meta_title",
    "meta_description",
    "meta_keywords",
    "product_url_slug",
    "url_slug",
    "url_key",
}


def _normalize_aliases(values: Any, *, include_canonical: Optional[str] = None) -> List[str]:
    if isinstance(values, str):
        values = [part.strip() for part in values.split(",")]
    if not isinstance(values, list):
        values = []
    seen = set()
    out: List[str] = []
    for item in [*([include_canonical] if include_canonical else []), *values]:
        text = str(item or "").strip()
        if not text:
            continue
        key = text.lower()
        if key in seen:
            continue
        seen.add(key)
        out.append(text)
    return out


def _covered_codes(attribute_code: str, covered: Any) -> List[str]:
    codes = _normalize_aliases(covered, include_canonical=attribute_code)
    normalized: List[str] = []
    seen = set()
    for code in codes:
        key = normalize_column_key(code)
        if key and key not in seen:
            seen.add(key)
            normalized.append(key)
    return normalized


def _normalize_slug_list(values: Any) -> List[str]:
    if isinstance(values, str):
        values = [part.strip() for part in values.split(",")]
    if not isinstance(values, list):
        return []
    seen = set()
    out: List[str] = []
    for item in values:
        slug = normalize_plp_path(item or "")
        if not slug or slug in seen:
            continue
        seen.add(slug)
        out.append(slug)
    return out


def _normalize_code_list(values: Any) -> List[str]:
    if isinstance(values, str):
        values = [part.strip() for part in values.split(",")]
    if not isinstance(values, list):
        return []
    seen = set()
    out: List[str] = []
    for item in values:
        code = str(item or "").strip().upper()
        if not code or code in seen:
            continue
        seen.add(code)
        out.append(code)
    return out


def _normalize_channel_list(values: Any) -> List[str]:
    if isinstance(values, str):
        values = [part.strip() for part in values.split(",")]
    if not isinstance(values, list):
        return []
    seen = set()
    out: List[str] = []
    for item in values:
        text = str(item or "").strip().lower()
        if not text:
            continue
        code = resolve_pipeline_channel_code(text)
        if code in seen:
            continue
        seen.add(code)
        out.append(code)
    return out


def _is_unrestricted_manual_attribute_code(attribute_code: str) -> bool:
    return normalize_column_key(attribute_code) in UNRESTRICTED_MANUAL_ATTRIBUTE_CODES


def _metadata_dict(
    vocab: ManualAttributeVocabulary,
    definition: Optional[MasterAttributeDefinition] = None,
) -> Dict[str, Any]:
    raw = vocab.raw_payload if isinstance(vocab.raw_payload, dict) else {}
    target_scopes = raw.get("target_scopes") if isinstance(raw.get("target_scopes"), dict) else {}
    return {
        "data_type": normalize_column_key(
            raw.get("data_type") or (definition.data_type if definition else None) or "text"
        ) or "text",
        "purpose": str(raw.get("purpose") or (definition.purpose if definition else None) or "presentation").strip()
        or "presentation",
        "is_global": bool(raw.get("is_global", False)),
        "hub_path_slugs": _normalize_slug_list(raw.get("hub_path_slugs")),
        "taxonomy_path_slugs": _normalize_slug_list(raw.get("taxonomy_path_slugs")),
        "collection_codes": _normalize_code_list(raw.get("collection_codes")),
        "target_scopes": target_scopes,
        "publish_to_magento": bool(raw.get("publish_to_magento", True)),
        "disabled_channels": _normalize_channel_list(raw.get("disabled_channels")),
    }


def _value_dict(row: ManualAttributeVocabularyValue) -> Dict[str, Any]:
    aliases = _normalize_aliases(
        row.aliases_json if isinstance(row.aliases_json, list) else [],
        include_canonical=row.value,
    )
    return {
        "value": str(row.value or "").strip(),
        "aliases": aliases,
        "sort_order": int(row.sort_order or 100),
        "is_active": bool(row.is_active),
        "notes": row.notes,
    }


def _attribute_dict(
    vocab: ManualAttributeVocabulary,
    values: List[ManualAttributeVocabularyValue],
    definition: Optional[MasterAttributeDefinition] = None,
) -> Dict[str, Any]:
    meta = _metadata_dict(vocab, definition)
    return {
        "attribute_code": normalize_column_key(vocab.attribute_code),
        "label": str(vocab.label or "").strip() or vocab.attribute_code,
        "is_locked": bool(vocab.is_locked),
        "unknown_policy": str(vocab.unknown_policy or "drop").strip().lower() or "drop",
        "covered_codes": _covered_codes(vocab.attribute_code, vocab.covered_codes_json),
        "is_active": bool(vocab.is_active),
        "sort_order": int(vocab.sort_order or 100),
        "notes": vocab.notes,
        "data_type": meta["data_type"],
        "purpose": meta["purpose"],
        "is_global": meta["is_global"],
        "hub_path_slugs": meta["hub_path_slugs"],
        "taxonomy_path_slugs": meta["taxonomy_path_slugs"],
        "collection_codes": meta["collection_codes"],
        "target_scopes": meta["target_scopes"],
        "publish_to_magento": meta["publish_to_magento"],
        "disabled_channels": meta["disabled_channels"],
        "values": [
            _value_dict(item)
            for item in sorted(values, key=lambda row: (row.sort_order, row.id))
            if item.is_active
        ],
    }


def list_manual_attributes(session: Session, *, include_inactive: bool = False) -> Dict[str, Any]:
    stmt = select(ManualAttributeVocabulary).order_by(
        ManualAttributeVocabulary.sort_order,
        ManualAttributeVocabulary.attribute_code,
        ManualAttributeVocabulary.id,
    )
    if not include_inactive:
        stmt = stmt.where(ManualAttributeVocabulary.is_active.is_(True))
    vocabs = list(session.scalars(stmt).all())
    if not vocabs:
        return {"attributes": [], "count": 0}
    vocab_ids = [row.id for row in vocabs]
    value_rows = session.scalars(
        select(ManualAttributeVocabularyValue)
        .where(ManualAttributeVocabularyValue.vocabulary_id.in_(vocab_ids))
        .order_by(
            ManualAttributeVocabularyValue.vocabulary_id,
            ManualAttributeVocabularyValue.sort_order,
            ManualAttributeVocabularyValue.id,
        )
    ).all()
    by_vocab: Dict[int, List[ManualAttributeVocabularyValue]] = {}
    for row in value_rows:
        by_vocab.setdefault(row.vocabulary_id, []).append(row)
    definitions = {
        normalize_column_key(row.attribute_code): row
        for row in session.scalars(
            select(MasterAttributeDefinition).where(
                MasterAttributeDefinition.attribute_code.in_(
                    [normalize_column_key(v.attribute_code) for v in vocabs if normalize_column_key(v.attribute_code)]
                )
            )
        ).all()
        if normalize_column_key(row.attribute_code)
    }
    attributes = [
        _attribute_dict(vocab, by_vocab.get(vocab.id, []), definitions.get(normalize_column_key(vocab.attribute_code)))
        for vocab in vocabs
    ]
    return {"attributes": attributes, "count": len(attributes)}


def ensure_default_manual_attributes(session: Session) -> Dict[str, Any]:
    """Idempotently seed built-in locked vocabularies when missing."""
    existing = {
        normalize_column_key(code)
        for (code,) in session.execute(select(ManualAttributeVocabulary.attribute_code)).all()
        if code
    }
    created = 0
    for item in DEFAULT_MANUAL_ATTRIBUTES:
        code = normalize_column_key(item["attribute_code"])
        if code in existing:
            continue
        _upsert_one_attribute(session, item)
        created += 1
    if created:
        session.flush()
    result = list_manual_attributes(session)
    result["seeded_count"] = created
    return result


def replace_manual_attributes(session: Session, payload: Dict[str, Any]) -> Dict[str, Any]:
    attributes = payload.get("attributes")
    if not isinstance(attributes, list):
        raise ValueError("attributes must be a list")

    replace_all = bool(payload.get("replace_all", False))
    saved = 0
    seen_codes: set[str] = set()
    for item in attributes:
        if not isinstance(item, dict):
            continue
        code = normalize_column_key(item.get("attribute_code") or "")
        if not code or code in seen_codes:
            continue
        seen_codes.add(code)
        _upsert_one_attribute(session, {**item, "attribute_code": code})
        saved += 1

    deleted = 0
    if replace_all:
        if seen_codes:
            stale = session.scalars(
                select(ManualAttributeVocabulary).where(
                    ~ManualAttributeVocabulary.attribute_code.in_(sorted(seen_codes))
                )
            ).all()
        else:
            stale = list(session.scalars(select(ManualAttributeVocabulary)).all())
        for row in stale:
            session.execute(
                delete(ManualAttributeVocabularyValue).where(
                    ManualAttributeVocabularyValue.vocabulary_id == row.id
                )
            )
            session.delete(row)
            deleted += 1

    session.flush()
    result = list_manual_attributes(session)
    result["saved_count"] = saved
    result["deleted_count"] = deleted
    return result


def _upsert_one_attribute(session: Session, item: Dict[str, Any]) -> ManualAttributeVocabulary:
    code = normalize_column_key(item.get("attribute_code") or "")
    if not code:
        raise ValueError("attribute_code is required")
    vocab = session.scalar(
        select(ManualAttributeVocabulary).where(ManualAttributeVocabulary.attribute_code == code)
    )
    if vocab is None:
        vocab = ManualAttributeVocabulary(attribute_code=code)
        session.add(vocab)
        session.flush()
    else:
        session.execute(
            delete(ManualAttributeVocabularyValue).where(
                ManualAttributeVocabularyValue.vocabulary_id == vocab.id
            )
        )

    unrestricted_values = _is_unrestricted_manual_attribute_code(code)

    unknown_policy = str(item.get("unknown_policy") or "drop").strip().lower() or "drop"
    if unknown_policy not in {"drop", "keep"}:
        unknown_policy = "drop"
    if unrestricted_values:
        unknown_policy = "keep"

    vocab.label = str(item.get("label") or "").strip() or code.replace("_", " ").title()
    vocab.is_locked = False if unrestricted_values else bool(item.get("is_locked", True))
    vocab.unknown_policy = unknown_policy
    vocab.covered_codes_json = _covered_codes(code, item.get("covered_codes"))
    vocab.is_active = bool(item.get("is_active", True))
    vocab.sort_order = int(item.get("sort_order") or 100)
    vocab.notes = str(item.get("notes") or "").strip() or None
    vocab.raw_payload = {
        **item,
        "attribute_code": code,
        "covered_codes": _covered_codes(code, item.get("covered_codes")),
        "data_type": normalize_column_key(item.get("data_type") or "text") or "text",
        "purpose": str(item.get("purpose") or "presentation").strip() or "presentation",
        "hub_path_slugs": _normalize_slug_list(item.get("hub_path_slugs")),
        "taxonomy_path_slugs": _normalize_slug_list(item.get("taxonomy_path_slugs")),
        "collection_codes": _normalize_code_list(item.get("collection_codes")),
        "target_scopes": item.get("target_scopes") if isinstance(item.get("target_scopes"), dict) else {},
        "publish_to_magento": bool(item.get("publish_to_magento", True)),
        "is_global": bool(item.get("is_global", False)),
        "disabled_channels": _normalize_channel_list(item.get("disabled_channels")),
    }

    seen_values: set[str] = set()
    if unrestricted_values:
        session.flush()
        return vocab
    for index, value_item in enumerate(item.get("values") or []):
        if not isinstance(value_item, dict):
            continue
        value = str(value_item.get("value") or "").strip()
        if not value:
            continue
        value_key = value.lower()
        if value_key in seen_values:
            continue
        seen_values.add(value_key)
        aliases = _normalize_aliases(value_item.get("aliases"), include_canonical=value)
        session.add(
            ManualAttributeVocabularyValue(
                vocabulary_id=vocab.id,
                value=value,
                aliases_json=aliases,
                sort_order=int(value_item.get("sort_order") if value_item.get("sort_order") is not None else index * 10),
                is_active=bool(value_item.get("is_active", True)),
                raw_payload=value_item,
                notes=str(value_item.get("notes") or "").strip() or None,
            )
        )
    session.flush()
    return vocab


def _manual_attribute_applies_to_channel(row: Dict[str, Any], channel_code: str) -> bool:
    channel = resolve_pipeline_channel_code(channel_code)
    disabled = set(_normalize_channel_list(row.get("disabled_channels")))
    if channel in disabled:
        return False
    if channel == "magento" and not bool(row.get("publish_to_magento", True)):
        return False
    return True


def _path_is_within_scope(path_slug: str, scope_slug: str) -> bool:
    return path_slug == scope_slug or path_slug.startswith(f"{scope_slug}/")


def _path_ancestor_slugs(path_slug: str) -> List[str]:
    parts = [part for part in normalize_plp_path(path_slug).split("/") if part]
    return ["/".join(parts[:index]) for index in range(1, len(parts) + 1)]


def _manual_attribute_matches_path(
    row: Dict[str, Any],
    *,
    path_slug: str,
    collection_code_by_path: Dict[str, str],
) -> bool:
    hubs = _normalize_slug_list(row.get("hub_path_slugs"))
    taxonomies = _normalize_slug_list(row.get("taxonomy_path_slugs"))
    collection_codes = set(_normalize_code_list(row.get("collection_codes")))
    if bool(row.get("is_global")):
        return True
    if not hubs and not taxonomies and not collection_codes:
        return False
    if any(_path_is_within_scope(path_slug, hub_slug) for hub_slug in hubs):
        return True
    if any(_path_is_within_scope(path_slug, taxonomy_slug) for taxonomy_slug in taxonomies):
        return True
    collection_code = collection_code_by_path.get(path_slug)
    if collection_code and collection_code in collection_codes:
        return True
    return False


def sync_manual_attribute_enabled_filters(
    session: Session,
    *,
    channel_code: str = "magento",
) -> Dict[str, Any]:
    channel = resolve_pipeline_channel_code(channel_code)
    attributes = [
        row
        for row in (list_manual_attributes(session).get("attributes") or [])
        if row.get("is_active") and _manual_attribute_applies_to_channel(row, channel)
    ]
    listing_paths = list(
        session.scalars(
            select(ChannelListingPath)
            .where(ChannelListingPath.is_active.is_(True))
            .order_by(ChannelListingPath.path_slug)
        ).all()
    )
    collection_code_by_path = {
        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)
    }
    listing_path_by_slug = {
        normalize_plp_path(path.path_slug or ""): path
        for path in listing_paths
        if normalize_plp_path(path.path_slug or "")
    }

    updated_paths = 0
    cleared_paths = 0
    touched_paths: List[str] = []
    for path in listing_paths:
        slug = normalize_plp_path(path.path_slug or "")
        if not slug:
            continue
        manual_filters: List[str] = []
        seen: Set[str] = set()
        for row in sorted(
            attributes,
            key=lambda item: (int(item.get("sort_order") or 100), str(item.get("attribute_code") or "")),
        ):
            if not _manual_attribute_matches_path(
                row,
                path_slug=slug,
                collection_code_by_path=collection_code_by_path,
            ):
                continue
            key = normalize_column_key(row.get("attribute_code") or "")
            if not key or key in seen:
                continue
            seen.add(key)
            manual_filters.append(key)

        facet_payload = dict(path.facet_terms) if isinstance(path.facet_terms, dict) else {}
        inherited_explicit: List[str] = []
        for ancestor_slug in _path_ancestor_slugs(slug):
            ancestor = listing_path_by_slug.get(ancestor_slug)
            ancestor_payload = dict(ancestor.facet_terms) if ancestor is not None and isinstance(ancestor.facet_terms, dict) else {}
            inherited_explicit = _merged_enabled_filters(
                inherited_explicit,
                ancestor_payload.get("explicit_enabled_filters") if isinstance(ancestor_payload.get("explicit_enabled_filters"), list) else [],
            )
        current_manual = _clean_filter_keys(
            facet_payload.get("manual_enabled_filters") if isinstance(facet_payload.get("manual_enabled_filters"), list) else []
        )
        next_manual = _clean_filter_keys(manual_filters)
        next_enabled = _merged_enabled_filters(inherited_explicit, next_manual)
        current_enabled = _clean_filter_keys(
            facet_payload.get("enabled_filters") if isinstance(facet_payload.get("enabled_filters"), list) else []
        )
        if current_manual == next_manual and current_enabled == next_enabled:
            continue
        if next_manual:
            facet_payload["manual_enabled_filters"] = next_manual
        else:
            facet_payload.pop("manual_enabled_filters", None)
        if next_enabled:
            facet_payload["enabled_filters"] = next_enabled
        else:
            facet_payload.pop("enabled_filters", None)
        path.facet_terms = facet_payload or None
        updated_paths += 1
        if not next_manual:
            cleared_paths += 1
        touched_paths.append(slug)

    if updated_paths:
        session.flush()
    return {
        "channel_code": channel,
        "attribute_count": len(attributes),
        "listing_paths_updated": updated_paths,
        "listing_paths_cleared": cleared_paths,
        "updated_path_slugs": touched_paths,
        "sample_paths": touched_paths[:25],
    }


def list_hub_scoped_attribute_rules(session: Session) -> List[Dict[str, Any]]:
    """Manual attrs with explicit hub/taxonomy/collection scope (used to strip product payloads)."""
    rules: List[Dict[str, Any]] = []
    for row in list_manual_attributes(session).get("attributes") or []:
        if not row.get("is_active"):
            continue
        if bool(row.get("is_global")):
            continue
        hubs = _normalize_slug_list(row.get("hub_path_slugs"))
        taxonomies = _normalize_slug_list(row.get("taxonomy_path_slugs"))
        collection_codes = _normalize_code_list(row.get("collection_codes"))
        if not hubs and not taxonomies and not collection_codes:
            continue
        code = normalize_column_key(row.get("attribute_code") or "")
        if not code:
            continue
        rules.append(
            {
                "attribute_code": code,
                "covered_codes": _covered_codes(code, row.get("covered_codes")),
                "is_global": False,
                "hub_path_slugs": hubs,
                "taxonomy_path_slugs": taxonomies,
                "collection_codes": collection_codes,
            }
        )
    return rules


def attribute_matches_product_scope(
    rule: Dict[str, Any],
    *,
    path_slugs: List[str],
    collection_code: Optional[str] = None,
) -> bool:
    """True when any product path/collection is within the attribute's hub/taxonomy/collection scope."""
    if bool(rule.get("is_global")):
        return True
    hubs = _normalize_slug_list(rule.get("hub_path_slugs"))
    taxonomies = _normalize_slug_list(rule.get("taxonomy_path_slugs"))
    collection_codes = set(_normalize_code_list(rule.get("collection_codes")))
    if not hubs and not taxonomies and not collection_codes:
        return True
    for path_slug in path_slugs:
        slug = normalize_plp_path(path_slug or "")
        if not slug:
            continue
        if any(_path_is_within_scope(slug, hub_slug) for hub_slug in hubs):
            return True
        if any(_path_is_within_scope(slug, taxonomy_slug) for taxonomy_slug in taxonomies):
            return True
    code = str(collection_code or "").strip().upper()
    if code and code in collection_codes:
        return True
    return False


def strip_hub_scoped_fields(
    fields: Dict[str, Any],
    *,
    path_slugs: List[str],
    collection_code: Optional[str] = None,
    scoped_rules: Optional[List[Dict[str, Any]]] = None,
) -> Dict[str, Any]:
    """Remove hub-scoped attribute codes that do not apply to this product's paths."""
    if not isinstance(fields, dict) or not fields or not scoped_rules:
        return fields
    drop_codes: Set[str] = set()
    for rule in scoped_rules:
        if attribute_matches_product_scope(
            rule,
            path_slugs=path_slugs,
            collection_code=collection_code,
        ):
            continue
        for code in rule.get("covered_codes") or [rule.get("attribute_code")]:
            key = normalize_column_key(code or "")
            if key:
                drop_codes.add(key)
    if not drop_codes:
        return fields
    for key in list(fields.keys()):
        if normalize_column_key(key) in drop_codes:
            fields.pop(key, None)
    return fields


def product_scope_path_slugs(
    *,
    category_l1: Optional[str] = None,
    collection: Optional[str] = None,
    listing_path_slugs: Optional[List[str]] = None,
) -> List[str]:
    """Resolve hub/collection path slugs for payload scoping (listing paths win, else category/collection)."""
    from db.collection_landing_pages import collection_path_slug

    seen: Set[str] = set()
    out: List[str] = []
    for item in listing_path_slugs or []:
        slug = normalize_plp_path(item or "")
        if slug and slug not in seen:
            seen.add(slug)
            out.append(slug)
    if out:
        return out
    hub = normalize_plp_path(category_l1 or "")
    if hub and hub not in seen:
        seen.add(hub)
        out.append(hub)
    if category_l1 and collection:
        coll_slug = normalize_plp_path(collection_path_slug(category_l1, collection))
        if coll_slug and coll_slug not in seen:
            out.append(coll_slug)
    return out


def ensure_manual_attribute_magento_aliases(
    session: Session,
    rows: List[Dict[str, Any]],
) -> Dict[str, Any]:
    """Ensure Magento channel aliases (create_new + option sync) for publishable manual attrs."""
    from db.channel_remote_attributes import SYNC_OPTIONS_FLAG
    from db.channel_attribute_provision import CREATE_NEW_ACTION
    from db.models import ChannelAttributeAlias

    ensured: List[str] = []
    skipped: List[str] = []
    for row in rows:
        code = normalize_column_key(row.get("attribute_code") or "")
        if not code:
            continue
        if not bool(row.get("publish_to_magento", True)):
            skipped.append(code)
            continue
        if "magento" in set(_normalize_channel_list(row.get("disabled_channels"))):
            skipped.append(code)
            continue
        data_type = normalize_column_key(row.get("data_type") or "text") or "text"
        alias = session.scalar(
            select(ChannelAttributeAlias)
            .where(ChannelAttributeAlias.channel_code == "magento")
            .where(ChannelAttributeAlias.canonical_code == code)
            .where(ChannelAttributeAlias.channel_attribute_code == code)
            .limit(1)
        )
        transform = dict(alias.transform_rule) if alias is not None and isinstance(alias.transform_rule, dict) else {}
        if data_type in {"select", "multiselect", "boolean", "yesno"}:
            transform[SYNC_OPTIONS_FLAG] = True
        if alias is None:
            session.add(
                ChannelAttributeAlias(
                    channel_code="magento",
                    canonical_code=code,
                    channel_attribute_code=code,
                    mapping_scope="dynamic",
                    action=CREATE_NEW_ACTION,
                    data_type=data_type,
                    transform_rule=transform or None,
                    is_active=True,
                    notes="Auto-created from manual attribute publish",
                )
            )
        else:
            alias.is_active = True
            alias.data_type = data_type or alias.data_type
            if alias.action != "use_existing":
                alias.action = CREATE_NEW_ACTION
            alias.transform_rule = transform or alias.transform_rule
            if not alias.notes:
                alias.notes = "Auto-created from manual attribute publish"
        ensured.append(code)
    if ensured:
        session.flush()
    return {"ensured_codes": ensured, "skipped_codes": skipped, "ensured_count": len(ensured)}


def publish_manual_attributes(
    session: Session,
    *,
    attribute_codes: Optional[List[str]] = None,
    active_only: bool = True,
    provision_magento: bool = True,
    push_filters: bool = True,
    magento_connection_id: Optional[int] = None,
    dry_run: bool = False,
) -> Dict[str, Any]:
    payload = list_manual_attributes(session, include_inactive=not active_only)
    rows = payload.get("attributes") or []
    wanted = {
        normalize_column_key(code)
        for code in (attribute_codes or [])
        if normalize_column_key(code)
    }
    if wanted:
        rows = [row for row in rows if normalize_column_key(row.get("attribute_code")) in wanted]
    if active_only:
        rows = [row for row in rows if row.get("is_active")]
    if not rows:
        return {"published_count": 0, "published_codes": []}

    values = []
    for index, row in enumerate(rows, start=1):
        code = normalize_column_key(row.get("attribute_code") or "")
        if not code:
            continue
        target_scopes = row.get("target_scopes") if isinstance(row.get("target_scopes"), dict) else {}
        merged_scopes = {
            **target_scopes,
            "manual_attribute": True,
            "is_locked": bool(row.get("is_locked")),
            "unknown_policy": row.get("unknown_policy") or "drop",
            "is_global": bool(row.get("is_global", False)),
            "hub_path_slugs": _normalize_slug_list(row.get("hub_path_slugs")),
            "taxonomy_path_slugs": _normalize_slug_list(row.get("taxonomy_path_slugs")),
            "collection_codes": _normalize_code_list(row.get("collection_codes")),
            "covered_codes": _covered_codes(code, row.get("covered_codes")),
            "publish_to_magento": bool(row.get("publish_to_magento", True)),
            "disabled_channels": _normalize_channel_list(row.get("disabled_channels")),
            "value_count": len(row.get("values") or []),
        }
        values.append(
            {
                "attribute_code": code,
                "label": str(row.get("label") or "").strip() or code.replace("_", " ").title(),
                "purpose": str(row.get("purpose") or "presentation").strip() or "presentation",
                "data_type": normalize_column_key(row.get("data_type") or "text") or "text",
                "target_scopes": merged_scopes,
                "is_required": False,
                "is_active": bool(row.get("is_active", True)),
                "notes": str(row.get("notes") or "").strip() or None,
            }
        )
    if not dry_run:
        stmt = pg_insert(MasterAttributeDefinition).values(values)
        stmt = stmt.on_conflict_do_update(
            constraint="uq_master_attribute_definition_code",
            set_={
                "label": stmt.excluded.label,
                "purpose": stmt.excluded.purpose,
                "data_type": stmt.excluded.data_type,
                "target_scopes": stmt.excluded.target_scopes,
                "is_required": stmt.excluded.is_required,
                "is_active": stmt.excluded.is_active,
                "notes": stmt.excluded.notes,
            },
        )
        session.execute(stmt)
        session.flush()
    filter_sync = sync_manual_attribute_enabled_filters(session, channel_code="magento")
    alias_sync = ensure_manual_attribute_magento_aliases(session, rows) if not dry_run else {
        "ensured_codes": [],
        "skipped_codes": [],
        "ensured_count": 0,
        "dry_run": True,
    }

    provision_result: Dict[str, Any] = {"status": "skipped", "reason": "disabled"}
    option_sync: Dict[str, Any] = {"status": "skipped", "reason": "disabled"}
    filter_push: Dict[str, Any] = {"status": "skipped", "reason": "disabled"}

    if provision_magento and not dry_run and alias_sync.get("ensured_codes"):
        from db.channel_attribute_provision import provision_channel_attributes
        from db.channel_remote_attributes import _seed_magento_options
        from db.models import MagentoConnection

        connection_id = magento_connection_id
        if connection_id is None:
            connection_id = session.scalar(
                select(MagentoConnection.id)
                .where(MagentoConnection.status == "active")
                .order_by(MagentoConnection.id)
                .limit(1)
            )
        if connection_id is None:
            provision_result = {"status": "skipped", "reason": "no_magento_connection"}
        else:
            provision_result = provision_channel_attributes(
                session,
                channel_code="magento",
                connection_id=int(connection_id),
                canonical_codes=alias_sync["ensured_codes"],
                dry_run=False,
            )
            option_results = []
            for row in rows:
                code = normalize_column_key(row.get("attribute_code") or "")
                if code not in set(alias_sync["ensured_codes"]):
                    continue
                if normalize_column_key(row.get("data_type") or "") not in {
                    "select",
                    "multiselect",
                    "boolean",
                    "yesno",
                }:
                    continue
                labels = [
                    str(item.get("value") or "").strip()
                    for item in (row.get("values") or [])
                    if str(item.get("value") or "").strip()
                ]
                if not labels:
                    continue
                option_results.append(
                    {
                        "attribute_code": code,
                        **_seed_magento_options(
                            session,
                            connection_id=int(connection_id),
                            attribute_code=code,
                            labels=labels,
                            dry_run=False,
                        ),
                    }
                )
            option_sync = {
                "status": "ok",
                "connection_id": int(connection_id),
                "attributes": len(option_results),
                "results": option_results,
            }

            if push_filters:
                from db.collection_landing_push import push_enabled_filters_to_magento

                path_slugs = list(filter_sync.get("updated_path_slugs") or [])
                filter_push = push_enabled_filters_to_magento(
                    session,
                    path_slugs=path_slugs,
                    connection_id=int(connection_id),
                    dry_run=False,
                )
    elif push_filters and not provision_magento and not dry_run:
        from db.collection_landing_push import push_enabled_filters_to_magento
        from db.models import MagentoConnection

        connection_id = magento_connection_id
        if connection_id is None:
            connection_id = session.scalar(
                select(MagentoConnection.id)
                .where(MagentoConnection.status == "active")
                .order_by(MagentoConnection.id)
                .limit(1)
            )
        if connection_id is None:
            filter_push = {"status": "skipped", "reason": "no_magento_connection"}
        else:
            filter_push = push_enabled_filters_to_magento(
                session,
                path_slugs=list(filter_sync.get("updated_path_slugs") or []),
                connection_id=int(connection_id),
                dry_run=False,
            )

    return {
        "published_count": len(values),
        "published_codes": [row["attribute_code"] for row in values],
        "filter_sync": filter_sync,
        "alias_sync": alias_sync,
        "provision": provision_result,
        "option_sync": option_sync,
        "filter_push": filter_push,
    }


def get_locked_vocabulary_for_code(
    session: Session,
    attribute_code: str,
) -> Optional[Dict[str, Any]]:
    """Return locked vocabulary covering this attribute code, if any."""
    code = normalize_column_key(attribute_code)
    if not code:
        return None
    payload = list_manual_attributes(session)
    for item in payload["attributes"]:
        if not item.get("is_locked") or not item.get("is_active"):
            continue
        covered = {normalize_column_key(c) for c in (item.get("covered_codes") or [])}
        covered.add(normalize_column_key(item["attribute_code"]))
        if code in covered:
            return item
    return None


def locked_values_for_code(session: Session, attribute_code: str) -> Optional[List[str]]:
    vocab = get_locked_vocabulary_for_code(session, attribute_code)
    if not vocab:
        return None
    return [str(item["value"]).strip() for item in vocab.get("values") or [] if str(item.get("value") or "").strip()]


def canonicalize_locked_attribute_value(
    session: Session,
    attribute_code: str,
    raw_value: Any,
) -> Tuple[Optional[str], str]:
    """Map a raw value through locked vocabulary.

    Returns (canonical_value_or_None, action) where action is:
      matched | dropped | kept | unlocked
    """
    text = str(raw_value or "").strip()
    vocab = get_locked_vocabulary_for_code(session, attribute_code)
    if not vocab:
        return (text or None, "unlocked")
    if not text:
        return (None, "dropped" if vocab.get("unknown_policy") == "drop" else "kept")

    needle = text.lower()
    for item in vocab.get("values") or []:
        canonical = str(item.get("value") or "").strip()
        aliases = [str(a).strip() for a in (item.get("aliases") or [])]
        candidates = {canonical.lower(), *(a.lower() for a in aliases if a)}
        if needle in candidates:
            return (canonical, "matched")

    if vocab.get("unknown_policy") == "keep":
        return (text, "kept")
    return (None, "dropped")


def apply_manual_attribute_vocabularies_to_payload(
    session: Session,
    payload: Dict[str, Any],
    *,
    locked_vocabs: Optional[List[Dict[str, Any]]] = None,
) -> Dict[str, Any]:
    """Rewrite payload keys covered by locked vocabularies; drop polluted values."""
    if not isinstance(payload, dict) or not payload:
        return payload

    if locked_vocabs is None:
        attributes = list_manual_attributes(session).get("attributes") or []
        locked = [item for item in attributes if item.get("is_locked") and item.get("is_active")]
    else:
        locked = locked_vocabs
    if not locked:
        return payload

    # Build lookup: normalized payload key → original keys
    key_map: Dict[str, List[str]] = {}
    for key in list(payload.keys()):
        normalized = normalize_column_key(key)
        if normalized:
            key_map.setdefault(normalized, []).append(key)

    def _match(vocab: Dict[str, Any], raw_value: str) -> Optional[str]:
        needle = raw_value.lower()
        for item in vocab.get("values") or []:
            canonical = str(item.get("value") or "").strip()
            aliases = [str(a).strip() for a in (item.get("aliases") or [])]
            candidates = {canonical.lower(), *(a.lower() for a in aliases if a)}
            if needle in candidates:
                return canonical
        return None

    for vocab in locked:
        covered = _covered_codes(vocab["attribute_code"], vocab.get("covered_codes"))
        for code in covered:
            original_keys = key_map.get(code) or []
            if not original_keys:
                continue
            current_value = None
            for key in original_keys:
                value = str(payload.get(key) or "").strip()
                if value:
                    current_value = value
                    break
            for key in original_keys:
                payload.pop(key, None)
            if current_value is None:
                continue
            canonical = _match(vocab, current_value)
            if canonical:
                payload[original_keys[0]] = canonical
                continue
            if vocab.get("unknown_policy") == "keep":
                payload[original_keys[0]] = current_value
    return payload