search/es_query_builder.py

"""
Elasticsearch query builder.
Converts parsed queries and search parameters into ES DSL queries.
Simplified architecture:
- filters and (text_recall or embedding_recall)
- function_score wrapper for boosting fields
"""
from dataclasses import dataclass
from typing import Dict, Any, List, Optional, Tuple
import numpy as np
from config import FunctionScoreConfig
from query.keyword_extractor import KEYWORDS_QUERY_BASE_KEY
class ESQueryBuilder:
    """Builds Elasticsearch DSL queries."""
    def __init__(
        self,
        match_fields: List[str],
        field_boosts: Optional[Dict[str, float]] = None,
        multilingual_fields: Optional[List[str]] = None,
        shared_fields: Optional[List[str]] = None,
        core_multilingual_fields: Optional[List[str]] = None,
        text_embedding_field: Optional[str] = None,
        image_embedding_field: Optional[str] = None,
        source_fields: Optional[List[str]] = None,
        function_score_config: Optional[FunctionScoreConfig] = None,
        default_language: str = "en",
        knn_text_boost: float = 20.0,
        knn_image_boost: float = 20.0,
        knn_text_k: int = 120,
        knn_text_num_candidates: int = 400,
        knn_text_k_long: int = 160,
        knn_text_num_candidates_long: int = 500,
        knn_image_k: int = 120,
        knn_image_num_candidates: int = 400,
        base_minimum_should_match: str = "66%",
        translation_minimum_should_match: str = "66%",
        keywords_minimum_should_match: str = "60%",
        translation_boost: float = 0.4,
        tie_breaker_base_query: float = 0.9,
        best_fields_boosts: Optional[Dict[str, float]] = None,
        best_fields_clause_boost: float = 2.0,
        phrase_field_boosts: Optional[Dict[str, float]] = None,
        phrase_match_base_fields: Optional[Tuple[str, ...]] = None,
        phrase_match_slop: int = 0,
        phrase_match_tie_breaker: float = 0.0,
        phrase_match_boost: float = 3.0,
    ):
        """
        Initialize query builder.
        Multi-language search (translation-based cross-language recall) is always enabled:
        queries are matched against detected-language and translated target-language clauses.
        Args:
            match_fields: Fields to search for text matching
            text_embedding_field: Field name for text embeddings
            image_embedding_field: Field name for image embeddings
            source_fields: Fields to return in search results (_source includes)
            function_score_config: Function score configuration
            default_language: Default language to use when detection fails or returns "unknown"
            knn_text_boost: Boost for text-embedding KNN clause
            knn_image_boost: Boost for image-embedding KNN clause
        """
        self.match_fields = match_fields
        self.field_boosts = field_boosts or {}
        self.multilingual_fields = multilingual_fields or []
        self.shared_fields = shared_fields or []
        self.core_multilingual_fields = core_multilingual_fields or []
        self.text_embedding_field = text_embedding_field
        self.image_embedding_field = image_embedding_field
        self.source_fields = source_fields
        self.function_score_config = function_score_config
        self.default_language = default_language
        self.knn_text_boost = float(knn_text_boost)
        self.knn_image_boost = float(knn_image_boost)
        self.knn_text_k = int(knn_text_k)
        self.knn_text_num_candidates = int(knn_text_num_candidates)
        self.knn_text_k_long = int(knn_text_k_long)
        self.knn_text_num_candidates_long = int(knn_text_num_candidates_long)
        self.knn_image_k = int(knn_image_k)
        self.knn_image_num_candidates = int(knn_image_num_candidates)
        self.base_minimum_should_match = base_minimum_should_match
        self.translation_minimum_should_match = translation_minimum_should_match
        self.keywords_minimum_should_match = str(keywords_minimum_should_match)
        self.translation_boost = float(translation_boost)
        self.tie_breaker_base_query = float(tie_breaker_base_query)
        default_best_fields = {
            base: self._get_field_boost(base)
            for base in self.core_multilingual_fields
            if base in self.multilingual_fields
        }
        self.best_fields_boosts = {
            str(base): float(boost)
            for base, boost in (best_fields_boosts or default_best_fields).items()
        }
        self.best_fields_clause_boost = float(best_fields_clause_boost)
        default_phrase_base_fields = tuple(phrase_match_base_fields or ("title", "qanchors"))
        default_phrase_fields = {
            base: self._get_field_boost(base)
            for base in default_phrase_base_fields
            if base in self.multilingual_fields
        }
        self.phrase_field_boosts = {
            str(base): float(boost)
            for base, boost in (phrase_field_boosts or default_phrase_fields).items()
        }
        self.phrase_match_slop = int(phrase_match_slop)
        self.phrase_match_tie_breaker = float(phrase_match_tie_breaker)
        self.phrase_match_boost = float(phrase_match_boost)
    @dataclass(frozen=True)
    class KNNClausePlan:
        field: str
        boost: float
        k: Optional[int] = None
        num_candidates: Optional[int] = None
        nested_path: Optional[str] = None
    @staticmethod
    def _vector_to_list(vector: Any) -> List[float]:
        if vector is None:
            return []
        if hasattr(vector, "tolist"):
            values = vector.tolist()
        else:
            values = list(vector)
        return [float(v) for v in values]
    @staticmethod
    def _query_token_count(parsed_query: Optional[Any]) -> int:
        if parsed_query is None:
            return 0
        query_tokens = getattr(parsed_query, "query_tokens", None) or []
        return len(query_tokens)
    def get_text_knn_plan(self, parsed_query: Optional[Any] = None) -> Optional[KNNClausePlan]:
        if not self.text_embedding_field:
            return None
        boost = self.knn_text_boost
        final_knn_k = self.knn_text_k
        final_knn_num_candidates = self.knn_text_num_candidates
        if self._query_token_count(parsed_query) >= 5:
            final_knn_k = self.knn_text_k_long
            final_knn_num_candidates = self.knn_text_num_candidates_long
            boost = self.knn_text_boost * 1.4
        return self.KNNClausePlan(
            field=str(self.text_embedding_field),
            boost=float(boost),
            k=int(final_knn_k),
            num_candidates=int(final_knn_num_candidates),
        )
    def get_image_knn_plan(self) -> Optional[KNNClausePlan]:
        if not self.image_embedding_field:
            return None
        nested_path, _, _ = str(self.image_embedding_field).rpartition(".")
        return self.KNNClausePlan(
            field=str(self.image_embedding_field),
            boost=float(self.knn_image_boost),
            k=int(self.knn_image_k),
            num_candidates=int(self.knn_image_num_candidates),
            nested_path=nested_path or None,
        )
    def build_text_knn_clause(
        self,
        query_vector: Any,
        *,
        parsed_query: Optional[Any] = None,
        query_name: str = "knn_query",
    ) -> Optional[Dict[str, Any]]:
        plan = self.get_text_knn_plan(parsed_query)
        if plan is None or query_vector is None:
            return None
        return {
            "knn": {
                "field": plan.field,
                "query_vector": self._vector_to_list(query_vector),
                "k": plan.k,
                "num_candidates": plan.num_candidates,
                "boost": plan.boost,
                "_name": query_name,
            }
        }
    def build_image_knn_clause(
        self,
        image_query_vector: Any,
        *,
        query_name: str = "image_knn_query",
    ) -> Optional[Dict[str, Any]]:
        plan = self.get_image_knn_plan()
        if plan is None or image_query_vector is None:
            return None
        image_knn_query = {
            "field": plan.field,
            "query_vector": self._vector_to_list(image_query_vector),
            "k": plan.k,
            "num_candidates": plan.num_candidates,
            "boost": plan.boost,
        }
        if plan.nested_path:
            return {
                "nested": {
                    "path": plan.nested_path,
                    "_name": query_name,
                    "query": {"knn": image_knn_query},
                    "score_mode": "max",
                }
            }
        return {
            "knn": {
                **image_knn_query,
                "_name": query_name,
            }
        }
    def build_exact_text_knn_rescore_clause(
        self,
        query_vector: Any,
        *,
        parsed_query: Optional[Any] = None,
        query_name: str = "exact_text_knn_query",
    ) -> Optional[Dict[str, Any]]:
        plan = self.get_text_knn_plan(parsed_query)
        if plan is None or query_vector is None:
            return None
        return {
            "script_score": {
                "_name": query_name,
                "query": {"exists": {"field": plan.field}},
                "script": {
                    "source": (
                        f"((dotProduct(params.query_vector, '{plan.field}') + 1.0) / 2.0) * params.boost"
                    ),
                    "params": {
                        "query_vector": self._vector_to_list(query_vector),
                        "boost": float(plan.boost),
                    },
                },
            }
        }
    def build_exact_image_knn_rescore_clause(
        self,
        image_query_vector: Any,
        *,
        query_name: str = "exact_image_knn_query",
    ) -> Optional[Dict[str, Any]]:
        plan = self.get_image_knn_plan()
        if plan is None or image_query_vector is None:
            return None
        script_score_query = {
            "query": {"exists": {"field": plan.field}},
            "script": {
                "source": (
                    f"((dotProduct(params.query_vector, '{plan.field}') + 1.0) / 2.0) * params.boost"
                ),
                "params": {
                    "query_vector": self._vector_to_list(image_query_vector),
                    "boost": float(plan.boost),
                },
            },
        }
        if plan.nested_path:
            return {
                "nested": {
                    "path": plan.nested_path,
                    "_name": query_name,
                    "score_mode": "max",
                    "query": {"script_score": script_score_query},
                }
            }
        return {"script_score": {"_name": query_name, **script_score_query}}
    def _apply_source_filter(self, es_query: Dict[str, Any]) -> None:
        """
        Apply tri-state _source semantics:
        - None: do not set _source (return all source fields)
        - []: _source=false
        - [..]: _source.includes=[..]
        """
        if self.source_fields is None:
            return
        if not isinstance(self.source_fields, list):
            raise ValueError("query_config.source_fields must be null or list[str]")
        if len(self.source_fields) == 0:
            es_query["_source"] = False
            return
        es_query["_source"] = {"includes": self.source_fields}
    def _split_filters_for_faceting(
        self,
        filters: Optional[Dict[str, Any]],
        facet_configs: Optional[List[Any]]
    ) -> tuple:
        """
        Split filters into conjunctive (query) and disjunctive (post_filter) based on facet configs.
        
        Disjunctive filters (multi-select facets):
        - Applied via post_filter (affects results but not aggregations)
        - Allows showing other options in the same facet even when filtered
        
        Conjunctive filters (standard facets):
        - Applied in query.bool.filter (affects both results and aggregations)
        - Standard drill-down behavior
        
        Args:
            filters: All filters from request
            facet_configs: Facet configurations with disjunctive flags
            
        Returns:
            (conjunctive_filters, disjunctive_filters)
        """
        if not filters or not facet_configs:
            return filters or {}, {}
        
        # Get fields that support multi-select
        multi_select_fields = set()
        for fc in facet_configs:
            if getattr(fc, 'disjunctive', False):
                # Handle specifications.xxx format
                if fc.field.startswith('specifications.'):
                    multi_select_fields.add('specifications')
                else:
                    multi_select_fields.add(fc.field)
        
        # Split filters
        conjunctive = {}
        disjunctive = {}
        
        for field, value in filters.items():
            if field in multi_select_fields:
                disjunctive[field] = value
            else:
                conjunctive[field] = value
        
        return conjunctive, disjunctive
    def build_query(
        self,
        query_text: str,
        query_vector: Optional[np.ndarray] = None,
        image_query_vector: Optional[np.ndarray] = None,
        filters: Optional[Dict[str, Any]] = None,
        range_filters: Optional[Dict[str, Any]] = None,
        facet_configs: Optional[List[Any]] = None,
        size: int = 10,
        from_: int = 0,
        enable_knn: bool = True,
        min_score: Optional[float] = None,
        parsed_query: Optional[Any] = None,
    ) -> Dict[str, Any]:
        """
        Build complete ES query with post_filter support for multi-select faceting.
        结构：filters and (text_recall or embedding_recall) + post_filter
        - conjunctive_filters: 应用在 query.bool.filter（影响结果和聚合）
        - disjunctive_filters: 应用在 post_filter（只影响结果，不影响聚合）
        - text_recall: 文本相关性召回（按实际 clause 语言动态字段）
        - embedding_recall: 向量召回（KNN）
        - function_score: 包装召回部分，支持提权字段
        Args:
            query_text: Query text for BM25 matching
            query_vector: Query embedding for KNN search
            filters: Exact match filters
            range_filters: Range filters for numeric fields (always applied in query)
            facet_configs: Facet configurations (used to identify multi-select facets)
            size: Number of results
            from_: Offset for pagination
            enable_knn: Whether to use KNN search
            min_score: Minimum score threshold
        Returns:
            ES query DSL dictionary
        """
        # Boolean AST path has been removed; keep a single text strategy.
        es_query = {
            "size": size,
            "from": from_
        }
        # Add _source filtering with explicit tri-state semantics.
        self._apply_source_filter(es_query)
        # 1. Build recall queries (text or embedding)
        recall_clauses = []
        # Text recall (always include if query_text exists)
        if query_text:
            recall_clauses.extend(self._build_advanced_text_query(query_text, parsed_query))
        # Embedding recall
        has_embedding = enable_knn and query_vector is not None and self.text_embedding_field
        has_image_embedding = enable_knn and image_query_vector is not None and self.image_embedding_field
        
        # 2. Split filters for multi-select faceting
        conjunctive_filters, disjunctive_filters = self._split_filters_for_faceting(
            filters, facet_configs
        )
        
        # Build filter clauses for query (conjunctive filters + range filters)
        filter_clauses = self._build_filters(conjunctive_filters, range_filters)
        product_title_exclusion_filter = self._build_product_title_exclusion_filter(parsed_query)
        if product_title_exclusion_filter:
            filter_clauses.append(product_title_exclusion_filter)
        # 3. Add KNN search clauses alongside lexical clauses under the same bool.should
        # Text KNN: k / num_candidates from config; long queries use *_long and higher boost
        if has_embedding:
            text_knn_clause = self.build_text_knn_clause(
                query_vector,
                parsed_query=parsed_query,
                query_name="knn_query",
            )
            if text_knn_clause:
                recall_clauses.append(text_knn_clause)
        if has_image_embedding:
            image_knn_clause = self.build_image_knn_clause(
                image_query_vector,
                query_name="image_knn_query",
            )
            if image_knn_clause:
                recall_clauses.append(image_knn_clause)
        # 4. Build main query structure: filters and recall
        if recall_clauses:
            if len(recall_clauses) == 1:
                recall_query = recall_clauses[0]
            else:
                recall_query = {
                    "bool": {
                        "should": recall_clauses,
                        "minimum_should_match": 1
                    }
                }
            recall_query = self._wrap_with_function_score(recall_query)
            if filter_clauses:
                es_query["query"] = {
                    "bool": {
                        "must": [recall_query],
                        "filter": filter_clauses
                    }
                }
            else:
                es_query["query"] = recall_query
        else:
            if filter_clauses:
                es_query["query"] = {
                    "bool": {
                        "must": [{"match_all": {}}],
                        "filter": filter_clauses
                    }
                }
            else:
                es_query["query"] = {"match_all": {}}
        # 5. Add post_filter for disjunctive (multi-select) filters
        if disjunctive_filters:
            post_filter_clauses = self._build_filters(disjunctive_filters, None)
            if post_filter_clauses:
                if len(post_filter_clauses) == 1:
                    es_query["post_filter"] = post_filter_clauses[0]
                else:
                    es_query["post_filter"] = {
                        "bool": {"filter": post_filter_clauses}
                    }
        # 6. Add minimum score filter
        if min_score is not None:
            es_query["min_score"] = min_score
        return es_query
    
    def _wrap_with_function_score(self, query: Dict[str, Any]) -> Dict[str, Any]:
        """
        Wrap query with function_score for boosting fields.
        
        Args:
            query: Base query to wrap
            
        Returns:
            Function score query or original query if no functions configured
        """
        functions = self._build_score_functions()
        
        # If no functions configured, return original query
        if not functions:
            return query
        
        # Build function_score query
        score_mode = self.function_score_config.score_mode if self.function_score_config else "sum"
        boost_mode = self.function_score_config.boost_mode if self.function_score_config else "multiply"
        
        function_score_query = {
            "function_score": {
                "query": query,
                "functions": functions,
                "score_mode": score_mode,
                "boost_mode": boost_mode
            }
        }
        
        return function_score_query
    
    def _build_score_functions(self) -> List[Dict[str, Any]]:
        """
        Build function_score functions from config.
        
        Returns:
            List of function score functions
        """
        functions = []
        if not self.function_score_config:
            return functions
        
        config_functions = self.function_score_config.functions or []
        
        for func_config in config_functions:
            func_type = func_config.get("type")
            
            if func_type == "filter_weight":
                # Filter + Weight
                functions.append({
                    "filter": func_config["filter"],
                    "weight": func_config.get("weight", 1.0)
                })
            
            elif func_type == "field_value_factor":
                # Field Value Factor
                functions.append({
                    "field_value_factor": {
                        "field": func_config["field"],
                        "factor": func_config.get("factor", 1.0),
                        "modifier": func_config.get("modifier", "none"),
                        "missing": func_config.get("missing", 1.0)
                    }
                })
            
            elif func_type == "decay":
                # Decay Function (gauss/exp/linear)
                decay_func = func_config.get("function", "gauss")
                field = func_config["field"]
                
                decay_params = {
                    "origin": func_config.get("origin", "now"),
                    "scale": func_config["scale"]
                }
                
                if "offset" in func_config:
                    decay_params["offset"] = func_config["offset"]
                if "decay" in func_config:
                    decay_params["decay"] = func_config["decay"]
                
                functions.append({
                    decay_func: {
                        field: decay_params
                    }
                })
        
        return functions
    def _format_field_with_boost(self, field_name: str, boost: float) -> str:
        if abs(float(boost) - 1.0) < 1e-9:
            return field_name
        return f"{field_name}^{round(boost, 2)}"
    def _get_field_boost(self, base_field: str, language: Optional[str] = None) -> float:
        # Language-specific override first (e.g. title.de), then base field (e.g. title)
        if language:
            lang_key = f"{base_field}.{language}"
            if lang_key in self.field_boosts:
                return float(self.field_boosts[lang_key])
        if base_field in self.field_boosts:
            return float(self.field_boosts[base_field])
        return 1.0
    def _match_field_strings(
        self,
        language: str,
        *,
        multilingual_fields: Optional[List[str]] = None,
        shared_fields: Optional[List[str]] = None,
        boost_overrides: Optional[Dict[str, float]] = None,
    ) -> List[str]:
        """Build ``multi_match`` / ``combined_fields`` field entries for one language code."""
        lang = (language or "").strip().lower()
        text_bases = multilingual_fields if multilingual_fields is not None else self.multilingual_fields
        term_fields = shared_fields if shared_fields is not None else self.shared_fields
        overrides = boost_overrides or {}
        out: List[str] = []
        for base in text_bases:
            path = f"{base}.{lang}"
            boost = float(overrides.get(base, self._get_field_boost(base, lang)))
            out.append(self._format_field_with_boost(path, boost))
        for shared in term_fields:
            boost = float(overrides.get(shared, self._get_field_boost(shared, None)))
            out.append(self._format_field_with_boost(shared, boost))
        return out
    def _build_best_fields_clause(self, language: str, query_text: str) -> Optional[Dict[str, Any]]:
        fields = self._match_field_strings(
            language,
            multilingual_fields=list(self.best_fields_boosts),
            shared_fields=[],
            boost_overrides=self.best_fields_boosts,
        )
        if not fields:
            return None
        return {
            "multi_match": {
                "query": query_text,
                "type": "best_fields",
                "fields": fields,
                "boost": self.best_fields_clause_boost,
            }
        }
    def _build_phrase_clause(self, language: str, query_text: str) -> Optional[Dict[str, Any]]:
        fields = self._match_field_strings(
            language,
            multilingual_fields=list(self.phrase_field_boosts),
            shared_fields=[],
            boost_overrides=self.phrase_field_boosts,
        )
        if not fields:
            return None
        clause: Dict[str, Any] = {
            "multi_match": {
                "query": query_text,
                "type": "phrase",
                "fields": fields,
                "boost": self.phrase_match_boost,
            }
        }
        if self.phrase_match_slop > 0:
            clause["multi_match"]["slop"] = self.phrase_match_slop
        if self.phrase_match_tie_breaker > 0:
            clause["multi_match"]["tie_breaker"] = self.phrase_match_tie_breaker
        return clause
    def _build_lexical_language_clause(
        self,
        lang: str,
        lang_query: str,
        clause_name: str,
        *,
        is_source: bool,
        keywords_query: Optional[str] = None,
    ) -> Optional[Dict[str, Any]]:
        combined_fields = self._match_field_strings(lang)
        if not combined_fields:
            return None
        minimum_should_match = (
            self.base_minimum_should_match if is_source else self.translation_minimum_should_match
        )
        kw = (keywords_query or "").strip()
        main_query = (lang_query or "").strip()
        combined_must: List[Dict[str, Any]] = [
            {
                "combined_fields": {
                    "query": main_query,
                    "fields": combined_fields,
                    "minimum_should_match": minimum_should_match,
                    "boost": 2.0,
                }
            }
        ]
        if kw and kw != main_query:
            combined_must.append(
                {
                    "combined_fields": {
                        "query": kw,
                        "fields": combined_fields,
                        "minimum_should_match": self.keywords_minimum_should_match,
                        "boost": 0.8,
                    }
                }
            )
        optional_mm = [
            clause
            for clause in (
                self._build_best_fields_clause(lang, main_query),
                self._build_phrase_clause(lang, main_query),
            )
            if clause
        ]
        should_clauses: List[Dict[str, Any]] = [{"bool": {"must": combined_must}}]
        should_clauses.extend(optional_mm)
        clause: Dict[str, Any] = {
            "bool": {
                "_name": clause_name,
                "should": should_clauses,
                "minimum_should_match": 1,
            }
        }
        if not is_source:
            clause["bool"]["boost"] = float(self.translation_boost)
        return clause
    def _build_advanced_text_query(
        self,
        query_text: str,
        parsed_query: Optional[Any] = None,
    ) -> List[Dict[str, Any]]:
        """
        Build advanced text query using base and translated lexical clauses.
        Unified implementation:
        - base_query: source-language clause
        - translation queries: target-language clauses from translations
        Args:
            query_text: Query text
            parsed_query: ParsedQuery object with analysis results
            
        Returns:
            Flat recall clauses to be merged with KNN clauses under query.bool.should
        """
        should_clauses = []
        source_lang = self.default_language
        translations: Dict[str, str] = {}
        if parsed_query:
            detected_lang = getattr(parsed_query, "detected_language", None)
            source_lang = detected_lang if detected_lang and detected_lang != "unknown" else self.default_language
            translations = getattr(parsed_query, "translations", None) or {}
        source_lang = str(source_lang or self.default_language).strip().lower() or self.default_language
        base_query_text = (
            getattr(parsed_query, "rewritten_query", None) if parsed_query else None
        ) or query_text
        kw_by_variant: Dict[str, str] = (
            getattr(parsed_query, "keywords_queries", None) or {}
            if parsed_query
            else {}
        )
        if base_query_text:
            base_clause = self._build_lexical_language_clause(
                source_lang,
                base_query_text,
                "base_query",
                is_source=True,
                keywords_query=(kw_by_variant.get(KEYWORDS_QUERY_BASE_KEY) or "").strip(),
            )
            if base_clause:
                should_clauses.append(base_clause)
        for lang, translated_text in translations.items():
            normalized_lang = str(lang or "").strip().lower()
            normalized_text = str(translated_text or "").strip()
            if not normalized_lang or not normalized_text:
                continue
            if normalized_lang == source_lang and normalized_text == base_query_text:
                continue
            trans_kw = (kw_by_variant.get(normalized_lang) or "").strip()
            trans_clause = self._build_lexical_language_clause(
                normalized_lang,
                normalized_text,
                f"base_query_trans_{normalized_lang}",
                is_source=False,
                keywords_query=trans_kw,
            )
            if trans_clause:
                should_clauses.append(trans_clause)
        # Fallback to a simple query when language fields cannot be resolved.
        if not should_clauses:
            fallback_fields = self.match_fields or ["title.en^1.0"]
            fallback_lexical = {
                "multi_match": {
                    "_name": "base_query_fallback",
                    "query": query_text,
                    "fields": fallback_fields,
                    "minimum_should_match": self.base_minimum_should_match,
                }
            }
            return [fallback_lexical]
        return should_clauses
    def _build_filters(
        self, 
        filters: Optional[Dict[str, Any]] = None,
        range_filters: Optional[Dict[str, 'RangeFilter']] = None
    ) -> List[Dict[str, Any]]:
        """
        构建过滤子句。
        
        Args:
            filters: 精确匹配过滤器字典
            range_filters: 范围过滤器（Dict[str, RangeFilter]，RangeFilter 是 Pydantic 模型）
        
        Returns:
            ES filter 子句列表
        """
        filter_clauses = []
        
        # 1. 处理精确匹配过滤
        if filters:
            for field, value in filters.items():
                # 特殊处理：specifications 嵌套过滤
                if field == "specifications":
                    if isinstance(value, dict):
                        # 单个规格过滤：{"name": "color", "value": "green"}
                        name = value.get("name")
                        spec_value = value.get("value")
                        if name and spec_value:
                            filter_clauses.append({
                                "nested": {
                                    "path": "specifications",
                                    "query": {
                                        "bool": {
                                            "must": [
                                                {"term": {"specifications.name": name}},
                                                {"term": {"specifications.value": spec_value}}
                                            ]
                                        }
                                    }
                                }
                            })
                    elif isinstance(value, list):
                        # 多个规格过滤：按 name 分组，相同维度 OR，不同维度 AND
                        # 例如：[{"name": "size", "value": "3"}, {"name": "size", "value": "4"}, {"name": "color", "value": "green"}]
                        # 应该生成：(size=3 OR size=4) AND color=green
                        from collections import defaultdict
                        specs_by_name = defaultdict(list)
                        for spec in value:
                            if isinstance(spec, dict):
                                name = spec.get("name")
                                spec_value = spec.get("value")
                                if name and spec_value:
                                    specs_by_name[name].append(spec_value)
                        
                        # 为每个 name 维度生成一个过滤子句
                        for name, values in specs_by_name.items():
                            if len(values) == 1:
                                # 单个值，直接生成 term 查询
                                filter_clauses.append({
                                    "nested": {
                                        "path": "specifications",
                                        "query": {
                                            "bool": {
                                                "must": [
                                                    {"term": {"specifications.name": name}},
                                                    {"term": {"specifications.value": values[0]}}
                                                ]
                                            }
                                        }
                                    }
                                })
                            else:
                                # 多个值，使用 should (OR) 连接
                                should_clauses = []
                                for spec_value in values:
                                    should_clauses.append({
                                        "bool": {
                                            "must": [
                                                {"term": {"specifications.name": name}},
                                                {"term": {"specifications.value": spec_value}}
                                            ]
                                        }
                                    })
                                filter_clauses.append({
                                    "nested": {
                                        "path": "specifications",
                                        "query": {
                                            "bool": {
                                                "should": should_clauses,
                                                "minimum_should_match": 1
                                            }
                                        }
                                    }
                                })
                    continue
                
                # *_all 语义：多值时为 AND（必须同时匹配所有值）
                if field.endswith("_all"):
                    es_field = field[:-4]  # 去掉 _all 后缀
                    if es_field == "specifications" and isinstance(value, list):
                        # specifications_all: 列表内每个规格条件都要满足（AND）
                        must_nested = []
                        for spec in value:
                            if isinstance(spec, dict):
                                name = spec.get("name")
                                spec_value = spec.get("value")
                                if name and spec_value:
                                    must_nested.append({
                                        "nested": {
                                            "path": "specifications",
                                            "query": {
                                                "bool": {
                                                    "must": [
                                                        {"term": {"specifications.name": name}},
                                                        {"term": {"specifications.value": spec_value}}
                                                    ]
                                                }
                                            }
                                        }
                                    })
                        if must_nested:
                            filter_clauses.append({"bool": {"must": must_nested}})
                    else:
                        # 普通字段 _all：多值用 must + 多个 term
                        if isinstance(value, list):
                            if value:
                                filter_clauses.append({
                                    "bool": {
                                        "must": [{"term": {es_field: v}} for v in value]
                                    }
                                })
                        else:
                            filter_clauses.append({"term": {es_field: value}})
                    continue
                
                # 普通字段过滤（默认多值为 OR）
                if isinstance(value, list):
                    # 多值匹配（OR）
                    filter_clauses.append({
                        "terms": {field: value}
                    })
                else:
                    # 单值精确匹配
                    filter_clauses.append({
                        "term": {field: value}
                    })
        
        # 2. 处理范围过滤（支持 RangeFilter Pydantic 模型或字典）
        if range_filters:
            for field, range_filter in range_filters.items():
                # 支持 Pydantic 模型或字典格式
                if hasattr(range_filter, 'model_dump'):
                    # Pydantic 模型
                    range_dict = range_filter.model_dump(exclude_none=True)
                elif isinstance(range_filter, dict):
                    # 已经是字典格式
                    range_dict = {k: v for k, v in range_filter.items() if v is not None}
                else:
                    # 其他格式，跳过
                    continue
                
                if range_dict:
                    filter_clauses.append({
                        "range": {field: range_dict}
                    })
        
        return filter_clauses
    @staticmethod
    def _build_product_title_exclusion_filter(parsed_query: Optional[Any]) -> Optional[Dict[str, Any]]:
        if parsed_query is None:
            return None
        profile = getattr(parsed_query, "product_title_exclusion_profile", None)
        if not profile or not getattr(profile, "is_active", False):
            return None
        should_clauses: List[Dict[str, Any]] = []
        for term in profile.all_zh_title_exclusions():
            should_clauses.append({"match_phrase": {"title.zh": {"query": term}}})
        for term in profile.all_en_title_exclusions():
            should_clauses.append({"match_phrase": {"title.en": {"query": term}}})
        if not should_clauses:
            return None
        return {
            "bool": {
                "must_not": [
                    {
                        "bool": {
                            "should": should_clauses,
                            "minimum_should_match": 1,
                        }
                    }
                ]
            }
        }
    def add_sorting(
        self,
        es_query: Dict[str, Any],
        sort_by: str,
        sort_order: str = "desc"
    ) -> Dict[str, Any]:
        """
        Add sorting to ES query.
        Args:
            es_query: Existing ES query
            sort_by: Field name for sorting (支持 'price' 自动映射)
            sort_order: Sort order: 'asc' or 'desc'
        Returns:
            Modified ES query
        """
        if not sort_by:
            return es_query
        if not sort_order:
            sort_order = "desc"
        # Auto-map 'price' to 'min_price' or 'max_price' based on sort_order
        if sort_by == "price":
            if sort_order.lower() == "asc":
                sort_by = "min_price"  # 价格从低到高
            else:
                sort_by = "max_price"  # 价格从高到低
        if "sort" not in es_query:
            es_query["sort"] = []
        # Add the specified sort
        sort_field = {
            sort_by: {
                "order": sort_order.lower()
            }
        }
        es_query["sort"].append(sort_field)
        return es_query
    def build_facets(
        self,
        facet_configs: Optional[List['FacetConfig']] = None,
        use_reverse_nested: bool = True
    ) -> Dict[str, Any]:
        """
        构建分面聚合。
        
        Args:
            facet_configs: 分面配置对象列表
            use_reverse_nested: 是否使用 reverse_nested 统计产品数量（默认 True）
                               如果为 False，将统计嵌套文档数量（性能更好但计数可能不准确）
            
            支持的字段类型：
                - 普通字段: 如 "category1_name"（terms 或 range 类型）
                - specifications: "specifications"（返回所有规格名称及其值）
                - specifications.{name}: 如 "specifications.color"（返回指定规格名称的值）
        
        Returns:
            ES aggregations 字典
        
        性能说明：
            - use_reverse_nested=True: 统计产品数量，准确性高但性能略差（通常影响 < 20%）
            - use_reverse_nested=False: 统计嵌套文档数量，性能更好但计数可能不准确
        """
        if not facet_configs:
            return {}
        
        aggs = {}
        
        for config in facet_configs:
            field = config.field
            size = config.size
            facet_type = config.type
            
            # 处理 specifications（所有规格名称）
            if field == "specifications":
                aggs["specifications_facet"] = {
                    "nested": {"path": "specifications"},
                    "aggs": {
                        "by_name": {
                            "terms": {
                                "field": "specifications.name",
                                "size": 20,
                                "order": {"_count": "desc"}
                            },
                            "aggs": {
                                "value_counts": {
                                    "terms": {
                                        "field": "specifications.value",
                                        "size": size,
                                        "order": {"_count": "desc"}
                                    }
                                }
                            }
                        }
                    }
                }
                continue
            
            # 处理 specifications.{name}（指定规格名称）
            if field.startswith("specifications."):
                name = field[len("specifications."):]
                agg_name = f"specifications_{name}_facet"
                # 使用 reverse_nested 统计产品（父文档）数量，而不是规格条目（嵌套文档）数量
                # 这样可以确保分面计数反映实际的产品数量，与搜索结果数量一致
                base_value_counts = {
                    "terms": {
                        "field": "specifications.value",
                        "size": size,
                        "order": {"_count": "desc"}
                    }
                }
                
                # 如果启用 reverse_nested，添加子聚合统计产品数量
                if use_reverse_nested:
                    base_value_counts["aggs"] = {
                        "product_count": {
                            "reverse_nested": {}
                        }
                    }
                
                aggs[agg_name] = {
                    "nested": {"path": "specifications"},
                    "aggs": {
                        "filter_by_name": {
                            "filter": {"term": {"specifications.name": name}},
                            "aggs": {
                                "value_counts": base_value_counts
                            }
                        }
                    }
                }
                continue
            
            # 处理普通字段
            agg_name = f"{field}_facet"
            
            if facet_type == 'terms':
                aggs[agg_name] = {
                    "terms": {
                        "field": field,
                        "size": size,
                        "order": {"_count": "desc"}
                    }
                }
            elif facet_type == 'range':
                if config.ranges:
                    aggs[agg_name] = {
                        "range": {
                            "field": field,
                            "ranges": config.ranges
                        }
                    }
        
        return aggs