refactor(jd): 反馈分组、列表统计、矩阵行与报告片段拆至 competitor_report

新增 consumer_feedback、run_context、list_mix、matrix_md、report_md_helpers；jd_competitor_report 从配置块后直接进入 build_competitor_markdown；移除未再使用的 ingredients 直引。

Made-with: Cursor

backend/crawler_copy/jd_pc_search/competitor_report/consumer_feedback.py

@@ -0,0 +1,161 @@
+"""评价与合并表按细类矩阵对齐：带前缀行、SKU→细类映射、按细类消费者反馈分组。"""
+from __future__ import annotations
+
+from pipeline.csv_schema import MERGED_FIELD_TO_CSV_HEADER
+
+from .comment_sentiment import _iter_comment_text_units
+from .constants import (
+    _COMMENT_CSV_BODY,
+    _COMMENT_CSV_SKU,
+    _MERGED_SHOP_CELL_KEYS,
+)
+from .csv_io import _cell, _md_cell
+from .matrix_group import _competitor_matrix_group_key, _merged_rows_grouped_for_matrix
+
+
+def _comment_lines_with_product_context(
+    comment_rows: list[dict[str, str]],
+    merged_rows: list[dict[str, str]],
+    *,
+    sku_header: str,
+    title_h: str,
+) -> list[str]:
+    """与 ``comment_rows`` 顺序对齐：带细类/SKU/品名/店铺前缀，供 §8.2 大模型抽样。"""
+    sku_meta: dict[str, tuple[str, str, str]] = {}
+    for row in merged_rows:
+        sku = _cell(row, sku_header).strip()
+        if not sku:
+            continue
+        gk = _competitor_matrix_group_key(row)
+        if not gk:
+            continue
+        sku_meta[sku] = (
+            gk,
+            _cell(row, title_h),
+            _cell(row, *_MERGED_SHOP_CELL_KEYS),
+        )
+    out: list[str] = []
+    for cr in comment_rows:
+        txt = _cell(cr, _COMMENT_CSV_BODY, "tagCommentContent")
+        if not txt:
+            continue
+        sku = _cell(cr, _COMMENT_CSV_SKU, "sku").strip()
+        meta = sku_meta.get(sku)
+        if meta:
+            gname, tit, shop = meta
+            prefix = (
+                f"【细类：{gname}｜SKU：{sku}｜品名：{_md_cell(tit, 80)}｜"
+                f"店铺：{_md_cell(shop, 40)}】"
+            )
+            out.append(prefix + txt)
+        else:
+            out.append(txt)
+    return out
+
+
+def _sku_to_matrix_group_map(
+    merged_rows: list[dict[str, str]], sku_header: str
+) -> dict[str, str]:
+    m: dict[str, str] = {}
+    for row in merged_rows:
+        sku = _cell(row, sku_header).strip()
+        if not sku:
+            continue
+        gk = _competitor_matrix_group_key(row)
+        if gk:
+            m[sku] = gk
+    return m
+
+
+def _comment_text_units_for_matrix_group(
+    gname: str,
+    merged_rows: list[dict[str, str]],
+    comment_rows_in_group: list[dict[str, str]],
+    sku_header: str,
+) -> list[str]:
+    """某细类下的评价正文列表；无 flat 时用该细类合并行的 comment_preview。"""
+    texts: list[str] = []
+    for row in comment_rows_in_group:
+        t = _cell(row, _COMMENT_CSV_BODY, "tagCommentContent")
+        if t:
+            texts.append(t)
+    if texts:
+        return texts
+    for row in merged_rows:
+        if _competitor_matrix_group_key(row) != gname:
+            continue
+        p = _cell(
+            row,
+            MERGED_FIELD_TO_CSV_HEADER["comment_preview"],
+            "comment_preview",
+        )
+        if p:
+            texts.append(p)
+    return texts
+
+
+def _consumer_feedback_by_matrix_group(
+    *,
+    merged_rows: list[dict[str, str]],
+    comment_rows: list[dict[str, str]],
+    sku_header: str,
+) -> list[tuple[str, list[dict[str, str]], list[str]]]:
+    """
+    与 §5 矩阵同序的细类列表；每项为 (细类名, 该类的 comments_flat 行, 用于场景统计的文本单元)。
+    评价 SKU 不在深入样本时归入「未归类（评价 SKU 无对应深入样本）」。
+    """
+    if not merged_rows:
+        if not comment_rows:
+            return []
+        texts = _iter_comment_text_units(comment_rows, [])
+        return [
+            (
+                "未归类（无深入合并表）",
+                list(comment_rows),
+                texts,
+            )
+        ]
+
+    sku_map = _sku_to_matrix_group_map(merged_rows, sku_header)
+    merged_by_sku: dict[str, dict[str, str]] = {}
+    for row in merged_rows:
+        s = _cell(row, sku_header).strip()
+        if s:
+            merged_by_sku[s] = row
+    by_g: dict[str, list[dict[str, str]]] = {}
+    for row in comment_rows:
+        sku = _cell(row, _COMMENT_CSV_SKU, "sku").strip()
+        g = sku_map.get(sku)
+        if g:
+            by_g.setdefault(g, []).append(row)
+            continue
+        if sku and sku in merged_by_sku:
+            # 深入样本存在但缺 detail_category_path（或路径无法解析为可读细类）：不参与按细类分析
+            continue
+        by_g.setdefault("未归类（评价 SKU 无对应深入样本）", []).append(row)
+
+    out: list[tuple[str, list[dict[str, str]], list[str]]] = []
+    used: set[str] = set()
+    for gname, _ in _merged_rows_grouped_for_matrix(merged_rows):
+        cr = by_g.get(gname, [])
+        tu = _comment_text_units_for_matrix_group(
+            gname, merged_rows, cr, sku_header
+        )
+        out.append((gname, cr, tu))
+        used.add(gname)
+    for gname, cr in sorted(by_g.items(), key=lambda x: (-len(x[1]), x[0])):
+        if gname in used:
+            continue
+        tu = _comment_text_units_for_matrix_group(
+            gname, merged_rows, cr, sku_header
+        )
+        out.append((gname, cr, tu))
+    return out
+
+
+__all__ = [
+    "_comment_lines_with_product_context",
+    "_comment_text_units_for_matrix_group",
+    "_consumer_feedback_by_matrix_group",
+    "_sku_to_matrix_group_map",
+]

backend/crawler_copy/jd_pc_search/competitor_report/list_mix.py

