章节 6：模块综合实战

学习目标

• 综合运用 Requests + Playwright + Scrapy 进行多模式数据采集
• 构建电商平台商品数据采集系统
• 搭建基于 Scrapy-Redis + MongoDB 的新闻资讯分布式爬虫集群

---

实战一：电商平台商品数据采集系统

项目概述

构建一个多模式电商数据采集系统，可采集京东、淘宝等电商平台的商品信息，包括标题、价格、评论数、店铺信息等。

技术选型：

场景	工具	说明
简单列表页	Requests + lxml	高效采集，适合静态页面
需要 JS 渲染的页面	Playwright	处理动态加载内容
大规模持续采集	Scrapy	框架化管理，自动重试调度
反爬对抗	代理池 + 指纹伪装	绕过 IP 限制和检测

项目结构

ecommerce_crawler/
├── config.py                      # 全局配置
├── proxy_pool.py                  # 代理池
├── utils.py                       # 工具函数
├── crawlers/
│   ├── __init__.py
│   ├── requests_crawler.py        # Requests 模式
│   ├── playwright_crawler.py      # Playwright 模式
│   └── scrapy_crawler/            # Scrapy 模式
│       ├── settings.py
│       ├── items.py
│       ├── pipelines.py
│       └── spiders/
│           └── product_spider.py
├── storage/
│   ├── __init__.py
│   ├── csv_writer.py              # CSV 存储
│   └── mongodb_writer.py          # MongoDB 存储
└── main.py                        # 统一入口

1. 配置与工具模块

# config.py
PROXY_POOL_CONFIG = {
    "redis_url": "redis://localhost:6379/0",
    "min_proxies": 20,  # 最少可用代理数
    "check_interval": 300,  # 校验间隔（秒）
}

REQUEST_CONFIG = {
    "timeout": 15,
    "retry_times": 3,
    "concurrency": 5,
    "qps_limit": 3,  # 每秒请求数限制
}

PLAYWRIGHT_CONFIG = {
    "headless": True,
    "viewport": {"width": 1920, "height": 1080},
    "user_agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/120.0.0.0 Safari/537.36"
    ),
}

STORAGE_CONFIG = {
    "type": "csv",  # csv | mongodb
    "csv_path": "data/products.csv",
    "mongodb_uri": "mongodb://localhost:27017/",
    "mongodb_db": "ecommerce",
    "mongodb_collection": "products",
}

# utils.py
import hashlib
import json
import time
from datetime import datetime


def generate_product_id(url, title):
    """生成商品唯一 ID"""
    raw = f"{url}|{title}"
    return hashlib.md5(raw.encode()).hexdigest()


def clean_price(price_str):
    """清洗价格字符串"""
    if not price_str:
        return 0.0
    price_str = price_str.replace("¥", "").replace("$", "") \
                         .replace(",", "").replace(" ", "")
    try:
        return float(price_str)
    except ValueError:
        return 0.0


def rate_limiter(max_qps=3):
    """请求限流装饰器"""
    last_time = [0]

    def decorator(func):
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_time[0]
            wait_time = 1.0 / max_qps - elapsed
            if wait_time > 0:
                time.sleep(wait_time)
            result = func(*args, **kwargs)
            last_time[0] = time.time()
            return result
        return wrapper
    return decorator


def save_json(data, filepath):
    """保存 JSON 文件"""
    with open(filepath, "a", encoding="utf-8") as f:
        f.write(json.dumps(data, ensure_ascii=False) + "\n")

2. Requests 模式采集器

# crawlers/requests_crawler.py
import requests
from lxml import etree
from config import REQUEST_CONFIG
from utils import clean_price, rate_limiter


