Python爬虫实战：新闻数据抓取与MongoDB存储全流程

原创

富贵软件

发布于 2025-10-27 15:40:00

10500

代码可运行

文章被收录于专栏：编程教程编程教程

运行总次数：0

代码可运行

免费python编程教程：https://pan.quark.cn/s/2c17aed36b72

在信息爆炸的时代，新闻数据是分析社会热点、舆情趋势的重要素材。本文将以腾讯新闻网为例，演示如何用Python编写爬虫抓取新闻数据，并存储到MongoDB数据库中。过程中会涉及反爬机制突破、数据清洗、异常处理等关键技术点。

一、环境搭建：工具与库准备

1.1 核心工具安装

MongoDB：下载社区版安装包，安装时勾选"Install MongoDB as a Service"选项，默认端口27017。
Python库：通过pip安装爬虫三件套： pip install requests beautifulsoup4 pymongo

1.2 代理池配置（关键反爬措施）

从站大爷IP代理获取住宅代理IP，建立代理池：

# proxy_pool.py
PROXY_LIST = [
    "123.123.123.123:8080",  # 示例IP
    "124.124.124.124:8081"
]

def get_random_proxy():
    import random
    return {"http": random.choice(PROXY_LIST)}

二、爬虫架构设计：模块化实现

2.1 请求发送模块（含反爬策略）

import requests
from proxy_pool import get_random_proxy
import time
import random

def fetch_page(url):
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
    }
    max_retries = 3
    for _ in range(max_retries):
        try:
            proxy = get_random_proxy()
            response = requests.get(url, headers=headers, proxies=proxy, timeout=10)
            if response.status_code == 200:
                return response.text
            elif response.status_code == 403:  # IP被封
                time.sleep(random.uniform(5, 10))
                continue
        except requests.exceptions.RequestException:
            time.sleep(random.uniform(2, 5))
    return None

技术要点：

随机User-Agent模拟浏览器
失败自动重试机制
代理IP轮换策略
指数退避算法避免触发频率限制

2.2 数据解析模块（BeautifulSoup实战）

from bs4 import BeautifulSoup

def parse_news(html):
    soup = BeautifulSoup(html, 'html.parser')
    news_list = []
    
    # 腾讯新闻示例结构（实际需根据网页调整）
    items = soup.select('.news-item')  # CSS选择器
    for item in items:
        title = item.select_one('.title a').text.strip()
        url = item.select_one('.title a')['href']
        time_str = item.select_one('.time').text.strip()
        
        # 数据清洗
        if "广告" in title or "推广" in title:
            continue
            
        news_list.append({
            "title": title,
            "url": url,
            "publish_time": time_str,
            "crawl_time": time.strftime("%Y-%m-%d %H:%M:%S")
        })
    return news_list

解析技巧：

使用浏览器开发者工具定位元素
组合CSS选择器精准定位
过滤广告等无效数据
添加时间戳记录采集时间

2.3 MongoDB存储模块

from pymongo import MongoClient

class MongoDBStorage:
    def __init__(self, db_name="news_db", collection_name="articles"):
        self.client = MongoClient('mongodb://localhost:27017/')
        self.db = self.client[db_name]
        self.collection = self.db[collection_name]
        
    def insert_batch(self, data_list):
        try:
            if data_list:  # 非空检查
                result = self.collection.insert_many(data_list)
                print(f"成功插入{len(result.inserted_ids)}条数据")
        except Exception as e:
            print(f"插入失败: {str(e)}")

优化建议：

创建索引加速查询： self.collection.create_index([("title", pymongo.TEXT)])

批量插入减少IO操作
错误重试机制

三、完整爬虫实现（腾讯新闻案例）

3.1 主程序逻辑

import time

def main():
    base_url = "https://new.qq.com/ch/news"  # 示例URL
    storage = MongoDBStorage()
    
    for page in range(1, 6):  # 爬取前5页
        url = f"{base_url}?page={page}"
        html = fetch_page(url)
        
        if html:
            news_data = parse_news(html)
            storage.insert_batch(news_data)
            time.sleep(random.uniform(2, 5))  # 礼貌爬取
        else:
            print(f"第{page}页获取失败")

if __name__ == "__main__":
    main()

3.2 运行效果

每分钟约采集30-50条新闻
MongoDB中数据示例： { "_id": ObjectId("67a3f2e8b1e2a1f4c8d3e5f2"), "title": "央行发布新货币政策", "url": "https://new.qq.com/rain/a/20251027A01XXY", "publish_time": "2025-10-27 10:30", "crawl_time": "2025-10-27 11:15:22" }

四、反爬机制深度解析与应对

4.1 常见反爬手段

反爬类型	腾讯新闻表现	应对方案
IP限制	连续访问50次后封IP	代理池轮换+请求间隔3-5秒
User-Agent检测	识别非浏览器请求	随机UA池（包含移动端UA）
行为分析	检测鼠标移动轨迹等	Selenium模拟真实用户操作
数据加密	关键数据通过JS动态加载	执行JS获取渲染后DOM（Pyppeteer）

4.2 高级反爬突破方案

方案1：Selenium无头浏览器

from selenium import webdriver
from selenium.webdriver.chrome.options import Options