@@ -0,0 +1,138 @@
+"""列表可见度代理指标、品牌/店铺扇图用的名称列表与计数。"""
+from __future__ import annotations
+
+from collections import Counter
+from typing import Any
+
+from pipeline.csv_schema import JD_SEARCH_CSV_HEADERS, MERGED_FIELD_TO_CSV_HEADER
+
+from .constants import (
+    _LEGACY_LIST_BRAND_TITLE_KEY,
+    _LEGACY_SHOP_NAME_KEY,
+    _LIST_BRAND_TITLE_HEADER,
+    _MERGED_SHOP_CELL_KEYS,
+)
+from .csv_io import _cell, _collect_prices
+from .price_stats import _price_stats_extended
+
+
+def _structure_names_for_pie_counter(row_names: list[str]) -> list[str]:
+    """
+    与 ``_counter_mix_top_rows_with_remainder`` / 列表品牌·店铺扇图同一套规则：
+    按 strip 后的名称逐行保留一条，便于 ``_brand_cr`` 与饼图 Counter 一致。
+    """
+    return [(x or "").strip() for x in row_names if (x or "").strip()]
+
+
+def _brand_cr(cnames: list[str]) -> tuple[float | None, float | None, str, str]:
+    """按名称计数返回 (第一大主体份额, 前三合计份额, 头部标签, 头部占比展示字符串)。"""
+    if not cnames:
+        return None, None, "", ""
+    cnt = Counter(cnames)
+    total = sum(cnt.values())
+    if total <= 0:
+        return None, None, "", ""
+    mc = cnt.most_common()
+    top1_n = mc[0][1] if mc else 0
+    top1 = mc[0][0] if mc else ""
+    cr1 = top1_n / total
+    top3_n = sum(n for _, n in mc[:3])
+    cr3 = top3_n / total
+    return cr1, cr3, top1, f"{100.0 * top1_n / total:.1f}%"
+
+
+def _counter_mix_top_rows_with_remainder(
+    row_names: list[str], *, top_n: int, remainder_label: str
+) -> list[tuple[str, int]]:
+    """
+    与列表品牌/店铺扇图一致：按 strip 后的名称计数；``most_common(top_n)`` 未覆盖的长尾合并为
+    ``remainder_label``，保证各块 count 之和等于可统计行数（与 ``_structure_names_for_pie_counter`` 总条数一致）。
+    """
+    c = Counter((x or "").strip() for x in row_names if (x or "").strip())
+    if not c:
+        return []
+    total = sum(c.values())
+    common = c.most_common(top_n)
+    accounted = sum(v for _, v in common)
+    rest = total - accounted
+    out: list[tuple[str, int]] = list(common)
+    if rest > 0:
+        out.append((remainder_label, rest))
+    return out
+
+
+def _search_list_proxies(rows: list[dict[str, str]]) -> dict[str, Any]:
+    """
+    基于 pc_search_export 的「列表可见度」指标，**不是**全渠道零售额或 TAM。
+    """
+    sku_k = JD_SEARCH_CSV_HEADERS["sku_id"]
+    shop_k = JD_SEARCH_CSV_HEADERS["shop_name"]
+    page_k = JD_SEARCH_CSV_HEADERS["page"]
+    cat_k = JD_SEARCH_CSV_HEADERS["leaf_category"]
+    skus: set[str] = set()
+    shops: set[str] = set()
+    pages: set[str] = set()
+    cats: set[str] = set()
+    for r in rows:
+        s = _cell(r, sku_k)
+        if s:
+            skus.add(s)
+        sh = _cell(r, shop_k)
+        if sh:
+            shops.add(sh)
+        pg = _cell(r, page_k)
+        if pg:
+            pages.add(pg)
+        c = _cell(r, cat_k)
+        if c:
+            cats.add(c)
+    prices = _collect_prices(rows)
+    pst = _price_stats_extended(prices)
+    return {
+        "total_rows": len(rows),
+        "unique_skus": len(skus),
+        "unique_shops": len(shops),
+        "unique_pages": len(pages),
+        "page_span": (min((int(p) for p in pages if p.isdigit()), default=None), max((int(p) for p in pages if p.isdigit()), default=None)),
+        "unique_leaf_cats": len(cats),
+        "list_price_stats": pst,
+    }
+
+
+def _structure_shops(rows: list[dict[str, str]], *, list_export: bool) -> list[str]:
+    if list_export:
+        return [
+            _cell(r, JD_SEARCH_CSV_HEADERS["shop_name"], _LEGACY_SHOP_NAME_KEY)
+            for r in rows
+            if _cell(r, JD_SEARCH_CSV_HEADERS["shop_name"], _LEGACY_SHOP_NAME_KEY)
+        ]
+    out: list[str] = []
+    for r in rows:
+        s = _cell(r, *_MERGED_SHOP_CELL_KEYS)
+        if s:
+            out.append(s)
+    return out
+
+
+def _structure_brands(rows: list[dict[str, str]], *, list_export: bool) -> list[str]:
+    if list_export:
+        return [
+            _cell(r, _LIST_BRAND_TITLE_HEADER, _LEGACY_LIST_BRAND_TITLE_KEY)
+            for r in rows
+            if _cell(r, _LIST_BRAND_TITLE_HEADER, _LEGACY_LIST_BRAND_TITLE_KEY)
+        ]
+    return [
+        _cell(r, MERGED_FIELD_TO_CSV_HEADER["detail_brand"], "detail_brand")
+        for r in rows
+        if _cell(r, MERGED_FIELD_TO_CSV_HEADER["detail_brand"], "detail_brand")
+    ]
+
+
+__all__ = [
+    "_brand_cr",
+    "_counter_mix_top_rows_with_remainder",
+    "_search_list_proxies",
+    "_structure_brands",
+    "_structure_names_for_pie_counter",
+    "_structure_shops",
+]

backend/crawler_copy/jd_pc_search/competitor_report/matrix_md.py

