"""Detect and block suffix-only collection paths (e.g. Stone Gray vs Anna Stone Gray)."""

from __future__ import annotations

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

from sqlalchemy import select
from sqlalchemy.orm import Session

from channel.url_canonical import normalize_plp_path, path_segments
from db.collection_landing_pages import canonical_collection, canonical_category, DEFAULT_CATEGORY_L1
from db.models import (
    ChannelListingPathTarget,
    MagentoCategoryRegistry,
    MasterCollectionRegistry,
    MasterProduct,
    MasterTaxonomyChannelLink,
    MasterTaxonomyNode,
    ShopifyCollectionRegistry,
)

MAGENTO_ROOT_PREFIXES = ("default category", "root catalog", "default")

# Magento structural nodes — not collection landing pages.
KITCHEN_NON_COLLECTION_NAMES = frozenset(
    {
        "Kitchen Cabinets",
        "Port & Bell",
        "CastleCraft",
        "Fabuwood",
        "NorthPoint Cabinetry",
        "Legion Manor",
        "Legacy",
        "SunnyWood",
        "HomePlus",
        "Base Cabinets",
        "Wall Cabinets",
        "Tall Cabinets",
        "Accessories & Trim",
        "Accessories",
        "Storage Accessories",
        "Cabinet Accessories",
        "Vanity",
        "Base",
        "Wall",
        "Panels and Fillers",
        "Mouldings",
        "Panels And Fillers",
        "All Products",
        "Default Category",
        "Bathroom Vanities",
        "Doors",
        "Interior Doors",
        "Exterior Doors",
        "Windows",
        "Showers",
        "Locks Hardware",
        "Header Menu",
    }
)

COLLECTION_DECORATOR_NAMES = frozenset(
    {
        "Accessories",
        "Accessories & Trim",
        "Cabinet Accessories",
        "Storage Accessories",
        "Port & Bell",
        "CastleCraft",
        "Fabuwood",
        "NorthPoint Cabinetry",
        "Legion Manor",
        "Legacy",
        "SunnyWood",
        "HomePlus",
    }
)

STRUCTURAL_COLLECTION_LABELS = frozenset(
    {
        "Port & Bell",
        "CastleCraft",
        "Fabuwood",
        "NorthPoint Cabinetry",
        "Legion Manor",
        "Legacy",
        "SunnyWood",
        "HomePlus",
    }
)

# Top-level Magento orphans outside the kitchen hub we should never treat as collections.
NON_KITCHEN_ROOT_PREFIXES = frozenset(
    {
        "all products",
        "bathroom vanities",
        "doors",
        "windows",
        "showers",
        "locks hardware",
        "header menu",
    }
)


def _norm_key(value: Optional[str]) -> str:
    return re.sub(r"[^a-z0-9]+", "", str(value or "").lower().replace("grey", "gray"))


def known_collection_names(
    session: Session,
    *,
    category_l1: Optional[str] = None,
) -> Set[str]:
    """Canonical collection display names from products, registries, and active taxonomy nodes.

    Active taxonomy collection nodes are included so preferred merged paths
    (e.g. Plymouth Espresso Maple) suppress suffix-only proposals
    (e.g. Plymouth Espresso) even when some SKUs still carry the short name.
    """
    from db.master_taxonomy_hierarchy import NODE_COLLECTION

    names: Set[str] = set()
    category = canonical_category(category_l1) if category_l1 else None

    product_stmt = (
        select(MasterProduct.collection)
        .where(MasterProduct.is_active.is_(True))
        .where(MasterProduct.collection.is_not(None))
        .where(MasterProduct.collection != "")
        .distinct()
    )
    if category:
        product_stmt = product_stmt.where(MasterProduct.category_l1.ilike(category))
    for value in session.scalars(product_stmt).all():
        coll = canonical_collection(value)
        if coll:
            names.add(coll)

    registry_stmt = select(MasterCollectionRegistry.name).where(MasterCollectionRegistry.is_active.is_(True))
    if category:
        hub_slug = normalize_plp_path(category)
        if hub_slug:
            registry_stmt = registry_stmt.where(MasterCollectionRegistry.path_slug.like(f"{hub_slug}/%"))
    for value in session.scalars(registry_stmt).all():
        coll = canonical_collection(value)
        if coll:
            names.add(coll)

    taxonomy_stmt = (
        select(MasterTaxonomyNode.name, MasterTaxonomyNode.path_slug)
        .where(MasterTaxonomyNode.is_active.is_(True))
        .where(MasterTaxonomyNode.node_kind == NODE_COLLECTION)
    )
    if category:
        hub_slug = normalize_plp_path(category)
        if hub_slug:
            taxonomy_stmt = taxonomy_stmt.where(MasterTaxonomyNode.path_slug.like(f"{hub_slug}/%"))
    for name, path_slug in session.execute(taxonomy_stmt).all():
        leaf = str(name or "").strip()
        if " / " in leaf:
            leaf = leaf.split(" / ")[-1].strip()
        if not leaf and path_slug:
            leaf = str(path_slug).rsplit("/", 1)[-1].replace("-", " ").strip()
        coll = canonical_collection(leaf)
        if coll:
            names.add(coll)

    return names


def _is_strict_suffix_name(short_name: str, long_name: str) -> bool:
    short_key = _norm_key(short_name)
    long_key = _norm_key(long_name)
    if not short_key or not long_key or short_key == long_key:
        return False
    return long_key.endswith(short_key) and len(long_name) > len(short_name)


