from __future__ import annotations

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

from sqlalchemy import false, func, or_, select
from sqlalchemy.orm import Session

from channel.url_canonical import normalize_plp_path
from db.collection_landing_pages import (
    canonical_category,
    canonical_collection,
    collection_path_slug,
    DEFAULT_CATEGORY_L1,
)
from db.models import (
    CollectionLandingPage,
    CollectionSkuOverride,
    MasterCollectionRegistry,
    MasterProduct,
    MasterProductCollectionMembership,
    ProductListingPathAssignment,
)
from db.tribeca_sku_parse import STYLE_COLLECTION_TO_CODE, split_styled_sku


MATCH_MASTER_COLLECTION = "master_collection"
MATCH_COLLECTION_REGISTRY = "collection_registry"
MATCH_EXPLICIT_MEMBERSHIP = "explicit_membership"
MATCH_SKU_PREFIX = "sku_prefix"
MATCH_MANUAL_INCLUDE = "manual_include"
MATCH_MANUAL_EXCLUDE = "manual_exclude"
MATCH_LISTING_PATH = "listing_path"

ACTION_INCLUDE = "include"
ACTION_EXCLUDE = "exclude"

REJECT_STYLE_MISMATCH = "style_code_mismatch"
MEMBERSHIP_MODE_NATIVE = "native"
MEMBERSHIP_MODE_SHARED = "shared"


def _collection_lookup_key(collection: Optional[str]) -> Optional[str]:
    coll = canonical_collection(collection)
    if not coll:
        return None
    return "".join(ch for ch in coll.lower() if ch.isalnum())


def expected_style_codes_for_collection(
    collection: Optional[str],
    prefixes: Optional[Iterable[str]] = None,
) -> Set[str]:
    codes: Set[str] = {str(p).strip().upper() for p in (prefixes or []) if str(p).strip()}
    code = default_style_code_for_collection(collection)
    if code:
        codes.add(code.upper())
    return codes


def primary_style_code_for_collection(collection: Optional[str]) -> Optional[str]:
    code = default_style_code_for_collection(collection)
    return code.upper() if code else None


def sku_primary_collection_key(sku: str) -> Optional[str]:
    """Normalized collection key implied by the SKU style code, when known."""
    style = master_style_code_for_sku(sku)
    if not style:
        return None
    style = style.upper()
    for key, code in STYLE_COLLECTION_TO_CODE.items():
        if code.upper() == style:
            return key
    return None


def evaluate_sku_collection_membership(
    session: Session,
    product: MasterProduct,
    *,
    path_slug: str,
    collection: Optional[str],
    prefixes: Iterable[str],
    manual_includes: Set[str],
    manual_excludes: Set[str],
    listing_assigned: Set[str],
    collection_registry_id: Optional[int] = None,
    explicit_member_skus: Optional[Set[str]] = None,
) -> Tuple[bool, Optional[str], Optional[str]]:
    """Return (included, match_source, reject_reason)."""
    sku = str(product.sku)
    if sku in manual_excludes:
        return False, None, MATCH_MANUAL_EXCLUDE
    if sku in manual_includes:
        return True, MATCH_MANUAL_INCLUDE, None
    membership_mode = str(getattr(product, "membership_mode", MEMBERSHIP_MODE_NATIVE) or MEMBERSHIP_MODE_NATIVE).strip().lower()

    style = master_style_code_for_sku(sku)
    expected_codes = expected_style_codes_for_collection(collection, prefixes)
    primary_key = sku_primary_collection_key(sku)
    if primary_key and expected_codes:
        if style and style.upper() not in expected_codes:
            return False, None, REJECT_STYLE_MISMATCH

    sources: List[str] = []
    if collection_registry_id and product.collection_registry_id == collection_registry_id:
        sources.append(MATCH_COLLECTION_REGISTRY)
    if membership_mode == MEMBERSHIP_MODE_SHARED and sku in (explicit_member_skus or set()):
        sources.append(MATCH_EXPLICIT_MEMBERSHIP)
    if membership_mode != MEMBERSHIP_MODE_SHARED and _master_collection_match(product, collection, session=session, path_slug=path_slug):
        if not (primary_key and expected_codes and style and style.upper() not in expected_codes):
            sources.append(MATCH_MASTER_COLLECTION)
    if membership_mode != MEMBERSHIP_MODE_SHARED and prefixes and sku_matches_prefixes(sku, prefixes):
        sources.append(MATCH_SKU_PREFIX)
    if sku in listing_assigned:
        if not (primary_key and expected_codes and style and style.upper() not in expected_codes):
            sources.append(MATCH_LISTING_PATH)

    if not sources:
        return False, None, None
    priority = (
        MATCH_MANUAL_INCLUDE,
        MATCH_COLLECTION_REGISTRY,
        MATCH_EXPLICIT_MEMBERSHIP,
        MATCH_SKU_PREFIX,
        MATCH_MASTER_COLLECTION,
        MATCH_LISTING_PATH,
    )
    for source in priority:
        if source in sources:
            return True, source, None
    return True, sources[0], None


