"""Curated Start Shopping vocabulary for shopping_collection / shopping_l1 / shopping_l2.

A value must be both:

1. curated by an operator — a manual taxonomy assignment or a saved Start Shopping
   config, and
2. structurally real — producible from an active master_taxonomy_node lineage.

Curation alone is not enough: manual assignments can point at taxonomy paths that no
longer exist (kitchen-cabinets/mouldings, kitchen-cabinets/tall,
wall-cabinets/double-door-wall-cabinet), and the shopping_l1/l2 columns derived from
those dead paths are exactly the duplicate variants that reached Magento.

Path inference (channel projection) and Magento option seeding both validate against
this vocabulary, so the taxonomy freeze cannot be bypassed by derived values.

An empty vocabulary means "nothing curated yet" and disables gating, so fresh
databases and focused unit tests keep their previous behaviour.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Dict, Iterable, List, Mapping, Optional, Protocol, Sequence, Set, Tuple

from sqlalchemy import select
from sqlalchemy.orm import Session

from channel.url_canonical import normalize_plp_path
from db.models import (
    CollectionLandingPage,
    ManualTaxonomyAssignment,
    ManualTaxonomyAssignmentCollection,
    MasterTaxonomyNode,
)
from db.source_imports import normalize_column_key

SHOPPING_TAXONOMY_CODES: Tuple[str, ...] = (
    "shopping_collection",
    "shopping_l1",
    "shopping_l2",
)


def is_shopping_taxonomy_code(attribute_code: Optional[str]) -> bool:
    return normalize_column_key(attribute_code or "") in set(SHOPPING_TAXONOMY_CODES)


def normalize_shopping_value(value: Optional[str]) -> str:
    """Comparison key: slug path form, so 'Panels and Fillers' == 'panels-and-fillers'."""
    return normalize_plp_path(str(value or ""))


@dataclass(frozen=True)
class ShoppingTaxonomyVocabulary:
    """Allowed shopping_* values keyed by canonical attribute code.

    unbacked_by_code holds curated values rejected for having no active taxonomy node,
    so callers can report why a value stopped being publishable.
    """

    values_by_code: Mapping[str, frozenset]
    unbacked_by_code: Mapping[str, frozenset] = field(default_factory=dict)

    @property
    def is_empty(self) -> bool:
        return not any(self.values_by_code.get(code) for code in SHOPPING_TAXONOMY_CODES)

    def unbacked_values(self, attribute_code: str) -> List[str]:
        return sorted(self.unbacked_by_code.get(normalize_column_key(attribute_code)) or set())

    def has_code(self, attribute_code: str) -> bool:
        return bool(self.values_by_code.get(normalize_column_key(attribute_code)))

    def allows(self, attribute_code: str, value: Optional[str]) -> bool:
        """True when the value is curated, or when nothing is curated for that code."""
        code = normalize_column_key(attribute_code)
        allowed = self.values_by_code.get(code)
        if not allowed:
            return True
        key = normalize_shopping_value(value)
        if not key:
            return True
        return key in allowed

    def curated_values(self, attribute_code: str) -> List[str]:
        return sorted(self.values_by_code.get(normalize_column_key(attribute_code)) or set())


class ShoppingVocabularySource(Protocol):
    """Read model contributing curated shopping_* values."""

    def curated_shopping_values(self) -> Dict[str, Set[str]]: ...


class ManualTaxonomyVocabularySource:
    """Values explicitly curated on manual taxonomy assignment rows."""

    def __init__(self, session: Session) -> None:
        self._session = session

    def curated_shopping_values(self) -> Dict[str, Set[str]]:
        out: Dict[str, Set[str]] = {code: set() for code in SHOPPING_TAXONOMY_CODES}
        rows = self._session.execute(
            select(ManualTaxonomyAssignment.shopping_l1, ManualTaxonomyAssignment.shopping_l2)
            .where(ManualTaxonomyAssignment.assignment_status == "active")
        ).all()
        for shopping_l1, shopping_l2 in rows:
            _add(out, "shopping_l1", shopping_l1)
            _add(out, "shopping_l2", shopping_l2)

        collection_rows = self._session.scalars(
            select(ManualTaxonomyAssignmentCollection.shopping_collection)
            .where(ManualTaxonomyAssignmentCollection.is_active.is_(True))
        ).all()
        for shopping_collection in collection_rows:
            _add(out, "shopping_collection", shopping_collection)
        return out


class StartShoppingConfigVocabularySource:
    """Values an operator selected in a collection's saved Start Shopping config."""

    def __init__(self, session: Session) -> None:
        self._session = session

    def curated_shopping_values(self) -> Dict[str, Set[str]]:
        from db.collection_landing_pages import (
            _normalize_start_shopping_config,
            _relative_shopping_collection_value,
            _relative_shopping_l2_value,
            _start_shopping_l1_value,
        )

        out: Dict[str, Set[str]] = {code: set() for code in SHOPPING_TAXONOMY_CODES}
        rows = self._session.execute(
            select(CollectionLandingPage.path_slug, CollectionLandingPage.start_shopping_config)
            .where(CollectionLandingPage.is_active.is_(True))
        ).all()
        for path_slug, raw_config in rows:
            _add(out, "shopping_collection", _relative_shopping_collection_value(path_slug))
            config = _normalize_start_shopping_config(raw_config)
            for item in config.get("items") or []:
                l1_value = _start_shopping_l1_value(item)
                _add(out, "shopping_l1", l1_value)
                for l2_path in item.get("l2_path_slugs") or []:
                    _add(out, "shopping_l2", _relative_shopping_l2_value(l2_path))
                for candidate in item.get("l2_values") or []:
                    slug = str((candidate or {}).get("slug") or "").strip()
                    if slug and l1_value:
                        _add(out, "shopping_l2", f"{l1_value}/{slug}")
        return out