def _is_strict_edge_extension_name(short_name: str, long_name: str) -> bool:
    short = canonical_collection(short_name)
    long = canonical_collection(long_name)
    short_key = _norm_key(short)
    long_key = _norm_key(long)
    if not short_key or not long_key or short_key == long_key:
        return False
    if _is_strict_suffix_name(short, long):
        return True
    return long.lower().startswith(f"{short.lower()} ") and len(long) > len(short)


def _canonical_names_without_suffix_aliases(candidates: Set[str]) -> Set[str]:
    canonical: Set[str] = set()
    for name in candidates:
        coll = canonical_collection(name)
        if not coll:
            continue
        if any(_is_strict_suffix_name(coll, other) for other in candidates if other != coll):
            continue
        canonical.add(coll)
    return canonical


def _magento_registry_collection_candidates(
    rows: List[MagentoCategoryRegistry],
    *,
    hub_title: str,
) -> Set[str]:
    """Leaf category names under the kitchen hub tree (any depth) plus level-2 orphans."""
    hub_key = _norm_key(hub_title)
    candidates: Set[str] = set()
    for row in rows:
        name = canonical_collection(row.name)
        if not name or name in KITCHEN_NON_COLLECTION_NAMES:
            continue
        stripped = _strip_magento_roots(str(row.path_names or row.name or ""))
        parts = [part.strip() for part in stripped.split("/") if part.strip()]
        if not parts:
            continue
        if parts[0].lower() == hub_title.lower() or _norm_key(parts[0]) == hub_key:
            if len(parts) >= 2:
                candidates.add(name)
            continue
        if len(parts) == 1 and int(row.level or 0) == 2:
            if _norm_key(parts[0]) in {_norm_key(item) for item in NON_KITCHEN_ROOT_PREFIXES}:
                continue
            candidates.add(name)
    return candidates


def known_collection_names_for_hub(
    session: Session,
    *,
    category_l1: Optional[str] = None,
    magento_connection_id: Optional[int] = None,
) -> Set[str]:
    """All collection-like names for a hub (master data + Magento registry leaves)."""
    names = known_collection_names(session, category_l1=category_l1)
    if magento_connection_id is None:
        return names
    rows = list(
        session.scalars(
            select(MagentoCategoryRegistry)
            .where(MagentoCategoryRegistry.connection_id == int(magento_connection_id))
            .where(MagentoCategoryRegistry.is_active.is_(True))
        ).all()
    )
    hub_title = canonical_category(category_l1) or DEFAULT_CATEGORY_L1
    names |= _magento_registry_collection_candidates(rows, hub_title=hub_title)
    return names


def canonical_collection_option_labels(
    session: Session,
    *,
    category_l1: Optional[str] = None,
    limit: int = 5000,
) -> List[str]:
    """Valid collection labels for Magento select option seeding (collection_style / family).

    Uses identified collection landing pages + active registry names (after taxonomy
    merge/collapse), not raw ``master_product.collection`` strings such as
    ``Snow White``, ``… - Accessories``, or ``Vanity Tops``.
    """
    from db.collection_landing_pages import list_collection_landing_pages

    hub = canonical_category(category_l1) if category_l1 else None
    candidates: Set[str] = set()

    for page in list_collection_landing_pages(
        session,
        category_l1=hub,
        include_inferred=True,
        limit=max(1, min(int(limit or 5000), 20000)),
    ):
        if page.get("is_active") is False:
            continue
        coll = canonical_collection(page.get("collection") or page.get("title"))
        if coll:
            candidates.add(coll)

    registry_stmt = select(MasterCollectionRegistry.name).where(MasterCollectionRegistry.is_active.is_(True))
    if hub:
        hub_slug = normalize_plp_path(hub)
        if hub_slug:
            registry_stmt = registry_stmt.where(MasterCollectionRegistry.path_slug.like(f"{hub_slug}/%"))
    for value in session.scalars(registry_stmt).all():
        coll = canonical_collection(value)
        if coll:
            candidates.add(coll)

    cleaned = _canonical_names_without_suffix_aliases(candidates)
    labels = [
        name
        for name in sorted(cleaned)
        if not is_polluted_collection_name(
            session,
            name,
            category_l1=hub,
            known_names=cleaned,
        )
    ]
    return labels[: max(1, min(int(limit or 5000), 20000))]


def is_suffix_only_polluted_collection(
    session: Session,
    collection_name: Optional[str],
    *,
    category_l1: Optional[str] = None,
    known_names: Optional[Set[str]] = None,
) -> bool:
    """True when *collection_name* is only the color/finish tail of a real collection."""
    name = canonical_collection(collection_name)
    if not name:
        return False
    catalog = known_names if known_names is not None else known_collection_names(session, category_l1=category_l1)
    if not catalog:
        return False
    return any(_is_strict_edge_extension_name(name, other) for other in catalog if other != name)


def _is_structural_collection_name(collection_name: Optional[str]) -> bool:
    name = canonical_collection(collection_name)
    if not name:
        return False
    return _norm_key(name) in {_norm_key(item) for item in STRUCTURAL_COLLECTION_LABELS}


