"""
  Full suggestion index builder.
  
  Build data from:
  - ES product index fields: title.{lang}, qanchors.{lang}
  - MySQL search logs: shoplazza_search_log.query (+ language metadata)
  """
  
  import json
  import logging
  import math
  import re
  from dataclasses import dataclass, field
  from datetime import datetime, timedelta, timezone
  from typing import Any, Dict, List, Optional, Tuple
  
  from sqlalchemy import text
  
  from config.tenant_config_loader import get_tenant_config_loader
  from utils.es_client import ESClient
  from suggestion.mapping import build_suggestion_mapping

  from config.env_config import ES_INDEX_NAMESPACE

  
  logger = logging.getLogger(__name__)
  
  
  def get_suggestion_index_name(tenant_id: str) -> str:

      """
      生成 suggestion 索引名称。
  
      命名规则：
        {ES_INDEX_NAMESPACE}search_suggestions_tenant_{tenant_id}
  
      通过 ES_INDEX_NAMESPACE 统一区分 prod/uat/test 等环境。
      """
      prefix = ES_INDEX_NAMESPACE or ""
      return f"{prefix}search_suggestions_tenant_{tenant_id}"

  
  
  @dataclass
  class SuggestionCandidate:
      text: str
      text_norm: str
      lang: str
      sources: set = field(default_factory=set)
      title_spu_ids: set = field(default_factory=set)
      qanchor_spu_ids: set = field(default_factory=set)
      query_count_7d: int = 0
      query_count_30d: int = 0
      lang_confidence: float = 1.0
      lang_source: str = "default"
      lang_conflict: bool = False
      top_spu_scores: Dict[str, float] = field(default_factory=dict)
  
      def add_product(self, source: str, spu_id: str, score: float) -> None:
          self.sources.add(source)
          if source == "title":
              self.title_spu_ids.add(spu_id)
          elif source == "qanchor":
              self.qanchor_spu_ids.add(spu_id)
          prev = self.top_spu_scores.get(spu_id)
          if prev is None or score > prev:
              self.top_spu_scores[spu_id] = score
  
      def add_query_log(self, is_7d: bool) -> None:
          self.sources.add("query_log")
          self.query_count_30d += 1
          if is_7d:
              self.query_count_7d += 1
  
  
  class SuggestionIndexBuilder:
      """Build and rebuild suggestion index."""
  
      def __init__(self, es_client: ESClient, db_engine: Any):
          self.es_client = es_client
          self.db_engine = db_engine
  
      @staticmethod
      def _normalize_text(value: str) -> str:
          text_value = (value or "").strip().lower()
          text_value = re.sub(r"\s+", " ", text_value)
          return text_value
  
      @staticmethod
      def _split_qanchors(value: Any) -> List[str]:
          if value is None:
              return []
          if isinstance(value, list):
              return [str(x).strip() for x in value if str(x).strip()]
          raw = str(value).strip()
          if not raw:
              return []
          parts = re.split(r"[,;|/\n\t]+", raw)
          out = [p.strip() for p in parts if p and p.strip()]
          if not out:
              return [raw]
          return out
  
      @staticmethod
      def _looks_noise(text_value: str) -> bool:
          if not text_value:
              return True
          if len(text_value) > 120:
              return True
          if re.fullmatch(r"[\W_]+", text_value):
              return True
          return False
  
      @staticmethod
      def _normalize_lang(lang: Optional[str]) -> Optional[str]:
          if not lang:
              return None
          token = str(lang).strip().lower().replace("-", "_")
          if not token:
              return None
          # en_us -> en, zh_cn -> zh, keep explicit zh_tw / pt_br
          if token in {"zh_tw", "pt_br"}:
              return token
          return token.split("_")[0]
  
      @staticmethod
      def _parse_request_params_language(raw: Any) -> Optional[str]:
          if raw is None:
              return None
          if isinstance(raw, dict):
              return raw.get("language")
          text_raw = str(raw).strip()
          if not text_raw:
              return None
          try:
              obj = json.loads(text_raw)
              if isinstance(obj, dict):
                  return obj.get("language")
          except Exception:
              return None
          return None
  
      @staticmethod
      def _detect_script_language(query: str) -> Tuple[Optional[str], float, str]:
          # CJK unified
          if re.search(r"[\u4e00-\u9fff]", query):
              return "zh", 0.98, "script"
          # Arabic
          if re.search(r"[\u0600-\u06FF]", query):
              return "ar", 0.98, "script"
          # Cyrillic
          if re.search(r"[\u0400-\u04FF]", query):
              return "ru", 0.95, "script"
          # Greek
          if re.search(r"[\u0370-\u03FF]", query):
              return "el", 0.95, "script"
          # Latin fallback
          if re.search(r"[a-zA-Z]", query):
              return "en", 0.55, "model"
          return None, 0.0, "default"
  
      def _resolve_query_language(
          self,
          query: str,
          log_language: Optional[str],
          request_params: Any,
          index_languages: List[str],
          primary_language: str,
      ) -> Tuple[str, float, str, bool]:
          """Resolve lang with priority: log field > request_params > script/model."""
          langs_set = set(index_languages or [])
          primary = self._normalize_lang(primary_language) or "en"
          if primary not in langs_set and langs_set:
              primary = index_languages[0]
  
          log_lang = self._normalize_lang(log_language)
          req_lang = self._normalize_lang(self._parse_request_params_language(request_params))
          conflict = bool(log_lang and req_lang and log_lang != req_lang)
  
          if log_lang and (not langs_set or log_lang in langs_set):
              return log_lang, 1.0, "log_field", conflict
  
          if req_lang and (not langs_set or req_lang in langs_set):
              return req_lang, 1.0, "request_params", conflict
  
          detected_lang, conf, source = self._detect_script_language(query)
          if detected_lang and (not langs_set or detected_lang in langs_set):
              return detected_lang, conf, source, conflict
  
          return primary, 0.3, "default", conflict
  
      @staticmethod
      def _score_product_hit(source: Dict[str, Any]) -> float:
          sales = float(source.get("sales") or 0.0)
          inventory = float(source.get("total_inventory") or 0.0)
          return math.log1p(max(sales, 0.0)) * 1.2 + math.log1p(max(inventory, 0.0)) * 0.4
  
      @staticmethod
      def _compute_rank_score(c: SuggestionCandidate) -> float:
          return (
              1.8 * math.log1p(c.query_count_30d)
              + 1.2 * math.log1p(c.query_count_7d)
              + 1.0 * math.log1p(len(c.qanchor_spu_ids))
              + 0.6 * math.log1p(len(c.title_spu_ids))
          )
  
      def _scan_products(self, tenant_id: str, batch_size: int = 500) -> List[Dict[str, Any]]:
          """Scan all product docs from tenant index using search_after."""
          from indexer.mapping_generator import get_tenant_index_name
  
          index_name = get_tenant_index_name(tenant_id)
          all_hits: List[Dict[str, Any]] = []
          search_after: Optional[List[Any]] = None
  
          while True:
              body: Dict[str, Any] = {
                  "size": batch_size,
                  "_source": ["spu_id", "title", "qanchors", "sales", "total_inventory"],
                  "sort": [{"spu_id": "asc"}],
                  "query": {"match_all": {}},
              }
              if search_after is not None:
                  body["search_after"] = search_after
  
              resp = self.es_client.client.search(index=index_name, body=body)
              hits = resp.get("hits", {}).get("hits", []) or []
              if not hits:
                  break
              all_hits.extend(hits)
              search_after = hits[-1].get("sort")
              if len(hits) < batch_size:
                  break
          return all_hits
  
      def _create_or_reset_index(self, tenant_id: str, index_languages: List[str], recreate: bool) -> str:
          index_name = get_suggestion_index_name(tenant_id)
          if recreate and self.es_client.index_exists(index_name):
              logger.info("Deleting existing suggestion index: %s", index_name)
              self.es_client.delete_index(index_name)
          if not self.es_client.index_exists(index_name):
              mapping = build_suggestion_mapping(index_languages=index_languages)
              ok = self.es_client.create_index(index_name, mapping)
              if not ok:
                  raise RuntimeError(f"Failed to create suggestion index: {index_name}")
          return index_name
  
      def rebuild_tenant_index(
          self,
          tenant_id: str,

          days: int = 365,

          recreate: bool = True,
          batch_size: int = 500,
          min_query_len: int = 1,
      ) -> Dict[str, Any]:
          tenant_loader = get_tenant_config_loader()
          tenant_cfg = tenant_loader.get_tenant_config(tenant_id)
          index_languages: List[str] = tenant_cfg.get("index_languages") or ["en", "zh"]
          primary_language: str = tenant_cfg.get("primary_language") or "en"
  
          index_name = self._create_or_reset_index(tenant_id, index_languages, recreate)
          key_to_candidate: Dict[Tuple[str, str], SuggestionCandidate] = {}
  
          # Step 1: product title/qanchors
          hits = self._scan_products(tenant_id, batch_size=batch_size)
          for hit in hits:
              src = hit.get("_source", {}) or {}
              spu_id = str(src.get("spu_id") or "")
              if not spu_id:
                  continue
              title_obj = src.get("title") or {}
              qanchor_obj = src.get("qanchors") or {}
              product_score = self._score_product_hit(src)
  
              for lang in index_languages:
                  title = ""
                  if isinstance(title_obj, dict):
                      title = str(title_obj.get(lang) or "").strip()
                  if title:
                      text_norm = self._normalize_text(title)
                      if not self._looks_noise(text_norm):
                          key = (lang, text_norm)
                          c = key_to_candidate.get(key)
                          if c is None:
                              c = SuggestionCandidate(text=title, text_norm=text_norm, lang=lang)
                              key_to_candidate[key] = c
                          c.add_product("title", spu_id=spu_id, score=product_score)
  
                  q_raw = None
                  if isinstance(qanchor_obj, dict):
                      q_raw = qanchor_obj.get(lang)
                  for q_text in self._split_qanchors(q_raw):
                      text_norm = self._normalize_text(q_text)
                      if self._looks_noise(text_norm):
                          continue
                      key = (lang, text_norm)
                      c = key_to_candidate.get(key)
                      if c is None:
                          c = SuggestionCandidate(text=q_text, text_norm=text_norm, lang=lang)
                          key_to_candidate[key] = c
                      c.add_product("qanchor", spu_id=spu_id, score=product_score + 0.6)
  
          # Step 2: query logs
          now = datetime.now(timezone.utc)
          since_30d = now - timedelta(days=days)
          since_7d = now - timedelta(days=7)
          query_sql = text(
              """
              SELECT query, language, request_params, create_time
              FROM shoplazza_search_log
              WHERE tenant_id = :tenant_id
                AND deleted = 0
                AND query IS NOT NULL
                AND query <> ''
                AND create_time >= :since_30d
              """
          )
          with self.db_engine.connect() as conn:
              rows = conn.execute(query_sql, {"tenant_id": int(tenant_id), "since_30d": since_30d}).fetchall()
  
          for row in rows:
              q = str(row.query or "").strip()
              if len(q) < min_query_len:
                  continue
              lang, conf, source, conflict = self._resolve_query_language(
                  query=q,
                  log_language=getattr(row, "language", None),
                  request_params=getattr(row, "request_params", None),
                  index_languages=index_languages,
                  primary_language=primary_language,
              )
              text_norm = self._normalize_text(q)
              if self._looks_noise(text_norm):
                  continue
              key = (lang, text_norm)
              c = key_to_candidate.get(key)
              if c is None:
                  c = SuggestionCandidate(text=q, text_norm=text_norm, lang=lang)
                  key_to_candidate[key] = c
              c.lang_confidence = max(c.lang_confidence, conf)
              c.lang_source = source if c.lang_source == "default" else c.lang_source
              c.lang_conflict = c.lang_conflict or conflict
  
              created_at = getattr(row, "create_time", None)
              if created_at is None:
                  is_7d = False
              else:
                  # DB datetime usually naive local time; compare conservatively
                  if isinstance(created_at, datetime) and created_at.tzinfo is None:
                      created_at = created_at.replace(tzinfo=timezone.utc)
                  is_7d = bool(created_at and created_at >= since_7d)
              c.add_query_log(is_7d=is_7d)
  
          # Step 3: build docs
          now_iso = datetime.now(timezone.utc).isoformat()
          docs: List[Dict[str, Any]] = []
          for (_, _), c in key_to_candidate.items():
              rank_score = self._compute_rank_score(c)
              # keep top 20 product ids by score
              top_spu_ids = [
                  item[0]
                  for item in sorted(c.top_spu_scores.items(), key=lambda kv: kv[1], reverse=True)[:20]
              ]
  
              completion_obj = {c.lang: {"input": [c.text], "weight": int(max(rank_score, 1.0) * 100)}}
              sat_obj = {c.lang: c.text}
              doc_id = f"{tenant_id}|{c.lang}|{c.text_norm}"
              docs.append(
                  {
                      "_id": doc_id,
                      "tenant_id": str(tenant_id),
                      "lang": c.lang,
                      "text": c.text,
                      "text_norm": c.text_norm,
                      "sources": sorted(c.sources),
                      "title_doc_count": len(c.title_spu_ids),
                      "qanchor_doc_count": len(c.qanchor_spu_ids),
                      "query_count_7d": c.query_count_7d,
                      "query_count_30d": c.query_count_30d,
                      "rank_score": float(rank_score),
                      "lang_confidence": float(c.lang_confidence),
                      "lang_source": c.lang_source,
                      "lang_conflict": bool(c.lang_conflict),
                      "top_spu_ids": top_spu_ids,
                      "status": 1,
                      "updated_at": now_iso,
                      "completion": completion_obj,
                      "sat": sat_obj,
                  }
              )
  
          if docs:
              result = self.es_client.bulk_index(index_name=index_name, docs=docs)
              self.es_client.refresh(index_name)
          else:
              result = {"success": 0, "failed": 0, "errors": []}
  
          return {
              "tenant_id": str(tenant_id),
              "index_name": index_name,
              "total_candidates": len(key_to_candidate),
              "indexed_docs": len(docs),
              "bulk_result": result,
          }