class ShoppingStructureSource(Protocol):
    """Values the live taxonomy structure can actually produce."""

    def structural_shopping_values(self) -> Dict[str, Set[str]]: ...


class ActiveTaxonomyNodeStructureSource:
    """shopping_* values derivable from active master_taxonomy_node lineages."""

    def __init__(self, session: Session) -> None:
        self._session = session

    def structural_shopping_values(self) -> Dict[str, Set[str]]:
        from db.collection_landing_pages import (
            _relative_shopping_collection_value,
            _relative_shopping_l1_value,
            _relative_shopping_l2_value,
        )

        out: Dict[str, Set[str]] = {code: set() for code in SHOPPING_TAXONOMY_CODES}
        path_slugs = self._session.scalars(
            select(MasterTaxonomyNode.path_slug).where(MasterTaxonomyNode.is_active.is_(True))
        ).all()
        for raw_path in path_slugs:
            path = normalize_plp_path(str(raw_path or ""))
            depth = len([part for part in path.split("/") if part])
            if depth == 2:
                _add(out, "shopping_collection", _relative_shopping_collection_value(path))
                _add(out, "shopping_l1", _relative_shopping_l1_value(path))
            elif depth >= 3:
                _add(out, "shopping_l2", _relative_shopping_l2_value(path))
        return out


def _add(bucket: Dict[str, Set[str]], code: str, value: Optional[str]) -> None:
    key = normalize_shopping_value(value)
    if key:
        bucket.setdefault(code, set()).add(key)


def load_shopping_taxonomy_vocabulary(
    session: Session,
    *,
    sources: Optional[Sequence[ShoppingVocabularySource]] = None,
    structure: Optional[ShoppingStructureSource] = None,
    require_active_node: bool = True,
) -> ShoppingTaxonomyVocabulary:
    resolved = sources if sources is not None else (
        ManualTaxonomyVocabularySource(session),
        StartShoppingConfigVocabularySource(session),
    )
    merged: Dict[str, Set[str]] = {code: set() for code in SHOPPING_TAXONOMY_CODES}
    for source in resolved:
        try:
            contributed = source.curated_shopping_values() or {}
        except Exception:
            # A missing table on an older deploy must not turn the gate into a hard failure.
            continue
        for code, values in contributed.items():
            merged.setdefault(normalize_column_key(code), set()).update(values)

    unbacked: Dict[str, Set[str]] = {}
    if require_active_node:
        structure_source = structure if structure is not None else ActiveTaxonomyNodeStructureSource(session)
        try:
            structural = structure_source.structural_shopping_values() or {}
        except Exception:
            structural = {}
        for code, curated in list(merged.items()):
            allowed = structural.get(code)
            # No structural values for a code means the taxonomy cannot vouch either way;
            # fall back to curation alone rather than dropping every value.
            if not allowed or not curated:
                continue
            rejected = curated - allowed
            if rejected:
                unbacked[code] = rejected
                merged[code] = curated & allowed

    return ShoppingTaxonomyVocabulary(
        values_by_code={code: frozenset(values) for code, values in merged.items()},
        unbacked_by_code={code: frozenset(values) for code, values in unbacked.items()},
    )


def partition_shopping_labels(
    vocabulary: ShoppingTaxonomyVocabulary,
    attribute_code: str,
    labels: Iterable[str],
) -> Tuple[List[str], List[str]]:
    """Split candidate option labels into (curated, uncurated) preserving order."""
    curated: List[str] = []
    uncurated: List[str] = []
    for label in labels:
        text = str(label or "").strip()
        if not text:
            continue
        if vocabulary.allows(attribute_code, text):
            curated.append(text)
        else:
            uncurated.append(text)
    return curated, uncurated