def _decorator_stripped_names(collection_name: str) -> Set[str]:
    """Remove brand/accessory tails that should not create separate collection nodes."""
    name = canonical_collection(collection_name)
    if not name:
        return set()
    out: Set[str] = set()
    name_key = _norm_key(name)
    for decorator in COLLECTION_DECORATOR_NAMES:
        decorator_key = _norm_key(decorator)
        if not decorator_key or decorator_key == name_key:
            continue
        for separator in (" ", " - ", " / ", "/"):
            suffix = f"{separator}{decorator}"
            if name.lower().endswith(suffix.lower()):
                base = canonical_collection(name[: -len(suffix)])
                if base:
                    out.add(base)
        prefix = f"{decorator} "
        if name.lower().startswith(prefix.lower()):
            base = canonical_collection(name[len(prefix):])
            if base:
                out.add(base)
    return out


def is_polluted_collection_name(
    session: Optional[Session],
    collection_name: Optional[str],
    *,
    category_l1: Optional[str] = None,
    known_names: Optional[Set[str]] = None,
) -> bool:
    """True when a collection-like label is structural or decorates a real collection."""
    name = canonical_collection(collection_name)
    if not name:
        return False
    if _is_structural_collection_name(name):
        return True

    catalog = known_names if known_names is not None else (
        known_collection_names(session, category_l1=category_l1) if session is not None else set()
    )
    if not catalog:
        return False
    if is_suffix_only_polluted_collection(
        session,
        name,
        category_l1=category_l1,
        known_names=catalog,
    ):
        return True

    name_key = _norm_key(name)
    catalog_keys = {_norm_key(item): item for item in catalog}
    for base in _decorator_stripped_names(name):
        base_key = _norm_key(base)
        if base_key in catalog_keys:
            return True
        if any(_is_strict_edge_extension_name(base, other) for other in catalog if _norm_key(other) != base_key):
            return True

    for known in catalog:
        known_key = _norm_key(known)
        if not known_key or known_key == name_key:
            continue
        if not (name_key.startswith(known_key) or name_key.endswith(known_key)):
            continue
        remainder = name_key.replace(known_key, "", 1) if name_key.startswith(known_key) else name_key[: -len(known_key)]
        if remainder and remainder in {_norm_key(item) for item in COLLECTION_DECORATOR_NAMES}:
            return True
    return False


def hub_and_collection_from_path(path_slug: str) -> Tuple[Optional[str], Optional[str]]:
    segments = path_segments(normalize_plp_path(path_slug))
    if len(segments) != 2:
        return None, None
    hub = segments[0].replace("-", " ").title()
    collection = canonical_collection(segments[1].replace("-", " ").title())
    return hub, collection


def _longest_known_collection_extending(
    session: Session,
    collection_name: str,
    *,
    category_l1: Optional[str] = None,
    magento_connection_id: Optional[int] = None,
    known_names: Optional[Set[str]] = None,
) -> Optional[str]:
    """Return the longest known collection name that ends with *collection_name*."""
    short = canonical_collection(collection_name)
    if not short:
        return None
    known = known_names if known_names is not None else known_collection_names_for_hub(
        session,
        category_l1=category_l1,
        magento_connection_id=magento_connection_id,
    )
    short_key = _norm_key(short)
    matches = [
        name
        for name in known
        if _norm_key(name) != short_key and _is_strict_edge_extension_name(short, name)
    ]
    if not matches:
        return None
    return max(matches, key=len)


def resolve_authoritative_collection_name(
    session: Session,
    *,
    sku: Optional[str] = None,
    collection: Optional[str] = None,
    category_l1: Optional[str] = None,
    magento_connection_id: Optional[int] = None,
    known_names: Optional[Set[str]] = None,
) -> Optional[str]:
    """Prefer full canonical collection names from SKU style codes and master catalog."""
    from db.tribeca_sku_parse import collection_name_from_sku

    coll = canonical_collection(collection)
    from_sku = canonical_collection(collection_name_from_sku(sku))
    if from_sku:
        if not coll or len(from_sku) >= len(coll) or _norm_key(from_sku).endswith(_norm_key(coll)):
            coll = from_sku

    if not coll:
        return None

    extended = _longest_known_collection_extending(
        session,
        coll,
        category_l1=category_l1,
        magento_connection_id=magento_connection_id,
        known_names=known_names,
    )
    return extended or coll


def is_invalid_root_collection_taxonomy_slug(
    path_slug: str,
    *,
    category_l1: Optional[str] = None,
) -> bool:
    """True when a single-segment slug is a bare collection, not the hub itself."""
    slug = normalize_plp_path(path_slug)
    if not slug or "/" in slug:
        return False
    hub_slug = normalize_plp_path(canonical_category(category_l1) or DEFAULT_CATEGORY_L1)
    return slug != hub_slug


def is_polluted_collection_taxonomy_slug(
    session: Session,
    path_slug: str,
    *,
    category_l1: Optional[str] = None,
    known_names: Optional[Set[str]] = None,
) -> bool:
    """True for taxonomy collection paths that should never be proposed or synced."""
    slug = normalize_plp_path(path_slug)
    if not slug:
        return True
    if is_invalid_root_collection_taxonomy_slug(slug, category_l1=category_l1):
        return True
    if "/" not in slug:
        return False
    return is_polluted_hub_collection_path(session, slug, known_names=known_names)