def default_style_code_for_collection(collection: Optional[str]) -> Optional[str]:
    """Infer default master SKU style code from collection display name."""
    coll = canonical_collection(collection)
    if not coll:
        return None
    key = "".join(ch for ch in coll.lower() if ch.isalnum())
    return STYLE_COLLECTION_TO_CODE.get(key)


def infer_sku_prefixes_for_collection(collection: Optional[str]) -> List[str]:
    code = default_style_code_for_collection(collection)
    return [code] if code else []


def master_style_code_for_sku(sku: str) -> Optional[str]:
    style, _ = split_styled_sku(str(sku or "").strip())
    return style or None


def sku_matches_prefixes(sku: str, prefixes: Iterable[str]) -> bool:
    style = master_style_code_for_sku(sku)
    if not style:
        return False
    allowed = {str(p).strip().upper() for p in prefixes if str(p).strip()}
    return style in allowed


def list_prefixes_from_json(value: Any) -> List[str]:
    if isinstance(value, dict):
        items = value.get("items")
        if isinstance(items, list):
            return [str(item).strip().upper() for item in items if str(item).strip()]
    if isinstance(value, list):
        return [str(item).strip().upper() for item in value if str(item).strip()]
    if isinstance(value, str) and value.strip():
        return [part.strip().upper() for part in value.replace(",", "\n").splitlines() if part.strip()]
    return []


def prefixes_to_json(prefixes: Iterable[str]) -> Optional[Dict[str, Any]]:
    items = [str(p).strip().upper() for p in prefixes if str(p).strip()]
    return {"items": items} if items else None


def discover_collection_skus(
    session: Session,
    *,
    path_slug: Optional[str] = None,
    category_l1: Optional[str] = None,
    collection: Optional[str] = None,
    sku_prefixes: Optional[Iterable[str]] = None,
    search: Optional[str] = None,
    limit: int = 500,
) -> Dict[str, Any]:
    """Return SKU rows grouped by how they match this collection."""
    ctx = _resolve_context(session, path_slug, category_l1, collection, sku_prefixes)
    category = ctx["category_l1"]
    coll = ctx["collection"]
    slug = ctx["path_slug"]
    prefixes = ctx["sku_prefixes"]

    overrides = _load_overrides(session, slug)
    manual_includes = {sku for sku, action in overrides.items() if action == ACTION_INCLUDE}
    manual_excludes = {sku for sku, action in overrides.items() if action == ACTION_EXCLUDE}

    listing_assigned = _listing_path_skus(session, slug)
    collection_registry_id = ctx.get("collection_registry_id")
    explicit_member_skus = _explicit_member_skus(
        session,
        path_slug=slug,
        collection_registry_id=collection_registry_id,
    )
    master_rows = _matching_master_products_for_collection(
        session,
        category_l1=category,
        collection=coll,
        path_slug=slug,
        sku_prefixes=prefixes,
        listing_skus=listing_assigned,
        manual_includes=manual_includes,
        search=search,
        collection_registry_id=collection_registry_id,
        explicit_member_skus=explicit_member_skus,
    )

    by_sku: Dict[str, Dict[str, Any]] = {}
    counts = {
        MATCH_MASTER_COLLECTION: 0,
        MATCH_COLLECTION_REGISTRY: 0,
        MATCH_EXPLICIT_MEMBERSHIP: 0,
        MATCH_SKU_PREFIX: 0,
        MATCH_MANUAL_INCLUDE: 0,
        MATCH_MANUAL_EXCLUDE: 0,
        MATCH_LISTING_PATH: 0,
    }

    for product in master_rows:
        sku = str(product.sku)
        included, source, reject_reason = evaluate_sku_collection_membership(
            session,
            product,
            path_slug=slug,
            collection=coll,
            prefixes=prefixes,
            manual_includes=manual_includes,
            manual_excludes=manual_excludes,
            listing_assigned=listing_assigned,
            collection_registry_id=collection_registry_id,
            explicit_member_skus=explicit_member_skus,
        )
        if not included:
            continue
        row = by_sku.setdefault(
            sku,
            {
                "master_sku": sku,
                "name": product.name,
                "brand": product.brand,
                "master_collection": product.collection,
                "sku_style_code": master_style_code_for_sku(sku),
                "match_sources": [],
                "included": False,
                "excluded": False,
            },
        )
        row["match_sources"] = [source] if source else []
        row["excluded"] = False
        row["included"] = True

    for sku in sorted(manual_includes - set(by_sku)):
        product = session.scalar(select(MasterProduct).where(MasterProduct.sku == sku).limit(1))
        by_sku[sku] = {
            "master_sku": sku,
            "name": product.name if product else None,
            "brand": product.brand if product else None,
            "master_collection": product.collection if product else None,
            "sku_style_code": master_style_code_for_sku(sku),
            "match_sources": [MATCH_MANUAL_INCLUDE],
            "included": True,
            "excluded": sku in manual_excludes,
        }

    rows = sorted(by_sku.values(), key=lambda item: str(item["master_sku"]))
    for row in rows:
        for source in row["match_sources"]:
            if source in counts:
                counts[source] += 1

    resolved = resolve_collection_skus(
        session,
        path_slug=slug,
        category_l1=category,
        collection=coll,
        sku_prefixes=prefixes,
    )
    included_rows = [row for row in rows if row.get("included") and not row.get("excluded")]
    return {
        "path_slug": slug,
        "category_l1": category,
        "collection": coll,
        "sku_prefixes": prefixes,
        "counts": counts,
        "resolved_sku_count": len(resolved["skus"]),
        "resolved_skus": resolved["skus"][:limit],
        "rows": included_rows[:limit],
        "all_rows": rows[:limit],
    }


