Python Web 爬虫实战

2025-12-17 约 2574 字预计阅读 6 分钟

学习目标：综合运用 requests、BeautifulSoup、asyncio 等技术，开发完整的 Web 爬虫，掌握反爬策略和数据存储。

1. 爬虫基础

1.1 爬虫工作流程

1. 发送请求 (requests/aiohttp)
       ↓
2. 获取响应 (HTML/JSON)
       ↓
3. 解析数据 (BeautifulSoup/lxml)
       ↓
4. 提取信息 (CSS选择器/XPath)
       ↓
5. 存储数据 (CSV/JSON/数据库)

1.2 安装依赖

        
pip install requests beautifulsoup4 lxml
pip install aiohttp  # 异步爬虫
pip install selenium  # 动态页面

2. 基础爬虫

2.1 获取页面

        
        
        
    
import requests
from bs4 import BeautifulSoup

def fetch_page(url: str) -> str:
    """获取网页内容"""
    headers = {
        'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) '
                       'AppleWebKit/537.36 (KHTML, like Gecko) '
                       'Chrome/120.0.0.0 Safari/537.36'
    }
    
    try:
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()
        response.encoding = response.apparent_encoding
        return response.text
    except requests.RequestException as e:
        print(f"请求失败: {e}")
        return None

# 获取页面
html = fetch_page('https://quotes.toscrape.com/')
if html:
    print(f"页面长度: {len(html)} 字符")

2.2 解析数据

        
        
        
    
def parse_quotes(html: str) -> list[dict]:
    """解析名言页面"""
    soup = BeautifulSoup(html, 'lxml')
    quotes = []
    
    for quote_div in soup.select('div.quote'):
        text = quote_div.select_one('span.text').get_text()
        author = quote_div.select_one('small.author').get_text()
        tags = [tag.get_text() for tag in quote_div.select('a.tag')]
        
        quotes.append({
            'text': text,
            'author': author,
            'tags': tags
        })
    
    return quotes

# 解析
quotes = parse_quotes(html)
for q in quotes[:3]:
    print(f"[{q['author']}] {q['text'][:50]}...")
    print(f"  标签: {', '.join(q['tags'])}")

2.3 分页爬取

        
        
        
    
import time

def scrape_all_pages(base_url: str, max_pages: int = 10) -> list[dict]:
    """爬取所有分页"""
    all_quotes = []
    
    for page in range(1, max_pages + 1):
        url = f"{base_url}/page/{page}/"
        print(f"正在爬取第 {page} 页: {url}")
        
        html = fetch_page(url)
        if not html:
            break
        
        quotes = parse_quotes(html)
        if not quotes:
            print(f"第 {page} 页无数据，停止")
            break
        
        all_quotes.extend(quotes)
        
        # 礼貌延迟
        time.sleep(1)
    
    return all_quotes

# 爬取
# all_quotes = scrape_all_pages('https://quotes.toscrape.com', 5)

3. 数据存储

3.1 JSON 存储

        
        
        
    
import json
from pathlib import Path

def save_to_json(data: list[dict], filepath: str) -> None:
    """保存为 JSON 文件"""
    Path(filepath).write_text(
        json.dumps(data, ensure_ascii=False, indent=2),
        encoding='utf-8'
    )
    print(f"已保存 {len(data)} 条数据到 {filepath}")

# save_to_json(all_quotes, 'quotes.json')

3.2 CSV 存储

        
import csv

def save_to_csv(data: list[dict], filepath: str) -> None:
    """保存为 CSV 文件"""
    if not data:
        return
    
    with open(filepath, 'w', newline='', encoding='utf-8-sig') as f:
        writer = csv.DictWriter(f, fieldnames=data[0].keys())
        writer.writeheader()
        writer.writerows(data)
    
    print(f"已保存 {len(data)} 条数据到 {filepath}")

# save_to_csv(all_quotes, 'quotes.csv')

3.3 SQLite 存储

        
        
        
    
import sqlite3