def is_polluted_hub_collection_path(
    session: Session,
    path_slug: str,
    *,
    known_names: Optional[Set[str]] = None,
) -> bool:
    """True for hub child collection paths that duplicate a canonical collection suffix."""
    hub, collection = hub_and_collection_from_path(path_slug)
    if not hub or not collection:
        return False
    return is_polluted_collection_name(
        session,
        collection,
        category_l1=hub,
        known_names=known_names,
    )


def list_polluted_collection_paths(
    session: Session,
    *,
    hub_path_slug: Optional[str] = None,
    include_inactive: bool = False,
) -> List[Dict[str, Any]]:
    """Return active taxonomy collection paths that look like suffix-only aliases."""
    hub = normalize_plp_path(hub_path_slug) if hub_path_slug else None
    category_l1 = hub.replace("-", " ").title() if hub else None
    known = known_collection_names(session, category_l1=category_l1)

    stmt = select(MasterTaxonomyNode).where(MasterTaxonomyNode.node_kind == "collection")
    if not include_inactive:
        stmt = stmt.where(MasterTaxonomyNode.is_active.is_(True))
    if hub:
        stmt = stmt.where(MasterTaxonomyNode.path_slug.like(f"{hub}/%"))
    rows = list(session.scalars(stmt.order_by(MasterTaxonomyNode.path_slug)).all())

    polluted: List[Dict[str, Any]] = []
    for node in rows:
        if not is_polluted_hub_collection_path(session, node.path_slug, known_names=known):
            continue
        _, collection = hub_and_collection_from_path(node.path_slug)
        polluted.append(
            {
                "node_id": node.id,
                "path_slug": node.path_slug,
                "name": node.name,
                "collection": collection,
                "is_active": bool(node.is_active),
            }
        )
    return polluted


def _strip_magento_roots(path_names: str) -> str:
    parts = [part.strip() for part in str(path_names or "").split("/") if part.strip()]
    while parts and parts[0].lower() in MAGENTO_ROOT_PREFIXES:
        parts.pop(0)
    return "/".join(parts)


def _magento_path_parts(row: MagentoCategoryRegistry) -> List[str]:
    stripped = _strip_magento_roots(str(row.path_names or row.name or ""))
    return [part.strip() for part in stripped.split("/") if part.strip()]


def _is_magento_default_category_root_row(row: MagentoCategoryRegistry) -> bool:
    """True when category is a direct child of Default Category (store root), not under Kitchen Cabinets."""
    parts = _magento_path_parts(row)
    if len(parts) != 1:
        return False
    if int(row.level or 0) != 2:
        return False
    return _norm_key(parts[0]) not in {_norm_key(item) for item in NON_KITCHEN_ROOT_PREFIXES}


def _is_magento_registry_row_under_hub(
    row: MagentoCategoryRegistry,
    *,
    hub_title: str,
    collection_name: Optional[str] = None,
) -> bool:
    """True when a registry row is nested under the kitchen hub (not a store-root orphan)."""
    name = canonical_collection(collection_name or row.name)
    if not name or name in KITCHEN_NON_COLLECTION_NAMES:
        return False
    parts = _magento_path_parts(row)
    if not parts or parts[0].lower() != hub_title.lower():
        return False
    if canonical_collection(parts[-1]) != name:
        return False
    return len(parts) >= 2


def _magento_registry_has_nested_hub_copy(
    rows: List[MagentoCategoryRegistry],
    collection_name: str,
    *,
    hub_title: str,
    exclude_category_id: Optional[int] = None,
) -> bool:
    """True when the same collection name also exists under Kitchen Cabinets in the registry."""
    name = canonical_collection(collection_name)
    if not name:
        return False
    for other in rows:
        if exclude_category_id is not None and int(other.category_id) == int(exclude_category_id):
            continue
        if _is_magento_registry_row_under_hub(other, hub_title=hub_title, collection_name=name):
            return True
    return False


def _managed_connection_ids(channel_code: str, connection_id: Optional[int]) -> Set[int]:
    """Native + compat connection ids that should match taxonomy links for a channel."""
    if connection_id is None:
        return set()
    channel = str(channel_code or "").strip().lower()
    native = int(connection_id)
    ids = {native}
    if channel == "shopify":
        from db.compat_connections import compat_connection_id, decode_compat_connection_id

        decoded_type, decoded_native = decode_compat_connection_id(native)
        if decoded_type == "shopify":
            ids.add(decoded_native)
            ids.add(compat_connection_id("shopify", decoded_native))
        else:
            ids.add(compat_connection_id("shopify", native))
    return ids


def _managed_remote_collection_ids(
    session: Session,
    *,
    channel_code: str,
    connection_id: Optional[int] = None,
) -> Set[str]:
    """Remote ids we actively own via taxonomy links or listing-path targets.

    Only *active* links count as managed; deactivated links are stale and their
    remote ids are treated as orphan candidates.
    """
    channel = str(channel_code or "").strip().lower()
    managed: Set[str] = set()
    connection_ids = _managed_connection_ids(channel, connection_id)

    link_stmt = (
        select(MasterTaxonomyChannelLink.remote_id)
        .where(MasterTaxonomyChannelLink.channel_code == channel)
        .where(MasterTaxonomyChannelLink.is_active.is_(True))
    )
    if connection_ids:
        link_stmt = link_stmt.where(
            MasterTaxonomyChannelLink.connection_id.in_(connection_ids)
            | MasterTaxonomyChannelLink.connection_id.is_(None)
        )
    for remote_id in session.scalars(link_stmt).all():
        text = str(remote_id or "").strip()
        if text:
            managed.add(text)

    target_stmt = (
        select(ChannelListingPathTarget.remote_id)
        .where(ChannelListingPathTarget.channel_code == channel)
        .where(ChannelListingPathTarget.is_active.is_(True))
    )
    if connection_ids:
        target_stmt = target_stmt.where(
            ChannelListingPathTarget.connection_id.in_(connection_ids)
            | ChannelListingPathTarget.connection_id.is_(None)
        )
    for remote_id in session.scalars(target_stmt).all():
        text = str(remote_id or "").strip()
        if text:
            managed.add(text)

    return managed


