from __future__ import annotations

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

import pandas as pd
from sqlalchemy import or_, select
from sqlalchemy.orm import Session

from db.channel_assignments import active_skus_for_channel
from db.channel_exports import resolve_pipeline_channel_code
from db.channel_taxonomy import master_path_from_fields, resolve_taxonomy_mappings_for_fields
from db.models import (
    MasterProduct,
    ProductChannelTaxonomyAssignment,
)


ACTIVE_STATUS = "active"
INACTIVE_STATUS = "inactive"

MAGENTO_KIND = "category"
MAGENTO_CATEGORY_KINDS = frozenset({"category", "collection"})
SHOPIFY_PRODUCT_CATEGORY_KIND = "product_category"
SHOPIFY_COLLECTION_KIND = "collection"


def _split_filter_values(value: Optional[str]) -> List[str]:
    text = str(value or "").strip()
    if not text:
        return []
    parts = [
        part.strip()
        for part in text.replace("|", "\n").replace(";", "\n").replace(",", "\n").splitlines()
        if part.strip()
    ]
    seen = set()
    out: List[str] = []
    for part in parts:
        key = part.lower()
        if key in seen:
            continue
        seen.add(key)
        out.append(part)
    return out


def _apply_multi_like(stmt, column, value: Optional[str]):
    values = _split_filter_values(value)
    if not values:
        return stmt
    return stmt.where(or_(*[column.ilike(f"%{item}%") for item in values]))


def list_sku_taxonomy_assignments(
    session: Session,
    *,
    channel_code: Optional[str] = None,
    connection_id: Optional[int] = None,
    master_sku: Optional[str] = None,
    status: str = ACTIVE_STATUS,
    limit: Optional[int] = None,
) -> List[Dict[str, Any]]:
    stmt = select(ProductChannelTaxonomyAssignment).order_by(
        ProductChannelTaxonomyAssignment.master_sku,
        ProductChannelTaxonomyAssignment.sort_order,
        ProductChannelTaxonomyAssignment.id,
    )
    if channel_code:
        stmt = stmt.where(ProductChannelTaxonomyAssignment.channel_code == resolve_pipeline_channel_code(channel_code))
    if connection_id is not None:
        stmt = stmt.where(ProductChannelTaxonomyAssignment.connection_id == connection_id)
    if master_sku:
        stmt = stmt.where(ProductChannelTaxonomyAssignment.master_sku == str(master_sku).strip())
    if status:
        stmt = stmt.where(ProductChannelTaxonomyAssignment.assignment_status == status)
    if limit:
        stmt = stmt.limit(limit)
    return [_assignment_dict(row) for row in session.scalars(stmt).all()]


def upsert_sku_taxonomy_assignment(
    session: Session,
    *,
    master_sku: str,
    channel_code: str,
    remote_id: str,
    taxonomy_kind: str = MAGENTO_KIND,
    connection_id: Optional[int] = None,
    remote_path: Optional[str] = None,
    assignment_status: str = ACTIVE_STATUS,
    match_source: str = "manual",
    sort_order: int = 0,
    notes: Optional[str] = None,
) -> Dict[str, Any]:
    master_sku = str(master_sku or "").strip()
    remote_id = str(remote_id or "").strip()
    channel = resolve_pipeline_channel_code(channel_code)
    if not master_sku or not remote_id:
        raise ValueError("master_sku and remote_id are required")
    product = session.scalar(select(MasterProduct.sku).where(MasterProduct.sku == master_sku))
    if not product:
        raise ValueError(f"Unknown master SKU: {master_sku}")

    existing = session.scalar(
        select(ProductChannelTaxonomyAssignment)
        .where(ProductChannelTaxonomyAssignment.master_sku == master_sku)
        .where(ProductChannelTaxonomyAssignment.channel_code == channel)
        .where(ProductChannelTaxonomyAssignment.taxonomy_kind == taxonomy_kind)
        .where(ProductChannelTaxonomyAssignment.remote_id == remote_id)
        .where(
            ProductChannelTaxonomyAssignment.connection_id.is_(None)
            if connection_id is None
            else ProductChannelTaxonomyAssignment.connection_id == connection_id
        )
    )
    payload = {
        "master_sku": master_sku,
        "channel_code": channel,
        "connection_id": connection_id,
        "taxonomy_kind": taxonomy_kind,
        "remote_id": remote_id,
        "remote_path": remote_path,
        "assignment_status": assignment_status,
        "match_source": match_source,
        "sort_order": sort_order,
        "notes": notes,
    }
    if existing is None:
        row = ProductChannelTaxonomyAssignment(**payload)
        session.add(row)
        session.flush()
        return _assignment_dict(row)
    for key, value in payload.items():
        setattr(existing, key, value)
    session.flush()
    return _assignment_dict(existing)


