c91e7f8c20
移除结构化 audience 字段，强化 voc_业务_2 源评论归因匹配与 Persona 引用展示，更新 README 与流水线默认清理 SQLite。 Co-authored-by: Cursor <cursoragent@cursor.com>
750 lines
27 KiB
Python
750 lines
27 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
ECharts 图表构建器：生成内联 ECharts JS 代码和静态库嵌入。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
from pathlib import Path
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
logger = logging.getLogger("voc.echarts")
|
||
|
||
# ECharts CDN（用于下载内联）
|
||
ECHARTS_CDN = "https://cdn.jsdelivr.net/npm/echarts@5.5.0/dist/echarts.min.js"
|
||
CACHE_FILE = Path(__file__).resolve().parent / ".echarts_cache.js"
|
||
|
||
|
||
def get_echarts_script(inline: bool = True) -> str:
|
||
"""返回 ECharts 库的 <script> 标签。inline=True 则内联，否则用 CDN。"""
|
||
if not inline:
|
||
return f'<script src="{ECHARTS_CDN}"></script>'
|
||
|
||
# 内联模式：从缓存或下载
|
||
if CACHE_FILE.is_file():
|
||
js = CACHE_FILE.read_text(encoding="utf-8")
|
||
logger.info("使用缓存的 ECharts (%s KB)", len(js) // 1024)
|
||
return f"<script>\n{js}\n</script>"
|
||
|
||
# 下载
|
||
logger.info("下载 ECharts 库...")
|
||
import urllib.request
|
||
try:
|
||
with urllib.request.urlopen(ECHARTS_CDN, timeout=30) as resp:
|
||
js = resp.read().decode("utf-8")
|
||
CACHE_FILE.write_text(js, encoding="utf-8")
|
||
logger.info("ECharts 已缓存 (%s KB)", len(js) // 1024)
|
||
return f"<script>\n{js}\n</script>"
|
||
except Exception as e:
|
||
logger.warning("下载 ECharts 失败: %s，降级到 CDN", e)
|
||
return f'<script src="{ECHARTS_CDN}"></script>'
|
||
|
||
|
||
# ── 图表 JS 生成 ──
|
||
|
||
ASIN_COLORS_BAR = ["#ef4444", "#f97316", "#eab308", "#8b5cf6", "#3b82f6"]
|
||
ASIN_COLORS_POS = ["#22c55e", "#86efac", "#4ade80", "#2dd4bf", "#60a5fa"]
|
||
|
||
|
||
def build_star_dist_chart(asins: List[str], star_data: Dict[str, Dict[int, int]]) -> str:
|
||
"""评分分布堆叠柱状图。"""
|
||
datasets = []
|
||
colors = ["#22c55e", "#86efac", "#d1d5db", "#fbbf24", "#ef4444"]
|
||
for star in [5, 4, 3, 2, 1]:
|
||
datasets.append({
|
||
"name": f"{star}★",
|
||
"type": "bar",
|
||
"stack": "total",
|
||
"data": [star_data.get(a, {}).get(star, 0) for a in asins],
|
||
"itemStyle": {"color": colors[5 - star]},
|
||
})
|
||
|
||
option = {
|
||
"tooltip": {"trigger": "axis", "axisPointer": {"type": "shadow"}},
|
||
"legend": {"bottom": 0, "textStyle": {"fontSize": 11}},
|
||
"grid": {"left": 50, "right": 20, "top": 20, "bottom": 40},
|
||
"xAxis": {"type": "category", "data": asins, "axisLabel": {"fontSize": 10}},
|
||
"yAxis": {"type": "value"},
|
||
"series": datasets,
|
||
}
|
||
return f"""new Chart(echarts.init(document.getElementById('starDistChart')), {{
|
||
type: 'bar',
|
||
data: {json.dumps(asins)},
|
||
datasets: {json.dumps(datasets, ensure_ascii=False)},
|
||
options: {{
|
||
responsive: true, maintainAspectRatio: false,
|
||
plugins: {{ legend: {{ position: 'bottom', labels: {{ font: {{ size: 11 }} }} }} }},
|
||
scales: {{ x: {{ stacked: true }}, y: {{ stacked: true, beginAtZero: true }} }}
|
||
}}
|
||
}});
|
||
// ECharts 版本:
|
||
(function() {{
|
||
var dom = document.getElementById('starDistChart');
|
||
var chart = echarts.init(dom);
|
||
chart.setOption({json.dumps(option, ensure_ascii=False)});
|
||
window.addEventListener('resize', function() {{ chart.resize(); }});
|
||
}})();"""
|
||
|
||
|
||
def build_horizontal_bar_chart(
|
||
element_id: str,
|
||
labels: List[str],
|
||
data: List[int],
|
||
colors: List[str] | None = None,
|
||
height: int = 360,
|
||
*,
|
||
asin_category_axis: Optional[str] = None,
|
||
) -> str:
|
||
"""水平柱状图。asin_category_axis='y' 时 Y 轴短码 tooltip 显示 ASIN 链接。"""
|
||
if colors is None:
|
||
colors = ["#ef4444"] * len(labels)
|
||
option = {
|
||
"tooltip": {"trigger": "axis", "axisPointer": {"type": "shadow"}},
|
||
"grid": {"left": 150, "right": 40, "top": 10, "bottom": 20},
|
||
"xAxis": {"type": "value"},
|
||
"yAxis": {"type": "category", "data": labels, "inverse": True,
|
||
"axisLabel": {"fontSize": 11}},
|
||
"series": [{
|
||
"type": "bar",
|
||
"data": [{"value": v, "itemStyle": {"color": c}} for v, c in zip(data, colors)],
|
||
"label": {"show": True, "position": "right", "fontSize": 10},
|
||
}],
|
||
}
|
||
return _echarts_init(element_id, option, height=height, asin_category_axis=asin_category_axis)
|
||
|
||
|
||
def build_asin_mini_chart(
|
||
element_id: str,
|
||
labels: List[str],
|
||
data: List[int],
|
||
color: str = "#ef4444",
|
||
) -> str:
|
||
"""单个 ASIN 的迷你柱状图。"""
|
||
option = {
|
||
"tooltip": {"trigger": "axis", "axisPointer": {"type": "shadow"}},
|
||
"grid": {"left": 100, "right": 30, "top": 5, "bottom": 15},
|
||
"xAxis": {"type": "value", "axisLabel": {"fontSize": 9}},
|
||
"yAxis": {"type": "category", "data": labels, "inverse": True,
|
||
"axisLabel": {"fontSize": 9}},
|
||
"series": [{
|
||
"type": "bar",
|
||
"data": [{"value": v, "itemStyle": {"color": color}} for v in data],
|
||
}],
|
||
}
|
||
return f"""(function() {{
|
||
var dom = document.getElementById('{element_id}');
|
||
var chart = echarts.init(dom);
|
||
chart.setOption({json.dumps(option, ensure_ascii=False)});
|
||
window.addEventListener('resize', function() {{ chart.resize(); }});
|
||
}})();"""
|
||
|
||
|
||
def build_grouped_bar_chart(
|
||
element_id: str,
|
||
labels: List[str],
|
||
datasets: List[Dict[str, Any]],
|
||
) -> str:
|
||
"""分组柱状图（多 ASIN 对比）。"""
|
||
series = [
|
||
{
|
||
"name": ds["label"],
|
||
"type": "bar",
|
||
"data": ds["data"],
|
||
"itemStyle": {"color": ds.get("color", "#3b82f6")},
|
||
}
|
||
for ds in datasets
|
||
]
|
||
option = {
|
||
"tooltip": {"trigger": "axis", "axisPointer": {"type": "shadow"}},
|
||
"legend": {"bottom": 0, "textStyle": {"fontSize": 10}},
|
||
"grid": {"left": 50, "right": 20, "top": 20, "bottom": 40},
|
||
"xAxis": {"type": "category", "data": labels, "axisLabel": {"fontSize": 10, "rotate": 20}},
|
||
"yAxis": {"type": "value"},
|
||
"series": series,
|
||
}
|
||
return f"""(function() {{
|
||
var dom = document.getElementById('{element_id}');
|
||
var chart = echarts.init(dom);
|
||
chart.setOption({json.dumps(option, ensure_ascii=False)});
|
||
window.addEventListener('resize', function() {{ chart.resize(); }});
|
||
}})();"""
|
||
|
||
|
||
# ── 布局与图表类型阈值 ──
|
||
|
||
THEME_COLORS_NEG = ["#ef4444", "#f97316", "#eab308", "#dc2626", "#b91c1c", "#fb923c"]
|
||
THEME_COLORS_POS = ["#22c55e", "#86efac", "#4ade80", "#2dd4bf", "#60a5fa", "#16a34a"]
|
||
|
||
|
||
def get_chart_layout(stats: Any, cfg: Optional[dict] = None) -> Dict[str, Any]:
|
||
"""根据 ASIN 数量计算图表布局参数。"""
|
||
cfg = cfg or {}
|
||
n = len(getattr(stats, "asins", []) or [])
|
||
large_th = int(cfg.get("large_asin_threshold", 12))
|
||
heatmap_th = int(cfg.get("heatmap_asin_threshold", 30))
|
||
large = n > large_th
|
||
heatmap = n > heatmap_th
|
||
return {
|
||
"asin_count": n,
|
||
"large_market": large,
|
||
"use_heatmap": heatmap,
|
||
"star_chart_height": max(320, n * 22) if large else 300,
|
||
"star_summary_height": 280,
|
||
"show_star_summary": large and n > 15,
|
||
"neg_theme_height": 480 if heatmap else (420 if large else 360),
|
||
"pos_theme_height": 480 if heatmap else (420 if large else 360),
|
||
}
|
||
|
||
|
||
# ── ASIN 短码 tooltip（悬浮显示代称 + 可点击 ASIN）──
|
||
|
||
def build_asin_tooltip_map(asin_short_codes: Dict[str, str]) -> Dict[str, Dict[str, str]]:
|
||
"""ASIN 短码 A → {asin, url}。"""
|
||
from data_loader import build_amazon_url
|
||
return {
|
||
short: {"asin": asin, "url": build_amazon_url(asin)}
|
||
for asin, short in asin_short_codes.items()
|
||
}
|
||
|
||
|
||
def build_asin_tooltip_bootstrap(asin_short_codes: Dict[str, str]) -> str:
|
||
"""注入全局 VOC_ASIN_MAP 与 axis tooltip formatter。"""
|
||
map_json = json.dumps(build_asin_tooltip_map(asin_short_codes), ensure_ascii=False)
|
||
return f"""// ASIN 短码 tooltip（悬浮：A · B0XXX 可点击跳转亚马逊）
|
||
window.VOC_ASIN_MAP = {map_json};
|
||
window.vocAsinTooltipHeader = function(code) {{
|
||
if (!code || String(code).indexOf('其余') === 0) return code;
|
||
var info = (window.VOC_ASIN_MAP || {{}})[code];
|
||
if (info && info.asin) {{
|
||
return code + ' · <a href="' + info.url + '" target="_blank" rel="noopener" style="color:#2563eb;text-decoration:underline">' + info.asin + '</a>';
|
||
}}
|
||
return code;
|
||
}};
|
||
window.vocApplyAsinAxisTooltip = function(opt, axisKey) {{
|
||
axisKey = axisKey || 'x';
|
||
var axis = axisKey === 'y' ? opt.yAxis : opt.xAxis;
|
||
if (!axis || axis.type !== 'category') return;
|
||
opt.tooltip = opt.tooltip || {{}};
|
||
opt.tooltip.trigger = 'axis';
|
||
opt.tooltip.enterable = true;
|
||
opt.tooltip.confine = true;
|
||
opt.tooltip.axisPointer = opt.tooltip.axisPointer || {{type: 'shadow'}};
|
||
opt.tooltip.formatter = function(params) {{
|
||
var items = Array.isArray(params) ? params : [params];
|
||
if (!items.length) return '';
|
||
var code = items[0].axisValue != null ? items[0].axisValue : (items[0].name || '');
|
||
var lines = [window.vocAsinTooltipHeader(code)];
|
||
items.forEach(function(p) {{
|
||
if (p.seriesName == null || p.value == null) return;
|
||
var v = p.value;
|
||
if (Array.isArray(v)) v = v[v.length - 1];
|
||
if (v && typeof v === 'object' && v.value != null) v = v.value;
|
||
lines.push((p.marker || '') + p.seriesName + ': ' + v);
|
||
}});
|
||
return lines.join('<br/>');
|
||
}};
|
||
}};"""
|
||
|
||
|
||
def _echarts_init(
|
||
element_id: str,
|
||
option: dict,
|
||
height: Optional[int] = None,
|
||
*,
|
||
asin_category_axis: Optional[str] = None,
|
||
) -> str:
|
||
h_js = f"dom.style.height='{height}px';" if height else ""
|
||
apply_js = ""
|
||
if asin_category_axis:
|
||
apply_js = f"if (window.vocApplyAsinAxisTooltip) window.vocApplyAsinAxisTooltip(opt, '{asin_category_axis}');"
|
||
return f"""(function() {{
|
||
var dom = document.getElementById('{element_id}');
|
||
if (!dom) return;
|
||
{h_js}
|
||
var opt = {json.dumps(option, ensure_ascii=False)};
|
||
{apply_js}
|
||
var chart = echarts.init(dom);
|
||
chart.setOption(opt);
|
||
window.addEventListener('resize', function() {{ chart.resize(); }});
|
||
}})();"""
|
||
|
||
|
||
def _aggregate_star_dist(asin_list: List[str], star_data: Dict[str, Dict[int, int]]) -> Dict[int, int]:
|
||
agg: Dict[int, int] = {s: 0 for s in (5, 4, 3, 2, 1)}
|
||
for asin in asin_list:
|
||
sd = star_data.get(asin, {})
|
||
for s in agg:
|
||
agg[s] += sd.get(s, 0)
|
||
return agg
|
||
|
||
|
||
def build_overall_charts(
|
||
neg_freq: dict,
|
||
pos_freq: dict,
|
||
neg_themes: list,
|
||
) -> str:
|
||
"""生成主题频次画像区的全市场图表（negChartOverall / posChartOverall）。"""
|
||
priority_color = {"P0": "#ef4444", "P1": "#f97316", "P2": "#eab308"}
|
||
theme_priority = {t["name"]: t.get("priority", "P2") for t in neg_themes}
|
||
|
||
neg_sorted = sorted(neg_freq.items(), key=lambda x: x[1], reverse=True)
|
||
pos_sorted = sorted(pos_freq.items(), key=lambda x: x[1], reverse=True)
|
||
neg_labels = [n for n, _ in neg_sorted]
|
||
neg_data = [c for _, c in neg_sorted]
|
||
pos_labels = [n for n, _ in pos_sorted]
|
||
pos_data = [c for _, c in pos_sorted]
|
||
neg_colors = [priority_color.get(theme_priority.get(n, "P2"), "#94a3b8") for n in neg_labels]
|
||
|
||
parts = []
|
||
parts.append("// 整体主题频次 - 差评")
|
||
parts.append(build_horizontal_bar_chart("negChartOverall", neg_labels, neg_data, neg_colors))
|
||
parts.append("// 整体主题频次 - 好评")
|
||
parts.append(build_horizontal_bar_chart("posChartOverall", pos_labels, pos_data, ["#22c55e"] * len(pos_labels)))
|
||
return "\n\n".join(parts)
|
||
|
||
|
||
def build_theme_mini_charts_html(
|
||
prefix: str,
|
||
theme_labels: List[str],
|
||
asin_count: int,
|
||
*,
|
||
large_threshold: int = 30,
|
||
) -> str:
|
||
"""大品类：每主题 Top N ASIN 小 multiples 的 HTML 容器。"""
|
||
if not theme_labels or asin_count <= large_threshold:
|
||
return ""
|
||
blocks = ['<div class="theme-mini-grid">']
|
||
for i, theme in enumerate(theme_labels):
|
||
blocks.append(
|
||
f'<div class="theme-mini-box"><h4>{theme}</h4>'
|
||
f'<div id="{prefix}Mini{i}" class="theme-mini-chart"></div></div>'
|
||
)
|
||
blocks.append("</div>")
|
||
return "".join(blocks)
|
||
|
||
|
||
def build_all_charts(
|
||
stats: Any,
|
||
neg_themes: List[Dict[str, Any]],
|
||
pos_themes: List[Dict[str, Any]],
|
||
neg_freq: Dict[str, int],
|
||
pos_freq: Dict[str, int],
|
||
per_asin_neg: Dict[str, Dict[str, int]],
|
||
per_asin_pos: Dict[str, Dict[str, int]],
|
||
asin_labels: Optional[Dict[str, str]] = None,
|
||
asin_short_codes: Optional[Dict[str, str]] = None,
|
||
layout: Optional[Dict[str, Any]] = None,
|
||
cfg: Optional[dict] = None,
|
||
) -> str:
|
||
"""生成所有图表的 JS 代码。"""
|
||
asin_labels = asin_labels or {}
|
||
asin_short_codes = asin_short_codes or asin_labels
|
||
layout = layout or get_chart_layout(stats, cfg)
|
||
parts: List[str] = [build_asin_tooltip_bootstrap(asin_short_codes)]
|
||
|
||
asins = [a.asin for a in stats.asins]
|
||
if layout["large_market"]:
|
||
from report_utils import sorted_asin_stats
|
||
asins = [a.asin for a in sorted_asin_stats(stats)]
|
||
else:
|
||
asins = [a.asin for a in sorted(stats.asins, key=lambda a: (-a.avg_rating, -a.total))]
|
||
axis_labels = [asin_short_codes.get(a, a) for a in asins]
|
||
star_data = {a.asin: a.star_dist for a in stats.asins}
|
||
|
||
parts.append("// 1. 评分分布堆叠图")
|
||
parts.append(_build_star_dist_echarts(
|
||
axis_labels, star_data, asins,
|
||
horizontal=layout["large_market"],
|
||
))
|
||
if layout.get("show_star_summary"):
|
||
from report_utils import sorted_asin_stats
|
||
ranked = [a.asin for a in sorted_asin_stats(stats)]
|
||
top_n = int((cfg or {}).get("star_summary_top_n", 15))
|
||
top_asins = ranked[:top_n]
|
||
rest_asins = ranked[top_n:]
|
||
summary_labels = [asin_short_codes.get(a, a) for a in top_asins]
|
||
summary_data = {a: star_data[a] for a in top_asins}
|
||
summary_asins = list(top_asins)
|
||
if rest_asins:
|
||
summary_labels.append(f"其余{len(rest_asins)}款")
|
||
summary_data["__other__"] = _aggregate_star_dist(rest_asins, star_data)
|
||
summary_asins.append("__other__")
|
||
parts.append("// 1b. 评分分布摘要")
|
||
parts.append(_build_star_dist_echarts(
|
||
summary_labels, summary_data, summary_asins,
|
||
element_id="starDistSummaryChart", horizontal=False,
|
||
))
|
||
|
||
neg_sorted = sorted(neg_freq.items(), key=lambda x: x[1], reverse=True)
|
||
pos_sorted = sorted(pos_freq.items(), key=lambda x: x[1], reverse=True)
|
||
neg_labels = [n for n, _ in neg_sorted]
|
||
neg_data = [c for _, c in neg_sorted]
|
||
pos_labels = [n for n, _ in pos_sorted]
|
||
pos_data = [c for _, c in pos_sorted]
|
||
|
||
priority_color = {"P0": "#ef4444", "P1": "#f97316", "P2": "#eab308"}
|
||
theme_priority = {t["name"]: t.get("priority", "P2") for t in neg_themes}
|
||
neg_colors = [priority_color.get(theme_priority.get(n, "P2"), "#94a3b8") for n in neg_labels]
|
||
|
||
parts.append("// 2. 差评主题")
|
||
parts.append(build_horizontal_bar_chart("negChart", neg_labels, neg_data, neg_colors))
|
||
parts.append("// 3. 好评主题")
|
||
parts.append(build_horizontal_bar_chart("posChart", pos_labels, pos_data, ["#22c55e"] * len(pos_labels)))
|
||
|
||
neg_top6 = neg_labels[:6]
|
||
pos_top6 = pos_labels[:6]
|
||
mini_top = int((cfg or {}).get("theme_mini_top_n", 8))
|
||
|
||
parts.append("// 4. 各 ASIN 差评主题对比")
|
||
parts.append(_build_per_asin_theme_chart(
|
||
"negChartPerAsin", neg_top6, per_asin_neg, asins,
|
||
axis_labels=axis_labels, asin_full_labels=asin_labels,
|
||
is_neg=True, use_heatmap=layout["use_heatmap"],
|
||
))
|
||
if layout["use_heatmap"]:
|
||
parts.append(_build_theme_mini_charts_js(
|
||
"neg", neg_top6, per_asin_neg, asins, asin_short_codes, asin_labels,
|
||
is_neg=True, top_n=mini_top,
|
||
))
|
||
|
||
parts.append("// 5. 各 ASIN 好评主题对比")
|
||
parts.append(_build_per_asin_theme_chart(
|
||
"posChartPerAsin", pos_top6, per_asin_pos, asins,
|
||
axis_labels=axis_labels, asin_full_labels=asin_labels,
|
||
is_neg=False, use_heatmap=layout["use_heatmap"],
|
||
))
|
||
if layout["use_heatmap"]:
|
||
parts.append(_build_theme_mini_charts_js(
|
||
"pos", pos_top6, per_asin_pos, asins, asin_short_codes, asin_labels,
|
||
is_neg=False, top_n=mini_top,
|
||
))
|
||
|
||
return "\n\n".join(parts)
|
||
|
||
|
||
def _build_per_asin_theme_chart(
|
||
element_id: str,
|
||
theme_labels: List[str],
|
||
per_asin_data: Dict[str, Dict[str, int]],
|
||
asins: List[str],
|
||
*,
|
||
axis_labels: Optional[List[str]] = None,
|
||
asin_full_labels: Optional[Dict[str, str]] = None,
|
||
is_neg: bool = True,
|
||
use_heatmap: bool = False,
|
||
asin_labels: Optional[Dict[str, str]] = None,
|
||
) -> str:
|
||
"""各 ASIN 主题对比：分组柱（X=ASIN，series=主题）或热力图。"""
|
||
if not theme_labels or not asins:
|
||
return f"// skip {element_id}: no data"
|
||
if asin_labels and not axis_labels:
|
||
axis_labels = [asin_labels.get(a, a) for a in asins]
|
||
axis_labels = axis_labels or asins
|
||
asin_full_labels = asin_full_labels or asin_labels or {}
|
||
|
||
if use_heatmap:
|
||
return _build_theme_heatmap(
|
||
element_id, theme_labels, per_asin_data, asins,
|
||
axis_labels, asin_full_labels, is_neg=is_neg,
|
||
)
|
||
|
||
colors = THEME_COLORS_NEG if is_neg else THEME_COLORS_POS
|
||
series = []
|
||
for i, theme in enumerate(theme_labels):
|
||
series.append({
|
||
"name": theme,
|
||
"type": "bar",
|
||
"data": [per_asin_data.get(asin, {}).get(theme, 0) for asin in asins],
|
||
"itemStyle": {"color": colors[i % len(colors)]},
|
||
})
|
||
option: Dict[str, Any] = {
|
||
"tooltip": {"trigger": "axis", "axisPointer": {"type": "shadow"}},
|
||
"legend": {"top": 0, "type": "scroll", "textStyle": {"fontSize": 10}},
|
||
"grid": {"left": 48, "right": 16, "top": 48, "bottom": 72},
|
||
"xAxis": {
|
||
"type": "category",
|
||
"data": axis_labels,
|
||
"axisLabel": {"fontSize": 9, "rotate": 45, "interval": 0},
|
||
},
|
||
"yAxis": {"type": "value"},
|
||
"series": series,
|
||
}
|
||
if len(asins) > 12:
|
||
end_pct = min(100, round(12 / len(asins) * 100))
|
||
option["dataZoom"] = [
|
||
{"type": "inside", "start": 0, "end": end_pct},
|
||
{"type": "slider", "start": 0, "end": end_pct, "bottom": 8, "height": 18},
|
||
]
|
||
option["grid"]["bottom"] = 96
|
||
return _echarts_init(element_id, option, asin_category_axis="x")
|
||
|
||
|
||
def _build_theme_heatmap(
|
||
element_id: str,
|
||
theme_labels: List[str],
|
||
per_asin_data: Dict[str, Dict[str, int]],
|
||
asins: List[str],
|
||
axis_labels: List[str],
|
||
asin_full_labels: Dict[str, str],
|
||
*,
|
||
is_neg: bool = True,
|
||
) -> str:
|
||
data = []
|
||
max_val = 1
|
||
for yi, theme in enumerate(theme_labels):
|
||
for xi, asin in enumerate(asins):
|
||
v = per_asin_data.get(asin, {}).get(theme, 0)
|
||
max_val = max(max_val, v)
|
||
data.append([xi, yi, v])
|
||
colors = ["#fef2f2", "#fca5a5", "#ef4444", "#b91c1c"] if is_neg else ["#f0fdf4", "#86efac", "#22c55e", "#15803d"]
|
||
end_pct = min(100, max(15, round(20 / max(len(asins), 1) * 100)))
|
||
option = {
|
||
"grid": {"left": 88, "right": 56, "top": 24, "bottom": 72},
|
||
"xAxis": {
|
||
"type": "category",
|
||
"data": axis_labels,
|
||
"splitArea": {"show": True},
|
||
"axisLabel": {"fontSize": 9, "rotate": 45, "interval": 0},
|
||
},
|
||
"yAxis": {
|
||
"type": "category",
|
||
"data": theme_labels,
|
||
"splitArea": {"show": True},
|
||
"axisLabel": {"fontSize": 10},
|
||
},
|
||
"visualMap": {
|
||
"min": 0,
|
||
"max": max_val,
|
||
"calculable": True,
|
||
"orient": "vertical",
|
||
"right": 8,
|
||
"top": "center",
|
||
"inRange": {"color": colors},
|
||
"text": ["高", "低"],
|
||
},
|
||
"dataZoom": [
|
||
{"type": "inside", "xAxisIndex": 0, "start": 0, "end": end_pct},
|
||
{"type": "slider", "xAxisIndex": 0, "bottom": 8, "height": 18, "start": 0, "end": end_pct},
|
||
],
|
||
"series": [{
|
||
"name": "命中数",
|
||
"type": "heatmap",
|
||
"data": data,
|
||
"label": {"show": False},
|
||
"emphasis": {"itemStyle": {"shadowBlur": 6, "shadowColor": "rgba(0,0,0,0.2)"}},
|
||
}],
|
||
}
|
||
full_map = {axis_labels[i]: asin_full_labels.get(asins[i], asins[i]) for i in range(len(asins))}
|
||
js_option = json.dumps(option, ensure_ascii=False)
|
||
return f"""(function() {{
|
||
var dom = document.getElementById('{element_id}');
|
||
if (!dom) return;
|
||
var opt = {js_option};
|
||
var fullMap = {json.dumps(full_map, ensure_ascii=False)};
|
||
opt.tooltip = {{
|
||
position: 'top',
|
||
enterable: true,
|
||
confine: true,
|
||
formatter: function(p) {{
|
||
if (!p.data) return '';
|
||
var code = (opt.xAxis.data[p.data[0]] || '');
|
||
var theme = (opt.yAxis.data[p.data[1]] || '');
|
||
var head = window.vocAsinTooltipHeader ? window.vocAsinTooltipHeader(code) : (fullMap[code] || code);
|
||
return head + '<br/>' + theme + ': ' + p.data[2];
|
||
}}
|
||
}};
|
||
var chart = echarts.init(dom);
|
||
chart.setOption(opt);
|
||
window.addEventListener('resize', function() {{ chart.resize(); }});
|
||
}})();"""
|
||
|
||
|
||
def _build_theme_mini_charts_js(
|
||
prefix: str,
|
||
theme_labels: List[str],
|
||
per_asin_data: Dict[str, Dict[str, int]],
|
||
asins: List[str],
|
||
asin_short: Dict[str, str],
|
||
asin_full: Dict[str, str],
|
||
*,
|
||
is_neg: bool = True,
|
||
top_n: int = 8,
|
||
) -> str:
|
||
_ = asin_full
|
||
parts = []
|
||
color = "#ef4444" if is_neg else "#22c55e"
|
||
for i, theme in enumerate(theme_labels):
|
||
ranked = sorted(
|
||
((asin, per_asin_data.get(asin, {}).get(theme, 0)) for asin in asins),
|
||
key=lambda x: (-x[1], x[0]),
|
||
)
|
||
top = [(a, c) for a, c in ranked if c > 0][:top_n]
|
||
if not top:
|
||
top = ranked[: min(top_n, len(ranked))]
|
||
labels = [asin_short.get(a, a) for a, _ in top]
|
||
values = [c for _, c in top]
|
||
parts.append(build_horizontal_bar_chart(
|
||
f"{prefix}Mini{i}", labels, values, [color] * len(labels),
|
||
height=200, asin_category_axis="y",
|
||
))
|
||
return "\n\n".join(parts)
|
||
|
||
|
||
def _build_star_dist_echarts(
|
||
chart_labels: List[str],
|
||
star_data: Dict[str, Dict[int, int]],
|
||
asins: List[str],
|
||
*,
|
||
element_id: str = "starDistChart",
|
||
horizontal: bool = False,
|
||
) -> str:
|
||
"""评分分布堆叠图；大品类用横向堆叠。"""
|
||
colors_5 = ["#22c55e", "#86efac", "#d1d5db", "#fbbf24", "#ef4444"]
|
||
series = []
|
||
for star in [5, 4, 3, 2, 1]:
|
||
series.append({
|
||
"name": f"{star}★",
|
||
"type": "bar",
|
||
"stack": "total",
|
||
"data": [star_data.get(a, {}).get(star, 0) for a in asins],
|
||
"itemStyle": {"color": colors_5[5 - star]},
|
||
})
|
||
if horizontal:
|
||
option = {
|
||
"tooltip": {"trigger": "axis", "axisPointer": {"type": "shadow"}},
|
||
"legend": {"top": 0, "textStyle": {"fontSize": 11}},
|
||
"grid": {"left": 72, "right": 24, "top": 36, "bottom": 24},
|
||
"xAxis": {"type": "value"},
|
||
"yAxis": {
|
||
"type": "category",
|
||
"data": chart_labels,
|
||
"inverse": True,
|
||
"axisLabel": {"fontSize": 10, "width": 64, "overflow": "truncate"},
|
||
},
|
||
"series": series,
|
||
}
|
||
else:
|
||
option = {
|
||
"tooltip": {"trigger": "axis", "axisPointer": {"type": "shadow"}},
|
||
"legend": {"bottom": 0, "textStyle": {"fontSize": 11}},
|
||
"grid": {"left": 50, "right": 20, "top": 20, "bottom": 40},
|
||
"xAxis": {
|
||
"type": "category",
|
||
"data": chart_labels,
|
||
"axisLabel": {"fontSize": 10, "rotate": 30 if len(chart_labels) > 8 else 0},
|
||
},
|
||
"yAxis": {"type": "value"},
|
||
"series": series,
|
||
}
|
||
axis_key = "y" if horizontal else "x"
|
||
return _echarts_init(element_id, option, asin_category_axis=axis_key)
|
||
|
||
|
||
# ── 图表数据计算辅助 ──
|
||
|
||
_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",
|
||
)
|
||
|
||
|
||
def _is_strong_keyword(kw: str) -> bool:
|
||
if len(kw) >= 12:
|
||
return True
|
||
return any(h in kw for h in _STRONG_HINTS)
|
||
|
||
|
||
def _match_theme_in_text(keywords: List[str], text: str) -> bool:
|
||
from report_utils import match_text
|
||
return match_text(keywords, text)
|
||
|
||
|
||
def calc_keyword_group_freq(
|
||
reviews: List[Any],
|
||
keywords: List[str],
|
||
is_neg: bool = True,
|
||
) -> int:
|
||
"""词组频次：组内任一词命中即计 1，同一评论只计 1 次；按星级过滤。"""
|
||
kws = [kw.lower().strip() for kw in keywords if kw and kw.strip()]
|
||
if not kws:
|
||
return 0
|
||
count = 0
|
||
for r in reviews:
|
||
if is_neg and r.rating > 2:
|
||
continue
|
||
if not is_neg and r.rating < 4:
|
||
continue
|
||
text = (r.title + " " + r.content).lower()
|
||
if _match_theme_in_text(kws, text):
|
||
count += 1
|
||
return count
|
||
|
||
|
||
def _theme_matches_review(
|
||
keywords: List[str],
|
||
review: Any,
|
||
extractions: Dict[int, Any],
|
||
*,
|
||
is_neg: bool,
|
||
) -> bool:
|
||
"""主题命中：优先结构化 category+aspect，回退原文 keyword。"""
|
||
from report_utils import match_theme_extraction
|
||
|
||
kws = [kw.lower().strip() for kw in keywords if kw and kw.strip()]
|
||
if not kws:
|
||
return False
|
||
ext = extractions.get(review.source_row)
|
||
if ext and match_theme_extraction(kws, ext, is_neg=is_neg):
|
||
return True
|
||
text = (review.title + " " + review.content).lower()
|
||
return _match_theme_in_text(kws, text)
|
||
|
||
|
||
def calc_theme_freq(
|
||
themes: List[Dict[str, Any]],
|
||
reviews: List[Any],
|
||
is_neg: bool = True,
|
||
extractions: Dict[int, Any] | None = None,
|
||
) -> Dict[str, int]:
|
||
"""用主题 keywords 统计频次（结构化 category+aspect 优先，原文回退）。"""
|
||
ext_map = extractions or {}
|
||
freq: Dict[str, int] = {}
|
||
for theme in themes:
|
||
keywords = [kw.lower().strip() for kw in theme.get("keywords", []) if kw.strip()]
|
||
count = 0
|
||
for r in reviews:
|
||
if is_neg and r.rating > 2:
|
||
continue
|
||
if not is_neg and r.rating < 4:
|
||
continue
|
||
if _theme_matches_review(keywords, r, ext_map, is_neg=is_neg):
|
||
count += 1
|
||
freq[theme["name"]] = count
|
||
return freq
|
||
|
||
|
||
def calc_per_asin_theme_freq(
|
||
themes: List[Dict[str, Any]],
|
||
reviews: List[Any],
|
||
is_neg: bool = True,
|
||
extractions: Dict[int, Any] | None = None,
|
||
) -> Dict[str, Dict[str, int]]:
|
||
"""按 ASIN 分别统计各主题频次（结构化优先）。"""
|
||
from collections import defaultdict
|
||
|
||
ext_map = extractions or {}
|
||
result: Dict[str, Dict[str, int]] = defaultdict(lambda: defaultdict(int))
|
||
for theme in themes:
|
||
keywords = [kw.lower().strip() for kw in theme.get("keywords", []) if kw.strip()]
|
||
name = theme["name"]
|
||
for r in reviews:
|
||
if is_neg and r.rating > 2:
|
||
continue
|
||
if not is_neg and r.rating < 4:
|
||
continue
|
||
if _theme_matches_review(keywords, r, ext_map, is_neg=is_neg):
|
||
result[r.asin][name] += 1
|
||
return {asin: dict(counts) for asin, counts in result.items()}
|