"""Taxonomy intent from master SKU import — match existing nodes or propose new ones."""

from __future__ import annotations

import re
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Sequence, Set, Tuple

from sqlalchemy import select
from sqlalchemy.orm import Session

from channel.url_canonical import normalize_plp_path
from db.catalog_intent_planner import ALL_PRODUCTS, category_targets_for_product_guarded, slugify_segment
from db.collection_landing_pages import canonical_collection, DEFAULT_CATEGORY_L1
from db.master_taxonomy_hierarchy import NODE_COLLECTION, NODE_HUB
from db.master_taxonomy_sync import push_taxonomy_nodes_to_channels, upsert_taxonomy_node
from db.models import MasterProduct, MasterTaxonomyNode

MATCH_MATCHED = "matched"
MATCH_PROPOSED = "proposed"
MATCH_AMBIGUOUS = "ambiguous"
MATCH_NONE = "none"


@dataclass
class _ProductLike:
    sku: str
    category_l1: Optional[str]
    category_l2: Optional[str]
    collection: Optional[str]
    brand: Optional[str]
    product_family: Optional[str]


@dataclass
class _IntentCaches:
    """Hot-path caches shared across every SKU/path during proposal summarization."""

    alias_map: Dict[str, str]
    known_by_hub: Dict[str, Set[str]] = field(default_factory=dict)
    authoritative_by_key: Dict[Tuple[str, str, str], Optional[str]] = field(default_factory=dict)

    def known_names(self, session: Session, category_l1: Optional[str]) -> Set[str]:
        from db.collection_landing_pages import canonical_category
        from db.collection_path_guard import known_collection_names_for_hub

        hub = canonical_category(category_l1) or DEFAULT_CATEGORY_L1
        key = _norm_key(hub)
        cached = self.known_by_hub.get(key)
        if cached is not None:
            return cached
        names = known_collection_names_for_hub(session, category_l1=hub)
        self.known_by_hub[key] = names
        return names

    def authoritative_collection(
        self,
        session: Session,
        product: _ProductLike,
    ) -> Optional[str]:
        from db.collection_path_guard import resolve_authoritative_collection_name

        hub = product.category_l1 or DEFAULT_CATEGORY_L1
        cache_key = (
            str(product.sku or "").strip(),
            str(product.collection or "").strip(),
            str(hub).strip(),
        )
        if cache_key in self.authoritative_by_key:
            return self.authoritative_by_key[cache_key]
        known = self.known_names(session, hub)
        value = resolve_authoritative_collection_name(
            session,
            sku=product.sku,
            collection=product.collection,
            category_l1=hub,
            known_names=known,
        )
        self.authoritative_by_key[cache_key] = value
        return value


