"""
Suggestion index builder (Phase 2).

Capabilities:
- Full rebuild to versioned index
- Atomic alias publish
- Incremental update from query logs with watermark
"""

import json
import logging
import math
import re
import unicodedata
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, Iterator, List, Optional, Tuple

from sqlalchemy import text

from config.loader import get_app_config
from config.tenant_config_loader import get_tenant_config_loader
from query.query_parser import detect_text_language_for_suggestions
from suggestion.mapping import build_suggestion_mapping
from utils.es_client import ESClient

logger = logging.getLogger(__name__)


def _index_prefix() -> str:
    return get_app_config().runtime.index_namespace or ""


def get_suggestion_alias_name(tenant_id: str) -> str:
    """Read alias for suggestion index (single source of truth)."""
    return f"{_index_prefix()}search_suggestions_tenant_{tenant_id}_current"


def get_suggestion_versioned_index_name(tenant_id: str, build_at: Optional[datetime] = None) -> str:
    """Versioned suggestion index name."""
    ts = (build_at or datetime.now(timezone.utc)).strftime("%Y%m%d%H%M%S%f")
    return f"{_index_prefix()}search_suggestions_tenant_{tenant_id}_v{ts}"


def get_suggestion_versioned_index_pattern(tenant_id: str) -> str:
    return f"{_index_prefix()}search_suggestions_tenant_{tenant_id}_v*"


def get_suggestion_meta_index_name() -> str:
    return f"{_index_prefix()}search_suggestions_meta"


@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)
    tag_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

    def add_product(self, source: str, spu_id: str) -> 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)
        elif source == "tag":
            self.tag_spu_ids.add(spu_id)

    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


@dataclass
class QueryDelta:
    tenant_id: str
    lang: str
    text: str
    text_norm: str
    delta_7d: int = 0
    delta_30d: int = 0
    lang_confidence: float = 1.0
    lang_source: str = "default"
    lang_conflict: bool = False


