From b6a52c04d20e2cfb377a306d74dbcf7b247ba8a4 Mon Sep 17 00:00:00 2001 From: hub-gif <2487812171@qq.com> Date: Fri, 17 Apr 2026 13:39:57 +0800 Subject: [PATCH] =?UTF-8?q?refactor(jd):=20=E5=8F=8D=E9=A6=88=E5=88=86?= =?UTF-8?q?=E7=BB=84=E3=80=81=E5=88=97=E8=A1=A8=E7=BB=9F=E8=AE=A1=E3=80=81?= =?UTF-8?q?=E7=9F=A9=E9=98=B5=E8=A1=8C=E4=B8=8E=E6=8A=A5=E5=91=8A=E7=89=87?= =?UTF-8?q?=E6=AE=B5=E6=8B=86=E8=87=B3=20competitor=5Freport?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 consumer_feedback、run_context、list_mix、matrix_md、report_md_helpers；jd_competitor_report 从配置块后直接进入 build_competitor_markdown；移除未再使用的 ingredients 直引。 Made-with: Cursor --- .../competitor_report/consumer_feedback.py | 161 +++++ .../competitor_report/list_mix.py | 138 ++++ .../competitor_report/matrix_md.py | 87 +++ .../competitor_report/report_md_helpers.py | 202 ++++++ .../competitor_report/run_context.py | 91 +++ .../jd_pc_search/jd_competitor_report.py | 619 +----------------- 6 files changed, 713 insertions(+), 585 deletions(-) create mode 100644 backend/crawler_copy/jd_pc_search/competitor_report/consumer_feedback.py create mode 100644 backend/crawler_copy/jd_pc_search/competitor_report/list_mix.py create mode 100644 backend/crawler_copy/jd_pc_search/competitor_report/matrix_md.py create mode 100644 backend/crawler_copy/jd_pc_search/competitor_report/report_md_helpers.py create mode 100644 backend/crawler_copy/jd_pc_search/competitor_report/run_context.py diff --git a/backend/crawler_copy/jd_pc_search/competitor_report/consumer_feedback.py b/backend/crawler_copy/jd_pc_search/competitor_report/consumer_feedback.py new file mode 100644 index 0000000..39cacb7 --- /dev/null +++ b/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", +] diff --git a/backend/crawler_copy/jd_pc_search/competitor_report/list_mix.py b/backend/crawler_copy/jd_pc_search/competitor_report/list_mix.py new file mode 100644 index 0000000..2ef927f --- /dev/null +++ b/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", +] diff --git a/backend/crawler_copy/jd_pc_search/competitor_report/matrix_md.py b/backend/crawler_copy/jd_pc_search/competitor_report/matrix_md.py new file mode 100644 index 0000000..19c3943 --- /dev/null +++ b/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"] diff --git a/backend/crawler_copy/jd_pc_search/competitor_report/report_md_helpers.py b/backend/crawler_copy/jd_pc_search/competitor_report/report_md_helpers.py new file mode 100644 index 0000000..a93abbc --- /dev/null +++ b/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/`` 存在则返回插图 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", +] diff --git a/backend/crawler_copy/jd_pc_search/competitor_report/run_context.py b/backend/crawler_copy/jd_pc_search/competitor_report/run_context.py new file mode 100644 index 0000000..3bd90cc --- /dev/null +++ b/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", +] diff --git a/backend/crawler_copy/jd_pc_search/jd_competitor_report.py b/backend/crawler_copy/jd_pc_search/jd_competitor_report.py index e4ce9c3..6b30677 100644 --- a/backend/crawler_copy/jd_pc_search/jd_competitor_report.py +++ b/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/`` 存在则返回插图 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( *,