def resolve_collection_skus(
    session: Session,
    *,
    path_slug: Optional[str] = None,
    category_l1: Optional[str] = None,
    collection: Optional[str] = None,
    sku_prefixes: Optional[Iterable[str]] = None,
) -> Dict[str, Any]:
    """Merged SKU set used for apply/sync."""
    ctx = _resolve_context(session, path_slug, category_l1, collection, sku_prefixes)
    category = ctx["category_l1"]
    coll = ctx["collection"]
    slug = ctx["path_slug"]
    prefixes = ctx["sku_prefixes"]

    overrides = _load_overrides(session, slug)
    manual_includes = {sku for sku, action in overrides.items() if action == ACTION_INCLUDE}
    manual_excludes = {sku for sku, action in overrides.items() if action == ACTION_EXCLUDE}

    matched: Dict[str, str] = {}
    listing_assigned = _listing_path_skus(session, slug)
    collection_registry_id = ctx.get("collection_registry_id")
    explicit_member_skus = _explicit_member_skus(
        session,
        path_slug=slug,
        collection_registry_id=collection_registry_id,
    )
    for product in _matching_master_products_for_collection(
        session,
        category_l1=category,
        collection=coll,
        path_slug=slug,
        sku_prefixes=prefixes,
        listing_skus=listing_assigned,
        manual_includes=manual_includes,
        collection_registry_id=collection_registry_id,
        explicit_member_skus=explicit_member_skus,
    ):
        sku = str(product.sku)
        included, source, _reject = evaluate_sku_collection_membership(
            session,
            product,
            path_slug=slug,
            collection=coll,
            prefixes=prefixes,
            manual_includes=manual_includes,
            manual_excludes=manual_excludes,
            listing_assigned=listing_assigned,
            collection_registry_id=collection_registry_id,
            explicit_member_skus=explicit_member_skus,
        )
        if not included or not source:
            continue
        matched[sku] = source

    for sku in manual_includes:
        if sku not in manual_excludes:
            matched[sku] = MATCH_MANUAL_INCLUDE

    skus = sorted(matched.items())
    return {
        "path_slug": slug,
        "category_l1": category,
        "collection": coll,
        "sku_prefixes": prefixes,
        "skus": [{"master_sku": sku, "match_source": source} for sku, source in skus],
        "sku_count": len(skus),
    }


def upsert_collection_sku_override(
    session: Session,
    *,
    path_slug: str,
    master_sku: str,
    action: str,
    notes: Optional[str] = None,
) -> Dict[str, Any]:
    slug = normalize_plp_path(path_slug)
    sku = str(master_sku or "").strip()
    act = str(action or "").strip().lower()
    if not slug:
        raise ValueError("path_slug is required")
    if not sku:
        raise ValueError("master_sku is required")
    if act not in {ACTION_INCLUDE, ACTION_EXCLUDE}:
        raise ValueError("action must be include or exclude")

    row = session.scalar(
        select(CollectionSkuOverride)
        .where(CollectionSkuOverride.path_slug == slug)
        .where(CollectionSkuOverride.master_sku == sku)
        .limit(1)
    )
    if row is None:
        row = CollectionSkuOverride(path_slug=slug, master_sku=sku, action=act, notes=notes)
        session.add(row)
    else:
        row.action = act
        row.notes = notes
    session.flush()
    return _override_dict(row)


