# -*- 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"), "抱怨产品损坏/不耐用"), (("pull", "snag", "tangle", "rips them out"), "抱怨拉扯毛发"), (("dull", "doesn't shave", "not smooth", "not a close", "does not trim", "barely trim"), "抱怨核心效果差"), (("razor burn", "irritat", "nick", "cut me", "bleeding", "knick"), "抱怨皮肤刺激/割伤"), (("waterproof", "shower", "easy to clean"), "称赞防水/易清洁"), (("love", "perfect", "great", "recommend", "amazing"), "整体满意推荐"), (("travel", "portable", "compact", "lightweight"), "称赞便携"), (("attachment", "head", "guard"), "提及配件/刀头"), ) 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": ("1_audience",), "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 名称 / 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 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 _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 _strip_physio_from_name(name: str, unsupported_zh: List[str]) -> str: out = name for zh in unsupported_zh: out = out.replace(zh, "") out = re.sub(r"\s+", "", out).strip() if len(out) < 2: return "用户群体" if not out.endswith(("用户", "者", "族", "群体")): out += "用户" return out[:6] 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 名称与 core_pain 中的生理标签：绑定簇内须 ≥min_review_count 条评论含对应英文词。""" 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: ref = _normalize_cluster_ref(persona.get("cluster_ref")) entry = catalog_map.get((ref["stage"], ref["label"])) if ref else None cluster = registry.get((ref["stage"], ref["label"])) if ref else None source_rows = list(cluster.source_rows) if cluster else [] 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 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", "") persona["name"] = _strip_physio_from_name(old_name, unsupported) 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)) counts_hint = (entry or {}).get("physio_review_counts") or {} logger.warning( "Persona 生理标签评论佐证不足（需≥%s条），已修正: %s → %s（移除 %s；簇内计数 %s）", min_review_count, old_name, persona["name"], persona["physio_removed_labels"], counts_hint, ) return personas def assign_persona_clusters( personas: List[Dict[str, Any]], all_clusters: Dict[str, List[ClusterData]], ) -> List[Dict[str, Any]]: """为每个 Persona 绑定聚类簇；LLM 已给 cluster_ref 则校验，否则自动映射。""" registry = build_cluster_registry(all_clusters) used_keys: Set[Tuple[str, int]] = set() for persona in personas: normalize_persona_dimension(persona) ref = _normalize_cluster_ref(persona.get("cluster_ref")) if ref and (ref["stage"], ref["label"]) in registry: persona["cluster_ref"] = ref used_keys.add((ref["stage"], ref["label"])) continue dim = persona.get("dimension", "B") 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 key in used_keys and stage == "1_audience": continue candidates.append(c) if not candidates: persona.pop("cluster_ref", None) persona["cluster_unassigned"] = True continue 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 persona["cluster_ref"] = {"stage": best.stage, "label": int(best.cluster_label)} persona.pop("cluster_unassigned", None) if best_score <= 0: persona["cluster_auto_mapped"] = True used_keys.add((best.stage, int(best.cluster_label))) return personas def enrich_persona_keywords( persona: Dict[str, Any], cluster: Optional[ClusterData], ) -> Dict[str, Any]: """从绑定簇 top_phrases 补充 keywords（过滤无信息词）。""" base = [k.lower().strip() for k in (persona.get("keywords") or []) if k and k.strip()] from_cluster: List[str] = [] if cluster: for seg in phrase_segments(cluster.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 cluster: merged = [ s for s in phrase_segments(cluster.top_phrases) if s not in _PERSONA_PHRASE_STOPWORDS and len(s) >= 4 ][:20] persona["keywords"] = merged[:40] return persona def match_persona_text(keywords: List[str], text: str) -> bool: """Persona 簇内二次过滤：比主题 match_text 更严格，避免泛词虚高。""" text = text.lower() expanded = [ k for k in expand_keywords(keywords) if k not in _PERSONA_PHRASE_STOPWORDS and len(k) >= 3 ] if not expanded: return False strong = [k for k in expanded if _is_strong_keyword(k)] weak = [k for k in expanded if k not in strong] strong_hits = sum(1 for k in strong if k in text) weak_hits = sum(1 for k in weak if len(k) >= 5 and k in text) if strong_hits >= 2: return True if strong_hits >= 1 and weak_hits >= 1: return True if not strong and weak_hits >= 2: return True if not strong and weak_hits >= 1 and len(expanded) <= 4: return True return False def resolve_persona_audience_label( persona: Dict[str, Any], all_clusters: Dict[str, List[ClusterData]], ) -> int: """推断 Persona 对应的 audience 簇标签（供根因 per_audience 使用）。""" 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 命中数/占比：以绑定聚类簇 source_rows 为主，簇内 keyword 为辅。""" total = len(reviews) row_map = {r.source_row: r for r in reviews} registry = build_cluster_registry(all_clusters or {}) cluster_ref_counts: Dict[Tuple[str, int], int] = {} for p in personas: ref = _normalize_cluster_ref(p.get("cluster_ref")) if ref: key = (ref["stage"], ref["label"]) cluster_ref_counts[key] = cluster_ref_counts.get(key, 0) + 1 for persona in personas: normalize_persona_dimension(persona) ref = _normalize_cluster_ref(persona.get("cluster_ref")) cluster = registry.get((ref["stage"], ref["label"])) if ref else None enrich_persona_keywords(persona, cluster) if not cluster: persona["source_rows"] = [] persona["hit_count"] = 0 persona["stat_method"] = "unassigned" persona["audience_label"] = -1 _apply_persona_pct_display(persona, total) continue base_rows = set(cluster.source_rows) key = (ref["stage"], ref["label"]) if ref else ("", -1) shared_cluster = cluster_ref_counts.get(key, 0) > 1 dim = persona.get("dimension", "B") if shared_cluster or (dim == "A" and len(base_rows) > 80): kws = persona.get("keywords") or [] filtered = { sr for sr in base_rows if sr in row_map and match_persona_text(kws, row_map[sr].title + " " + row_map[sr].content) } min_keep = max(10, int(len(base_rows) * 0.03)) if filtered and len(filtered) >= min_keep: hit_rows = filtered persona["stat_method"] = "cluster+keywords" else: hit_rows = base_rows persona["stat_method"] = "cluster" else: hit_rows = base_rows persona["stat_method"] = "cluster" persona["source_rows"] = sorted(hit_rows) persona["hit_count"] = len(hit_rows) persona["cluster_review_count"] = len(base_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 "用户反馈摘要" def pick_persona_quotes( personas: List[Dict[str, Any]], reviews: List[ReviewRecord], ) -> List[Dict[str, Any]]: """为每个 Persona 选取代表性引用，优先差评；无匹配时不强行填充。""" used_rows: Set[int] = set() row_map = {r.source_row: r for r in reviews} results = [] for persona in personas: bound_rows = set(persona.get("source_rows") or []) if bound_rows: matched = [row_map[sr] for sr in sorted(bound_rows) if sr in row_map] else: kws = expand_keywords(persona.get("keywords") or []) matched = [r for r in reviews if match_persona_text(kws, r.title + " " + r.content)] neg = [r for r in matched if r.rating <= 2 and r.source_row not in used_rows] pos = [r for r in matched if r.rating >= 4 and r.source_row not in used_rows] pool = neg if neg else pos pool = [r for r in pool if _persona_quote_relevant(persona, r.content, r.rating <= 2)] if not pool: results.append({ "persona": persona.get("name", "?"), "quote": None, "asin": None, "rating": None, "amazon_url": None, "is_neg": False, "cn_summary": None, }) continue pool.sort(key=lambda r: abs(len(r.content) - 180)) r = pool[0] used_rows.add(r.source_row) results.append({ "persona": persona.get("name", "?"), "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), }) 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'
搜索/证据：{evidence}
' f'
结论：{reason}
' ) if competitor != "—": rows += f'
补充：{competitor}
' else: rows = ( f'
频次证据：{evidence}
' f'
影响 Persona：{persona}
' f'
分类原因：{reason}
' f'
竞品现状：{competitor}
' ) return ( f'
' f'
{icon}
' f'
' f'
{item_name}
' f"{rows}" f"
" ) 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'