def deactivate_sku_taxonomy_assignment(
    session: Session,
    *,
    assignment_id: Optional[int] = None,
    master_sku: Optional[str] = None,
    channel_code: Optional[str] = None,
    remote_id: Optional[str] = None,
    taxonomy_kind: Optional[str] = None,
    connection_id: Optional[int] = None,
) -> Dict[str, Any]:
    """Soft-delete a per-SKU taxonomy assignment."""
    row = None
    if assignment_id is not None:
        row = session.get(ProductChannelTaxonomyAssignment, int(assignment_id))
    elif master_sku and channel_code:
        stmt = (
            select(ProductChannelTaxonomyAssignment)
            .where(ProductChannelTaxonomyAssignment.master_sku == master_sku.strip())
            .where(ProductChannelTaxonomyAssignment.channel_code == channel_code.strip().lower())
            .where(ProductChannelTaxonomyAssignment.assignment_status == ACTIVE_STATUS)
        )
        if remote_id:
            stmt = stmt.where(ProductChannelTaxonomyAssignment.remote_id == str(remote_id).strip())
        if taxonomy_kind:
            stmt = stmt.where(ProductChannelTaxonomyAssignment.taxonomy_kind == taxonomy_kind)
        if connection_id is not None:
            stmt = stmt.where(
                (ProductChannelTaxonomyAssignment.connection_id == connection_id)
                | (ProductChannelTaxonomyAssignment.connection_id.is_(None))
            )
        row = session.scalar(stmt.limit(1))
    if row is None:
        raise ValueError("Assignment not found")
    row.assignment_status = INACTIVE_STATUS
    session.flush()
    return {"deactivated": True, "assignment": _assignment_dict(row)}


def import_sku_taxonomy_assignments_csv(session: Session, df: pd.DataFrame) -> Dict[str, int]:
    inserted = 0
    updated = 0
    skipped = 0
    for _, row in df.iterrows():
        master_sku = _cell(row, "master_sku", "sku")
        channel_code = _cell(row, "channel_code", "channel")
        remote_id = _cell(row, "remote_id", "category_id", "taxonomy_id")
        if not master_sku or not channel_code or not remote_id:
            skipped += 1
            continue
        channel = resolve_pipeline_channel_code(channel_code)
        if channel not in {"magento", "shopify"}:
            skipped += 1
            continue
        connection_id = _optional_int(row.get("connection_id"))
        taxonomy_kind = _cell(row, "taxonomy_kind", "kind") or _default_kind(channel)
        existing = session.scalar(
            select(ProductChannelTaxonomyAssignment.id)
            .where(ProductChannelTaxonomyAssignment.master_sku == master_sku)
            .where(ProductChannelTaxonomyAssignment.channel_code == channel)
            .where(ProductChannelTaxonomyAssignment.taxonomy_kind == taxonomy_kind)
            .where(ProductChannelTaxonomyAssignment.remote_id == remote_id)
            .where(
                ProductChannelTaxonomyAssignment.connection_id.is_(None)
                if connection_id is None
                else ProductChannelTaxonomyAssignment.connection_id == connection_id
            )
        )
        upsert_sku_taxonomy_assignment(
            session,
            master_sku=master_sku,
            channel_code=channel,
            remote_id=remote_id,
            taxonomy_kind=taxonomy_kind,
            connection_id=connection_id,
            remote_path=_cell(row, "remote_path", "path", "handle_or_path"),
            assignment_status=_cell(row, "assignment_status", "status") or ACTIVE_STATUS,
            match_source=_cell(row, "match_source") or "csv_import",
            sort_order=_optional_int(row.get("sort_order")) or 0,
            notes=_cell(row, "notes"),
        )
        if existing is None:
            inserted += 1
        else:
            updated += 1
    session.flush()
    return {"inserted": inserted, "updated": updated, "skipped": skipped}


def resolve_assignments_for_sku(
    session: Session,
    *,
    master_sku: str,
    channel_code: str,
    connection_id: Optional[int] = None,
    taxonomy_kind: Optional[str] = None,
) -> List[ProductChannelTaxonomyAssignment]:
    channel = resolve_pipeline_channel_code(channel_code)
    stmt = (
        select(ProductChannelTaxonomyAssignment)
        .where(ProductChannelTaxonomyAssignment.master_sku == master_sku)
        .where(ProductChannelTaxonomyAssignment.channel_code == channel)
        .where(ProductChannelTaxonomyAssignment.assignment_status == ACTIVE_STATUS)
        .order_by(ProductChannelTaxonomyAssignment.sort_order, ProductChannelTaxonomyAssignment.id)
    )
    if connection_id is not None:
        stmt = stmt.where(
            (ProductChannelTaxonomyAssignment.connection_id == connection_id)
            | (ProductChannelTaxonomyAssignment.connection_id.is_(None))
        )
    if taxonomy_kind:
        stmt = stmt.where(ProductChannelTaxonomyAssignment.taxonomy_kind == taxonomy_kind)
    rows = list(session.scalars(stmt).all())
    if connection_id is not None:
        conn_rows = [row for row in rows if row.connection_id == connection_id]
        if conn_rows:
            return conn_rows
    return rows


def sku_has_master_taxonomy_placements(session: Session, master_sku: str) -> bool:
    """True when the SKU is assigned to at least one active master_taxonomy_node listing path."""
    from db.channel_listing_path import resolve_listing_paths_for_sku
    from db.master_taxonomy_hierarchy import active_lineage_path_set

    sku = str(master_sku or "").strip()
    if not sku:
        return False
    slugs = [p["path_slug"] for p in resolve_listing_paths_for_sku(session, sku) if p.get("path_slug")]
    if not slugs:
        return False
    active_paths = active_lineage_path_set(session)
    count = sum(1 for slug in slugs if str(slug).strip() in active_paths)
    return bool(count)


def _coalesce_magento_category_path(*, remote_path: str, plp_path_slug: str = "") -> str:
    """Prefer full PLP slugs over leaf-only Magento labels (e.g. ``single-door``)."""
    from channel.url_canonical import normalize_plp_path

    remote = str(remote_path or "").strip()
    plp = normalize_plp_path(plp_path_slug or "")
    if not remote:
        return plp
    # Leaf-only remote_path is common when links were saved with category name only.
    if plp and "/" not in remote.replace("\\", "/"):
        return plp
    return remote


