Python定时爬取新闻网站头条：从零到一的自动化实践

目录

一、为什么需要定时爬取新闻？

（一）传统方式的局限性

（二）自动化方案的优势

（三）典型应用场景

二、技术选型与工具准备

（一）核心库介绍

（二）环境配置

（三）目标网站分析

三、核心代码实现

（一）基础爬取功能

（二）数据存储模块

（三）定时任务调度

四、进阶优化方案

（一）动态代理池实现

（二）新闻去重策略

（三）异常通知机制

五、部署与维护建议

（一）服务器部署方案

（二）监控与日志分析

六、常见问题Q&A

七、总结与展望

---

在信息爆炸的时代，新闻头条是公众了解时事的重要窗口。无论是数据分析师追踪热点趋势，还是产品经理监控竞品动态，定时获取新闻头条都成为一项基础需求。本文将以爬取某主流新闻网站（以"腾讯新闻"为例）为例，介绍如何用Python实现定时爬取、数据存储和异常处理的完整流程，帮助读者快速搭建自己的新闻监控系统。

---

一、为什么需要定时爬取新闻？

（一）传统方式的局限性

手动访问新闻网站存在三大痛点：

1. 时效性差：热点事件可能每分钟都在更新，人工刷新无法实时捕捉
2. 效率低下：每天多次访问同一网站，重复操作浪费大量时间
3. 数据易丢失：浏览历史可能被清理，重要新闻难以追溯

（二）自动化方案的优势

通过Python定时爬取可实现：

• 每10分钟自动获取最新头条（频率可调）
• 数据持久化存储到数据库或Excel
• 配合邮件/企业微信推送关键新闻
• 历史数据可视化分析趋势

（三）典型应用场景

• 金融行业：监控政策类新闻对股市的影响
• 电商领域：跟踪竞品营销活动报道
• 媒体机构：自动收集热点选题素材
• 学术研究：构建新闻语料库用于NLP训练

---

二、技术选型与工具准备

（一）核心库介绍

库名称	用途	版本要求
requests	发送HTTP请求获取网页内容	≥2.25.1
BeautifulSoup	解析HTML提取新闻标题和链接	≥4.9.3
schedule	实现定时任务调度	≥1.1.0
sqlite3	轻量级数据库存储新闻数据	内置
logging	记录爬虫运行日志	内置

（二）环境配置

创建虚拟环境（推荐）：

python -m venv news_spider_env
source news_spider_env/bin/activate # Linux/Mac
.\news_spider_env\Scripts\activate # Windows

安装依赖库：

pip install requests beautifulsoup4 schedule

（三）目标网站分析

以腾讯新闻首页为例（https://www.**.com/）：

1. 使用浏览器开发者工具检查：
   • 头条新闻位于<div class="Q-topWrap">内
   • 每条新闻包含<a class="linkto">标签
   • 标题在<h2>标签中，链接在href属性
2. 请求头设置：
   • 添加User-Agent模拟浏览器访问
   • 设置Referer避免被反爬

---

三、核心代码实现

（一）基础爬取功能