{meta["empty"]}

' cards.append( f'
' f'
{meta["label"]}
' f'
{meta["title"]}
' f"{body}" f"
" ) return f'
{"".join(cards)}
' 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 or not pq.get("quote"): return None asin = pq.get("asin") or "" text = pq.get("quote") or "" rating = pq.get("rating") row_map = {r.source_row: r for r in reviews} 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=-1, ) return None def enrich_rootcause_quotes( rootcauses: List[Dict[str, Any]], personas: List[Dict[str, Any]], per_aud: Dict[int, Dict[str, List[ClusterData]]], loader: DataLoader, reviews: List[ReviewRecord], persona_quotes: Optional[List[Dict[str, Any]]] = None, ) -> List[Dict[str, Any]]: """从真实评论回填根因引用；匹配失败时放宽至 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} 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) ] 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, ) candidates.sort(key=lambda r: abs(len(r.content) - 160)) for r in candidates[:2]: _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'{item.get("words", "")}' f'{item.get("count", 0)}' f'{item.get("polarity_label", "")}' f'{item.get("meaning", "")}' f"{related}" ) 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('"', """) return f'{display}' 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'{label}' 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'详情' return ( f'{label}{asin_link}' f'{a.total}' f'{avg_display}' f'{int(a.pos_rate * 100)}%' f'{int(a.neg_rate * 100)}%{star_cell}' ) return ( f'{label}{asin_link}' f'{a.total}' f'{avg_display}' f'{int(a.pos_rate * 100)}%' f'{int(a.neg_rate * 100)}%' f'{sd.get(5, 0)}{sd.get(4, 0)}' f'{sd.get(3, 0)}{sd.get(2, 0)}' f'{sd.get(1, 0)}' ) 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 = ( "竞品ASIN有效评论数" "均分正评率(≥4★)差评率(≤2★)" "5★4★3★2★1★" ) rows = [ build_asin_table_row(a, asin_labels, asin_link=asin_link_fn(a.asin)) for a in sorted_asin_stats(stats) ] table = f"{thead}{''.join(rows)}
" return table, "", "" ranked = sorted_asin_stats(stats) top = ranked[:top_n] rest = ranked[top_n:] compact_head = ( "竞品ASIN有效评论数" "均分正评率(≥4★)差评率(≤2★)" "星分布" ) 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'其余 {len(rest)} 款合计' f'{rt}{ra}—' f'——' ) summary = f"{compact_head}{''.join(summary_rows)}
" full_head = ( "竞品ASIN有效评论数" "均分正评率(≥4★)差评率(≤2★)" "5★4★3★2★1★" ) appendix_rows = [ build_asin_table_row(a, asin_labels, asin_link=asin_link_fn(a.asin)) for a in ranked ] appendix = ( f'
展开全部 {n} 个 ASIN 明细表' f'
{full_head}' f"{''.join(appendix_rows)}
" ) note = ( f'