def save_to_sqlite(data: list[dict], db_path: str = 'quotes.db') -> None:
    """保存到 SQLite 数据库"""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS quotes (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            text TEXT NOT NULL,
            author TEXT,
            tags TEXT
        )
    ''')
    
    for item in data:
        cursor.execute(
            'INSERT INTO quotes (text, author, tags) VALUES (?, ?, ?)',
            (item['text'], item['author'], ','.join(item['tags']))
        )
    
    conn.commit()
    print(f"已保存 {len(data)} 条数据到数据库")
    conn.close()

# save_to_sqlite(all_quotes)

4. 异步爬虫

        
        
        
    
import asyncio
import aiohttp
from bs4 import BeautifulSoup

async def fetch_async(session: aiohttp.ClientSession, url: str) -> str:
    """异步获取页面"""
    headers = {'User-Agent': 'Mozilla/5.0'}
    
    async with session.get(url, headers=headers, timeout=aiohttp.ClientTimeout(total=10)) as resp:
        return await resp.text()

async def scrape_page(session: aiohttp.ClientSession, url: str) -> list[dict]:
    """爬取单个页面"""
    html = await fetch_async(session, url)
    return parse_quotes(html)

async def scrape_concurrent(base_url: str, max_pages: int = 10) -> list[dict]:
    """并发爬取多个页面"""
    urls = [f"{base_url}/page/{i}/" for i in range(1, max_pages + 1)]
    
    async with aiohttp.ClientSession() as session:
        tasks = [scrape_page(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
    
    all_quotes = []
    for result in results:
        if isinstance(result, list):
            all_quotes.extend(result)
        else:
            print(f"错误: {result}")
    
    return all_quotes

# 运行异步爬虫
# all_quotes = asyncio.run(scrape_concurrent('https://quotes.toscrape.com', 10))

5. 反爬策略应对

5.1 设置请求头

        
        
        
    
HEADERS = {
    'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) ...',
    'Accept': 'text/html,application/xhtml+xml,...',
    'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
    'Referer': 'https://www.google.com/',
}

# 随机 User-Agent
import random

USER_AGENTS = [
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0.0.0',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) Chrome/120.0.0.0',
    'Mozilla/5.0 (X11; Linux x86_64) Firefox/121.0',
]

headers = {'User-Agent': random.choice(USER_AGENTS)}

5.2 代理 IP

        
        
        
    
def get_proxies() -> dict:
    """获取代理"""
    return {
        'http': 'http://proxy.example.com:8080',
        'https': 'http://proxy.example.com:8080',
    }

# 代理池
PROXY_POOL = [
    'http://1.2.3.4:8080',
    'http://5.6.7.8:8080',
]

def get_random_proxy():
    return random.choice(PROXY_POOL)

# 使用代理
# response = requests.get(url, proxies={'http': get_random_proxy()})

5.3 限速与重试

        
        
        
    
import time
from functools import wraps

def rate_limit(calls: int = 1, period: float = 1.0):
    """限速装饰器"""
    min_interval = period / calls
    last_called = [0.0]
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            wait = min_interval - elapsed
            if wait > 0:
                time.sleep(wait)
            last_called[0] = time.time()
            return func(*args, **kwargs)
        return wrapper
    return decorator

def retry(times: int = 3, delay: float = 1.0):
    """重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(times):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    print(f"第 {attempt+1} 次尝试失败: {e}")
                    if attempt < times - 1:
                        time.sleep(delay * (attempt + 1))
            raise Exception(f"重试 {times} 次后仍失败")
        return wrapper
    return decorator

@rate_limit(calls=2, period=1.0)
@retry(times=3, delay=1.0)
def fetch_with_retry(url: str) -> str:
    """带限速和重试的请求"""
    response = requests.get(url, headers=HEADERS, timeout=10)
    response.raise_for_status()
    return response.text

6. 完整爬虫项目

        
        
        
    
"""
新闻爬虫 - 综合示例
爬取新闻标题、链接、日期，保存到数据库
"""

import requests
from bs4 import BeautifulSoup
import sqlite3
import time
import logging
from dataclasses import dataclass
from typing import Optional

logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s: %(message)s')
logger = logging.getLogger(__name__)

@dataclass
class NewsItem:
    """新闻数据"""
    title: str
    url: str
    date: str
    category: str

class NewsScraper:
    """新闻爬虫"""
    
    def __init__(self, db_path: str = 'news.db'):
        self.session = requests.Session()
        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0 ...'
        })
        self.db_path = db_path
        self._init_db()
    
    def _init_db(self):
        """初始化数据库"""
        conn = sqlite3.connect(self.db_path)
        conn.execute('''
            CREATE TABLE IF NOT EXISTS news (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                title TEXT UNIQUE,
                url TEXT,
                date TEXT,
                category TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        conn.commit()
        conn.close()
    
    def fetch(self, url: str) -> Optional[str]:
        """获取页面"""
        try:
            resp = self.session.get(url, timeout=10)
            resp.raise_for_status()
            return resp.text
        except Exception as e:
            logger.error(f"获取失败 {url}: {e}")
            return None
    
    def parse(self, html: str, category: str) -> list[NewsItem]:
        """解析新闻列表"""
        soup = BeautifulSoup(html, 'lxml')
        news_list = []
        
        for article in soup.select('article.news-item'):
            title_el = article.select_one('h2.title a')
            if not title_el:
                continue
            
            news = NewsItem(
                title=title_el.get_text(strip=True),
                url=title_el.get('href', ''),
                date=article.select_one('time').get('datetime', ''),
                category=category
            )
            news_list.append(news)
        
        return news_list
    
    def save(self, news_list: list[NewsItem]) -> int:
        """保存到数据库"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        saved = 0
        
        for news in news_list:
            try:
                cursor.execute(
                    'INSERT OR IGNORE INTO news (title, url, date, category) VALUES (?,?,?,?)',
                    (news.title, news.url, news.date, news.category)
                )
                if cursor.rowcount > 0:
                    saved += 1
            except Exception as e:
                logger.error(f"保存失败: {e}")
        
        conn.commit()
        conn.close()
        logger.info(f"保存 {saved} 条新闻")
        return saved
    
    def run(self, categories: dict[str, str]):
        """运行爬虫"""
        total = 0
        
        for category, url in categories.items():
            logger.info(f"爬取分类: {category}")
            
            html = self.fetch(url)
            if not html:
                continue
            
            news_list = self.parse(html, category)
            total += self.save(news_list)
            
            time.sleep(2)  # 礼貌延迟
        
        logger.info(f"完成，共保存 {total} 条新闻")
        return total

# 使用
# scraper = NewsScraper()
# scraper.run({
#     '科技': 'https://example.com/tech',
#     '财经': 'https://example.com/finance',
# })

7. 爬虫礼仪与法律

爬虫注意事项：

1. 遵守 robots.txt
   - 检查目标网站的爬虫协议
   - 尊重 Disallow 规则

2. 控制频率
   - 设置合理延迟（1-3 秒）
   - 避免对服务器造成压力

3. 数据使用
   - 仅用于个人学习
   - 不商用、不传播
   - 尊重版权

4. 法律风险
   - 不爬取个人隐私数据
   - 不爬取需要授权的数据
   - 不进行恶意攻击

        
        
        
    
# 检查 robots.txt
import urllib.robotparser

def check_robots(url: str) -> bool:
    """检查是否允许爬取"""
    rp = urllib.robotparser.RobotFileParser()
    rp.set_url(f"https://{url.split('/')[2]}/robots.txt")
    rp.read()
    return rp.can_fetch('*', url)

# allowed = check_robots('https://example.com/page')

8. 小结

技术	用途
requests	HTTP 请求
BeautifulSoup	HTML 解析
aiohttp	异步请求
lxml	高性能解析
SQLite/CSV/JSON	数据存储
代理/限速/重试	反爬应对

9. 练习题

爬取一个公开的图书网站，提取书名、作者、价格并存入数据库。
使用 asyncio 实现并发爬虫，对比与同步爬虫的速度差异。
为爬虫添加日志记录和错误重试机制。
爬取数据后使用 pandas 进行简单分析（如按作者统计数量）。

下节预告：我们将使用 pandas 和 matplotlib 进行数据分析实战。

目录

Python Web 爬虫实战

1. 爬虫基础

1.1 爬虫工作流程

1.2 安装依赖

2. 基础爬虫

2.1 获取页面

2.2 解析数据

2.3 分页爬取

3. 数据存储

3.1 JSON 存储

3.2 CSV 存储

3.3 SQLite 存储

4. 异步爬虫

5. 反爬策略应对

5.1 设置请求头

5.2 代理 IP

5.3 限速与重试

6. 完整爬虫项目

7. 爬虫礼仪与法律

8. 小结

9. 练习题