全网搜索采集师 - MCP/RSS/浏览器搜索三合一

本文重点

️ 数据采集 Agent MCP/RSS/浏览器搜索三合一全网采集 用 OpenClaw 构建自动化数据采集 Agent，整合三种采集方式，实现全网信息监控。 --- 采集架构 ┌─────────────────┐ ...

目录

🕷️ 数据采集 Agent

MCP/RSS/浏览器搜索三合一全网采集

用 OpenClaw 构建自动化数据采集 Agent，整合三种采集方式，实现全网信息监控。

---

<## id="section-0">📊 采集架构</##>

                    ┌─────────────────┐
                    │  数据采集 Agent  │
                    └────────┬────────┘
                             │
        ┌────────────────────┼────────────────────┐
        │                    │                    │
        ▼                    ▼                    ▼
   ┌─────────┐         ┌─────────┐         ┌─────────┐
   │  MCP    │         │   RSS   │         │ 浏览器  │
   │  工具   │         │  订阅   │         │  搜索   │
   └────┬────┘         └────┬────┘         └────┬────┘
        │                   │                   │
        ▼                   ▼                   ▼
   API 接口            新闻网站            搜索引擎
   数据库              博客 RSS            社交媒体

---

<## id="section-1">🔧 第 1 步：MCP 工具采集</##>

<### id="section-2">什么是 MCP？</###>

MCP (Model Context Protocol) 是 OpenClaw 的工具调用协议，可以调用各种 API 获取数据。

<### id="section-3">配置 MCP 工具</###>

# 1. 查看可用 MCP 工具
mcporter list
2. 配置 WebSearch MCP（阿里百炼）

mcporter call WebSearch.bailian_web_search query="AI 最新进展" count=10
3. 配置代码解释器（数据处理）

mcporter call code_interpreter_mcp.execute_code code="import pandas as pd; ..."

<### id="section-4">数据采集指令模板</###>

你是一个数据采集专家，请使用 MCP 工具帮我采集以下信息：
【采集目标】
• 主题：[填写主题，如"AI 大模型 2026 年 3 月更新"]

• 时间范围：[如"最近 7 天"]

• 数量要求：[如"20-30 条"]
【采集步骤】
1. 使用 WebSearch 搜索关键词
2. 提取标题、链接、发布时间、摘要
3. 去重并整理为表格
4. 输出 Markdown 格式
【输出格式】
| 标题 | 来源 | 时间 | 链接 |
|------|------|------|------|
| ...  | ...  | ...  | ...  |

<### id="section-5">自动化采集脚本</###>

# ~/Desktop/workScript/python/projects/bvm_collect/mcp_collector.py
MCP 自动化采集脚本
import subprocess
import json
from datetime import datetime
class MCPDataCollector:
    def __init__(self):
        self.results = []
    
    def search(self, query, count=10):
        """使用 MCP WebSearch 搜索"""
        cmd = f'mcporter call WebSearch.bailian_web_search query="{query}" count={count}'
        result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
        return json.loads(result.stdout)
    
    def collect(self, keywords):
        """批量采集"""
        for keyword in keywords:
            print(f"采集：{keyword}")
            data = self.search(keyword)
            self.results.extend(data.get('pages', []))
        return self.results
    
    def save(self, filename):
        """保存结果"""
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(self.results, f, ensure_ascii=False, indent=2)
        print(f"已保存 {len(self.results)} 条数据到 {filename}")
使用示例

if __name__ == "__main__":
    collector = MCPDataCollector()
    keywords = ["OpenClaw 更新", "AI Agent 最新", "大模型发布"]
    results = collector.collect(keywords)
    collector.save(f"data/mcp_collect_{datetime.now().strftime('%Y%m%d')}.json")

---

<## id="section-6">📰 第 2 步：RSS 订阅采集</##>

<### id="section-7">配置 RSS 订阅源</###>

【推荐 RSS 订阅源】
科技新闻：
• CSDN: https://blog.csdn.net/rss/list

• 知乎日报：https://daily.zhihu.com/rss

• 36Kr: https://36kr.com/feed
AI 专项：
• OpenAI Blog: https://openai.com/blog/rss

