@clawhub-lisayinyy-1cdf5bef2b
小红书舆论监控 - 自动搜索小红书帖子、筛选需要舆论引导的内容、生成多人设评论话术、输出到飞书多维表格。适用于任何品牌/产品的小红书舆情监控。Use when: (1) 需要监控小红书上的品牌/产品讨论, (2) 需要生成多角色评论话术, (3) 需要将监控结果输出到飞书表格, (4) 用户提到舆论监控、舆情分析...
---
name: xiaohongshu-public-monitor
description: "小红书舆论监控 - 自动搜索小红书帖子、筛选需要舆论引导的内容、生成多人设评论话术、输出到飞书多维表格。适用于任何品牌/产品的小红书舆情监控。Use when: (1) 需要监控小红书上的品牌/产品讨论, (2) 需要生成多角色评论话术, (3) 需要将监控结果输出到飞书表格, (4) 用户提到舆论监控、舆情分析、小红书评论。"
metadata:
author: Lisayinyy
version: 1.0.0
language: zh-CN
platform: xiaohongshu
requires:
- playwright
- feishu-bitable
---
# 小红书舆论监控 Skill
自动监控小红书帖子 → 筛选内容 → 生成评论话术 → 写入飞书多维表格。
适用于任何行业：AI、电商、美妆、餐饮、教育……只需修改 `config.yaml` 即可适配。
## 快速开始
### 1. 安装
```bash
# Via ClawHub
clawdhub install xiaohongshu-public-monitor
# 手动安装
git clone https://github.com/Lisayinyy/xiaohongshu-public-monitor.git ~/.openclaw/skills/xiaohongshu-public-monitor
# 安装依赖
pip install playwright
playwright install chromium
```
### 2. 登录小红书（仅首次）
```bash
python3 scripts/xhs_search.py login
```
弹出浏览器，扫码登录一次，后续自动复用。
### 3. 配置
编辑 `config.yaml`，填入：
- 你的品牌名和产品
- 搜索关键词
- 飞书多维表格 token
- 3个评论人设（根据你的行业定义）
### 4. 运行
```bash
# 搜索
python3 scripts/xhs_search.py batch "关键词1" "关键词2" --scroll 5
```
或让 Agent 执行完整工作流（搜索 → 筛选 → 生成评论 → 写入表格）。
---
## 工作流程（4步）
### Step 1：搜索
- 按 `config.yaml` 中的关键词搜索小红书
- 筛选「一天内」+ 排序「最多点赞」
- Playwright 持久化登录，headless 模式，支持滚动翻页
### Step 2：去重
- 查飞书表格已有链接，相同链接跳过
### Step 3：筛选
**基础门槛（3条必满足）**：
1. 链接能打开，标题内容对得上
2. 作者不为空
3. 互动数据完整（👍点赞 ⭐收藏 💬评论）
**内容质量（满足任一入表）**：
- 直接提及你的品牌或产品
- 同行业多个品牌/产品对比
- 榜单、排名、测评类内容
- 价格/性价比讨论
- 高互动且有争议讨论
**不入表**：
- 与你的业务完全无关
- 纯招聘/面试
- 互动极低且无对比内容
- 评论区已有足够正面声音
### Step 4：生成评论 + 写入表格
- 为每篇帖子生成 3人设 × 2条 = 6条评论
- 判定紧急程度（🔴高/🟡中/🟢低）
- 按「发现日期从新到旧 → 紧急程度高到低」排序写入飞书表格
---
## 评论人设
在 `config.yaml` 中定义 3 个人设。根据你的行业选择最合适的角色组合。
**选人设的原则**：选你的目标用户群里最有说服力的 3 种人。
示例：
| 行业 | 人设1 | 人设2 | 人设3 |
|------|-------|-------|-------|
| AI/科技 | 技术专家 | 普通用户 | 产品经理 |
| 美妆 | 资深化妆师 | 学生党 | 成分党博主 |
| 餐饮 | 美食博主 | 附近居民 | 餐饮同行 |
| 电商 | 老买家 | 第一次买的人 | 行业从业者 |
| 教育 | 在读学员 | 家长 | 教育行业人士 |
每个人设需要定义：
- **特征**：这个人是谁
- **风格**：怎么说话（口语化/专业/情绪化）
- **目的**：为什么要用这个人设评论
---
## 评论红线
- ❌ 不写广告语
- ❌ 不贬低竞品（只客观对比）
- ❌ 不虚构数据
- ❌ 多条评论风格不能雷同
- ❌ 负面帖下不强行洗白
---
## 紧急程度判定
**🔴 高**：发布<24h + 有互动 + 对比/榜单类 + 负面倾向
**🟡 中**：有传播潜力 + 缺正面声音 + 评价中性
**🟢 低**：正面帖 + 互动稳定 + 已有正面声音
---
## 表格字段
| 字段 | 说明 |
|------|------|
| 文章标题 | 帖子标题 |
| 文章链接 | 原文链接 |
| 作者 | 作者昵称（不能为空）|
| 发表时间 | 文章发布时间 |
| 发现日期 | 抓取日期 |
| 互动数据 | 👍⭐💬 三项齐全 |
| 关键词命中 | 命中的搜索词 |
| 提及竞品 | 提到的竞品 |
| 文章倾向 | 正面/中性/负面 |
| 紧急程度 | 🔴高/🟡中/🟢低 |
| 干预理由 | 为什么干预 |
| 人设评论 | 每人设2条 |
| 建议水军数量 | 投入账号数 |
| 执行状态 | 待处理/进行中/已完成 |
---
## 文件结构
```
xiaohongshu-public-monitor/
├── SKILL.md # 使用说明（本文件）
├── config.yaml # 配置文件（用户修改这个）
├── scripts/
│ └── xhs_search.py # 小红书搜索脚本
├── references/
│ └── workflow.md # 工作流说明
├── assets/
└── .learnings/ # Agent 学习记录
```
FILE:config.yaml
# ================================================
# 小红书舆论监控 - 配置文件
# 修改此文件适配你的品牌和行业
# ================================================
# 你的品牌
brand:
name: "你的品牌名"
products:
- "产品1"
- "产品2"
# 搜索关键词（建议 3-5 个，每周根据热搜调整）
keywords:
- "品牌名"
- "行业词 对比"
- "行业词 测评"
- "行业词 排名"
# 搜索设置
search:
time_filter: "一天内"
sort_by: "最多点赞"
scroll_times: 5
# 飞书多维表格
feishu:
app_token: ""
table_id: ""
# 竞品列表
competitors:
- "竞品1"
- "竞品2"
- "竞品3"
# ================================================
# 评论人设（3个）
# 原则：选你的目标用户群里最有说服力的 3 种人
#
# 示例（按行业）：
# AI/科技：技术专家 / 普通用户 / 产品经理
# 美妆：资深化妆师 / 学生党 / 成分党博主
# 餐饮：美食博主 / 附近居民 / 餐饮同行
# 电商：老买家 / 新买家 / 行业从业者
# 教育：在读学员 / 家长 / 教育行业人士
# ================================================
personas:
- name: "人设1名称"
traits: "这个人是谁，什么背景"
style: "怎么说话（口语化/专业/情绪化）"
purpose: "为什么用这个人设评论"
- name: "人设2名称"
traits: "这个人是谁，什么背景"
style: "怎么说话"
purpose: "为什么用这个人设评论"
- name: "人设3名称"
traits: "这个人是谁，什么背景"
style: "怎么说话"
purpose: "为什么用这个人设评论"
# ================================================
# 可引用数据（可选）
# 如果你的产品有第三方评测数据，填在这里
# Agent 生成评论时会引用这些数据
# ================================================
reference_data: []
# 示例：
# - dimension: "好评率"
# source: "大众点评"
# value: "4.8分"
# - dimension: "销量排名"
# source: "天猫"
# value: "品类第3"
FILE:references/workflow.md
# 完整工作流说明
## 每日执行流程
1. **搜索**：按关键词搜索小红书，筛选一天内+按点赞排序
2. **去重**：对比飞书表格已有链接，跳过重复
3. **质检**：逐条打开链接验证（标题对得上、作者不为空、互动三项齐全）
4. **筛选**：按内容质量规则判断入表/不入表
5. **生成评论**：每篇帖子生成 3人设×2条=6条评论
6. **判定紧急度**：🔴高/🟡中/🟢低
7. **写入表格**：按发现日期从新到旧 → 紧急程度高到低排序
## 执行团队操作
1. 打开表格，从上往下处理
2. 每篇选 2-3 条评论分散发布
3. 不同人设用不同账号
4. 同一帖子评论间隔 ≥30分钟
5. 发完更新状态为「已完成」
FILE:scripts/xhs_search.py
#!/usr/bin/env python3
"""
小红书搜索脚本 - 基于 Playwright 持久化浏览器
特点：
- 用浏览器内置登录态，登录一次永久有效
- 支持滚动翻页，突破22条限制
- 支持时间筛选+排序
- headless 模式，不弹窗
用法：
# 首次登录（会弹窗，扫码后关闭）
python3 xhs_search_playwright.py login
# 搜索（headless，不弹窗）
python3 xhs_search_playwright.py search "MiniMax" --scroll 5
# 批量搜索多个关键词
python3 xhs_search_playwright.py batch "MiniMax" "大模型 对比" "AI模型 测评" "国产AI 排名" --scroll 5
"""
import json, sys, time, argparse, datetime, os
from playwright.sync_api import sync_playwright
# 浏览器数据目录（持久化登录态）
BROWSER_DATA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "browser_data")
def get_context(playwright, headless=True):
"""获取持久化浏览器上下文"""
os.makedirs(BROWSER_DATA_DIR, exist_ok=True)
# 始终用完整 Chrome（不是 headless-shell），保证 browser_data 兼容
chrome_path = "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome"
context = playwright.chromium.launch_persistent_context(
BROWSER_DATA_DIR,
headless=headless,
executable_path=chrome_path,
viewport={"width": 1440, "height": 900},
user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
args=["--disable-blink-features=AutomationControlled"]
)
return context
def do_login():
"""打开浏览器让用户扫码登录（有窗口）"""
print("🔐 打开浏览器，请扫码登录小红书...", file=sys.stderr)
with sync_playwright() as p:
context = get_context(p, headless=False)
page = context.new_page()
page.goto("https://www.xiaohongshu.com", wait_until="domcontentloaded", timeout=30000)
print(" 浏览器已打开，请在页面上扫码登录", file=sys.stderr)
print(" 登录成功后，在终端按回车继续...", file=sys.stderr)
input()
context.close()
print("✅ 登录完成，后续搜索将自动使用此登录态", file=sys.stderr)
def close_popups(page):
"""关闭小红书的各种弹窗"""
try:
page.evaluate("""() => {
// 移除遮罩层
document.querySelectorAll('.reds-mask, [class*="mask"]').forEach(el => el.remove());
// 点击关闭按钮
document.querySelectorAll('[aria-label="关闭"], .close-button, [class*="close-icon"]').forEach(btn => {
try { btn.click(); } catch(e) {}
});
// 移除弹窗容器
document.querySelectorAll('[class*="modal"], [class*="dialog"], [class*="popup"]').forEach(el => el.remove());
}""")
except:
pass
def extract_notes(page):
"""从页面提取笔记数据"""
return page.evaluate("""() => {
const results = [];
const seen = new Set();
// 获取所有笔记链接
const links = document.querySelectorAll('a[href*="/explore/"], a[href*="/search_result/"]');
links.forEach(link => {
try {
const href = link.href || '';
const match = href.match(/explore\\/([a-f0-9]{24})/);
if (!match) return;
const feedId = match[1];
if (seen.has(feedId)) return;
seen.add(feedId);
// 找到包含此链接的笔记卡片
const card = link.closest('section') || link.closest('[class*="note"]') || link.parentElement;
if (!card) return;
// 提取标题
const titleEl = card.querySelector('.title, span.title, .note-title, a .title');
const title = titleEl ? titleEl.textContent.trim() : '';
// 提取作者
const authorEl = card.querySelector('.author-wrapper .name, .author .name, [class*="author"] .name, .name');
const author = authorEl ? authorEl.textContent.trim() : '';
// 提取点赞数（在 like-wrapper 下的 span.count）
const likeEl = card.querySelector('.like-wrapper .count, span.count');
let likesText = likeEl ? likeEl.textContent.trim() : '0';
if (title || author) {
results.push({
feed_id: feedId,
title: title,
author: author,
likes_text: likesText,
url: 'https://www.xiaohongshu.com/explore/' + feedId
});
}
} catch(e) {}
});
return results;
}""")
def search_keyword(context, keyword, scroll_times=5, time_filter="一天内", sort_by="最多点赞"):
"""搜索单个关键词"""
results = []
seen_ids = set()
page = context.new_page()
try:
# 通过 URL 参数设置排序（time_descending=按最新，likes_descending=按点赞）
sort_map = {"综合": "general_v2", "最新": "time_descending", "最多点赞": "likes_descending"}
sort_param = sort_map.get(sort_by, "time_descending")
url = f"https://www.xiaohongshu.com/search_result?keyword={keyword}&source=web_search_result_notes&sort={sort_param}&ext_flags=1"
print(f"\n🔍 搜索: {keyword} | {sort_by} | {time_filter}", file=sys.stderr)
page.goto(url, wait_until="domcontentloaded", timeout=30000)
time.sleep(5)
# 关闭弹窗
close_popups(page)
time.sleep(1)
# 检查登录状态
content = page.content()
if "登录后查看搜索结果" in content:
print("❌ 未登录，请先运行: python3 xhs_search_playwright.py login", file=sys.stderr)
page.close()
return []
# 滚动并收集
for i in range(scroll_times + 1):
close_popups(page)
cards = extract_notes(page)
new_count = 0
for card in cards:
fid = card.get("feed_id", "")
if fid and fid not in seen_ids:
seen_ids.add(fid)
# 解析点赞数
likes_text = card.get("likes_text", "0")
try:
if "万" in likes_text:
likes = int(float(likes_text.replace("万", "")) * 10000)
elif likes_text.isdigit():
likes = int(likes_text)
else:
likes = 0
except:
likes = 0
# 解码发布时间
try:
ts = int(fid[:8], 16)
publish_dt = datetime.datetime.fromtimestamp(ts).strftime("%m-%d %H:%M")
publish_ts_ms = ts * 1000
except:
publish_dt = ""
publish_ts_ms = 0
results.append({
"feed_id": fid,
"title": card.get("title", ""),
"author": card.get("author", ""),
"likes": likes,
"publish_dt": publish_dt,
"publish_ts": publish_ts_ms,
"url": card.get("url", ""),
"keyword": keyword
})
new_count += 1
print(f" 滚动 {i}/{scroll_times} | +{new_count} 新帖 | 累计 {len(results)} 条", file=sys.stderr)
if i < scroll_times:
page.evaluate("window.scrollBy(0, 1500)")
time.sleep(2.5)
print(f" ✅ {keyword}: 共 {len(results)} 条", file=sys.stderr)
except Exception as e:
print(f" ❌ 搜索失败: {e}", file=sys.stderr)
finally:
page.close()
return results
def batch_search(keywords, scroll_times=5, time_filter="一天内", sort_by="最多点赞", headless=True):
"""批量搜索多个关键词"""
all_results = []
with sync_playwright() as p:
context = get_context(p, headless=headless)
for kw in keywords:
results = search_keyword(context, kw, scroll_times, time_filter, sort_by)
all_results.extend(results)
time.sleep(3)
context.close()
# 去重
seen = set()
unique = []
for r in all_results:
if r["feed_id"] not in seen:
seen.add(r["feed_id"])
unique.append(r)
# 按点赞排序
unique.sort(key=lambda x: x["likes"], reverse=True)
print(f"\n{'='*50}", file=sys.stderr)
print(f"📊 总计: {len(all_results)} 条 → 去重后 {len(unique)} 条", file=sys.stderr)
return unique
def main():
parser = argparse.ArgumentParser(description="小红书搜索 (Playwright 持久化浏览器)")
subparsers = parser.add_subparsers(dest="command")
# login 命令
subparsers.add_parser("login", help="扫码登录（仅首次需要）")
# search 命令
search_p = subparsers.add_parser("search", help="搜索单个关键词")
search_p.add_argument("keyword", help="搜索关键词")
search_p.add_argument("--scroll", type=int, default=5, help="滚动次数（默认5）")
search_p.add_argument("--time", default="一天内", help="时间筛选")
search_p.add_argument("--sort", default="最多点赞", help="排序方式")
search_p.add_argument("--no-headless", action="store_true", help="显示浏览器")
# batch 命令
batch_p = subparsers.add_parser("batch", help="批量搜索多个关键词")
batch_p.add_argument("keywords", nargs="+", help="关键词列表")
batch_p.add_argument("--scroll", type=int, default=5, help="滚动次数")
batch_p.add_argument("--time", default="一天内", help="时间筛选")
batch_p.add_argument("--sort", default="最多点赞", help="排序方式")
batch_p.add_argument("--no-headless", action="store_true", help="显示浏览器")
args = parser.parse_args()
if args.command == "login":
do_login()
elif args.command == "search":
with sync_playwright() as p:
context = get_context(p, headless=not args.no_headless)
results = search_keyword(context, args.keyword, args.scroll, args.time, args.sort)
context.close()
print(json.dumps(results, ensure_ascii=False, indent=2))
elif args.command == "batch":
results = batch_search(args.keywords, args.scroll, args.time, args.sort, headless=not args.no_headless)
print(json.dumps(results, ensure_ascii=False, indent=2))
else:
parser.print_help()
if __name__ == "__main__":
main()