class RequestsCrawler:
    """适用于静态页面的采集器"""

    def __init__(self, proxy_pool=None):
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                          "AppleWebKit/537.36 Chrome/120.0.0.0 Safari/537.36"),
            "Accept": "text/html,application/xhtml+xml,*/*",
        })
        self.proxy_pool = proxy_pool

    @rate_limiter(max_qps=3)
    def fetch_page(self, url):
        """获取页面 HTML"""
        kwargs = {"timeout": REQUEST_CONFIG["timeout"]}
        if self.proxy_pool:
            proxy = self.proxy_pool.get_proxy()
            if proxy:
                kwargs["proxies"] = {"http": proxy, "https": proxy}

        resp = self.session.get(url, **kwargs)
        resp.raise_for_status()
        resp.encoding = resp.apparent_encoding
        return resp.text

    def parse_product_list(self, html):
        """解析商品列表页"""
        tree = etree.HTML(html)
        products = []

        # 京东商品列表示例解析
        items = tree.xpath('//div[@class="gl-item"]')
        for item in items:
            product = {
                "title": item.xpath('.//div[@class="p-name"]/a/em/text()'),
                "price": item.xpath('.//div[@class="p-price"]/strong/i/text()'),
                "shop": item.xpath('.//span[@class="J_im_icon"]/a/text()'),
                "url": item.xpath('.//div[@class="p-name"]/a/@href'),
            }
            # 清洗数据
            if product["title"]:
                product["title"] = "".join(product["title"]).strip()
                product["price"] = clean_price(
                    product["price"][0] if product["price"] else "0"
                )
                products.append(product)

        return products

    def scrape_category(self, category_url, max_pages=5):
        """采集整个品类"""
        all_products = []
        for page in range(1, max_pages + 1):
            url = f"{category_url}?page={page}"
            try:
                html = self.fetch_page(url)
                products = self.parse_product_list(html)
                all_products.extend(products)
                print(f"第 {page} 页: 采集 {len(products)} 个商品")
            except Exception as e:
                print(f"第 {page} 页失败: {e}")
                continue
        return all_products

3. Playwright 模式采集器

# crawlers/playwright_crawler.py
import asyncio
from playwright.async_api import async_playwright
from config import PLAYWRIGHT_CONFIG


class PlaywrightCrawler:
    """适用于动态渲染页面的采集器"""

    def __init__(self, max_pages=3):
        self.max_pages = max_pages
        self.semaphore = asyncio.Semaphore(max_pages)

    async def create_context(self, browser):
        """创建伪装上下文"""
        context = await browser.new_context(
            viewport=PLAYWRIGHT_CONFIG["viewport"],
            user_agent=PLAYWRIGHT_CONFIG["user_agent"],
            locale="zh-CN",
            timezone_id="Asia/Shanghai",
        )
        # 注入反检测脚本
        await context.add_init_script("""
            Object.defineProperty(navigator, 'webdriver', {
                get: () => undefined,
            });
        """)
        return context

    async def scrape_product_detail(self, browser, url):
        """采集商品详情页"""
        async with self.semaphore:
            context = await self.create_context(browser)
            page = await context.new_page()

            try:
                await page.goto(url, wait_until="networkidle", timeout=30000)

                # 等待关键元素加载
                await page.wait_for_selector("#product-name", timeout=5000)

                # 提取商品信息
                product = await page.evaluate("""
                    () => ({
                        title: document.querySelector('#product-name')?.innerText,
                        price: document.querySelector('.price')?.innerText,
                        description: document.querySelector('#description')?.innerText,
                        images: Array.from(
                            document.querySelectorAll('.product-image img')
                        ).map(img => img.src),
                        specs: Array.from(
                            document.querySelectorAll('.specs-table tr')
                        ).map(row => ({
                            key: row.querySelector('th')?.innerText,
                            value: row.querySelector('td')?.innerText,
                        })),
                    })
                """)
                product["url"] = url

                # 滚动到底部加载更多内容
                await page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
                await asyncio.sleep(1)

                return product

            except Exception as e:
                print(f"采集失败 {url}: {e}")
                return {"url": url, "error": str(e)}
            finally:
                await context.close()

    async def run(self, urls):
        """批量采集商品详情"""
        async with async_playwright() as p:
            browser = await p.chromium.launch(
                headless=PLAYWRIGHT_CONFIG["headless"],
                args=["--disable-blink-features=AutomationControlled"],
            )
            tasks = [self.scrape_product_detail(browser, url) for url in urls]
            results = await asyncio.gather(*tasks)
            await browser.close()
            return results