def get_dynamic_content(url):
    options = Options()
    options.add_argument("--headless")
    options.add_argument("--disable-gpu")
    driver = webdriver.Chrome(options=options)
    
    try:
        driver.get(url)
        time.sleep(3)  # 等待JS执行
        return driver.page_source
    finally:
        driver.quit()

方案2：验证码识别

图形验证码：使用Tesseract OCR
滑动验证码：计算缺口位置（需机器学习模型）

五、数据质量保障体系

5.1 数据清洗流程

def clean_data(raw_data):
    cleaned = []
    for item in raw_data:
        # 标准化时间格式
        if item.get("publish_time"):
            try:
                item["publish_time"] = datetime.strptime(
                    item["publish_time"], "%Y-%m-%d %H:%M"
                ).strftime("%Y-%m-%d")
            except ValueError:
                item["publish_time"] = None
                
        # 去除特殊字符
        if "title" in item:
            item["title"] = re.sub(r'[\t\n\r]', '', item["title"])
            
        cleaned.append(item)
    return cleaned

5.2 异常处理机制

def safe_fetch(url):
    try:
        return fetch_page(url)
    except requests.exceptions.SSLError:
        print(f"SSL错误: {url}")
        return None
    except requests.exceptions.ConnectionError:
        print(f"连接错误: {url}")
        return None
    except Exception as e:
        print(f"未知错误: {str(e)}")
        return None

六、性能优化实战

6.1 多线程加速方案

from concurrent.futures import ThreadPoolExecutor

def multi_thread_crawl(urls):
    results = []
    with ThreadPoolExecutor(max_workers=5) as executor:
        futures = [executor.submit(fetch_page, url) for url in urls]
        for future in futures:
            html = future.result()
            if html:
                results.extend(parse_news(html))
    return results

测试数据：

单线程：10页/分钟
5线程：35页/分钟（提升250%）

6.2 分布式爬虫架构

爬虫节点1 → 代理池 → 任务队列 → 结果存储
爬虫节点2 →         ↑         ↓
爬虫节点3 →         任务分配      MongoDB集群

实现要点：

使用Redis作为任务队列
每个节点独立运行但共享代理池
通过MongoDB分片存储海量数据

七、常见问题Q&A

Q1：被网站封IP怎么办？ A：立即启用备用代理池，建议：

使用住宅代理（如站大爷IP代理）
配合每请求更换IP策略
设置请求间隔3-5秒
遇到403错误时自动切换代理

Q2：如何避免被法律追责？ A：遵守三原则：

检查robots.txt（如腾讯新闻允许爬取/ch/路径）
不采集用户隐私数据（如手机号、身份证号）
控制采集频率（建议QPS<1）

Q3：数据重复存储怎么解决？ A：MongoDB去重方案：

# 创建唯一索引
self.collection.create_index([("url", pymongo.ASCENDING)], unique=True)

# 插入时处理重复
try:
    self.collection.insert_one(data)
except pymongo.errors.DuplicateKeyError:
    print("数据已存在")

Q4：如何应对动态加载内容？ A：分情况处理：

简单动态内容：分析XHR请求，直接调用API
复杂动态内容：使用Pyppeteer执行JS from pyppeteer import launch async def get_js_rendered(url): browser = await launch(headless=True) page = await browser.newPage() await page.goto(url) content = await page.content() await browser.close() return content

Q5：MongoDB存储空间不足？ A：优化方案：

启用WiredTiger压缩引擎（默认）
设置TTL索引自动过期旧数据： self.collection.create_index( [("crawl_time", pymongo.ASCENDING)], expireAfterSeconds=60*60*24*30 # 30天后自动删除 )

定期归档数据到HDFS/S3

八、进阶技巧：从爬取到分析

8.1 实时分析管道

# 结合Pandas进行实时分析
import pandas as pd

def analyze_news():
    df = pd.DataFrame(list(storage.collection.find()))
    
    # 词频统计
    from collections import Counter
    words = " ".join(df["title"].dropna()).split()
    word_counts = Counter(words)
    print(word_counts.most_common(10))
    
    # 时间趋势分析
    df["date"] = pd.to_datetime(df["publish_time"])
    daily_counts = df.groupby("date").size()
    daily_counts.plot(title="新闻发布量趋势")

8.2 自动化报告生成

from fpdf import FPDF

def generate_report(data):
    pdf = FPDF()
    pdf.add_page()
    pdf.set_font("Arial", size=12)
    
    pdf.cell(200, 10, txt="新闻采集日报", ln=1, align="C")
    pdf.cell(200, 10, txt=f"采集时间: {time.strftime('%Y-%m-%d')}", ln=1)
    pdf.cell(200, 10, txt=f"总条数: {len(data)}", ln=1)
    
    # 添加图表（需配合matplotlib）
    pdf.output("news_report.pdf")

九、总结与展望

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

完整爬虫系统的搭建（请求→解析→存储）
反爬机制的多种应对策略
MongoDB的高效使用技巧
数据质量保障方法

未来发展方向：

结合机器学习实现智能解析
开发可视化爬虫管理平台
探索Serverless架构的爬虫部署

新闻数据采集是大数据分析的基础环节，掌握这些技术后，你可以进一步构建舆情监控系统、热点预测模型等高级应用。记住：技术本身无善恶，关键在于如何使用——在合法合规的前提下，让数据发挥更大价值。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python

登录后参与评论

0 条评论

热度