• Anthropic: https://www.anthropic.com/rss

• Google AI: https://ai.google/rss

<### id="section-8">RSS 采集指令</###>

你是一个 RSS 内容分析专家，请帮我处理以下 RSS 订阅内容：
【RSS 源】
• [粘贴 RSS 链接或内容]
【处理要求】
1. 提取最近 10 篇文章
2. 筛选与"AI Agent"相关的内容
3. 提取标题、链接、发布时间、摘要
4. 按时间倒序排列
【输出格式】
<## id="section-9">📰 [日期] AI Agent 相关新闻
<### id="section-10">1. [标题]
• 来源：[网站名]

• 时间：[发布时间]

• 摘要：[100 字以内]

• 链接：[URL]

<### id="section-11">RSS 监控脚本</###>

# ~/Desktop/workScript/python/projects/bvm_collect/rss_monitor.py
RSS 监控脚本
import feedparser
import json
from datetime import datetime, timedelta
class RSSMonitor:
    def __init__(self):
        self.feeds = []
    
    def add_feed(self, url, name):
        """添加 RSS 订阅源"""
        self.feeds.append({'url': url, 'name': name})
    
    def fetch_all(self, days=7):
        """获取最近 N 天的内容"""
        all_entries = []
        cutoff_date = datetime.now() - timedelta(days=days)
        
        for feed in self.feeds:
            print(f"抓取：{feed['name']}")
            fp = feedparser.parse(feed['url'])
            for entry in fp.entries:
                try:
                    published = datetime(*entry.published_parsed[:6])
                    if published >= cutoff_date:
                        all_entries.append({
                            'source': feed['name'],
                            'title': entry.title,
                            'link': entry.link,
                            'published': published.isoformat(),
                            'summary': entry.summary[:200] if hasattr(entry, 'summary') else ''
                        })
                except:
                    continue
        
        return sorted(all_entries, key=lambda x: x['published'], reverse=True)
    
    def save(self, entries, filename):
        """保存结果"""
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(entries, f, ensure_ascii=False, indent=2)
        print(f"已保存 {len(entries)} 条内容")
使用示例

if __name__ == "__main__":
    monitor = RSSMonitor()
    monitor.add_feed("https://36kr.com/feed", "36Kr")
    monitor.add_feed("https://blog.csdn.net/rss/list", "CSDN")
    entries = monitor.fetch_all(days=7)
    monitor.save(entries, f"data/rss_monitor_{datetime.now().strftime('%Y%m%d')}.json")

---

<## id="section-12">🌐 第 3 步：浏览器搜索采集</##>

<### id="section-13">使用 browser 工具采集</###>

你是一个数据采集专家，请使用 browser 工具帮我采集信息：
【采集目标】
• 网站：[如"https://www.huxiu.com/"]

• 主题：[如"AI 相关文章"]

• 数量：[如"10 篇"]
【采集步骤】
1. 访问目标网站
2. 搜索关键词或浏览相关分类
3. 提取文章标题、链接、发布时间、摘要
4. 整理为表格
【注意事项】
• 注意反爬机制，控制访问频率

• 提取内容要完整

• 链接要绝对路径

<### id="section-14">浏览器自动化脚本</###>

