classifier.py

"""The tier classifier and the facet registry — the single source of truth.

Everything in this package flows from one function, :func:`classify_request`,
and one piece of state, the :class:`FacetRegistry`. Given a category and a set
of query parameters, the classifier decides which of the indexing tiers a URL
belongs to and emits the matching signals (canonical URL + meta-robots string).

The five-tier model
-------------------

* **Tier 1 — INDEXABLE.** Base category pages plus a *curated* set of facet
  combinations (single, occasionally two facets) that map to real search
  demand. Self-canonical, ``index, follow``, belongs in the sitemap.
* **Tier 2 — CANONICALIZED.** Presentation-only variants (sort / view /
  per-page). Crawlable; emit ``rel="canonical"`` to the underlying page so
  Google consolidates them.
* **Tier 3 — NOINDEX.** Uncurated facet combinations (and, by default,
  pagination beyond page 1). Crawlable but ``noindex, follow``. **Never**
  cross-canonical to a different URL — canonical is ``None`` here.
* **Tier 4 — BLOCKED.** Tracking / session params, arbitrary range inputs,
  internal search. Blocked in ``robots.txt``; no indexing value to preserve.
* **Tier 5 — not exposed as URLs.** Pure client UI state (hover, compare-mode
  toggles). Never reaches the address bar, so the classifier never sees it and
  emits nothing — it is documented here, not represented in the enum.

Design notes
------------

* **Parameter-driven, not URL-pattern-driven.** The classifier inspects the
  *parameters*, not a regex over the path. Adding a new facet defaults to
  Tier 3 (noindex) — promotion to Tier 1 is an explicit registry opt-in, so a
  new facet can never silently promote thousands of URLs.
* **Nullable canonical.** ``noindex`` + a cross-canonical to a *different* URL
  is contradictory and Google may drop both signals. Tier-3 pages therefore
  return ``canonical=None`` (the return type is ``str | None``).
* **Consistent normalization.** Facet values are normalized (lowercase + trim)
  on both registry write and classification read, so ``{"color": "Black"}``
  matches ``?color=black``.
"""

from __future__ import annotations

import enum
from collections.abc import Iterable, Iterator, Mapping, Sequence
from typing import NamedTuple
from urllib.parse import parse_qsl, quote, urlencode

from .normalize import normalize_key, normalize_value

__all__ = [
    "IndexingTier",
    "Classification",
    "FacetRegistry",
    "classify_request",
    "register_indexable_combination",
    "default_registry",
    "reset_default_registry",
    "build_category_url",
    "DEFAULT_BASE_URL",
    "TRACKING_PARAMS",
    "SESSION_PARAMS",
    "SEARCH_PARAMS",
    "RANGE_PARAMS",
    "RANGE_SUFFIXES",
    "TRANSPARENT_PARAMS",
    "PAGINATION_PARAMS",
    "is_blocked_param",
]

DEFAULT_BASE_URL = "https://example.com"


class IndexingTier(enum.Enum):
    """The four URL-emitting tiers. Tier 5 produces no URL, so it is absent."""

    INDEXABLE = "indexable"  # Tier 1
    CANONICALIZED = "canonicalized"  # Tier 2
    NOINDEX = "noindex"  # Tier 3
    BLOCKED = "blocked"  # Tier 4

    def __str__(self) -> str:  # pragma: no cover - cosmetic
        return self.value


class Classification(NamedTuple):
    """The classifier's result. Unpacks as ``(tier, canonical, meta_robots)``."""

    tier: IndexingTier
    canonical: str | None
    meta_robots: str


# --- Parameter taxonomy -----------------------------------------------------
#
# Parameters fall into five buckets. The first three (tracking/session,
# range, search) are Tier-4 "blocked" families. Transparent params are
# presentation-only (Tier 2). Pagination is handled specially. Anything else
# is treated as a facet.

TRACKING_PARAMS = frozenset(
    {
        "gclid",
        "fbclid",
        "msclkid",
        "yclid",
        "mc_cid",
        "mc_eid",
        "igshid",
        "_ga",
        "ref",
        "referrer",
        "affiliate",
        "aff_id",
    }
)

SESSION_PARAMS = frozenset(
    {
        "sessionid",
        "session_id",
        "sid",
        "sessid",
        "phpsessid",
        "jsessionid",
        "aspsessionid",
    }
)

SEARCH_PARAMS = frozenset({"q", "query", "search", "keyword", "keywords"})

# Explicit range params plus a suffix rule that catches arbitrary range inputs
# such as ``rating_min`` / ``length_to``.
RANGE_PARAMS = frozenset({"price_min", "price_max", "min_price", "max_price", "price"})
RANGE_SUFFIXES = ("_min", "_max", "_from", "_to")

# Presentation-only parameters: they change how results are shown, not which
# results, so every variant should consolidate onto the underlying page.
TRANSPARENT_PARAMS = frozenset({"sort", "view", "per_page"})

PAGINATION_PARAMS = frozenset({"page"})