def _magento_outbound_membership_allowed(
    *,
    plp_path_slug: str = "",
    taxonomy_kind: str = "",
) -> bool:
    """Filter Magento category_links to L3 leaf + collection + all-products only."""
    from db.master_taxonomy_product_paths import is_magento_outbound_category_membership

    slug = str(plp_path_slug or "").strip()
    kind = str(taxonomy_kind or "").strip().lower()
    if slug:
        return is_magento_outbound_category_membership(slug, taxonomy_kind=kind)
    # Assignment / mapping fallbacks without a PLP slug: keep collections; drop unknown hubs.
    if kind == "collection":
        return True
    # Without a path we cannot tell hub/facet from leaf — keep (legacy mappings).
    return True


def _plp_slug_for_magento_remote(
    session: Session,
    *,
    remote_id: str,
    connection_id: Optional[int],
) -> str:
    """Best-effort PLP slug for a Magento remote category id via channel links."""
    from db.models import MasterTaxonomyChannelLink, MasterTaxonomyNode
    from db.master_taxonomy_hierarchy import active_lineage_path_set

    rid = str(remote_id or "").strip()
    if not rid:
        return ""
    active_paths = active_lineage_path_set(session)
    stmt = (
        select(MasterTaxonomyNode.path_slug)
        .join(
            MasterTaxonomyChannelLink,
            MasterTaxonomyChannelLink.taxonomy_node_id == MasterTaxonomyNode.id,
        )
        .where(MasterTaxonomyChannelLink.channel_code == "magento")
        .where(MasterTaxonomyChannelLink.is_active.is_(True))
        .where(MasterTaxonomyChannelLink.remote_id == rid)
        .limit(1)
    )
    if connection_id is not None:
        stmt = stmt.where(MasterTaxonomyChannelLink.connection_id == connection_id)
    for path_slug in session.scalars(stmt).all():
        slug = str(path_slug or "").strip()
        if slug and slug in active_paths:
            return slug
    return ""


def _append_magento_category_target(
    *,
    remote_id: str,
    remote_path: str,
    seen_ids: Set[int],
    seen_paths: Set[str],
    category_ids: List[int],
    category_paths: List[str],
    plp_path_slug: str = "",
    taxonomy_kind: str = "",
) -> None:
    if not _magento_outbound_membership_allowed(
        plp_path_slug=plp_path_slug,
        taxonomy_kind=taxonomy_kind,
    ):
        return
    rid = str(remote_id or "").strip()
    if rid.isdigit():
        category_id = int(rid)
        if category_id not in seen_ids:
            seen_ids.add(category_id)
            category_ids.append(category_id)
    path = _coalesce_magento_category_path(remote_path=remote_path, plp_path_slug=plp_path_slug)
    if path and path not in seen_paths:
        seen_paths.add(path)
        category_paths.append(path)


def _resolve_magento_assignments_for_sku(
    session: Session,
    *,
    master_sku: str,
    connection_id: Optional[int],
) -> List[ProductChannelTaxonomyAssignment]:
    stmt = (
        select(ProductChannelTaxonomyAssignment)
        .where(ProductChannelTaxonomyAssignment.master_sku == master_sku)
        .where(ProductChannelTaxonomyAssignment.channel_code == "magento")
        .where(ProductChannelTaxonomyAssignment.assignment_status == ACTIVE_STATUS)
        .where(ProductChannelTaxonomyAssignment.taxonomy_kind.in_(tuple(MAGENTO_CATEGORY_KINDS)))
        .order_by(ProductChannelTaxonomyAssignment.sort_order, ProductChannelTaxonomyAssignment.id)
    )
    if connection_id is not None:
        stmt = stmt.where(
            (ProductChannelTaxonomyAssignment.connection_id == connection_id)
            | (ProductChannelTaxonomyAssignment.connection_id.is_(None))
        )
    rows = list(session.scalars(stmt).all())
    if connection_id is not None:
        conn_rows = [row for row in rows if row.connection_id == connection_id]
        if conn_rows:
            return conn_rows
    return rows