def remove_collection_sku_override(session: Session, *, path_slug: str, master_sku: str) -> bool:
    slug = normalize_plp_path(path_slug)
    sku = str(master_sku or "").strip()
    row = session.scalar(
        select(CollectionSkuOverride)
        .where(CollectionSkuOverride.path_slug == slug)
        .where(CollectionSkuOverride.master_sku == sku)
        .limit(1)
    )
    if row is None:
        return False
    session.delete(row)
    session.flush()
    return True


def list_collection_sku_overrides(session: Session, path_slug: str) -> List[Dict[str, Any]]:
    slug = normalize_plp_path(path_slug)
    rows = session.scalars(
        select(CollectionSkuOverride)
        .where(CollectionSkuOverride.path_slug == slug)
        .order_by(CollectionSkuOverride.master_sku)
    ).all()
    return [_override_dict(row) for row in rows]


def _resolve_context(
    session: Session,
    path_slug: Optional[str],
    category_l1: Optional[str],
    collection: Optional[str],
    sku_prefixes: Optional[Iterable[str]],
) -> Dict[str, Any]:
    from db.master_taxonomy_merge import resolve_canonical_path_slug

    slug = normalize_plp_path(path_slug) if path_slug else None
    category = canonical_category(category_l1)
    coll = canonical_collection(collection)
    prefixes: List[str] = []

    if slug:
        slug = resolve_canonical_path_slug(session, slug) or slug
        row = session.scalar(select(CollectionLandingPage).where(CollectionLandingPage.path_slug == slug).limit(1))
        if row is None:
            row = session.scalar(
                select(CollectionLandingPage)
                .where(CollectionLandingPage.path_slug == normalize_plp_path(path_slug or ""))
                .limit(1)
            )
        if row:
            category = category or canonical_category(row.category_l1)
            coll = coll or canonical_collection(row.collection)
            prefixes = list_prefixes_from_json(row.sku_prefixes)

    if not coll and slug:
        _, tail = _split_slug_tail(slug)
        coll = canonical_collection(tail)

    category = category or DEFAULT_CATEGORY_L1
    if not slug and coll:
        slug = collection_path_slug(category, coll)
    elif not slug:
        raise ValueError("path_slug or collection is required")

    if sku_prefixes is not None:
        prefixes = [str(p).strip().upper() for p in sku_prefixes if str(p).strip()]
    elif not prefixes:
        prefixes = infer_sku_prefixes_for_collection(coll)

    collection_registry_id = _resolve_collection_registry_id(session, slug=slug, collection=coll)

    return {
        "path_slug": slug,
        "category_l1": category,
        "collection": coll,
        "sku_prefixes": prefixes,
        "collection_registry_id": collection_registry_id,
    }


def _master_products_for_scope(
    session: Session,
    *,
    category_l1: Optional[str],
    search: Optional[str] = None,
    limit: Optional[int] = None,
) -> List[MasterProduct]:
    stmt = _scoped_master_products_stmt(category_l1=category_l1, search=search)
    if limit:
        stmt = stmt.limit(limit)
    return list(session.scalars(stmt).all())


def _scoped_master_products_stmt(
    *,
    category_l1: Optional[str],
    search: Optional[str] = None,
):
    stmt = select(MasterProduct).where(MasterProduct.is_active.is_(True))
    if category_l1:
        stmt = stmt.where(MasterProduct.category_l1.ilike(canonical_category(category_l1) or category_l1))
    if search:
        pattern = f"%{search.strip()}%"
        stmt = stmt.where(
            or_(
                MasterProduct.sku.ilike(pattern),
                MasterProduct.name.ilike(pattern),
                MasterProduct.collection.ilike(pattern),
            )
        )
    return stmt.order_by(MasterProduct.sku)


def _matching_master_products_for_collection(
    session: Session,
    *,
    category_l1: Optional[str],
    collection: Optional[str],
    path_slug: Optional[str] = None,
    sku_prefixes: Optional[Iterable[str]],
    listing_skus: Optional[Set[str]] = None,
    manual_includes: Optional[Set[str]] = None,
    search: Optional[str] = None,
    collection_registry_id: Optional[int] = None,
    explicit_member_skus: Optional[Set[str]] = None,
) -> List[MasterProduct]:
    """Return active master products that match this collection in SQL (no prefetch limit)."""
    stmt = _matching_master_products_stmt(
        session,
        category_l1=category_l1,
        collection=collection,
        path_slug=path_slug,
        sku_prefixes=sku_prefixes,
        listing_skus=listing_skus,
        manual_includes=manual_includes,
        search=search,
        collection_registry_id=collection_registry_id,
        explicit_member_skus=explicit_member_skus,
    )
    return list(session.scalars(stmt).all())