import requests
from bs4 import BeautifulSoup
import sqlite3
import logging
from datetime import datetime# 配置日志
logging.basicConfig(filename='news_spider.log',level=logging.INFO,format='%(asctime)s - %(levelname)s - %(message)s'
)def get_news_headers():"""返回带反爬头的请求头"""return {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36','Referer': 'https://www.***.com/'}def fetch_news():"""获取腾讯新闻头条"""url = 'https://www.***.com/'try:response = requests.get(url, headers=get_news_headers(), timeout=10)response.raise_for_status()  # 检查请求是否成功soup = BeautifulSoup(response.text, 'html.parser')news_list = []top_wrap = soup.find('div', class_='Q-topWrap')if top_wrap:for item in top_wrap.find_all('a', class_='linkto'):title = item.find('h2').get_text(strip=True) if item.find('h2') else '无标题'link = item['href'] if 'href' in item.attrs else '#'# 处理相对链接if not link.startswith('http'):link = f'https:{link}' if link.startswith('//') else f'https://www.qq.com{link}'news_list.append((title, link))return news_listexcept requests.exceptions.RequestException as e:logging.error(f'请求失败: {str(e)}')return []

（二）数据存储模块

def init_db():"""初始化SQLite数据库"""conn = sqlite3.connect('news.db')cursor = conn.cursor()cursor.execute('''CREATE TABLE IF NOT EXISTS headlines (id INTEGER PRIMARY KEY AUTOINCREMENT,title TEXT NOT NULL,url TEXT NOT NULL,fetch_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP)''')conn.commit()conn.close()def save_to_db(news_list):"""存储新闻到数据库"""conn = sqlite3.connect('news.db')cursor = conn.cursor()for title, url in news_list:try:cursor.execute('INSERT INTO headlines (title, url) VALUES (?, ?)',(title, url))except sqlite3.IntegrityError:logging.warning(f'重复新闻已跳过: {title}')conn.commit()conn.close()logging.info(f'成功存储 {len(news_list)} 条新闻')

（三）定时任务调度

import schedule
import timedef job():"""定时任务执行函数"""logging.info('开始执行新闻爬取任务...')news_list = fetch_news()if news_list:save_to_db(news_list)else:logging.warning('本次未获取到有效新闻')def start_scheduler(interval_minutes=10):"""启动定时调度"""schedule.every(interval_minutes).minutes.do(job)logging.info(f'定时任务已启动，每{interval_minutes}分钟执行一次')while True:schedule.run_pending()time.sleep(1)if __name__ == '__main__':init_db()start_scheduler(10)  # 每10分钟执行一次

四、进阶优化方案

（一）动态代理池实现

import random
from fake_useragent import UserAgentclass ProxyPool:def __init__(self):self.proxies = [{'http': 'http://123.123.123.123:8080'},  # 示例代理# 实际应从代理API获取或维护代理列表]def get_random_proxy(self):return random.choice(self.proxies)def fetch_with_proxy():"""带代理的请求示例"""proxy_pool = ProxyPool()ua = UserAgent()try:proxy = proxy_pool.get_random_proxy()response = requests.get('https://www.***.com/',headers={'User-Agent': ua.random},proxies=proxy,timeout=10)# 处理响应...except Exception as e:logging.error(f'代理请求失败: {str(e)}')

（二）新闻去重策略

def is_duplicate(title, url):"""检查新闻是否已存在"""conn = sqlite3.connect('news.db')cursor = conn.cursor()cursor.execute('SELECT 1 FROM headlines WHERE title=? OR url=?',(title, url))exists = cursor.fetchone()conn.close()return exists is not None# 修改后的save_to_db函数
def save_to_db_v2(news_list):"""带去重的存储"""conn = sqlite3.connect('news.db')cursor = conn.cursor()for title, url in news_list:if not is_duplicate(title, url):try:cursor.execute('INSERT INTO headlines (title, url) VALUES (?, ?)',(title, url))except sqlite3.Error as e:logging.error(f'数据库错误: {str(e)}')conn.commit()conn.close()

（三）异常通知机制

import smtplib
from email.mime.text import MIMETextdef send_alert(subject, content):"""发送邮件警报"""msg = MIMEText(content)msg['Subject'] = subjectmsg['From'] = '[e-mail]'msg['To'] = '[e-mail]'try:with smtplib.SMTP_SSL('smtp.example.com', 465) as server:server.login('[e-mail]', 'your_password')server.send_message(msg)logging.info('警报邮件已发送')except Exception as e:logging.error(f'邮件发送失败: {str(e)}')# 在job函数中添加异常处理
def job():try:logging.info('开始执行新闻爬取任务...')news_list = fetch_news()if news_list:save_to_db(news_list)else:send_alert('新闻爬取警告', '本次未获取到有效新闻')except Exception as e:send_alert('新闻爬取错误', f'任务执行失败: {str(e)}')logging.error(str(e))

五、部署与维护建议

（一）服务器部署方案

云服务器选择：

• 轻量级应用：腾讯云/阿里云1核2G实例
• 预算有限：使用Vultr/Linode的$5/月方案

屏幕会话管理：

# 启动tmux会话
tmux new -s news_spider
# 在会话中运行python脚本
python spider.py
# 按Ctrl+B再按D退出会话（程序继续运行）# 重新连接会话
tmux attach -t news_spider

系统服务化（Linux）：
创建/etc/systemd/system/news_spider.service：

[Unit]
Description=News Spider Service
After=network.target[Service]
User=ubuntu
WorkingDirectory=/home/ubuntu/news_spider
ExecStart=/home/ubuntu/news_spider_env/bin/python spider.py
Restart=always[Install]
WantedBy=multi-user.target

启用服务：

sudo systemctl daemon-reload
sudo systemctl start news_spider
sudo systemctl enable news_spider

（二）监控与日志分析

日志轮转配置：
创建/etc/logrotate.d/news_spider：

/home/ubuntu/news_spider/news_spider.log {
daily
rotate 7
compress
missingok
notifempty
}

关键指标监控：

• 成功获取新闻数
• 数据库写入次数
• 代理请求成功率
• 任务执行耗时

---

六、常见问题Q&A

Q1：被网站封IP怎么办？
A：立即启用备用代理池，建议使用住宅代理（如站大爷IP代理），配合每请求更换IP策略。可修改请求代码：

def get_random_proxy():# 从代理API获取或轮询代理列表return {'http': 'http://proxy.example.com:8080'}def fetch_news():proxy = get_random_proxy()try:return requests.get('https://www.qq.com/',proxies=proxy,timeout=10).textexcept:# 代理失效时切换代理重试proxy = get_random_proxy()# 再次尝试...

Q2：如何应对网站结构变化？
A：建立CSS选择器监控机制，当连续3次爬取失败时发送警报。示例检测代码：

def check_structure(soup):top_wrap = soup.find('div', class_='Q-topWrap')if not top_wrap or len(top_wrap.find_all('a')) < 5:send_alert('网站结构变更', '检测到腾讯新闻页面结构变化')return Falsereturn True

Q3：定时任务不准时怎么办？
A：使用更精确的定时库apscheduler替代schedule：

from apscheduler.schedulers.blocking import BlockingSchedulerdef precise_job():# 原job函数内容scheduler = BlockingScheduler()
scheduler.add_job(precise_job, 'interval', minutes=10, jitter=30)
scheduler.start()

Q4：如何获取新闻发布时间？
A：部分新闻网站在HTML中包含时间标签，可通过以下方式提取：

def extract_publish_time(item):time_tag = item.find('span', class_='publish-time')if time_tag:time_str = time_tag.get_text(strip=True)# 解析时间字符串（示例）try:return datetime.strptime(time_str, '%Y-%m-%d %H:%M')except ValueError:return datetime.now()return datetime.now()

Q5：数据量大了如何优化存储？
A：考虑以下方案：

1. 分表存储：按年月创建表（headlines_202307）
2. 列式存储：改用Parquet格式（配合pandas）
3. 索引优化：为title和url字段创建索引

CREATE INDEX idx_title ON headlines(title);
CREATE INDEX idx_url ON headlines(url);

七、总结与展望

通过本文的实践，我们实现了：

• 每10分钟自动爬取腾讯新闻头条
• 数据持久化存储与去重
• 完善的错误处理和通知机制
• 可扩展的代理和部署方案

未来改进方向：

1. 增加新闻内容正文抓取
2. 实现自然语言处理分析热点
3. 开发Web界面展示历史数据
4. 容器化部署（Docker+K8s）

新闻爬虫的本质是信息获取的自动化，在遵守robots.txt和版权法规的前提下，合理使用爬虫技术可以极大提升工作效率。建议读者从本例出发，逐步扩展到更多新闻源和更复杂的数据处理场景。