def _is_tracking(key: str) -> bool:
    return key in TRACKING_PARAMS or key in SESSION_PARAMS or key.startswith("utm_")


def _is_range(key: str) -> bool:
    return key in RANGE_PARAMS or key.endswith(RANGE_SUFFIXES)


def _is_search(key: str) -> bool:
    return key in SEARCH_PARAMS


def is_blocked_param(key: str) -> bool:
    """Return ``True`` if a (normalized) param key belongs to a Tier-4 family."""
    key = normalize_key(key)
    return _is_tracking(key) or _is_range(key) or _is_search(key)


def _is_transparent(key: str) -> bool:
    return key in TRANSPARENT_PARAMS


def _is_pagination(key: str) -> bool:
    return key in PAGINATION_PARAMS


# --- URL construction -------------------------------------------------------


def build_category_url(
    base_url: str,
    category: str,
    facets: Mapping[str, Sequence[str]] | None = None,
) -> str:
    """Build an absolute, properly URL-encoded category URL.

    ``facets`` maps a facet key to one or more values. Pairs are emitted in
    canonical (sorted) order and encoded via :func:`urllib.parse.urlencode`, so
    the result is stable regardless of the input ordering.
    """
    path = f"{base_url.rstrip('/')}/category/{quote(category)}"
    if not facets:
        return path
    pairs = sorted((key, value) for key, values in facets.items() for value in values)
    return f"{path}?{urlencode(pairs)}"


# --- Facet registry ---------------------------------------------------------


def _facet_signature(facets: Mapping[str, Sequence[str]]) -> frozenset[tuple[str, str]]:
    """Order-independent identity of a facet selection.

    A multi-valued selection (``?color=black&color=red``) yields multiple
    members and so will never match a curated single-value combination — which
    is exactly what we want: multi-select defaults to Tier 3.
    """
    return frozenset(
        (key, value) for key, values in facets.items() for value in values
    )


def _normalize_facets(
    facets: Mapping[str, object] | None,
) -> dict[str, list[str]]:
    """Normalize a facet mapping to ``{key: [normalized values...]}``.

    Accepts string or iterable-of-string values, applies the shared key/value
    normalization, drops empties, and de-duplicates values within a facet.
    """
    out: dict[str, list[str]] = {}
    if not facets:
        return out
    for raw_key, raw_value in facets.items():
        key = normalize_key(str(raw_key))
        if isinstance(raw_value, (list, tuple, set)):
            raw_values: Iterable[object] = raw_value
        else:
            raw_values = [raw_value]
        seen: list[str] = []
        for item in raw_values:
            value = normalize_value(str(item))
            if value and value not in seen:
                seen.append(value)
        if seen:
            out[key] = sorted(seen)
    return out


class FacetRegistry:
    """The curated set of Tier-1 combinations — the single source of truth.

    The classifier reads it to decide whether a facet combination is indexable;
    the sitemap generator reads the *same* registry to emit exactly those URLs.
    Default behaviour is Tier 3 (noindex): a combination is indexable only if it
    was explicitly registered here.
    """

    def __init__(self) -> None:
        # Categories whose base page is indexable (every registered category).
        self._categories: list[str] = []
        # Per category: the curated facet combinations, normalized.
        self._combos: dict[str, list[dict[str, list[str]]]] = {}
        # Per category: signatures for O(1) membership tests.
        self._signatures: dict[str, set[frozenset[tuple[str, str]]]] = {}

    def register_indexable_combination(
        self,
        category: str,
        facets: Mapping[str, object] | None = None,
    ) -> None:
        """Promote a combination to Tier 1.

        Registering with no facets (or ``None``) marks the *base category page*
        as indexable so it appears in the sitemap. Registering with facets adds
        a curated combination. Values are normalized on write to match the
        normalization applied at classification time.
        """
        category = normalize_value(category)
        if category not in self._categories:
            self._categories.append(category)
            self._combos.setdefault(category, [])
            self._signatures.setdefault(category, set())

        normalized = _normalize_facets(facets)
        if not normalized:
            return  # base-page-only registration

        signature = _facet_signature(normalized)
        if signature not in self._signatures[category]:
            self._signatures[category].add(signature)
            self._combos[category].append(normalized)

    def is_indexable(
        self, category: str, facets: Mapping[str, Sequence[str]]
    ) -> bool:
        """Return ``True`` if this exact facet combination is curated Tier 1."""
        category = normalize_value(category)
        signature = _facet_signature(facets)
        return signature in self._signatures.get(category, set())

    def categories(self) -> list[str]:
        """The registered categories (base pages), in registration order."""
        return list(self._categories)

    def combinations(self, category: str) -> list[dict[str, list[str]]]:
        """The curated facet combinations for a category, in registration order."""
        return [dict(combo) for combo in self._combos.get(normalize_value(category), [])]

    def iter_indexable_urls(self, base_url: str) -> Iterator[str]:
        """Yield every Tier-1 URL: each base category page then its combos."""
        for category in self._categories:
            yield build_category_url(base_url, category)
            for combo in self._combos[category]:
                yield build_category_url(base_url, category, combo)