def summarize_import_taxonomy_intent(session: Session, records: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Analyze import rows for hierarchy node matches, proposals, and per-SKU placement."""
    from db.master_taxonomy_merge import load_path_alias_map

    index = _NodeIndex.load(session)
    caches = _IntentCaches(alias_map=load_path_alias_map(session))
    proposed_by_slug: Dict[str, Dict[str, Any]] = {}
    ambiguous: List[Dict[str, Any]] = []
    sku_rows: List[Dict[str, Any]] = []
    matched_counts = {"hub": 0, "category": 0, "collection": 0}

    for record in records:
        sku = str(record.get("sku") or "").strip()
        if not sku:
            continue
        product = _product_from_record(record)
        attrs = record.get("attributes") or {}
        paths = _paths_for_taxonomy_intent(session, product, attrs, caches=caches)
        row_matches: List[Dict[str, Any]] = []
        row_proposed: List[str] = []
        row_ambiguous: List[Dict[str, Any]] = []

        for path in paths:
            resolution = resolve_path_taxonomy(session, path, product, index=index, caches=caches)
            slug = resolution.get("path_slug") or ""
            if resolution["status"] == MATCH_MATCHED:
                matched_counts[resolution.get("node_kind") or "category"] = (
                    matched_counts.get(resolution.get("node_kind") or "category", 0) + 1
                )
                row_matches.append(resolution)
            elif resolution["status"] == MATCH_PROPOSED and slug:
                if _is_polluted_collection_proposal(session, slug, product=product, caches=caches):
                    continue
                row_proposed.append(slug)
                prop = proposed_by_slug.setdefault(
                    slug,
                    {
                        **(resolution.get("proposed") or {}),
                        "path_slug": slug,
                        "status": MATCH_PROPOSED,
                        "source_skus": [],
                    },
                )
                if sku not in prop["source_skus"]:
                    prop["source_skus"].append(sku)
            elif resolution["status"] == MATCH_AMBIGUOUS:
                row_ambiguous.append(resolution)
                ambiguous.append({**resolution, "sku": sku, "source_path": path})

        sku_rows.append(
            {
                "sku": sku,
                "category_l1": product.category_l1,
                "collection": product.collection,
                "matched_nodes": row_matches,
                "proposed_path_slugs": row_proposed,
                "ambiguous": row_ambiguous,
            }
        )

    return {
        "row_count": len(sku_rows),
        "matched_counts": matched_counts,
        "proposed_nodes": [
            node
            for node in proposed_by_slug.values()
            if not _is_polluted_collection_proposal(session, str(node.get("path_slug") or ""), caches=caches)
        ],
        "ambiguous": ambiguous[:100],
        "sku_samples": sku_rows[:50],
        "sku_rows": sku_rows,
    }


def resolve_path_taxonomy(
    session: Session,
    path: str,
    product: _ProductLike,
    *,
    index: Optional["_NodeIndex"] = None,
    caches: Optional[_IntentCaches] = None,
) -> Dict[str, Any]:
    """Resolve one category/collection path label to an existing node or proposal."""
    from db.master_taxonomy_merge import load_path_alias_map, resolve_canonical_path_slug

    index = index or _NodeIndex.load(session)
    caches = caches or _IntentCaches(alias_map=load_path_alias_map(session))
    path, path_slug = _canonical_path_for_product(session, path, product, caches=caches)
    if not path_slug:
        return {"status": MATCH_NONE, "path_slug": None, "name": path}

    # Prefer merged canonical path (e.g. plymouth-espresso → plymouth-espresso-maple).
    canonical_slug = resolve_canonical_path_slug(session, path_slug, alias_map=caches.alias_map) or path_slug
    if canonical_slug != path_slug:
        path_slug = canonical_slug
        canonical_node = index.by_slug.get(path_slug)
        if canonical_node and canonical_node.is_active:
            return _matched_node(canonical_node)
        path = (canonical_node.name if canonical_node and canonical_node.name else path_slug.replace("-", " "))

    if _is_polluted_collection_proposal(session, path_slug, product=product, caches=caches):
        return {"status": MATCH_NONE, "path_slug": path_slug, "name": path}

    # Prefer an active longer sibling (maple) over a short/inactive path (espresso).
    preferred = _prefer_active_extended_collection_node(index, path_slug)
    if preferred is not None:
        return _matched_node(preferred)

    node = index.by_slug.get(path_slug)
    if node and node.is_active:
        return _matched_node(node)

    leaf_key = _norm_key(_leaf_from_path(path))
    candidates = index.by_leaf_key.get(leaf_key, [])
    if len(candidates) == 1:
        return _matched_node(candidates[0])
    if len(candidates) > 1:
        return {
            "status": MATCH_AMBIGUOUS,
            "path_slug": path_slug,
            "name": path,
            "candidates": [_node_brief(n) for n in candidates],
        }

    parent_slug = _parent_path_slug(path_slug)
    parent_id = None
    if parent_slug and parent_slug in index.by_slug:
        parent_id = index.by_slug[parent_slug].id

    node_kind = _infer_node_kind(path, product)
    if node_kind == NODE_COLLECTION and parent_slug is None:
        return {"status": MATCH_NONE, "path_slug": path_slug, "name": path}

    display_name = path.replace("/", " / ")
    proposed = {
        "name": display_name,
        "path_slug": path_slug,
        "node_kind": node_kind,
        "parent_id": parent_id,
        "parent_path_slug": parent_slug,
    }
    # Inactive node with the same path_slug: propose reactivation (carry id),
    # but never when a preferred longer active sibling already exists.
    if node is not None and not node.is_active:
        preferred = _prefer_active_extended_collection_node(index, path_slug)
        if preferred is not None:
            return _matched_node(preferred)
        proposed["id"] = node.id
        proposed["name"] = node.name or display_name
        proposed["node_kind"] = node.node_kind or node_kind
        if node.parent_id:
            proposed["parent_id"] = node.parent_id
    return {
        "status": MATCH_PROPOSED,
        "path_slug": path_slug,
        "name": proposed["name"],
        "node_kind": proposed["node_kind"],
        "parent_path_slug": parent_slug,
        "parent_id": proposed.get("parent_id"),
        "id": proposed.get("id"),
        "proposed": proposed,
    }


def _prefer_active_extended_collection_node(
    index: "_NodeIndex",
    path_slug: str,
) -> Optional[MasterTaxonomyNode]:
    """Return the longest active sibling whose leaf extends this path leaf with '-'."""
    slug = normalize_plp_path(path_slug)
    if not slug or "/" not in slug:
        return None
    parent = _parent_path_slug(slug)
    leaf = slug.rsplit("/", 1)[-1]
    if not parent or not leaf:
        return None
    prefix = f"{parent}/{leaf}-"
    matches = [
        node
        for candidate_slug, node in index.by_slug.items()
        if node.is_active and candidate_slug.startswith(prefix)
    ]
    if not matches:
        return None
    return max(matches, key=lambda node: (len(node.path_slug or ""), node.path_slug or ""))


def apply_taxonomy_proposals(
    session: Session,
    proposals: Sequence[Dict[str, Any]],
    *,
    dry_run: bool = False,
    sync_channels: bool = False,
    magento_connection_id: Optional[int] = None,
    shopify_connection_id: Optional[int] = None,
    magento_api: Optional[Any] = None,
    shopify_client: Optional[Any] = None,
    shopify_shop_code: Optional[str] = None,
    allow_invent: Optional[bool] = None,
) -> Dict[str, Any]:
    """Create approved taxonomy nodes (parents first), optionally push to channels."""
    from db.taxonomy_invent_policy import can_auto_invent

    invent_ok = can_auto_invent(allow_invent=allow_invent)
    approved = [p for p in proposals if p.get("approved", True) is not False]
    ordered = sorted(approved, key=lambda row: str(row.get("path_slug") or "").count("/"))

    created: List[Dict[str, Any]] = []
    updated: List[Dict[str, Any]] = []
    skipped: List[Dict[str, Any]] = []
    node_ids: List[int] = []

    if dry_run:
        return {
            "dry_run": True,
            "would_create": len(ordered) if invent_ok else 0,
            "invent_frozen": not invent_ok,
            "proposals": ordered[:50],
        }

    if not invent_ok:
        return {
            "created_count": 0,
            "updated_count": 0,
            "skipped_count": len(ordered),
            "invent_frozen": True,
            "created": [],
            "updated": [],
            "skipped": [{"path_slug": p.get("path_slug"), "reason": "invent_frozen"} for p in ordered[:50]],
            "node_ids": [],
            "channel_push": None,
            "message": "Taxonomy invent is frozen; proposals were not materialized",
        }

    nodes_by_slug = {
        normalize_plp_path(node.path_slug): node
        for node in session.scalars(select(MasterTaxonomyNode)).all()
        if normalize_plp_path(node.path_slug)
    }

    for prop in ordered:
        parent_id = prop.get("parent_id")
        parent_slug = prop.get("parent_path_slug")
        if not parent_id and parent_slug:
            parent_row = nodes_by_slug.get(normalize_plp_path(str(parent_slug)))
            parent_id = parent_row.id if parent_row else None

        path_slug = normalize_plp_path(str(prop.get("path_slug") or ""))
        payload = {
            "name": prop.get("name") or _title_from_slug(path_slug),
            "path_slug": path_slug,
            "node_kind": prop.get("node_kind") or NODE_HUB,
            "parent_id": parent_id,
            "code": prop.get("code"),
            "sort_order": prop.get("sort_order"),
            "is_active": True,
        }
        if prop.get("id"):
            payload["id"] = prop["id"]
        elif path_slug:
            # Reuse inactive/existing nodes with the same natural key so pipeline
            # apply does not crash with "path_slug already in use".
            existing = nodes_by_slug.get(path_slug)
            if existing is not None:
                payload["id"] = existing.id
        result = upsert_taxonomy_node(session, payload, allow_invent=True)
        node_ids.append(int(result["node"]["id"]))
        node_row = session.get(MasterTaxonomyNode, int(result["node"]["id"]))
        if node_row is not None and path_slug:
            nodes_by_slug[path_slug] = node_row
        if result.get("created"):
            created.append(result["node"])
        else:
            updated.append(result["node"])

    session.flush()
    push_stats = None
    if sync_channels and node_ids:
        push_stats = push_taxonomy_nodes_to_channels(
            session,
            node_ids=node_ids,
            magento_connection_id=magento_connection_id,
            shopify_connection_id=shopify_connection_id,
            magento_api=magento_api,
            shopify_client=shopify_client,
            shopify_shop_code=shopify_shop_code,
            dry_run=False,
            create_missing=True,
        )

    return {
        "created_count": len(created),
        "updated_count": len(updated),
        "skipped_count": len(skipped),
        "invent_frozen": False,
        "created": created[:25],
        "updated": updated[:25],
        "node_ids": node_ids,
        "channel_push": push_stats,
    }


class _NodeIndex:
    def __init__(self) -> None:
        self.by_slug: Dict[str, MasterTaxonomyNode] = {}
        self.by_leaf_key: Dict[str, List[MasterTaxonomyNode]] = {}

    @classmethod
    def load(cls, session: Session) -> "_NodeIndex":
        index = cls()
        # Index all nodes by path_slug (including inactive) so proposal generation
        # does not re-propose slugs that already exist. Leaf matching stays active-only
        # to avoid ambiguous matches against retired names.
        for node in session.scalars(select(MasterTaxonomyNode)).all():
            slug = normalize_plp_path(node.path_slug)
            if not slug:
                continue
            existing = index.by_slug.get(slug)
            if existing is None or (not existing.is_active and node.is_active):
                index.by_slug[slug] = node
            if node.is_active:
                leaf = _norm_key(_leaf_title(node.name))
                index.by_leaf_key.setdefault(leaf, []).append(node)
        return index


def _as_master_product(product: _ProductLike) -> MasterProduct:
    row = MasterProduct()
    row.sku = product.sku
    row.category_l1 = product.category_l1
    row.collection = product.collection
    row.brand = product.brand
    row.product_family = product.product_family
    return row


def _authoritative_collection(
    session: Session,
    product: _ProductLike,
    *,
    caches: Optional[_IntentCaches] = None,
) -> Optional[str]:
    if caches is not None:
        return caches.authoritative_collection(session, product)
    from db.collection_path_guard import resolve_authoritative_collection_name

    return resolve_authoritative_collection_name(
        session,
        sku=product.sku,
        collection=product.collection,
        category_l1=product.category_l1,
    )


def _paths_for_taxonomy_intent(
    session: Session,
    product: _ProductLike,
    attrs: Optional[Dict[str, Any]] = None,
    *,
    caches: Optional[_IntentCaches] = None,
) -> List[str]:
    """Guarded catalog paths plus canonical hub/collection paths from the product row."""
    row = _as_master_product(product)
    coll = _authoritative_collection(session, product, caches=caches)
    if coll and coll != canonical_collection(product.collection):
        row.collection = coll
        product = _ProductLike(
            sku=product.sku,
            category_l1=product.category_l1,
            category_l2=product.category_l2,
            collection=coll,
            brand=product.brand,
            product_family=product.product_family,
        )

    known = None
    if caches is not None:
        known = caches.known_names(session, product.category_l1)
    paths = list(
        category_targets_for_product_guarded(
            session,
            row,
            attrs or {},
            known_names=known,
        )
    )
    if coll:
        hub = product.category_l1 or DEFAULT_CATEGORY_L1
        extra = f"{hub}/{coll}"
        slug = normalize_plp_path(extra)
        seen = {_norm_key(normalize_plp_path(path)) for path in paths}
        if slug and _norm_key(slug) not in seen and not _is_polluted_collection_proposal(
            session, slug, product=product, caches=caches
        ):
            paths.append(extra)
    return [path for path in paths if _norm(path) != _norm(ALL_PRODUCTS)]


def _canonical_path_for_product(
    session: Session,
    path: str,
    product: _ProductLike,
    *,
    caches: Optional[_IntentCaches] = None,
) -> tuple[str, str]:
    """Prefer full canonical collection names under the hub."""
    coll = _authoritative_collection(session, product, caches=caches) or canonical_collection(
        product.collection or ""
    )
    hub = product.category_l1 or DEFAULT_CATEGORY_L1
    leaf = _leaf_from_path(path)
    normalized_path = str(path or "").strip()

    if coll:
        if leaf and _norm_key(leaf) != _norm_key(coll):
            if _norm_key(coll).endswith(_norm_key(leaf)) and len(coll) > len(leaf):
                normalized_path = f"{hub}/{coll}"
        elif "/" not in normalized_path and _norm_key(normalized_path) != _norm_key(hub):
            normalized_path = f"{hub}/{coll}"
        elif normalized_path == coll or _norm_key(normalized_path) == _norm_key(coll):
            normalized_path = f"{hub}/{coll}"

    slug = _path_label_to_slug(normalized_path)
    return normalized_path, slug


def _is_polluted_collection_proposal(
    session: Session,
    path_slug: str,
    *,
    product: Optional[_ProductLike] = None,
    caches: Optional[_IntentCaches] = None,
) -> bool:
    from db.collection_path_guard import is_polluted_collection_taxonomy_slug

    category_l1 = (product.category_l1 if product else None) or DEFAULT_CATEGORY_L1
    known = caches.known_names(session, category_l1) if caches is not None else None
    return is_polluted_collection_taxonomy_slug(
        session,
        path_slug,
        category_l1=category_l1,
        known_names=known,
    )


def _product_from_record(record: Dict[str, Any]) -> _ProductLike:
    return _ProductLike(
        sku=str(record.get("sku") or "").strip(),
        category_l1=_clean(record.get("category_l1")),
        category_l2=_clean(record.get("category_l2")),
        collection=_clean(record.get("collection")),
        brand=_clean(record.get("brand")),
        product_family=_clean(record.get("product_family")),
    )


def _infer_node_kind(path: str, product: _ProductLike) -> str:
    coll = canonical_collection(product.collection or "")
    if "/" not in path:
        hub = product.category_l1 or DEFAULT_CATEGORY_L1
        if _norm_key(path) == _norm_key(hub):
            return NODE_HUB
        return NODE_HUB
    if coll and path.endswith(coll):
        return NODE_COLLECTION
    if coll and _norm_key(coll) == _norm_key(_leaf_from_path(path)):
        return NODE_COLLECTION
    return "category"


def _path_label_to_slug(path: str) -> str:
    parts = [p.strip() for p in str(path or "").split("/") if p.strip()]
    if not parts:
        return ""
    return normalize_plp_path("/".join(slugify_segment(part) for part in parts))


def _parent_path_slug(path_slug: str) -> Optional[str]:
    parts = [p for p in normalize_plp_path(path_slug).split("/") if p]
    if len(parts) <= 1:
        return None
    return "/".join(parts[:-1])


def _leaf_from_path(path: str) -> str:
    parts = [p.strip() for p in str(path or "").split("/") if p.strip()]
    return parts[-1] if parts else str(path or "").strip()


def _title_from_slug(slug: str) -> str:
    return " / ".join(part.replace("-", " ").title() for part in slug.split("/") if part)


def _matched_node(node: MasterTaxonomyNode) -> Dict[str, Any]:
    return {
        "status": MATCH_MATCHED,
        "id": node.id,
        "path_slug": node.path_slug,
        "name": node.name,
        "node_kind": node.node_kind,
        "parent_id": node.parent_id or 0,
    }


def _node_brief(node: MasterTaxonomyNode) -> Dict[str, Any]:
    return {"id": node.id, "path_slug": node.path_slug, "name": node.name, "node_kind": node.node_kind}


def _leaf_title(name: str) -> str:
    text = str(name or "").strip()
    if " / " in text:
        return text.split(" / ")[-1].strip()
    return text


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


def _norm(value: Optional[str]) -> str:
    return re.sub(r"\s+", " ", str(value or "").strip().lower())


def _clean(value: Any) -> Optional[str]:
    text = str(value or "").strip()
    return text or None