def _matching_master_products_stmt(
    session: Session,
    *,
    category_l1: Optional[str],
    collection: Optional[str],
    path_slug: Optional[str] = None,
    sku_prefixes: Optional[Iterable[str]],
    listing_skus: Optional[Set[str]] = None,
    manual_includes: Optional[Set[str]] = None,
    search: Optional[str] = None,
    collection_registry_id: Optional[int] = None,
    explicit_member_skus: Optional[Set[str]] = None,
):
    stmt = _scoped_master_products_stmt(category_l1=category_l1, search=search)
    match_parts = []
    if collection_registry_id:
        match_parts.append(MasterProduct.collection_registry_id == int(collection_registry_id))
    explicit_clause = _sql_explicit_membership_match(
        path_slug=path_slug,
        collection_registry_id=collection_registry_id,
        explicit_member_skus=explicit_member_skus,
    )
    if explicit_clause is not None:
        match_parts.append(explicit_clause)
    collection_clause = _sql_master_collection_match(session, path_slug, collection)
    if collection_clause is not None:
        match_parts.append(collection_clause)
    prefix_clause = _sql_sku_prefix_match(sku_prefixes)
    if prefix_clause is not None:
        match_parts.append(prefix_clause)

    listing = {str(sku).strip() for sku in (listing_skus or set()) if str(sku).strip()}
    manual = {str(sku).strip() for sku in (manual_includes or set()) if str(sku).strip()}
    extra_skus = listing | manual
    if extra_skus:
        match_parts.append(MasterProduct.sku.in_(sorted(extra_skus)))

    if not match_parts:
        return stmt.where(false())

    return stmt.where(or_(*match_parts))


def _sql_normalized_collection_expr():
    return func.replace(func.lower(func.coalesce(MasterProduct.collection, "")), "grey", "gray")


def _sql_master_collection_match(
    session: Session,
    path_slug: Optional[str],
    collection: Optional[str],
):
    from db.master_taxonomy_merge import collection_names_for_path_slug

    names: List[str] = []
    if path_slug:
        names = collection_names_for_path_slug(session, path_slug, fallback_collection=collection)
    elif collection:
        coll = canonical_collection(collection)
        if coll:
            names = [coll.lower().replace("grey", "gray")]
    if not names:
        return None
    expr = _sql_normalized_collection_expr()
    return or_(*(expr == name for name in names))


def _sql_sku_prefix_match(sku_prefixes: Optional[Iterable[str]]):
    prefixes = [str(prefix).strip().upper() for prefix in (sku_prefixes or []) if str(prefix).strip()]
    if not prefixes:
        return None
    return or_(*(func.upper(MasterProduct.sku).like(f"{prefix}-%") for prefix in prefixes))


def _sql_explicit_membership_match(
    *,
    path_slug: Optional[str],
    collection_registry_id: Optional[int],
    explicit_member_skus: Optional[Set[str]] = None,
):
    member_skus = {str(sku).strip() for sku in (explicit_member_skus or set()) if str(sku).strip()}
    if member_skus:
        return (
            MasterProduct.sku.in_(sorted(member_skus))
            & (func.lower(func.coalesce(MasterProduct.membership_mode, MEMBERSHIP_MODE_NATIVE)) == MEMBERSHIP_MODE_SHARED)
        )

    match_parts = []
    slug = normalize_plp_path(path_slug or "")
    if slug:
        match_parts.append(MasterProductCollectionMembership.path_slug == slug)
    if collection_registry_id:
        match_parts.append(MasterProductCollectionMembership.collection_registry_id == int(collection_registry_id))
    if not match_parts:
        return None

    membership_skus = (
        select(MasterProductCollectionMembership.master_sku)
        .where(MasterProductCollectionMembership.is_active.is_(True))
        .where(or_(*match_parts))
    )
    return (
        MasterProduct.sku.in_(membership_skus)
        & (func.lower(func.coalesce(MasterProduct.membership_mode, MEMBERSHIP_MODE_NATIVE)) == MEMBERSHIP_MODE_SHARED)
    )


def _master_collection_match(
    product: MasterProduct,
    collection: Optional[str],
    *,
    session: Optional[Session] = None,
    path_slug: Optional[str] = None,
) -> bool:
    if session is not None and path_slug:
        from db.master_taxonomy_merge import collection_names_for_path_slug

        names = collection_names_for_path_slug(session, path_slug, fallback_collection=collection)
        product_coll = canonical_collection(product.collection)
        if product_coll:
            return product_coll.lower().replace("grey", "gray") in names
    coll = canonical_collection(collection)
    if not coll:
        return False
    product_coll = canonical_collection(product.collection)
    return bool(product_coll and product_coll.lower() == coll.lower())


