#!/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",
    analysis_kind: str = "content",
) -> 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, kind=%s, shared_key=%s, chars=%s] (logged once per process key)",
            MODEL_NAME,
            analysis_kind,
            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, kind=%s, lang=%s, shared_key=%s, tail_key=%s]:",
        MODEL_NAME,
        analysis_kind,
        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 [kind=%s, lang=%s, shared_key=%s, tail_key=%s, prompt_chars=%s, prefix_chars=%s]",
        analysis_kind,
        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, kind=%s, lang=%s, shared_key=%s, tail_key=%s]:",
                    MODEL_NAME,
                    analysis_kind,
                    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 [kind=%s, lang=%s, shared_key=%s, tail_key=%s, generated_chars=%s, completion_tokens=%s, prompt_tokens=%s, total_tokens=%s]",
                    analysis_kind,
                    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,
    analysis_kind: str = "content",
    category_taxonomy_profile: Optional[str] = None,
) -> List[Dict[str, str]]:
    """解析markdown表格内容"""
    schema = _get_analysis_schema(
        analysis_kind,
        category_taxonomy_profile=category_taxonomy_profile,
    )
    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("|")]
            if parts and parts[0] == "":
                parts = parts[1:]
            if parts and parts[-1] == "":
                parts = parts[:-1]

            if len(parts) >= 2:
                row = {"seq_no": parts[0]}
                for field_index, field_name in enumerate(schema.result_fields, start=1):
                    cell = parts[field_index] if len(parts) > field_index else ""
                    row[field_name] = _normalize_markdown_table_cell(cell)
                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,
    analysis_kind: str,
    category_taxonomy_profile: Optional[str] = None,
) -> None:
    schema = _get_analysis_schema(
        analysis_kind,
        category_taxonomy_profile=category_taxonomy_profile,
    )
    expected = len(batch_data)
    actual = len(parsed_results)
    if actual != expected:
        logger.warning(
            "Parsed row count mismatch for kind=%s batch=%s lang=%s: expected=%s actual=%s",
            analysis_kind,
            batch_num,
            target_lang,
            expected,
            actual,
        )

    if not schema.quality_fields:
        logger.info(
            "Parsed Quality Summary [kind=%s, batch=%s, lang=%s]: rows=%s/%s",
            analysis_kind,
            batch_num,
            target_lang,
            actual,
            expected,
        )
        return

    missing_summary = ", ".join(
        f"missing_{field}="
        f"{sum(1 for item in parsed_results if not str(item.get(field) or '').strip())}"
        for field in schema.quality_fields
    )
    logger.info(
        "Parsed Quality Summary [kind=%s, batch=%s, lang=%s]: rows=%s/%s, %s",
        analysis_kind,
        batch_num,
        target_lang,
        actual,
        expected,
        missing_summary,
    )


def process_batch(
    batch_data: List[Dict[str, str]],
    batch_num: int,
    target_lang: str = "zh",
    analysis_kind: str = "content",
    category_taxonomy_profile: Optional[str] = None,
) -> List[Dict[str, Any]]:
    """处理一个批次的数据"""
    schema = _get_analysis_schema(
        analysis_kind,
        category_taxonomy_profile=category_taxonomy_profile,
    )
    logger.info(f"\n{'#' * 80}")
    logger.info(
        "Processing Batch %s (%s items, kind=%s)",
        batch_num,
        len(batch_data),
        analysis_kind,
    )

    # 创建提示词
    shared_context, user_prompt, assistant_prefix = create_prompt(
        batch_data,
        target_lang=target_lang,
        analysis_kind=analysis_kind,
        category_taxonomy_profile=category_taxonomy_profile,
    )

    # 如果提示词创建失败（例如不支持的 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, kind=%s, target_lang=%s; "
            "marking entire batch as failed without calling LLM",
            batch_num,
            analysis_kind,
            target_lang,
        )
        return [
            _make_empty_analysis_result(
                item,
                target_lang,
                schema,
                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,
            analysis_kind=analysis_kind,
        )

        # 解析结果
        parsed_results = parse_markdown_table(
            raw_response,
            analysis_kind=analysis_kind,
            category_taxonomy_profile=category_taxonomy_profile,
        )
        _log_parsed_result_quality(
            batch_data,
            parsed_results,
            target_lang,
            batch_num,
            analysis_kind,
            category_taxonomy_profile,
        )

        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,
                    schema=schema,
                )
                results_with_ids.append(result)
                logger.info(
                    "Mapped: kind=%s seq=%s -> original_id=%s",
                    analysis_kind,
                    parsed_item.get("seq_no"),
                    source_product.get("id"),
                )

        # 保存批次 JSON 日志到独立文件
        batch_log = {
            "batch_num": batch_num,
            "analysis_kind": analysis_kind,
            "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_{analysis_kind}_{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, schema, 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,
    analysis_kind: str = "content",
    category_taxonomy_profile: 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 []

    _get_analysis_schema(
        analysis_kind,
        category_taxonomy_profile=category_taxonomy_profile,
    )
    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_analysis_result(
            product,
            target_lang,
            analysis_kind,
            category_taxonomy_profile=category_taxonomy_profile,
        )
        if cached:
            logger.info(
                f"[analyze_products] Cache hit for title='{title[:50]}...', "
                f"kind={analysis_kind}, 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)}, kind={analysis_kind}, target_lang={target_lang}"
            )
            batch_results = process_batch(
                batch,
                batch_num=batch_num,
                target_lang=target_lang,
                analysis_kind=analysis_kind,
                category_taxonomy_profile=category_taxonomy_profile,
            )

            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_analysis_result(
                        product,
                        target_lang,
                        item,
                        analysis_kind,
                        category_taxonomy_profile=category_taxonomy_profile,
                    )
                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, kind=%s, target_lang=%s",
            max_workers,
            total_batches,
            bs,
            analysis_kind,
            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,
                analysis_kind=analysis_kind,
                category_taxonomy_profile=category_taxonomy_profile,
            )

        # 按 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_analysis_result(
                        product,
                        target_lang,
                        item,
                        analysis_kind,
                        category_taxonomy_profile=category_taxonomy_profile,
                    )
                except Exception:
                    # 已在内部记录 warning
                    pass

    return [item for item in results_by_index if item is not None]