#!/usr/bin/env python3
  """
  商品内容理解与属性补充模块（product_enrich）
  
  提供基于 LLM 的商品锚文本 / 语义属性 / 标签等分析能力，
  供 indexer 与 API 在内存中调用（不再负责 CSV 读写）。
  """
  
  import os
  import json
  import logging

  import re

  import time
  import hashlib

  import uuid
  import threading

  from dataclasses import dataclass, field

  from collections import OrderedDict

  from datetime import datetime

  from concurrent.futures import ThreadPoolExecutor

  from typing import List, Dict, Tuple, Any, Optional, FrozenSet

  
  import redis
  import requests
  from pathlib import Path
  

  from config.loader import get_app_config

  from config.tenant_config_loader import SOURCE_LANG_CODE_MAP

  from indexer.product_enrich_prompts import (
      SYSTEM_MESSAGE,
      USER_INSTRUCTION_TEMPLATE,
      LANGUAGE_MARKDOWN_TABLE_HEADERS,
      SHARED_ANALYSIS_INSTRUCTION,

      CATEGORY_TAXONOMY_PROFILES,

  )

  
  # 配置
  BATCH_SIZE = 20

  # enrich-content LLM 批次并发 worker 上限（线程池；仅对 uncached batch 并发）
  _APP_CONFIG = get_app_config()
  CONTENT_UNDERSTANDING_MAX_WORKERS = int(_APP_CONFIG.product_enrich.max_workers)

  # 华北2（北京）：https://dashscope.aliyuncs.com/compatible-mode/v1
  # 新加坡：https://dashscope-intl.aliyuncs.com/compatible-mode/v1
  # 美国（弗吉尼亚）：https://dashscope-us.aliyuncs.com/compatible-mode/v1
  API_BASE_URL = "https://dashscope-us.aliyuncs.com/compatible-mode/v1"
  MODEL_NAME = "qwen-flash"
  API_KEY = os.environ.get("DASHSCOPE_API_KEY")
  MAX_RETRIES = 3
  RETRY_DELAY = 5  # 秒
  REQUEST_TIMEOUT = 180  # 秒

  LOGGED_SHARED_CONTEXT_CACHE_SIZE = 256

  PROMPT_INPUT_MIN_ZH_CHARS = 20
  PROMPT_INPUT_MAX_ZH_CHARS = 100
  PROMPT_INPUT_MIN_WORDS = 16
  PROMPT_INPUT_MAX_WORDS = 80

  
  # 日志路径
  OUTPUT_DIR = Path("output_logs")
  LOG_DIR = OUTPUT_DIR / "logs"
  
  # 设置独立日志（不影响全局 indexer.log）
  LOG_DIR.mkdir(parents=True, exist_ok=True)
  timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
  log_file = LOG_DIR / f"product_enrich_{timestamp}.log"
  verbose_log_file = LOG_DIR / "product_enrich_verbose.log"

  _logged_shared_context_keys: "OrderedDict[str, None]" = OrderedDict()

  _logged_shared_context_lock = threading.Lock()
  
  _content_understanding_executor: Optional[ThreadPoolExecutor] = None
  _content_understanding_executor_lock = threading.Lock()
  
  
  def _get_content_understanding_executor() -> ThreadPoolExecutor:
      """
      使用模块级单例线程池，避免同一进程内多次请求叠加创建线程池导致并发失控。
      """
      global _content_understanding_executor
      with _content_understanding_executor_lock:
          if _content_understanding_executor is None:
              _content_understanding_executor = ThreadPoolExecutor(
                  max_workers=CONTENT_UNDERSTANDING_MAX_WORKERS,
                  thread_name_prefix="product-enrich-llm",
              )
          return _content_understanding_executor

  
  # 主日志 logger：执行流程、批次信息等
  logger = logging.getLogger("product_enrich")
  logger.setLevel(logging.INFO)
  
  if not logger.handlers:
      formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
  
      file_handler = logging.FileHandler(log_file, encoding="utf-8")
      file_handler.setFormatter(formatter)
  
      stream_handler = logging.StreamHandler()
      stream_handler.setFormatter(formatter)
  
      logger.addHandler(file_handler)
      logger.addHandler(stream_handler)
  
      # 避免日志向根 logger 传播，防止写入 logs/indexer.log 等其他文件
      logger.propagate = False
  
  # 详尽日志 logger：专门记录 LLM 请求与响应
  verbose_logger = logging.getLogger("product_enrich_verbose")
  verbose_logger.setLevel(logging.INFO)
  
  if not verbose_logger.handlers:
      verbose_formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
      verbose_file_handler = logging.FileHandler(verbose_log_file, encoding="utf-8")
      verbose_file_handler.setFormatter(verbose_formatter)
      verbose_logger.addHandler(verbose_file_handler)
      verbose_logger.propagate = False
  

  logger.info("Verbose LLM logs are written to: %s", verbose_log_file)
  

  
  # Redis 缓存（用于 anchors / 语义属性）

  _REDIS_CONFIG = _APP_CONFIG.infrastructure.redis

  ANCHOR_CACHE_PREFIX = _REDIS_CONFIG.anchor_cache_prefix
  ANCHOR_CACHE_EXPIRE_DAYS = int(_REDIS_CONFIG.anchor_cache_expire_days)

  _anchor_redis: Optional[redis.Redis] = None
  
  try:
      _anchor_redis = redis.Redis(

          host=_REDIS_CONFIG.host,
          port=_REDIS_CONFIG.port,
          password=_REDIS_CONFIG.password,

          decode_responses=True,

          socket_timeout=_REDIS_CONFIG.socket_timeout,
          socket_connect_timeout=_REDIS_CONFIG.socket_connect_timeout,
          retry_on_timeout=_REDIS_CONFIG.retry_on_timeout,

          health_check_interval=10,
      )
      _anchor_redis.ping()
      logger.info("Redis cache initialized for product anchors and semantic attributes")
  except Exception as e:
      logger.warning(f"Failed to initialize Redis for anchors cache: {e}")
      _anchor_redis = None
  

  _missing_prompt_langs = sorted(set(SOURCE_LANG_CODE_MAP) - set(LANGUAGE_MARKDOWN_TABLE_HEADERS))
  if _missing_prompt_langs:
      raise RuntimeError(
          f"Missing product_enrich prompt config for languages: {_missing_prompt_langs}"
      )

  # 多值字段分隔

  _MULTI_VALUE_FIELD_SPLIT_RE = re.compile(r"[，、,;|/\n\t]+")

  # 表格单元格中视为「无内容」的占位
  _MARKDOWN_EMPTY_CELL_LITERALS: Tuple[str, ...] = ("-","–", "—", "none", "null", "n/a", "无")
  _MARKDOWN_EMPTY_CELL_TOKENS_CF: FrozenSet[str] = frozenset(
      lit.casefold() for lit in _MARKDOWN_EMPTY_CELL_LITERALS
  )
  
  def _normalize_markdown_table_cell(raw: Optional[str]) -> str:
      """strip；将占位符统一视为空字符串。"""
      s = str(raw or "").strip()
      if not s:
          return ""
      if s.casefold() in _MARKDOWN_EMPTY_CELL_TOKENS_CF:
          return ""
      return s

  _CORE_INDEX_LANGUAGES = ("zh", "en")

  _DEFAULT_ENRICHMENT_SCOPES = ("generic", "category_taxonomy")
  _DEFAULT_CATEGORY_TAXONOMY_PROFILE = "apparel"

  _CONTENT_ANALYSIS_ATTRIBUTE_FIELD_MAP = (

      ("tags", "enriched_tags"),
      ("target_audience", "target_audience"),
      ("usage_scene", "usage_scene"),
      ("season", "season"),
      ("key_attributes", "key_attributes"),
      ("material", "material"),
      ("features", "features"),
  )

  _CONTENT_ANALYSIS_RESULT_FIELDS = (

      "title",
      "category_path",
      "tags",
      "target_audience",
      "usage_scene",
      "season",
      "key_attributes",
      "material",
      "features",
      "anchor_text",
  )

  _CONTENT_ANALYSIS_MEANINGFUL_FIELDS = (

      "tags",

      "target_audience",
      "usage_scene",
      "season",
      "key_attributes",
      "material",
      "features",

      "anchor_text",

  )

  _CONTENT_ANALYSIS_FIELD_ALIASES = {

      "tags": ("tags", "enriched_tags"),
  }

  _CONTENT_ANALYSIS_QUALITY_FIELDS = ("title", "category_path", "anchor_text")

  
  
  @dataclass(frozen=True)
  class AnalysisSchema:
      name: str
      shared_instruction: str
      markdown_table_headers: Dict[str, List[str]]
      result_fields: Tuple[str, ...]
      meaningful_fields: Tuple[str, ...]

      cache_version: str = "v1"

      field_aliases: Dict[str, Tuple[str, ...]] = field(default_factory=dict)

      quality_fields: Tuple[str, ...] = ()
  
      def get_headers(self, target_lang: str) -> Optional[List[str]]:

          return self.markdown_table_headers.get(target_lang)

  
  
  _ANALYSIS_SCHEMAS: Dict[str, AnalysisSchema] = {
      "content": AnalysisSchema(
          name="content",
          shared_instruction=SHARED_ANALYSIS_INSTRUCTION,
          markdown_table_headers=LANGUAGE_MARKDOWN_TABLE_HEADERS,
          result_fields=_CONTENT_ANALYSIS_RESULT_FIELDS,
          meaningful_fields=_CONTENT_ANALYSIS_MEANINGFUL_FIELDS,

          cache_version="v2",

          field_aliases=_CONTENT_ANALYSIS_FIELD_ALIASES,
          quality_fields=_CONTENT_ANALYSIS_QUALITY_FIELDS,
      ),

  }
  

  def _build_taxonomy_profile_schema(profile: str, config: Dict[str, Any]) -> AnalysisSchema:

      return AnalysisSchema(
          name=f"taxonomy:{profile}",
          shared_instruction=config["shared_instruction"],

          markdown_table_headers=config["markdown_table_headers"],
          result_fields=tuple(field["key"] for field in config["fields"]),
          meaningful_fields=tuple(field["key"] for field in config["fields"]),

          cache_version="v1",

      )
  
  
  _CATEGORY_TAXONOMY_PROFILE_SCHEMAS: Dict[str, AnalysisSchema] = {
      profile: _build_taxonomy_profile_schema(profile, config)
      for profile, config in CATEGORY_TAXONOMY_PROFILES.items()

  }
  

  _CATEGORY_TAXONOMY_PROFILE_ATTRIBUTE_FIELD_MAPS: Dict[str, Tuple[Tuple[str, str], ...]] = {

      profile: tuple((field["key"], field["label"]) for field in config["fields"])
      for profile, config in CATEGORY_TAXONOMY_PROFILES.items()

  }
  

  def get_supported_category_taxonomy_profiles() -> Tuple[str, ...]:
      return tuple(_CATEGORY_TAXONOMY_PROFILE_SCHEMAS.keys())
  
  

  def _normalize_category_taxonomy_profile(category_taxonomy_profile: Optional[str] = None) -> str:
      profile = str(category_taxonomy_profile or _DEFAULT_CATEGORY_TAXONOMY_PROFILE).strip()
      if profile not in _CATEGORY_TAXONOMY_PROFILE_SCHEMAS:

          supported = ", ".join(get_supported_category_taxonomy_profiles())
          raise ValueError(
              f"Unsupported category_taxonomy_profile: {profile}. Supported profiles: {supported}"
          )

      return profile

  def _get_analysis_schema(
      analysis_kind: str,
      *,
      category_taxonomy_profile: Optional[str] = None,
  ) -> AnalysisSchema:
      if analysis_kind == "content":
          return _ANALYSIS_SCHEMAS["content"]
      if analysis_kind == "taxonomy":
          profile = _normalize_category_taxonomy_profile(category_taxonomy_profile)
          return _CATEGORY_TAXONOMY_PROFILE_SCHEMAS[profile]
      raise ValueError(f"Unsupported analysis_kind: {analysis_kind}")
  
  
  def _get_taxonomy_attribute_field_map(
      category_taxonomy_profile: Optional[str] = None,
  ) -> Tuple[Tuple[str, str], ...]:
      profile = _normalize_category_taxonomy_profile(category_taxonomy_profile)
      return _CATEGORY_TAXONOMY_PROFILE_ATTRIBUTE_FIELD_MAPS[profile]
  
  

  def _normalize_enrichment_scopes(
      enrichment_scopes: Optional[List[str]] = None,

  ) -> Tuple[str, ...]:

      requested = _DEFAULT_ENRICHMENT_SCOPES if not enrichment_scopes else tuple(enrichment_scopes)

      normalized: List[str] = []
      seen = set()

      for enrichment_scope in requested:
          scope = str(enrichment_scope).strip()
          if scope not in {"generic", "category_taxonomy"}:
              raise ValueError(f"Unsupported enrichment_scope: {scope}")
          if scope in seen:

              continue

          seen.add(scope)
          normalized.append(scope)

      return tuple(normalized)
  
  

  def split_multi_value_field(text: Optional[str]) -> List[str]:
      """将 LLM/业务中的多值字符串拆成短语列表（strip 后去空）。"""
      if text is None:
          return []
      s = str(text).strip()
      if not s:
          return []
      return [p.strip() for p in _MULTI_VALUE_FIELD_SPLIT_RE.split(s) if p.strip()]
  
  

  def _append_lang_phrase_map(target: Dict[str, List[str]], lang: str, raw_value: Any) -> None:
      parts = split_multi_value_field(raw_value)
      if not parts:
          return
      existing = target.get(lang) or []
      merged = list(dict.fromkeys([str(x).strip() for x in existing if str(x).strip()] + parts))
      if merged:
          target[lang] = merged
  
  

  def _get_or_create_named_value_entry(
      target: List[Dict[str, Any]],
      name: str,
      *,
      default_value: Optional[Dict[str, Any]] = None,
  ) -> Dict[str, Any]:
      for item in target:
          if item.get("name") == name:
              value = item.get("value")
              if isinstance(value, dict):
                  return item
              break
  
      entry = {"name": name, "value": default_value or {}}
      target.append(entry)
      return entry
  
  
  def _append_named_lang_phrase_map(

      target: List[Dict[str, Any]],
      name: str,
      lang: str,
      raw_value: Any,
  ) -> None:

      entry = _get_or_create_named_value_entry(target, name=name, default_value={})
      _append_lang_phrase_map(entry["value"], lang=lang, raw_value=raw_value)

  def _get_product_id(product: Dict[str, Any]) -> str:
      return str(product.get("id") or product.get("spu_id") or "").strip()
  
  

  def _get_analysis_field_aliases(field_name: str, schema: AnalysisSchema) -> Tuple[str, ...]:
      return schema.field_aliases.get(field_name, (field_name,))

  def _get_analysis_field_value(row: Dict[str, Any], field_name: str, schema: AnalysisSchema) -> Any:
      for alias in _get_analysis_field_aliases(field_name, schema):

          if alias in row:
              return row.get(alias)
      return None
  
  
  def _has_meaningful_value(value: Any) -> bool:
      if value is None:
          return False
      if isinstance(value, str):
          return bool(value.strip())
      if isinstance(value, dict):
          return any(_has_meaningful_value(v) for v in value.values())
      if isinstance(value, list):
          return any(_has_meaningful_value(v) for v in value)
      return bool(value)
  
  
  def _make_empty_analysis_result(
      product: Dict[str, Any],
      target_lang: str,

      schema: AnalysisSchema,

      error: Optional[str] = None,
  ) -> Dict[str, Any]:
      result = {
          "id": _get_product_id(product),
          "lang": target_lang,
          "title_input": str(product.get("title") or "").strip(),
      }

      for field in schema.result_fields:

          result[field] = ""
      if error:
          result["error"] = error
      return result
  
  
  def _normalize_analysis_result(
      result: Dict[str, Any],
      product: Dict[str, Any],
      target_lang: str,

      schema: AnalysisSchema,

  ) -> Dict[str, Any]:

      normalized = _make_empty_analysis_result(product, target_lang, schema)

      if not isinstance(result, dict):
          return normalized
  
      normalized["lang"] = str(result.get("lang") or target_lang).strip() or target_lang

      normalized["title_input"] = str(
          product.get("title") or result.get("title_input") or ""
      ).strip()
  

      for field in schema.result_fields:
          normalized[field] = str(_get_analysis_field_value(result, field, schema) or "").strip()

  
      if result.get("error"):
          normalized["error"] = str(result.get("error"))
      return normalized
  
  

  def _has_meaningful_analysis_content(result: Dict[str, Any], schema: AnalysisSchema) -> bool:
      return any(_has_meaningful_value(result.get(field)) for field in schema.meaningful_fields)
  
  
  def _append_analysis_attributes(
      target: List[Dict[str, Any]],
      row: Dict[str, Any],
      lang: str,
      schema: AnalysisSchema,
      field_map: Tuple[Tuple[str, str], ...],
  ) -> None:
      for source_name, output_name in field_map:
          raw = _get_analysis_field_value(row, source_name, schema)
          if not raw:
              continue
          _append_named_lang_phrase_map(
              target,
              name=output_name,
              lang=lang,
              raw_value=raw,
          )

  def _apply_index_content_row(result: Dict[str, Any], row: Dict[str, Any], lang: str) -> None:
      if not row or row.get("error"):
          return
  

      content_schema = _get_analysis_schema("content")
      anchor_text = str(_get_analysis_field_value(row, "anchor_text", content_schema) or "").strip()

      if anchor_text:
          _append_lang_phrase_map(result["qanchors"], lang=lang, raw_value=anchor_text)
  

      for source_name, output_name in _CONTENT_ANALYSIS_ATTRIBUTE_FIELD_MAP:
          raw = _get_analysis_field_value(row, source_name, content_schema)

          if not raw:
              continue

          _append_named_lang_phrase_map(

              result["enriched_attributes"],
              name=output_name,
              lang=lang,
              raw_value=raw,
          )
          if output_name == "enriched_tags":

              _append_lang_phrase_map(result["enriched_tags"], lang=lang, raw_value=raw)
  
  

  def _apply_index_taxonomy_row(
      result: Dict[str, Any],
      row: Dict[str, Any],
      lang: str,
      *,
      category_taxonomy_profile: Optional[str] = None,
  ) -> None:

      if not row or row.get("error"):
          return
  
      _append_analysis_attributes(
          result["enriched_taxonomy_attributes"],
          row=row,
          lang=lang,

          schema=_get_analysis_schema(
              "taxonomy",
              category_taxonomy_profile=category_taxonomy_profile,
          ),
          field_map=_get_taxonomy_attribute_field_map(category_taxonomy_profile),

      )
  
  

  def _normalize_index_content_item(item: Dict[str, Any]) -> Dict[str, str]:

      item_id = _get_product_id(item)

      return {
          "id": item_id,
          "title": str(item.get("title") or "").strip(),
          "brief": str(item.get("brief") or "").strip(),
          "description": str(item.get("description") or "").strip(),
          "image_url": str(item.get("image_url") or "").strip(),

      }
  
  
  def build_index_content_fields(
      items: List[Dict[str, Any]],
      tenant_id: Optional[str] = None,

      enrichment_scopes: Optional[List[str]] = None,
      category_taxonomy_profile: Optional[str] = None,

  ) -> List[Dict[str, Any]]:
      """
      高层入口：生成与 ES mapping 对齐的内容理解字段。
  
      输入项需包含：
      - `id` 或 `spu_id`
      - `title`
      - 可选 `brief` / `description` / `image_url`

      - 可选 `enrichment_scopes`，默认同时执行 `generic` 与 `category_taxonomy`

      - 可选 `category_taxonomy_profile`，默认 `apparel`

  
      返回项结构：
      - `id`
      - `qanchors`
      - `enriched_tags`
      - `enriched_attributes`

      - `enriched_taxonomy_attributes`

      - 可选 `error`
  
      其中：
      - `qanchors.{lang}` 为短语数组
      - `enriched_tags.{lang}` 为标签数组
      """

      requested_enrichment_scopes = _normalize_enrichment_scopes(enrichment_scopes)

      normalized_taxonomy_profile = _normalize_category_taxonomy_profile(category_taxonomy_profile)

      normalized_items = [_normalize_index_content_item(item) for item in items]
      if not normalized_items:
          return []

  
      results_by_id: Dict[str, Dict[str, Any]] = {
          item["id"]: {
              "id": item["id"],
              "qanchors": {},
              "enriched_tags": {},
              "enriched_attributes": [],

              "enriched_taxonomy_attributes": [],

          }
          for item in normalized_items
      }
  

      for lang in _CORE_INDEX_LANGUAGES:

          if "generic" in requested_enrichment_scopes:

              try:
                  rows = analyze_products(
                      products=normalized_items,
                      target_lang=lang,
                      batch_size=BATCH_SIZE,
                      tenant_id=tenant_id,
                      analysis_kind="content",

                      category_taxonomy_profile=normalized_taxonomy_profile,

                  )
              except Exception as e:
                  logger.warning("build_index_content_fields content enrichment failed for lang=%s: %s", lang, e)
                  for item in normalized_items:
                      results_by_id[item["id"]].setdefault("error", str(e))

                  continue

              for row in rows or []:
                  item_id = str(row.get("id") or "").strip()
                  if not item_id or item_id not in results_by_id:
                      continue
                  if row.get("error"):
                      results_by_id[item_id].setdefault("error", row["error"])
                      continue
                  _apply_index_content_row(results_by_id[item_id], row=row, lang=lang)
  

      if "category_taxonomy" in requested_enrichment_scopes:

          for lang in _CORE_INDEX_LANGUAGES:
              try:
                  taxonomy_rows = analyze_products(
                      products=normalized_items,
                      target_lang=lang,
                      batch_size=BATCH_SIZE,
                      tenant_id=tenant_id,
                      analysis_kind="taxonomy",
                      category_taxonomy_profile=normalized_taxonomy_profile,
                  )
              except Exception as e:
                  logger.warning(
                      "build_index_content_fields taxonomy enrichment failed for profile=%s lang=%s: %s",
                      normalized_taxonomy_profile,
                      lang,
                      e,
                  )
                  for item in normalized_items:
                      results_by_id[item["id"]].setdefault("error", str(e))
                  continue

              for row in taxonomy_rows or []:
                  item_id = str(row.get("id") or "").strip()
                  if not item_id or item_id not in results_by_id:
                      continue
                  if row.get("error"):
                      results_by_id[item_id].setdefault("error", row["error"])
                      continue
                  _apply_index_taxonomy_row(
                      results_by_id[item_id],
                      row=row,
                      lang=lang,
                      category_taxonomy_profile=normalized_taxonomy_profile,
                  )

      return [results_by_id[item["id"]] for item in normalized_items]
  
  

  def _normalize_space(text: str) -> str:
      return re.sub(r"\s+", " ", (text or "").strip())
  
  
  def _contains_cjk(text: str) -> bool:
      return bool(re.search(r"[\u3400-\u4dbf\u4e00-\u9fff\uf900-\ufaff]", text or ""))
  
  
  def _truncate_by_chars(text: str, max_chars: int) -> str:
      return text[:max_chars].strip()
  
  
  def _truncate_by_words(text: str, max_words: int) -> str:
      words = re.findall(r"\S+", text or "")
      return " ".join(words[:max_words]).strip()
  
  
  def _detect_prompt_input_lang(text: str) -> str:
      # 简化处理：包含 CJK 时按中文类文本处理，否则统一按空格分词类语言处理。
      return "zh" if _contains_cjk(text) else "en"
  
  
  def _build_prompt_input_text(product: Dict[str, Any]) -> str:
      """
      生成真正送入 prompt 的商品文本。
  
      规则：
      - 默认使用 title
      - 若文本过短，则依次补 brief / description
      - 若文本过长，则按语言粗粒度截断
      """
      fields = [
          _normalize_space(str(product.get("title") or "")),
          _normalize_space(str(product.get("brief") or "")),
          _normalize_space(str(product.get("description") or "")),
      ]
      parts: List[str] = []
  
      def join_parts() -> str:
          return " | ".join(part for part in parts if part).strip()
  
      for field in fields:
          if not field:
              continue
          if field not in parts:
              parts.append(field)
          candidate = join_parts()
          if _detect_prompt_input_lang(candidate) == "zh":
              if len(candidate) >= PROMPT_INPUT_MIN_ZH_CHARS:
                  return _truncate_by_chars(candidate, PROMPT_INPUT_MAX_ZH_CHARS)
          else:
              if len(re.findall(r"\S+", candidate)) >= PROMPT_INPUT_MIN_WORDS:
                  return _truncate_by_words(candidate, PROMPT_INPUT_MAX_WORDS)
  
      candidate = join_parts()
      if not candidate:
          return ""
      if _detect_prompt_input_lang(candidate) == "zh":
          return _truncate_by_chars(candidate, PROMPT_INPUT_MAX_ZH_CHARS)
      return _truncate_by_words(candidate, PROMPT_INPUT_MAX_WORDS)
  
  

  def _make_analysis_cache_key(

      product: Dict[str, Any],

      target_lang: str,

      analysis_kind: str,

      category_taxonomy_profile: Optional[str] = None,

  ) -> str:

      """构造缓存 key，仅由分析类型、prompt 实际输入文本内容与目标语言决定。"""

      schema = _get_analysis_schema(
          analysis_kind,
          category_taxonomy_profile=category_taxonomy_profile,
      )

      prompt_input = _build_prompt_input_text(product)
      h = hashlib.md5(prompt_input.encode("utf-8")).hexdigest()

      prompt_contract = {
          "schema_name": schema.name,
          "cache_version": schema.cache_version,
          "system_message": SYSTEM_MESSAGE,
          "user_instruction_template": USER_INSTRUCTION_TEMPLATE,
          "shared_instruction": schema.shared_instruction,
          "assistant_headers": schema.get_headers(target_lang),
          "result_fields": schema.result_fields,
          "meaningful_fields": schema.meaningful_fields,
          "field_aliases": schema.field_aliases,
      }
      prompt_contract_hash = hashlib.md5(
          json.dumps(prompt_contract, ensure_ascii=False, sort_keys=True).encode("utf-8")
      ).hexdigest()[:12]
      return (
          f"{ANCHOR_CACHE_PREFIX}:{analysis_kind}:{prompt_contract_hash}:"
          f"{target_lang}:{prompt_input[:4]}{h}"
      )

  def _make_anchor_cache_key(

      product: Dict[str, Any],

      target_lang: str,

  ) -> str:
      return _make_analysis_cache_key(product, target_lang, analysis_kind="content")
  
  
  def _get_cached_analysis_result(
      product: Dict[str, Any],
      target_lang: str,
      analysis_kind: str,

      category_taxonomy_profile: Optional[str] = None,

  ) -> Optional[Dict[str, Any]]:
      if not _anchor_redis:
          return None

      schema = _get_analysis_schema(
          analysis_kind,
          category_taxonomy_profile=category_taxonomy_profile,
      )

      try:

          key = _make_analysis_cache_key(
              product,
              target_lang,
              analysis_kind,
              category_taxonomy_profile=category_taxonomy_profile,
          )

          raw = _anchor_redis.get(key)
          if not raw:
              return None

          result = _normalize_analysis_result(
              json.loads(raw),
              product=product,
              target_lang=target_lang,
              schema=schema,
          )
          if not _has_meaningful_analysis_content(result, schema):

              return None
          return result

      except Exception as e:

          logger.warning("Failed to get %s analysis cache: %s", analysis_kind, e)

          return None
  
  

  def _get_cached_anchor_result(
      product: Dict[str, Any],
      target_lang: str,
  ) -> Optional[Dict[str, Any]]:
      return _get_cached_analysis_result(product, target_lang, analysis_kind="content")
  
  
  def _set_cached_analysis_result(

      product: Dict[str, Any],

      target_lang: str,
      result: Dict[str, Any],

      analysis_kind: str,

      category_taxonomy_profile: Optional[str] = None,

  ) -> None:
      if not _anchor_redis:
          return

      schema = _get_analysis_schema(
          analysis_kind,
          category_taxonomy_profile=category_taxonomy_profile,
      )

      try:

          normalized = _normalize_analysis_result(
              result,
              product=product,
              target_lang=target_lang,
              schema=schema,
          )
          if not _has_meaningful_analysis_content(normalized, schema):

              return

          key = _make_analysis_cache_key(
              product,
              target_lang,
              analysis_kind,
              category_taxonomy_profile=category_taxonomy_profile,
          )

          ttl = ANCHOR_CACHE_EXPIRE_DAYS * 24 * 3600

          _anchor_redis.setex(key, ttl, json.dumps(normalized, ensure_ascii=False))

      except Exception as e:

          logger.warning("Failed to set %s analysis cache: %s", analysis_kind, e)
  
  
  def _set_cached_anchor_result(
      product: Dict[str, Any],
      target_lang: str,
      result: Dict[str, Any],
  ) -> None:
      _set_cached_analysis_result(product, target_lang, result, analysis_kind="content")

  def _build_assistant_prefix(headers: List[str]) -> str:
      header_line = "| " + " | ".join(headers) + " |"
      separator_line = "|" + "----|" * len(headers)
      return f"{header_line}\n{separator_line}\n"

  def _build_shared_context(products: List[Dict[str, str]], schema: AnalysisSchema) -> str:
      shared_context = schema.shared_instruction

      for idx, product in enumerate(products, 1):

          prompt_input = _build_prompt_input_text(product)
          shared_context += f"{idx}. {prompt_input}\n"

      return shared_context

  def _hash_text(text: str) -> str:
      return hashlib.md5((text or "").encode("utf-8")).hexdigest()[:12]
  
  
  def _mark_shared_context_logged_once(shared_context_key: str) -> bool:

      with _logged_shared_context_lock:
          if shared_context_key in _logged_shared_context_keys:
              _logged_shared_context_keys.move_to_end(shared_context_key)
              return False

          _logged_shared_context_keys[shared_context_key] = None
          if len(_logged_shared_context_keys) > LOGGED_SHARED_CONTEXT_CACHE_SIZE:
              _logged_shared_context_keys.popitem(last=False)
          return True

  def reset_logged_shared_context_keys() -> None:
      """测试辅助：清理已记录的共享 prompt key。"""

      with _logged_shared_context_lock:
          _logged_shared_context_keys.clear()

  
  def create_prompt(
      products: List[Dict[str, str]],
      target_lang: str = "zh",

      analysis_kind: str = "content",

      category_taxonomy_profile: Optional[str] = None,

  ) -> Tuple[Optional[str], Optional[str], Optional[str]]:

      """根据目标语言创建共享上下文、本地化输出要求和 Partial Mode assistant 前缀。"""

      schema = _get_analysis_schema(
          analysis_kind,
          category_taxonomy_profile=category_taxonomy_profile,
      )

      markdown_table_headers = schema.get_headers(target_lang)

      if not markdown_table_headers:
          logger.warning(

              "Unsupported target_lang for markdown table headers: kind=%s lang=%s",
              analysis_kind,

              target_lang,
          )
          return None, None, None

      shared_context = _build_shared_context(products, schema)

      language_label = SOURCE_LANG_CODE_MAP.get(target_lang, target_lang)
      user_prompt = USER_INSTRUCTION_TEMPLATE.format(language=language_label).strip()
      assistant_prefix = _build_assistant_prefix(markdown_table_headers)
      return shared_context, user_prompt, assistant_prefix
  
  
  def _merge_partial_response(assistant_prefix: str, generated_content: str) -> str:
      """将 Partial Mode 的 assistant 前缀与补全文本拼成完整 markdown。"""
      generated = (generated_content or "").lstrip()
      prefix_lines = [line.strip() for line in assistant_prefix.strip().splitlines()]
      generated_lines = generated.splitlines()
  
      if generated_lines:
          first_line = generated_lines[0].strip()
          if prefix_lines and first_line == prefix_lines[0]:
              generated_lines = generated_lines[1:]
              if generated_lines and len(prefix_lines) > 1 and generated_lines[0].strip() == prefix_lines[1]:
                  generated_lines = generated_lines[1:]
          elif len(prefix_lines) > 1 and first_line == prefix_lines[1]:
              generated_lines = generated_lines[1:]
  
      suffix = "\n".join(generated_lines).lstrip("\n")
      if suffix:
          return f"{assistant_prefix}{suffix}"
      return assistant_prefix
  
  
  def call_llm(
      shared_context: str,
      user_prompt: str,
      assistant_prefix: str,
      target_lang: str = "zh",