def _magento_registry_unmanaged_purge_reason(
    row: MagentoCategoryRegistry,
    *,
    hub_title: str,
    managed_remote_ids: Set[str],
) -> Optional[Tuple[str, str]]:
    """Return (collection_name, reason) when a registry row is outside the taxonomy allowlist."""
    if str(row.category_id) in managed_remote_ids:
        return None
    collection_name = _magento_registry_row_collection_name(row, hub_title=hub_title)
    if not collection_name:
        return None
    return collection_name, "not_in_taxonomy"


def list_polluted_remote_targets_from_taxonomy(
    session: Session,
    *,
    hub_path_slug: str = "kitchen-cabinets",
    magento_connection_id: Optional[int] = None,
    shopify_connection_id: Optional[int] = None,
    include_inactive_nodes: bool = True,
) -> List[Dict[str, Any]]:
    """Remote ids from deactivated taxonomy nodes whose links are no longer managed."""
    from db.compat_connections import resolve_native_connection_id

    polluted_paths = list_polluted_collection_paths(
        session,
        hub_path_slug=hub_path_slug,
        include_inactive=include_inactive_nodes,
    )
    polluted_paths = [row for row in polluted_paths if not row.get("is_active")]
    if not polluted_paths:
        return []

    node_ids = [int(row["node_id"]) for row in polluted_paths]
    links = list(
        session.scalars(
            select(MasterTaxonomyChannelLink).where(MasterTaxonomyChannelLink.taxonomy_node_id.in_(node_ids))
        ).all()
    )
    nodes_by_id = {
        int(node.id): node
        for node in session.scalars(select(MasterTaxonomyNode).where(MasterTaxonomyNode.id.in_(node_ids))).all()
    }

    magento_native = resolve_native_connection_id("magento", magento_connection_id)
    shopify_native = resolve_native_connection_id("shopify", shopify_connection_id)
    magento_managed = (
        _managed_remote_collection_ids(session, channel_code="magento", connection_id=int(magento_native))
        if magento_native is not None
        else set()
    )
    shopify_managed = (
        _managed_remote_collection_ids(session, channel_code="shopify", connection_id=int(shopify_native))
        if shopify_native is not None
        else set()
    )
    targets: List[Dict[str, Any]] = []
    for link in links:
        channel = str(link.channel_code or "").strip().lower()
        if channel == "magento":
            allowed = _managed_connection_ids("magento", magento_native)
            if magento_native is not None and link.connection_id not in allowed and link.connection_id is not None:
                continue
        if channel == "shopify":
            allowed = _managed_connection_ids("shopify", shopify_native)
            if shopify_native is not None and link.connection_id not in allowed and link.connection_id is not None:
                continue
        node = nodes_by_id.get(int(link.taxonomy_node_id))
        if node is None or node.is_active:
            continue
        remote_id = str(link.remote_id or "").strip()
        if not remote_id:
            continue
        if channel == "magento" and remote_id in magento_managed:
            continue
        if channel == "shopify" and remote_id in shopify_managed:
            continue
        targets.append(
            {
                "source": "taxonomy_link",
                "link_id": int(link.id),
                "channel_code": channel,
                "connection_id": link.connection_id,
                "remote_id": remote_id,
                "path_slug": node.path_slug if node else None,
                "name": node.name if node else None,
                "link_is_active": bool(link.is_active),
                "node_is_active": bool(node.is_active) if node else None,
                "purge_reason": "deactivated_taxonomy_node",
            }
        )
    return targets


def _magento_registry_row_collection_name(
    row: MagentoCategoryRegistry,
    *,
    hub_title: str,
) -> Optional[str]:
    """Return the collection leaf name when a registry row is a collection landing category."""
    name = canonical_collection(row.name)
    if not name or name in KITCHEN_NON_COLLECTION_NAMES:
        return None
    stripped = _strip_magento_roots(str(row.path_names or row.name or ""))
    parts = [part.strip() for part in stripped.split("/") if part.strip()]
    if not parts:
        return None
    if parts[0].lower() == hub_title.lower():
        if len(parts) >= 2:
            return name
        return None
    if len(parts) == 1 and int(row.level or 0) == 2:
        if _norm_key(parts[0]) in {_norm_key(item) for item in NON_KITCHEN_ROOT_PREFIXES}:
            return None
        return name
    return None