# A module-level default registry for convenience. Inject a dedicated registry
# into ``classify_request`` (and the sitemap generator) to keep things isolated.
default_registry = FacetRegistry()


def register_indexable_combination(
    category: str, facets: Mapping[str, object] | None = None
) -> None:
    """Register a Tier-1 combination on the module-level default registry."""
    default_registry.register_indexable_combination(category, facets)


def reset_default_registry() -> None:
    """Clear the module-level default registry (handy for tests/demos)."""
    global default_registry
    default_registry = FacetRegistry()


# --- The classifier ---------------------------------------------------------


def _coerce_params(
    query_params: Mapping[str, object] | Sequence[tuple[str, object]] | str | None,
) -> dict[str, list[str]]:
    """Coerce assorted query-param inputs into ``{key: [values...]}``.

    Accepts a query string, a mapping (values may be scalars or lists), or a
    sequence of pairs. Keys/values are kept *raw* here; normalization happens in
    the classifier so blocked-param detection sees the original casing too.
    """
    if query_params is None:
        return {}
    if isinstance(query_params, str):
        pairs: list[tuple[str, str]] = parse_qsl(query_params, keep_blank_values=True)
    elif isinstance(query_params, Mapping):
        pairs = []
        for key, value in query_params.items():
            if isinstance(value, (list, tuple, set)):
                pairs.extend((str(key), str(v)) for v in value)
            else:
                pairs.append((str(key), str(value)))
    else:
        pairs = [(str(k), str(v)) for k, v in query_params]

    out: dict[str, list[str]] = {}
    for key, value in pairs:
        out.setdefault(key, []).append(value)
    return out


def _page_number(pagination: Mapping[str, Sequence[str]]) -> int:
    """Extract the page number, defaulting to 1 for absent/invalid values."""
    values = pagination.get("page")
    if not values:
        return 1
    try:
        return int(values[0])
    except (TypeError, ValueError):
        return 1


def classify_request(
    category: str,
    query_params: Mapping[str, object] | Sequence[tuple[str, object]] | str | None = None,
    *,
    base_url: str = DEFAULT_BASE_URL,
    registry: FacetRegistry | None = None,
) -> Classification:
    """Classify a faceted request into a tier and emit the matching signals.

    Returns a :class:`Classification` (a ``(tier, canonical, meta_robots)``
    NamedTuple). ``canonical`` is ``None`` for Tier 3/4 — see the module
    docstring for why a noindex page must not cross-canonicalize.
    """
    registry = registry if registry is not None else default_registry
    category_norm = normalize_value(category)
    params = _coerce_params(query_params)

    blocked: dict[str, list[str]] = {}
    transparent: dict[str, list[str]] = {}
    pagination: dict[str, list[str]] = {}
    facets: dict[str, list[str]] = {}

    for raw_key, raw_values in params.items():
        key = normalize_key(raw_key)
        if is_blocked_param(key):
            # Presence alone is disqualifying; value cleanliness is irrelevant.
            blocked[key] = list(raw_values)
            continue
        clean = [v for v in (normalize_value(x) for x in raw_values) if v != ""]
        if not clean:
            continue  # empty value — dropped, contributes nothing
        if _is_transparent(key):
            transparent[key] = clean
        elif _is_pagination(key):
            pagination[key] = clean
        else:
            facets[key] = clean

    # Tier 4 — any tracking / session / range / search param blocks the URL.
    if blocked:
        return Classification(IndexingTier.BLOCKED, None, "noindex, nofollow")

    # Pagination beyond page 1 -> Tier 3 (crawl deeper, but don't index).
    if pagination and _page_number(pagination) > 1:
        return Classification(IndexingTier.NOINDEX, None, "noindex, follow")

    # A leftover ``page=1`` is a duplicate of the unpaginated page: treat it as
    # a presentation variant so it canonicalizes.
    has_presentation = bool(transparent) or bool(pagination)

    if facets:
        if registry.is_indexable(category_norm, facets):
            self_url = build_category_url(base_url, category_norm, facets)
            if has_presentation:
                # e.g. ?color=black&sort=price — a sort variant of a Tier-1
                # facet page. Consolidate onto the clean facet URL.
                return Classification(
                    IndexingTier.CANONICALIZED, self_url, "index, follow"
                )
            return Classification(IndexingTier.INDEXABLE, self_url, "index, follow")
        # Uncurated facet combination -> Tier 3. No cross-canonical (None).
        return Classification(IndexingTier.NOINDEX, None, "noindex, follow")

    base = build_category_url(base_url, category_norm)
    if has_presentation:
        # Presentation-only variant of the base category page -> canonical to it.
        return Classification(IndexingTier.CANONICALIZED, base, "index, follow")

    # Bare base category page -> Tier 1, self-canonical.
    return Classification(IndexingTier.INDEXABLE, base, "index, follow")