4. Scrapy 模式采集器（百度搜索结果）

# crawlers/scrapy_crawler/spiders/product_spider.py
import scrapy
from scrapy_redis.spiders import RedisSpider


class ProductSpider(RedisSpider):
    """分布式商品采集爬虫"""
    name = "product_spider"
    redis_key = "product:start_urls"

    def parse(self, response):
        """解析商品列表页"""
        products = response.css(".gl-item")

        for item in products:
            product_url = item.css(".p-name a::attr(href)").get()
            if product_url:
                if not product_url.startswith("http"):
                    product_url = "https:" + product_url

                yield scrapy.Request(
                    url=product_url,
                    callback=self.parse_detail,
                    meta={"from_list": response.url},
                )

        # 翻页
        next_page = response.css(".pn-next::attr(href)").get()
        if next_page:
            yield response.follow(next_page, callback=self.parse)

    def parse_detail(self, response):
        """解析商品详情页"""
        yield {
            "title": response.css(".sku-name::text").get("").strip(),
            "price": response.css(".p-price .price::text").get("").strip(),
            "url": response.url,
            "shop": response.css(".J-hove-wrap .ellipsis::text").get("").strip(),
        }

5. 存储模块

# storage/csv_writer.py
import csv
import os


class CsvWriter:
    def __init__(self, filepath, fieldnames):
        self.filepath = filepath
        self.fieldnames = fieldnames
        self._ensure_file()

    def _ensure_file(self):
        """确保文件存在并写入表头"""
        if not os.path.exists(self.filepath):
            with open(self.filepath, "w", newline="", encoding="utf-8-sig") as f:
                writer = csv.DictWriter(f, fieldnames=self.fieldnames)
                writer.writeheader()

    def write(self, data):
        """写入一行数据"""
        with open(self.filepath, "a", newline="", encoding="utf-8-sig") as f:
            writer = csv.DictWriter(f, fieldnames=self.fieldnames)
            writer.writerow(data)


# storage/mongodb_writer.py
from pymongo import MongoClient


class MongoWriter:
    def __init__(self, uri, db_name, collection_name):
        self.client = MongoClient(uri)
        self.db = self.client[db_name]
        self.collection = self.db[collection_name]

    def write(self, data):
        """写入数据（去重）"""
        # 使用 url 去重
        self.collection.update_one(
            {"url": data["url"]},
            {"$set": data},
            upsert=True,
        )

    def close(self):
        self.client.close()

6. 主入口

# main.py
import argparse
from config import STORAGE_CONFIG
from crawlers.requests_crawler import RequestsCrawler
from storage.csv_writer import CsvWriter
from storage.mongodb_writer import MongoWriter


def get_storage():
    """获取存储实例"""
    if STORAGE_CONFIG["type"] == "csv":
        return CsvWriter(
            filepath=STORAGE_CONFIG["csv_path"],
            fieldnames=["title", "price", "url", "shop", "created_at"],
        )
    elif STORAGE_CONFIG["type"] == "mongodb":
        return MongoWriter(
            uri=STORAGE_CONFIG["mongodb_uri"],
            db_name=STORAGE_CONFIG["mongodb_db"],
            collection_name=STORAGE_CONFIG["mongodb_collection"],
        )


def run_requests_mode(category_url, pages=5):
    """运行 Requests 模式采集"""
    print("=== 启动 Requests 模式采集 ===")
    crawler = RequestsCrawler()
    products = crawler.scrape_category(category_url, max_pages=pages)

    storage = get_storage()
    for p in products:
        storage.write(p)
    print(f"采集完成，共 {len(products)} 条数据")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="电商数据采集系统")
    parser.add_argument("--mode", choices=["requests", "playwright", "scrapy"],
                        default="requests")
    parser.add_argument("--url", required=True, help="起始 URL")
    parser.add_argument("--pages", type=int, default=5)
    args = parser.parse_args()

    if args.mode == "requests":
        run_requests_mode(args.url, args.pages)