def _resolve_magento_from_master_taxonomy(
    session: Session,
    *,
    master_sku: str,
    connection_id: Optional[int],
) -> Tuple[List[int], List[str]]:
    """Canonical Magento categories from master taxonomy listing paths only."""
    from db.channel_listing_path import expand_listing_path_targets, resolve_listing_paths_for_sku
    from db.models import MasterTaxonomyChannelLink, MasterTaxonomyNode

    category_ids: List[int] = []
    category_paths: List[str] = []
    seen_ids: Set[int] = set()
    seen_paths: Set[str] = set()

    for target in expand_listing_path_targets(
        session,
        master_sku=master_sku,
        channel_code="magento",
        connection_id=connection_id,
    )[0]:
        kind = str(target.get("taxonomy_kind") or "")
        if kind not in MAGENTO_CATEGORY_KINDS:
            continue
        _append_magento_category_target(
            remote_id=str(target.get("remote_id") or ""),
            remote_path=str(target.get("remote_path") or ""),
            plp_path_slug=str(target.get("plp_path_slug") or ""),
            taxonomy_kind=kind,
            seen_ids=seen_ids,
            seen_paths=seen_paths,
            category_ids=category_ids,
            category_paths=category_paths,
        )

    # Listing-path targets are canonical. Do not union product_channel_taxonomy_assignment
    # remotes on top — those often retain leaf-collision IDs (Wall/Single Door, Frost,
    # Accessories/Fillers) after PLP targets were repaired.
    if category_ids or category_paths:
        return category_ids, category_paths

    for row in _resolve_magento_assignments_for_sku(
        session,
        master_sku=master_sku,
        connection_id=connection_id,
    ):
        if str(row.match_source or "") not in {"master_taxonomy", "bulk_assign", "taxonomy_merge", "manual"}:
            continue
        plp_slug = _plp_slug_for_magento_remote(
            session,
            remote_id=str(row.remote_id or ""),
            connection_id=connection_id,
        )
        _append_magento_category_target(
            remote_id=str(row.remote_id or ""),
            remote_path=str(row.remote_path or ""),
            plp_path_slug=plp_slug,
            taxonomy_kind=str(row.taxonomy_kind or ""),
            seen_ids=seen_ids,
            seen_paths=seen_paths,
            category_ids=category_ids,
            category_paths=category_paths,
        )

    if category_ids or category_paths:
        return category_ids, category_paths

    for path in resolve_listing_paths_for_sku(session, master_sku):
        slug = str(path.get("path_slug") or "").strip()
        if not slug:
            continue
        from db.master_taxonomy_hierarchy import active_lineage_node_by_slug

        node = active_lineage_node_by_slug(session, slug)
        if node is None:
            continue
        link_stmt = (
            select(MasterTaxonomyChannelLink)
            .where(MasterTaxonomyChannelLink.taxonomy_node_id == node.id)
            .where(MasterTaxonomyChannelLink.channel_code == "magento")
            .where(MasterTaxonomyChannelLink.is_active.is_(True))
            .where(MasterTaxonomyChannelLink.taxonomy_kind.in_(tuple(MAGENTO_CATEGORY_KINDS)))
        )
        if connection_id is not None:
            link_stmt = link_stmt.where(MasterTaxonomyChannelLink.connection_id == connection_id)
        for link in session.scalars(link_stmt).all():
            _append_magento_category_target(
                remote_id=str(link.remote_id or ""),
                remote_path=str(link.remote_path or ""),
                plp_path_slug=slug,
                taxonomy_kind=str(link.taxonomy_kind or ""),
                seen_ids=seen_ids,
                seen_paths=seen_paths,
                category_ids=category_ids,
                category_paths=category_paths,
            )

    return category_ids, category_paths


def resolve_magento_category_targets(
    session: Session,
    *,
    master_sku: str,
    canonical_fields: Dict[str, Any],
    connection_id: Optional[int],
) -> Tuple[List[int], List[str]]:
    """Resolve Magento category targets for a master SKU.

    When the SKU is on a master taxonomy listing path, only canonical taxonomy-linked
    categories are returned (legacy multi-path intent is ignored).
    """
    if sku_has_master_taxonomy_placements(session, master_sku):
        return _resolve_magento_from_master_taxonomy(
            session,
            master_sku=master_sku,
            connection_id=connection_id,
        )

    category_ids: List[int] = []
    category_paths: List[str] = []
    seen_ids: Set[int] = set()
    seen_paths: Set[str] = set()

    from db.channel_listing_path import expand_listing_path_targets

    for target in expand_listing_path_targets(
        session,
        master_sku=master_sku,
        channel_code="magento",
        connection_id=connection_id,
    )[0]:
        kind = str(target.get("taxonomy_kind") or "")
        if kind not in MAGENTO_CATEGORY_KINDS:
            continue
        _append_magento_category_target(
            remote_id=str(target.get("remote_id") or ""),
            remote_path=str(target.get("remote_path") or ""),
            plp_path_slug=str(target.get("plp_path_slug") or ""),
            taxonomy_kind=kind,
            seen_ids=seen_ids,
            seen_paths=seen_paths,
            category_ids=category_ids,
            category_paths=category_paths,
        )

    if category_ids or category_paths:
        return category_ids, category_paths

    for row in _resolve_magento_assignments_for_sku(
        session,
        master_sku=master_sku,
        connection_id=connection_id,
    ):
        plp_slug = _plp_slug_for_magento_remote(
            session,
            remote_id=str(row.remote_id or ""),
            connection_id=connection_id,
        )
        _append_magento_category_target(
            remote_id=str(row.remote_id or ""),
            remote_path=str(row.remote_path or ""),
            plp_path_slug=plp_slug,
            taxonomy_kind=str(row.taxonomy_kind or ""),
            seen_ids=seen_ids,
            seen_paths=seen_paths,
            category_ids=category_ids,
            category_paths=category_paths,
        )

    if category_ids or category_paths:
        return category_ids, category_paths

    for mapping in resolve_taxonomy_mappings_for_fields(
        session,
        channel_code="magento",
        fields=canonical_fields,
        connection_id=connection_id,
    ):
        rule = mapping.transform_rule or {}
        for category_id in _category_ids_from_rule(rule):
            if category_id not in seen_ids:
                seen_ids.add(category_id)
                category_ids.append(category_id)
        for path in _category_paths_from_rule(rule):
            if path not in seen_paths:
                seen_paths.add(path)
                category_paths.append(path)
        remote_id = str(mapping.channel_remote_id or "").strip()
        if remote_id.isdigit():
            category_id = int(remote_id)
            if category_id not in seen_ids:
                seen_ids.add(category_id)
                category_ids.append(category_id)
            continue
        path = str(mapping.channel_handle_or_path or "").strip()
        if path and path not in seen_paths:
            seen_paths.add(path)
            category_paths.append(path)
    return category_ids, category_paths


