"""
  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",

                      # Expose the best-matching image entry (url, score) so SKU selection
                      # can promote the SKU whose image_src matches the winning url.
                      "inner_hits": {
                          "name": f"{query_name}_hits",
                          "size": 1,
                          "_source": ["url"],
                      },

                  }
              }
          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},

                      # Same rationale as build_image_knn_clause: carry the winning url + score
                      # so downstream SKU selection can consume it without a second ES round-trip.
                      "inner_hits": {
                          "name": f"{query_name}_hits",
                          "size": 1,
                          "_source": ["url"],
                      },

                  }
              }
          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: