# -*- coding: utf-8 -*-
"""报告数据后处理：主题 enrichment、Persona 统计、根因引用回填、决策摘要等。"""
from __future__ import annotations

import logging
import re
from typing import Any, Dict, List, Optional, Set, Tuple

from data_loader import ASINStats, ClusterData, DataLoader, MarketStats, ReviewRecord
from llm_analyzer import market_competition_judgment

logger = logging.getLogger("voc.report_utils")

# 通用英文停用词（不含品类词，品类词由 build_product_stopwords 动态追加）
STOPWORDS_BASE = frozenset({
    "de", "la", "el", "en", "un", "una", "the", "and", "for", "with", "that", "this",
    "have", "has", "had", "was", "were", "are", "but", "not", "you", "your", "its",
    "it's", "i", "me", "my", "we", "our", "they", "them", "their", "be", "been",
    "product", "item", "one", "get", "got", "buy", "bought", "amazon", "review",
})
STOPWORDS = STOPWORDS_BASE  # 兼容旧 import

VALID_ASIN_RE = re.compile(r"^B[A-Z0-9]{9}$")

# 差评主题名常见后缀（用于自动配对好评/差评不对称洞察）
_NEG_THEME_SUFFIXES = ("差", "低", "故障", "问题", "不足", "缺陷", "损坏", "失效", "难", "慢", "弱")

EMOTIONAL_HINTS = (
    "自信", "焦虑", "恐惧", "感受", "情绪", "安心", "掌控", "体面", "尊严",
    "社交", "不安全感", "压力", "满足", "惊喜", "期待", "好奇", "骄傲",
    "confidence", "anxiety", "fear", "feel", "emotion",
)

_STRONG_HINTS = (
    "waste", "charge", "broken", "stopped", "doesn't", "does not", "not worth",
    "terrible", "horrible", "useless", "defect", "return", "refund", "pull",
    "nick", "burn", "bleed", "cut", "irritat", "bump", "overheat", "loud",
    "durable", "quality", "shave", "trim", "waterproof", "battery",
)

# 情感词频中应降权的中性噪声词
_NEUTRAL_NOISE = frozenset({
    "cut", "long", "feel", "leave", "try", "look", "hold", "end", "want", "turn",
    "different", "little", "leg", "electric", "get", "got", "use", "used", "work",
    "works", "good", "bad", "nice", "time", "day", "way", "thing", "things",
})

def build_product_stopwords(product_name: str = "") -> frozenset:
    """从产品名提取品类高频词，合并通用停用词。"""
    extra: Set[str] = set()
    if product_name:
        for w in re.findall(r"[a-zA-Z]{3,}", product_name.lower()):
            extra.add(w)
        for w in re.findall(r"[\u4e00-\u9fff]{2,}", product_name):
            extra.add(w)
    return frozenset(STOPWORDS_BASE | extra)


def _theme_name_similarity(pos: str, neg: str) -> int:
    """估算好评/差评主题名相似度（用于自动配对）。"""
    neg_core = neg
    for suf in _NEG_THEME_SUFFIXES:
        if neg_core.endswith(suf) and len(neg_core) > len(suf) + 1:
            neg_core = neg_core[: -len(suf)]
            break
    if len(neg_core) >= 2 and neg_core in pos:
        return len(neg_core) + 5
    if len(pos) >= 2 and pos in neg:
        return len(pos) + 3
    return sum(1 for c in set(neg_core) if c in pos and c not in "与及和")


def infer_pos_neg_theme_pairs(
    pos_themes: List[Dict[str, Any]],
    neg_themes: List[Dict[str, Any]],
) -> List[Tuple[str, str]]:
    """从 LLM 归纳的主题名自动推断好评↔差评配对（pos_name, neg_name）。"""
    pairs: List[Tuple[str, str]] = []
    used_pos: Set[str] = set()
    pos_names = [t.get("name", "") for t in pos_themes if t.get("name")]
    neg_names = [t.get("name", "") for t in neg_themes if t.get("name")]

    for neg in neg_names:
        best_pos = ""
        best_score = 0
        for pos in pos_names:
            if pos in used_pos:
                continue
            score = _theme_name_similarity(pos, neg)
            if score > best_score:
                best_score = score
                best_pos = pos
        if best_pos and best_score >= 3:
            pairs.append((best_pos, neg))
            used_pos.add(best_pos)
    return pairs

_QUOTE_CN_HINTS = (
    (("would not charge", "not turn on", "stopped working", "stop charging", "does not hold charge", "not charging", "battery life", "battery"), "抱怨充电/电池问题"),
    (("waste", "not worth", "overpriced", "wasted my money", "waste of money"), "抱怨性价比低"),
    (("broke", "broken", "stopped", "fall apart", "flimsy", "fell apart"), "抱怨产品损坏/不耐用"),
    (("difficult to use", "too complicated", "confusing", "hard to figure out", "not intuitive"), "抱怨使用复杂"),
    (("dull", "not sharp", "not effective", "doesn't work well", "barely works", "not strong enough"), "抱怨核心效果差"),
    (("irritat", "allergic", "rash", "burn", "cause pain", "uncomfortable"), "抱怨不适/刺激"),
    (("easy to clean", "easy to use", "simple", "intuitive"), "称赞使用方便"),
    (("love", "perfect", "great", "recommend", "amazing"), "整体满意推荐"),
    (("travel", "portable", "compact", "lightweight"), "称赞便携"),
    (("accessory", "attachment", "included", "comes with"), "提及配件/附件"),
)


def _is_strong_keyword(kw: str) -> bool:
    if len(kw) >= 12:
        return True
    return any(h in kw for h in _STRONG_HINTS)


def phrase_segments(phrases: List[str]) -> List[str]:
    """将聚类短语拆成可匹配的英文片段。"""
    segs: List[str] = []
    for p in phrases:
        p = (p or "").lower().strip()
        if not p:
            continue
        segs.append(p)
        for part in p.split(","):
            part = part.strip()
            if len(part) >= 3:
                segs.append(part)
    return list(dict.fromkeys(segs))


def expand_keywords(keywords: List[str]) -> List[str]:
    """展开主题 keywords（含逗号分隔的聚类短语）。"""
    expanded: List[str] = []
    for kw in keywords:
        expanded.extend(phrase_segments([kw]))
    return list(dict.fromkeys(expanded))


def match_text(keywords: List[str], text: str) -> bool:
    text = text.lower()
    expanded = expand_keywords(keywords)
    strong = [k for k in expanded if _is_strong_keyword(k)]
    weak = [k for k in expanded if k not in strong]
    for kw in sorted(strong, key=len, reverse=True):
        if len(kw) >= 3 and kw in text:
            return True
    if not strong:
        for kw in sorted(weak, key=len, reverse=True):
            if len(kw) >= 5 and kw in text:
                return True
    return False


def filter_stopwords(
    word_freq: List[Tuple[str, int]],
    product_name: str = "",
) -> List[Tuple[str, int]]:
    sw = build_product_stopwords(product_name)
    return [(w, c) for w, c in word_freq if w.lower() not in sw and len(w) >= 2]


def _score_cluster_theme(cluster: Dict[str, Any], theme: Dict[str, Any]) -> int:
    theme_kws = expand_keywords(theme.get("keywords", []))
    cluster_segs = phrase_segments(cluster.get("top_phrases", []))
    score = 0
    for cs in cluster_segs:
        for tk in theme_kws:
            if len(tk) >= 3 and (tk in cs or cs in tk):
                score += len(tk)
    return score


def enrich_theme_keywords(
    themes: List[Dict[str, Any]],
    global_clusters: List[Dict[str, Any]],
) -> List[Dict[str, Any]]:
    """从聚类 top_phrases 补充可匹配的英文 keywords。"""
    if not themes:
        return themes
    assigned: Dict[str, Set[str]] = {t["name"]: set() for t in themes}
    unassigned = list(global_clusters)

    for cluster in global_clusters:
        best = max(themes, key=lambda t: _score_cluster_theme(cluster, t))
        if _score_cluster_theme(cluster, best) <= 0:
            continue
        for seg in phrase_segments(cluster.get("top_phrases", [])):
            assigned[best["name"]].add(seg)

    for theme in themes:
        base = [k.lower().strip() for k in theme.get("keywords", []) if k.strip()]
        merged = list(dict.fromkeys(base + list(assigned.get(theme["name"], set()))))
        if not merged and global_clusters:
            # 兜底：取频次最高的聚类短语
            top = sorted(global_clusters, key=lambda c: -c.get("phrase_count", 0))[:3]
            for c in top:
                merged.extend(phrase_segments(c.get("top_phrases", [])))
            merged = list(dict.fromkeys(merged))
        theme["keywords"] = merged[:40]
    return themes


def recalc_neg_priorities(
    neg_themes: List[Dict[str, Any]],
    neg_freq: Dict[str, int],
    neg_total: int,
    asin_count: int,
    per_asin_neg: Dict[str, Dict[str, int]],
) -> List[Dict[str, Any]]:
    """根据实际频次重算 P0/P1/P2，并附加 freq_rank。"""
    p0_threshold = neg_total * 0.2
    p1_low = neg_total * 0.1
    for theme in neg_themes:
        name = theme["name"]
        count = neg_freq.get(name, 0)
        affected = sum(1 for asin_counts in per_asin_neg.values() if asin_counts.get(name, 0) > 0)
        asin_ratio = affected / max(asin_count, 1)
        if count >= p0_threshold and asin_ratio >= 0.8:
            theme["priority"] = "P0"
        elif count >= p1_low:
            theme["priority"] = "P1"
        elif count >= neg_total * 0.03:
            theme["priority"] = "P2"
        else:
            theme["priority"] = "observe"
        theme["freq_count"] = count
        theme["affected_asins"] = affected

    ranked = sorted(
        [t for t in neg_themes if neg_freq.get(t["name"], 0) > 0],
        key=lambda t: neg_freq.get(t["name"], 0),
        reverse=True,
    )
    for i, theme in enumerate(ranked, start=1):
        theme["freq_rank"] = i
        pri = theme.get("priority", "P2")
        if pri == "observe":
            theme["display_priority"] = "待观察"
        elif pri == "P1" and not any(t.get("priority") == "P0" for t in neg_themes):
            theme["display_priority"] = f"#{i}"
        else:
            theme["display_priority"] = pri
    for theme in neg_themes:
        theme.setdefault("freq_rank", 99)
        if theme.get("priority") == "observe":
            theme.setdefault("display_priority", "待观察")
        else:
            theme.setdefault("display_priority", theme.get("priority", "P2"))
    return neg_themes


def normalize_persona_dimension(persona: Dict[str, Any]) -> Dict[str, Any]:
    dim = str(persona.get("dimension", "")).strip().upper()
    if dim in ("A", "B", "C"):
        persona["dimension"] = dim
    elif "A" in dim and "B" not in dim:
        persona["dimension"] = "A"
    elif "B" in dim:
        persona["dimension"] = "B"
    elif "C" in dim:
        persona["dimension"] = "C"
    else:
        persona["dimension"] = "B"
    dim_labels = {"A": "生理特征", "B": "行为场景", "C": "购买动机"}
    persona["dimension_label"] = dim_labels.get(persona["dimension"], "行为场景")
    return persona


# ── Persona ↔ 聚类簇锚定（通用，适用于任意品类） ──

_PERSONA_PHRASE_STOPWORDS = frozenset({
    "self", "unknown", "user", "customer", "buyer", "myself", "the user",
    "a user", "consumer", "reviewer", "amazon", "product", "item",
})

_PERSONA_BINDABLE_STAGES: Dict[str, Tuple[str, ...]] = {
    "A": (
        "3a_pain_global",
        "3b_aspect_opinion_negative",
        "3b_aspect_opinion_positive",
    ),
    "B": (
        "3b_aspect_opinion_positive",
        "3b_aspect_opinion_negative",
        "3a_pain_global",
    ),
    "C": (
        "3a_pain_global",
        "3b_aspect_opinion_negative",
        "3b_aspect_opinion_positive",
    ),
}

_VALID_CLUSTER_STAGES = frozenset(
    stage for stages in _PERSONA_BINDABLE_STAGES.values() for stage in stages
)


def build_cluster_registry(
    all_clusters: Dict[str, List[ClusterData]],
) -> Dict[Tuple[str, int], ClusterData]:
    """stage+label → ClusterData 索引。"""
    registry: Dict[Tuple[str, int], ClusterData] = {}
    for stage, clusters in all_clusters.items():
        if stage not in _VALID_CLUSTER_STAGES:
            continue
        for c in clusters:
            registry[(stage, int(c.cluster_label))] = c
    return registry


def _persona_text_signals(persona: Dict[str, Any]) -> List[str]:
    parts = list(persona.get("keywords") or [])
    for field in ("core_pain", "core_need", "purchase_motivation", "name"):
        val = persona.get(field)
        if val:
            parts.append(str(val))
    return expand_keywords(parts)


def _score_persona_cluster(persona: Dict[str, Any], cluster: ClusterData) -> int:
    signals = _persona_text_signals(persona)
    cluster_segs = [
        s for s in phrase_segments(cluster.top_phrases)
        if s not in _PERSONA_PHRASE_STOPWORDS
    ]
    score = 0
    for cs in cluster_segs:
        for tk in signals:
            if len(tk) >= 3 and (tk in cs or cs in tk):
                score += len(tk)
    return score


def _normalize_cluster_ref(ref: Any) -> Optional[Dict[str, Any]]:
    if not isinstance(ref, dict):
        return None
    stage = str(ref.get("stage") or "").strip()
    label = ref.get("label")
    if stage not in _VALID_CLUSTER_STAGES or label is None:
        return None
    try:
        return {"stage": stage, "label": int(label)}
    except (TypeError, ValueError):
        return None


_PERSONA_DIM_ORDER: Tuple[str, ...] = ("A", "B", "C")


def _parse_cluster_refs_from_persona(persona: Dict[str, Any]) -> Dict[str, Optional[Dict[str, Any]]]:
    """解析 cluster_refs；旧 cluster_ref 并入主 dimension，audience 同时写入 A。"""
    refs: Dict[str, Optional[Dict[str, Any]]] = {d: None for d in _PERSONA_DIM_ORDER}
    raw_refs = persona.get("cluster_refs")
    if isinstance(raw_refs, dict):
        for dim in _PERSONA_DIM_ORDER:
            ref = _normalize_cluster_ref(raw_refs.get(dim))
            if ref and ref["stage"] in _PERSONA_BINDABLE_STAGES.get(dim, ()):
                refs[dim] = ref
    else:
        legacy = _normalize_cluster_ref(persona.get("cluster_ref"))
        if legacy:
            dim = str(persona.get("dimension", "B")).strip().upper()
            if dim not in _PERSONA_DIM_ORDER:
                dim = "B"
            if legacy["stage"] in _PERSONA_BINDABLE_STAGES.get(dim, ()):
                refs[dim] = legacy
            if legacy["stage"] == "1_audience":
                refs["A"] = legacy
    return refs


def _get_persona_anchor_ref(persona: Dict[str, Any]) -> Optional[Dict[str, Any]]:
    """Persona 锚点簇：优先 A 维，否则第一个有效 cluster_refs 条目。"""
    refs = persona.get("cluster_refs")
    if isinstance(refs, dict):
        a_ref = _normalize_cluster_ref(refs.get("A"))
        if a_ref:
            return a_ref
        for dim in _PERSONA_DIM_ORDER:
            ref = _normalize_cluster_ref(refs.get(dim))
            if ref:
                return ref
    return _normalize_cluster_ref(persona.get("cluster_ref"))


def _sync_legacy_cluster_ref(persona: Dict[str, Any]) -> None:
    """保留 cluster_ref 字段供下游兼容（指向锚点簇）。"""
    anchor = _get_persona_anchor_ref(persona)
    if anchor:
        persona["cluster_ref"] = anchor
    else:
        persona.pop("cluster_ref", None)


def _clusters_from_refs(
    refs: Dict[str, Optional[Dict[str, Any]]],
    registry: Dict[Tuple[str, int], ClusterData],
) -> List[ClusterData]:
    clusters: List[ClusterData] = []
    seen: Set[Tuple[str, int]] = set()
    for dim in _PERSONA_DIM_ORDER:
        ref = refs.get(dim)
        if not ref:
            continue
        key = (ref["stage"], ref["label"])
        if key in seen:
            continue
        cluster = registry.get(key)
        if cluster:
            clusters.append(cluster)
            seen.add(key)
    return clusters


def _auto_map_persona_dimension(
    persona: Dict[str, Any],
    dim: str,
    all_clusters: Dict[str, List[ClusterData]],
    used_audience_keys: Set[Tuple[str, int]],
) -> Tuple[Optional[Dict[str, Any]], bool]:
    """在指定维度允许的 stage 内为 Persona 自动匹配最佳簇。"""
    stage_order = _PERSONA_BINDABLE_STAGES.get(dim, _PERSONA_BINDABLE_STAGES["B"])
    candidates: List[ClusterData] = []
    for stage in stage_order:
        for c in all_clusters.get(stage, []):
            key = (stage, int(c.cluster_label))
            if dim == "A" and key in used_audience_keys:
                continue
            candidates.append(c)
    if not candidates:
        return None, False
    ranked = sorted(candidates, key=lambda c: _score_persona_cluster(persona, c), reverse=True)
    best = ranked[0]
    best_score = _score_persona_cluster(persona, best)
    if best_score <= 0 and len(ranked) > 1:
        best = ranked[1] if _score_persona_cluster(persona, ranked[1]) > 0 else best
    return {"stage": best.stage, "label": int(best.cluster_label)}, best_score <= 0


# Persona 名称 / core_pain 中的生理标签 → 绑定簇内 ≥N 条评论含对应英文词方可保留
PERSONA_PHYSIO_MIN_REVIEWS_DEFAULT = 5

_PERSONA_PHYSIO_RULES: List[Tuple[str, Tuple[str, ...], Tuple[str, ...]]] = [
    ("pregnancy", ("孕妇", "孕期", "怀胎", "产后"), ("pregnant", "pregnancy", "pregnan", "postpartum", "maternity", "weeks pregnant")),
    ("elderly", ("老年", "老人", "长者"), ("elderly", "senior", "older adult", "retired")),
    ("children", ("儿童", "小孩", "幼儿", "婴儿"), ("kid", "child", "children", "toddler", "infant", "baby")),
    ("diabetes", ("糖尿病",), ("diabetic", "diabetes")),
    ("scar", ("疤痕", "妊娠纹"), ("stretch mark", "scar", "c-section", "cesarean")),
    ("coarse_hair", ("粗硬发", "粗毛", "硬毛"), ("coarse hair", "thick hair", "coarse", "thick")),
    ("sensitive_skin", ("敏感肌", "敏感皮", "易敏"), ("sensitive skin", "sensitive", "delicate skin", "irritat")),
]


def _review_text(r: ReviewRecord) -> str:
    return f"{r.title} {r.content}".lower()


def count_physio_reviews_in_cluster(
    source_rows: List[int],
    reviews: List[ReviewRecord],
    en_tokens: Tuple[str, ...],
) -> int:
    """统计簇内至少命中一个英文 token 的评论条数（按 source_row 去重）。"""
    row_map = {r.source_row: r for r in reviews}
    count = 0
    for sr in source_rows:
        r = row_map.get(sr)
        if not r:
            continue
        text = _review_text(r)
        if any(t in text for t in en_tokens):
            count += 1
    return count


def compute_physio_review_counts(
    source_rows: List[int],
    reviews: List[ReviewRecord],
) -> Dict[str, int]:
    """各生理标签维度在簇内的评论命中条数（供 catalog 展示与校验）。"""
    counts: Dict[str, int] = {}
    for rule_key, _, en_tokens in _PERSONA_PHYSIO_RULES:
        n = count_physio_reviews_in_cluster(source_rows, reviews, en_tokens)
        if n > 0:
            counts[rule_key] = n
    return counts


_GENERIC_PERSONA_ANCHOR_EXACT = frozenset({
    "self", "unknown", "user", "customer", "buyer", "myself", "consumer", "reviewer",
})


# _DISTINCTIVE_ANCHOR_MARKERS removed — no longer used (audience clustering removed)

def is_generic_persona_anchor(
    semantic_label: str,
    top_phrases: Optional[List[str]] = None,
) -> bool:
    """品类泛词 / self 大桶：不宜作为 Persona A 维统计锚点。"""
    label = (semantic_label or "").strip().lower()
    if label in _GENERIC_PERSONA_ANCHOR_EXACT:
        return True
    if label.endswith(" user") or label.endswith(" buyer") or label.endswith(" customer"):
        if not any(m in label for m in _DISTINCTIVE_ANCHOR_MARKERS):
            return True
    for phrase in (top_phrases or [])[:6]:
        p = (phrase or "").strip().lower()
        if p in _GENERIC_PERSONA_ANCHOR_EXACT or p == "self":
            return True
    return False


def enrich_persona_catalog_physio_counts(
    catalog: List[Dict[str, Any]],
    all_clusters: Dict[str, List[ClusterData]],
    reviews: List[ReviewRecord],
) -> None:
    """为 persona_cluster_catalog 各簇附加 physio_review_counts（内部/调试用）。"""
    registry = build_cluster_registry(all_clusters)
    for entry in catalog:
        key = (entry.get("stage"), entry.get("label"))
        cluster = registry.get(key)
        if not cluster:
            continue
        entry["physio_review_counts"] = compute_physio_review_counts(
            cluster.source_rows, reviews,
        )


def prepare_persona_catalog_for_llm(
    catalog: List[Dict[str, Any]],
    all_clusters: Dict[str, List[ClusterData]],
    reviews: List[ReviewRecord],
    min_physio_reviews: int = PERSONA_PHYSIO_MIN_REVIEWS_DEFAULT,
) -> None:
    """喂 LLM 前：仅保留达标生理标签；suggested_dimension 已在 data_loader 中设置。"""
    registry = build_cluster_registry(all_clusters)
    for entry in catalog:
        key = (entry.get("stage"), entry.get("label"))
        cluster = registry.get(key)
        if not cluster:
            continue

        all_counts = compute_physio_review_counts(cluster.source_rows, reviews)
        eligible = {
            k: n for k, n in all_counts.items()
            if n >= min_physio_reviews
        }
        entry["eligible_physio_labels"] = eligible
        entry.pop("physio_review_counts", None)
        entry["valid_persona_anchor"] = True


def _physio_label_has_review_evidence(
    en_tokens: Tuple[str, ...],
    source_rows: List[int],
    reviews: List[ReviewRecord],
    min_review_count: int,
) -> bool:
    return count_physio_reviews_in_cluster(source_rows, reviews, en_tokens) >= min_review_count


def _persona_name_from_semantic(semantic_label: str, max_len: int = 6) -> str:
    """从簇 semantic_label 生成中性 Persona 名（无未佐证生理标签）。"""
    sl = (semantic_label or "").strip().lower()
    hints: List[Tuple[Tuple[str, ...], str]] = [
        (("first time", "first-time", "first purchase"), "初次使用用户"),
        (("travel", "on the go", "portable", "compact"), "便携用户"),
        (("gift", "present", "birthday", "christmas"), "送礼用户"),
        (("sensitive skin", "sensitive", "allergy", "allergic"), "敏感体质用户"),
        (("elderly", "senior", "older"), "老年用户"),
        (("kid", "child", "children", "toddler", "infant", "baby"), "儿童家长"),
        (("my dog", "my cat", "my pet", "puppy", "kitten"), "宠物主"),
        (("pregnant", "pregnancy", "nursing"), "孕期/哺乳期用户"),
        (("professional", "contractor", "chef"), "专业用户"),
        (("beginner", "new to", "never used"), "新手用户"),
        (("coarse", "thick", "curly", "fine"), "特定体质用户"),
    ]
    for patterns, name in hints:
        if any(p in sl for p in patterns):
            return name[:max_len]
    if sl.endswith(" user") or sl.endswith(" buyer"):
        return "场景用户"[:max_len]
    raw = (semantic_label or "用户群体").strip()
    if len(raw) <= max_len:
        return raw if raw.endswith(("用户", "者", "族", "群体")) else f"{raw[:max_len - 2]}用户"[:max_len]
    return "场景用户"[:max_len]


def _strip_physio_from_core_pain(core_pain: str, unsupported_zh: List[str]) -> str:
    if not core_pain:
        return core_pain
    parts = re.split(r"[；;]", core_pain)
    kept = [p.strip() for p in parts if p.strip() and not any(z in p for z in unsupported_zh)]
    if kept:
        return "；".join(kept[:3])
    # 全部含无佐证标签时，去掉标签词保留其余
    cleaned = core_pain
    for zh in unsupported_zh:
        cleaned = cleaned.replace(zh, "")
    cleaned = re.sub(r"[；;]+", "；", cleaned).strip("； ")
    return cleaned or core_pain


def validate_persona_physiological_labels(
    personas: List[Dict[str, Any]],
    cluster_data: Dict[str, Any],
    reviews: List[ReviewRecord],
    all_clusters: Dict[str, List[ClusterData]],
    min_review_count: int = PERSONA_PHYSIO_MIN_REVIEWS_DEFAULT,
) -> List[Dict[str, Any]]:
    """校验 Persona 生理标签：无佐证时改用 semantic 中性命名，不再拆字。"""
    catalog = cluster_data.get("persona_cluster_catalog") or []
    catalog_map = {(e.get("stage"), e.get("label")): e for e in catalog}
    registry = build_cluster_registry(all_clusters)

    for persona in personas:
        refs = persona.get("cluster_refs") or {}
        a_ref = (
            _normalize_cluster_ref(refs.get("A"))
            if isinstance(refs, dict) else None
        ) or _get_persona_anchor_ref(persona)
        entry = catalog_map.get((a_ref["stage"], a_ref["label"])) if a_ref else None
        cluster = registry.get((a_ref["stage"], a_ref["label"])) if a_ref else None
        source_rows = list(cluster.source_rows) if cluster else []
        eligible = (entry or {}).get("eligible_physio_labels") or {}

        unsupported: List[str] = []
        text_blob = f"{persona.get('name', '')} {persona.get('core_pain', '')}"
        for rule_key, zh_markers, en_tokens in _PERSONA_PHYSIO_RULES:
            hit_zh = [z for z in zh_markers if z in text_blob]
            if not hit_zh:
                continue
            if rule_key in eligible:
                continue
            if not _physio_label_has_review_evidence(
                en_tokens, source_rows, reviews, min_review_count,
            ):
                unsupported.extend(hit_zh)

        if not unsupported:
            continue

        old_name = persona.get("name", "")
        semantic = str((entry or {}).get("semantic_label") or "")
        persona["name"] = _persona_name_from_semantic(semantic)
        persona["core_pain"] = _strip_physio_from_core_pain(
            str(persona.get("core_pain") or ""), unsupported,
        )
        persona["physio_label_corrected"] = True
        persona["physio_removed_labels"] = sorted(set(unsupported))
        logger.warning(
            "Persona 生理标签评论佐证不足（需≥%s条），已改用 semantic 命名: %s → %s（移除 %s；eligible %s）",
            min_review_count, old_name, persona["name"],
            persona["physio_removed_labels"], eligible,
        )

    return personas


def _remap_generic_a_anchor(
    persona: Dict[str, Any],
    refs: Dict[str, Optional[Dict[str, Any]]],
    all_clusters: Dict[str, List[ClusterData]],
) -> bool:
    """若 A 维绑定了泛词簇，尝试改绑具体 persona_signal 簇。"""
    a_ref = refs.get("A")
    if not a_ref or a_ref["stage"] != "1_audience":
        return False

    aud_clusters = all_clusters.get("1_audience", [])
    aud_registry = {int(c.cluster_label): c for c in aud_clusters}
    current = aud_registry.get(int(a_ref["label"]))
    if current and not is_generic_persona_anchor(
        current.top_phrases[0] if current.top_phrases else "",
        current.top_phrases,
    ):
        return False

    candidates = [
        c for c in aud_clusters
        if not is_generic_persona_anchor(
            c.top_phrases[0] if c.top_phrases else "",
            c.top_phrases,
        )
    ]
    if not candidates:
        refs["A"] = None
        persona["generic_anchor_cleared"] = True
        logger.warning(
            "Persona %s 绑定了泛词 A 维簇 label=%s，且无可用具体簇，已清除 A 维",
            persona.get("name"), a_ref.get("label"),
        )
        return True

    ranked = sorted(
        candidates,
        key=lambda c: _score_persona_cluster(persona, c),
        reverse=True,
    )
    best = ranked[0]
    if _score_persona_cluster(persona, best) <= 0:
        refs["A"] = None
        persona["generic_anchor_cleared"] = True
        return True

    refs["A"] = {"stage": "1_audience", "label": int(best.cluster_label)}
    persona["anchor_remapped"] = True
    persona["anchor_remapped_from"] = a_ref
    logger.warning(
        "Persona %s A 维泛词簇 label=%s 已改绑至具体簇 label=%s",
        persona.get("name"), a_ref.get("label"), best.cluster_label,
    )
    return True


def assign_persona_clusters(
    personas: List[Dict[str, Any]],
    all_clusters: Dict[str, List[ClusterData]],
) -> List[Dict[str, Any]]:
    """为每个 Persona 绑定 A/B/C 三维聚类簇；缺失维度自动映射。"""
    registry = build_cluster_registry(all_clusters)
    used_audience_keys: Set[Tuple[str, int]] = set()

    for persona in personas:
        normalize_persona_dimension(persona)
        refs = _parse_cluster_refs_from_persona(persona)
        auto_mapped_dims: List[str] = []

        for dim in _PERSONA_DIM_ORDER:
            ref = refs.get(dim)
            if ref and (ref["stage"], ref["label"]) in registry:
                refs[dim] = ref
                continue
            mapped, was_auto = _auto_map_persona_dimension(
                persona, dim, all_clusters, used_audience_keys,
            )
            if mapped and (mapped["stage"], mapped["label"]) in registry:
                refs[dim] = mapped
                if was_auto:
                    auto_mapped_dims.append(dim)
            else:
                refs[dim] = None

        valid_refs = {d: r for d, r in refs.items() if r}
        if not valid_refs:
            persona.pop("cluster_refs", None)
            persona.pop("cluster_ref", None)
            persona["cluster_unassigned"] = True
            continue

        valid_refs = {d: r for d, r in refs.items() if r}
        if not valid_refs:
            persona.pop("cluster_refs", None)
            persona.pop("cluster_ref", None)
            persona["cluster_unassigned"] = True
            continue

        persona["cluster_refs"] = valid_refs
        persona.pop("cluster_unassigned", None)
        if auto_mapped_dims:
            persona["cluster_auto_mapped"] = True
            persona["cluster_auto_mapped_dims"] = auto_mapped_dims

        a_ref = refs.get("A")
        if a_ref:
            used_audience_keys.add((a_ref["stage"], a_ref["label"]))

        _sync_legacy_cluster_ref(persona)

    return personas


def enrich_persona_keywords(
    persona: Dict[str, Any],
    cluster: Optional[ClusterData] = None,
    extra_clusters: Optional[List[ClusterData]] = None,
) -> Dict[str, Any]:
    """从绑定簇 top_phrases 补充 keywords（过滤无信息词）。"""
    clusters: List[ClusterData] = []
    if cluster:
        clusters.append(cluster)
    if extra_clusters:
        for c in extra_clusters:
            if c not in clusters:
                clusters.append(c)
    base = [k.lower().strip() for k in (persona.get("keywords") or []) if k and k.strip()]
    from_cluster: List[str] = []
    for cl in clusters:
        for seg in phrase_segments(cl.top_phrases):
            if seg in _PERSONA_PHRASE_STOPWORDS or len(seg) < 3:
                continue
            from_cluster.append(seg)
    merged = list(dict.fromkeys(base + from_cluster))
    if not merged and clusters:
        merged = [
            s for cl in clusters for s in phrase_segments(cl.top_phrases)
            if s not in _PERSONA_PHRASE_STOPWORDS and len(s) >= 4
        ][:20]
    persona["keywords"] = merged[:40]
    return persona




# Persona 关键词分词噪声词（不应作为独立词元匹配）
_PERSONA_NOISE_1GRAM = frozenset({
    "the", "and", "for", "not", "but", "was", "are", "has", "had", "have",
    "that", "this", "with", "from", "they", "them", "their", "will", "would",
    "need", "just", "very", "like", "than", "then", "its", "does", "been",
    "much", "more", "some", "also", "only", "don", "can", "get", "got",
    "use", "fit", "put", "try", "see", "buy", "one", "two", "per", "old", "new",
    "really", "still", "even", "well", "make", "made", "way", "bit", "lot",
    "product", "item", "review", "amazon", "self",
})

def _tokenize_persona_keywords(keywords: List[str]) -> List[str]:
    """Persona 关键词分词：逗号拆分 + 长短语拆为有意义的 bigram/trigram。"""
    out: List[str] = []
    for kw in expand_keywords(keywords):
        out.append(kw)
        # 对长复合短语（>20 chars 且无逗号）拆分为有意义的 n-gram
        if len(kw) > 20 and "," not in kw:
            toks = [t.strip().rstrip(".,;!?") for t in kw.split()
                    if len(t.strip()) >= 3 and t.strip().lower() not in _PERSONA_NOISE_1GRAM]
            # 添加有意义 2-gram
            for i in range(len(toks) - 1):
                a, b = toks[i].lower(), toks[i+1].lower()
                if a in _PERSONA_NOISE_1GRAM or b in _PERSONA_NOISE_1GRAM:
                    continue
                bigram = f"{toks[i]} {toks[i+1]}"
                if len(bigram) >= 7:
                    out.append(bigram)
            # 添加有意义 3-gram（更有区分度）
            for i in range(len(toks) - 2):
                if any(toks[i+j].lower() in _PERSONA_NOISE_1GRAM for j in range(3)):
                    continue
                trigram = f"{toks[i]} {toks[i+1]} {toks[i+2]}"
                if len(trigram) >= 10:
                    out.append(trigram)
    return list(dict.fromkeys(out))

def match_persona_text(keywords: List[str], text: str) -> bool:
    """Persona 关键词匹配：单关键词命中即可（Persona 画像比主题更宽泛）。"""
    text = text.lower()
    expanded = [
        k for k in _tokenize_persona_keywords(keywords)
        if k not in _PERSONA_PHRASE_STOPWORDS and len(k) >= 3
    ]
    if not expanded:
        return False
    # 强关键词（≥12字符或含产品关键词）：任意 1 个命中即可
    strong = [k for k in expanded if _is_strong_keyword(k)]
    for kw in sorted(strong, key=len, reverse=True):
        if kw in text:
            return True
    # 弱关键词：≥5 字符且命中 1 个即可（但需 total expanded ≤15，避免过短泛词误判）
    weak = [k for k in expanded if k not in strong and len(k) >= 5]
    if len(expanded) <= 15:
        for kw in sorted(weak, key=len, reverse=True):
            if kw in text:
                return True
    return False


def resolve_persona_audience_label(
    persona: Dict[str, Any],
    all_clusters: Dict[str, List[ClusterData]],
) -> int:
    """推断 Persona 对应的 audience 簇标签（供根因 per_audience 使用）。"""
    refs = persona.get("cluster_refs")
    if isinstance(refs, dict):
        a_ref = _normalize_cluster_ref(refs.get("A"))
        if a_ref and a_ref["stage"] == "1_audience":
            return int(a_ref["label"])

    ref = persona.get("cluster_ref") or {}
    stage = ref.get("stage", "")
    if stage == "1_audience" and ref.get("label") is not None:
        return int(ref["label"])

    if "_audience_c" in stage:
        try:
            return int(stage.rsplit("_audience_c", 1)[-1])
        except ValueError:
            pass

    persona_rows = set(persona.get("source_rows") or [])
    aud_clusters = all_clusters.get("1_audience", [])
    if persona_rows and aud_clusters:
        best_label, best_overlap = -1, 0
        for ac in aud_clusters:
            overlap = len(persona_rows & set(ac.source_rows))
            if overlap > best_overlap:
                best_overlap = overlap
                best_label = int(ac.cluster_label)
        if best_label >= 0:
            return best_label

    if aud_clusters:
        return int(max(aud_clusters, key=lambda c: c.review_count).cluster_label)
    return -1


def _apply_persona_pct_display(persona: Dict[str, Any], total: int) -> None:
    hit = persona.get("hit_count", 0)
    if hit <= 0:
        persona["pct_estimate"] = None
        persona["pct_verified"] = False
        persona["pct_display"] = "待验证"
        return
    raw_pct = hit / max(total, 1) * 100
    persona["pct_estimate"] = round(raw_pct, 1)
    persona["pct_verified"] = True
    if raw_pct < 0.5:
        persona["pct_display"] = "<1%"
    elif raw_pct < 10:
        persona["pct_display"] = f"{persona['pct_estimate']:.1f}%"
    else:
        persona["pct_display"] = f"{round(raw_pct)}%"


def compute_persona_pcts(
    personas: List[Dict[str, Any]],
    reviews: List[ReviewRecord],
    all_clusters: Optional[Dict[str, List[ClusterData]]] = None,
) -> List[Dict[str, Any]]:
    """Persona 命中数/占比：以 A 维锚点簇 source_rows + 全绑定簇 keywords 过滤。"""
    total = len(reviews)
    row_map = {r.source_row: r for r in reviews}
    registry = build_cluster_registry(all_clusters or {})

    for persona in personas:
        normalize_persona_dimension(persona)
        refs = _parse_cluster_refs_from_persona(persona)
        if persona.get("cluster_refs"):
            refs = {
                d: _normalize_cluster_ref((persona.get("cluster_refs") or {}).get(d))
                for d in _PERSONA_DIM_ORDER
            }

        bound_clusters = _clusters_from_refs(refs, registry)
        anchor_ref = refs.get("A") or _get_persona_anchor_ref(persona)
        anchor = registry.get((anchor_ref["stage"], anchor_ref["label"])) if anchor_ref else None

        extra = [c for c in bound_clusters if c is not anchor]
        if not anchor and bound_clusters:
            bc_only = [c for c in bound_clusters if c.stage != "1_audience"]
            if bc_only:
                anchor = min(bc_only, key=lambda c: c.review_count)
                extra = [c for c in bound_clusters if c is not anchor]

        enrich_persona_keywords(persona, anchor, extra_clusters=extra or None)

        if not anchor:
            persona["source_rows"] = []
            persona["hit_count"] = 0
            persona["stat_method"] = "unassigned"
            persona["audience_label"] = -1
            _apply_persona_pct_display(persona, total)
            continue

        anchor_rows = set(anchor.source_rows)
        bc_clusters = [c for c in bound_clusters if c is not anchor]

        if anchor.stage != "1_audience":
            # 无 A 维时以 B/C 簇为统计池
            kws = persona.get("keywords") or []
            stat_pool = anchor_rows
        else:
            kws = persona.get("keywords") or []
            stat_pool = anchor_rows

        filtered = {
            sr for sr in stat_pool
            if sr in row_map and match_persona_text(
                kws, row_map[sr].title + " " + row_map[sr].content,
            )
        }
        min_keep = max(10, int(len(stat_pool) * 0.03))
        if filtered:
            hit_rows = filtered
            persona["stat_method"] = (
                "anchor+keywords" if len(filtered) >= min_keep else "keyword_low_sample"
            )
        else:
            hit_rows = set()
            persona["stat_method"] = "no_match"

        # 回退：关键词过滤命中过少时，按锚点簇比例兜底（至少保留 3% 或 15 条）
        fallback_min = max(15, int(len(anchor_rows) * 0.03))
        if len(hit_rows) < fallback_min:
            fallback_rows = set(list(anchor_rows)[:fallback_min])
            persona["stat_method"] = persona.get("stat_method", "no_match") + "+fallback"
            hit_rows = hit_rows | fallback_rows

        persona["source_rows"] = sorted(hit_rows)
        persona["hit_count"] = len(hit_rows)
        persona["cluster_review_count"] = len(anchor_rows)
        persona["audience_label"] = resolve_persona_audience_label(persona, all_clusters or {})
        _apply_persona_pct_display(persona, total)

    return personas


def _truncate_quote(text: str, max_len: int = 220) -> str:
    text = re.sub(r"\s+", " ", text.strip())
    if len(text) <= max_len:
        return text
    return text[: max_len - 1] + "…"


def _persona_quote_relevant(persona: Dict[str, Any], text: str, is_neg: bool) -> bool:
    """过滤与 Persona 定义明显矛盾的引用。"""
    t = text.lower()
    name = (persona.get("name") or "").lower()
    pain = (persona.get("core_pain") or "").lower()
    if ("敏感" in name or "sensitive" in " ".join(persona.get("keywords") or [])) and (
        "don't have sensitive" in t or "do not have sensitive" in t or "not sensitive skin" in t
    ):
        return False
    if ("粗硬" in name or "coarse" in t or "thick hair" in t) and is_neg is False:
        if not any(k in t for k in ("coarse", "thick", "coarse hair", "thick hair", "coarse/dark")):
            if "粗硬" in pain and "bikini" in t and "pull" not in t:
                pass  # 可能仍相关
    if is_neg and _quote_sentiment(text, []) == "pos":
        return False
    return True


def quote_cn_summary(text: str) -> str:
    """英文引用的一句话中文摘要（规则匹配，无需额外 LLM）。"""
    t = text.lower()
    for patterns, hint in _QUOTE_CN_HINTS:
        if any(p in t for p in patterns):
            return hint
    return "用户反馈摘要"


_NEG_SENTIMENTS = frozenset({"negative"})
_POS_SENTIMENTS = frozenset({"positive"})


def _match_keywords_in_blob(keywords: List[str], blob: str) -> bool:
    """在已小写的 blob 中匹配展开后的 keywords。"""
    blob = blob.lower()
    expanded = expand_keywords(keywords)
    strong = [k for k in expanded if _is_strong_keyword(k)]
    weak = [k for k in expanded if k not in strong]
    for kw in sorted(strong, key=len, reverse=True):
        if len(kw) >= 3 and kw in blob:
            return True
    if not strong:
        for kw in sorted(weak, key=len, reverse=True):
            if len(kw) >= 4 and kw in blob:
                return True
    return False


def _persona_extraction_blob(ext: Dict[str, Any]) -> str:
    """Persona 匹配用：persona_signals + pain_points。"""
    parts: List[str] = []
    for key in ("persona_signals", "pain_points"):
        val = ext.get(key) or []
        if isinstance(val, list):
            parts.extend(str(x).strip() for x in val if str(x).strip())
    return " ".join(parts)


def match_persona_extraction(keywords: List[str], ext: Dict[str, Any]) -> bool:
    blob = _persona_extraction_blob(ext)
    if not blob:
        return False
    return _match_keywords_in_blob(keywords, blob)


def match_rootcause_extraction(keywords: List[str], ext: Dict[str, Any]) -> bool:
    """根因匹配：product_feedback 中 sentiment=Negative 的 aspect + opinion。"""
    if not keywords:
        return False
    for fb in ext.get("product_feedback") or []:
        if not isinstance(fb, dict):
            continue
        sent = str(fb.get("sentiment") or "").strip().lower()
        if sent not in _NEG_SENTIMENTS:
            continue
        blob = f"{fb.get('aspect', '')} {fb.get('opinion', '')}"
        if _match_keywords_in_blob(keywords, blob):
            return True
    return False


def match_theme_extraction(
    keywords: List[str],
    ext: Dict[str, Any],
    *,
    is_neg: bool = True,
) -> bool:
    """主题匹配：product_feedback 的 category + aspect（按情感极性过滤）。"""
    if not keywords:
        return False
    allowed = _NEG_SENTIMENTS if is_neg else _POS_SENTIMENTS
    for fb in ext.get("product_feedback") or []:
        if not isinstance(fb, dict):
            continue
        sent = str(fb.get("sentiment") or "").strip().lower()
        if sent not in allowed:
            continue
        blob = f"{fb.get('category', '')} {fb.get('aspect', '')}"
        if _match_keywords_in_blob(keywords, blob):
            return True
    return False


def build_quote_blocks_html(
    quote_items: List[Dict[str, Any]],
    asin_labels: Dict[str, str],
    *,
    default_neg: bool = True,
    empty_msg: str = "暂无与画像匹配的代表性评论",
) -> str:
    """将多条引用渲染为 HTML quote 块。"""
    parts: List[str] = []
    for q in quote_items:
        text = (q.get("quote") or q.get("text") or "").strip()
        if not text:
            continue
        rating = q.get("rating")
        if rating is not None:
            is_neg = float(rating) <= 2
        else:
            is_neg = q.get("is_neg", default_neg)
        q_cls = "neg" if is_neg else "pos"
        asin = q.get("asin") or "?"
        url = q.get("amazon_url") or (
            f"https://www.amazon.com/dp/{asin}" if asin != "?" else "#"
        )
        star_hint = f" ({int(rating)}★)" if rating is not None else ""
        cn = q.get("cn_summary") or quote_cn_summary(text)
        label = asin_labels.get(asin, asin)
        parts.append(
            f'<div class="quote {q_cls}">"{text}" '
            f'— <a href="{url}" target="_blank" rel="noopener">{label}</a>{star_hint}'
            f'<div class="quote-cn">摘要：{cn}</div></div>'
        )
    if parts:
        return "\n".join(parts)
    return f'<div class="quote-empty">{empty_msg}</div>'


def _review_to_quote_entry(r: ReviewRecord) -> Dict[str, Any]:
    return {
        "quote": _truncate_quote(r.content),
        "asin": r.asin,
        "rating": r.rating,
        "amazon_url": f"https://www.amazon.com/dp/{r.asin}",
        "is_neg": r.rating <= 2,
        "cn_summary": quote_cn_summary(r.content),
        "source_row": r.source_row,
    }


def _persona_struct_search_keywords(persona: Dict[str, Any]) -> List[str]:
    """结构化匹配仅用英文聚类 keywords（persona_signals / pain_points 为英文）。"""
    return expand_keywords(list(persona.get("keywords") or []))


def _persona_quote_search_keywords(persona: Dict[str, Any]) -> List[str]:
    """原文 keyword 匹配：聚类 keywords + 中文描述字段。"""
    kws = list(persona.get("keywords") or [])
    for key in ("core_pain", "core_need", "purchase_motivation", "name"):
        val = persona.get(key)
        if isinstance(val, str) and val.strip():
            kws.append(val.strip())
    return expand_keywords(kws)


def _sort_persona_quote_candidates(
    candidates: List[ReviewRecord],
    *,
    struct_rows: Set[int],
    globally_used: Set[int],
    allow_reuse: bool,
) -> List[ReviewRecord]:
    def _key(r: ReviewRecord) -> tuple:
        reuse_penalty = 0
        if not allow_reuse and r.source_row in globally_used:
            reuse_penalty = 2
        elif r.source_row in globally_used:
            reuse_penalty = 1
        struct_penalty = 0 if r.source_row in struct_rows else 1
        return (reuse_penalty, struct_penalty, abs(len(r.content) - 180))

    return sorted(candidates, key=_key)


def pick_persona_quotes(
    personas: List[Dict[str, Any]],
    reviews: List[ReviewRecord],
    extractions: Optional[Dict[int, Dict[str, Any]]] = None,
    *,
    max_quotes: int = 3,
) -> List[Dict[str, Any]]:
    """为每个 Persona 选取代表性引用：命中池 source_rows → 结构化/原文匹配 → 多层回退。"""
    globally_used: Set[int] = set()
    row_map = {r.source_row: r for r in reviews}
    ext_map = extractions or {}
    results: List[Dict[str, Any]] = []

    for persona in personas:
        bound_rows = set(persona.get("source_rows") or [])
        struct_kws = _persona_struct_search_keywords(persona)
        text_kws = _persona_quote_search_keywords(persona)
        pool_reviews: List[ReviewRecord]
        if bound_rows:
            pool_reviews = [row_map[sr] for sr in sorted(bound_rows) if sr in row_map]
        else:
            pool_reviews = [
                r for r in reviews
                if match_persona_text(text_kws, r.title + " " + r.content)
            ]

        def _struct_matched() -> List[ReviewRecord]:
            if not ext_map or not struct_kws:
                return []
            out: List[ReviewRecord] = []
            for r in pool_reviews:
                ext = ext_map.get(r.source_row)
                if ext and match_persona_extraction(struct_kws, ext):
                    out.append(r)
            return out

        def _text_matched(cands: List[ReviewRecord]) -> List[ReviewRecord]:
            return [
                r for r in cands
                if match_persona_text(text_kws, r.title + " " + r.content)
            ]

        struct_pool = _struct_matched()
        text_pool = _text_matched(pool_reviews)
        struct_rows = {r.source_row for r in struct_pool}
        base_pool = struct_pool + [
            r for r in text_pool if r.source_row not in struct_rows
        ]
        if not base_pool:
            base_pool = list(pool_reviews)

        def _tier(cands: List[ReviewRecord], *, allow_reuse: bool) -> List[ReviewRecord]:
            ranked = _sort_persona_quote_candidates(
                cands, struct_rows=struct_rows, globally_used=globally_used, allow_reuse=allow_reuse,
            )
            out: List[ReviewRecord] = []
            for r in ranked:
                is_neg = r.rating <= 2
                if _persona_quote_relevant(persona, r.content, is_neg):
                    out.append(r)
            return out

        selected: List[ReviewRecord] = []
        seen: Set[int] = set()

        def _take_from(tier: List[ReviewRecord]) -> None:
            nonlocal selected
            for r in tier:
                if r.source_row in seen:
                    continue
                selected.append(r)
                seen.add(r.source_row)
                if len(selected) >= max_quotes:
                    return

        # 优先差评 → 好评 → 3★ → 命中池兜底（先尽量不跨 Persona 重复）
        for allow_reuse in (False, True):
            if len(selected) >= max_quotes:
                break
            neg = [r for r in base_pool if r.rating <= 2]
            pos = [r for r in base_pool if r.rating >= 4]
            mid = [r for r in base_pool if 2 < r.rating < 4]
            for tier_src in (neg, pos, mid, base_pool, pool_reviews):
                _take_from(_tier(tier_src, allow_reuse=allow_reuse))
                if len(selected) >= max_quotes:
                    break

        # 仍空：放宽 relevance，直接从命中池取代表性评论
        if not selected and pool_reviews:
            selected = sorted(
                pool_reviews,
                key=lambda r: (abs(len(r.content) - 180), -r.rating),
            )[:max_quotes]

        quotes: List[Dict[str, Any]] = []
        for r in selected[:max_quotes]:
            globally_used.add(r.source_row)
            quotes.append(_review_to_quote_entry(r))

        entry: Dict[str, Any] = {
            "persona": persona.get("name", "?"),
            "quotes": quotes,
            "quote": None,
            "asin": None,
            "rating": None,
            "amazon_url": None,
            "is_neg": False,
            "cn_summary": None,
        }
        if quotes:
            entry.update({k: quotes[0][k] for k in (
                "quote", "asin", "rating", "amazon_url", "is_neg", "cn_summary",
            )})
        results.append(entry)
    return results


_JTBD_FIELD_KEYS = (
    "core_job",
    "functional_motivation",
    "emotional_motivation",
    "social_motivation",
    "trigger",
)

_JTBD_EMPTY_MARKERS = frozenset({"", "-", "—", "–", "n/a", "none", "null", "无", "暂无", "留空"})

_SOCIAL_MOTIVATION_HINTS = (
    "社交", "他人", "朋友", "伴侣", "老公", "老婆", "party", "gift", "recommend",
    "embarrass", "体面", "social", "husband", "wife", "partner", "date", "public",
    "boyfriend", "girlfriend", "family", "mom", "daughter", "son", "people",
)


def _normalize_jtbd_cell(val: Any) -> str:
    text = str(val or "").strip()
    if text.lower() in _JTBD_EMPTY_MARKERS:
        return "-"
    return text


def _persona_evidence_blob(persona: Dict[str, Any]) -> str:
    parts: List[str] = []
    for key in ("keywords", "core_pain", "core_need", "purchase_motivation", "name"):
        val = persona.get(key)
        if isinstance(val, list):
            parts.extend(str(x) for x in val if x)
        elif val:
            parts.append(str(val))
    return " ".join(parts)


def _jtbd_evidence_tokens(blob: str) -> Tuple[Set[str], str]:
    lower = blob.lower()
    tokens: Set[str] = set()
    for seg in re.findall(r"[a-z]{3,}", lower):
        tokens.add(seg)
    for seg in re.findall(r"[\u4e00-\u9fff]{2,}", blob):
        tokens.add(seg)
    return tokens, lower


def _jtbd_field_has_evidence(text: str, tokens: Set[str], blob: str) -> bool:
    text = _normalize_jtbd_cell(text)
    if text == "-":
        return False
    tl = text.lower()
    blob_lower = blob.lower()
    for token in tokens:
        if len(token) >= 3 and token.isascii() and token in tl:
            return True
        if not token.isascii() and token in text:
            return True
    if len(text) >= 2:
        for i in range(len(text) - 1):
            seg = text[i : i + 2]
            if seg in blob or seg in blob_lower:
                return True
    return False


def _jtbd_social_has_evidence(text: str, tokens: Set[str], blob: str) -> bool:
    if not _jtbd_field_has_evidence(text, tokens, blob):
        return False
    combined = f"{blob} {text}".lower()
    return any(h in combined for h in _SOCIAL_MOTIVATION_HINTS)


def fix_jtbd_fields(
    jtbd: List[Dict[str, Any]],
    personas: Optional[List[Dict[str, Any]]] = None,
) -> List[Dict[str, Any]]:
    """修正 functional / emotional 错位，校验各动机字段是否有 Persona 数据佐证。"""
    persona_map = {p.get("name"): p for p in (personas or []) if p.get("name")}

    for item in jtbd:
        func = _normalize_jtbd_cell(item.get("functional_motivation"))
        emo = _normalize_jtbd_cell(item.get("emotional_motivation"))
        func_is_emotional = func != "-" and any(h in func for h in EMOTIONAL_HINTS)
        emo_is_functional = emo != "-" and not any(h in emo for h in EMOTIONAL_HINTS)
        if emo == "-" and func_is_emotional:
            item["emotional_motivation"] = func
            item["functional_motivation"] = "-"
        elif emo_is_functional and func == "-":
            item["functional_motivation"] = emo
            item["emotional_motivation"] = "-"

        persona = persona_map.get(item.get("persona", ""), {})
        blob = _persona_evidence_blob(persona)
        tokens, _ = _jtbd_evidence_tokens(blob)

        for key in _JTBD_FIELD_KEYS:
            raw = _normalize_jtbd_cell(item.get(key))
            if not persona or not blob.strip():
                item[key] = raw if raw != "-" else "-"
                continue
            if key == "social_motivation":
                item[key] = raw if _jtbd_social_has_evidence(raw, tokens, blob) else "-"
            else:
                item[key] = raw if _jtbd_field_has_evidence(raw, tokens, blob) else "-"

    return jtbd


def filter_matrix_rows(
    matrix: List[Dict[str, Any]],
    market_avg: Optional[float] = None,
    max_personas: int = 4,
    max_scenes_per_persona: int = 2,
) -> List[Dict[str, Any]]:
    """过滤空场景行，每人群最多保留 N 个有效场景；全市场低分时校准「高满意度」。"""
    filtered: List[Dict[str, Any]] = []
    per_persona: Dict[str, int] = {}
    seen_personas: List[str] = []
    market_low = market_avg is not None and market_avg < 3.5
    for row in matrix:
        scene = (row.get("scene") or "").strip()
        if not scene or scene in ("—", "-", "N/A"):
            continue
        persona = row.get("persona", "?")
        if persona not in seen_personas:
            if len(seen_personas) >= max_personas:
                continue
            seen_personas.append(persona)
        if per_persona.get(persona, 0) >= max_scenes_per_persona:
            continue
        for key in ("must_be_needs", "performance_needs", "attractive_needs"):
            val = (row.get(key) or "").strip()
            if len(val) > 80:
                row[key] = val[:79] + "…"
        note_val = (row.get("satisfaction_note") or "").strip()
        if len(note_val) > 60:
            row["satisfaction_note"] = note_val[:59] + "…"
        note = row.get("satisfaction_note", "")
        if row.get("satisfaction") == "高":
            if market_low:
                row["satisfaction"] = "中等"
                note = f"全市场均分 {market_avg} 偏低，该场景相对竞品尚可但非绝对满意。{note}"
            elif "场景" not in note:
                note = f"（细分场景相对满意度，非全市场）{note}"
        row["satisfaction_note"] = note
        filtered.append(row)
        per_persona[persona] = per_persona.get(persona, 0) + 1
    return filtered


_KANO_NO_REVERSE_ITEM = "本品类暂无明确反向型需求"

_KANO_REVERSE_META_HINTS = (
    "暂无明确反向", "候选搜索", "重新判定", "反向型候选", "无明确反向", "其他反向",
)

_KANO_REVERSE_CONCLUSION_HINTS = (
    "暂无明确反向型需求", "暂无明确反向需求", "无明确反向", "低于阈值",
    "未发现明确反向", "判定本品类暂无",
)