共 {n} 个竞品：摘要表默认展示评论量 Top {top_n}；' f'「星分布」列悬停查看 5★–1★ 明细。

' ) 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 '暂无矩阵数据。' 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'{pname}' f'
~{pct}{hit_suffix}' ) 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}
{ev}' 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'
{note}' rows_html.append( f'' f"{persona_cell}" f"{scene_cell}" f"{_format_matrix_need(row.get('must_be_needs', ''))}" f"{_format_matrix_need(row.get('performance_needs', ''))}" f"{_format_matrix_need(row.get('attractive_needs', ''))}" f'{sat_cell}' f"" ) 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 ( '
' "" '' '' '' '' '' '' "" f"{inner}
用户群使用场景（When/Where）' '
仅评论中佐证≥5条的场景		基本型需求期望型需求魅力型需求当前满意度
" ) return '

暂无矩阵数据。

' 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'
' f"查看全部 {n} 个 ASIN" f'
' f'{" / ".join(asin_link_with_label(a.asin, asin_labels) for a in sorted_asin_stats(stats))}' f"
" ) _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}
{desc}" 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}）
（{hint}）" 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'{cell_name}{count}' f'{pct}{label}' f'{scope}' ) 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'{name}{count}' f'{pct}{kano_cell}' ) 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}
{line2}" def persona_display_meta(persona: Dict[str, Any], total_reviews: int) -> 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 >= 10 else ("低样本" if verified and hit > 0 else "待验证") return ( f"占比 ~{pct}（命中 {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