def _load_overrides(session: Session, path_slug: str) -> Dict[str, str]:
    rows = session.scalars(
        select(CollectionSkuOverride).where(CollectionSkuOverride.path_slug == path_slug)
    ).all()
    return {str(row.master_sku): str(row.action) for row in rows}


def _listing_path_skus(session: Session, path_slug: str) -> Set[str]:
    from db.models import ChannelListingPath

    path = session.scalar(
        select(ChannelListingPath).where(ChannelListingPath.path_slug == path_slug).limit(1)
    )
    if path is None:
        return set()
    rows = session.scalars(
        select(ProductListingPathAssignment.master_sku)
        .where(ProductListingPathAssignment.listing_path_id == path.id)
        .where(ProductListingPathAssignment.assignment_status == "active")
    ).all()
    return {str(sku) for sku in rows}


def _explicit_member_skus(
    session: Session,
    *,
    path_slug: str,
    collection_registry_id: Optional[int],
) -> Set[str]:
    clauses = [MasterProductCollectionMembership.is_active.is_(True)]
    match_parts = []
    slug = normalize_plp_path(path_slug)
    if slug:
        match_parts.append(MasterProductCollectionMembership.path_slug == slug)
    if collection_registry_id:
        match_parts.append(MasterProductCollectionMembership.collection_registry_id == int(collection_registry_id))
    if not match_parts:
        return set()
    clauses.append(or_(*match_parts))
    rows = session.scalars(
        select(MasterProductCollectionMembership.master_sku).where(*clauses)
    ).all()
    return {str(sku).strip() for sku in rows if str(sku).strip()}


def _sku_is_included(sources: List[str]) -> bool:
    if MATCH_MANUAL_EXCLUDE in sources:
        return False
    return any(
        source in sources
        for source in (
            MATCH_COLLECTION_REGISTRY,
            MATCH_EXPLICIT_MEMBERSHIP,
            MATCH_MASTER_COLLECTION,
            MATCH_SKU_PREFIX,
            MATCH_MANUAL_INCLUDE,
            MATCH_LISTING_PATH,
        )
    )


def _resolve_collection_registry_id(
    session: Session,
    *,
    slug: Optional[str],
    collection: Optional[str],
) -> Optional[int]:
    if slug:
        row = session.scalar(
            select(MasterCollectionRegistry.id)
            .where(MasterCollectionRegistry.path_slug == slug)
            .where(MasterCollectionRegistry.is_active.is_(True))
            .limit(1)
        )
        if row:
            return int(row)
        landing = session.scalar(
            select(CollectionLandingPage.collection_registry_id)
            .where(CollectionLandingPage.path_slug == slug)
            .limit(1)
        )
        if landing:
            return int(landing)
    coll = canonical_collection(collection)
    if coll:
        row = session.scalar(
            select(MasterCollectionRegistry.id)
            .where(MasterCollectionRegistry.name.ilike(coll))
            .where(MasterCollectionRegistry.is_active.is_(True))
            .limit(1)
        )
        if row:
            return int(row)
    return None


def _split_slug_tail(slug: str) -> Tuple[str, Optional[str]]:
    parts = [part for part in normalize_plp_path(slug).split("/") if part]
    if len(parts) < 2:
        return (parts[0] if parts else DEFAULT_CATEGORY_L1), None
    return parts[0], parts[-1]


def deactivate_sku_from_other_collection_listing_paths(
    session: Session,
    *,
    master_sku: str,
    keep_path_slug: str,
) -> int:
    """Remove a SKU from collection listing paths other than the canonical keep path."""
    from db.master_taxonomy_merge import resolve_canonical_path_slug
    from db.models import ChannelListingPath

    sku = str(master_sku or "").strip()
    keep_slug = resolve_canonical_path_slug(session, keep_path_slug) or normalize_plp_path(keep_path_slug)
    if not sku or not keep_slug:
        return 0
    allowed_slugs = _allowed_collection_path_slugs_for_sku(session, master_sku=sku)
    allowed_slugs.add(keep_slug)

    keep_path = session.scalar(
        select(ChannelListingPath).where(ChannelListingPath.path_slug == keep_slug).limit(1)
    )
    keep_id = keep_path.id if keep_path is not None else None
    deactivated = 0
    for assignment, path_slug in session.execute(
        select(ProductListingPathAssignment, ChannelListingPath.path_slug)
        .join(ChannelListingPath, ChannelListingPath.id == ProductListingPathAssignment.listing_path_id)
        .where(ProductListingPathAssignment.master_sku == sku)
        .where(ProductListingPathAssignment.assignment_status == "active")
        .where(ChannelListingPath.path_kind == "collection")
    ).all():
        if keep_id is not None and assignment.listing_path_id == keep_id:
            continue
        normalized_path_slug = normalize_plp_path(str(path_slug or ""))
        if normalized_path_slug and normalized_path_slug in allowed_slugs:
            continue
        assignment.assignment_status = "inactive"
        deactivated += 1
    if deactivated:
        session.flush()
    return deactivated