@@ -0,0 +1,87 @@
+"""竞品矩阵 Markdown 行：配料格与整行管道表单元。"""
+from __future__ import annotations
+
+from pipeline.csv_schema import MERGED_FIELD_TO_CSV_HEADER, merged_csv_effective_total_sales
+
+from .constants import (
+    _COMMENT_FUZZ_KEYS,
+    _DETAIL_PRICE_FINAL_CSV_KEYS,
+    _LEGACY_RANK_TAGLINE_KEY,
+    _LEGACY_SELLING_POINT_KEY,
+    _LIST_SHOW_PRICE_CELL_KEYS,
+    _MERGED_SHOP_CELL_KEYS,
+    _RANK_TAGLINE_KEY,
+    _SELLING_POINT_KEY,
+)
+from .csv_io import _cell, _detail_category_path_cell, _md_cell
+from .ingredients import (
+    _ingredients_from_product_attributes,
+    _ingredients_single_line,
+    _is_ingredient_url_blob,
+)
+
+
+def _matrix_ingredients_cell(row: dict[str, str], *, max_len: int = 420) -> str:
+    """
+    优先 ``detail_body_ingredients``（配料 OCR/文本）；旧合并表可能为 ``detail_body_image_urls``。
+    若为 URL 串则尝试 ``detail_product_attributes`` 中的「配料/配料表：」片段。
+    """
+    raw = _cell(
+        row,
+        MERGED_FIELD_TO_CSV_HEADER["detail_body_ingredients"],
+        "detail_body_ingredients",
+        "detail_body_image_urls",
+    )
+    if raw and not _is_ingredient_url_blob(raw):
+        return _md_cell(_ingredients_single_line(raw), max_len)
+    from_attr = _ingredients_from_product_attributes(
+        _cell(
+            row,
+            MERGED_FIELD_TO_CSV_HEADER["detail_product_attributes"],
+            "detail_product_attributes",
+        )
+    )
+    if from_attr:
+        return _md_cell(from_attr, max_len)
+    if raw and _is_ingredient_url_blob(raw):
+        return _md_cell(
+            "（详情长图链接，无配料正文；可在采集侧开启配料识别后重新跑批次）",
+            max_len,
+        )
+    return "—"
+
+
+def _competitor_matrix_md_line(
+    row: dict[str, str], *, sku_header: str, title_h: str
+) -> str:
+    sku = _md_cell(_cell(row, sku_header), 14)
+    title = _md_cell(_cell(row, title_h), 56)
+    brand = _md_cell(
+        _cell(row, MERGED_FIELD_TO_CSV_HEADER["detail_brand"], "detail_brand"), 16
+    )
+    pj = _md_cell(_cell(row, *_LIST_SHOW_PRICE_CELL_KEYS), 10)
+    df = _md_cell(_cell(row, *_DETAIL_PRICE_FINAL_CSV_KEYS), 10)
+    shop = _md_cell(_cell(row, *_MERGED_SHOP_CELL_KEYS), 22)
+    sell = _md_cell(_cell(row, _SELLING_POINT_KEY, _LEGACY_SELLING_POINT_KEY), 36)
+    rank = _md_cell(
+        _cell(row, _RANK_TAGLINE_KEY, _LEGACY_RANK_TAGLINE_KEY), 28
+    )
+    cat = _md_cell(_detail_category_path_cell(row), 24)
+    ing = _matrix_ingredients_cell(row)
+    ts_eff = merged_csv_effective_total_sales(row)
+    cc = _md_cell(ts_eff or _cell(row, *_COMMENT_FUZZ_KEYS), 14)
+    prev = _md_cell(
+        _cell(
+            row,
+            MERGED_FIELD_TO_CSV_HEADER["comment_preview"],
+            "comment_preview",
+        ),
+        72,
+    )
+    return (
+        f"| {sku} | {title} | {brand} | {pj} | {df} | {shop} | {sell} | {rank} | "
+        f"{cat} | {ing} | {cc} | {prev} |"
+    )
+
+
+__all__ = ["_competitor_matrix_md_line", "_matrix_ingredients_cell"]

backend/crawler_copy/jd_pc_search/competitor_report/report_md_helpers.py