# ~/Desktop/workScript/python/projects/bvm_collect/browser_collector.py
浏览器自动化采集脚本
from playwright.sync_api import sync_playwright
import json
from datetime import datetime
class BrowserCollector:
    def __init__(self, headless=True):
        self.headless = headless
        self.results = []
    
    def collect(self, url, search_query=None):
        """采集网页内容"""
        with sync_playwright() as p:
            browser = p.chromium.launch(headless=self.headless)
            page = browser.new_page()
            
            # 访问页面
            print(f"访问：{url}")
            page.goto(url, wait_until='networkidle')
            
            # 如果有搜索词，执行搜索
            if search_query:
                search_box = page.query_selector('input[type="search"], input[name="q"]')
                if search_box:
                    search_box.fill(search_query)
                    search_box.press('Enter')
                    page.wait_for_load_state('networkidle')
            
            # 提取文章列表（根据实际网站结构调整选择器）
            articles = page.query_selector_all('article, .post, .news-item')
            for article in articles[:10]:  # 限制数量
                try:
                    title_el = article.query_selector('h1, h2, h3, a.title')
                    link_el = article.query_selector('a')
                    time_el = article.query_selector('time, .date, .publish-time')
                    summary_el = article.query_selector('p, .summary, .excerpt')
                    
                    if title_el and link_el:
                        self.results.append({
                            'title': title_el.inner_text().strip(),
                            'link': link_el.get_attribute('href'),
                            'time': time_el.inner_text().strip() if time_el else '',
                            'summary': summary_el.inner_text().strip()[:200] if summary_el else ''
                        })
                except Exception as e:
                    continue
            
            browser.close()
        
        return self.results
    
    def save(self, filename):
        """保存结果"""
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(self.results, f, ensure_ascii=False, indent=2)
        print(f"已保存 {len(self.results)} 条数据")
使用示例

if __name__ == "__main__":
    collector = BrowserCollector(headless=True)
    collector.collect("https://www.huxiu.com/", search_query="AI")
    collector.save(f"data/browser_collect_{datetime.now().strftime('%Y%m%d')}.json")

---

<## id="section-15">🤖 三合一采集 Agent</##>

<### id="section-16">整合三种采集方式</###>

你是一个全网数据采集 Agent，请整合以下三种方式采集信息：
【采集任务】
• 主题：[如"2026 年 3 月 AI 大模型进展"]

• 时间范围：[如"最近 7 天"]

• 目标数量：[如"50 条"]
【采集方式】
1. MCP 工具：搜索 20 条
   - 使用 WebSearch MCP
   - 关键词：[关键词列表]
2. RSS 订阅：监控 10 个来源
   - RSS 源：[RSS 链接列表]
   - 筛选相关主题
3. 浏览器采集：补充 20 条
   - 目标网站：[网站列表]
   - 搜索关键词
【去重与整理】
1. 合并三种方式的结果
2. 按标题去重
3. 按时间倒序排列
4. 输出 Markdown 表格
【输出格式】
<## id="section-17">📊 数据采集报告
采集时间：[日期]
采集主题：[主题]
数据来源：MCP [X]条 + RSS [X]条 + 浏览器 [X]条
| 标题 | 来源 | 时间 | 类型 | 链接 |
|------|------|------|------|------|
| ...  | ...  | ...  | ...  | ...  |

<### id="section-18">完整采集工作流</###>

# ~/Desktop/workScript/python/projects/bvm_collect/universal_collector.py
三合一通用采集器
import json
from datetime import datetime
from mcp_collector import MCPDataCollector
from rss_monitor import RSSMonitor
from browser_collector import BrowserCollector
class UniversalDataCollector:
    def __init__(self):
        self.mcp = MCPDataCollector()
        self.rss = RSSMonitor()
        self.browser = BrowserCollector(headless=True)
        self.all_data = []
    
    def collect_all(self, topic, keywords, rss_feeds, websites):
        """整合三种方式采集"""
        print(f"=== 开始采集：{topic} ===")
        
        # 1. MCP 采集
        print("\n【1/3】MCP 工具采集...")
        mcp_data = self.mcp.collect(keywords)
        print(f"MCP 采集：{len(mcp_data)} 条")
        
        # 2. RSS 采集
        print("\n【2/3】RSS 订阅采集...")
        for url, name in rss_feeds:
            self.rss.add_feed(url, name)
        rss_data = self.rss.fetch_all(days=7)
        print(f"RSS 采集：{len(rss_data)} 条")
        
        # 3. 浏览器采集
        print("\n【3/3】浏览器采集...")
        browser_data = []
        for website in websites:
            self.browser.collect(website, search_query=topic)
            browser_data.extend(self.browser.results)
        print(f"浏览器采集：{len(browser_data)} 条")
        
        # 合并数据
        self.all_data = mcp_data + rss_data + browser_data
        print(f"\n总计：{len(self.all_data)} 条")
        
        return self.all_data
    
    def deduplicate(self):
        """去重"""
        seen_titles = set()
        unique_data = []
        for item in self.all_data:
            title = item.get('title', '')
            if title not in seen_titles:
                seen_titles.add(title)
                unique_data.append(item)
        self.all_data = unique_data
        print(f"去重后：{len(self.all_data)} 条")
        return self.all_data
    
    def save_report(self, topic):
        """生成报告"""
        filename = f"data/report_{topic}_{datetime.now().strftime('%Y%m%d')}.md"
        with open(filename, 'w', encoding='utf-8') as f:
            f.write(f"# 📊 数据采集报告\n\n")
            f.write(f"采集时间：{datetime.now().strftime('%Y-%m-%d %H:%M')}\n")
            f.write(f"采集主题：{topic}\n")
            f.write(f"数据总量：{len(self.all_data)} 条\n\n")
            f.write(f"## 数据列表\n\n")
            for i, item in enumerate(self.all_data, 1):
                f.write(f"### {i}. {item.get('title', '无标题')}\n")
                f.write(f"- 来源：{item.get('source', '未知')}\n")
                f.write(f"- 时间：{item.get('published', item.get('time', '未知'))}\n")
                f.write(f"- 链接：{item.get('link', '无链接')}\n\n")
        print(f"报告已保存：{filename}")