def resolve_shopify_product_category_id(
    session: Session,
    *,
    master_sku: str,
    canonical_fields: Dict[str, Any],
    connection_id: Optional[int] = None,
) -> Optional[str]:
    rows = resolve_assignments_for_sku(
        session,
        master_sku=master_sku,
        channel_code="shopify",
        connection_id=connection_id,
        taxonomy_kind=SHOPIFY_PRODUCT_CATEGORY_KIND,
    )
    if rows:
        return str(rows[0].remote_id).strip() or None
    from shopify.taxonomy_sync import suggest_shopify_product_category

    suggestion = suggest_shopify_product_category(
        session,
        category_l1=canonical_fields.get("category_l1"),
        category_l2=canonical_fields.get("category_l2"),
        category_l3=canonical_fields.get("category_l3"),
        collection=canonical_fields.get("collection"),
        product_name=canonical_fields.get("title") or canonical_fields.get("name"),
    )
    return suggestion.get("taxonomy_id") if suggestion else None


def resolve_shopify_collection_ids(
    session: Session,
    *,
    master_sku: str,
    connection_id: Optional[int] = None,
) -> List[str]:
    seen: Set[str] = set()
    out: List[str] = []
    for row in resolve_assignments_for_sku(
        session,
        master_sku=master_sku,
        channel_code="shopify",
        connection_id=connection_id,
        taxonomy_kind=SHOPIFY_COLLECTION_KIND,
    ):
        collection_id = str(row.remote_id or "").strip()
        if collection_id and collection_id not in seen:
            seen.add(collection_id)
            out.append(collection_id)

    from db.channel_listing_path import expand_listing_path_targets

    for target in expand_listing_path_targets(
        session,
        master_sku=master_sku,
        channel_code="shopify",
        connection_id=connection_id,
    )[0]:
        if target.get("taxonomy_kind") != SHOPIFY_COLLECTION_KIND:
            continue
        collection_id = str(target.get("remote_id") or "").strip()
        if collection_id and collection_id not in seen:
            seen.add(collection_id)
            out.append(collection_id)
    return out


def resolve_taxonomy_targets_for_push(
    session: Session,
    *,
    master_sku: str,
    channel_code: str,
    canonical_fields: Dict[str, Any],
    connection_id: Optional[int] = None,
) -> Dict[str, Any]:
    """Stable taxonomy snapshot included in channel payload hashes."""
    from db.channel_listing_path import expand_listing_path_targets

    channel = resolve_pipeline_channel_code(channel_code)
    _, plp_paths = expand_listing_path_targets(
        session,
        master_sku=master_sku,
        channel_code=channel,
        connection_id=connection_id,
    )
    if channel == "magento":
        category_ids, category_paths = resolve_magento_category_targets(
            session,
            master_sku=master_sku,
            canonical_fields=canonical_fields,
            connection_id=connection_id,
        )
        return {
            "magento_category_ids": category_ids,
            "magento_category_paths": category_paths,
            "plp_path_slugs": plp_paths,
            "pdp_slug": canonical_fields.get("product_url_slug"),
        }
    if channel == "shopify":
        return {
            "shopify_product_category": resolve_shopify_product_category_id(
                session,
                master_sku=master_sku,
                canonical_fields=canonical_fields,
                connection_id=connection_id,
            ),
            "shopify_collection_ids": resolve_shopify_collection_ids(
                session,
                master_sku=master_sku,
                connection_id=connection_id,
            ),
            "plp_path_slugs": plp_paths,
            "pdp_slug": canonical_fields.get("product_url_slug"),
        }
    return {}


def resolve_filtered_master_skus(
    session: Session,
    *,
    skus: Optional[Iterable[str]] = None,
    channel_code: Optional[str] = None,
    only_assigned: bool = True,
    category_l1: Optional[str] = None,
    category_l2: Optional[str] = None,
    category_l3: Optional[str] = None,
    collection: Optional[str] = None,
    product_family: Optional[str] = None,
    search: Optional[str] = None,
    master_filters: Optional[Any] = None,
    limit: Optional[int] = None,
) -> List[str]:
    """Filter master SKUs for bulk taxonomy assignment workflows."""
    from db.master_product_filters import apply_master_attribute_filters, parse_master_filters

    stmt = select(MasterProduct.sku).where(MasterProduct.is_active.is_(True))
    if channel_code and only_assigned:
        assigned = active_skus_for_channel(session, resolve_pipeline_channel_code(channel_code))
        if not assigned:
            return []
        stmt = stmt.where(MasterProduct.sku.in_(sorted(assigned)))
    if skus:
        wanted = {str(s).strip() for s in skus if str(s).strip()}
        if wanted:
            stmt = stmt.where(MasterProduct.sku.in_(sorted(wanted)))
    stmt = _apply_multi_like(stmt, MasterProduct.category_l1, category_l1)
    stmt = _apply_multi_like(stmt, MasterProduct.category_l2, category_l2)
    stmt = _apply_multi_like(stmt, MasterProduct.category_l3, category_l3)
    stmt = _apply_multi_like(stmt, MasterProduct.collection, collection)
    stmt = _apply_multi_like(stmt, MasterProduct.product_family, product_family)
    if search:
        values = _split_filter_values(search)
        if values:
            stmt = stmt.where(or_(*[
                (MasterProduct.sku.ilike(f"%{item}%")) | (MasterProduct.name.ilike(f"%{item}%"))
                for item in values
            ]))
    stmt = apply_master_attribute_filters(stmt, filters=parse_master_filters(master_filters))
    stmt = stmt.order_by(MasterProduct.sku)
    if limit:
        stmt = stmt.limit(limit)
    return [str(sku).strip() for (sku,) in session.execute(stmt).all() if str(sku).strip()]