---

实战二：新闻资讯分布式爬虫集群

项目概述

构建一个基于 Scrapy-Redis + MongoDB 的分布式爬虫集群，采集多个新闻网站的资讯内容，支持增量更新。

1. 项目结构

news_crawler/
├── settings.py
├── items.py
├── pipelines.py
├── spiders/
│   ├── __init__.py
│   ├── news_spider.py           # 通用新闻爬虫
│   └── weibo_hot_spider.py      # 微博热搜爬虫（需 Playwright）
├── middlewares.py
├── docker-compose.yml           # Redis + MongoDB 容器
└── start_crawler.sh             # 启动脚本

2. 定义 Item

# items.py
import scrapy


class NewsItem(scrapy.Item):
    title = scrapy.Field()         # 新闻标题
    url = scrapy.Field()          # 原文链接
    source = scrapy.Field()       # 来源（如 新浪新闻）
    content = scrapy.Field()      # 正文
    summary = scrapy.Field()      # 摘要
    author = scrapy.Field()       # 作者
    publish_time = scrapy.Field() # 发布时间
    keywords = scrapy.Field()     # 关键词列表
    category = scrapy.Field()     # 分类
    crawl_time = scrapy.Field()   # 采集时间
    images = scrapy.Field()       # 图片列表

3. Spider——通用新闻爬虫

# spiders/news_spider.py
import scrapy
from scrapy_redis.spiders import RedisSpider
from datetime import datetime
from news_crawler.items import NewsItem


class NewsSpider(RedisSpider):
    """分布式新闻爬虫——从 Redis 接收起始 URL"""
    name = "news_spider"
    redis_key = "news:start_urls"

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # 站点解析规则
        self.site_rules = {
            "news.sina.com.cn": {
                "title": "h1.main-title::text",
                "content": "#article .article-content p::text",
                "pub_time": ".date::text",
                "source": ".source::text",
            },
            "www.163.com": {
                "title": "h1.post_title::text",
                "content": ".post_body p::text",
                "pub_time": ".post_time::text",
                "source": ".post_info span:nth-child(1)::text",
            },
        }

    def get_rules(self, url):
        """根据 URL 获取对应站点的解析规则"""
        import tldextract
        ext = tldextract.extract(url)
        domain = f"{ext.domain}.{ext.suffix}"
        return self.site_rules.get(domain, {
            "title": "h1::text",
            "content": "article p::text",
            "pub_time": "time::text",
        })

    def parse(self, response):
        """解析新闻页面"""
        rules = self.get_rules(response.url)

        item = NewsItem()
        item["url"] = response.url
        item["title"] = response.css(rules["title"]).get("").strip()
        item["content"] = "\n".join(
            p.strip() for p in response.css(rules["content"]).getall() if p.strip()
        )
        item["summary"] = item["content"][:200] + "..." if len(item["content"]) > 200 else item["content"]

        # 发布时间
        pub_time = response.css(rules.get("pub_time", "time::text")).get("")
        item["publish_time"] = pub_time.strip()
        item["crawl_time"] = datetime.now().isoformat()

        # 来源
        source = response.css(rules.get("source", ".source::text")).get("")
        item["source"] = source.strip() or response.url.split("/")[2]

        # 提取分类（从 URL 路径推断）
        path_parts = response.url.split("/")
        item["category"] = path_parts[3] if len(path_parts) > 3 else "general"

        yield item

        # 提取页内相关链接
        for link in response.css("a[href^='http']::attr(href)").getall():
            if any(kw in link for kw in ["news", "article", "detail"]):
                yield scrapy.Request(link, callback=self.parse)

4. 微博热搜爬虫（Playwright + Scrapy）

