from __future__ import annotations
  
  from dataclasses import dataclass
  from pathlib import Path
  from types import SimpleNamespace
  from typing import Any, Dict, List
  

  import numpy as np

  import yaml
  
  from config import (
      ConfigLoader,
      FunctionScoreConfig,
      IndexConfig,
      QueryConfig,

      RerankConfig,
      SPUConfig,
      SearchConfig,
  )
  from context import create_request_context
  from search.searcher import Searcher
  
  
  @dataclass
  class _FakeParsedQuery:
      original_query: str
      query_normalized: str
      rewritten_query: str
      detected_language: str = "en"
      translations: Dict[str, str] = None
      query_vector: Any = None
      domain: str = "default"
  
      def to_dict(self) -> Dict[str, Any]:
          return {
              "original_query": self.original_query,
              "query_normalized": self.query_normalized,
              "rewritten_query": self.rewritten_query,
              "detected_language": self.detected_language,
              "translations": self.translations or {},
              "domain": self.domain,
          }
  
  
  class _FakeQueryParser:
      def parse(self, query: str, tenant_id: str, generate_vector: bool, context: Any):
          return _FakeParsedQuery(
              original_query=query,
              query_normalized=query,
              rewritten_query=query,
              translations={},
          )
  
  
  class _FakeQueryBuilder:
      def build_query(self, **kwargs):
          return {
              "query": {"match_all": {}},
              "size": kwargs["size"],
              "from": kwargs["from_"],
          }
  
      def build_facets(self, facets: Any):
          return {}
  
      def add_sorting(self, es_query: Dict[str, Any], sort_by: str, sort_order: str):
          return es_query
  
  
  class _FakeESClient:
      def __init__(self, total_hits: int = 5000):
          self.calls: List[Dict[str, Any]] = []
          self.total_hits = total_hits
  
      @staticmethod
      def _apply_source_filter(src: Dict[str, Any], source_spec: Any) -> Dict[str, Any]:
          if source_spec is None:
              return dict(src)
          if source_spec is False:
              return {}
          if isinstance(source_spec, dict):
              includes = source_spec.get("includes") or []
          elif isinstance(source_spec, list):
              includes = source_spec
          else:
              includes = []
          if not includes:
              return dict(src)
          return {k: v for k, v in src.items() if k in set(includes)}
  
      @staticmethod
      def _full_source(doc_id: str) -> Dict[str, Any]:
          return {
              "spu_id": doc_id,
              "title": {"en": f"product-{doc_id}"},
              "brief": {"en": f"brief-{doc_id}"},
              "vendor": {"en": f"vendor-{doc_id}"},
              "skus": [],
          }
  

      def search(
          self,
          index_name: str,
          body: Dict[str, Any],
          size: int,
          from_: int,
          include_named_queries_score: bool = False,
      ):

          self.calls.append(

              {
                  "index_name": index_name,
                  "body": body,
                  "size": size,
                  "from_": from_,
                  "include_named_queries_score": include_named_queries_score,
              }

          )
          ids_query = (((body or {}).get("query") or {}).get("ids") or {}).get("values")
          source_spec = (body or {}).get("_source")
  
          if isinstance(ids_query, list):
              # Return reversed order intentionally; caller should restore original ranking order.
              ids = [str(i) for i in ids_query][::-1]
              hits = []
              for doc_id in ids:
                  src = self._apply_source_filter(self._full_source(doc_id), source_spec)
                  hit = {"_id": doc_id, "_score": 1.0}
                  if source_spec is not False:
                      hit["_source"] = src
                  hits.append(hit)
          else:
              end = min(from_ + size, self.total_hits)
              hits = []
              for i in range(from_, end):
                  doc_id = str(i)
                  src = self._apply_source_filter(self._full_source(doc_id), source_spec)
                  hit = {"_id": doc_id, "_score": float(self.total_hits - i)}
                  if source_spec is not False:
                      hit["_source"] = src
                  hits.append(hit)
  
          return {
              "took": 8,
              "hits": {
                  "total": {"value": self.total_hits},
                  "max_score": hits[0]["_score"] if hits else 0.0,
                  "hits": hits,
              },
          }
  
  

  def _build_search_config(*, rerank_enabled: bool = True, rerank_window: int = 384):

      return SearchConfig(
          field_boosts={"title.en": 3.0},
          indexes=[IndexConfig(name="default", label="default", fields=["title.en"])],
          query_config=QueryConfig(enable_text_embedding=False, enable_query_rewrite=False),

          function_score=FunctionScoreConfig(),
          rerank=RerankConfig(enabled=rerank_enabled, rerank_window=rerank_window),
          spu_config=SPUConfig(enabled=False),
          es_index_name="test_products",

          es_settings={},

      )
  
  
  def _build_searcher(config: SearchConfig, es_client: _FakeESClient) -> Searcher:
      searcher = Searcher(
          es_client=es_client,
          config=config,
          query_parser=_FakeQueryParser(),
      )
      searcher.query_builder = _FakeQueryBuilder()
      return searcher
  
  

  class _FakeTextEncoder:
      def __init__(self, vectors: Dict[str, List[float]]):
          self.vectors = {
              key: np.array(value, dtype=np.float32)
              for key, value in vectors.items()
          }
  
      def encode(self, sentences, priority: int = 0, **kwargs):
          if isinstance(sentences, str):
              sentences = [sentences]
          return np.array([self.vectors[text] for text in sentences], dtype=object)
  
  

  def test_config_loader_rerank_enabled_defaults_true(tmp_path: Path):
      config_data = {
          "es_index_name": "test_products",
          "field_boosts": {"title.en": 3.0},
          "indexes": [{"name": "default", "label": "default", "fields": ["title.en"]}],
          "query_config": {"supported_languages": ["en"], "default_language": "en"},
          "spu_config": {"enabled": False},

          "function_score": {"score_mode": "sum", "boost_mode": "multiply", "functions": []},

          "rerank": {"rerank_window": 384},

      }
      config_path = tmp_path / "config.yaml"
      config_path.write_text(yaml.safe_dump(config_data), encoding="utf-8")
  
      loader = ConfigLoader(config_path)
      loaded = loader.load_config(validate=False)
  
      assert loaded.rerank.enabled is True
  
  
  def test_searcher_reranks_top_window_by_default(monkeypatch):
      es_client = _FakeESClient()
      searcher = _build_searcher(_build_search_config(rerank_enabled=True), es_client)
      context = create_request_context(reqid="t1", uid="u1")
  
      monkeypatch.setattr(
          "search.searcher.get_tenant_config_loader",
          lambda: SimpleNamespace(get_tenant_config=lambda tenant_id: {"index_languages": ["en"]}),
      )
  
      called: Dict[str, Any] = {"count": 0, "docs": 0}
  
      def _fake_run_rerank(**kwargs):
          called["count"] += 1
          called["docs"] = len(kwargs["es_response"]["hits"]["hits"])
          return kwargs["es_response"], None, []
  
      monkeypatch.setattr("search.rerank_client.run_rerank", _fake_run_rerank)
  
      result = searcher.search(
          query="toy",
          tenant_id="162",
          from_=20,
          size=10,
          context=context,
          enable_rerank=None,
      )
  
      assert called["count"] == 1

      # 应当对配置的 rerank_window 条文档做重排预取
      window = searcher.config.rerank.rerank_window
      assert called["docs"] == window

      assert es_client.calls[0]["from_"] == 0

      assert es_client.calls[0]["size"] == window

      assert es_client.calls[0]["include_named_queries_score"] is True

      assert es_client.calls[0]["body"]["_source"] == {"includes": ["title"]}
      assert len(es_client.calls) == 2
      assert es_client.calls[1]["size"] == 10
      assert es_client.calls[1]["from_"] == 0
      assert es_client.calls[1]["body"]["query"]["ids"]["values"] == [str(i) for i in range(20, 30)]
      assert len(result.results) == 10
      assert result.results[0].spu_id == "20"
      assert result.results[0].brief == "brief-20"
  
  
  def test_searcher_rerank_prefetch_source_follows_doc_template(monkeypatch):
      es_client = _FakeESClient()
      searcher = _build_searcher(_build_search_config(rerank_enabled=True), es_client)
      context = create_request_context(reqid="t1b", uid="u1b")
  
      monkeypatch.setattr(
          "search.searcher.get_tenant_config_loader",
          lambda: SimpleNamespace(get_tenant_config=lambda tenant_id: {"index_languages": ["en"]}),
      )
      monkeypatch.setattr("search.rerank_client.run_rerank", lambda **kwargs: (kwargs["es_response"], None, []))
  
      searcher.search(
          query="toy",
          tenant_id="162",
          from_=0,
          size=5,
          context=context,
          enable_rerank=None,
          rerank_doc_template="{title} {vendor} {brief}",
      )
  
      assert es_client.calls[0]["body"]["_source"] == {"includes": ["brief", "title", "vendor"]}
  
  
  def test_searcher_skips_rerank_when_request_explicitly_false(monkeypatch):
      es_client = _FakeESClient()
      searcher = _build_searcher(_build_search_config(rerank_enabled=True), es_client)
      context = create_request_context(reqid="t2", uid="u2")
  
      monkeypatch.setattr(
          "search.searcher.get_tenant_config_loader",
          lambda: SimpleNamespace(get_tenant_config=lambda tenant_id: {"index_languages": ["en"]}),
      )
  
      called: Dict[str, int] = {"count": 0}
  
      def _fake_run_rerank(**kwargs):
          called["count"] += 1
          return kwargs["es_response"], None, []
  
      monkeypatch.setattr("search.rerank_client.run_rerank", _fake_run_rerank)
  
      searcher.search(
          query="toy",
          tenant_id="162",
          from_=20,
          size=10,
          context=context,
          enable_rerank=False,
      )
  
      assert called["count"] == 0
      assert es_client.calls[0]["from_"] == 20
      assert es_client.calls[0]["size"] == 10

      assert es_client.calls[0]["include_named_queries_score"] is False

      assert len(es_client.calls) == 1
  
  
  def test_searcher_skips_rerank_when_page_exceeds_window(monkeypatch):
      es_client = _FakeESClient()

      searcher = _build_searcher(_build_search_config(rerank_enabled=True, rerank_window=384), es_client)

      context = create_request_context(reqid="t3", uid="u3")
  
      monkeypatch.setattr(
          "search.searcher.get_tenant_config_loader",
          lambda: SimpleNamespace(get_tenant_config=lambda tenant_id: {"index_languages": ["en"]}),
      )
  
      called: Dict[str, int] = {"count": 0}
  
      def _fake_run_rerank(**kwargs):
          called["count"] += 1
          return kwargs["es_response"], None, []
  
      monkeypatch.setattr("search.rerank_client.run_rerank", _fake_run_rerank)
  
      searcher.search(
          query="toy",
          tenant_id="162",
          from_=995,
          size=10,
          context=context,
          enable_rerank=None,
      )
  
      assert called["count"] == 0
      assert es_client.calls[0]["from_"] == 995
      assert es_client.calls[0]["size"] == 10

      assert es_client.calls[0]["include_named_queries_score"] is False

      assert len(es_client.calls) == 1

  
  
  def test_searcher_promotes_sku_when_option1_matches_translated_query(monkeypatch):
      es_client = _FakeESClient(total_hits=1)
      searcher = _build_searcher(_build_search_config(rerank_enabled=False), es_client)
      context = create_request_context(reqid="sku-text", uid="u-sku-text")
  
      monkeypatch.setattr(
          "search.searcher.get_tenant_config_loader",
          lambda: SimpleNamespace(get_tenant_config=lambda tenant_id: {"index_languages": ["en", "zh"]}),
      )
  
      class _TranslatedQueryParser:
          text_encoder = None
  
          def parse(self, query: str, tenant_id: str, generate_vector: bool, context: Any):
              return _FakeParsedQuery(
                  original_query=query,
                  query_normalized=query,
                  rewritten_query=query,
                  translations={"en": "black dress"},
              )
  
      searcher.query_parser = _TranslatedQueryParser()
  
      def _full_source_with_skus(doc_id: str) -> Dict[str, Any]:
          return {
              "spu_id": doc_id,
              "title": {"en": f"product-{doc_id}"},
              "brief": {"en": f"brief-{doc_id}"},
              "vendor": {"en": f"vendor-{doc_id}"},

              "option1_name": "Color",

              "image_url": "https://img/default.jpg",
              "skus": [
                  {"sku_id": "sku-red", "option1_value": "Red", "image_src": "https://img/red.jpg"},
                  {"sku_id": "sku-black", "option1_value": "Black", "image_src": "https://img/black.jpg"},
              ],
          }
  
      monkeypatch.setattr(_FakeESClient, "_full_source", staticmethod(_full_source_with_skus))
  
      result = searcher.search(
          query="黑色 连衣裙",
          tenant_id="162",
          from_=0,
          size=1,
          context=context,
          enable_rerank=False,
      )
  
      assert len(result.results) == 1
      assert result.results[0].skus[0].sku_id == "sku-black"
      assert result.results[0].image_url == "https://img/black.jpg"
  
  
  def test_searcher_promotes_sku_by_embedding_when_query_has_no_direct_option_match(monkeypatch):
      es_client = _FakeESClient(total_hits=1)
      searcher = _build_searcher(_build_search_config(rerank_enabled=False), es_client)
      context = create_request_context(reqid="sku-embed", uid="u-sku-embed")
  
      monkeypatch.setattr(
          "search.searcher.get_tenant_config_loader",
          lambda: SimpleNamespace(get_tenant_config=lambda tenant_id: {"index_languages": ["en"]}),
      )
  
      encoder = _FakeTextEncoder(
          {
              "linen summer dress": [0.8, 0.2],

              "color:Red": [1.0, 0.0],
              "color:Blue": [0.0, 1.0],

          }
      )
  
      class _EmbeddingQueryParser:
          text_encoder = encoder
  
          def parse(self, query: str, tenant_id: str, generate_vector: bool, context: Any):
              return _FakeParsedQuery(
                  original_query=query,
                  query_normalized=query,
                  rewritten_query=query,
                  translations={},
                  query_vector=np.array([0.0, 1.0], dtype=np.float32),
              )
  
      searcher.query_parser = _EmbeddingQueryParser()
  
      def _full_source_with_skus(doc_id: str) -> Dict[str, Any]:
          return {
              "spu_id": doc_id,
              "title": {"en": f"product-{doc_id}"},
              "brief": {"en": f"brief-{doc_id}"},
              "vendor": {"en": f"vendor-{doc_id}"},

              "option1_name": "Color",

              "image_url": "https://img/default.jpg",
              "skus": [
                  {"sku_id": "sku-red", "option1_value": "Red", "image_src": "https://img/red.jpg"},
                  {"sku_id": "sku-blue", "option1_value": "Blue", "image_src": "https://img/blue.jpg"},
              ],
          }
  
      monkeypatch.setattr(_FakeESClient, "_full_source", staticmethod(_full_source_with_skus))
  
      result = searcher.search(
          query="linen summer dress",
          tenant_id="162",
          from_=0,
          size=1,
          context=context,
          enable_rerank=False,
      )
  
      assert len(result.results) == 1
      assert result.results[0].skus[0].sku_id == "sku-blue"
      assert result.results[0].image_url == "https://img/blue.jpg"