def bulk_assign_sku_taxonomy(
    session: Session,
    *,
    channel_code: str,
    remote_ids: List[str],
    taxonomy_kind: str,
    connection_id: Optional[int] = None,
    skus: Optional[Iterable[str]] = None,
    only_assigned: bool = True,
    category_l1: Optional[str] = None,
    category_l2: Optional[str] = None,
    category_l3: Optional[str] = None,
    collection: Optional[str] = None,
    product_family: Optional[str] = None,
    search: Optional[str] = None,
    master_filters: Optional[Any] = None,
    remote_path: Optional[str] = None,
    replace_existing: bool = False,
    dry_run: bool = False,
    limit: Optional[int] = None,
) -> Dict[str, Any]:
    """Assign one or more taxonomy targets to a filtered SKU set."""
    channel = resolve_pipeline_channel_code(channel_code)
    if channel not in {"magento", "shopify"}:
        raise ValueError(f"Unsupported channel: {channel_code}")
    targets = [str(remote_id or "").strip() for remote_id in remote_ids if str(remote_id or "").strip()]
    if not targets:
        raise ValueError("remote_ids is required")

    kind = str(taxonomy_kind or "").strip() or _default_kind(channel)
    master_skus = resolve_filtered_master_skus(
        session,
        skus=skus,
        channel_code=channel,
        only_assigned=only_assigned,
        category_l1=category_l1,
        category_l2=category_l2,
        category_l3=category_l3,
        collection=collection,
        product_family=product_family,
        search=search,
        master_filters=master_filters,
        limit=limit,
    )
    if not master_skus:
        return {
            "channel_code": channel,
            "taxonomy_kind": kind,
            "sku_count": 0,
            "assignment_count": 0,
            "deactivated": 0,
        }

    deactivated = 0
    if replace_existing and not dry_run:
        for master_sku in master_skus:
            rows = resolve_assignments_for_sku(
                session,
                master_sku=master_sku,
                channel_code=channel,
                connection_id=connection_id,
                taxonomy_kind=kind,
            )
            for row in rows:
                row.assignment_status = INACTIVE_STATUS
                deactivated += 1
        session.flush()

    candidates: List[Dict[str, Any]] = []
    for master_sku in master_skus:
        for remote_id in targets:
            candidates.append(
                _candidate(
                    master_sku,
                    channel,
                    connection_id,
                    kind,
                    remote_id,
                    remote_path,
                    "bulk_assign",
                )
            )

    upserted = 0
    if not dry_run:
        for item in candidates:
            upsert_sku_taxonomy_assignment(session, **item)
            upserted += 1

    return {
        "channel_code": channel,
        "taxonomy_kind": kind,
        "connection_id": connection_id,
        "sku_count": len(master_skus),
        "remote_ids": targets,
        "assignment_count": len(candidates),
        "upserted": upserted if not dry_run else 0,
        "would_upsert": len(candidates) if dry_run else upserted,
        "deactivated": deactivated,
        "replace_existing": replace_existing,
        "samples": candidates[:25],
    }


def bulk_remove_sku_taxonomy(
    session: Session,
    *,
    channel_code: str,
    taxonomy_kind: Optional[str] = None,
    remote_ids: Optional[List[str]] = None,
    connection_id: Optional[int] = None,
    skus: Optional[Iterable[str]] = None,
    only_assigned: bool = False,
    category_l1: Optional[str] = None,
    category_l2: Optional[str] = None,
    category_l3: Optional[str] = None,
    collection: Optional[str] = None,
    product_family: Optional[str] = None,
    search: Optional[str] = None,
    master_filters: Optional[Any] = None,
    dry_run: bool = False,
) -> Dict[str, Any]:
    """Deactivate assignments for filtered SKUs (optionally scoped to remote_ids/kind)."""
    channel = resolve_pipeline_channel_code(channel_code)
    master_skus = resolve_filtered_master_skus(
        session,
        skus=skus,
        channel_code=channel,
        only_assigned=only_assigned,
        category_l1=category_l1,
        category_l2=category_l2,
        category_l3=category_l3,
        collection=collection,
        product_family=product_family,
        search=search,
        master_filters=master_filters,
    )
    if not master_skus:
        return {"channel_code": channel, "deactivated": 0}

    stmt = (
        select(ProductChannelTaxonomyAssignment)
        .where(ProductChannelTaxonomyAssignment.channel_code == channel)
        .where(ProductChannelTaxonomyAssignment.master_sku.in_(sorted(master_skus)))
        .where(ProductChannelTaxonomyAssignment.assignment_status == ACTIVE_STATUS)
    )
    if taxonomy_kind:
        stmt = stmt.where(ProductChannelTaxonomyAssignment.taxonomy_kind == taxonomy_kind)
    if connection_id is not None:
        stmt = stmt.where(
            (ProductChannelTaxonomyAssignment.connection_id == connection_id)
            | (ProductChannelTaxonomyAssignment.connection_id.is_(None))
        )
    if remote_ids:
        targets = {str(remote_id).strip() for remote_id in remote_ids if str(remote_id).strip()}
        stmt = stmt.where(ProductChannelTaxonomyAssignment.remote_id.in_(sorted(targets)))

    rows = list(session.scalars(stmt).all())
    if not dry_run:
        for row in rows:
            row.assignment_status = INACTIVE_STATUS
        session.flush()
    return {
        "channel_code": channel,
        "sku_count": len(master_skus),
        "deactivated": len(rows),
        "samples": [_assignment_dict(row) for row in rows[:25]],
    }