def list_polluted_magento_categories_from_registry(
    session: Session,
    *,
    connection_id: int,
    hub_path_slug: str = "kitchen-cabinets",
) -> List[Dict[str, Any]]:
    """Magento registry collection rows not referenced by the active master taxonomy allowlist."""
    hub = normalize_plp_path(hub_path_slug)
    hub_title = hub.replace("-", " ").title()

    rows = list(
        session.scalars(
            select(MagentoCategoryRegistry)
            .where(MagentoCategoryRegistry.connection_id == int(connection_id))
            .where(MagentoCategoryRegistry.is_active.is_(True))
            .order_by(MagentoCategoryRegistry.path_names, MagentoCategoryRegistry.name)
        ).all()
    )
    managed = _managed_remote_collection_ids(
        session,
        channel_code="magento",
        connection_id=int(connection_id),
    )

    targets: List[Dict[str, Any]] = []
    for row in rows:
        unmanaged = _magento_registry_unmanaged_purge_reason(
            row,
            hub_title=hub_title,
            managed_remote_ids=managed,
        )
        if not unmanaged:
            continue
        collection_name, reason = unmanaged
        targets.append(
            {
                "source": "magento_registry",
                "channel_code": "magento",
                "connection_id": int(connection_id),
                "remote_id": str(row.category_id),
                "path_slug": normalize_plp_path(f"{hub}/{collection_name}"),
                "name": collection_name,
                "path_names": row.path_names,
                "level": row.level,
                "purge_reason": reason,
            }
        )
    return targets


def _is_structural_collection_title(title: str) -> bool:
    """True for category/hub labels that are not collection landing pages."""
    name = canonical_collection(title) or str(title or "").strip()
    if not name:
        return True
    structural = {_norm_key(item) for item in KITCHEN_NON_COLLECTION_NAMES}
    structural |= {_norm_key(item) for item in STRUCTURAL_COLLECTION_LABELS}
    return _norm_key(name) in structural


def list_polluted_shopify_collections_from_registry(
    session: Session,
    *,
    connection_id: int,
    hub_path_slug: str = "kitchen-cabinets",
    magento_connection_id: Optional[int] = None,
) -> List[Dict[str, Any]]:
    """Shopify registry collection landings not referenced by the active master taxonomy allowlist.

    Structural category labels (Wall Cabinets, Tall Cabinets, etc.) are skipped — same as Magento.
    Duplicate remote collections that share a managed title are still reported, with the managed
    remote id(s) attached so operators can tell taxonomy-owned vs orphan duplicates apart.
    """
    hub = normalize_plp_path(hub_path_slug)
    managed = _managed_remote_collection_ids(
        session,
        channel_code="shopify",
        connection_id=int(connection_id),
    )

    rows = list(
        session.scalars(
            select(ShopifyCollectionRegistry)
            .where(ShopifyCollectionRegistry.connection_id == int(connection_id))
            .order_by(ShopifyCollectionRegistry.title)
        ).all()
    )

    managed_by_title: Dict[str, List[str]] = {}
    for row in rows:
        remote_id = str(row.collection_id or "").strip()
        if not remote_id or remote_id not in managed:
            continue
        title_key = _norm_key(row.title)
        if not title_key:
            continue
        managed_by_title.setdefault(title_key, []).append(remote_id)

    targets: List[Dict[str, Any]] = []
    for row in rows:
        remote_id = str(row.collection_id or "").strip()
        if not remote_id or remote_id in managed:
            continue
        title = str(row.title or "").strip()
        if not title or _is_structural_collection_title(title):
            continue
        collection_name = canonical_collection(title) or title
        same_title_managed = list(managed_by_title.get(_norm_key(title), []))
        targets.append(
            {
                "source": "shopify_registry",
                "channel_code": "shopify",
                "connection_id": int(connection_id),
                "remote_id": remote_id,
                # Inferred from registry title — not proof that this remote owns the taxonomy node.
                "path_slug": normalize_plp_path(f"{hub}/{collection_name}"),
                "path_slug_source": "registry_title",
                "name": title,
                "handle": row.handle,
                "purge_reason": "not_in_taxonomy",
                "managed_same_title_remote_ids": same_title_managed,
                "note": (
                    "duplicate_title_of_managed_collection"
                    if same_title_managed
                    else "unmanaged_collection_landing"
                ),
            }
        )
    return targets


def list_polluted_remote_catalog_targets(
    session: Session,
    *,
    hub_path_slug: str = "kitchen-cabinets",
    magento_connection_id: Optional[int] = None,
    shopify_connection_id: Optional[int] = None,
    include_inactive_nodes: bool = True,
) -> List[Dict[str, Any]]:
    """Union of taxonomy links and pulled channel registries for suffix-only collections."""
    from db.compat_connections import resolve_native_connection_id

    merged: Dict[tuple[str, str, Optional[int]], Dict[str, Any]] = {}

    def add(target: Dict[str, Any]) -> None:
        key = (
            str(target.get("channel_code") or ""),
            str(target.get("remote_id") or ""),
            target.get("connection_id"),
        )
        if not key[0] or not key[1]:
            return
        existing = merged.get(key)
        if existing is None:
            merged[key] = dict(target)
            return
        sources = set(str(existing.get("sources") or existing.get("source") or "").split(","))
        sources.add(str(target.get("source") or ""))
        existing["sources"] = ",".join(sorted(s for s in sources if s))

    for target in list_polluted_remote_targets_from_taxonomy(
        session,
        hub_path_slug=hub_path_slug,
        magento_connection_id=magento_connection_id,
        shopify_connection_id=shopify_connection_id,
        include_inactive_nodes=include_inactive_nodes,
    ):
        add(target)

    magento_native = resolve_native_connection_id("magento", magento_connection_id)
    if magento_native is not None:
        for target in list_polluted_magento_categories_from_registry(
            session,
            connection_id=int(magento_native),
            hub_path_slug=hub_path_slug,
        ):
            add(target)

    shopify_native = resolve_native_connection_id("shopify", shopify_connection_id)
    if shopify_native is not None:
        for target in list_polluted_shopify_collections_from_registry(
            session,
            connection_id=int(shopify_native),
            hub_path_slug=hub_path_slug,
            magento_connection_id=magento_connection_id,
        ):
            add(target)

    return sorted(merged.values(), key=lambda row: (row.get("channel_code") or "", row.get("name") or ""))