def _allowed_collection_path_slugs_for_sku(session: Session, *, master_sku: str) -> Set[str]:
    sku = str(master_sku or "").strip()
    if not sku:
        return set()

    allowed: Set[str] = set()
    product = session.scalar(select(MasterProduct).where(MasterProduct.sku == sku).limit(1))
    if product is not None:
        primary_slug = normalize_plp_path(
            collection_path_slug(
                canonical_category(product.category_l1) or DEFAULT_CATEGORY_L1,
                canonical_collection(product.collection),
            )
        )
        if primary_slug:
            allowed.add(primary_slug)

    for slug in session.scalars(
        select(MasterProductCollectionMembership.path_slug)
        .where(MasterProductCollectionMembership.master_sku == sku)
        .where(MasterProductCollectionMembership.is_active.is_(True))
    ).all():
        normalized = normalize_plp_path(str(slug or ""))
        if normalized:
            allowed.add(normalized)
    return allowed


def reconcile_collection_sku_assignments(
    session: Session,
    *,
    path_slug: str,
    category_l1: Optional[str] = None,
    collection: Optional[str] = None,
    dry_run: bool = True,
    cleanup_other_paths: bool = True,
) -> Dict[str, Any]:
    """Apply authoritative SKU membership for one collection and optionally clean cross-path pollution."""
    from db.collection_landing_pages import apply_collection_to_skus

    apply_result = apply_collection_to_skus(
        session,
        path_slug=path_slug,
        category_l1=category_l1,
        collection=collection,
        dry_run=dry_run,
    )
    slug = apply_result.get("path_slug") or path_slug
    resolved = resolve_collection_skus(
        session,
        path_slug=slug,
        category_l1=apply_result.get("category_l1"),
        collection=apply_result.get("collection"),
    )
    keep_skus = [item["master_sku"] for item in (resolved.get("skus") or [])]
    removed_elsewhere = 0
    if cleanup_other_paths and not dry_run:
        for sku in keep_skus:
            removed_elsewhere += deactivate_sku_from_other_collection_listing_paths(
                session,
                master_sku=sku,
                keep_path_slug=slug,
            )
    return {
        **apply_result,
        "cleanup_other_paths": cleanup_other_paths,
        "removed_from_other_collection_paths": removed_elsewhere,
        "expected_style_code": primary_style_code_for_collection(apply_result.get("collection")),
    }


def reconcile_all_collection_assignments(
    session: Session,
    *,
    category_l1: Optional[str] = None,
    dry_run: bool = True,
) -> Dict[str, Any]:
    """Reconcile SKU listing assignments for every taxonomy-backed collection."""
    from db.collection_landing_pages import list_collection_landing_pages

    rows = list_collection_landing_pages(
        session,
        category_l1=category_l1,
        include_inferred=True,
        limit=1000,
    )
    results: List[Dict[str, Any]] = []
    totals = {"collections": 0, "sku_count": 0, "removed_from_other_collection_paths": 0}
    for row in rows:
        if row.get("path_slug") in {None, ""}:
            continue
        item = reconcile_collection_sku_assignments(
            session,
            path_slug=str(row["path_slug"]),
            category_l1=row.get("category_l1"),
            collection=row.get("collection"),
            dry_run=dry_run,
            cleanup_other_paths=True,
        )
        results.append(
            {
                "path_slug": item.get("path_slug"),
                "collection": item.get("collection"),
                "sku_count": item.get("sku_count") or 0,
                "removed_from_other_collection_paths": item.get("removed_from_other_collection_paths") or 0,
            }
        )
        totals["collections"] += 1
        totals["sku_count"] += int(item.get("sku_count") or 0)
        totals["removed_from_other_collection_paths"] += int(item.get("removed_from_other_collection_paths") or 0)
    return {"dry_run": dry_run, "totals": totals, "collections": results[:100]}


def _magento_product_sync_from_channel_result(magento_result: Dict[str, Any]) -> Dict[str, Any]:
    for item in magento_result.get("results") or []:
        product_sync = item.get("product_sync")
        if isinstance(product_sync, dict):
            return product_sync
    product_sync = magento_result.get("product_sync")
    return product_sync if isinstance(product_sync, dict) else {}