def auto_assign_from_path_mappings(
    session: Session,
    *,
    channel_code: str,
    connection_id: Optional[int] = None,
    only_assigned: bool = True,
    dry_run: bool = False,
) -> Dict[str, Any]:
    """Materialize channel_taxonomy_mapping hits into per-SKU assignment rows."""
    channel = resolve_pipeline_channel_code(channel_code)
    if channel not in {"magento", "shopify"}:
        raise ValueError(f"Unsupported channel: {channel_code}")

    stmt = select(MasterProduct).where(MasterProduct.is_active.is_(True))
    if only_assigned:
        assigned = active_skus_for_channel(session, channel)
        if not assigned:
            return {"channel_code": channel, "candidates": 0, "upserted": 0, "skipped_existing": 0}
        stmt = stmt.where(MasterProduct.sku.in_(sorted(assigned)))
    products = list(session.scalars(stmt).all())

    existing_keys = {
        (row.master_sku, row.taxonomy_kind, row.remote_id, row.connection_id)
        for row in session.scalars(
            select(ProductChannelTaxonomyAssignment).where(ProductChannelTaxonomyAssignment.channel_code == channel)
        ).all()
    }

    candidates: List[Dict[str, Any]] = []
    for product in products:
        fields = _master_fields(product)
        mappings = resolve_taxonomy_mappings_for_fields(
            session,
            channel_code=channel,
            fields=fields,
            connection_id=connection_id,
        )
        for mapping in mappings:
            kind = str(mapping.channel_taxonomy_kind or MAGENTO_KIND).strip() or MAGENTO_KIND
            if channel == "shopify" and kind == "collection":
                kind = SHOPIFY_COLLECTION_KIND
            remote_id = str(mapping.channel_remote_id or "").strip()
            if not remote_id and kind == MAGENTO_KIND:
                rule = mapping.transform_rule or {}
                ids = rule.get("category_ids") or rule.get("category_id")
                if ids:
                    values = ids if isinstance(ids, list) else str(ids).replace("|", ",").split(",")
                    for value in values:
                        text = str(value or "").strip()
                        if text.isdigit():
                            candidates.append(
                                _candidate(
                                    product.sku,
                                    channel,
                                    connection_id,
                                    kind,
                                    text,
                                    mapping.channel_handle_or_path,
                                    "auto_path_map",
                                )
                            )
                continue
            if not remote_id:
                continue
            candidates.append(
                _candidate(
                    product.sku,
                    channel,
                    connection_id,
                    kind,
                    remote_id,
                    mapping.channel_handle_or_path,
                    "auto_path_map",
                )
            )

    upserted = 0
    skipped_existing = 0
    if not dry_run:
        for item in candidates:
            key = (item["master_sku"], item["taxonomy_kind"], item["remote_id"], item["connection_id"])
            if key in existing_keys:
                skipped_existing += 1
                continue
            upsert_sku_taxonomy_assignment(session, **item)
            existing_keys.add(key)
            upserted += 1
    return {
        "channel_code": channel,
        "connection_id": connection_id,
        "product_count": len(products),
        "candidate_count": len(candidates),
        "upserted": upserted if not dry_run else 0,
        "would_upsert": len(candidates) if dry_run else upserted,
        "skipped_existing": skipped_existing,
        "samples": candidates[:25],
    }


def auto_assign_shopify_product_categories(
    session: Session,
    *,
    connection_id: Optional[int] = None,
    only_assigned: bool = True,
    dry_run: bool = False,
) -> Dict[str, Any]:
    """Suggest Shopify Standard Product Taxonomy categories from master category text."""
    stmt = select(MasterProduct).where(MasterProduct.is_active.is_(True))
    if only_assigned:
        assigned = active_skus_for_channel(session, "shopify")
        if not assigned:
            return {"channel_code": "shopify", "candidates": 0, "upserted": 0}
        stmt = stmt.where(MasterProduct.sku.in_(sorted(assigned)))
    products = list(session.scalars(stmt).all())

    existing = {
        row.master_sku
        for row in session.scalars(
            select(ProductChannelTaxonomyAssignment)
            .where(ProductChannelTaxonomyAssignment.channel_code == "shopify")
            .where(ProductChannelTaxonomyAssignment.taxonomy_kind == SHOPIFY_PRODUCT_CATEGORY_KIND)
            .where(ProductChannelTaxonomyAssignment.assignment_status == ACTIVE_STATUS)
        ).all()
    }

    candidates: List[Dict[str, Any]] = []
    from shopify.taxonomy_sync import suggest_shopify_product_category

    for product in products:
        if product.sku in existing:
            continue
        fields = _master_fields(product)
        suggestion = suggest_shopify_product_category(
            session,
            category_l1=fields.get("category_l1"),
            category_l2=fields.get("category_l2"),
            category_l3=fields.get("category_l3"),
            collection=fields.get("collection"),
            product_name=fields.get("title") or product.name,
        )
        if not suggestion:
            continue
        candidates.append(
            _candidate(
                product.sku,
                "shopify",
                connection_id,
                SHOPIFY_PRODUCT_CATEGORY_KIND,
                suggestion["taxonomy_id"],
                suggestion.get("full_name"),
                "auto_keyword",
            )
        )

    upserted = 0
    if not dry_run:
        for item in candidates:
            upsert_sku_taxonomy_assignment(session, **item)
            upserted += 1
    return {
        "channel_code": "shopify",
        "connection_id": connection_id,
        "product_count": len(products),
        "candidate_count": len(candidates),
        "upserted": upserted if not dry_run else 0,
        "would_upsert": len(candidates) if dry_run else upserted,
        "samples": candidates[:25],
    }