@@ -0,0 +1,202 @@
+"""报告 Markdown 片段：Mermaid、场景摘要、规则策略提示、插图路径与解读段落。"""
+from __future__ import annotations
+
+import re
+from collections import Counter
+from pathlib import Path
+from typing import Any
+
+from .csv_io import _md_cell
+
+
+def _mermaid_pie_focus_keywords(hits: Counter[str], *, top_k: int = 8) -> str:
+    """关注词全局 Top 的 Mermaid pie（便于渲染或导出工具识别）。"""
+    top = hits.most_common(top_k)
+    if not top:
+        return ""
+    rest = list(hits.most_common())
+    if len(rest) > top_k:
+        others_n = sum(n for _, n in rest[top_k:])
+    else:
+        others_n = 0
+    lines = ["```mermaid", 'pie title 关注词命中次数（全局 Top，子串计数）']
+    for w, n in top:
+        if n <= 0:
+            continue
+        label = (w or "?").replace('"', "'").replace("\n", " ")[:18]
+        lines.append(f'    "{label}（{n}）" : {n}')
+    if others_n > 0:
+        lines.append(f'    "其余词合计" : {others_n}')
+    lines.append("```")
+    return "\n".join(lines)
+
+
+def _scenario_summary_bullets(counter: Counter[str], n_texts: int, top_k: int = 5) -> list[str]:
+    if n_texts <= 0 or not counter:
+        return []
+    ordered = counter.most_common()
+    lines: list[str] = []
+    head = ordered[:top_k]
+    parts = []
+    for label, cnt in head:
+        pct = 100.0 * cnt / n_texts
+        parts.append(f"「{label}」约 **{cnt}** 条（占有效文本 **{pct:.0f}%**）")
+    lines.append(
+        "用户自述的用途/场景（基于预设词组，**非语义分类**）：" + "；".join(parts) + "。"
+    )
+    tail = [lbl for lbl, n in ordered[top_k:] if n > 0]
+    if tail:
+        lines.append(f"另有提及较少的场景标签：{'、'.join(tail)}。")
+    return lines
+
+
+def _strategy_hints(
+    *,
+    cr1: float | None,
+    pst: dict[str, Any],
+    hits: Counter[str],
+    n_comments: int,
+    scen_counts: Counter[str],
+    scen_n_texts: int,
+) -> list[str]:
+    """基于规则的「提示性」结论，均标注待验证。"""
+    hints: list[str] = []
+    if cr1 is not None and cr1 >= 0.45:
+        hints.append(
+            "样本内品牌集中度较高（第一大品牌份额偏高），头部玩家占据显著曝光；新原料/解决方案宜明确差异化价值主张（**需线下渠道与招商信息交叉验证**）。"
+        )
+    elif cr1 is not None and cr1 < 0.25:
+        hints.append(
+            "样本内品牌较分散，品类或关键词下竞争格局未固化，存在定位与叙事空间（**需扩大样本页数与关键词矩阵验证**）。"
+        )
+    if pst.get("stdev") and pst.get("mean") and pst["mean"] > 0:
+        cv = pst["stdev"] / pst["mean"]
+        if cv > 0.35:
+            hints.append(
+                "价格离散度较高，同时存在偏低价与偏高价陈列，可分别对标「性价比带」与「品质/功能带」竞品（**终端到手价受促销影响，非成本结构**）。"
+            )
+    if hits:
+        top = hits.most_common(3)
+        top_s = "、".join(w for w, _ in top)
+        hints.append(
+            f"评价文本中「{top_s}」等主题出现较多，可作为消费者沟通与产品卖点的假设输入（**非严格主题模型，建议人工抽样复核**）。"
+        )
+    if n_comments < 5:
+        hints.append(
+            "有效评价样本偏少，消费者洞察部分仅作方向参考，正式结论建议加大 SKU 数或评论分页。"
+        )
+    if scen_n_texts >= 5 and scen_counts:
+        top_lbl, top_n = scen_counts.most_common(1)[0]
+        share = top_n / scen_n_texts
+        if share >= 0.25:
+            hints.append(
+                f"用途/场景中「{top_lbl}」在约 {100 * share:.0f}% 的有效评价自述中出现，可作为沟通场景与卖点的优先假设（**词组规则，建议抽样核对原句**）。"
+            )
+    if not hints:
+        hints.append(
+            "当前样本下自动规则未触发强信号；请结合业务目标人工解读对比矩阵与原始 CSV。"
+        )
+    return hints
+
+
+def _embed_chart(run_dir: Path, filename: str, caption: str = "") -> list[str]:
+    """若 ``report_assets/<filename>`` 存在则返回插图 Markdown 片段。"""
+    if not (run_dir / "report_assets" / filename).is_file():
+        return []
+    cap = (caption or "").strip()
+    out: list[str] = []
+    if cap:
+        out.append(f"*{cap}*")
+        out.append("")
+    out.append(f"![](report_assets/{filename})")
+    out.append("")
+    return out
+
+
+def _scenario_group_asset_slug(group: str, index: int) -> str:
+    """与 ``pipeline.reporting.charts`` 中场景分组图文件名规则一致（勿改格式）。"""
+    raw = (group or "").strip()
+    core = re.sub(r"[^\w\u4e00-\u9fff-]", "", raw)[:20]
+    if not core:
+        core = "group"
+    return f"i{index:02d}_{core}"
+
+
+def _focus_scenario_combo_bar_filename(group: str, index: int) -> str:
+    """关注词 + 使用场景并排条形图（与 ``pipeline.reporting.charts.save_combo_focus_scenario_bar`` 同源）。"""
+    slug = _scenario_group_asset_slug(group, index)
+    return f"chart_focus_and_scenarios_bar__{slug}.png"
+
+
+def _matrix_prices_sales_chart_filename(group: str, index: int) -> str:
+    """与 ``pipeline.reporting.charts.generate_report_charts`` 中 ``chart_matrix_prices_sales__*`` 一致。"""
+    slug = _scenario_group_asset_slug(group, index)
+    return f"chart_matrix_prices_sales__{slug}.png"
+
+
+def _lines_4_reading_brand(
+    *,
+    cr1: float | None,
+    cr3: float | None,
+    top: str,
+    brand_rows_n: int,
+    n_structure: int,
+) -> list[str]:
+    if cr1 is None or not (top or "").strip():
+        return []
+    lines = [
+        "",
+        "**数据解读（规则摘要）**：",
+        "",
+        f"- 在含品牌字段的 **{brand_rows_n}** 条列表行（占本章结构样本 **{n_structure}** 行）中，"
+        f"「{_md_cell(top.strip(), 36)}」曝光约占 **{100 * cr1:.1f}%**（按行计，同一 SKU 多行会重复计）。",
+    ]
+    if cr3 is not None:
+        lines.append(
+            f"- 前三品牌合计约 **{100 * cr3:.1f}%**；若该比例偏高，说明搜索页品牌集中度高，"
+            "新品需搭配清晰的差异定位与资源投放，避免与头部在泛词下正面撞车。"
+        )
+    lines.append("")
+    return lines
+
+
+def _lines_4_reading_shop(
+    *,
+    cr1: float | None,
+    cr3: float | None,
+    top: str,
+    shop_rows_n: int,
+    n_structure: int,
+) -> list[str]:
+    if not shop_rows_n:
+        return []
+    lines = [
+        "",
+        "**数据解读（规则摘要）**：",
+        "",
+        f"- 含店铺名的列表行共 **{shop_rows_n}** 条（结构样本 **{n_structure}** 行），反映搜索曝光下的店铺格局。",
+    ]
+    if cr1 is not None and (top or "").strip():
+        lines.append(
+            f"- 第一大店铺「{_md_cell(top.strip(), 40)}」约占 **{100 * cr1:.1f}%**；"
+            "该指标刻画的是**列表可见度**而非销量，适合用于判断货架被哪些店铺占据。"
+        )
+    if cr3 is not None:
+        lines.append(
+            f"- 前三店铺合计约 **{100 * cr3:.1f}%**；若集中度高，可考虑从店铺矩阵、旗舰店/专营店布局等角度拆解竞争。"
+        )
+    lines.append("")
+    return lines
+
+
+__all__ = [
+    "_embed_chart",
+    "_focus_scenario_combo_bar_filename",
+    "_lines_4_reading_brand",
+    "_lines_4_reading_shop",
+    "_matrix_prices_sales_chart_filename",
+    "_mermaid_pie_focus_keywords",
+    "_scenario_group_asset_slug",
+    "_scenario_summary_bullets",
+    "_strategy_hints",
+]

backend/crawler_copy/jd_pc_search/competitor_report/run_context.py

@@ -0,0 +1,91 @@
+"""运行目录解析、关键词推断、pc_search_raw 检索规模读取。"""
+from __future__ import annotations
+
+import json
+import re
+from collections import Counter
+from pathlib import Path
+from typing import Any
+
+
+def _run_batch_label(run_dir: Path) -> str:
+    name = run_dir.name
+    m = re.match(r"^(\d{8})_(\d{6})_", name)
+    if m:
+        return f"{m.group(1)} {m.group(2)}"
+    return name
+
+
+def _resolve_existing_run_dir(raw: str | Path | None) -> Path | None:
+    if raw is None:
+        return None
+    s = str(raw).strip()
+    if not s:
+        return None
+    p = Path(s).expanduser()
+    if not p.is_absolute():
+        p = (Path.cwd() / p).resolve()
+    else:
+        p = p.resolve()
+    return p
+
+
+def _infer_keyword(run_dir: Path, meta: dict[str, Any] | None) -> str:
+    if meta:
+        k = str(meta.get("keyword") or "").strip()
+        if k:
+            return k
+    m = re.match(r"^\d{8}_\d{6}_(.+)$", run_dir.name)
+    if m:
+        return m.group(1).strip()
+    return ""
+
+
+def _pc_search_result_count_from_raw(
+    run_dir: Path,
+) -> tuple[int | None, str, list[int], int, int]:
+    """
+    从 ``pc_search_raw/*.json`` 读取 ``data.resultCount``（京东 PC 搜索接口返回的检索命中规模）。
+    多文件时取众数；返回 (众数值, data.listKeyWord 首见值, 出现过的不同取值升序, 解析到的样本文件数)。
+    """
+    raw_dir = run_dir / "pc_search_raw"
+    if not raw_dir.is_dir():
+        return None, "", [], 0, 0
+    counts: list[int] = []
+    list_kw = ""
+    n_files = 0
+    for p in sorted(raw_dir.glob("*.json")):
+        try:
+            obj = json.loads(p.read_text(encoding="utf-8"))
+        except (json.JSONDecodeError, OSError, UnicodeError):
+            continue
+        n_files += 1
+        if not isinstance(obj, dict):
+            continue
+        data = obj.get("data")
+        if not isinstance(data, dict):
+            continue
+        rc = data.get("resultCount")
+        val: int | None = None
+        if isinstance(rc, int) and not isinstance(rc, bool) and rc >= 0:
+            val = rc
+        elif isinstance(rc, str) and rc.strip().isdigit():
+            val = int(rc.strip())
+        if val is not None:
+            counts.append(val)
+        lk = data.get("listKeyWord")
+        if isinstance(lk, str) and lk.strip() and not list_kw:
+            list_kw = lk.strip()
+    if not counts:
+        return None, list_kw, [], n_files, 0
+    consensus_rc, _freq = Counter(counts).most_common(1)[0]
+    uniques = sorted(set(counts))
+    return consensus_rc, list_kw, uniques, n_files, len(counts)
+
+
+__all__ = [
+    "_infer_keyword",
+    "_pc_search_result_count_from_raw",
+    "_resolve_existing_run_dir",
+    "_run_batch_label",
+]