使用示例

if __name__ == "__main__":
    collector = UniversalDataCollector()
    
    # 配置采集参数
    topic = "AI 大模型"
    keywords = ["GPT-5", "Claude 4", "Gemini 3"]
    rss_feeds = [
        ("https://36kr.com/feed", "36Kr"),
        ("https://blog.csdn.net/rss/list", "CSDN")
    ]
    websites = [
        "https://www.huxiu.com/",
        "https://36kr.com/"
    ]
    
    # 执行采集
    collector.collect_all(topic, keywords, rss_feeds, websites)
    collector.deduplicate()
    collector.save_report(topic)

---

<## id="section-19">📋 实战案例：监控 OpenClaw 更新</##>

<### id="section-20">配置采集任务</###>

【采集任务】监控 OpenClaw 最新动态
【MCP 关键词】
• OpenClaw 更新

• OpenClaw 发布

• OpenClaw version
【RSS 订阅源】
• OpenClaw GitHub: https://github.com/openclaw/openclaw/releases.atom

• CSDN OpenClaw: https://so.csdn.net/so/search/rss.do?q=OpenClaw
【浏览器采集】
• https://github.com/openclaw/openclaw

• https://docs.openclaw.ai
【采集频率】每天上午 9 点
【输出】Markdown 报告 + JSON 数据

<### id="section-21">定时任务配置</###>

# Crontab 配置（每天 9 点执行）
0 9   * cd ~/Desktop/workScript/python/projects/bvm_collect && python3 universal_collector.py >> logs/collector.log 2>&1

---

<## id="section-22">🎯 总结</##>

<### id="section-23">三种采集方式对比</###>

| 方式 | 优点 | 缺点 | 适用场景 |
|------|------|------|---------|
| MCP 工具 | 快速、结构化 | 依赖 API | 搜索引擎、数据库 |
| RSS 订阅 | 实时、稳定 | 需要网站支持 RSS | 新闻网站、博客 |
| 浏览器采集 | 灵活、全面 | 速度慢、易被反爬 | 动态网站、社交媒体 |

<### id="section-24">最佳实践</###>

1. 优先使用 MCP - 快速、稳定
2. RSS 做补充 - 监控固定来源
3. 浏览器兜底 - 采集动态内容
4. 定期去重 - 避免重复数据
5. 保存原始数据 - 便于回溯分析

---

📅 2026-03-31 整理

OCnote Agent 教学 - 数据采集实战系列

核心要点

选对模型，事半功倍。选错模型，花钱买气受。适合 = 能力匹配 + 预算允许 + 用得顺手

上一篇
文案撰写专家
下一篇
数据采集专家

相关主题

后端架构师
6 0
可爱型文案专家
8 0
诙谐型文案专家
11 0
严谨型文案专家
4 0
文案撰写专家
6 0