20.1 网络爬虫基础

20.1.1 爬虫定义与原理

20.1.2 法律与道德规范

表19-1 爬虫合法性要点

20.2 请求库使用

20.2.1 requests库

基本使用：

import requests

def fetch_page(url):
    try:
        response = requests.get(
            url,
            headers={
                'User-Agent': 'Mozilla/5.0',
                'Accept-Language': 'zh-CN'
            },
            timeout=5
        )
        response.raise_for_status()  # 检查HTTP状态码
        return response.text
    except requests.exceptions.RequestException as e:
        print(f"请求失败: {e}")
        return None

# 示例使用
html = fetch_page('https://example.com')

20.2.2 高级请求技巧

# 会话保持
session = requests.Session()
session.get('https://example.com/login', params={'user': 'test'})

# 代理设置
proxies = {
    'http': 'http://10.10.1.10:3128',
    'https': 'http://10.10.1.10:1080'
}
response = requests.get(url, proxies=proxies)

# 文件下载
with requests.get('https://example.com/image.jpg', stream=True) as r:
    with open('image.jpg', 'wb') as f:
        for chunk in r.iter_content(chunk_size=8192):
            f.write(chunk)

20.3 数据解析技术

20.3.1 BeautifulSoup解析

from bs4 import BeautifulSoup

def parse_html(html):
    soup = BeautifulSoup(html, 'lxml')
    
    # CSS选择器
    titles = soup.select('h1.article-title')
    
    # 属性提取
    links = [a['href'] for a in soup.find_all('a', class_='external')]
    
    # 文本处理
    content = soup.find('div', id='content').get_text(strip=True, separator='\n')
    
    return {
        'titles': [t.text for t in titles],
        'links': links,
        'content': content
    }

20.3.2 XPath与lxml

from lxml import etree

def xpath_parse(html):
    tree = etree.HTML(html)
    
    # 提取商品价格
    prices = tree.xpath('//div[@class="price"]/text()')
    
    # 提取嵌套数据
    items = []
    for item in tree.xpath('//div[@class="product"]'):
        items.append({
            'name': item.xpath('.//h2/text()')[0],
            'sku': item.xpath('./@data-sku')[0]
        })
    
    return {'prices': prices, 'items': items}

20.4 动态页面处理

20.4.1 Selenium自动化

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

def selenium_crawl(url):
    options = webdriver.ChromeOptions()
    options.add_argument('--headless')  # 无头模式
    driver = webdriver.Chrome(options=options)
    
    try:
        driver.get(url)
        
        # 等待元素加载
        element = WebDriverWait(driver, 10).until(
            EC.presence_of_element_located((By.CSS_SELECTOR, ".dynamic-content"))
        )
        
        # 执行JavaScript
        driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
        
        # 获取渲染后页面
        html = driver.page_source
        return html
    finally:
        driver.quit()

20.4.2 接口逆向分析

import json

def api_crawl():
    # 分析XHR请求
    api_url = 'https://api.example.com/data'
    params = {
        'page': 1,
        'size': 20,
        'timestamp': int(time.time()*1000)
    }
    
    response = requests.get(api_url, params=params)
    data = response.json()
    
    # 解析JSON数据
    for item in data['list']:
        print(f"商品: {item['name']}, 价格: {item['price']}")

20.5 数据存储方案

20.5.1 文件存储

import csv
import json

def save_to_csv(data, filename):
    with open(filename, 'w', newline='', encoding='utf-8') as f:
        writer = csv.DictWriter(f, fieldnames=data[0].keys())
        writer.writeheader()
        writer.writerows(data)

def save_to_json(data, filename):
    with open(filename, 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=2)

20.5.2 数据库存储

import sqlite3
import pymongo

# SQLite存储
def sqlite_save(data):
    conn = sqlite3.connect('data.db')
    c = conn.cursor()
    c.execute('''CREATE TABLE IF NOT EXISTS products
               (id TEXT, name TEXT, price REAL)''')
    c.executemany('INSERT INTO products VALUES (?,?,?)',
                 [(d['id'], d['name'], d['price']) for d in data])
    conn.commit()

# MongoDB存储
def mongo_save(data):
    client = pymongo.MongoClient('mongodb://localhost:27017/')
    db = client['web_data']
    collection = db['products']
    collection.insert_many(data)

20.6 反爬应对策略

20.6.1 常见反爬机制

表19-2 常见反爬技术与应对

20.6.2 高级反反爬技巧

# 代理IP池示例
class ProxyPool:
    def __init__(self):
        self.proxies = [
            'http://ip1:port',
            'http://ip2:port',
            # ...
        ]
        self.current = 0
    
    def get_proxy(self):
        proxy = self.proxies[self.current % len(self.proxies)]
        self.current += 1
        return {'http': proxy, 'https': proxy}

# 请求头随机生成
from fake_useragent import UserAgent
ua = UserAgent()

def get_random_headers():
    return {
        'User-Agent': ua.random,
        'Referer': 'https://www.google.com/',
        'Accept-Encoding': 'gzip, deflate, br'
    }

20.7 应用举例

案例1：电商商品爬虫

import requests
from bs4 import BeautifulSoup
import time
import random

def ecommerce_crawler(base_url, max_page=10):
    products = []
    
    for page in range(1, max_page+1):
        # 带延迟的请求
        time.sleep(random.uniform(1, 3))
        
        url = f"{base_url}?page={page}"
        html = fetch_page(url)
        if not html:
            continue
        
        soup = BeautifulSoup(html, 'lxml')
        items = soup.select('.product-item')
        
        for item in items:
            try:
                products.append({
                    'name': item.select_one('.name').text.strip(),
                    'price': float(item.select_one('.price').text.replace('yen', '')),
                    'sku': item['data-sku'],
                    'rating': item.select_one('.rating').text.strip()
                })
            except Exception as e:
                print(f"解析失败: {e}")
    
    save_to_csv(products, 'products.csv')
    return products

# 使用示例
ecommerce_crawler('https://example.com/products')

案例2：新闻聚合爬虫

import schedule
import datetime

def news_monitor():
    sources = [
        'https://news.source1.com/rss',
        'https://news.source2.com/api/latest'
    ]
    
    all_news = []
    
    for url in sources:
        try:
            if 'rss' in url:
                # 解析RSS
                news = parse_rss(url)
            else:
                # 调用API
                news = parse_news_api(url)
            all_news.extend(news)
        except Exception as e:
            print(f"爬取失败 {url}: {e}")
    
    # 去重存储
    store_news(all_news)
    print(f"{datetime.datetime.now()} 已抓取{len(all_news)}条新闻")

# 定时任务
schedule.every(1).hours.do(news_monitor)

while True:
    schedule.run_pending()
    time.sleep(60)

20.8 知识图谱

20.9 学习总结

核心要点：

• 掌握HTTP请求与响应处理
• 熟练使用主流解析工具
• 理解动态页面加载原理
• 能够应对常见反爬措施

实践建议：

• 遵守爬虫道德规范
• 添加随机请求延迟
• 实现异常处理机制
• 定期维护代理池

进阶方向：

• 分布式爬虫架构
• 验证码智能识别
• 数据清洗与分析
• 反爬JS逆向工程

常见陷阱：

• 触发网站防护机制
• 页面结构变更导致解析失败
• 未处理编码问题
• 法律风险意识不足

持续更新Python编程学习日志与技巧，敬请关注！

#编程# #学习# #python# #在头条记录我的2025#