class SuggestionIndexBuilder:
    """Build and update suggestion index."""

    def __init__(self, es_client: ESClient, db_engine: Any):
        self.es_client = es_client
        self.db_engine = db_engine

    def _format_allocation_failure(self, index_name: str) -> str:
        health = self.es_client.wait_for_index_ready(index_name=index_name, timeout="5s")
        explain = self.es_client.get_allocation_explain(index_name=index_name)

        parts = [
            f"Suggestion index '{index_name}' was created but is not allocatable/readable yet",
            f"health_status={health.get('status')}",
            f"timed_out={health.get('timed_out')}",
        ]
        if health.get("error"):
            parts.append(f"health_error={health['error']}")

        if explain:
            unassigned = explain.get("unassigned_info") or {}
            if unassigned.get("reason"):
                parts.append(f"unassigned_reason={unassigned['reason']}")
            if unassigned.get("last_allocation_status"):
                parts.append(f"last_allocation_status={unassigned['last_allocation_status']}")

            for node in explain.get("node_allocation_decisions") or []:
                node_name = node.get("node_name") or node.get("node_id") or "unknown-node"
                for decider in node.get("deciders") or []:
                    if decider.get("decision") == "NO":
                        parts.append(
                            f"{node_name}:{decider.get('decider')}={decider.get('explanation')}"
                        )
                        return "; ".join(parts)

        return "; ".join(parts)

    def _create_fresh_versioned_index(
        self,
        tenant_id: str,
        mapping: Dict[str, Any],
        max_attempts: int = 5,
    ) -> str:
        for attempt in range(1, max_attempts + 1):
            index_name = get_suggestion_versioned_index_name(tenant_id)
            if self.es_client.index_exists(index_name):
                logger.warning(
                    "Suggestion index name collision before create for tenant=%s index=%s attempt=%s/%s",
                    tenant_id,
                    index_name,
                    attempt,
                    max_attempts,
                )
                continue

            if self.es_client.create_index(index_name, mapping):
                return index_name

            if self.es_client.index_exists(index_name):
                logger.warning(
                    "Suggestion index name collision during create for tenant=%s index=%s attempt=%s/%s",
                    tenant_id,
                    index_name,
                    attempt,
                    max_attempts,
                )
                continue

            raise RuntimeError(f"Failed to create suggestion index: {index_name}")

        raise RuntimeError(
            f"Failed to allocate a unique suggestion index name for tenant={tenant_id} after {max_attempts} attempts"
        )

    def _ensure_new_index_ready(self, index_name: str) -> None:
        health = self.es_client.wait_for_index_ready(index_name=index_name, timeout="5s")
        if health.get("ok"):
            return
        raise RuntimeError(self._format_allocation_failure(index_name))

    @staticmethod
    def _to_utc(dt: Any) -> Optional[datetime]:
        if dt is None:
            return None
        if isinstance(dt, datetime):
            if dt.tzinfo is None:
                return dt.replace(tzinfo=timezone.utc)
            return dt.astimezone(timezone.utc)
        return None

    @staticmethod
    def _normalize_text(value: str) -> str:
        text_value = unicodedata.normalize("NFKC", (value or "")).strip().lower()
        text_value = re.sub(r"\s+", " ", text_value)
        return text_value

    @staticmethod
    def _prepare_title_for_suggest(title: str, max_len: int = 120) -> str:
        """
        Keep title-derived suggestions concise:
        - keep raw title when short enough
        - for long titles, keep the leading phrase before common separators
        - fallback to hard truncate
        """
        raw = str(title or "").strip()
        if not raw:
            return ""
        if len(raw) <= max_len:
            return raw

        head = re.split(r"[，,;；|/\\\\(（\\[【]", raw, maxsplit=1)[0].strip()
        if 1 < len(head) <= max_len:
            return head

        truncated = raw[:max_len].rstrip(" ，,;；|/\\\\-—–()（）[]【】")
        return truncated or raw[:max_len]

    @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 _iter_product_tags(raw: Any) -> List[str]:
        if raw is None:
            return []
        if isinstance(raw, list):
            return [str(x).strip() for x in raw if str(x).strip()]
        s = str(raw).strip()
        if not s:
            return []
        parts = re.split(r"[，、,;|/\n\t]+", s)
        out = [p.strip() for p in parts if p and p.strip()]
        return out if out else [s]

    def _iter_multilang_product_tags(
        self,
        raw: Any,
        index_languages: List[str],
        primary_language: str,
    ) -> List[Tuple[str, str]]:
        if isinstance(raw, dict):
            pairs: List[Tuple[str, str]] = []
            for lang in index_languages:
                for tag in self._iter_product_tags(raw.get(lang)):
                    pairs.append((lang, tag))
            return pairs

        pairs = []
        for tag in self._iter_product_tags(raw):
            tag_lang, _, _ = detect_text_language_for_suggestions(
                tag,
                index_languages=index_languages,
                primary_language=primary_language,
            )
            pairs.append((tag_lang, tag))
        return pairs

    @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
        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

    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

        det_lang, conf, det_source = detect_text_language_for_suggestions(
            query,
            index_languages=index_languages,
            primary_language=primary,
        )
        if det_lang and (not langs_set or det_lang in langs_set):
            return det_lang, conf, det_source, conflict

        return primary, 0.3, "default", conflict

    @staticmethod
    def _compute_rank_score(
        query_count_30d: int,
        query_count_7d: int,
        qanchor_doc_count: int,
        title_doc_count: int,
        tag_doc_count: int = 0,
    ) -> float:
        return (
            1.8 * math.log1p(max(query_count_30d, 0))
            + 1.2 * math.log1p(max(query_count_7d, 0))
            + 1.0 * math.log1p(max(qanchor_doc_count, 0))
            + 0.85 * math.log1p(max(tag_doc_count, 0))
            + 0.6 * math.log1p(max(title_doc_count, 0))
        )

    @classmethod
    def _compute_rank_score_from_candidate(cls, c: SuggestionCandidate) -> float:
        return cls._compute_rank_score(
            query_count_30d=c.query_count_30d,
            query_count_7d=c.query_count_7d,
            qanchor_doc_count=len(c.qanchor_spu_ids),
            title_doc_count=len(c.title_spu_ids),
            tag_doc_count=len(c.tag_spu_ids),
        )

    def _iter_products(self, tenant_id: str, batch_size: int = 500) -> Iterator[Dict[str, Any]]:
        """Stream 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)
        search_after: Optional[List[Any]] = None
        print(f"[DEBUG] Python using index: {index_name} for tenant {tenant_id}")
        total_processed = 0
        while True:
            body: Dict[str, Any] = {
                "size": batch_size,
                "_source": ["id", "spu_id", "title", "qanchors", "enriched_tags"],
                "sort": [
                    {"spu_id": {"order": "asc", "missing": "_last"}},
                    {"id.keyword": {"order": "asc", "missing": "_last"}},
                ],
                "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
            for hit in hits:
                total_processed += 1
                yield hit
            search_after = hits[-1].get("sort")
            if len(hits) < batch_size:
                break
        print(f"[DEBUG] Python processed total products: {total_processed} for tenant {tenant_id}")
    

    def _iter_query_log_rows(
        self,
        tenant_id: str,
        since: datetime,
        until: datetime,
        fetch_size: int = 2000,
    ) -> Iterator[Any]:
        """Stream search logs from MySQL with bounded time range."""
        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_time
              AND create_time < :until_time
            ORDER BY create_time ASC
            """
        )

        with self.db_engine.connect().execution_options(stream_results=True) as conn:
            result = conn.execute(
                query_sql,
                {
                    "tenant_id": int(tenant_id),
                    "since_time": since,
                    "until_time": until,
                },
            )
            while True:
                rows = result.fetchmany(fetch_size)
                if not rows:
                    break
                for row in rows:
                    yield row

    def _ensure_meta_index(self) -> str:
        meta_index = get_suggestion_meta_index_name()
        if self.es_client.index_exists(meta_index):
            return meta_index
        body = {
            "settings": {
                "number_of_shards": 1,
                "number_of_replicas": 0,
                "refresh_interval": "1s",
            },
            "mappings": {
                "properties": {
                    "tenant_id": {"type": "keyword"},
                    "active_alias": {"type": "keyword"},
                    "active_index": {"type": "keyword"},
                    "last_full_build_at": {"type": "date"},
                    "last_incremental_build_at": {"type": "date"},
                    "last_incremental_watermark": {"type": "date"},
                    "updated_at": {"type": "date"},
                }
            },
        }
        if not self.es_client.create_index(meta_index, body):
            raise RuntimeError(f"Failed to create suggestion meta index: {meta_index}")
        return meta_index

    def _get_meta(self, tenant_id: str) -> Dict[str, Any]:
        meta_index = self._ensure_meta_index()
        try:
            resp = self.es_client.client.get(index=meta_index, id=str(tenant_id))
            return resp.get("_source", {}) or {}
        except Exception:
            return {}

    def _upsert_meta(self, tenant_id: str, patch: Dict[str, Any]) -> None:
        meta_index = self._ensure_meta_index()
        current = self._get_meta(tenant_id)
        now_iso = datetime.now(timezone.utc).isoformat()
        merged = {
            "tenant_id": str(tenant_id),
            **current,
            **patch,
            "updated_at": now_iso,
        }
        self.es_client.client.index(index=meta_index, id=str(tenant_id), document=merged, refresh="wait_for")

    def _cleanup_old_versions(self, tenant_id: str, keep_versions: int, protected_indices: Optional[List[str]] = None) -> List[str]:
        if keep_versions < 1:
            keep_versions = 1
        protected = set(protected_indices or [])
        pattern = get_suggestion_versioned_index_pattern(tenant_id)
        all_indices = self.es_client.list_indices(pattern)
        if len(all_indices) <= keep_versions:
            return []

        # Names are timestamp-ordered by suffix; keep newest N.
        kept = set(sorted(all_indices)[-keep_versions:])
        dropped: List[str] = []
        for idx in sorted(all_indices):
            if idx in kept or idx in protected:
                continue
            if self.es_client.delete_index(idx):
                dropped.append(idx)
        return dropped

    def _publish_alias(self, tenant_id: str, index_name: str, keep_versions: int = 2) -> Dict[str, Any]:
        alias_name = get_suggestion_alias_name(tenant_id)
        current_indices = self.es_client.get_alias_indices(alias_name)

        actions: List[Dict[str, Any]] = []
        for idx in current_indices:
            actions.append({"remove": {"index": idx, "alias": alias_name}})
        actions.append({"add": {"index": index_name, "alias": alias_name}})

        if not self.es_client.update_aliases(actions):
            raise RuntimeError(f"Failed to publish alias {alias_name} -> {index_name}")

        dropped = self._cleanup_old_versions(
            tenant_id=tenant_id,
            keep_versions=keep_versions,
            protected_indices=[index_name],
        )

        self._upsert_meta(
            tenant_id,
            {
                "active_alias": alias_name,
                "active_index": index_name,
            },
        )

        return {
            "alias": alias_name,
            "previous_indices": current_indices,
            "current_index": index_name,
            "dropped_old_indices": dropped,
        }

    def _resolve_incremental_target_index(self, tenant_id: str) -> Optional[str]:
        """Resolve active suggestion index for incremental updates (alias only)."""
        alias_name = get_suggestion_alias_name(tenant_id)
        aliased = self.es_client.get_alias_indices(alias_name)
        if aliased:
            # alias should map to one index in this design
            return sorted(aliased)[-1]
        return None

    def _build_full_candidates(
        self,
        tenant_id: str,
        index_languages: List[str],
        primary_language: str,
        days: int,
        batch_size: int,
        min_query_len: int,
    ) -> Dict[Tuple[str, str], SuggestionCandidate]:
        key_to_candidate: Dict[Tuple[str, str], SuggestionCandidate] = {}

        # Step 1: product title/qanchors
        for hit in self._iter_products(tenant_id, batch_size=batch_size):
            src = hit.get("_source", {}) or {}
            product_id = str(src.get("spu_id") or src.get("id") or hit.get("_id") or "")
            if not product_id:
                continue
            title_obj = src.get("title") or {}
            qanchor_obj = src.get("qanchors") or {}

            for lang in index_languages:
                title = ""
                if isinstance(title_obj, dict):
                    title = self._prepare_title_for_suggest(title_obj.get(lang) or "")
                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=product_id)

                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=product_id)

            for tag_lang, tag in self._iter_multilang_product_tags(
                src.get("enriched_tags"),
                index_languages=index_languages,
                primary_language=primary_language,
            ):
                text_norm = self._normalize_text(tag)
                if self._looks_noise(text_norm):
                    continue
                key = (tag_lang, text_norm)
                c = key_to_candidate.get(key)
                if c is None:
                    c = SuggestionCandidate(text=tag, text_norm=text_norm, lang=tag_lang)
                    key_to_candidate[key] = c
                c.add_product("tag", spu_id=product_id)

        # Step 2: query logs
        now = datetime.now(timezone.utc)
        since = now - timedelta(days=days)
        since_7d = now - timedelta(days=7)

        for row in self._iter_query_log_rows(tenant_id=tenant_id, since=since, until=now):
            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 = self._to_utc(getattr(row, "create_time", None))
            is_7d = bool(created_at and created_at >= since_7d)
            c.add_query_log(is_7d=is_7d)

        return key_to_candidate

    def _candidate_to_doc(self, tenant_id: str, c: SuggestionCandidate, now_iso: str) -> Dict[str, Any]:
        rank_score = self._compute_rank_score_from_candidate(c)
        completion_obj = {c.lang: {"input": [c.text], "weight": int(max(rank_score, 1.0) * 100)}}
        sat_obj = {c.lang: c.text}
        return {
            "_id": f"{tenant_id}|{c.lang}|{c.text_norm}",
            "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),
            "tag_doc_count": len(c.tag_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),
            "status": 1,
            "updated_at": now_iso,
            "completion": completion_obj,
            "sat": sat_obj,
        }

    def rebuild_tenant_index(
        self,
        tenant_id: str,
        days: int = 365,
        batch_size: int = 500,
        min_query_len: int = 1,
        publish_alias: bool = True,
        keep_versions: int = 2,
    ) -> Dict[str, Any]:
        """
        Full rebuild.

        Phase2 default behavior:
        - write to versioned index
        - atomically publish alias
        """
        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"

        alias_publish: Optional[Dict[str, Any]] = None
        index_name: Optional[str] = None
        try:
            mapping = build_suggestion_mapping(index_languages=index_languages)
            index_name = self._create_fresh_versioned_index(
                tenant_id=tenant_id,
                mapping=mapping,
            )
            self._ensure_new_index_ready(index_name)

            key_to_candidate = self._build_full_candidates(
                tenant_id=tenant_id,
                index_languages=index_languages,
                primary_language=primary_language,
                days=days,
                batch_size=batch_size,
                min_query_len=min_query_len,
            )

            now_iso = datetime.now(timezone.utc).isoformat()
            docs = [self._candidate_to_doc(tenant_id, c, now_iso) for c in key_to_candidate.values()]

            if docs:
                bulk_result = self.es_client.bulk_index(index_name=index_name, docs=docs)
                self.es_client.refresh(index_name)
            else:
                bulk_result = {"success": 0, "failed": 0, "errors": []}

            if publish_alias:
                alias_publish = self._publish_alias(
                    tenant_id=tenant_id,
                    index_name=index_name,
                    keep_versions=keep_versions,
                )

            now_utc = datetime.now(timezone.utc).isoformat()
            meta_patch: Dict[str, Any] = {
                "last_full_build_at": now_utc,
                "last_incremental_watermark": now_utc,
            }
            if publish_alias:
                meta_patch["active_index"] = index_name
                meta_patch["active_alias"] = get_suggestion_alias_name(tenant_id)
            self._upsert_meta(tenant_id, meta_patch)

            return {
                "mode": "full",
                "tenant_id": str(tenant_id),
                "index_name": index_name,
                "alias_published": bool(alias_publish),
                "alias_publish": alias_publish,
                "total_candidates": len(key_to_candidate),
                "indexed_docs": len(docs),
                "bulk_result": bulk_result,
            }
        except Exception:
            if index_name and not alias_publish:
                self.es_client.delete_index(index_name)
            raise

    def _build_incremental_deltas(
        self,
        tenant_id: str,
        index_languages: List[str],
        primary_language: str,
        since: datetime,
        until: datetime,
        min_query_len: int,
    ) -> Dict[Tuple[str, str], QueryDelta]:
        now = datetime.now(timezone.utc)
        since_7d = now - timedelta(days=7)
        deltas: Dict[Tuple[str, str], QueryDelta] = {}

        for row in self._iter_query_log_rows(tenant_id=tenant_id, since=since, until=until):
            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)
            item = deltas.get(key)
            if item is None:
                item = QueryDelta(
                    tenant_id=str(tenant_id),
                    lang=lang,
                    text=q,
                    text_norm=text_norm,
                    lang_confidence=conf,
                    lang_source=source,
                    lang_conflict=conflict,
                )
                deltas[key] = item

            created_at = self._to_utc(getattr(row, "create_time", None))
            item.delta_30d += 1
            if created_at and created_at >= since_7d:
                item.delta_7d += 1

            if conf > item.lang_confidence:
                item.lang_confidence = conf
                item.lang_source = source
            item.lang_conflict = item.lang_conflict or conflict

        return deltas

    def _delta_to_upsert_doc(self, delta: QueryDelta, now_iso: str) -> Dict[str, Any]:
        rank_score = self._compute_rank_score(
            query_count_30d=delta.delta_30d,
            query_count_7d=delta.delta_7d,
            qanchor_doc_count=0,
            title_doc_count=0,
            tag_doc_count=0,
        )
        return {
            "tenant_id": delta.tenant_id,
            "lang": delta.lang,
            "text": delta.text,
            "text_norm": delta.text_norm,
            "sources": ["query_log"],
            "title_doc_count": 0,
            "qanchor_doc_count": 0,
            "tag_doc_count": 0,
            "query_count_7d": delta.delta_7d,
            "query_count_30d": delta.delta_30d,
            "rank_score": float(rank_score),
            "lang_confidence": float(delta.lang_confidence),
            "lang_source": delta.lang_source,
            "lang_conflict": bool(delta.lang_conflict),
            "status": 1,
            "updated_at": now_iso,
            "completion": {
                delta.lang: {
                    "input": [delta.text],
                    "weight": int(max(rank_score, 1.0) * 100),
                }
            },
            "sat": {delta.lang: delta.text},
        }

    @staticmethod
    def _build_incremental_update_script() -> str:
        return """
            if (ctx._source == null || ctx._source.isEmpty()) {
                ctx._source = params.upsert;
                return;
            }

            if (ctx._source.query_count_30d == null) { ctx._source.query_count_30d = 0; }
            if (ctx._source.query_count_7d == null) { ctx._source.query_count_7d = 0; }
            if (ctx._source.qanchor_doc_count == null) { ctx._source.qanchor_doc_count = 0; }
            if (ctx._source.title_doc_count == null) { ctx._source.title_doc_count = 0; }
            if (ctx._source.tag_doc_count == null) { ctx._source.tag_doc_count = 0; }

            ctx._source.query_count_30d += params.delta_30d;
            ctx._source.query_count_7d += params.delta_7d;

            if (ctx._source.sources == null) { ctx._source.sources = new ArrayList(); }
            if (!ctx._source.sources.contains('query_log')) { ctx._source.sources.add('query_log'); }

            if (ctx._source.lang_conflict == null) { ctx._source.lang_conflict = false; }
            ctx._source.lang_conflict = ctx._source.lang_conflict || params.lang_conflict;

            if (ctx._source.lang_confidence == null || params.lang_confidence > ctx._source.lang_confidence) {
                ctx._source.lang_confidence = params.lang_confidence;
                ctx._source.lang_source = params.lang_source;
            }

            int q30 = ctx._source.query_count_30d;
            int q7 = ctx._source.query_count_7d;
            int qa = ctx._source.qanchor_doc_count;
            int td = ctx._source.title_doc_count;
            int tg = ctx._source.tag_doc_count;

            double score = 1.8 * Math.log(1 + q30)
                         + 1.2 * Math.log(1 + q7)
                         + 1.0 * Math.log(1 + qa)
                         + 0.85 * Math.log(1 + tg)
                         + 0.6 * Math.log(1 + td);
            ctx._source.rank_score = score;
            ctx._source.status = 1;
            ctx._source.updated_at = params.now_iso;
            ctx._source.text = params.text;
            ctx._source.lang = params.lang;
            ctx._source.text_norm = params.text_norm;

            if (ctx._source.completion == null) { ctx._source.completion = new HashMap(); }
            Map c = new HashMap();
            c.put('input', params.completion_input);
            c.put('weight', params.completion_weight);
            ctx._source.completion.put(params.lang, c);

            if (ctx._source.sat == null) { ctx._source.sat = new HashMap(); }
            ctx._source.sat.put(params.lang, params.text);
        """

    def _build_incremental_actions(self, target_index: str, deltas: Dict[Tuple[str, str], QueryDelta]) -> List[Dict[str, Any]]:
        now_iso = datetime.now(timezone.utc).isoformat()
        script_source = self._build_incremental_update_script()
        actions: List[Dict[str, Any]] = []

        for delta in deltas.values():
            upsert_doc = self._delta_to_upsert_doc(delta=delta, now_iso=now_iso)
            upsert_rank = float(upsert_doc.get("rank_score") or 0.0)
            action = {
                "_op_type": "update",
                "_index": target_index,
                "_id": f"{delta.tenant_id}|{delta.lang}|{delta.text_norm}",
                "scripted_upsert": True,
                "script": {
                    "lang": "painless",
                    "source": script_source,
                    "params": {
                        "delta_30d": int(delta.delta_30d),
                        "delta_7d": int(delta.delta_7d),
                        "lang_confidence": float(delta.lang_confidence),
                        "lang_source": delta.lang_source,
                        "lang_conflict": bool(delta.lang_conflict),
                        "now_iso": now_iso,
                        "lang": delta.lang,
                        "text": delta.text,
                        "text_norm": delta.text_norm,
                        "completion_input": [delta.text],
                        "completion_weight": int(max(upsert_rank, 1.0) * 100),
                        "upsert": upsert_doc,
                    },
                },
                "upsert": upsert_doc,
            }
            actions.append(action)

        return actions

    def incremental_update_tenant_index(
        self,
        tenant_id: str,
        min_query_len: int = 1,
        fallback_days: int = 7,
        overlap_minutes: int = 30,
        bootstrap_if_missing: bool = True,
        bootstrap_days: int = 30,
        batch_size: int = 500,
    ) -> 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"

        target_index = self._resolve_incremental_target_index(tenant_id)
        if not target_index:
            if not bootstrap_if_missing:
                raise RuntimeError(
                    f"No active suggestion index for tenant={tenant_id}. "
                    "Run full rebuild first or enable bootstrap_if_missing."
                )
            full_result = self.rebuild_tenant_index(
                tenant_id=tenant_id,
                days=bootstrap_days,
                batch_size=batch_size,
                min_query_len=min_query_len,
                publish_alias=True
            )
            return {
                "mode": "incremental",
                "tenant_id": str(tenant_id),
                "bootstrapped": True,
                "bootstrap_result": full_result,
            }

        meta = self._get_meta(tenant_id)
        watermark_raw = meta.get("last_incremental_watermark") or meta.get("last_full_build_at")
        now = datetime.now(timezone.utc)
        default_since = now - timedelta(days=fallback_days)
        since = None
        if isinstance(watermark_raw, str) and watermark_raw.strip():
            try:
                since = self._to_utc(datetime.fromisoformat(watermark_raw.replace("Z", "+00:00")))
            except Exception:
                since = None
        if since is None:
            since = default_since
        since = since - timedelta(minutes=max(overlap_minutes, 0))
        if since < default_since:
            since = default_since

        deltas = self._build_incremental_deltas(
            tenant_id=tenant_id,
            index_languages=index_languages,
            primary_language=primary_language,
            since=since,
            until=now,
            min_query_len=min_query_len,
        )

        actions = self._build_incremental_actions(target_index=target_index, deltas=deltas)
        bulk_result = self.es_client.bulk_actions(actions)
        self.es_client.refresh(target_index)

        now_iso = now.isoformat()
        self._upsert_meta(
            tenant_id,
            {
                "last_incremental_build_at": now_iso,
                "last_incremental_watermark": now_iso,
                "active_index": target_index,
                "active_alias": get_suggestion_alias_name(tenant_id),
            },
        )

        return {
            "mode": "incremental",
            "tenant_id": str(tenant_id),
            "target_index": target_index,
            "query_window": {
                "since": since.isoformat(),
                "until": now_iso,
                "overlap_minutes": int(overlap_minutes),
            },
            "updated_terms": len(deltas),
            "bulk_result": bulk_result,
        }