Python网络爬虫实战指南-CFANZ编程社区

概述

网络爬虫是一种自动化程序，用于从网站中提取和收集数据。Python因其简洁的语法和强大的第三方库支持，成为了爬虫开发的首选语言。本文将介绍如何使用Python构建高效、稳定的网络爬虫。

核心技术栈

1. 基础库介绍

requests: 发送HTTP请求的优雅库
BeautifulSoup: HTML/XML解析利器
lxml: 高性能的XML和HTML解析器
selenium: 模拟浏览器行为，处理动态内容

2. 环境准备

pip install requests beautifulsoup4 lxml selenium fake-useragent

实战案例：新闻网站爬取

基础爬虫实现

import requests
from bs4 import BeautifulSoup
from fake_useragent import UserAgent
import time
import csv

class NewsSpider:
    def __init__(self):
        self.session = requests.Session()
        self.ua = UserAgent()
        self.headers = {
            'User-Agent': self.ua.random,
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
            'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
            'Accept-Encoding': 'gzip, deflate',
            'Connection': 'keep-alive',
        }
    
    def get_page(self, url, retry=3):
        """获取页面内容，带重试机制"""
        for i in range(retry):
            try:
                response = self.session.get(url, headers=self.headers, timeout=10)
                response.raise_for_status()
                return response
            except requests.exceptions.RequestException as e:
                print(f"请求失败 ({i+1}/{retry}): {e}")
                if i < retry - 1:
                    time.sleep(2 ** i)  # 指数退避
        return None
    
    def parse_news_list(self, html):
        """解析新闻列表页面"""
        soup = BeautifulSoup(html, 'lxml')
        news_items = []
        
        # 根据实际网站结构调整选择器
        articles = soup.select('div.news-item')
        
        for article in articles:
            try:
                title = article.select_one('h3.title a').get_text(strip=True)
                url = article.select_one('h3.title a')['href']
                pub_time = article.select_one('.pub-time').get_text(strip=True)
                summary = article.select_one('.summary').get_text(strip=True)
                
                news_items.append({
                    'title': title,
                    'url': url,
                    'pub_time': pub_time,
                    'summary': summary
                })
            except AttributeError:
                continue  # 跳过解析失败的条目
        
        return news_items

数据持久化与处理

def save_to_csv(self, data, filename='news_data.csv'):
    """保存数据到CSV文件"""
    with open(filename, 'w', newline='', encoding='utf-8') as f:
        if data:
            writer = csv.DictWriter(f, fieldnames=data[0].keys())
            writer.writeheader()
            writer.writerows(data)
    print(f"数据已保存到 {filename}")

def crawl_news(self, base_url, pages=5):
    """批量爬取新闻数据"""
    all_news = []
    
    for page in range(1, pages + 1):
        print(f"正在爬取第 {page} 页...")
        url = f"{base_url}?page={page}"
        
        response = self.get_page(url)
        if response:
            news_items = self.parse_news_list(response.text)
            all_news.extend(news_items)
            print(f"第 {page} 页获取到 {len(news_items)} 条新闻")
        
        # 礼貌性延时
        time.sleep(1)
    
    return all_news

高级技巧

1. 处理反爬机制

import random
from urllib.parse import urljoin

class AdvancedSpider(NewsSpider):
    def __init__(self):
        super().__init__()
        self.proxies_pool = []  # 代理池
        self.delay_range = (1, 3)  # 随机延时范围
    
    def random_delay(self):
        """随机延时，模拟人类行为"""
        delay = random.uniform(*self.delay_range)
        time.sleep(delay)
    
    def rotate_user_agent(self):
        """轮换User-Agent"""
        self.headers['User-Agent'] = self.ua.random

2. 异常处理与监控

import logging
from functools import wraps

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('spider.log'),
        logging.StreamHandler()
    ]
)

def retry_on_failure(max_retries=3):
    """装饰器：失败重试"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    logging.warning(f"尝试 {attempt + 1} 失败: {e}")
                    if attempt == max_retries - 1:
                        raise
                    time.sleep(2 ** attempt)
        return wrapper
    return decorator

使用示例

if __name__ == "__main__":
    spider = NewsSpider()
    
    # 爬取新闻数据
    news_data = spider.crawl_news("https://example-news.com/news", pages=3)
    
    # 保存数据
    spider.save_to_csv(news_data, 'latest_news.csv')
    
    print(f"爬取完成，共获取 {len(news_data)} 条新闻")