#!/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 collections import OrderedDict

  from datetime import datetime

  from concurrent.futures import ThreadPoolExecutor

  from typing import List, Dict, Tuple, Any, Optional
  
  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,
  )

  
  # 配置
  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]+")

  _CORE_INDEX_LANGUAGES = ("zh", "en")

  _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"),
  )
  _ANALYSIS_RESULT_FIELDS = (
      "title",
      "category_path",
      "tags",
      "target_audience",
      "usage_scene",
      "season",
      "key_attributes",
      "material",
      "features",
      "anchor_text",
  )
  _ANALYSIS_MEANINGFUL_FIELDS = (
      "tags",

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

      "anchor_text",

  )

  _ANALYSIS_FIELD_ALIASES = {
      "tags": ("tags", "enriched_tags"),
  }

  
  
  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 _append_enriched_attribute(
      target: List[Dict[str, Any]],
      name: str,
      lang: str,
      raw_value: Any,
  ) -> None:
      for value in split_multi_value_field(raw_value):
          if any(
              item.get("name") == name
              and isinstance(item.get("value"), dict)
              and item["value"].get(lang) == value
              for item in target
          ):
              continue
          target.append({"name": name, "value": {lang: 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) -> Tuple[str, ...]:
      return _ANALYSIS_FIELD_ALIASES.get(field_name, (field_name,))
  
  
  def _get_analysis_field_value(row: Dict[str, Any], field_name: str) -> Any:
      for alias in _get_analysis_field_aliases(field_name):
          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,
      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 _ANALYSIS_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,
  ) -> Dict[str, Any]:
      normalized = _make_empty_analysis_result(product, target_lang)
      if not isinstance(result, dict):
          return normalized
  
      normalized["lang"] = str(result.get("lang") or target_lang).strip() or target_lang
      normalized["title"] = str(result.get("title") or "").strip()
      normalized["category_path"] = str(result.get("category_path") or "").strip()
      normalized["title_input"] = str(
          product.get("title") or result.get("title_input") or ""
      ).strip()
  
      for field in _ANALYSIS_RESULT_FIELDS:
          if field in {"title", "category_path"}:
              continue
          normalized[field] = str(_get_analysis_field_value(result, field) or "").strip()
  
      if result.get("error"):
          normalized["error"] = str(result.get("error"))
      return normalized
  
  
  def _has_meaningful_analysis_content(result: Dict[str, Any]) -> bool:
      return any(_has_meaningful_value(result.get(field)) for field in _ANALYSIS_MEANINGFUL_FIELDS)
  
  

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

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

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

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

          if not raw:
              continue

          _append_enriched_attribute(
              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 _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,
  ) -> List[Dict[str, Any]]:
      """
      高层入口：生成与 ES mapping 对齐的内容理解字段。
  
      输入项需包含：
      - `id` 或 `spu_id`
      - `title`
      - 可选 `brief` / `description` / `image_url`
  
      返回项结构：
      - `id`
      - `qanchors`
      - `enriched_tags`
      - `enriched_attributes`
      - 可选 `error`
  
      其中：
      - `qanchors.{lang}` 为短语数组
      - `enriched_tags.{lang}` 为标签数组
      """
      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": [],
          }
          for item in normalized_items
      }
  
      for lang in _CORE_INDEX_LANGUAGES:
          try:
              rows = analyze_products(
                  products=normalized_items,
                  target_lang=lang,
                  batch_size=BATCH_SIZE,
                  tenant_id=tenant_id,
              )
          except Exception as e:
              logger.warning("build_index_content_fields 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)
  
      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_anchor_cache_key(

      product: Dict[str, Any],

      target_lang: str,

  ) -> str:

      """构造缓存 key，仅由 prompt 实际输入文本内容 + 目标语言决定。"""
      prompt_input = _build_prompt_input_text(product)
      h = hashlib.md5(prompt_input.encode("utf-8")).hexdigest()
      return f"{ANCHOR_CACHE_PREFIX}:{target_lang}:{prompt_input[:4]}{h}"

  
  
  def _get_cached_anchor_result(

      product: Dict[str, Any],

      target_lang: str,

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

          key = _make_anchor_cache_key(product, target_lang)

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

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

      except Exception as e:
          logger.warning(f"Failed to get anchor cache: {e}")
          return None
  
  
  def _set_cached_anchor_result(

      product: Dict[str, Any],

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

  ) -> None:
      if not _anchor_redis:
          return
      try:

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

          key = _make_anchor_cache_key(product, target_lang)

          ttl = ANCHOR_CACHE_EXPIRE_DAYS * 24 * 3600

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

      except Exception as e:
          logger.warning(f"Failed to set anchor cache: {e}")
  
  

  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]]) -> str:
      shared_context = SHARED_ANALYSIS_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",
  ) -> Tuple[str, str, str]:
      """根据目标语言创建共享上下文、本地化输出要求和 Partial Mode assistant 前缀。"""
      markdown_table_headers = LANGUAGE_MARKDOWN_TABLE_HEADERS.get(target_lang)
      if not markdown_table_headers:
          logger.warning(
              "Unsupported target_lang for markdown table headers: %s",
              target_lang,
          )
          return None, None, None
      shared_context = _build_shared_context(products)
      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",
  ) -> Tuple[str, str]:
      """调用大模型 API（带重试机制），使用 Partial Mode 强制 markdown 表格前缀。"""

      headers = {
          "Authorization": f"Bearer {API_KEY}",
          "Content-Type": "application/json",
      }

      shared_context_key = _hash_text(shared_context)
      localized_tail_key = _hash_text(f"{target_lang}\n{user_prompt}\n{assistant_prefix}")
      combined_user_prompt = f"{shared_context.rstrip()}\n\n{user_prompt.strip()}"

  
      payload = {
          "model": MODEL_NAME,
          "messages": [
              {
                  "role": "system",

                  "content": SYSTEM_MESSAGE,

              },
              {
                  "role": "user",

                  "content": combined_user_prompt,
              },
              {
                  "role": "assistant",
                  "content": assistant_prefix,
                  "partial": True,

              },
          ],
          "temperature": 0.3,
          "top_p": 0.8,
      }
  
      request_data = {
          "headers": {k: v for k, v in headers.items() if k != "Authorization"},
          "payload": payload,
      }
  

      if _mark_shared_context_logged_once(shared_context_key):
          logger.info(f"\n{'=' * 80}")
          logger.info(
              "LLM Shared Context [model=%s, shared_key=%s, chars=%s] (logged once per process key)",
              MODEL_NAME,
              shared_context_key,
              len(shared_context),
          )
          logger.info("\nSystem Message:\n%s", SYSTEM_MESSAGE)
          logger.info("\nShared Context:\n%s", shared_context)

  
      verbose_logger.info(f"\n{'=' * 80}")

      verbose_logger.info(
          "LLM Request [model=%s, lang=%s, shared_key=%s, tail_key=%s]:",
          MODEL_NAME,
          target_lang,
          shared_context_key,
          localized_tail_key,
      )

      verbose_logger.info(json.dumps(request_data, ensure_ascii=False, indent=2))

      verbose_logger.info(f"\nCombined User Prompt:\n{combined_user_prompt}")
      verbose_logger.info(f"\nShared Context:\n{shared_context}")
      verbose_logger.info(f"\nLocalized Requirement:\n{user_prompt}")
      verbose_logger.info(f"\nAssistant Prefix:\n{assistant_prefix}")
  
      logger.info(
          "\nLLM Request Variant [lang=%s, shared_key=%s, tail_key=%s, prompt_chars=%s, prefix_chars=%s]",
          target_lang,
          shared_context_key,
          localized_tail_key,
          len(user_prompt),
          len(assistant_prefix),
      )
      logger.info("\nLocalized Requirement:\n%s", user_prompt)
      logger.info("\nAssistant Prefix:\n%s", assistant_prefix)

  
      # 创建session，禁用代理
      session = requests.Session()
      session.trust_env = False  # 忽略系统代理设置
  
      try:
          # 重试机制
          for attempt in range(MAX_RETRIES):
              try:
                  response = session.post(
                      f"{API_BASE_URL}/chat/completions",
                      headers=headers,
                      json=payload,
                      timeout=REQUEST_TIMEOUT,
                      proxies={"http": None, "https": None},  # 明确禁用代理
                  )
  
                  response.raise_for_status()
                  result = response.json()

                  usage = result.get("usage") or {}
  
                  verbose_logger.info(
                      "\nLLM Response [model=%s, lang=%s, shared_key=%s, tail_key=%s]:",
                      MODEL_NAME,
                      target_lang,
                      shared_context_key,
                      localized_tail_key,
                  )
                  verbose_logger.info(json.dumps(result, ensure_ascii=False, indent=2))

                  generated_content = result["choices"][0]["message"]["content"]
                  full_markdown = _merge_partial_response(assistant_prefix, generated_content)

                  logger.info(
                      "\nLLM Response Summary [lang=%s, shared_key=%s, tail_key=%s, generated_chars=%s, completion_tokens=%s, prompt_tokens=%s, total_tokens=%s]",
                      target_lang,
                      shared_context_key,
                      localized_tail_key,
                      len(generated_content or ""),
                      usage.get("completion_tokens"),
                      usage.get("prompt_tokens"),
                      usage.get("total_tokens"),
                  )
                  logger.info("\nGenerated Content:\n%s", generated_content)
                  logger.info("\nMerged Markdown:\n%s", full_markdown)

                  verbose_logger.info(f"\nGenerated Content:\n{generated_content}")
                  verbose_logger.info(f"\nMerged Markdown:\n{full_markdown}")

                  return full_markdown, json.dumps(result, ensure_ascii=False)

  
              except requests.exceptions.ProxyError as e:
                  logger.warning(f"Attempt {attempt + 1}/{MAX_RETRIES}: Proxy error - {str(e)}")
                  if attempt < MAX_RETRIES - 1:
                      logger.info(f"Retrying in {RETRY_DELAY} seconds...")
                      time.sleep(RETRY_DELAY)
                  else:
                      raise
  
              except requests.exceptions.RequestException as e:
                  logger.warning(f"Attempt {attempt + 1}/{MAX_RETRIES}: Request error - {str(e)}")
                  if attempt < MAX_RETRIES - 1:
                      logger.info(f"Retrying in {RETRY_DELAY} seconds...")
                      time.sleep(RETRY_DELAY)
                  else:
                      raise
  
              except Exception as e:
                  logger.error(f"Unexpected error on attempt {attempt + 1}/{MAX_RETRIES}: {str(e)}")
                  if attempt < MAX_RETRIES - 1:
                      logger.info(f"Retrying in {RETRY_DELAY} seconds...")
                      time.sleep(RETRY_DELAY)
                  else:
                      raise
  
      finally:
          session.close()
  
  
  def parse_markdown_table(markdown_content: str) -> List[Dict[str, str]]:
      """解析markdown表格内容"""
      lines = markdown_content.strip().split("\n")
      data = []
      data_started = False
  
      for line in lines:
          line = line.strip()
          if not line:
              continue
  
          # 表格行处理
          if line.startswith("|"):
              # 分隔行（---- 或 :---: 等；允许空格，如 "| ---- | ---- |"）
              sep_chars = line.replace("|", "").strip().replace(" ", "")
              if sep_chars and set(sep_chars) <= {"-", ":"}:
                  data_started = True
                  continue
  
              # 首个表头行：无论语言如何，统一跳过
              if not data_started:
                  # 等待下一行数据行
                  continue
  
              # 解析数据行
              parts = [p.strip() for p in line.split("|")]
              parts = [p for p in parts if p]  # 移除空字符串
  
              if len(parts) >= 2:
                  row = {
                      "seq_no": parts[0],
                      "title": parts[1],  # 商品标题（按目标语言）
                      "category_path": parts[2] if len(parts) > 2 else "",  # 品类路径

                      "tags": parts[3] if len(parts) > 3 else "",  # 细分标签

                      "target_audience": parts[4] if len(parts) > 4 else "",  # 适用人群
                      "usage_scene": parts[5] if len(parts) > 5 else "",  # 使用场景
                      "season": parts[6] if len(parts) > 6 else "",  # 适用季节
                      "key_attributes": parts[7] if len(parts) > 7 else "",  # 关键属性
                      "material": parts[8] if len(parts) > 8 else "",  # 材质说明
                      "features": parts[9] if len(parts) > 9 else "",  # 功能特点

                      "anchor_text": parts[10] if len(parts) > 10 else "",  # 锚文本

                  }
                  data.append(row)
  
      return data
  
  

  def _log_parsed_result_quality(
      batch_data: List[Dict[str, str]],
      parsed_results: List[Dict[str, str]],
      target_lang: str,
      batch_num: int,
  ) -> None:
      expected = len(batch_data)
      actual = len(parsed_results)
      if actual != expected:
          logger.warning(
              "Parsed row count mismatch for batch=%s lang=%s: expected=%s actual=%s",
              batch_num,
              target_lang,
              expected,
              actual,
          )
  
      missing_anchor = sum(1 for item in parsed_results if not str(item.get("anchor_text") or "").strip())
      missing_category = sum(1 for item in parsed_results if not str(item.get("category_path") or "").strip())
      missing_title = sum(1 for item in parsed_results if not str(item.get("title") or "").strip())
  
      logger.info(
          "Parsed Quality Summary [batch=%s, lang=%s]: rows=%s/%s, missing_title=%s, missing_category=%s, missing_anchor=%s",
          batch_num,
          target_lang,
          actual,
          expected,
          missing_title,
          missing_category,
          missing_anchor,
      )
  
  

  def process_batch(
      batch_data: List[Dict[str, str]],
      batch_num: int,
      target_lang: str = "zh",

  ) -> List[Dict[str, Any]]:

      """处理一个批次的数据"""
      logger.info(f"\n{'#' * 80}")
      logger.info(f"Processing Batch {batch_num} ({len(batch_data)} items)")
  
      # 创建提示词

      shared_context, user_prompt, assistant_prefix = create_prompt(
          batch_data,
          target_lang=target_lang,
      )
  
      # 如果提示词创建失败（例如不支持的 target_lang），本次批次整体失败，不再继续调用 LLM
      if shared_context is None or user_prompt is None or assistant_prefix is None:
          logger.error(
              "Failed to create prompt for batch %s, target_lang=%s; "
              "marking entire batch as failed without calling LLM",
              batch_num,
              target_lang,
          )
          return [

              _make_empty_analysis_result(
                  item,
                  target_lang,
                  error=f"prompt_creation_failed: unsupported target_lang={target_lang}",
              )

              for item in batch_data
          ]

  
      # 调用LLM
      try:

          raw_response, full_response_json = call_llm(
              shared_context,
              user_prompt,
              assistant_prefix,
              target_lang=target_lang,
          )

  
          # 解析结果
          parsed_results = parse_markdown_table(raw_response)

          _log_parsed_result_quality(batch_data, parsed_results, target_lang, batch_num)

  
          logger.info(f"\nParsed Results ({len(parsed_results)} items):")
          logger.info(json.dumps(parsed_results, ensure_ascii=False, indent=2))
  
          # 映射回原始ID
          results_with_ids = []
          for i, parsed_item in enumerate(parsed_results):
              if i < len(batch_data):

                  source_product = batch_data[i]
                  result = _normalize_analysis_result(
                      parsed_item,
                      product=source_product,
                      target_lang=target_lang,
                  )

                  results_with_ids.append(result)

                  logger.info(
                      "Mapped: seq=%s -> original_id=%s",
                      parsed_item.get("seq_no"),
                      source_product.get("id"),
                  )

  
          # 保存批次 JSON 日志到独立文件
          batch_log = {
              "batch_num": batch_num,
              "timestamp": datetime.now().isoformat(),
              "input_products": batch_data,
              "raw_response": raw_response,
              "full_response_json": full_response_json,
              "parsed_results": parsed_results,
              "final_results": results_with_ids,
          }
  

          # 并发写 batch json 日志时，保证文件名唯一避免覆盖
          batch_call_id = uuid.uuid4().hex[:12]
          batch_log_file = LOG_DIR / f"batch_{batch_num:04d}_{timestamp}_{batch_call_id}.json"

          with open(batch_log_file, "w", encoding="utf-8") as f:
              json.dump(batch_log, f, ensure_ascii=False, indent=2)
  
          logger.info(f"Batch log saved to: {batch_log_file}")
  
          return results_with_ids
  
      except Exception as e:
          logger.error(f"Error processing batch {batch_num}: {str(e)}", exc_info=True)
          # 返回空结果，保持ID映射
          return [

              _make_empty_analysis_result(item, target_lang, error=str(e))

              for item in batch_data
          ]
  
  
  def analyze_products(
      products: List[Dict[str, str]],
      target_lang: str = "zh",
      batch_size: Optional[int] = None,
      tenant_id: Optional[str] = None,
  ) -> List[Dict[str, Any]]:
      """
      库调用入口：根据输入+语言，返回锚文本及各维度信息。
  
      Args:
          products: [{"id": "...", "title": "..."}]
          target_lang: 输出语言
          batch_size: 批大小，默认使用全局 BATCH_SIZE
      """
      if not API_KEY:
          raise RuntimeError("DASHSCOPE_API_KEY is not set, cannot call LLM")
  
      if not products:
          return []
  

      results_by_index: List[Optional[Dict[str, Any]]] = [None] * len(products)
      uncached_items: List[Tuple[int, Dict[str, str]]] = []
  
      for idx, product in enumerate(products):
          title = str(product.get("title") or "").strip()
          if not title:
              uncached_items.append((idx, product))
              continue
  

          cached = _get_cached_anchor_result(product, target_lang)

          if cached:
              logger.info(
                  f"[analyze_products] Cache hit for title='{title[:50]}...', "

                  f"lang={target_lang}"

              )
              results_by_index[idx] = cached
              continue
  
          uncached_items.append((idx, product))
  
      if not uncached_items:
          return [item for item in results_by_index if item is not None]

  
      # call_llm 一次处理上限固定为 BATCH_SIZE（默认 20）：
      # - 尽可能攒批处理；
      # - 即便调用方传入更大的 batch_size，也会自动按上限拆批。
      req_bs = BATCH_SIZE if batch_size is None else int(batch_size)
      bs = max(1, min(req_bs, BATCH_SIZE))

      total_batches = (len(uncached_items) + bs - 1) // bs

      batch_jobs: List[Tuple[int, List[Tuple[int, Dict[str, str]]], List[Dict[str, str]]]] = []

      for i in range(0, len(uncached_items), bs):

          batch_num = i // bs + 1

          batch_slice = uncached_items[i : i + bs]
          batch = [item for _, item in batch_slice]

          batch_jobs.append((batch_num, batch_slice, batch))
  
      # 只有一个批次时走串行，减少线程池创建开销与日志/日志文件的不可控交织
      if total_batches <= 1 or CONTENT_UNDERSTANDING_MAX_WORKERS <= 1:
          for batch_num, batch_slice, batch in batch_jobs:
              logger.info(
                  f"[analyze_products] Processing batch {batch_num}/{total_batches}, "
                  f"size={len(batch)}, target_lang={target_lang}"
              )
              batch_results = process_batch(batch, batch_num=batch_num, target_lang=target_lang)
  
              for (original_idx, product), item in zip(batch_slice, batch_results):
                  results_by_index[original_idx] = item
                  title_input = str(item.get("title_input") or "").strip()
                  if not title_input:
                      continue
                  if item.get("error"):
                      # 不缓存错误结果，避免放大临时故障
                      continue
                  try:
                      _set_cached_anchor_result(product, target_lang, item)
                  except Exception:
                      # 已在内部记录 warning
                      pass
      else:
          max_workers = min(CONTENT_UNDERSTANDING_MAX_WORKERS, len(batch_jobs))

          logger.info(

              "[analyze_products] Using ThreadPoolExecutor for uncached batches: "
              "max_workers=%s, total_batches=%s, bs=%s, target_lang=%s",
              max_workers,
              total_batches,
              bs,
              target_lang,

          )

          # 只把“LLM 调用 + markdown 解析”放到线程里；Redis get/set 保持在主线程，避免并发写入带来额外风险。
          # 注意：线程池是模块级单例，因此这里的 max_workers 主要用于日志语义（实际并发受单例池上限约束）。
          executor = _get_content_understanding_executor()
          future_by_batch_num: Dict[int, Any] = {}
          for batch_num, _batch_slice, batch in batch_jobs:
              future_by_batch_num[batch_num] = executor.submit(
                  process_batch, batch, batch_num=batch_num, target_lang=target_lang
              )
  
          # 按 batch_num 回填，确保输出稳定（results_by_index 是按原始 input index 映射的）
          for batch_num, batch_slice, _batch in batch_jobs:
              batch_results = future_by_batch_num[batch_num].result()
              for (original_idx, product), item in zip(batch_slice, batch_results):
                  results_by_index[original_idx] = item
                  title_input = str(item.get("title_input") or "").strip()
                  if not title_input:
                      continue
                  if item.get("error"):
                      # 不缓存错误结果，避免放大临时故障
                      continue
                  try:
                      _set_cached_anchor_result(product, target_lang, item)
                  except Exception:
                      # 已在内部记录 warning
                      pass