# Python Web 爬虫实战

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

---

## 1. 爬虫基础

### 1.1 爬虫工作流程

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

### 1.2 安装依赖

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

---

## 2. 基础爬虫

### 2.1 获取页面

```python
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 解析数据

```python
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 分页爬取

```python
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 存储

```python
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 存储

```python
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 存储

```python
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. 异步爬虫

```python
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 设置请求头

```python
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

```python
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 限速与重试

```python
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. 完整爬虫项目

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

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. 法律风险
   - 不爬取个人隐私数据
   - 不爬取需要授权的数据
   - 不进行恶意攻击
```

```python
# 检查 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. 练习题

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

---

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