feat(demo): 第八章文本挖掘探针脚本（分词/TF-IDF/共现/LDA/可选LLM）

Made-with: Cursor

backend/pipeline/demos/chapter8_text_mining_probe.py

@@ -0,0 +1,443 @@
+"""
+第八章「文本挖掘探针」独立脚本（**不修改**主报告代码）。
+
+流程（按细类分组）：清洗（jieba 分词 + 停用词）→ 词频 / TF-IDF → 共现对 → LDA 主题
+→ 规则化叙事小结 → 可选 LLM 解读（需 ``--live-llm`` 且配置好 ``AI_crawler``）。
+
+依赖（请自行安装）::
+
+    pip install jieba scikit-learn numpy
+
+用法（在 ``backend`` 目录下）::
+
+    python -m pipeline.demos.chapter8_text_mining_probe --run-dir \"../data/JD/pipeline_runs/20260413_104252_低GI\"
+    python -m pipeline.demos.chapter8_text_mining_probe --run-dir \"...\" --out chapter8_probe.md
+    python -m pipeline.demos.chapter8_text_mining_probe --run-dir \"...\" --live-llm --max-llm-groups 2
+
+输出：默认写入 ``<run_dir>/chapter8_text_mining_probe.md``。
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import sys
+from collections import Counter
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+
+BACKEND_ROOT = Path(__file__).resolve().parents[2]
+if str(BACKEND_ROOT) not in sys.path:
+    sys.path.insert(0, str(BACKEND_ROOT))
+os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")
+
+import django  # noqa: E402
+
+django.setup()
+
+JCR_ROOT = BACKEND_ROOT / "crawler_copy" / "jd_pc_search"
+if str(JCR_ROOT) not in sys.path:
+    sys.path.insert(0, str(JCR_ROOT))
+
+import jd_competitor_report as jcr  # noqa: E402
+import jd_keyword_pipeline as kpl  # noqa: E402
+
+from pipeline.csv_schema import MERGED_FIELD_TO_CSV_HEADER  # noqa: E402
+
+try:
+    import jieba  # noqa: WPS433
+    from sklearn.decomposition import LatentDirichletAllocation  # noqa: WPS433
+    from sklearn.feature_extraction.text import (  # noqa: WPS433
+        CountVectorizer,
+        TfidfVectorizer,
+    )
+except ImportError as e:
+    print(
+        "缺少依赖，请先安装：pip install jieba scikit-learn numpy\n"
+        f"原始错误: {e}",
+        file=sys.stderr,
+    )
+    sys.exit(1)
+
+
+# 精简中文停用词（可换外部文件）；与业务无关，仅用于探针
+_STOP_BASIC: frozenset[str] = frozenset(
+    """
+    的 了 和 是 在 也 有 就 都 很 啊 还 吗 吧 呢 呀 哦 噢 哈 呵 与 及 或 等 为 被 让 从 到 把 而 又 对 中 这 那 其 一个 一些 没有 不是 可以 这样 我们 你们 他们 它们 它 会 要 能 去 来 做 用 给 自己 这个 那个 什么 怎么 如果 因为 所以 但是 而且 然后 还是 或者 还有 就是 只是 只是 已经 还是
+    非常 真的 比较 特别 感觉 觉得 认为 看到 收到 东西 商品 产品 卖家 买家 店铺 京东 物流 快递 包装 评价 评论 购买 买 卖 收到 天 次 个 款 种 条 块
+    """.split()
+)
+
+
+def _load_run(
+    run_dir: Path,
+) -> tuple[str, list[dict[str, str]], list[dict[str, str]]]:
+    run_dir = run_dir.resolve()
+    merged_path = run_dir / kpl.FILE_MERGED_CSV
+    if not merged_path.is_file():
+        raise FileNotFoundError(f"缺少合并表: {merged_path}")
+    _, merged_rows = jcr._read_csv_rows(merged_path)
+    _, comment_rows = jcr._read_csv_rows(run_dir / kpl.FILE_COMMENTS_FLAT_CSV)
+    meta_path = run_dir / kpl.FILE_RUN_META_JSON
+    meta: dict[str, Any] | None = None
+    if meta_path.is_file():
+        try:
+            meta = json.loads(meta_path.read_text(encoding="utf-8"))
+        except json.JSONDecodeError:
+            meta = None
+    kw = ""
+    if meta and str(meta.get("keyword") or "").strip():
+        kw = str(meta.get("keyword")).strip()
+    return kw, merged_rows, comment_rows
+
+
+def _cut_one(text: str) -> list[str]:
+    s = (text or "").strip()
+    if not s:
+        return []
+    raw = jieba.lcut(s)
+    out: list[str] = []
+    for w in raw:
+        w = w.strip()
+        if len(w) < 2:
+            continue
+        if w in _STOP_BASIC:
+            continue
+        if re.match(r"^[0-9\s\W_]+$", w):
+            continue
+        out.append(w)
+    return out
+
+
+def _docs_cut(texts: list[str]) -> list[str]:
+    """每条评论 → 空格连接的分词串，供 sklearn。"""
+    rows: list[str] = []
+    for t in texts:
+        toks = _cut_one(t)
+        if toks:
+            rows.append(" ".join(toks))
+    return rows
+
+
+def _term_freq_top(cut_docs: list[str], top_k: int) -> list[tuple[str, int]]:
+    c: Counter[str] = Counter()
+    for line in cut_docs:
+        for w in line.split():
+            c[w] += 1
+    return c.most_common(top_k)
+
+
+def _tfidf_top(cut_docs: list[str], top_k: int) -> list[tuple[str, float]]:
+    if not cut_docs:
+        return []
+    vec = TfidfVectorizer(max_features=min(2000, max(50, len(cut_docs) * 3)))
+    try:
+        X = vec.fit_transform(cut_docs)
+    except ValueError:
+        return []
+    feats = np.array(vec.get_feature_names_out())
+    scores = np.asarray(X.mean(axis=0)).ravel()
+    idx = np.argsort(-scores)[:top_k]
+    return [(str(feats[i]), float(scores[i])) for i in idx]
+
+
+def _cooc_top_pairs(
+    cut_docs: list[str],
+    vocab_cap: int,
+    pair_top: int,
+) -> list[tuple[str, str, int]]:
+    """同一条评论内无序词对共现（过滤低频词）。"""
+    term_freq = Counter()
+    for line in cut_docs:
+        for w in set(line.split()):
+            term_freq[w] += 1
+    top_terms = {w for w, _ in term_freq.most_common(vocab_cap)}
+    pair_c: Counter[tuple[str, str]] = Counter()
+    for line in cut_docs:
+        toks = sorted({w for w in line.split() if w in top_terms})
+        for i in range(len(toks)):
+            for j in range(i + 1, len(toks)):
+                a, b = toks[i], toks[j]
+                if a > b:
+                    a, b = b, a
+                pair_c[(a, b)] += 1
+    return [(a, b, n) for (a, b), n in pair_c.most_common(pair_top)]
+
+
+def _lda_topics(
+    cut_docs: list[str],
+    n_topics: int,
+    n_top_words: int,
+) -> tuple[list[list[str]], str]:
+    if len(cut_docs) < 4:
+        return [], "文本条数过少，跳过 LDA。"
+    n_topics = max(2, min(n_topics, len(cut_docs) // 2))
+    try:
+        vec = CountVectorizer(max_df=0.95, min_df=2, max_features=800)
+        X = vec.fit_transform(cut_docs)
+    except ValueError as e:
+        return [], f"LDA 向量化失败：{e}"
+    if X.shape[0] < 3 or X.shape[1] < 3:
+        return [], "矩阵过稀疏，跳过 LDA。"
+    lda = LatentDirichletAllocation(
+        n_components=n_topics,
+        max_iter=30,
+        learning_method="batch",
+        random_state=42,
+        n_jobs=1,
+    )
+    try:
+        lda.fit(X)
+    except Exception as e:
+        return [], f"LDA 拟合失败：{e}"
+    names = vec.get_feature_names_out()
+    topics: list[list[str]] = []
+    for topic_idx, topic in enumerate(lda.components_):
+        top_ix = np.argsort(-topic)[:n_top_words]
+        topics.append([str(names[i]) for i in top_ix])
+    return topics, ""
+
+
+def _md_escape(s: str) -> str:
+    return (s or "").replace("|", "\\|").replace("\n", " ")
+
+
+def _narrative_stub(
+    n_raw: int,
+    n_cut: int,
+    tf_top: list[tuple[str, int]],
+    tfidf_top: list[tuple[str, float]],
+    cooc: list[tuple[str, str, int]],
+    lda_topics: list[list[str]],
+    lda_note: str,
+) -> str:
+    lines: list[str] = [
+        f"本细类有效评论约 **{n_cut}** 条（原始非空 **{n_raw}** 条，经分词去停用后用于建模）。",
+        "",
+        "**词频 Top**："
+        + (
+            "、".join(f"「{w}」({n})" for w, n in tf_top[:12])
+            if tf_top
+            else "（无）"
+        )
+        + "。",
+        "",
+        "**TF-IDF 加权 Top**（相对区分度）："
+        + (
+            "、".join(f"「{w}」({s:.3f})" for w, s in tfidf_top[:12])
+            if tfidf_top
+            else "（无）"
+        )
+        + "。",
+        "",
+        "**共现较强的词对**（同条评论内，供联想维度）："
+        + (
+            "；".join(f"「{a}」-「{b}」({c})" for a, b, c in cooc[:10])
+            if cooc
+            else "（无）"
+        )
+        + "。",
+        "",
+    ]
+    if lda_note:
+        lines.append(f"*LDA：{lda_note}*")
+        lines.append("")
+    elif lda_topics:
+        lines.append("**LDA 主题（无监督，仅作探索）**：")
+        for i, words in enumerate(lda_topics):
+            lines.append(f"- 主题 {i + 1}：{'、'.join(words)}")
+        lines.append("")
+    lines.append(
+        "> 以上为主题探索与统计摘要，**不等同**于业务结论；若与星级、规则词表冲突，以人工抽样为准。"
+    )
+    return "\n".join(lines)
+
+
+def _maybe_llm_block(
+    texts: list[str],
+    scores: list[int | None] | None,
+    keyword: str,
+    live: bool,
+) -> str:
+    if not live:
+        return (
+            "> **LLM 解读**：未启用（请加 ``--live-llm``；需本机 ``AI_crawler`` 等与大模型调用环境可用）。"
+        )
+    try:
+        from pipeline.llm.generate import generate_comment_sentiment_analysis_llm  # noqa: WPS433
+    except Exception as e:
+        return f"> **LLM 解读**：导入失败：{e}"
+    if len(texts) < 2:
+        return "> **LLM 解读**：有效评论不足 2 条，跳过。"
+    try:
+        pl = jcr.build_comment_sentiment_llm_payload(
+            texts,
+            scores=scores,
+            max_samples_positive=10,
+            max_samples_negative=12,
+            max_samples_mixed=6,
+            semantic_pool_max=24,
+            max_chars_per_review=320,
+            shuffle_seed=keyword or "probe",
+        )
+        pl["keyword"] = keyword
+        pl["probe_note"] = "chapter8_text_mining_probe 脚本生成，非生产流水线。"
+        body = generate_comment_sentiment_analysis_llm(pl)
+    except Exception as e:
+        return f"> **LLM 解读**调用失败：{e}"
+    return "#### LLM 深入解读（探针）\n\n" + body.strip()
+
+
+def build_markdown(
+    run_dir: Path,
+    *,
+    min_texts: int,
+    lda_topics_n: int,
+    top_k_words: int,
+    cooc_vocab: int,
+    cooc_pairs: int,
+    live_llm: bool,
+    max_llm_groups: int,
+) -> str:
+    kw, merged, comments = _load_run(run_dir)
+    sku_h = MERGED_FIELD_TO_CSV_HEADER["sku_id"]
+    groups = jcr._consumer_feedback_by_matrix_group(
+        merged_rows=merged,
+        comment_rows=comments,
+        sku_header=sku_h,
+    )
+    # 与正文同序评分，便于 LLM payload
+    all_texts, all_scores = jcr._iter_comment_text_units_and_scores(comments, merged)
+    text_to_score: dict[str, int | None] = dict(zip(all_texts, all_scores))
+
+    lines: list[str] = [
+        "# 八、消费者反馈与用户画像（文本挖掘探针 · 实验稿）",
+        "",
+        f"- **运行目录**：`{run_dir}`",
+        f"- **监测词（run_meta）**：{kw or '—'}",
+        f"- **生成脚本**：`pipeline.demos.chapter8_text_mining_probe`",
+        "",
+        "## 8.0 说明",
+        "",
+        "本稿为**独立探针**，流程参考「清洗 → 词频/TF-IDF → 共现 → LDA → 叙事小结 →（可选）LLM」。"
+        "与线上一致的部分：**细类划分与 SKU 归因**复用 ``jd_competitor_report._consumer_feedback_by_matrix_group``；"
+        "其余为 **jieba + sklearn** 的开放词表分析，**不替代**正式报告中的规则统计。",
+        "",
+        "---",
+        "",
+    ]
+
+    llm_used = 0
+    for gname, _cr_rows, texts in groups:
+        n_raw = len([t for t in texts if (t or "").strip()])
+        if n_raw < min_texts:
+            lines.extend(
+                [
+                    f"## {gname}",
+                    "",
+                    f"*本细类有效文本 {n_raw} 条，低于 ``--min-texts``={min_texts}，跳过。*",
+                    "",
+                    "---",
+                    "",
+                ]
+            )
+            continue
+
+        cut_docs = _docs_cut(texts)
+        if len(cut_docs) < 2:
+            lines.extend(
+                [
+                    f"## {gname}",
+                    "",
+                    "*分词后不足 2 条，跳过。*",
+                    "",
+                    "---",
+                    "",
+                ]
+            )
+            continue
+
+        tf_top = _term_freq_top(cut_docs, top_k_words)
+        tfidf_top = _tfidf_top(cut_docs, top_k_words)
+        cooc = _cooc_top_pairs(cut_docs, cooc_vocab, cooc_pairs)
+        lda_t, lda_err = _lda_topics(cut_docs, lda_topics_n, 12)
+
+        lines.extend([f"## {gname}", ""])
+        lines.append(_narrative_stub(
+            n_raw,
+            len(cut_docs),
+            tf_top,
+            tfidf_top,
+            cooc,
+            lda_t,
+            lda_err,
+        ))
+        lines.extend(["", "---", ""])
+
+        if live_llm and llm_used < max_llm_groups:
+            sub_scores = [text_to_score.get(t) for t in texts]
+            block = _maybe_llm_block(texts, sub_scores, kw, live=True)
+            lines.append(block)
+            lines.extend(["", "---", ""])
+            llm_used += 1
+    if not live_llm:
+        lines.append("")
+        lines.append(_maybe_llm_block([], None, kw, live=False))
+
+    lines.append("")
+    lines.append("*（完）*")
+    return "\n".join(lines)
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser(description="第八章文本挖掘探针（独立脚本）")
+    ap.add_argument(
+        "--run-dir",
+        type=Path,
+        required=True,
+        help="pipeline_runs 下某批次目录（含 keyword_pipeline_merged.csv、comments_flat.csv）",
+    )
+    ap.add_argument(
+        "--out",
+        type=Path,
+        default=None,
+        help="输出 Markdown 路径（默认 <run_dir>/chapter8_text_mining_probe.md）",
+    )
+    ap.add_argument("--min-texts", type=int, default=8, help="细类最少评论条数才分析")
+    ap.add_argument("--lda-topics", type=int, default=4, help="LDA 主题数上限（会按样本量裁剪）")
+    ap.add_argument("--top-k-words", type=int, default=30, help="词频/TF-IDF 展示长度")
+    ap.add_argument("--cooc-vocab", type=int, default=80, help="共现矩阵保留的高频词数")
+    ap.add_argument("--cooc-pairs", type=int, default=25, help="输出词对数量")
+    ap.add_argument(
+        "--live-llm",
+        action="store_true",
+        help="对前若干个细类调用 §8.2 同款 LLM（需环境可用）",
+    )
+    ap.add_argument(
+        "--max-llm-groups",
+        type=int,
+        default=1,
+        help="最多对几个细类调用 LLM（避免费用与时间）",
+    )
+    args = ap.parse_args()
+
+    md = build_markdown(
+        args.run_dir,
+        min_texts=args.min_texts,
+        lda_topics_n=args.lda_topics,
+        top_k_words=args.top_k_words,
+        cooc_vocab=args.cooc_vocab,
+        cooc_pairs=args.cooc_pairs,
+        live_llm=args.live_llm,
+        max_llm_groups=max(0, args.max_llm_groups),
+    )
+    out = args.out or (args.run_dir.resolve() / "chapter8_text_mining_probe.md")
+    out.write_text(md, encoding="utf-8")
+    print(f"已写入: {out}", flush=True)
+
+
+if __name__ == "__main__":
+    main()

backend/requirements.txt

@@ -8,3 +8,7 @@ python-docx>=1.1
 reportlab>=4.0
 matplotlib>=3.8
 playwright>=1.40
+
+# 可选：pipeline.demos.chapter8_text_mining_probe（第八章文本挖掘探针）
+# jieba>=0.42
+# scikit-learn>=1.4