def _magento_channel_sync_complete(magento_result: Dict[str, Any], *, expected_total: int) -> bool:
    for err in magento_result.get("errors") or []:
        if isinstance(err, dict) and err.get("aborted"):
            return False
    product_sync = _magento_product_sync_from_channel_result(magento_result)
    processed = int(product_sync.get("processed") or 0)
    return processed >= expected_total


def sync_collection_reconcile_channels(
    session: Session,
    *,
    path_slug: str,
    master_skus: List[str],
    magento_connection_id: Optional[int] = None,
    shopify_connection_id: Optional[int] = None,
    sync_magento: bool = False,
    sync_shopify: bool = False,
    on_progress: Optional[Any] = None,
) -> Dict[str, Any]:
    """Push reconciled SKU membership to Magento listing paths and Shopify collections."""
    channel_sync: Dict[str, Any] = {}
    total = len(master_skus)
    if sync_magento and magento_connection_id:
        from db.collection_landing_push import sync_magento_collection_products_channel

        if on_progress:
            on_progress(
                {
                    "stage": "magento_sync",
                    "completed": 0,
                    "total": total,
                    "notes": f"Syncing {total} SKU(s) to Magento category",
                }
            )
        channel_sync["magento"] = sync_magento_collection_products_channel(
            session,
            connection_id=int(magento_connection_id),
            path_slug=path_slug,
            apply_listing_paths=False,
            dry_run=False,
            on_progress=on_progress,
            master_skus=master_skus,
        )
        magento_result = channel_sync["magento"]
        product_sync = _magento_product_sync_from_channel_result(magento_result)
        assigned = int(product_sync.get("assigned") or 0)
        failed = int(product_sync.get("failed") or 0)
        if on_progress:
            on_progress(
                {
                    "stage": "magento_sync",
                    "completed": total,
                    "total": total,
                    "pushed": assigned,
                    "failed": failed,
                    "notes": f"Magento sync complete ({assigned} assigned, {failed} failed)",
                }
            )
        if sync_shopify and shopify_connection_id and not _magento_channel_sync_complete(
            magento_result, expected_total=total
        ):
            channel_sync["shopify"] = {
                "skipped": True,
                "reason": "magento_sync_incomplete",
                "magento_errors": (magento_result.get("errors") or [])[:5],
            }
            return channel_sync
    if sync_shopify and shopify_connection_id:
        from db.channel_sku_mapping import mapping_by_master
        from db.models import ShopifyConnection
        from shopify.collection_sync import assign_product_to_collections
        from shopify.connections import build_client
        from shopify.product_sync import _lookup_product_id

        shop_row = session.get(ShopifyConnection, int(shopify_connection_id))
        if shop_row is None:
            raise ValueError("Shopify connection not found")
        client = build_client(shop_row)
        channel_skus = mapping_by_master(
            session,
            "shopify",
            master_skus,
            connection_id=int(shopify_connection_id),
        )
        assigned = 0
        errors: List[str] = []
        if on_progress:
            on_progress(
                {
                    "stage": "shopify_sync",
                    "completed": 0,
                    "total": total,
                    "notes": f"Syncing {total} SKU(s) to Shopify collection",
                }
            )
        for index, master_sku in enumerate(master_skus, start=1):
            channel_sku = channel_skus.get(master_sku, master_sku)
            product_id = _lookup_product_id(client, channel_sku)
            if not product_id:
                continue
            try:
                assign_product_to_collections(
                    session,
                    client,
                    product_id=str(product_id),
                    master_sku=master_sku,
                    fields={"sku": master_sku},
                    connection_id=int(shopify_connection_id),
                )
                assigned += 1
            except Exception as exc:
                errors.append(f"{master_sku}: {exc}")
            if on_progress and (index == 1 or index % 25 == 0 or index == total):
                on_progress(
                    {
                        "stage": "shopify_sync",
                        "completed": index,
                        "total": total,
                        "pushed": assigned,
                        "failed": len(errors),
                        "notes": f"Shopify sync {index}/{total} ({assigned} assigned)",
                    }
                )
        channel_sync["shopify"] = {
            "assigned": assigned,
            "resolved": len(master_skus),
            "errors": errors[:25],
        }
    return channel_sync


def _override_dict(row: CollectionSkuOverride) -> Dict[str, Any]:
    return {
        "id": row.id,
        "path_slug": row.path_slug,
        "master_sku": row.master_sku,
        "action": row.action,
        "notes": row.notes,
        "updated_at": row.updated_at.isoformat() if row.updated_at else None,
    }