backend/crawler_copy/jd_pc_search/jd_competitor_report.py

@@ -65,11 +65,6 @@ from competitor_report.comment_sentiment import (  # noqa: E402
     _merge_comment_previews,
     _parse_comment_score,
 )
-from competitor_report.ingredients import (  # noqa: E402
-    _ingredients_from_product_attributes,
-    _ingredients_single_line,
-    _is_ingredient_url_blob,
-)
 from competitor_report.llm_group_payloads import (  # noqa: E402
     build_comment_groups_llm_payload,
     build_matrix_groups_llm_payload,
@@ -86,6 +81,40 @@ from competitor_report.matrix_group import (  # noqa: E402
     _merged_rows_grouped_for_matrix,
 )
 from competitor_report.price_stats import _price_stats_extended  # noqa: E402
+from competitor_report.consumer_feedback import (  # noqa: E402
+    _comment_lines_with_product_context,
+    _consumer_feedback_by_matrix_group,
+    _sku_to_matrix_group_map,
+)
+from competitor_report.list_mix import (  # noqa: E402
+    _brand_cr,
+    _counter_mix_top_rows_with_remainder,
+    _search_list_proxies,
+    _structure_brands,
+    _structure_names_for_pie_counter,
+    _structure_shops,
+)
+from competitor_report.matrix_md import (  # noqa: E402
+    _competitor_matrix_md_line,
+    _matrix_ingredients_cell,
+)
+from competitor_report.report_md_helpers import (  # noqa: E402
+    _embed_chart,
+    _focus_scenario_combo_bar_filename,
+    _lines_4_reading_brand,
+    _lines_4_reading_shop,
+    _matrix_prices_sales_chart_filename,
+    _mermaid_pie_focus_keywords,
+    _scenario_group_asset_slug,
+    _scenario_summary_bullets,
+    _strategy_hints,
+)
+from competitor_report.run_context import (  # noqa: E402
+    _infer_keyword,
+    _pc_search_result_count_from_raw,
+    _resolve_existing_run_dir,
+    _run_batch_label,
+)
 
 # ---------------------------------------------------------------------------
 # 运行配置（按需改这里；与 competitor_report.constants 中默认关注词等配合使用）
@@ -100,586 +129,6 @@ OVERRIDE_MAX_SKUS: int | None = None
 OVERRIDE_PAGE_START: int | None = None
 OVERRIDE_PAGE_TO: int | None = None
 
-def _comment_lines_with_product_context(
-    comment_rows: list[dict[str, str]],
-    merged_rows: list[dict[str, str]],
-    *,
-    sku_header: str,
-    title_h: str,
-) -> list[str]:
-    """与 ``comment_rows`` 顺序对齐：带细类/SKU/品名/店铺前缀，供 §8.2 大模型抽样。"""
-    sku_meta: dict[str, tuple[str, str, str]] = {}
-    for row in merged_rows:
-        sku = _cell(row, sku_header).strip()
-        if not sku:
-            continue
-        gk = _competitor_matrix_group_key(row)
-        if not gk:
-            continue
-        sku_meta[sku] = (
-            gk,
-            _cell(row, title_h),
-            _cell(row, *_MERGED_SHOP_CELL_KEYS),
-        )
-    out: list[str] = []
-    for cr in comment_rows:
-        txt = _cell(cr, _COMMENT_CSV_BODY, "tagCommentContent")
-        if not txt:
-            continue
-        sku = _cell(cr, _COMMENT_CSV_SKU, "sku").strip()
-        meta = sku_meta.get(sku)
-        if meta:
-            gname, tit, shop = meta
-            prefix = (
-                f"【细类：{gname}｜SKU：{sku}｜品名：{_md_cell(tit, 80)}｜"
-                f"店铺：{_md_cell(shop, 40)}】"
-            )
-            out.append(prefix + txt)
-        else:
-            out.append(txt)
-    return out
-
-
-
-def _mermaid_pie_focus_keywords(hits: Counter[str], *, top_k: int = 8) -> str:
-    """关注词全局 Top 的 Mermaid pie（便于渲染或导出工具识别）。"""
-    top = hits.most_common(top_k)
-    if not top:
-        return ""
-    rest = list(hits.most_common())
-    if len(rest) > top_k:
-        others_n = sum(n for _, n in rest[top_k:])
-    else:
-        others_n = 0
-    lines = ["```mermaid", 'pie title 关注词命中次数（全局 Top，子串计数）']
-    for w, n in top:
-        if n <= 0:
-            continue
-        label = (w or "?").replace('"', "'").replace("\n", " ")[:18]
-        lines.append(f'    "{label}（{n}）" : {n}')
-    if others_n > 0:
-        lines.append(f'    "其余词合计" : {others_n}')
-    lines.append("```")
-    return "\n".join(lines)
-
-
-
-
-def _scenario_summary_bullets(counter: Counter[str], n_texts: int, top_k: int = 5) -> list[str]:
-    if n_texts <= 0 or not counter:
-        return []
-    ordered = counter.most_common()
-    lines: list[str] = []
-    head = ordered[:top_k]
-    parts = []
-    for label, cnt in head:
-        pct = 100.0 * cnt / n_texts
-        parts.append(f"「{label}」约 **{cnt}** 条（占有效文本 **{pct:.0f}%**）")
-    lines.append(
-        "用户自述的用途/场景（基于预设词组，**非语义分类**）：" + "；".join(parts) + "。"
-    )
-    tail = [lbl for lbl, n in ordered[top_k:] if n > 0]
-    if tail:
-        lines.append(f"另有提及较少的场景标签：{'、'.join(tail)}。")
-    return lines
-
-
-def _sku_to_matrix_group_map(
-    merged_rows: list[dict[str, str]], sku_header: str
-) -> dict[str, str]:
-    m: dict[str, str] = {}
-    for row in merged_rows:
-        sku = _cell(row, sku_header).strip()
-        if not sku:
-            continue
-        gk = _competitor_matrix_group_key(row)
-        if gk:
-            m[sku] = gk
-    return m
-
-
-def _comment_text_units_for_matrix_group(
-    gname: str,
-    merged_rows: list[dict[str, str]],
-    comment_rows_in_group: list[dict[str, str]],
-    sku_header: str,
-) -> list[str]:
-    """某细类下的评价正文列表；无 flat 时用该细类合并行的 comment_preview。"""
-    texts: list[str] = []
-    for row in comment_rows_in_group:
-        t = _cell(row, _COMMENT_CSV_BODY, "tagCommentContent")
-        if t:
-            texts.append(t)
-    if texts:
-        return texts
-    for row in merged_rows:
-        if _competitor_matrix_group_key(row) != gname:
-            continue
-        p = _cell(
-            row,
-            MERGED_FIELD_TO_CSV_HEADER["comment_preview"],
-            "comment_preview",
-        )
-        if p:
-            texts.append(p)
-    return texts
-
-
-
-
-def _consumer_feedback_by_matrix_group(
-    *,
-    merged_rows: list[dict[str, str]],
-    comment_rows: list[dict[str, str]],
-    sku_header: str,
-) -> list[tuple[str, list[dict[str, str]], list[str]]]:
-    """
-    与 §5 矩阵同序的细类列表；每项为 (细类名, 该类的 comments_flat 行, 用于场景统计的文本单元)。
-    评价 SKU 不在深入样本时归入「未归类（评价 SKU 无对应深入样本）」。
-    """
-    if not merged_rows:
-        if not comment_rows:
-            return []
-        texts = _iter_comment_text_units(comment_rows, [])
-        return [
-            (
-                "未归类（无深入合并表）",
-                list(comment_rows),
-                texts,
-            )
-        ]
-
-    sku_map = _sku_to_matrix_group_map(merged_rows, sku_header)
-    merged_by_sku: dict[str, dict[str, str]] = {}
-    for row in merged_rows:
-        s = _cell(row, sku_header).strip()
-        if s:
-            merged_by_sku[s] = row
-    by_g: dict[str, list[dict[str, str]]] = {}
-    for row in comment_rows:
-        sku = _cell(row, _COMMENT_CSV_SKU, "sku").strip()
-        g = sku_map.get(sku)
-        if g:
-            by_g.setdefault(g, []).append(row)
-            continue
-        if sku and sku in merged_by_sku:
-            # 深入样本存在但缺 detail_category_path（或路径无法解析为可读细类）：不参与按细类分析
-            continue
-        by_g.setdefault("未归类（评价 SKU 无对应深入样本）", []).append(row)
-
-    out: list[tuple[str, list[dict[str, str]], list[str]]] = []
-    used: set[str] = set()
-    for gname, _ in _merged_rows_grouped_for_matrix(merged_rows):
-        cr = by_g.get(gname, [])
-        tu = _comment_text_units_for_matrix_group(
-            gname, merged_rows, cr, sku_header
-        )
-        out.append((gname, cr, tu))
-        used.add(gname)
-    for gname, cr in sorted(by_g.items(), key=lambda x: (-len(x[1]), x[0])):
-        if gname in used:
-            continue
-        tu = _comment_text_units_for_matrix_group(
-            gname, merged_rows, cr, sku_header
-        )
-        out.append((gname, cr, tu))
-    return out
-
-
-def _run_batch_label(run_dir: Path) -> str:
-    name = run_dir.name
-    m = re.match(r"^(\d{8})_(\d{6})_", name)
-    if m:
-        return f"{m.group(1)} {m.group(2)}"
-    return name
-
-
-def _resolve_existing_run_dir(raw: str | Path | None) -> Path | None:
-    if raw is None:
-        return None
-    s = str(raw).strip()
-    if not s:
-        return None
-    p = Path(s).expanduser()
-    if not p.is_absolute():
-        p = (Path.cwd() / p).resolve()
-    else:
-        p = p.resolve()
-    return p
-
-
-def _infer_keyword(run_dir: Path, meta: dict[str, Any] | None) -> str:
-    if meta:
-        k = str(meta.get("keyword") or "").strip()
-        if k:
-            return k
-    m = re.match(r"^\d{8}_\d{6}_(.+)$", run_dir.name)
-    if m:
-        return m.group(1).strip()
-    return ""
-
-
-def _pc_search_result_count_from_raw(
-    run_dir: Path,
-) -> tuple[int | None, str, list[int], int, int]:
-    """
-    从 ``pc_search_raw/*.json`` 读取 ``data.resultCount``（京东 PC 搜索接口返回的检索命中规模）。
-    多文件时取众数；返回 (众数值, data.listKeyWord 首见值, 出现过的不同取值升序, 解析到的样本文件数)。
-    """
-    raw_dir = run_dir / "pc_search_raw"
-    if not raw_dir.is_dir():
-        return None, "", [], 0, 0
-    counts: list[int] = []
-    list_kw = ""
-    n_files = 0
-    for p in sorted(raw_dir.glob("*.json")):
-        try:
-            obj = json.loads(p.read_text(encoding="utf-8"))
-        except (json.JSONDecodeError, OSError, UnicodeError):
-            continue
-        n_files += 1
-        if not isinstance(obj, dict):
-            continue
-        data = obj.get("data")
-        if not isinstance(data, dict):
-            continue
-        rc = data.get("resultCount")
-        val: int | None = None
-        if isinstance(rc, int) and not isinstance(rc, bool) and rc >= 0:
-            val = rc
-        elif isinstance(rc, str) and rc.strip().isdigit():
-            val = int(rc.strip())
-        if val is not None:
-            counts.append(val)
-        lk = data.get("listKeyWord")
-        if isinstance(lk, str) and lk.strip() and not list_kw:
-            list_kw = lk.strip()
-    if not counts:
-        return None, list_kw, [], n_files, 0
-    consensus_rc, _freq = Counter(counts).most_common(1)[0]
-    uniques = sorted(set(counts))
-    return consensus_rc, list_kw, uniques, n_files, len(counts)
-
-
-def _structure_names_for_pie_counter(row_names: list[str]) -> list[str]:
-    """
-    与 ``_counter_mix_top_rows_with_remainder`` / 列表品牌·店铺扇图同一套规则：
-    按 strip 后的名称逐行保留一条，便于 ``_brand_cr`` 与饼图 Counter 一致。
-    """
-    return [(x or "").strip() for x in row_names if (x or "").strip()]
-
-
-def _brand_cr(cnames: list[str]) -> tuple[float | None, float | None, str, str]:
-    """按名称计数返回 (第一大主体份额, 前三合计份额, 头部标签, 头部占比展示字符串)。"""
-    if not cnames:
-        return None, None, "", ""
-    cnt = Counter(cnames)
-    total = sum(cnt.values())
-    if total <= 0:
-        return None, None, "", ""
-    mc = cnt.most_common()
-    top1_n = mc[0][1] if mc else 0
-    top1 = mc[0][0] if mc else ""
-    cr1 = top1_n / total
-    top3_n = sum(n for _, n in mc[:3])
-    cr3 = top3_n / total
-    return cr1, cr3, top1, f"{100.0 * top1_n / total:.1f}%"
-
-
-def _counter_mix_top_rows_with_remainder(
-    row_names: list[str], *, top_n: int, remainder_label: str
-) -> list[tuple[str, int]]:
-    """
-    与列表品牌/店铺扇图一致：按 strip 后的名称计数；``most_common(top_n)`` 未覆盖的长尾合并为
-    ``remainder_label``，保证各块 count 之和等于可统计行数（与 ``_structure_names_for_pie_counter`` 总条数一致）。
-    """
-    c = Counter((x or "").strip() for x in row_names if (x or "").strip())
-    if not c:
-        return []
-    total = sum(c.values())
-    common = c.most_common(top_n)
-    accounted = sum(v for _, v in common)
-    rest = total - accounted
-    out: list[tuple[str, int]] = list(common)
-    if rest > 0:
-        out.append((remainder_label, rest))
-    return out
-
-
-
-
-def _search_list_proxies(rows: list[dict[str, str]]) -> dict[str, Any]:
-    """
-    基于 pc_search_export 的「列表可见度」指标，**不是**全渠道零售额或 TAM。
-    """
-    sku_k = JD_SEARCH_CSV_HEADERS["sku_id"]
-    shop_k = JD_SEARCH_CSV_HEADERS["shop_name"]
-    page_k = JD_SEARCH_CSV_HEADERS["page"]
-    cat_k = JD_SEARCH_CSV_HEADERS["leaf_category"]
-    skus: set[str] = set()
-    shops: set[str] = set()
-    pages: set[str] = set()
-    cats: set[str] = set()
-    for r in rows:
-        s = _cell(r, sku_k)
-        if s:
-            skus.add(s)
-        sh = _cell(r, shop_k)
-        if sh:
-            shops.add(sh)
-        pg = _cell(r, page_k)
-        if pg:
-            pages.add(pg)
-        c = _cell(r, cat_k)
-        if c:
-            cats.add(c)
-    prices = _collect_prices(rows)
-    pst = _price_stats_extended(prices)
-    return {
-        "total_rows": len(rows),
-        "unique_skus": len(skus),
-        "unique_shops": len(shops),
-        "unique_pages": len(pages),
-        "page_span": (min((int(p) for p in pages if p.isdigit()), default=None), max((int(p) for p in pages if p.isdigit()), default=None)),
-        "unique_leaf_cats": len(cats),
-        "list_price_stats": pst,
-    }
-
-
-
-
-def _structure_shops(rows: list[dict[str, str]], *, list_export: bool) -> list[str]:
-    if list_export:
-        return [
-            _cell(r, JD_SEARCH_CSV_HEADERS["shop_name"], _LEGACY_SHOP_NAME_KEY)
-            for r in rows
-            if _cell(r, JD_SEARCH_CSV_HEADERS["shop_name"], _LEGACY_SHOP_NAME_KEY)
-        ]
-    out: list[str] = []
-    for r in rows:
-        s = _cell(r, *_MERGED_SHOP_CELL_KEYS)
-        if s:
-            out.append(s)
-    return out
-
-
-def _structure_brands(rows: list[dict[str, str]], *, list_export: bool) -> list[str]:
-    if list_export:
-        return [
-            _cell(r, _LIST_BRAND_TITLE_HEADER, _LEGACY_LIST_BRAND_TITLE_KEY)
-            for r in rows
-            if _cell(r, _LIST_BRAND_TITLE_HEADER, _LEGACY_LIST_BRAND_TITLE_KEY)
-        ]
-    return [
-        _cell(r, MERGED_FIELD_TO_CSV_HEADER["detail_brand"], "detail_brand")
-        for r in rows
-        if _cell(r, MERGED_FIELD_TO_CSV_HEADER["detail_brand"], "detail_brand")
-    ]
-
-
-
-
-def _matrix_ingredients_cell(row: dict[str, str], *, max_len: int = 420) -> str:
-    """
-    优先 ``detail_body_ingredients``（配料 OCR/文本）；旧合并表可能为 ``detail_body_image_urls``。
-    若为 URL 串则尝试 ``detail_product_attributes`` 中的「配料/配料表：」片段。
-    """
-    raw = _cell(
-        row,
-        MERGED_FIELD_TO_CSV_HEADER["detail_body_ingredients"],
-        "detail_body_ingredients",
-        "detail_body_image_urls",
-    )
-    if raw and not _is_ingredient_url_blob(raw):
-        return _md_cell(_ingredients_single_line(raw), max_len)
-    from_attr = _ingredients_from_product_attributes(
-        _cell(
-            row,
-            MERGED_FIELD_TO_CSV_HEADER["detail_product_attributes"],
-            "detail_product_attributes",
-        )
-    )
-    if from_attr:
-        return _md_cell(from_attr, max_len)
-    if raw and _is_ingredient_url_blob(raw):
-        return _md_cell(
-            "（详情长图链接，无配料正文；可在采集侧开启配料识别后重新跑批次）",
-            max_len,
-        )
-    return "—"
-
-
-def _competitor_matrix_md_line(
-    row: dict[str, str], *, sku_header: str, title_h: str
-) -> str:
-    sku = _md_cell(_cell(row, sku_header), 14)
-    title = _md_cell(_cell(row, title_h), 56)
-    brand = _md_cell(
-        _cell(row, MERGED_FIELD_TO_CSV_HEADER["detail_brand"], "detail_brand"), 16
-    )
-    pj = _md_cell(_cell(row, *_LIST_SHOW_PRICE_CELL_KEYS), 10)
-    df = _md_cell(_cell(row, *_DETAIL_PRICE_FINAL_CSV_KEYS), 10)
-    shop = _md_cell(_cell(row, *_MERGED_SHOP_CELL_KEYS), 22)
-    sell = _md_cell(_cell(row, _SELLING_POINT_KEY, _LEGACY_SELLING_POINT_KEY), 36)
-    rank = _md_cell(
-        _cell(row, _RANK_TAGLINE_KEY, _LEGACY_RANK_TAGLINE_KEY), 28
-    )
-    cat = _md_cell(_detail_category_path_cell(row), 24)
-    ing = _matrix_ingredients_cell(row)
-    ts_eff = merged_csv_effective_total_sales(row)
-    cc = _md_cell(ts_eff or _cell(row, *_COMMENT_FUZZ_KEYS), 14)
-    prev = _md_cell(
-        _cell(
-            row,
-            MERGED_FIELD_TO_CSV_HEADER["comment_preview"],
-            "comment_preview",
-        ),
-        72,
-    )
-    return (
-        f"| {sku} | {title} | {brand} | {pj} | {df} | {shop} | {sell} | {rank} | "
-        f"{cat} | {ing} | {cc} | {prev} |"
-    )
-
-
-def _strategy_hints(
-    *,
-    cr1: float | None,
-    pst: dict[str, Any],
-    hits: Counter[str],
-    n_comments: int,
-    scen_counts: Counter[str],
-    scen_n_texts: int,
-) -> list[str]:
-    """基于规则的「提示性」结论，均标注待验证。"""
-    hints: list[str] = []
-    if cr1 is not None and cr1 >= 0.45:
-        hints.append(
-            "样本内品牌集中度较高（第一大品牌份额偏高），头部玩家占据显著曝光；新原料/解决方案宜明确差异化价值主张（**需线下渠道与招商信息交叉验证**）。"
-        )
-    elif cr1 is not None and cr1 < 0.25:
-        hints.append(
-            "样本内品牌较分散，品类或关键词下竞争格局未固化，存在定位与叙事空间（**需扩大样本页数与关键词矩阵验证**）。"
-        )
-    if pst.get("stdev") and pst.get("mean") and pst["mean"] > 0:
-        cv = pst["stdev"] / pst["mean"]
-        if cv > 0.35:
-            hints.append(
-                "价格离散度较高，同时存在偏低价与偏高价陈列，可分别对标「性价比带」与「品质/功能带」竞品（**终端到手价受促销影响，非成本结构**）。"
-            )
-    if hits:
-        top = hits.most_common(3)
-        top_s = "、".join(w for w, _ in top)
-        hints.append(
-            f"评价文本中「{top_s}」等主题出现较多，可作为消费者沟通与产品卖点的假设输入（**非严格主题模型，建议人工抽样复核**）。"
-        )
-    if n_comments < 5:
-        hints.append(
-            "有效评价样本偏少，消费者洞察部分仅作方向参考，正式结论建议加大 SKU 数或评论分页。"
-        )
-    if scen_n_texts >= 5 and scen_counts:
-        top_lbl, top_n = scen_counts.most_common(1)[0]
-        share = top_n / scen_n_texts
-        if share >= 0.25:
-            hints.append(
-                f"用途/场景中「{top_lbl}」在约 {100 * share:.0f}% 的有效评价自述中出现，可作为沟通场景与卖点的优先假设（**词组规则，建议抽样核对原句**）。"
-            )
-    if not hints:
-        hints.append(
-            "当前样本下自动规则未触发强信号；请结合业务目标人工解读对比矩阵与原始 CSV。"
-        )
-    return hints
-
-
-def _embed_chart(run_dir: Path, filename: str, caption: str = "") -> list[str]:
-    """若 ``report_assets/<filename>`` 存在则返回插图 Markdown 片段。"""
-    if not (run_dir / "report_assets" / filename).is_file():
-        return []
-    cap = (caption or "").strip()
-    out: list[str] = []
-    if cap:
-        out.append(f"*{cap}*")
-        out.append("")
-    out.append(f"![](report_assets/{filename})")
-    out.append("")
-    return out
-
-
-def _scenario_group_asset_slug(group: str, index: int) -> str:
-    """与 ``pipeline.reporting.charts`` 中场景分组图文件名规则一致（勿改格式）。"""
-    raw = (group or "").strip()
-    core = re.sub(r"[^\w\u4e00-\u9fff-]", "", raw)[:20]
-    if not core:
-        core = "group"
-    return f"i{index:02d}_{core}"
-
-
-def _focus_scenario_combo_bar_filename(group: str, index: int) -> str:
-    """关注词 + 使用场景并排条形图（与 ``pipeline.reporting.charts.save_combo_focus_scenario_bar`` 同源）。"""
-    slug = _scenario_group_asset_slug(group, index)
-    return f"chart_focus_and_scenarios_bar__{slug}.png"
-
-
-def _matrix_prices_sales_chart_filename(group: str, index: int) -> str:
-    """与 ``pipeline.reporting.charts.generate_report_charts`` 中 ``chart_matrix_prices_sales__*`` 一致。"""
-    slug = _scenario_group_asset_slug(group, index)
-    return f"chart_matrix_prices_sales__{slug}.png"
-
-
-def _lines_4_reading_brand(
-    *,
-    cr1: float | None,
-    cr3: float | None,
-    top: str,
-    brand_rows_n: int,
-    n_structure: int,
-) -> list[str]:
-    if cr1 is None or not (top or "").strip():
-        return []
-    lines = [
-        "",
-        "**数据解读（规则摘要）**：",
-        "",
-        f"- 在含品牌字段的 **{brand_rows_n}** 条列表行（占本章结构样本 **{n_structure}** 行）中，"
-        f"「{_md_cell(top.strip(), 36)}」曝光约占 **{100 * cr1:.1f}%**（按行计，同一 SKU 多行会重复计）。",
-    ]
-    if cr3 is not None:
-        lines.append(
-            f"- 前三品牌合计约 **{100 * cr3:.1f}%**；若该比例偏高，说明搜索页品牌集中度高，"
-            "新品需搭配清晰的差异定位与资源投放，避免与头部在泛词下正面撞车。"
-        )
-    lines.append("")
-    return lines
-
-
-def _lines_4_reading_shop(
-    *,
-    cr1: float | None,
-    cr3: float | None,
-    top: str,
-    shop_rows_n: int,
-    n_structure: int,
-) -> list[str]:
-    if not shop_rows_n:
-        return []
-    lines = [
-        "",
-        "**数据解读（规则摘要）**：",
-        "",
-        f"- 含店铺名的列表行共 **{shop_rows_n}** 条（结构样本 **{n_structure}** 行），反映搜索曝光下的店铺格局。",
-    ]
-    if cr1 is not None and (top or "").strip():
-        lines.append(
-            f"- 第一大店铺「{_md_cell(top.strip(), 40)}」约占 **{100 * cr1:.1f}%**；"
-            "该指标刻画的是**列表可见度**而非销量，适合用于判断货架被哪些店铺占据。"
-        )
-    if cr3 is not None:
-        lines.append(
-            f"- 前三店铺合计约 **{100 * cr3:.1f}%**；若集中度高，可考虑从店铺矩阵、旗舰店/专营店布局等角度拆解竞争。"
-        )
-    lines.append("")
-    return lines
-
 
 def build_competitor_markdown(
     *,