# spiders/weibo_hot_spider.py
import scrapy
import json
from scrapy_redis.spiders import RedisSpider


class WeiboHotSpider(RedisSpider):
    """微博热搜爬虫——通过 API 接口采集"""
    name = "weibo_hot"
    redis_key = "weibo:start_urls"
    allowed_domains = ["weibo.com", "weibo.cn"]

    def parse(self, response):
        """解析微博热搜列表"""
        # 微博热搜 API（示例）
        data = json.loads(response.text)
        hot_list = data.get("data", {}).get("realtime", [])

        for item in hot_list:
            yield {
                "title": item.get("word"),
                "rank": item.get("rank"),
                "hot_score": item.get("raw_hot"),
                "category": "微博热搜",
                "url": f"https://s.weibo.com/weibo?q={item.get('word')}",
                "crawl_time": response.meta.get("crawl_time"),
            }

5. Pipeline——MongoDB 存储

# pipelines.py
from itemadapter import ItemAdapter
from datetime import datetime
import re


class NewsCleanPipeline:
    """新闻数据清洗"""

    def process_item(self, item, spider):
        adapter = ItemAdapter(item)

        # 清洗标题
        if adapter.get("title"):
            adapter["title"] = re.sub(r"\s+", " ", adapter["title"]).strip()

        # 清洗正文：去除多余空白
        if adapter.get("content"):
            adapter["content"] = re.sub(r"\n\s*\n", "\n", adapter["content"]).strip()

        # 提取关键词（简单分词模拟）
        if adapter.get("title"):
            common_words = ["的", "了", "是", "在", "和", "有", "不", "就", "也", "都"]
            words = re.findall(r"[\u4e00-\u9fa5]{2,4}", adapter["title"])
            adapter["keywords"] = [w for w in words if w not in common_words][:5]

        return item


class MongoPipeline:
    """MongoDB 存储 Pipeline"""

    def __init__(self, mongo_uri, mongo_db):
        self.mongo_uri = mongo_uri
        self.mongo_db = mongo_db

    @classmethod
    def from_crawler(cls, crawler):
        return cls(
            mongo_uri=crawler.settings.get("MONGO_URI"),
            mongo_db=crawler.settings.get("MONGO_DB", "news"),
        )

    def open_spider(self, spider):
        import pymongo
        self.client = pymongo.MongoClient(self.mongo_uri)
        self.db = self.client[self.mongo_db]

    def close_spider(self, spider):
        self.client.close()

    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        # 使用 url 去重（upsert）
        self.db["articles"].update_one(
            {"url": adapter.get("url")},
            {"$set": adapter.asdict()},
            upsert=True,
        )
        return item


class StatsPipeline:
    """统计 Pipeline——记录爬取数量"""

    def process_item(self, item, spider):
        spider.crawler.stats.inc_value("news_count")
        return item

6. 配置与部署

# settings.py
# === 分布式配置 ===
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"
REDIS_URL = "redis://redis:6379/0"  # Docker 容器名
SCHEDULER_PERSIST = True           # 持久化队列

# === MongoDB 存储 ===
MONGO_URI = "mongodb://mongo:27017"
MONGO_DB = "news_db"

# === Pipeline ===
ITEM_PIPELINES = {
    "news_crawler.pipelines.NewsCleanPipeline": 100,
    "news_crawler.pipelines.MongoPipeline": 200,
    "news_crawler.pipelines.StatsPipeline": 300,
}

# === 并发与限速 ===
CONCURRENT_REQUESTS = 32
CONCURRENT_REQUESTS_PER_DOMAIN = 8
DOWNLOAD_DELAY = 0.5
AUTOTHROTTLE_ENABLED = True
AUTOTHROTTLE_TARGET_CONCURRENCY = 4.0

# === 扩展配置 ===
EXTENSIONS = {
    "scrapy.extensions.telnet.TelnetConsole": None,  # 关闭 telnet
}

7. Docker 部署

# docker-compose.yml
version: '3.8'