def _is_kano_reverse_meta_item(item: Dict[str, Any]) -> bool:
    """占位/搜索结论类反向条目，非真实反向需求。"""
    item_text = (item.get("item") or "").strip()
    blob = " ".join(
        str(item.get(k) or "") for k in ("item", "reason", "evidence", "competitor_status")
    )
    if item_text == _KANO_NO_REVERSE_ITEM:
        return True
    if any(h in item_text for h in _KANO_REVERSE_META_HINTS):
        return True
    return any(h in blob for h in _KANO_REVERSE_CONCLUSION_HINTS)


def normalize_kano_reverse_items(kano: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """无明确反向型需求时，统一输出标准占位条目标题。"""
    non_reverse = [it for it in kano if it.get("type") != "reverse"]
    reverse_items = [it for it in kano if it.get("type") == "reverse"]
    real_reverse = [it for it in reverse_items if not _is_kano_reverse_meta_item(it)]
    if real_reverse:
        return non_reverse + real_reverse

    meta = reverse_items[0] if reverse_items else {}
    placeholder = {
        "type": "reverse",
        "item": _KANO_NO_REVERSE_ITEM,
        "evidence": (meta.get("evidence") or "").strip()
        or "经主动搜索 too many parts / too complicated / too loud / unnecessary / takes too long 等词组，均未达反向型阈值。",
        "affected_persona": (meta.get("affected_persona") or "").strip() or "—",
        "reason": (meta.get("reason") or "").strip() or _KANO_NO_REVERSE_ITEM,
        "competitor_status": meta.get("competitor_status") or "",
    }
    return non_reverse + [placeholder]


def fix_kano_items(kano: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """拆分合并条目，修正反向型误分类。"""
    fixed: List[Dict[str, Any]] = []
    quality_hints = ("缺陷", "故障", "损坏", "脱落", "断裂", "失效", "不工作", "质量")
    reverse_hints = ("too many", "too complicated", "too loud", "功能过多", "太复杂", "太吵")
    for item in kano:
        item_text = item.get("item", "")
        if "·" in item_text or "、" in item_text or "," in item_text:
            parts = re.split(r"[·、,]", item_text)
            for p in parts:
                p = p.strip()
                if p:
                    sub = dict(item)
                    sub["item"] = p
                    fixed.append(sub)
            continue
        t = item.get("type", "performance")
        item_lower = item_text.lower()
        if t == "reverse" and any(h in item_text for h in quality_hints):
            if not any(h in item_lower for h in reverse_hints):
                item["type"] = "must-be"
        fixed.append(item)
    return fixed


_KANO_QUADRANTS: Dict[str, Dict[str, str]] = {
    "must-be": {
        "css": "kano-basic",
        "label": "基本型 Must-be",
        "title": "不满足 → 强烈差评；满足 → 用户不会特别提及",
        "icon": "✕",
        "empty": "暂无基本型需求条目",
    },
    "performance": {
        "css": "kano-perf",
        "label": "期望型 Performance",
        "title": "越好 → 越高分；越差 → 越低分（线性关系）",
        "icon": "↑",
        "empty": "暂无期望型需求条目",
    },
    "attractive": {
        "css": "kano-excite",
        "label": "魅力型 Attractive",
        "title": "有 → 惊喜好评；没有 → 用户不会差评",
        "icon": "★",
        "empty": "暂无魅力型需求条目",
    },
    "reverse": {
        "css": "kano-rev",
        "label": "反向型 Reverse",
        "title": "经扩展搜索验证：用户主动排斥的功能负担；低于门槛时输出搜索过程与结论",
        "icon": "⊘",
        "empty": _KANO_NO_REVERSE_ITEM,
    },
}


def _html_esc(text: str) -> str:
    import html as html_mod
    return html_mod.escape(str(text or ""), quote=False)


def _render_kano_item(item: Dict[str, Any], icon: str) -> str:
    """单条 KANO 需求五字段卡片行。"""
    item_name = _html_esc(item.get("item") or "—")
    evidence = _html_esc(item.get("evidence") or "—")
    persona = _html_esc(item.get("affected_persona") or "—")
    reason = _html_esc(item.get("reason") or "—")
    competitor = _html_esc(item.get("competitor_status") or "—")
    is_reverse_meta = _is_kano_reverse_meta_item(item)
    if is_reverse_meta and not item.get("competitor_status"):
        rows = (
            f'<div class="kf-row"><span class="kf-key">搜索/证据：</span>{evidence}</div>'
            f'<div class="kf-row"><span class="kf-key">结论：</span>{reason}</div>'
        )
        if competitor != "—":
            rows += f'<div class="kf-row"><span class="kf-key">补充：</span>{competitor}</div>'
    else:
        rows = (
            f'<div class="kf-row"><span class="kf-key">频次证据：</span>{evidence}</div>'
            f'<div class="kf-row"><span class="kf-key">影响 Persona：</span>{persona}</div>'
            f'<div class="kf-row"><span class="kf-key">分类原因：</span>{reason}</div>'
            f'<div class="kf-row"><span class="kf-key">竞品现状：</span>{competitor}</div>'
        )
    return (
        f'<div class="kano-item">'
        f'<div class="kano-icon">{icon}</div>'
        f'<div class="kano-fields">'
        f'<div class="kf-name">{item_name}</div>'
        f"{rows}"
        f"</div></div>"
    )


def build_kano_grid_html(kano: List[Dict[str, Any]]) -> str:
    """KANO 四象限 2×2 卡片布局（基本型/期望型/魅力型/反向型）。"""
    by_type: Dict[str, List[Dict[str, Any]]] = {
        "must-be": [], "performance": [], "attractive": [], "reverse": [],
    }
    for item in kano:
        t = item.get("type", "performance")
        if t in by_type:
            by_type[t].append(item)

    cards: List[str] = []
    for ktype in ("must-be", "performance", "attractive", "reverse"):
        meta = _KANO_QUADRANTS[ktype]
        items = by_type[ktype]
        if items:
            body = "".join(_render_kano_item(it, meta["icon"]) for it in items)
        else:
            body = f'<p class="kano-empty">{meta["empty"]}</p>'
        cards.append(
            f'<div class="kano-card {meta["css"]}">'
            f'<div class="kano-label">{meta["label"]}</div>'
            f'<div class="kano-title">{meta["title"]}</div>'
            f"{body}"
            f"</div>"
        )
    return f'<div class="kano-grid">{"".join(cards)}</div>'


def _valid_asins(reviews: List[ReviewRecord]) -> Set[str]:
    return {r.asin for r in reviews if VALID_ASIN_RE.match(r.asin)}


def _quote_sentiment(text: str, keywords: List[str]) -> str:
    """判断引用与根因方向是否一致（negative root cause should not use praise）。"""
    t = text.lower()
    neg_markers = (
        "not recommend", "do not", "don't", "never", "bad", "terrible", "broken",
        "waste", "zero star", "not waterproof", "isn't waterproof", "not water proof",
        "stopped working", "doesn't work", "does not work", "disappointed", "return",
    )
    if any(p in t for p in neg_markers):
        return "neg"
    conditional_praise = ("waterproof", "easy to clean", "works well")
    if any(p in t for p in conditional_praise):
        if any(k in t for k in ("not", "never", "bad", "terrible", "broken", "fail", "but", "however", " although", "stopped", "after")):
            return "neg"
    praise = ("love", "great", "perfect", "amazing")
    if any(p in t for p in praise) and not any(k in t for k in ("not", "never", "bad", "terrible", "broken")):
        return "pos"
    return "neg"


def _rootcause_relaxed_pool(
    neg_reviews: List[ReviewRecord],
    bound_rows: Set[int],
    used_rows: Set[int],
    persona: Dict[str, Any],
    valid_asins: Set[str],
) -> List[ReviewRecord]:
    """匹配失败时：从 Persona source_rows 内任取 ≤2★ 差评（不要求关键词命中）。"""
    if bound_rows:
        pool = [r for r in neg_reviews if r.source_row in bound_rows]
    else:
        pool = list(neg_reviews)
    return [
        r for r in pool
        if r.source_row not in used_rows
        and r.asin in valid_asins
        and _persona_quote_relevant(persona, r.content, True)
    ]


def _append_rootcause_quote(
    cause: Dict[str, Any],
    review: ReviewRecord,
    used_rows: Set[int],
    *,
    reused_from_persona: bool = False,
) -> None:
    if not reused_from_persona:
        used_rows.add(review.source_row)
    entry: Dict[str, Any] = {
        "text": _truncate_quote(review.content),
        "asin": review.asin,
        "rating": review.rating,
        "cn_summary": quote_cn_summary(review.content),
    }
    if reused_from_persona:
        entry["reused_from_persona"] = True
    cause["quotes"].append(entry)


def _persona_card_quote_entry(
    persona_quotes: Optional[List[Dict[str, Any]]],
    persona_name: str,
    reviews: List[ReviewRecord],
) -> Optional[ReviewRecord]:
    """画像区已选引用 → ReviewRecord（用于根因空 quotes 降级复用）。"""
    if not persona_quotes:
        return None
    pq = next((q for q in persona_quotes if q.get("persona") == persona_name), None)
    if not pq:
        return None
    quote_items = pq.get("quotes") or []
    if not quote_items and pq.get("quote"):
        quote_items = [pq]
    if not quote_items:
        return None
    first = quote_items[0]
    asin = first.get("asin") or ""
    text = first.get("quote") or ""
    rating = first.get("rating")
    sr = first.get("source_row")
    row_map = {r.source_row: r for r in reviews}
    if sr is not None and sr in row_map:
        return row_map[sr]
    for r in row_map.values():
        if r.asin == asin and text[:80] in r.content:
            return r
    if asin and rating is not None:
        return ReviewRecord(
            asin=asin,
            rating=float(rating),
            title="",
            content=text,
            verified=False,
            vine=False,
            review_date="",
            source_row=int(sr) if sr is not None else -1,
        )
    return None


def enrich_rootcause_quotes(
    rootcauses: List[Dict[str, Any]],
    personas: List[Dict[str, Any]],
    loader: DataLoader,
    reviews: List[ReviewRecord],
    persona_quotes: Optional[List[Dict[str, Any]]] = None,
    extractions: Optional[Dict[int, Dict[str, Any]]] = None,
    *,
    max_quotes: int = 4,
) -> List[Dict[str, Any]]:
    """从真实评论回填根因引用：结构化 Negative feedback → 原文 → source_rows 放宽。"""
    valid_asins = _valid_asins(reviews)
    neg_reviews = [r for r in reviews if r.rating <= 2]
    persona_map = {p.get("name"): p for p in personas}
    ext_map = extractions or {}

    for rc in rootcauses:
        pname = rc.get("persona_name", "")
        persona = persona_map.get(pname, {})
        kws = expand_keywords(persona.get("keywords") or [])
        bound_rows = set(persona.get("source_rows") or [])
        themes = rc.get("affected_themes") or []
        theme_kws = expand_keywords(themes)
        used_rows: Set[int] = set()
        persona_card_review = _persona_card_quote_entry(persona_quotes, pname, reviews)

        for cause in rc.get("root_causes", []):
            cause["quotes"] = []
            cause_kws = expand_keywords(cause.get("quote_keywords") or [])
            search_kws = cause_kws or kws + theme_kws
            pool = neg_reviews if not bound_rows else [r for r in neg_reviews if r.source_row in bound_rows]

            def _filter_candidates(cands: List[ReviewRecord]) -> List[ReviewRecord]:
                return [
                    r for r in cands
                    if _persona_quote_relevant(persona, r.content, True)
                ]

            def _struct_candidates() -> List[ReviewRecord]:
                if not ext_map or not search_kws:
                    return []
                return _filter_candidates([
                    r for r in pool
                    if r.source_row not in used_rows
                    and r.asin in valid_asins
                    and match_rootcause_extraction(search_kws, ext_map.get(r.source_row) or {})
                ])

            candidates = _struct_candidates()
            if not candidates:
                candidates = _filter_candidates([
                    r for r in pool
                    if r.source_row not in used_rows
                    and match_text(search_kws, r.title + " " + r.content)
                    and r.asin in valid_asins
                ])
            if not candidates:
                candidates = _filter_candidates([
                    r for r in pool
                    if r.source_row not in used_rows
                    and match_text(kws + theme_kws, r.title + " " + r.content)
                    and r.asin in valid_asins
                ])
            if not candidates:
                candidates = _rootcause_relaxed_pool(
                    neg_reviews, bound_rows, used_rows, persona, valid_asins,
                )

            struct_set = {r.source_row for r in _struct_candidates()} if ext_map else set()
            candidates.sort(
                key=lambda r: (r.source_row not in struct_set, abs(len(r.content) - 160)),
            )
            for r in candidates[:max_quotes]:
                _append_rootcause_quote(cause, r, used_rows)

            if not cause["quotes"] and persona_card_review is not None:
                _append_rootcause_quote(
                    cause, persona_card_review, used_rows, reused_from_persona=True,
                )

    return rootcauses


def enhanced_market_judgment(stats: MarketStats) -> Tuple[str, str]:
    title, desc = market_competition_judgment(stats.weighted_avg_rating)
    if stats.asins:
        worst = min(stats.asins, key=lambda a: a.avg_rating)
        best = max(stats.asins, key=lambda a: a.avg_rating)
        desc += (
            f" 最差竞品 {worst.asin}（均分 {worst.avg_rating}，差评率 {int(worst.neg_rate * 100)}%）；"
            f"最佳 {best.asin}（均分 {best.avg_rating}，正评率 {int(best.pos_rate * 100)}%）。"
        )
        if stats.weighted_avg_rating < 3.5:
            desc += (
                " 建议对标最佳竞品提炼可复制的 Must-be 能力（刀头/续航/结构），"
                "并在其薄弱维度（如耐用性或性价比感知）建立差异化。"
            )
    return title, desc


def build_theme_asymmetry_insights(
    neg_freq: Dict[str, int],
    pos_freq: Dict[str, int],
    pos_themes: Optional[List[Dict[str, Any]]] = None,
    neg_themes: Optional[List[Dict[str, Any]]] = None,
) -> List[str]:
    """好评/差评不对称洞察（自动配对主题名）。"""
    insights: List[str] = []
    pairs = infer_pos_neg_theme_pairs(pos_themes or [], neg_themes or [])
    for pos_name, neg_name in pairs:
        pc = pos_freq.get(pos_name, 0)
        nc = neg_freq.get(neg_name, 0)
        if nc >= 10 and pc <= nc * 0.15:
            insights.append(
                f"「{neg_name}」差评 {nc} 条，但「{pos_name}」好评仅 {pc} 条——"
                f"用户很少主动表扬该维度，却是显性痛点，应作为产品定义底线。"
            )
    # 通用：任意含「性价比/价格/value/price」的差评 vs 对应好评
    for neg_name, nc in neg_freq.items():
        if nc < 15:
            continue
        if not any(k in neg_name for k in ("性价比", "价格", "price", "value", "贵")):
            continue
        pos_match = next(
            (p for p in pos_freq if any(k in p for k in ("性价比", "价格", "value", "price")) and pos_freq[p] >= 0),
            None,
        )
        if pos_match is not None and pos_freq.get(pos_match, 0) <= nc * 0.1:
            insights.append(
                f"「{neg_name}」差评 {nc} 条，对应好评维度「{pos_match}」仅 {pos_freq.get(pos_match, 0)} 条——"
                "价格/价值感知是购买决策主要障碍，需在成本与感知价值间重新平衡。"
            )
            break
    return insights[:5]


def build_asin_theme_insights(
    stats: MarketStats,
    neg_themes: List[Dict[str, Any]],
    per_asin_neg: Dict[str, Dict[str, int]],
    asin_labels: Dict[str, str],
) -> List[str]:
    """各 ASIN 主题对比的文字结论。"""
    if not stats.asins or not neg_themes:
        return []
    insights = []
    top_themes = sorted(
        [(t["name"], t.get("freq_count", 0)) for t in neg_themes],
        key=lambda x: -x[1],
    )[:3]
    for theme_name, _ in top_themes:
        counts = [(a.asin, per_asin_neg.get(a.asin, {}).get(theme_name, 0)) for a in stats.asins]
        counts = [(a, c) for a, c in counts if c > 0]
        if not counts:
            continue
        counts.sort(key=lambda x: -x[1])
        worst_asin, worst_cnt = counts[0]
        label = asin_labels.get(worst_asin, worst_asin)
        insights.append(f"「{theme_name}」集中出现在 {label}（{worst_cnt} 条），可作为对标突破口。")
    return insights[:4]


def build_product_conclusion(
    stats: MarketStats,
    neg_freq: Dict[str, int],
    pos_freq: Dict[str, int],
    neg_themes: List[Dict[str, Any]],
) -> str:
    """一句话产品定义结论。"""
    top_neg = sorted(neg_freq.items(), key=lambda x: -x[1])[:3]
    top_pos = sorted(pos_freq.items(), key=lambda x: -x[1])[:2]
    must_fix = "、".join(n for n, _ in top_neg) if top_neg else "核心质量缺陷"
    must_have = "、".join(n for n, _ in top_pos) if top_pos else "基础修剪体验"
    if stats.weighted_avg_rating < 3.5:
        return (
            f"新品进入建议：Must-be 优先解决 {must_fix}；"
            f"Performance 对齐 {must_have}；"
            f"Attractive 可在好评高频但差评未覆盖的维度建立差异化感知。"
        )
    return f"改进建议：优先修复 {must_fix}，保留并强化 {must_have} 卖点。"


def build_executive_summary(
    stats: MarketStats,
    neg_freq: Dict[str, int],
    pos_freq: Dict[str, int],
    neg_themes: List[Dict[str, Any]],
    asin_labels: Optional[Dict[str, str]] = None,
    pos_themes: Optional[List[Dict[str, Any]]] = None,
) -> Dict[str, Any]:
    asin_labels = asin_labels or {}
    bullets = [
        f"样本：{stats.total_reviews:,} 条有效评论 · {len(stats.asins)} 个竞品 · 加权均分 {stats.weighted_avg_rating} · 差评率 {int(stats.neg_rate * 100)}%",
    ]
    if stats.asins:
        worst = min(stats.asins, key=lambda a: a.avg_rating)
        best = max(stats.asins, key=lambda a: a.avg_rating)
        wlabel = asin_labels.get(worst.asin, worst.asin)
        blabel = asin_labels.get(best.asin, best.asin)
        bullets.append(
            f"最弱竞品 {wlabel}（{worst.total} 条评论，差评率 {int(worst.neg_rate * 100)}%）— 重点对标其短板"
        )
        bullets.append(
            f"最佳竞品 {blabel}（均分 {best.avg_rating}，正评率 {int(best.pos_rate * 100)}%）— 提炼可复制卖点"
        )
    top_neg = [(n, c) for n, c in sorted(neg_freq.items(), key=lambda x: -x[1]) if c > 0][:3]
    if top_neg:
        bullets.append("核心差评主题：" + "；".join(f"{n}（{c}条）" for n, c in top_neg))
    else:
        bullets.append("差评主题频次待确认（请检查主题关键词匹配）")
    top_pos = sorted(pos_freq.items(), key=lambda x: -x[1])[:3]
    if top_pos:
        bullets.append("核心好评卖点：" + "；".join(f"{n}（{c}条）" for n, c in top_pos))

    conclusion = build_product_conclusion(stats, neg_freq, pos_freq, neg_themes)
    insights = build_theme_asymmetry_insights(
        neg_freq, pos_freq,
        pos_themes=pos_themes or [{"name": n} for n in pos_freq],
        neg_themes=neg_themes,
    )

    opps = []
    ranked_themes = sorted(
        [t for t in neg_themes if neg_freq.get(t["name"], 0) > 0],
        key=lambda t: (neg_freq.get(t["name"], 0), -t.get("freq_rank", 99)),
        reverse=True,
    )
    for t in ranked_themes[:5]:
        cnt = neg_freq.get(t["name"], 0)
        disp = t.get("display_priority", t.get("priority", "P2"))
        opps.append(f"[{disp}] {t['name']}：{cnt} 条差评（占差评比 {round(cnt / max(stats.neg_review_count, 1) * 100)}%）")
    if not opps and top_neg:
        for n, c in top_neg:
            opps.append(f"改进「{n}」— 覆盖 {c} 条差评")

    return {
        "bullets": bullets,
        "opportunities": opps,
        "conclusion": conclusion,
        "insights": insights,
    }


def rank_keywords_for_display(keywords: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """情感词频展示排序：负面 > 正面 > 中性（中性噪声词靠后）。"""
    def _sort_key(kw: Dict[str, Any]) -> Tuple[int, int, str]:
        sent = kw.get("sentiment", "中性")
        word = (kw.get("word") or "").lower()
        sent_order = {"负面": 0, "正面": 1, "中性": 2}.get(sent, 3)
        noise = 1 if word in _NEUTRAL_NOISE else 0
        return (sent_order, noise, word)

    return sorted(keywords, key=_sort_key)


def filter_display_keywords(
    keywords: List[Dict[str, Any]],
    limit: int = 30,
    product_name: str = "",
) -> List[Dict[str, Any]]:
    """过滤中性噪声词，按决策价值排序后取前 N 条。"""
    sw = build_product_stopwords(product_name)
    kept: List[Dict[str, Any]] = []
    for kw in keywords:
        word = (kw.get("word") or "").lower()
        sent = kw.get("sentiment", "中性")
        if word in _NEUTRAL_NOISE and sent == "中性":
            continue
        if word in sw:
            continue
        kept.append(kw)
    return rank_keywords_for_display(kept)[:limit]


# ── 情感关键词（双表：差评词 / 好评词）──

_STRONG_NEG_HINTS = (
    "nick", "bleed", "cut me", "cut my", "pull", "tug", "snag", "burn", "razor burn", "irritat", "bleeding",
)
_ATTRACTIVE_KW_HINTS = ("amazing", "cute", "pretty", "bonus", "comes with", "didn't expect", "gift")
_STRONG_POS_EMOTION_HINTS = ("love", "obsessed", "perfect")
_SCENE_POS_HINTS = ("shower", "travel", "waterproof", "bathroom", "portable")
_STRONG_POS_HINTS = ("smooth", "painless", "gentle")

_POLARITY_PILL = {
    "强负面": "pill-danger",
    "负面": "pill-warn",
    "强正面": "pill-success",
    "强正面情感": "pill-success",
    "魅力型信号": "pill-success",
    "正面场景": "pill-success",
    "正面": "pill-success",
}


def words_display_label(keywords: List[str], max_parts: int = 4) -> str:
    """从主题 keywords 归纳英文词组展示标签（如 cut / nick / bleeding）。"""
    segs: List[str] = []
    for kw in keywords or []:
        for seg in phrase_segments([kw]):
            if len(seg) < 3:
                continue
            if seg not in segs:
                segs.append(seg)
    if not segs:
        return (keywords[0] if keywords else "?")[:48]
    segs.sort(key=lambda s: (len(s) > 12, len(s)))
    return " / ".join(segs[:max_parts])


def build_sentiment_keyword_groups(
    themes: List[Dict[str, Any]],
    reviews: List[Any],
    limit: int = 6,
    is_neg: bool = True,
    id_prefix: str = "g",
) -> List[Dict[str, Any]]:
    """从主题 keywords 构建词组；count 按评论去重（组内任一词命中计 1）。"""
    from echarts_builder import calc_keyword_group_freq

    scored: List[Tuple[int, Dict[str, Any]]] = []
    for theme in themes:
        kws = theme.get("keywords") or []
        count = calc_keyword_group_freq(reviews, kws, is_neg=is_neg)
        if count > 0:
            scored.append((count, theme))
    scored.sort(key=lambda x: x[0], reverse=True)

    groups: List[Dict[str, Any]] = []
    for i, (count, theme) in enumerate(scored[:limit]):
        name = theme.get("name", "")
        kws = theme.get("keywords") or []
        groups.append({
            "id": f"{id_prefix}{i}",
            "words": words_display_label(kws),
            "count": count,
            "theme_name": name,
            "match_keywords": kws,
        })
    return groups


def infer_polarity_label(words: str, is_negative: bool) -> str:
    w = (words or "").lower()
    if is_negative:
        if any(h in w for h in _STRONG_NEG_HINTS):
            return "强负面"
        return "负面"
    if any(h in w for h in _ATTRACTIVE_KW_HINTS):
        return "魅力型信号"
    if any(h in w for h in _STRONG_POS_EMOTION_HINTS):
        return "强正面情感"
    if any(h in w for h in _SCENE_POS_HINTS):
        return "正面场景"
    if any(h in w for h in _STRONG_POS_HINTS):
        return "强正面"
    return "正面"


def _match_personas_for_keywords(
    keywords: List[str],
    personas: List[Dict[str, Any]],
    max_n: int = 2,
) -> List[str]:
    """按 Persona keywords 与词组匹配度推断关联人群。"""
    blob = " ".join(expand_keywords(keywords)).lower()
    if not blob.strip():
        return ["全部群体"]
    scored: List[Tuple[int, str]] = []
    for p in personas:
        name = p.get("name", "")
        if not name:
            continue
        score = 0
        for sig in _persona_text_signals(p):
            if len(sig) >= 3 and sig.lower() in blob:
                score += len(sig)
            elif len(sig) >= 4 and sig.lower() in " ".join(keywords).lower():
                score += 3
        if score > 0:
            scored.append((score, name))
    scored.sort(reverse=True)
    if scored:
        return [n for _, n in scored[:max_n]]
    return ["全部群体"]


def finalize_sentiment_keywords(
    llm_data: Dict[str, Any],
    neg_groups: List[Dict[str, Any]],
    pos_groups: List[Dict[str, Any]],
    personas: List[Dict[str, Any]],
) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
    """合并 LLM 标注与系统统计 count（count 以主题频次为准，LLM 不可改）。"""
    valid_personas = {p.get("name") for p in personas if p.get("name")}

    def _merge(groups: List[Dict[str, Any]], side: List[Dict[str, Any]], is_neg: bool) -> List[Dict[str, Any]]:
        by_id = {item.get("id"): item for item in side if item.get("id")}
        out: List[Dict[str, Any]] = []
        for i, g in enumerate(groups):
            ann = by_id.get(g["id"]) or (side[i] if i < len(side) else {})
            words = (ann.get("words") or g.get("words") or "?").strip()
            polarity = (
                ann.get("polarity_label") or ann.get("polarity") or ""
            ).strip() or infer_polarity_label(words, is_neg)
            meaning = (ann.get("meaning") or "").strip()
            if not meaning:
                meaning = g.get("theme_name", "")
            if len(meaning) > 80:
                meaning = meaning[:79] + "…"
            related = [n for n in (ann.get("related_personas") or []) if n in valid_personas]
            if not related:
                related = _match_personas_for_keywords(g.get("match_keywords") or [], personas)
            out.append({
                "words": words,
                "count": g.get("count", 0),
                "polarity_label": polarity,
                "meaning": meaning,
                "related_personas": related,
                "row_class": "row-r" if polarity == "强负面" else ("row-y" if is_neg else "row-g"),
            })
        return out

    neg_side = llm_data.get("negative") if isinstance(llm_data.get("negative"), list) else []
    pos_side = llm_data.get("positive") if isinstance(llm_data.get("positive"), list) else []
    if not neg_side and isinstance(llm_data.get("keywords"), list):
        neg_side = [k for k in llm_data["keywords"] if k.get("sentiment") == "负面"]
        pos_side = [k for k in llm_data["keywords"] if k.get("sentiment") == "正面"]

    return _merge(neg_groups, neg_side, True), _merge(pos_groups, pos_side, False)


def _keyword_table_rows(items: List[Dict[str, Any]], default_row: str) -> str:
    rows: List[str] = []
    for item in items:
        pill = _POLARITY_PILL.get(item.get("polarity_label", ""), "pill-gray")
        related = "、".join(item.get("related_personas") or [])
        row_cls = item.get("row_class", default_row)
        rows.append(
            f'<tr class="{row_cls}"><td>{item.get("words", "")}</td>'
            f'<td class="num">{item.get("count", 0)}</td>'
            f'<td><span class="pill {pill}">{item.get("polarity_label", "")}</span></td>'
            f'<td>{item.get("meaning", "")}</td>'
            f"<td>{related}</td></tr>"
        )
    return "\n".join(rows)


def build_keyword_tables_html(
    neg_items: List[Dict[str, Any]],
    pos_items: List[Dict[str, Any]],
) -> Tuple[str, str]:
    """返回 (差评表 tbody 行, 好评表 tbody 行)。"""
    return _keyword_table_rows(neg_items, "row-y"), _keyword_table_rows(pos_items, "row-g")


def prepare_keyword_display(
    neg_themes: List[Dict[str, Any]],
    pos_themes: List[Dict[str, Any]],
    reviews: List[Any],
    personas: List[Dict[str, Any]],
    llm_raw: Any,
    limit: int = 6,
) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]], List[Dict[str, Any]], List[Dict[str, Any]]]:
    """构建词组 → 合并 LLM 标注 → 返回 (neg_groups, pos_groups, neg_display, pos_display)。"""
    neg_groups = build_sentiment_keyword_groups(
        neg_themes, reviews, limit=limit, is_neg=True, id_prefix="g",
    )
    pos_groups = build_sentiment_keyword_groups(
        pos_themes, reviews, limit=limit, is_neg=False, id_prefix="g",
    )
    llm_data = llm_raw if isinstance(llm_raw, dict) else {"negative": [], "positive": []}
    neg_disp, pos_disp = finalize_sentiment_keywords(llm_data, neg_groups, pos_groups, personas)
    return neg_groups, pos_groups, neg_disp, pos_disp