# Collections that are never orphan candidates — Shopify system pages and store roots.
ORPHAN_PROTECTED_HANDLES = frozenset(
    {
        "frontpage",
        "home",
        "home-page",
        "all",
        "all-products",
    }
)

# Handle/title patterns that mark a remote collection as test/scratch junk (Tier 1).
_ORPHAN_TEST_PATTERNS = (
    re.compile(r"testing", re.IGNORECASE),
    re.compile(r"^\[test\]", re.IGNORECASE),
    re.compile(r"^test[-_\s]", re.IGNORECASE),
    re.compile(r"[-_]copy\d*$", re.IGNORECASE),
    re.compile(r"custom[-_\s]?smart[-_\s]?collection", re.IGNORECASE),
    re.compile(r"\bdo not use\b", re.IGNORECASE),
    re.compile(r"\bdeprecated\b", re.IGNORECASE),
)

# Tiers that describe how safe a remote-only collection is to delete.
ORPHAN_TIER_TEST = "test"
ORPHAN_TIER_EMPTY_UNLINKED = "empty_unlinked"
ORPHAN_TIER_REVIEW = "review"
ORPHAN_TIER_KEEP = "keep"

# Default tiers eligible for automated purge (most conservative).
ORPHAN_DEFAULT_PURGE_TIERS = frozenset({ORPHAN_TIER_TEST})


def _looks_like_test_collection(title: Optional[str], handle: Optional[str]) -> bool:
    for value in (title, handle):
        text = str(value or "").strip()
        if not text:
            continue
        if any(pattern.search(text) for pattern in _ORPHAN_TEST_PATTERNS):
            return True
    return False


def _title_matches_known_collection(title: Optional[str], known_keys: Set[str]) -> bool:
    """Best-effort check that a remote title corresponds to a canonical collection/label."""
    title_key = _norm_key(title)
    if not title_key:
        return False
    for known_key in known_keys:
        if len(known_key) < 4:
            continue
        if known_key in title_key or title_key in known_key:
            return True
    return False


def _classify_orphan_collection(
    *,
    title: Optional[str],
    handle: Optional[str],
    product_count: Optional[int],
    collection_type: Optional[str],
    is_known: bool,
) -> Tuple[str, str]:
    """Return (tier, reason) describing how safe an orphan collection is to delete."""
    count = int(product_count or 0)
    ctype = str(collection_type or "manual").strip().lower()
    if _looks_like_test_collection(title, handle) and count == 0:
        return ORPHAN_TIER_TEST, "test/scratch naming with no products"
    if is_known:
        return ORPHAN_TIER_KEEP, "title matches a known canonical collection"
    if count > 0:
        return ORPHAN_TIER_REVIEW, f"has {count} product(s) — review before delete"
    if ctype == "smart":
        return ORPHAN_TIER_REVIEW, "smart collection — automation rules may still apply"
    return ORPHAN_TIER_EMPTY_UNLINKED, "manual, no products, not linked, not in master catalog"


def list_orphan_shopify_collections(
    session: Session,
    *,
    connection_id: int,
    hub_path_slug: str = "kitchen-cabinets",
    magento_connection_id: Optional[int] = None,
) -> List[Dict[str, Any]]:
    """Shopify collections in the registry that we no longer actively manage."""
    hub_title = normalize_plp_path(hub_path_slug).replace("-", " ").title()
    known_names = known_collection_names_for_hub(
        session,
        category_l1=hub_title,
        magento_connection_id=magento_connection_id,
    )
    known_keys = {_norm_key(name) for name in known_names if name}
    known_keys |= {_norm_key(name) for name in KITCHEN_NON_COLLECTION_NAMES}
    managed = _managed_remote_collection_ids(
        session,
        channel_code="shopify",
        connection_id=int(connection_id),
    )

    rows = session.scalars(
        select(ShopifyCollectionRegistry)
        .where(ShopifyCollectionRegistry.connection_id == int(connection_id))
        .order_by(ShopifyCollectionRegistry.title)
    ).all()

    orphans: List[Dict[str, Any]] = []
    for row in rows:
        remote_id = str(row.collection_id or "").strip()
        if not remote_id:
            continue
        handle = str(row.handle or "").strip()
        if handle.lower() in ORPHAN_PROTECTED_HANDLES:
            continue
        if remote_id in managed:
            continue
        title = str(row.title or "").strip()
        is_known = _title_matches_known_collection(title, known_keys)
        tier, reason = _classify_orphan_collection(
            title=title,
            handle=handle,
            product_count=row.product_count,
            collection_type=row.collection_type,
            is_known=is_known,
        )
        orphans.append(
            {
                "source": "shopify_orphan",
                "channel_code": "shopify",
                "connection_id": int(connection_id),
                "remote_id": remote_id,
                "name": title,
                "title": title,
                "handle": handle,
                "collection_type": row.collection_type,
                "product_count": int(row.product_count or 0),
                "is_known": is_known,
                "orphan_tier": tier,
                "orphan_reason": reason,
            }
        )
    return orphans


