项目2: Web爬虫¶

难度: ⭐⭐⭐ 中级时间: 3-4小时 涉及知识: requests, BeautifulSoup, 正则表达式, 文件操作

🎯 项目目标¶

创建一个简单的Web爬虫，能够： 1. 发送HTTP请求获取网页内容 2. 解析HTML提取数据 3. 处理分页和多个页面 4. 保存数据到文件 5. 遵守爬虫礼仪

📋 需求¶

功能需求¶

Bash

# 命令行使用
python crawler.py "https://example.com/news" --output data.json --pages 5

# 功能
- 爬取指定网站的文章标题和链接
- 支持分页
- 保存为JSON或CSV格式
- 添加延迟避免请求过快
- 显示爬取进度

技术要求¶

使用requests发送HTTP请求
使用BeautifulSoup解析HTML
使用argparse处理命令行参数
适当的错误处理
遵守robots.txt

🚀 实现步骤¶

步骤1: 环境准备¶

Bash

pip install requests beautifulsoup4 lxml

步骤2: 基础爬虫¶

Python

# crawler.py
import requests
from bs4 import BeautifulSoup
import argparse
import json
import time
from urllib.parse import urljoin, urlparse

def fetch_page(url, headers=None):
    """获取网页内容"""
    default_headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
    }
    headers = headers or default_headers

    try:
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()
        return response.text
    except requests.RequestException as e:
        print(f"请求失败: {e}")
        return None

def parse_articles(html, base_url):
    """解析文章列表"""
    soup = BeautifulSoup(html, 'lxml')
    articles = []

    # 根据实际网站调整选择器
    # 这里以示例选择器为例
    for item in soup.find_all('article') or soup.find_all('div', class_='post'):
        title_tag = item.find('h2') or item.find('h3') or item.find('a')
        link_tag = item.find('a')

        if title_tag and link_tag:
            title = title_tag.get_text(strip=True)
            href = link_tag.get('href', '')
            full_url = urljoin(base_url, href)

            articles.append({
                'title': title,
                'url': full_url
            })

    return articles

def save_data(data, filepath):
    """保存数据到JSON文件"""
    with open(filepath, 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=2)
    print(f"数据已保存到: {filepath}")

def main():
    parser = argparse.ArgumentParser(description='简单Web爬虫')
    parser.add_argument('url', help='目标网站URL')
    parser.add_argument('--output', default='output.json', help='输出文件')
    parser.add_argument('--delay', type=float, default=1.0, help='请求间隔(秒)')

    args = parser.parse_args()

    print(f"开始爬取: {args.url}")

    html = fetch_page(args.url)
    if html:
        articles = parse_articles(html, args.url)
        print(f"找到 {len(articles)} 篇文章")

        save_data(articles, args.output)

    print("爬取完成!")

if __name__ == '__main__':
    main()

步骤3: 添加分页支持¶

Python

def crawl_with_pagination(base_url, max_pages=5, delay=1.0):
    """爬取多页数据"""
    all_articles = []

    for page in range(1, max_pages + 1):
        # 根据网站的分页URL格式调整
        url = f"{base_url}?page={page}"

        print(f"正在爬取第 {page} 页...")
        html = fetch_page(url)

        if not html:
            break

        articles = parse_articles(html, base_url)
        if not articles:
            print("没有更多数据")
            break

        all_articles.extend(articles)
        print(f"第 {page} 页: 找到 {len(articles)} 篇文章")

        # 延迟，避免请求过快
        if page < max_pages:
            time.sleep(delay)

    return all_articles

步骤4: 完整实现¶

Python

# advanced_crawler.py
import requests
from bs4 import BeautifulSoup
import argparse
import json
import csv
import time
import re
from urllib.parse import urljoin, urlparse
from pathlib import Path

class WebCrawler:
    def __init__(self, delay=1.0, headers=None):
        self.delay = delay
        self.session = requests.Session()
        self.session.headers.update(headers or {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        })
        self.visited_urls = set()

    def fetch(self, url):
        """获取网页内容"""
        if url in self.visited_urls:
            return None

        try:
            print(f"Fetching: {url}")
            response = self.session.get(url, timeout=10)
            response.raise_for_status()
            self.visited_urls.add(url)
            return response.text
        except requests.RequestException as e:
            print(f"Error fetching {url}: {e}")
            return None

    def parse(self, html, selector_config):
        """根据配置解析HTML"""
        soup = BeautifulSoup(html, 'lxml')
        results = []

        items = soup.select(selector_config['container'])
        for item in items:
            data = {}
            for field, sel in selector_config['fields'].items():
                elem = item.select_one(sel)
                if elem:
                    if field == 'link':
                        data[field] = elem.get('href', '')
                    else:
                        data[field] = elem.get_text(strip=True)
            if data:
                results.append(data)

        return results

    def crawl(self, start_url, selector_config, max_pages=5):
        """爬取多个页面"""
        all_results = []
        base_url = f"{urlparse(start_url).scheme}://{urlparse(start_url).netloc}"

        for page in range(1, max_pages + 1):
            # 构建分页URL（根据实际情况调整）
            page_url = f"{start_url}?page={page}"

            html = self.fetch(page_url)
            if not html:
                break

            results = self.parse(html, selector_config)
            if not results:
                print(f"No more data on page {page}")
                break

            # 补全URL
            for item in results:
                if 'link' in item:
                    item['link'] = urljoin(base_url, item['link'])

            all_results.extend(results)
            print(f"Page {page}: {len(results)} items")

            if page < max_pages:
                time.sleep(self.delay)

        return all_results

    def save(self, data, filepath):
        """保存数据"""
        path = Path(filepath)

        if path.suffix == '.json':
            with open(path, 'w', encoding='utf-8') as f:
                json.dump(data, f, ensure_ascii=False, indent=2)
        elif path.suffix == '.csv':
            if data:
                with open(path, 'w', newline='', encoding='utf-8') as f:
                    writer = csv.DictWriter(f, fieldnames=data[0].keys())
                    writer.writeheader()
                    writer.writerows(data)

        print(f"Saved {len(data)} items to {filepath}")

def main():
    parser = argparse.ArgumentParser(description='Web Crawler')
    parser.add_argument('url', help='Start URL')
    parser.add_argument('--output', default='output.json', help='Output file')
    parser.add_argument('--pages', type=int, default=5, help='Max pages')
    parser.add_argument('--delay', type=float, default=1.0, help='Delay between requests')

    args = parser.parse_args()

    # 配置选择器（需要根据目标网站调整）
    selector_config = {
        'container': 'article, .post, .item',
        'fields': {
            'title': 'h2, h3, .title',
            'link': 'a',
            'summary': '.summary, .excerpt, p'
        }
    }

    crawler = WebCrawler(delay=args.delay)
    data = crawler.crawl(args.url, selector_config, max_pages=args.pages)
    crawler.save(data, args.output)

if __name__ == '__main__':
    main()

📝 扩展挑战¶

添加正则表达式支持 - 使用正则提取特定模式的数据
多线程爬取 - 使用concurrent.futures加速
数据去重 - 基于URL或标题去重
增量爬取 - 只爬取新内容
遵守robots.txt - 使用robotparser

🎯 完成标准¶

💡 提示¶

先在一个简单的网站上测试
使用浏览器的开发者工具查看HTML结构
注意网站的robots.txt文件
不要请求过快，避免被封IP
考虑使用代理IP（高级）

📚 参考资源¶

🚀 下一步¶

完成后，尝试： - 项目3: ML模型训练 - 挑战更复杂的网站 - 学习Scrapy框架

记住: 爬虫要遵守网站规则和法律法规！