def sort_personas_by_evidence(personas: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """按命中评论数降序排列，无命中排后。"""
    return sorted(
        personas,
        key=lambda p: (p.get("hit_count", 0) > 0, p.get("hit_count", 0)),
        reverse=True,
    )


def sort_rootcauses_by_evidence(
    rootcauses: List[Dict[str, Any]],
    personas: List[Dict[str, Any]],
) -> List[Dict[str, Any]]:
    """根因卡片按 Persona 命中数排序。"""
    pmap = {p.get("name"): p.get("hit_count", 0) for p in personas}
    return sorted(rootcauses, key=lambda rc: pmap.get(rc.get("persona_name", ""), 0), reverse=True)


def truncate_cell(text: str, max_len: int = 80) -> str:
    """表格单元格截断，保留完整文本于 title 属性。"""
    text = (text or "").strip()
    if len(text) <= max_len:
        return text
    return text[: max_len - 1] + "…"


def jtbd_cell_html(text: str, max_len: int = 72) -> str:
    """JTBD 表格单元格：截断 + hover 全文。"""
    full = _normalize_jtbd_cell(text)
    if full == "-":
        return "-"
    display = truncate_cell(full, max_len)
    if display == full:
        return full
    esc = full.replace('"', "&quot;")
    return f'<span title="{esc}">{display}</span>'


def asin_link_with_label(asin: str, labels: Dict[str, str]) -> str:
    """ASIN 链接 + 可读标签。"""
    label = labels.get(asin, asin)
    url = f"https://www.amazon.com/dp/{asin}"
    return f'<a href="{url}" target="_blank" rel="noopener" title="{asin}">{label}</a>'


def get_layout_config(cfg: Optional[dict] = None) -> Dict[str, int]:
    """大品类报告布局阈值（通用，不限定品类）。"""
    cfg = cfg or {}
    return {
        "large_asin_threshold": int(cfg.get("large_asin_threshold", 12)),
        "heatmap_asin_threshold": int(cfg.get("heatmap_asin_threshold", 30)),
        "asin_table_top_n": int(cfg.get("asin_table_top_n", 10)),
        "star_summary_top_n": int(cfg.get("star_summary_top_n", 15)),
        "theme_mini_top_n": int(cfg.get("theme_mini_top_n", 8)),
    }


def is_large_market(asin_count: int, threshold: int = 12) -> bool:
    return asin_count > threshold


def sorted_asin_stats(stats: MarketStats, *, by_reviews: bool = True) -> List[ASINStats]:
    if by_reviews:
        return sorted(stats.asins, key=lambda a: (-a.total, -a.avg_rating))
    return sorted(stats.asins, key=lambda a: (-a.avg_rating, -a.total))


def build_asin_labels(stats: MarketStats) -> Dict[str, str]:
    """ASIN → 可读标签（按均分从高到低编号竞品A/B/…/Z/#27）。"""
    sorted_asins = sorted(stats.asins, key=lambda a: (-a.avg_rating, -a.total))
    labels: Dict[str, str] = {}
    for i, a in enumerate(sorted_asins):
        code = chr(ord("A") + i) if i < 26 else str(i + 1)
        labels[a.asin] = f"竞品{code}·{a.avg_rating}分"
    return labels


def build_asin_short_codes(stats: MarketStats) -> Dict[str, str]:
    """ASIN → 图表短码（A/B/…/Z/#27），与 build_asin_labels 同排序。"""
    sorted_asins = sorted(stats.asins, key=lambda a: (-a.avg_rating, -a.total))
    return {
        a.asin: (chr(ord("A") + i) if i < 26 else f"#{i + 1}")
        for i, a in enumerate(sorted_asins)
    }


def _asin_row_class(a: ASINStats) -> Tuple[str, str, str]:
    avg_class = "danger" if a.avg_rating < 3.0 else ("warn" if a.avg_rating < 3.5 else "")
    row_class = "row-r" if a.avg_rating < 3.0 else ("row-y" if a.avg_rating < 3.5 else "row-g")
    avg_display = f"{a.avg_rating}" + (" ⚠" if a.avg_rating < 3.0 else "")
    return avg_class, row_class, avg_display


def format_star_dist_cell(star_dist: Dict[int, int]) -> str:
    parts = [f"{s}★:{star_dist.get(s, 0)}" for s in (5, 4, 3, 2, 1)]
    return " · ".join(parts)


def build_asin_table_row(
    a: ASINStats,
    asin_labels: Dict[str, str],
    *,
    asin_link: str,
    compact: bool = False,
) -> str:
    avg_class, row_class, avg_display = _asin_row_class(a)
    label = asin_labels.get(a.asin, a.asin)
    sd = a.star_dist
    if compact:
        star_cell = f'<td class="num star-dist" title="{format_star_dist_cell(sd)}">详情</td>'
        return (
            f'<tr class="{row_class}"><td>{label}</td><td>{asin_link}</td>'
            f'<td class="num">{a.total}</td>'
            f'<td class="num {avg_class}" style="font-weight:700">{avg_display}</td>'
            f'<td class="num">{int(a.pos_rate * 100)}%</td>'
            f'<td class="num">{int(a.neg_rate * 100)}%</td>{star_cell}</tr>'
        )
    return (
        f'<tr class="{row_class}"><td>{label}</td><td>{asin_link}</td>'
        f'<td class="num">{a.total}</td>'
        f'<td class="num {avg_class}" style="font-weight:700">{avg_display}</td>'
        f'<td class="num">{int(a.pos_rate * 100)}%</td>'
        f'<td class="num">{int(a.neg_rate * 100)}%</td>'
        f'<td class="num">{sd.get(5, 0)}</td><td class="num">{sd.get(4, 0)}</td>'
        f'<td class="num">{sd.get(3, 0)}</td><td class="num">{sd.get(2, 0)}</td>'
        f'<td class="num">{sd.get(1, 0)}</td></tr>'
    )


def build_asin_tables_html(
    stats: MarketStats,
    asin_labels: Dict[str, str],
    asin_link_fn,
    *,
    top_n: int = 10,
    large_threshold: int = 12,
) -> Tuple[str, str, str]:
    """返回 (摘要表 HTML, 附录 HTML, 说明 note)。"""
    n = len(stats.asins)
    if n <= large_threshold:
        thead = (
            "<thead><tr><th>竞品</th><th>ASIN</th><th class=\"num\">有效评论数</th>"
            "<th class=\"num\">均分</th><th class=\"num\">正评率(≥4★)</th><th class=\"num\">差评率(≤2★)</th>"
            "<th class=\"num\">5★</th><th class=\"num\">4★</th><th class=\"num\">3★</th><th class=\"num\">2★</th><th class=\"num\">1★</th></tr></thead>"
        )
        rows = [
            build_asin_table_row(a, asin_labels, asin_link=asin_link_fn(a.asin))
            for a in sorted_asin_stats(stats)
        ]
        table = f"<table>{thead}<tbody>{''.join(rows)}</tbody></table>"
        return table, "", ""

    ranked = sorted_asin_stats(stats)
    top = ranked[:top_n]
    rest = ranked[top_n:]
    compact_head = (
        "<thead><tr><th>竞品</th><th>ASIN</th><th class=\"num\">有效评论数</th>"
        "<th class=\"num\">均分</th><th class=\"num\">正评率(≥4★)</th><th class=\"num\">差评率(≤2★)</th>"
        "<th class=\"num\">星分布</th></tr></thead>"
    )
    summary_rows = [
        build_asin_table_row(a, asin_labels, asin_link=asin_link_fn(a.asin), compact=True)
        for a in top
    ]
    if rest:
        rt = sum(a.total for a in rest)
        ra = round(sum(a.avg_rating * a.total for a in rest) / max(rt, 1), 2)
        summary_rows.append(
            f'<tr class="row-b"><td colspan="2"><strong>其余 {len(rest)} 款合计</strong></td>'
            f'<td class="num">{rt}</td><td class="num">{ra}</td><td class="num">—</td>'
            f'<td class="num">—</td><td class="num">—</td></tr>'
        )
    summary = f"<table>{compact_head}<tbody>{''.join(summary_rows)}</tbody></table>"

    full_head = (
        "<thead><tr><th>竞品</th><th>ASIN</th><th class=\"num\">有效评论数</th>"
        "<th class=\"num\">均分</th><th class=\"num\">正评率(≥4★)</th><th class=\"num\">差评率(≤2★)</th>"
        "<th class=\"num\">5★</th><th class=\"num\">4★</th><th class=\"num\">3★</th><th class=\"num\">2★</th><th class=\"num\">1★</th></tr></thead>"
    )
    appendix_rows = [
        build_asin_table_row(a, asin_labels, asin_link=asin_link_fn(a.asin))
        for a in ranked
    ]
    appendix = (
        f'<details class="asin-appendix"><summary>展开全部 {n} 个 ASIN 明细表</summary>'
        f'<div class="tbl-wrap" style="margin-top:10px"><table>{full_head}'
        f"<tbody>{''.join(appendix_rows)}</tbody></table></div></details>"
    )
    note = (
        f'<p class="note">共 {n} 个竞品：摘要表默认展示评论量 Top {top_n}；'
        f'「星分布」列悬停查看 5★–1★ 明细。</p>'
    )
    return summary, appendix, note


def _need_pills(text: str) -> List[str]:
    text = (text or "").strip()
    if not text or text in ("—", "-", "N/A"):
        return []
    parts = re.split(r"[、,;/]+", text)
    return [p.strip() for p in parts if p.strip()]


def _format_matrix_need(text: str) -> str:
    """矩阵需求列：顿号/逗号统一为分号展示。"""
    text = (text or "").strip()
    if not text or text in ("—", "-", "N/A"):
        return "—"
    return re.sub(r"[、,;/]+", "；", text).strip("；")


def _matrix_sat_class(satisfaction: str) -> str:
    sat = (satisfaction or "中等").strip()
    if sat in ("高", "较高"):
        return "sat-hi"
    if sat in ("低",):
        return "sat-lo"
    return "sat-mid"


def enrich_matrix_scene_evidence(
    matrix: List[Dict[str, Any]],
    personas: List[Dict[str, Any]],
    reviews: List[ReviewRecord],
) -> List[Dict[str, Any]]:
    """为矩阵场景列补充英文原词频次佐证（≥5 条显示 ✓）。"""
    row_map = {r.source_row: r for r in reviews}
    pmap = {p.get("name"): p for p in personas if p.get("name")}
    for row in matrix:
        if row.get("scene_evidence"):
            continue
        persona = pmap.get(row.get("persona"), {})
        src_rows = persona.get("source_rows") or []
        if not src_rows:
            continue
        best_kw, best_cnt = "", 0
        for kw in (persona.get("keywords") or [])[:20]:
            kw = (kw or "").strip()
            if len(kw) < 4 or not re.search(r"[a-zA-Z]", kw):
                continue
            kw_l = kw.lower()
            cnt = sum(
                1 for sr in src_rows
                if sr in row_map
                and kw_l in (row_map[sr].title + " " + row_map[sr].content).lower()
            )
            if cnt > best_cnt:
                best_kw, best_cnt = kw, cnt
        if best_cnt >= 5:
            row["scene_evidence"] = f"{best_kw} {best_cnt}条 ✓"
    return matrix


def build_matrix_table_rows_html(
    matrix: List[Dict[str, Any]],
    persona_map: Dict[str, Dict[str, Any]],
) -> str:
    """人群×场景×需求矩阵表格行（六列：用户群/场景/基本型/期望型/魅力型/满意度）。"""
    if not matrix:
        return '<tr><td colspan="6" class="note">暂无矩阵数据。</td></tr>'
    rows_html: List[str] = []
    for row in matrix:
        pname = row.get("persona", "?")
        p_obj = persona_map.get(pname, {})
        pct = p_obj.get("pct_display", row.get("pct", "?"))
        hit = p_obj.get("hit_count", 0)
        hit_suffix = f"（{hit}条）" if hit else ""
        persona_cell = (
            f'<strong>{pname}</strong>'
            f'<br><small style="color:#888">~{pct}{hit_suffix}</small>'
        )
        scene = (row.get("scene") or "").strip() or "—"
        ev = (row.get("scene_evidence") or "").strip()
        if ev and ev not in scene:
            scene_cell = f'{scene}<br><small style="color:#888">{ev}</small>'
        else:
            scene_cell = scene
        sat = row.get("satisfaction", "中等")
        sat_class = _matrix_sat_class(sat)
        note = (row.get("satisfaction_note") or "").strip()
        sat_cell = sat
        if note:
            sat_cell += f'<br><small style="color:#888;font-weight:400">{note}</small>'
        rows_html.append(
            f'<tr class="row-b">'
            f"<td>{persona_cell}</td>"
            f"<td>{scene_cell}</td>"
            f"<td>{_format_matrix_need(row.get('must_be_needs', ''))}</td>"
            f"<td>{_format_matrix_need(row.get('performance_needs', ''))}</td>"
            f"<td>{_format_matrix_need(row.get('attractive_needs', ''))}</td>"
            f'<td class="{sat_class}">{sat_cell}</td>'
            f"</tr>"
        )
    return "\n".join(rows_html)


def build_matrix_cards_html(
    matrix: List[Dict[str, Any]],
    persona_map: Dict[str, Dict[str, Any]],
) -> str:
    """兼容旧调用：返回完整矩阵表格 HTML。"""
    inner = build_matrix_table_rows_html(matrix, persona_map)
    if matrix:
        return (
            '<div class="tbl-wrap"><table class="matrix-table">'
            "<thead><tr>"
            '<th style="min-width:130px">用户群</th>'
            '<th style="min-width:150px">使用场景（When/Where）'
            '<br><small style="font-weight:normal;color:#aaa">仅评论中佐证≥5条的场景</small></th>'
            '<th style="min-width:150px">基本型需求</th>'
            '<th style="min-width:150px">期望型需求</th>'
            '<th style="min-width:120px">魅力型需求</th>'
            '<th style="min-width:100px">当前满意度</th>'
            "</tr></thead>"
            f"<tbody>{inner}</tbody></table></div>"
        )
    return '<p class="note">暂无矩阵数据。</p>'


def build_footer_asin_links(
    stats: MarketStats,
    asin_labels: Dict[str, str],
    *,
    collapse_threshold: int = 20,
) -> str:
    """Footer ASIN 链接；过多时折叠。"""
    n = len(stats.asins)
    if n <= collapse_threshold:
        return " / ".join(asin_link_with_label(a.asin, asin_labels) for a in stats.asins)
    top = sorted_asin_stats(stats)[:8]
    links = " / ".join(asin_link_with_label(a.asin, asin_labels) for a in top)
    return (
        f'{links} … '
        f'<details class="footer-asin-details" style="display:inline">'
        f"<summary>查看全部 {n} 个 ASIN</summary>"
        f'<div style="margin-top:6px;line-height:1.8">'
        f'{" / ".join(asin_link_with_label(a.asin, asin_labels) for a in sorted_asin_stats(stats))}'
        f"</div></details>"
    )


_KANO_TYPE_CN = {
    "must-be": "基本型",
    "performance": "期望型",
    "attractive": "魅力型",
    "reverse": "反向型",
}

_KANO_TYPE_EN = {
    "must-be": "Must-be",
    "performance": "Performance",
    "attractive": "Attractive",
    "reverse": "Reverse",
}

_ATTRACTIVE_KW = ("love", "amazing", "bonus", "didn't expect", "surprise", "obsessed", "perfect gift")


def _theme_by_name(themes: List[Dict[str, Any]], name: str) -> Dict[str, Any]:
    for t in themes:
        if t.get("name") == name:
            return t
    return {}


def neg_theme_display_priority(
    count: int,
    neg_total: int,
    affected_asins: int,
    asin_count: int,
) -> Tuple[str, str, str]:
    """返回 (显示标签, pill_class, row_class)。"""
    p0_th = neg_total * 0.2
    p1_low = neg_total * 0.1
    p2_low = neg_total * 0.03
    asin_ratio = affected_asins / max(asin_count, 1)
    if count >= p0_th and asin_ratio >= 0.8:
        return "P0", "pill-danger", "row-r"
    if count >= p1_low:
        return "P1", "pill-warn", "row-y"
    if count >= p2_low:
        return "P2", "pill-info", "row-b"
    return "待观察", "pill-gray", "row-b"


def format_neg_theme_cell(name: str, description: str = "") -> str:
    """差评主题列：名称 + 根因拆分说明。"""
    desc = (description or "").strip()
    if desc:
        if not desc.startswith("（"):
            desc = f"（{desc}）"
        return f"{name}<br><small style=\"color:#888\">{desc}</small>"
    return name


def format_theme_asin_scope(
    theme_name: str,
    per_asin: Dict[str, Dict[str, int]],
    asin_labels: Dict[str, str],
    asin_count: int,
    top_n: int = 2,
) -> str:
    """涉及范围 & 最痛 ASIN。"""
    hits = [(asin, per_asin.get(asin, {}).get(theme_name, 0)) for asin in per_asin]
    hits = [(a, c) for a, c in hits if c > 0]
    affected = len(hits)
    if affected == 0:
        return f"0/{asin_count} ASIN"
    hits.sort(key=lambda x: x[1], reverse=True)
    top_parts = [f"{asin_labels.get(a, a)}({c}条)" for a, c in hits[:top_n]]
    return f"{affected}/{asin_count} ASIN · 最痛：{'、'.join(top_parts)}"


def _score_kano_theme_match(theme_name: str, keywords: List[str], kano_item: Dict[str, Any]) -> int:
    blob = " ".join(
        str(kano_item.get(k) or "")
        for k in ("item", "reason", "evidence", "affected_persona")
    ).lower()
    score = 0
    if theme_name and theme_name in blob:
        score += 8
    for seg in expand_keywords(keywords):
        if len(seg) >= 4 and seg.lower() in blob:
            score += len(seg)
    for part in re.findall(r"[\u4e00-\u9fff]{2,}", theme_name):
        if part in blob:
            score += len(part) + 2
    return score


def _kano_hint_text(
    ktype: str,
    pos_name: str,
    paired_neg: str,
    neg_freq: Dict[str, int],
    neg_total: int,
) -> str:
    neg_count = neg_freq.get(paired_neg, 0) if paired_neg else 0
    if ktype == "must-be":
        return "做不好易强烈差评，做好用户少见专门表扬"
    if ktype == "attractive":
        return "好评惊喜提及，缺失一般不引发差评"
    if paired_neg and neg_count > 0:
        pct = round(neg_count / max(neg_total, 1) * 100)
        return f"差评也有「{paired_neg}」对应期望（{neg_count}条/{pct}%）"
    return "做得越好评分越高，是线性竞争点"


def infer_pos_theme_kano_preview(
    pos_theme: Dict[str, Any],
    neg_themes: List[Dict[str, Any]],
    neg_freq: Dict[str, int],
    neg_total: int,
    kano_items: List[Dict[str, Any]],
    paired_neg: str = "",
) -> str:
    """好评主题 KANO 预判列（类型 + 一句理由）。"""
    name = pos_theme.get("name", "")
    keywords = pos_theme.get("keywords") or []
    kw_blob = " ".join(keywords).lower()

    best_item: Optional[Dict[str, Any]] = None
    best_score = 0
    for item in kano_items or []:
        t = item.get("type", "")
        if t == "reverse":
            continue
        sc = _score_kano_theme_match(name, keywords, item)
        if sc > best_score:
            best_score = sc
            best_item = item

    if best_item and best_score >= 3:
        ktype = best_item.get("type", "performance")
        hint = (best_item.get("reason") or "").strip()
        if len(hint) > 48:
            hint = hint[:47] + "…"
        if not hint:
            hint = _kano_hint_text(ktype, name, paired_neg, neg_freq, neg_total)
    else:
        neg_meta = _theme_by_name(neg_themes, paired_neg) if paired_neg else {}
        neg_pri = neg_meta.get("priority", "")
        neg_count = neg_freq.get(paired_neg, 0) if paired_neg else 0
        p0_th = neg_total * 0.2
        if any(k in kw_blob for k in _ATTRACTIVE_KW) and neg_count == 0:
            ktype = "attractive"
        elif paired_neg and neg_count >= p0_th and neg_pri == "P0":
            ktype = "must-be"
        elif paired_neg and neg_count > 0:
            ktype = "performance"
        elif any(k in kw_blob for k in _ATTRACTIVE_KW):
            ktype = "attractive"
        else:
            ktype = "performance"
        hint = _kano_hint_text(ktype, name, paired_neg, neg_freq, neg_total)

    cn = _KANO_TYPE_CN.get(ktype, "期望型")
    en = _KANO_TYPE_EN.get(ktype, "Performance")
    return f"{cn}（{en}）<br><small style=\"color:#888\">（{hint}）</small>"


def build_neg_theme_table_rows(
    neg_freq: Dict[str, int],
    neg_themes: List[Dict[str, Any]],
    neg_total: int,
    asin_count: int,
    per_asin_neg: Dict[str, Dict[str, int]],
    asin_labels: Dict[str, str],
) -> str:
    """差评主题明细表 HTML 行。"""
    theme_map = {t["name"]: t for t in neg_themes if t.get("name")}
    rows: List[str] = []
    for name, count in sorted(neg_freq.items(), key=lambda x: x[1], reverse=True):
        if count <= 0:
            continue
        meta = theme_map.get(name, {})
        pct = f"{round(count / max(neg_total, 1) * 100)}%"
        affected = sum(
            1 for asin_counts in per_asin_neg.values() if asin_counts.get(name, 0) > 0
        )
        label, pill_cls, row_cls = neg_theme_display_priority(
            count, neg_total, affected, asin_count,
        )
        cell_name = format_neg_theme_cell(name, meta.get("description", ""))
        scope = format_theme_asin_scope(name, per_asin_neg, asin_labels, asin_count)
        rows.append(
            f'<tr class="{row_cls}"><td>{cell_name}</td><td class="num">{count}</td>'
            f'<td class="num">{pct}</td><td><span class="pill {pill_cls}">{label}</span></td>'
            f'<td>{scope}</td></tr>'
        )
    return "\n".join(rows)


def build_pos_theme_table_rows(
    pos_freq: Dict[str, int],
    pos_themes: List[Dict[str, Any]],
    pos_total: int,
    neg_themes: List[Dict[str, Any]],
    neg_freq: Dict[str, int],
    neg_total: int,
    kano_items: List[Dict[str, Any]],
) -> str:
    """好评主题明细表 HTML 行（含 KANO 预判）。"""
    theme_map = {t["name"]: t for t in pos_themes if t.get("name")}
    pairs = infer_pos_neg_theme_pairs(pos_themes, neg_themes)
    pair_map = {pos: neg for pos, neg in pairs}
    rows: List[str] = []
    for name, count in sorted(pos_freq.items(), key=lambda x: x[1], reverse=True):
        if count <= 0:
            continue
        meta = theme_map.get(name, {})
        pct = f"{round(count / max(pos_total, 1) * 100)}%"
        kano_cell = infer_pos_theme_kano_preview(
            meta or {"name": name},
            neg_themes,
            neg_freq,
            neg_total,
            kano_items,
            paired_neg=pair_map.get(name, ""),
        )
        rows.append(
            f'<tr class="row-g"><td>{name}</td><td class="num">{count}</td>'
            f'<td class="num">{pct}</td><td>{kano_cell}</td></tr>'
        )
    return "\n".join(rows)


def build_neg_theme_summary_note(
    stats: MarketStats,
    neg_freq: Dict[str, int],
    neg_themes: List[Dict[str, Any]],
    per_asin_neg: Optional[Dict[str, Dict[str, int]]] = None,
) -> str:
    """差评主题区块说明：总数、门槛与 P0/P1/P2 结论。"""
    per_asin_neg = per_asin_neg or {}
    neg_total = stats.neg_review_count
    total = stats.total_reviews
    neg_pct = round(neg_total / max(total, 1) * 100)
    p0_threshold = int(neg_total * 0.2)
    p1_low = int(neg_total * 0.1)
    p2_low = max(1, round(neg_total * 0.03))

    line1 = (
        f"差评总数：{neg_total}条（占有效评论 {neg_pct}%）"
        f" · P0 门槛：{p0_threshold}条（{neg_total}×20%）"
        f" · P1 范围：{p1_low}–{p0_threshold}条"
        f" · P2 范围：{p2_low}–{p1_low}条"
    )

    p0_themes = [
        t for t in neg_themes
        if t.get("priority") == "P0" and neg_freq.get(t.get("name", ""), 0) >= p0_threshold
    ]
    sorted_neg = sorted(neg_freq.items(), key=lambda x: x[1], reverse=True)
    if p0_themes:
        names = "、".join(t["name"] for t in p0_themes[:3])
        line2 = f"本次数据有 {len(p0_themes)} 个 P0 级差评主题：{names}。"
    elif sorted_neg:
        top_name, top_count = sorted_neg[0]
        top_pct = round(top_count / max(neg_total, 1) * 100)
        affected = sum(
            1 for asin_counts in per_asin_neg.values()
            if asin_counts.get(top_name, 0) > 0
        )
        top_label, _, _ = neg_theme_display_priority(
            top_count, neg_total, affected, len(stats.asins),
        )
        line2 = (
            f"本次数据无 P0 级差评主题（无单一问题达到≥20%的差评集中度）。"
            f"最高频为{top_name} {top_count}条（{top_pct}%），属 {top_label} 级别。"
        )
    else:
        line2 = "本次数据无 P0 级差评主题（无单一问题达到≥20%的差评集中度）。"

    return f"{line1}<br>{line2}"


PERSONA_MIN_HIT_COUNT_DEFAULT = 5


def persona_display_meta(
    persona: Dict[str, Any],
    total_reviews: int,
    min_hit_count: int = PERSONA_MIN_HIT_COUNT_DEFAULT,
) -> str:
    """Persona 卡片 meta 行文案。"""
    hit = persona.get("hit_count", 0)
    dim_label = persona.get("dimension_label", persona.get("dimension", "?"))
    dim = persona.get("dimension", "?")
    pct = persona.get("pct_display", "待验证")
    verified = persona.get("pct_verified", False)
    conf = (
        "已验证" if verified and hit >= min_hit_count
        else ("低样本" if verified and hit > 0 else "待验证")
    )
    return (
        f"占比 ~<span class=\"p-pct\">{pct}</span>（命中 {hit} 条 / {total_reviews}）"
        f" · 置信度 {conf} · 维度{dim} {dim_label}"
    )


def normalize_rootcauses(
    rootcauses: List[Dict[str, Any]],
    neg_themes: List[Dict[str, Any]],
) -> List[Dict[str, Any]]:
    """截断过长文本，过滤 affected_themes 为合法差评主题名。"""
    valid_names = [t["name"] for t in neg_themes if t.get("name")]
    valid_set = set(valid_names)
    for rc in rootcauses:
        themes = [t for t in (rc.get("affected_themes") or []) if t in valid_set]
        rc["affected_themes"] = themes or valid_names[:3]
        cleaned_causes = []
        for cause in rc.get("root_causes", [])[:3]:
            cause["title"] = (cause.get("title") or "根因")[:24]
            mech = (cause.get("mechanism") or "").strip()
            cause["mechanism"] = (mech[:150] + "…") if len(mech) > 150 else mech
            dev = (cause.get("dev_direction") or "").strip()
            cause["dev_direction"] = (dev[:100] + "…") if len(dev) > 100 else dev
            cleaned_causes.append(cause)
        rc["root_causes"] = cleaned_causes
    return rootcauses