hc/MND-IA

Fork 0

hc 8b0eadc876 首次提交

2025-12-31 19:58:09 +08:00

6.5 KiB

Raw Permalink Blame History

Skill F: 新闻联播爬虫使用指南

功能说明

Skill F (Crawler) 使用 cloudscraper + BeautifulSoup 自动抓取新闻联播官方网站的文字稿，并转换为符合 Skill A (Analyst) 要求的格式。

技术特点

绕过防护: 使用 cloudscraper 自动处理 Cloudflare 等反爬机制
稳定可靠: 同步爬取，无需处理异步复杂性
精准解析: 基于实际网站 DOM 结构（.content-section）

数据源

网站: https://cn.govopendata.com/xinwenlianbo/
更新频率: 每日 19:00 后更新
数据格式: 新闻标题 + 正文内容

安装依赖

pip install cloudscraper beautifulsoup4

基础使用

1. 命令行测试

# 测试爬虫功能
python skills/crawler.py

2. Python 代码调用

from skills.crawler import XinwenLianboCrawler

# 创建爬虫实例
crawler = XinwenLianboCrawler()

# 爬取今日新闻
news_list = crawler.crawl_sync()

# 爬取指定日期
news_list = crawler.crawl_sync(date="20251229")

# 输出格式：
# [
#   {"title": "新闻标题1", "text": "新闻正文1..."},
#   {"title": "新闻标题2", "text": "新闻正文2..."},
#   ...
# ]

3. 批量爬取

from skills.crawler import XinwenLianboCrawler

crawler = XinwenLianboCrawler()

# 爬取一段时间范围的新闻
results = crawler.crawl_date_range(
    start_date="20251220",
    end_date="20251229"
)

# results 是一个字典: {date: news_list}
for date, news_list in results.items():
    print(f"{date}: {len(news_list)} 条新闻")

与 Skill A 集成

from skills.crawler import XinwenLianboCrawler
from skills.analyst import NewsAnalyzer

# 1. 爬取新闻
crawler = XinwenLianboCrawler()
news_list = crawler.crawl_sync()

# 2. 分析新闻
analyzer = NewsAnalyzer()
analysis_results = analyzer.batch_analyze(news_list)

# 3. 生成叙事 JSON
narrative_json = analyzer.generate_narrative_json(analysis_results)

print(f"识别到 {len(narrative_json['narratives'])} 个投资叙事")

在主流程中使用

# 方式1: 使用爬虫（默认）
python main.py --mode daily

# 方式2: 不使用爬虫（使用测试数据）
python main.py --mode daily --no-crawler

数据缓存

爬虫会自动缓存抓取的新闻到 data/news_cache/ 目录：

data/news_cache/
├── xinwenlianbo_20251229.json
├── xinwenlianbo_20251230.json
└── ...

当网络失败时，系统会自动尝试从缓存读取。

批量爬取历史数据

from skills.crawler import XinwenLianboCrawler

crawler = XinwenLianboCrawler()

# 爬取一段时间范围的新闻
results = crawler.crawl_date_range(
    start_date="20251220",
    end_date="20251229"
)

# results 是一个字典: {date: news_list}
for date, news_list in results.items():
    print(f"{date}: {len(news_list)} 条新闻")

自定义解析逻辑

如果网站结构发生变化，可以修改 _parse_content() 方法：

def _parse_content(self, markdown: str, html: str) -> List[Dict[str, str]]:
    """
    自定义解析逻辑
    
    Args:
        markdown: Crawl4AI 提取的 Markdown
        html: 原始 HTML
    
    Returns:
        [{"title": "", "text": ""}]
    """
    # 在这里添加你的解析逻辑
    pass

故障排查

问题1: 爬虫返回空列表

可能原因:

网站结构变化
当天新闻尚未更新（新闻联播通常 19:30 播出，网站 20:00 后更新）
网络连接问题
被 Cloudflare 拦截（虽然使用了 cloudscraper，但仍可能发生）

解决方法:

# 检查爬虫输出
python skills/crawler.py

# 查看缓存文件
cat data/news_cache/xinwenlianbo_*.json

# 检查网站是否可访问
curl https://cn.govopendata.com/xinwenlianbo/20251229/

问题2: cloudscraper 安装失败

解决方法:

# 升级 pip
pip install --upgrade pip

# 单独安装 cloudscraper
pip install cloudscraper --upgrade

# 如果仍然失败，安装依赖
pip install requests[security] pyparsing

问题3: 解析结果不准确

解决方法: 编辑 skills/crawler.py，检查 _parse_content() 方法中的 CSS 选择器是否与网站实际结构匹配：

# 当前使用的选择器
sections = soup.select('.content-section')
title_tag = section.select_one('.content-heading')
content_tag = section.select_one('.content-body')

可以手动访问网站，使用浏览器开发者工具（F12）查看实际的 DOM 结构。

性能优化

1. 使用缓存

爬虫会自动缓存结果，避免重复请求：

# 首次爬取（从网站）
news_list = crawler.crawl_sync(date="20251229")

# 再次爬取（从缓存，如果网络失败）
news_list = crawler.crawl_sync(date="20251229")  # 自动读取缓存

2. 批量爬取时添加延迟

crawler = XinwenLianboCrawler()

# crawl_date_range 已内置 2 秒延迟
results = crawler.crawl_date_range("20251220", "20251229")

定时任务配置

Linux/Mac (crontab)

# 编辑 crontab
crontab -e

# 添加定时任务（每天 20:30 执行）
30 20 * * * cd /path/to/MND-IA && python skills/crawler.py >> logs/crawler.log 2>&1

Windows (任务计划程序)

打开"任务计划程序"
创建基本任务
触发器：每天 20:30
操作：启动程序 python.exe
参数：E:\02projects\MND-IA\skills\crawler.py

API 参考

XinwenLianboCrawler

`init(base_url)`

初始化爬虫
base_url: 新闻联播网站基础 URL
自动创建 cloudscraper 会话

`crawl_sync(date=None)`

同步爬取接口（推荐使用）
date: 日期字符串 (YYYYMMDD)，默认今天
返回: List[Dict[str, str]]

`crawl_xinwenlianbo(date=None)`

爬取核心方法
date: 日期字符串
返回: List[Dict[str, str]]

`crawl_date_range(start_date, end_date)`

批量爬取日期范围
start_date: 开始日期 (YYYYMMDD)
end_date: 结束日期 (YYYYMMDD)
返回: Dict[str, List[Dict]]
自动添加 2 秒延迟避免请求过快

扩展建议

1. 添加更多新闻源

可以参考 Skill F 的实现，添加其他新闻源爬虫：

财联社快讯
人民日报
发改委官网
各部委新闻发布

2. 使用 LLM 增强提取

# 配置 LLM 客户端
from openai import OpenAI

client = OpenAI(api_key="your-api-key")

# 使用 LLM 辅助提取（需要实现相关代码）
news_list = await crawler.crawl_xinwenlianbo(use_llm=True)

3. 集成到实时监控

监听新闻联播更新，实时触发分析流程。

许可证

MIT License

6.5 KiB Raw Permalink Blame History Unescape Escape