def export_shopify_assignment_rows(
    session: Session,
    *,
    connection_id: Optional[int] = None,
    only_assigned: bool = True,
) -> List[Dict[str, Any]]:
    stmt = select(MasterProduct).where(MasterProduct.is_active.is_(True))
    if only_assigned:
        assigned = active_skus_for_channel(session, "shopify")
        if not assigned:
            return []
        stmt = stmt.where(MasterProduct.sku.in_(sorted(assigned)))
    products = list(session.scalars(stmt).all())
    rows: List[Dict[str, Any]] = []
    for product in products:
        fields = _master_fields(product)
        rows.append(
            {
                "master_sku": product.sku,
                "name": product.name,
                "category_l1": product.category_l1,
                "category_l2": product.category_l2,
                "category_l3": product.category_l3,
                "collection": product.collection,
                "shopify_product_category": resolve_shopify_product_category_id(
                    session,
                    master_sku=product.sku,
                    canonical_fields=fields,
                    connection_id=connection_id,
                ),
                "shopify_collection_ids": "|".join(
                    resolve_shopify_collection_ids(
                        session,
                        master_sku=product.sku,
                        connection_id=connection_id,
                    )
                ),
            }
        )
    return rows


def export_magento_category_assignment_rows(
    session: Session,
    *,
    connection_id: Optional[int] = None,
    only_assigned: bool = True,
) -> List[Dict[str, Any]]:
    """Wide export for dashboard: master SKU + Magento category ids/paths."""
    stmt = select(MasterProduct).where(MasterProduct.is_active.is_(True))
    if only_assigned:
        assigned = active_skus_for_channel(session, "magento")
        if not assigned:
            return []
        stmt = stmt.where(MasterProduct.sku.in_(sorted(assigned)))
    products = list(session.scalars(stmt).all())
    rows: List[Dict[str, Any]] = []
    for product in products:
        fields = _master_fields(product)
        category_ids, category_paths = resolve_magento_category_targets(
            session,
            master_sku=product.sku,
            canonical_fields=fields,
            connection_id=connection_id,
        )
        rows.append(
            {
                "master_sku": product.sku,
                "name": product.name,
                "category_l1": product.category_l1,
                "category_l2": product.category_l2,
                "category_l3": product.category_l3,
                "collection": product.collection,
                "magento_category_ids": "|".join(str(i) for i in category_ids),
                "magento_category_paths": "|".join(category_paths),
                "assignment_count": len(category_ids) + len(category_paths),
            }
        )
    return rows


def _candidate(
    master_sku: str,
    channel_code: str,
    connection_id: Optional[int],
    taxonomy_kind: str,
    remote_id: str,
    remote_path: Optional[str],
    match_source: str,
) -> Dict[str, Any]:
    return {
        "master_sku": master_sku,
        "channel_code": channel_code,
        "connection_id": connection_id,
        "taxonomy_kind": taxonomy_kind,
        "remote_id": remote_id,
        "remote_path": remote_path,
        "match_source": match_source,
    }


def _master_fields(product: MasterProduct) -> Dict[str, Any]:
    fields = {
        "sku": product.sku,
        "title": product.name,
        "name": product.name,
        "category_l1": product.category_l1,
        "category_l2": product.category_l2,
        "category_l3": product.category_l3,
        "collection": product.collection,
        "product_family": product.product_family,
    }
    l1, l2, l3, coll = master_path_from_fields(fields)
    fields["category_l1"] = l1 or fields["category_l1"]
    fields["category_l2"] = l2 or fields["category_l2"]
    fields["category_l3"] = l3 or fields["category_l3"]
    fields["collection"] = coll or fields["collection"]
    return fields


def _assignment_dict(row: ProductChannelTaxonomyAssignment) -> Dict[str, Any]:
    return {
        "id": row.id,
        "master_sku": row.master_sku,
        "channel_code": row.channel_code,
        "connection_id": row.connection_id,
        "taxonomy_kind": row.taxonomy_kind,
        "remote_id": row.remote_id,
        "remote_path": row.remote_path,
        "assignment_status": row.assignment_status,
        "match_source": row.match_source,
        "sort_order": row.sort_order,
        "notes": row.notes,
        "updated_at": row.updated_at.isoformat() if row.updated_at else None,
    }


def _default_kind(channel: str) -> str:
    if channel == "shopify":
        return SHOPIFY_PRODUCT_CATEGORY_KIND
    return MAGENTO_KIND


def _cell(row: pd.Series, *names: str) -> Optional[str]:
    for name in names:
        if name in row.index:
            text = str(row.get(name) or "").strip()
            if text and text.lower() not in {"nan", "none", "null"}:
                return text
    return None


def _optional_int(value: Any) -> Optional[int]:
    text = str(value or "").strip()
    if not text:
        return None
    try:
        return int(text)
    except (TypeError, ValueError):
        return None


def _category_ids_from_rule(rule: Dict[str, Any]) -> List[int]:
    raw = rule.get("category_ids", rule.get("category_id"))
    if raw is None:
        return []
    values = raw if isinstance(raw, list) else str(raw).replace("|", ",").split(",")
    ids: List[int] = []
    for value in values:
        text = str(value or "").strip()
        if text.isdigit():
            ids.append(int(text))
    return ids


def _category_paths_from_rule(rule: Dict[str, Any]) -> List[str]:
    raw = rule.get("category_paths", rule.get("category_path", rule.get("path")))
    if raw is None:
        return []
    values = raw if isinstance(raw, list) else str(raw).split("|")
    return [str(value or "").strip() for value in values if str(value or "").strip()]