services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data

  mongo:
    image: mongo:7
    ports:
      - "27017:27017"
    volumes:
      - mongo_data:/data/db

  crawler-worker-1:
    build: .
    depends_on:
      - redis
      - mongo
    environment:
      - REDIS_URL=redis://redis:6379/0
      - MONGO_URI=mongodb://mongo:27017
    command: >
      sh -c "sleep 10 &&
             scrapy crawl news_spider"

  crawler-worker-2:
    build: .
    depends_on:
      - redis
      - mongo
    environment:
      - REDIS_URL=redis://redis:6379/0
      - MONGO_URI=mongodb://mongo:27017
    command: >
      sh -c "sleep 10 &&
             scrapy crawl news_spider"

volumes:
  redis_data:
  mongo_data:

8. 运维管理

# 1. 启动基础设施
docker-compose up -d redis mongo

# 2. 向 Redis 注入新闻起始 URL
redis-cli lpush news:start_urls "https://news.sina.com.cn/china/"
redis-cli lpush news:start_urls "https://www.163.com/news/"
redis-cli lpush news:start_urls "https://www.qq.com/news/"

# 3. 启动爬虫节点
docker-compose up -d crawler-worker-1 crawler-worker-2

# 4. 查看爬取统计
# 进入 MongoDB 查看数据量
docker exec -it news-crawler_mongo_1 mongosh news_db
> db.articles.countDocuments()
> db.articles.find().limit(5).pretty()

# 5. 扩容——增加爬虫节点
docker-compose up -d --scale crawler-worker=4

# 6. 增量爬取（去重集合保留，不会重复爬已爬 URL）
# 只需注入新的 URL 即可

9. 监控与告警

# monitor.py —— 简易监控脚本
import redis
from pymongo import MongoClient
import time


class CrawlerMonitor:
    def __init__(self):
        self.redis = redis.from_url("redis://localhost:6379/0")
        self.mongo = MongoClient("mongodb://localhost:27017/")["news_db"]

    def check_queue(self):
        """检查等待队列"""
        queue_size = self.redis.llen("news:start_urls")
        print(f"[队列] 等待爬取的 URL: {queue_size}")

    def check_dupefilter(self):
        """检查去重集合"""
        dupe_size = self.redis.scard("news_spider:dupefilter")
        print(f"[去重] 已爬取去重: {dupe_size}")

    def check_storage(self):
        """检查存储"""
        count = self.mongo["articles"].count_documents({})
        print(f"[存储] MongoDB 文章总数: {count}")

    def run(self, interval=30):
        while True:
            print(f"\n=== 爬虫监控 @ {time.strftime('%Y-%m-%d %H:%M:%S')} ===")
            self.check_queue()
            self.check_dupefilter()
            self.check_storage()
            time.sleep(interval)


if __name__ == "__main__":
    monitor = CrawlerMonitor()
    monitor.run()

---

小结

1. 电商采集系统：根据场景灵活选择 Requests（静态列表页）、Playwright（动态详情页）和 Scrapy（大规模持续采集）
2. 分布式新闻集群：Scrapy-Redis 实现任务分发与去重共享，MongoDB 作为统一存储，Docker Compose 一键部署
3. 工程设计要点：配置集中管理、存储抽象层、代理池解耦、监控告警机制

练习

1. 电商采集：使用 Requests + lxml 采集京东指定品类（如手机）的商品标题、价格和链接，存储为 CSV。
2. 详情页采集：使用 Playwright 采集 5 个商品详情页，提取完整的商品规格参数表。
3. 分布式爬虫搭建：在本地启动 Redis + MongoDB（可使用 Docker），搭建一个 2 节点的 Scrapy-Redis 分布式爬虫集群。
4. 增量采集：配置 Scrapy-Redis 的持久化去重，验证重复 URL 不会被二次爬取。
5. 监控系统：编写一个简单的爬虫监控脚本，每 30 秒输出一次队列长度、已爬数量和存储总数。