def list_orphan_magento_categories(
    session: Session,
    *,
    connection_id: int,
    hub_path_slug: str = "kitchen-cabinets",
) -> List[Dict[str, Any]]:
    """Magento kitchen-hub categories we no longer actively manage."""
    hub = normalize_plp_path(hub_path_slug)
    hub_title = hub.replace("-", " ").title()
    known_names = known_collection_names_for_hub(
        session,
        category_l1=hub_title,
        magento_connection_id=int(connection_id),
    )
    known_keys = {_norm_key(name) for name in known_names if name}
    known_keys |= {_norm_key(name) for name in KITCHEN_NON_COLLECTION_NAMES}
    managed = _managed_remote_collection_ids(
        session,
        channel_code="magento",
        connection_id=int(connection_id),
    )

    rows = session.scalars(
        select(MagentoCategoryRegistry)
        .where(MagentoCategoryRegistry.connection_id == int(connection_id))
        .where(MagentoCategoryRegistry.is_active.is_(True))
        .order_by(MagentoCategoryRegistry.path_names, MagentoCategoryRegistry.name)
    ).all()

    orphans: List[Dict[str, Any]] = []
    for row in rows:
        collection_name = _magento_registry_row_collection_name(row, hub_title=hub_title)
        if not collection_name:
            continue
        remote_id = str(row.category_id)
        if remote_id in managed:
            continue
        is_known = _title_matches_known_collection(collection_name, known_keys)
        tier, reason = _classify_orphan_collection(
            title=collection_name,
            handle=None,
            product_count=None,
            collection_type="manual",
            is_known=is_known,
        )
        orphans.append(
            {
                "source": "magento_orphan",
                "channel_code": "magento",
                "connection_id": int(connection_id),
                "remote_id": remote_id,
                "name": collection_name,
                "title": collection_name,
                "path_names": row.path_names,
                "level": row.level,
                "is_known": is_known,
                "orphan_tier": tier,
                "orphan_reason": reason,
            }
        )
    return orphans


def list_orphan_remote_collections(
    session: Session,
    *,
    hub_path_slug: str = "kitchen-cabinets",
    magento_connection_id: Optional[int] = None,
    shopify_connection_id: Optional[int] = None,
) -> List[Dict[str, Any]]:
    """Remote-only collections/categories (present remotely, not actively managed by us)."""
    from db.compat_connections import resolve_native_connection_id

    orphans: List[Dict[str, Any]] = []

    magento_native = resolve_native_connection_id("magento", magento_connection_id)
    if magento_native is not None:
        orphans.extend(
            list_orphan_magento_categories(
                session,
                connection_id=int(magento_native),
                hub_path_slug=hub_path_slug,
            )
        )

    shopify_native = resolve_native_connection_id("shopify", shopify_connection_id)
    if shopify_native is not None:
        orphans.extend(
            list_orphan_shopify_collections(
                session,
                connection_id=int(shopify_native),
                hub_path_slug=hub_path_slug,
                magento_connection_id=magento_connection_id,
            )
        )

    tier_rank = {
        ORPHAN_TIER_TEST: 0,
        ORPHAN_TIER_EMPTY_UNLINKED: 1,
        ORPHAN_TIER_REVIEW: 2,
        ORPHAN_TIER_KEEP: 3,
    }
    return sorted(
        orphans,
        key=lambda row: (
            tier_rank.get(row.get("orphan_tier"), 9),
            row.get("channel_code") or "",
            row.get("name") or "",
        ),
    )


def filter_suffix_only_category_targets(
    session: Session,
    targets: Iterable[str],
    *,
    category_l1: Optional[str] = None,
    magento_connection_id: Optional[int] = None,
    known_names: Optional[Set[str]] = None,
) -> List[str]:
    """Drop polluted collection paths such as Ocean Blue or Snow White Accessories."""
    from db.catalog_intent_planner import ALL_PRODUCTS

    hub = canonical_category(category_l1) or DEFAULT_CATEGORY_L1
    hub_slug = normalize_plp_path(hub)
    known = known_names if known_names is not None else known_collection_names_for_hub(
        session,
        category_l1=hub,
        magento_connection_id=magento_connection_id,
    )
    out: List[str] = []
    seen: Set[str] = set()
    for path in targets:
        text = str(path or "").strip()
        if not text:
            continue
        key = _norm_key(text)
        if key in seen:
            continue
        if text == ALL_PRODUCTS:
            out.append(text)
            seen.add(key)
            continue
        if "/" not in text and normalize_plp_path(text) != hub_slug and _norm_key(text) != _norm_key(hub):
            continue
        leaf = text.split("/")[-1].strip() if "/" in text else text
        if is_polluted_collection_name(
            session,
            leaf,
            category_l1=hub,
            known_names=known,
        ):
            continue
        out.append(text)
        seen.add(key)
    return out