
Xhs Topic Scout
- 6 installs
- 5 repo stars
- Updated April 7, 2026
- cyhzzz/finance_aigc_skills
Researches Xiaohongshu finance topics by collecting hot news and evaluating real notes to produce 5-10 actionable topics and angles.
About
This skill researches finance topics for Xiaohongshu by gathering hot news and assessing real notes via a browser session, yielding actionable topics and angles. A securities or fund account creator uses it to plan Xiaohongshu content.
- Combines hot-news collection with real Xiaohongshu note sentiment
- Requires the web-access skill and a logged-in browser
Xhs Topic Scout by the numbers
- 6 all-time installs (skills.sh)
- Ranked #1,559 of 1,879 Marketing & SEO skills by installs in the Skillselion catalog
- Data as of Jul 29, 2026 (Skillselion catalog sync)
npx skills add https://github.com/cyhzzz/finance_aigc_skills --skill xhs-topic-scoutAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 6 |
|---|---|
| repo stars | ★ 5 |
| Last updated | April 7, 2026 |
| Repository | cyhzzz/finance_aigc_skills ↗ |
What it does
Researches Xiaohongshu finance topics by collecting hot news and evaluating real notes to produce 5-10 actionable topics and angles.
Files
小红书财经选题调研工具
依赖说明
必需依赖
| 依赖 | 用途 | 检查/启动方式 |
|---|---|---|
| web-access skill | CDP浏览器操控 + 小红书调研 | 首次使用需安装 |
| Chrome/Edge 浏览器 | 携带用户登录态访问小红书 | 自动尝试启动 |
web-access skill 安装
首次使用前必须安装。执行任务前先检查:
# 检查 web-access skill 是否已安装
ls ~/.claude/skills/web-access/SKILL.md 2>/dev/null && echo "已安装" || echo "未安装"判断逻辑:
- 若已安装 → 直接进入"依赖检查与启动流程"
- 若未安装 → 引导用户说:"帮我安装 web-access skill,仓库地址是 https://github.com/eze-is/web-access"
安装后说明:该 skill 原为 Claude Code 设计,安装后需理解其核心原理和工作逻辑,结合当前环境进行适配使用。
依赖检查与启动流程
按顺序执行,每步检查结果:
# Step 1: 检查 CDP Proxy 是否运行,未运行则自动启动
curl -s http://localhost:3456/health || \
node ~/.claude/skills/web-access/scripts/cdp-proxy.mjs &
# Step 2: 检查是否已连接到浏览器
curl -s http://localhost:3456/health
# 期望: {"connected": true, "sessions": N} (N > 0)
# Step 3: 若未连接,尝试自动启动 Chrome
# Windows:
start chrome --remote-debugging-port=9222
# 或 Edge:
start msedge --remote-debugging-port=9222
# Step 4: 等待连接后再次检查
sleep 3 && curl -s http://localhost:3456/health
# Step 5: 若仍失败,检查是否有浏览器进程
# Windows:
tasklist | findstr /i "chrome edge"降级策略
| 依赖失效 | 降级方案 |
|---|---|
| Chrome/Edge 无法启动 | 提示用户手动启动:chrome --remote-debugging-port=9222 |
| CDP 仍未连接 | 使用 WebFetch/curl 抓取财经新闻作为备选 |
| 小红书无法访问 | 基于行业知识评估竞争度,标注"待验证" |
| 所有外部源均失败 | 提示用户提供近期市场概况,生成基于模拟的选题报告 |
---
技能定位
为申万宏源证券官方/KOS账号提供小红书选题参考,核心目标是引导客户关注和咨询,获取线索。
核心工作流
Phase 1: 热点采集
→ 财联社深度 + 新浪财经7×24 + 东方财富操盘必读
Phase 2: 选题发散
→ 从热点中提取候选话题(10-15个)
→ 热点分类 + 候选话题生成
Phase 3: 小红书市场调研
→ CDP真实检索,评估竞争度和内容缺口
Phase 4: 选题分析
→ 选题评估框架 + 避坑指南 + 切入角度
Phase 5: 输出选题报告 + 热点连续性评估
→ 5-10个选题,含新闻来源、小红书现状、切入思路
→ 评估近期热点连续性(持续热点 = 换角度持续聊)---
Phase 1: 热点采集
数据来源(按优先级)
| 来源 | 栏目 | URL |
|---|---|---|
| 财联社 | 深度 | https://www.cls.cn/depth?id=1000 |
| 新浪财经 | 7×24全球直播 | https://finance.sina.com.cn/7x24/ |
| 东方财富 | 操盘必读 | https://finance.eastmoney.com/a/cdfsd_2.html |
采集命令
通过CDP打开上述页面,获取页面文本内容:
# 财联社
curl -s "http://localhost:3456/new?url=https://www.cls.cn/depth?id=1000"
# 新浪财经
curl -s "http://localhost:3456/new?url=https://finance.sina.com.cn/7x24/"
# 东方财富
curl -s "http://localhost:3456/new?url=https://finance.eastmoney.com/a/cdfsd_2.html"热点提取原则
从页面文本中提取:
- 与A股/投资/理财相关的话题
- 有具体数据或事件
- 有大众关注度(阅读量、讨论度)
具体新闻URL提取要求
必须为每个热点话题提取其来源页面的具体URL:
1. 东方财富:快讯链接格式为 https://finance.eastmoney.com/a/xxx,可直接提取 ✅ 2. 财联社:动态渲染页面,返回纯文本无URL。使用格式:财联社电报 + 时间 + 阅读量 作为来源标识 3. 新浪7×24:动态页面,无具体URL。使用格式:新浪财经7×24 + 时间 作为来源标识
禁止:
- ❌ 只写栏目首页(如
https://www.cls.cn/depth?id=1000) - ❌ 只写来源名称(如"财联社深度")
正确示例:
| 东方财富 | https://finance.eastmoney.com/a/202603213679644262.html | 宇树科技IPO... |
| 财联社 | 财联社电报 13:46 阅6869 | A股下跌... |
| 新浪财经 | 新浪财经7×24 13:45 | 市场动态... |---
Phase 2: 选题发散
将热点分为5大类,对应不同选题角度:
| 类别 | 选题角度 |
|---|---|
| 行情/指数 | 解读分析、操作建议 |
| 板块/概念 | 板块机会、ETF/基金推荐 |
| 产品/事件 | 投资逻辑、产品分析 |
| 国际/宏观 | 影响分析、避险策略 |
| 公司/个股 | 基本面分析、投资逻辑 |
关键词由候选话题决定,不要预设。根据Phase 1提取的真实热点中的关键词来设计选题角度。
候选话题生成
从热点中提取10-15个候选话题,原则:
- 与资本市场/投资相关
- 有社会关注度
- 小红书用户关心的话题
最终输出要求:必须精选5-10个选题,不足5个则需补充热点,不足10个也需达到5个。
---
Phase 3: 小红书市场调研(CDP)
调研目标
对于每个候选话题,在小红书搜索并评估: 1. 话题阶段:萌芽期 / 爆发期 / 成熟期 / 衰退期 2. 竞争激烈度:高 / 中 / 低 3. 内容缺口:哪些角度已有人做,哪些是机会
CDP操作流程
Step 1: 构造搜索URL 小红书搜索URL格式:https://www.xiaohongshu.com/search_result?keyword={关键词}&type=51 其中关键词需要URL编码,如"黄金" → %E9%BB%84%E9%87%91
Step 2: 创建CDP Tab并等待加载
curl -s "http://localhost:3456/new?url=https://www.xiaohongshu.com/search_result?keyword=%E9%BB%84%E9%87%91&type=51"返回的targetId用于后续操作。
Step 3: 等待页面完全加载
sleep 5 # 等待JS渲染完成Step 4: 获取笔记列表数据
curl -s -X POST "http://localhost:3456/eval?target={targetId}" \
-d 'JSON.stringify({
notes: Array.from(document.querySelectorAll(".note-item, .search-result-item")).slice(0, 15).map(el => ({
title: el.querySelector(".title, .note-content-title")?.textContent?.trim() || el.textContent?.trim()?.slice(0, 60),
time: el.textContent?.match(/(\d+分钟前|\d+小时前|\d+天前|\d+月前)/)?.[0] || "未知",
likes: el.querySelector(".liked-count, .like-count")?.textContent?.trim() || "0"
}))
})'Step 5: 如果Step 4返回空或错误,尝试备选方案
# 方案A: 等待更长时间后重试
sleep 5 && curl -s -X POST "http://localhost:3456/eval?target={targetId}" \
-d 'document.body.innerText.slice(0, 3000)'
# 方案B: 截图查看页面状态
curl -s "http://localhost:3456/screenshot?target={targetId}&file=/tmp/xhs_search.png"Step 6: 提取数据后关闭Tab
curl -s "http://localhost:3456/close?target={targetId}"常见问题处理
| 问题 | 原因 | 解决方案 |
|---|---|---|
| 返回targetId为Infinity | URL编码问题或页面跳转 | 检查关键词URL编码是否正确 |
| 笔记列表为空 | JS未渲染完成 | 增加sleep时间,或使用截图确认页面状态 |
| 页面加载超时 | 网络问题 | 重试或跳过该话题 |
话题阶段判断
| 阶段 | 特征 | 建议 |
|---|---|---|
| 萌芽期 | 新笔记少,互动不高 | 提前布局,适合首发 |
| 爆发期 | 新笔记爆发,互动激增 | 赶紧跟进,适合热点 |
| 成熟期 | 大量笔记,互动趋稳 | 差异化角度切入 |
| 衰退期 | 新笔记减少,老内容为主 | 换话题 |
竞争度评估
| 评估 | 信号 |
|---|---|
| 高 | 多个100+互动笔记,标题党严重 |
| 中 | 有几个30-50互动笔记 |
| 低 | 笔记少,互动低,投资角度空白 |
---
Phase 4: 选题分析
选题评估框架
双维度评分:
| 维度 | 评估标准 | 满分 |
|---|---|---|
| 新闻热度 | 今日阅读量、讨论度 | /10 |
| 小红书机会 | 竞争度低、内容缺口大 | /10 |
总分 = 新闻热度 + 小红书机会
| 总分 | 评级 | 建议 |
|---|---|---|
| 16-20 | ⭐⭐⭐⭐⭐ 强烈推荐 | 今天发最好 |
| 12-15 | ⭐⭐⭐⭐ 推荐 | 尽快发布 |
| 8-11 | ⭐⭐⭐ 一般 | 需要差异化 |
| <8 | ⭐⭐ 谨慎 | 不建议做 |
避坑指南 & 切入角度设计
详细指南:references/选题方法论.md(含5大核心方法论、标题四关键词、避坑指南)
引导咨询策略
每个选题结尾需要设计:
- 互动引导:"你现在仓位如何?"、"亏了多少?"
- 咨询引导:"想了解更多,可以私信咨询"
- 关注引导:"关注我,持续跟踪..."
---
Phase 5: 输出选题报告
报告结构
# 小红书选题调研报告
**调研日期**:
**调研目的**:为申万宏源证券官方/KOS账号提供选题参考
**核心诉求**:引导客户关注和咨询,获取线索
---
## 一、热点财经新闻来源
| 来源 | 具体新闻链接 | 核心热点 |
|------|------------|---------|
| 财联社 | https://www.cls.cn/telegraph/xxx | ... |
| 新浪财经 | https://finance.sina.com.cn/7x24/xxx | ... |
| 东方财富 | https://finance.eastmoney.com/a/xxx | ... |
**重要**:链接必须是具体新闻页URL,不能只写栏目首页。
---
## 二、热点财经新闻分析
### 核心热点排序
| 排名 | 热点话题 | 热度指数 | 与A股关联度 |
|------|---------|---------|-------------|
### 重点话题分析
#### 【话题A】XXX
**新闻背景**:
- 来源:[链接]
- 核心数据/事件
**核心矛盾/亮点**:
---
## 三、小红书发帖情况分析
来源:[小红书搜索链接]
### 话题A:XXX
| 验证项 | 结果 |
|--------|------|
| **发帖时间** | X小时内 / X天前 |
| **笔记数量** | 多/中/少 |
| **互动情况** | 最高X互动 |
| **内容角度** | XXX类多,XXX类少 |
| **竞争激烈度** | 高/中/低 |
---
## 四、选题建议(5-10个)⭐
**重要**:最终输出必须包含**5-10个选题**,不多不少。若候选话题不足,需返回热点采集阶段补充。
### 选题一:XXX
| 项目 | 内容 |
|------|------|
| **新闻来源** | [来源] |
| **小红书现状** | [搜索链接] |
| **选题角度** | ... |
| **切入思路** | ... |
| **差异化点** | ... |
---
## 五、执行建议
### 内容形式
| 选题 | 建议形式 | 理由 |
|------|---------|------|
### 引导咨询策略
- **互动引导**:
- **咨询引导**:
- **风险提示**:
---
**报告生成时间**:
**数据来源**:财联社、新浪财经、东方财富、小红书搜索---
快速执行命令
CDP命令参考:references/cdp-commands.md
输出位置
选题报告输出到:./output/选题调研报告_{YYYYMMDD}.md(相对路径,skill目录下)
---
热点连续性评估(Phase 5 附)
目的
评估选题在近期是否持续出现,判断热点生命周期。
检查方法
# 列出最近7天的选题报告
ls -t ./output/ | head -7
# 读取最近报告内容
cat ./output/选题调研报告_YYYYMMDD.md评估逻辑
| 情况 | 含义 | 处理方式 |
|---|---|---|
| 首次出现 | 新话题 | 优先推荐 |
| 间歇出现 | 周期性话题 | 正常推荐 |
| 持续出现 | 热点持续发酵 | 换角度持续聊,在报告中标注 🔥 热点持续中,建议换角度切入 |
热点连续性标注
在每个选题后添加标注:
| 标注 | 含义 |
|---|---|
🆕 新话题 | 首次出现 |
🔄 周期性话题 | 间歇出现 |
🔥 热点持续中 | 连续2次以上出现,建议换角度切入 |
📈 热点爆发 | 连续出现且热度上升,强烈推荐 |
注意事项
⚠️ 热点连续性标记仅作为参考,不作为排除依据。持续热点意味着市场关注度高,是内容消费的黄金期,应换不同角度持续产出。
---
注意事项
1. 投资关联性为核心:每个选题都必须可以自然转向证券开户或基金投资 2. 固定输出5-10个选题:不多不少,确保质量。若候选不足,需返回热点采集阶段补充 3. 投资关联度 ≥80分:低于80分的选题会被排除 4. 合规性:不包含市场预测、荐股信息 5. 时效性:热点话题优先,今天的新闻今天发最好
{
"skill_name": "xhs-topic-scout",
"evals": [
{
"id": 1,
"prompt": "帮我做一个小红书财经选题调研,主题是今天的A股市场热点",
"expected_output": "输出5-10个选题报告,包含新闻来源、小红书现状分析、切入角度",
"files": []
},
{
"id": 2,
"prompt": "我需要一个证券账号的小红书选题,主要关注黄金和ETF投资方向",
"expected_output": "输出5-10个关于黄金和ETF的选题,包含差异化切入思路",
"files": []
},
{
"id": 3,
"prompt": "今天有什么热点可以做成小红书内容?帮我调研一下适合基金理财账号的选题",
"expected_output": "输出5-10个基金理财相关选题,带有避坑指南和风险提示",
"files": []
}
]
}
xhs-topic-scout
小红书财经选题调研工具
功能
为证券/基金/理财账号提供小红书选题参考,输出5-10个可执行的选题。
核心工作流
Phase 1: 热点采集(财联社 + 新浪财经 + 东方财富)
Phase 2: 选题发散(提取候选话题)
Phase 3: 小红书市场调研(CDP真实检索)
Phase 4: 选题分析(评估框架 + 避坑指南)
Phase 5: 输出选题报告快速开始
1. 确保 CDP Proxy 运行中:
node ~/.claude/skills/web-access/scripts/cdp-proxy.mjs2. 执行 Python 脚本抓取热点:
cd D:/project/skills/xhs-topic-scout/scripts
python fetch_hot_topics.py3. 根据 SKILL.md 引导进行小红书调研
文件结构
xhs-topic-scout/
├── SKILL.md # 主技能文件
├── README.md # 说明文件
├── scripts/
│ └── fetch_hot_topics.py # 热点抓取脚本
└── references/
└── 选题方法论.md # 选题方法论参考输出
选题报告保存到:./output/选题调研报告_{YYYYMMDD}.md(相对路径)
CDP 操作命令参考
热点采集命令
# 财联社深度
curl -s "http://localhost:3456/new?url=https://www.cls.cn/depth?id=1000"
# 新浪财经7×24
curl -s "http://localhost:3456/new?url=https://finance.sina.com.cn/7x24/"
# 东方财富操盘必读
curl -s "http://localhost:3456/new?url=https://finance.eastmoney.com/a/cdfsd_2.html"小红书搜索命令
# 搜索话题(type=51获取笔记)
curl -s "http://localhost:3456/new?url=https://www.xiaohongshu.com/search_result?keyword={关键词}&type=51"
# 获取笔记列表(标题+时间)
curl -s -X POST "http://localhost:3456/eval?target={targetId}" \
-d 'JSON.stringify({
notes: Array.from(document.querySelectorAll(".note-item")).slice(0,12).map(el => ({
title: el.querySelector(".title span")?.textContent?.trim() || el.textContent?.trim()?.slice(0,50),
time: el.textContent?.match(/(\d+分钟前|\d+小时前|\d+天前)/)?.[0] || "未知"
}))
})'
# 点击具体笔记获取完整互动数据
curl -s "http://localhost:3456/new?url=https://www.xiaohongshu.com/explore/{笔记ID}"
# 获取笔记详情
curl -s -X POST "http://localhost:3456/eval?target={targetId}" \
-d 'JSON.stringify({
title: document.querySelector("h1")?.textContent?.trim() || document.title,
liked: document.querySelectorAll(".count")[0]?.textContent?.trim() || "",
collected: document.querySelectorAll(".count")[1]?.textContent?.trim() || "",
commented: document.querySelectorAll(".count")[2]?.textContent?.trim() || "",
content: document.querySelector(".note-content")?.textContent?.trim()?.slice(0,500) || ""
})'健康检查
# 检查CDP连接
curl -s http://localhost:3456/health财经选题方法论
核心目标
从"不炒股的普通人" → "有投资兴趣的潜在客户"
---
5大核心方法论
1. 痛点驱动
从生活痛点切入,而非专业概念
❌ 错误:"如何理解通货膨胀" ✅ 正确:"为什么你的钱越来越不值钱了?"
原理:
- 大众不关心专业概念
- 只关心对自己的影响
- 痛点是更强的吸引力
2. 情绪共鸣
引发情绪,而非单纯传递信息
❌ 错误:"黄金价格上涨20%" ✅ 正确:"5年前买黄金,现在赚196万!你错过了吗?"
情绪类型:
- 焦虑:房价、养老、就业
- 希望:赚钱、副业、投资
- 好奇:富豪生活、内幕揭秘
- 愤怒:不公平、被割韭菜
3. 数据冲击
用数据说话,增加可信度
❌ 错误:"黄金是很好的投资" ✅ 正确:"2700克金条卖出,赚196万!年化收益18%!"
数据要素:
- 具体数字(不要"很多",要"196万")
- 对比数据(5年前 vs 现在)
- 可视化(图表、对比)
4. 故事表达
用故事讲道理,而非说教
❌ 错误:"定投的重要性" ✅ 正确:"打工30年攒300万,靠的不是工资,而是..."
故事要素:
- 主角(普通人更好)
- 冲突(困难、挑战)
- 转折(发现、改变)
- 结果(成功、收获)
5. 实用导向
给出可操作建议,而非泛泛而谈
❌ 错误:"投资要分散风险" ✅ 正确:"普通人如何投资黄金?这3种方式最靠谱"
实用要素:
- 具体步骤(第一步、第二步、第三步)
- 避坑指南(不要做什么)
- 工具推荐(用什么)
---
选题公式
公式1:数字+冲突+结果
模板: [具体数字] + [冲突/对比] + [结果]
案例:
- "月入200万的AI生意,普通人能做吗?"
- "5年前买黄金,现在赚196万!"
- "公司利润2.7亿,年终奖1.8亿!你的呢?"
公式2:痛点+解决方案
模板: [痛点] + [如何解决]
案例:
- "工资不够花?这个副业让你月入过万"
- "存钱30年不如投资3年?"
- "年终奖发了,怎么理财才能不贬值?"
公式3:故事+启示
模板: [故事] + [启示/教训]
案例:
- "返乡大哥打工30年攒300万,秘诀竟然是..."
- "我朋友炒股亏了50万,原因是..."
公式4:对比+结论
模板: [A vs B] + [结论]
案例:
- "存钱 vs 投资,哪个更划算?"
- "买房 vs 买黄金,现在该选哪个?"
---
标题四关键词测试
好的标题包含以下4个关键词中的至少2个:
| 关键词 | 触发词 |
|---|---|
| 好奇 | "为什么"、"怎么"、"揭秘"、"讲明白" |
| 悬念 | "结果却"、"没想到"、"但是"、"最后" |
| 独特 | "第一次"、"独家"、"罕见"、"只有" |
| 颠覆 | "颠覆认知"、"打破常识"、"反直觉"、"竟然" |
测试方法:
- 输入标题
- 统计包含几个关键词
- <2个 = 需要优化
---
避坑指南
❌ 禁止项
- 市场预测("会涨到"、"要跌到")
- 荐股信息("推荐买入"、"目标价XXX")
- 卖点堆砌(>3个核心卖点)
- 纯数据堆砌(无情绪价值)
- 虚假互动(刷赞、刷收藏)
✅ 必需项
- 有明确人群定位("上班族"、"月光族")
- 有冲突/反转/共鸣
- 有具体场景或案例
- 有数据支撑或专业背书
- 有开放性问题引导互动
---
选题评估标准
1. 传播性(40%)
评估维度:
- 是否有情绪共鸣?
- 是否有数据冲击?
- 是否有故事性?
- 标题是否吸引人?
2. 相关性(30%)
评估维度:
- 与大众生活相关吗?
- 能自然转向投资吗?
- 目标受众明确吗?
3. 实用性(20%)
评估维度:
- 能给出具体建议吗?
- 有可操作性吗?
- 有避坑指南吗?
4. 合规性(10%)
评估维度:
- 不触碰敏感话题?
- 不传播虚假信息?
- 可以加入风险提示?
---
小红书平台适配
选题公式
结构:适用人群 + 效果显著 + 场景化 + 体验 + 背书 + 对比
标题模板:
《人群 + 效果 + 强情绪词》
例:《学生党必看!低风险理财!月入3000也能存钱》内容要素
1. 商品/方法名称完整露出 2. 亲身体验(口语化) 3. 场景化应用(3种场景) 4. 专业背书(数据、术语) 5. 对比分析(性价比)
验证指标
- 收藏率 > 5%(核心指标,说明有用)
- 点赞率(可能虚假)
- 评论率(看评论质量)
---
引导咨询话术
互动引导
- "你现在仓位如何?亏了多少?"
- "手里的基金现在怎么样了?"
- "你买过黄金吗?赚了多少?"
咨询引导
- "想了解更多ETF操作,可以私信咨询"
- "想布局XX板块,可以扫码开户"
- "有问题可以评论区聊聊"
关注引导
- "关注我,持续跟踪XX话题"
- "下期讲XX,点击关注不迷路"
风险提示
以上内容仅为个人观点,不构成投资建议。市场有风险,投资需谨慎。
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
财经热点抓取工具
通过 CDP Proxy API 获取热点新闻页面内容
依赖:
- web-access skill (CDP Proxy)
- Python 3.x
降级策略:
- 如果CDP不可用,可以手动复制页面内容进行分析
- 如果Python脚本不可用,可以直接使用CDP命令获取热点
"""
import json
import os
import subprocess
import time
import re
import urllib.request
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional
class HotTopicsFetcher:
"""热点抓取器 - 基于CDP"""
CDP_BASE_URL = os.environ.get("CDP_PROXY_URL", "http://localhost:3456")
# 数据来源配置(均为财经新闻源,无需关键词筛选)
SOURCES = {
"cls": {
"name": "财联社",
"url": "https://www.cls.cn/depth?id=1000"
},
"sina": {
"name": "新浪财经",
"url": "https://finance.sina.com.cn/7x24/"
},
"eastmoney": {
"name": "东方财富",
"url": "https://finance.eastmoney.com/a/cdfsd_2.html"
}
}
def __init__(self):
self.targets = {}
def check_health(self) -> bool:
"""检查CDP Proxy健康状态"""
try:
with urllib.request.urlopen(f"{self.CDP_BASE_URL}/health", timeout=5) as resp:
data = json.loads(resp.read().decode())
return data.get("connected", False)
except Exception:
return False
def create_tab(self, url: str) -> Optional[str]:
"""创建新tab并返回targetId"""
try:
with urllib.request.urlopen(f"{self.CDP_BASE_URL}/new?url={url}", timeout=15) as resp:
data = json.loads(resp.read().decode())
return data.get("targetId")
except Exception as e:
print(f"[FAIL] create tab: {e}")
return None
def get_page_text(self, target_id: str, max_length: int = 5000) -> str:
"""获取页面文本内容"""
try:
# 等待页面加载
time.sleep(3)
# 执行JS获取页面文本
req = urllib.request.Request(
f"{self.CDP_BASE_URL}/eval?target={target_id}",
data=b'document.body?.innerText?.slice(0, 5000) || ""',
headers={"Content-Type": "application/x-www-form-urlencoded"}
)
with urllib.request.urlopen(req, timeout=15) as resp:
data = json.loads(resp.read().decode())
return data.get("value", "")
except Exception as e:
print(f"[FAIL] get page text: {e}")
return ""
def close_tab(self, target_id: str):
"""关闭tab"""
try:
with urllib.request.urlopen(f"{self.CDP_BASE_URL}/close?target={target_id}", timeout=5):
pass
except Exception:
pass
def extract_news(self, text: str, source_name: str) -> List[Dict]:
"""从页面文本中提取新闻"""
news_items = []
# 按行分割,过滤有效行
lines = text.split('\n')
current_item = {}
for line in lines:
line = line.strip()
if not line:
continue
# 时间模式匹配 (10:44, 10:43:27等)
time_match = re.match(r'^(\d{1,2}:\d{2}(?::\d{2})?)$', line)
if time_match and current_item.get('text'):
# 保存上一个条目
if current_item.get('text'):
current_item['time'] = time_match.group(1)
news_items.append(current_item)
current_item = {'text': '', 'time': time_match.group(1)}
continue
# 阅读量模式匹配
read_match = re.search(r'([\d.]+[万万亿]?\s*阅读)', line)
if read_match:
current_item['read_count'] = read_match.group(1)
# 收集文本内容
if line and not line.startswith('财联社') and not line.startswith('新浪财经'):
current_item['text'] = (current_item.get('text', '') + ' ' + line).strip()
# 保存最后一个条目
if current_item.get('text'):
news_items.append(current_item)
return news_items
def filter_by_keywords(self, news: List[Dict], keywords: List[str]) -> List[Dict]:
"""根据关键词过滤新闻"""
filtered = []
for item in news:
text = item.get('text', '')
if any(kw in text for kw in keywords):
filtered.append(item)
return filtered
def fetch_all_sources(self) -> Dict:
"""抓取所有来源的新闻"""
results = {}
print("[START] Fetching hot topics...")
for source_id, config in self.SOURCES.items():
print(f"\n[FETCH] {config['name']}...")
# 创建tab
target_id = self.create_tab(config['url'])
if not target_id:
print(f"[FAIL] {config['name']}")
continue
self.targets[source_id] = target_id
# 获取页面文本
text = self.get_page_text(target_id)
if not text:
print(f"[FAIL] {config['name']} - empty content")
continue
# 提取新闻(财经新闻源,无需关键词筛选)
news = self.extract_news(text, config['name'])
results[source_id] = {
"name": config['name'],
"url": config['url'],
"all_count": len(news),
"news": news[:20] # 最多保留20条
}
print(f"[OK] {config['name']}: {len(news)} news items")
# 请求间隔
time.sleep(1)
return results
def save_to_file(self, data: Dict, output_path: str = "/tmp") -> str:
"""保存数据到JSON文件"""
Path(output_path).mkdir(parents=True, exist_ok=True)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"hot_topics_{timestamp}.json"
filepath = Path(output_path) / filename
output_data = {
"fetch_time": datetime.now().isoformat(),
"sources_count": len(data),
"total_news": sum(s.get("all_count", 0) for s in data.values()),
"data": data
}
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(output_data, f, ensure_ascii=False, indent=2)
print(f"\n[SAVE] {filepath}")
return str(filepath)
def print_summary(self, data: Dict):
"""打印摘要"""
print("\n" + "="*60)
print("[SUMMARY] Hot Topics")
print("="*60)
for source_id, source_data in data.items():
print(f"\n[{source_data['name']}]")
for i, news in enumerate(source_data.get('news', [])[:5], 1):
text = news.get('text', '')[:60]
print(f" {i}. {text}...")
print("\n" + "="*60)
def cleanup(self):
"""清理所有tab"""
for target_id in self.targets.values():
self.close_tab(target_id)
def main():
import argparse
parser = argparse.ArgumentParser(
description='财经热点抓取工具 - 通过CDP获取财联社/新浪/东财热点',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
数据来源:
- 财联社深度: https://www.cls.cn/depth?id=1000
- 新浪财经7×24: https://finance.sina.com.cn/7x24/
- 东方财富操盘必读: https://finance.eastmoney.com/a/cdfsd_2.html
示例:
python fetch_hot_topics.py # 默认保存到 /tmp
python fetch_hot_topics.py --output ./data # 保存到 ./data
python fetch_hot_topics.py --cdp-url http://localhost:3456 # 指定CDP地址
"""
)
parser.add_argument('--output', default='/tmp', help='输出目录 (default: /tmp)')
parser.add_argument('--cdp-url', default=None, help='CDP Proxy地址')
args = parser.parse_args()
fetcher = HotTopicsFetcher()
if args.cdp_url:
fetcher.CDP_BASE_URL = args.cdp_url
# 检查CDP连接
if not fetcher.check_health():
print("[FAIL] CDP Proxy not connected. Start with: node ~/.claude/skills/web-access/scripts/cdp-proxy.mjs")
return
print("[OK] CDP Proxy connected")
# 抓取数据
data = fetcher.fetch_all_sources()
if not data:
print("[FAIL] No data fetched")
return
# 保存
filepath = fetcher.save_to_file(data, args.output)
# 打印摘要
fetcher.print_summary(data)
# 清理
fetcher.cleanup()
if __name__ == "__main__":
main()