scrapy

命令行工具

创建项目

scrapy startproject myproject [project_dir]

创建爬虫

scrapy genspider mydomain mydomain.com

开始爬虫

scrapy crawl spider

列出当前项目中所有可用的spider

scrapy list
#(venv) xiaogang@Mac all_tutorial % scrapy list                 
#quotes
#scrapyorg

在浏览器中打开特定的url,检查爬虫看见了什么

scrapy view <url>

使用命令行的模式对特定的url尝试解析，解析的代码可以直接放到parse中使用，确保正确性

scrapy shell  url

运行单独的爬虫文件

scrapy runspider spider_name

Scrapy.spider

基础的爬虫

import scrapy
# the basic spider class
class Simple_Demo(scrapy.Spider):
    name = 'example' # 爬虫的唯一标识
    allowed_domains = ['example.com'] # 允许爬取的域名列表
    start_urls = ['http://example.com'] # 爬虫开始爬取的URL列表
    def start_requests(self):
        urls = ['http://example.com/page1', 'http://example.com/page2']
        for url in urls:
            yield scrapy.Request(url=url, callback=self.parse, headers={'X-Custom': 'value'}) # 不写默认是start_urls
    custom_settings = {
        'DOWNLOAD_DELAY': 1,
        'USER_AGENT': 'MyBot/1.0'
    } # 爬虫特定的设置会覆盖项目全局设置

    def parse(self, response): # 默认的处理response方法，通常在call_back中指定
        self.logger.info('正在处理: %s', response.url)
        yield {
            'title': response.css('title::text').get(),
            'url': response.url
        }

    def closed(self, reason):
        self.logger.info('爬虫结束运行，原因: %s', reason)

定义规则的爬虫

import scrapy
from scrapy.linkextractors import LinkExtractor
from scrapy.spiders import CrawlSpider, Rule


class ScrapyorgSpider(CrawlSpider):
    name = "scrapyorg"
    allowed_domains = ["scrapy.org"]
    start_urls = ["https://scrapy.org"]

    rules = (Rule(LinkExtractor(allow=r"Items/",# 匹配包含"Items/"的URL
                                deny= r"shop/"),# 排除包含"shop/"的URL
                  callback="parse_item", # 处理相应的回调方法
                  follow=True),)# 是否继续跟踪从这些页面中提取的链接

    def parse_item(self, response):
        item = {}
        #item["domain_id"] = response.xpath('//input[@id="sid"]/@value').get()
        #item["name"] = response.xpath('//div[@id="name"]').get()
        #item["description"] = response.xpath('//div[@id="description"]').get()
        return item

XmlFeedSpider

class scrapy.spiders.XMLFeedSpide

XMLFeedSpider是为解析XML提要而设计的，它通过使用特定的节点名对这些提要进行迭代。

from scrapy.spiders import XMLFeedSpider
from myproject.items import TestItem

class MySpider(XMLFeedSpider):
    name = 'example.com'
    allowed_domains = ['example.com']
    start_urls = ['http://www.example.com/feed.xml']
    iterator = 'iternodes'  # This is actually unnecessary, since it's the default value
    itertag = 'item'

    def parse_node(self, response, node):
        self.logger.info('Hi, this is a <%s> node!: %s', self.itertag, ''.join(node.getall()))

        item = TestItem()
        item['id'] = node.xpath('@id').get()
        item['name'] = node.xpath('name').get()
        item['description'] = node.xpath('description').get()
        return item

CSVFeedSpider

这个spider与xmlFeedSpider非常相似，只是它迭代行，而不是节点

from scrapy.spiders import CSVFeedSpider
from myproject.items import TestItem

class MySpider(CSVFeedSpider):
    name = 'example.com'
    allowed_domains = ['example.com']
    start_urls = ['http://www.example.com/feed.csv']
    delimiter = ';'
    quotechar = "'"
    headers = ['id', 'name', 'description']

    def parse_row(self, response, row):
        self.logger.info('Hi, this is a row!: %r', row)

        item = TestItem()
        item['id'] = row['id']
        item['name'] = row['name']
        item['description'] = row['description']
        return item

xpath css

https://www.osgeo.cn/scrapy/topics/selectors.html

xpath

基本表达式

表达式 说明 示例
nodename 选择所有名为nodename的节点 div
/ 从根节点选择 /html
// 从当前节点选择匹配的节点(不考虑位置) //div
. 当前节点 ./p
.. 父节点 ../div
@ 选择属性 @class 
//div           # 选择所有div元素
/html/body      # 选择body元素(必须是html的子元素)
//div[@class]   # 选择所有带有class属性的div
//a/@href       # 选择所有a标签的href属性值
//text()        # 选择文本节点

from lxml import html


html_content = """
<!DOCTYPE html>
<html>
<head>
    <title>示例网页</title>
</head>
<body>
    <div id="header" class="page-header">
        <h1>欢迎来到我的网站</h1>
        <p class="subtitle">探索精彩内容</p>
    </div>

    <div id="content">
        <div class="article">
            <h2 class="article-title">第一篇文章</h2>
            <p class="article-meta">作者: 张三 | 日期: 2023-05-15</p>
            <div class="article-content">
                <p>这是第一篇文章的内容...</p>
                <p>继续阅读更多<a href="/read-more" class="read-more">详细内容</a></p>
            </div>
        </div>

        <div class="article" id="special-article">
            <h2 class="article-title">特别推荐文章</h2>
            <p class="article-meta">作者: 李四 | 日期: 2023-05-10</p>
            <div class="article-content">
                <p>这篇是特别推荐的内容！</p>
                <ul>
                    <li>要点1</li>
                    <li>要点2</li>
                    <li>要点3</li>
                </ul>
            </div>
        </div>
    </div>

    <div id="footer" class="page-footer">
        <p>© 2023 我的网站. 保留所有权利.</p>
    </div>
</body>
</html>

"""

from lxml import html

# 解析HTML
html_content = """上面提供的HTML代码"""
tree = html.fromstring(html_content)

# 1. 提取主标题
title = tree.xpath('//h1/text()')[0]
print(f"主标题: {title}")

# 2. 提取所有文章标题
article_titles = tree.xpath('//h2[@class="article-title"]/text()')
print("\n文章标题:")
for idx, title in enumerate(article_titles, 1):
    print(f"{idx}. {title}")

# 3. 提取特别推荐文章的要点
special_points = tree.xpath('//div[@id="special-article"]//li/text()')
print("\n特别推荐文章的要点:")
for point in special_points:
    print(f"- {point}")

# 4. 提取页脚版权信息
copyright = tree.xpath('//div[@class="page-footer"]/p/text()')[0]
print(f"\n版权信息: {copyright}")

//div[contains(@class, 'article')]  # 选择class包含"article"的div
(//div[@class='article'])[2]  # 选择第二个article div
//div[.//a[@class='read-more']]  # 选择包含read-more链接的div
//p[@class='article-meta' and contains(text(), '李四')]  # 选择作者是李四的meta信息

处理动态生成的class：

//div[contains(concat(' ', normalize-space(@class), ' article ')]  # 精确匹配class单词

处理多层嵌套：

//div[@id='content']/div[1]/div[@class='article-content']/p[1]/text()

处理空白文本：

normalize-space(//p[@class='subtitle']/text())  # 去除多余空格

css

from parsel import Selector

# 解析HTML
html_content = """上面提供的HTML代码"""
sel = Selector(text=html_content)

# 1. 提取主标题
title = sel.css('h1::text').get()
print(f"主标题: {title}")

# 2. 提取所有文章标题
article_titles = sel.css('h2.article-title::text').getall()
print("\n文章标题:")
for idx, title in enumerate(article_titles, 1):
    print(f"{idx}. {title}")

# 3. 提取特别推荐文章的要点
special_points = sel.css('#special-article li::text').getall()
print("\n特别推荐文章的要点:")
for point in special_points:
    print(f"- {point}")

# 4. 提取页脚版权信息
copyright = sel.css('div.page-footer p::text').get()
print(f"\n版权信息: {copyright}")

# 5. 提取链接和链接文本
link = sel.css('a.read-more::attr(href)').get()
link_text = sel.css('a.read-more::text').get()
print(f"\n链接: {link}, 文本: {link_text}")

部分匹配class

div[class*="article"]  /* 选择class包含"article"的div */

选择第n个元素

div.article:nth-of-type(2)  /* 选择第二个article div */

选择有特定子元素的元素

div:has(a.read-more)  /* 选择包含read-more链接的div (注意: 非CSS标准，但在jQuery和Scrapy中可用) */

多条件选择

p.article-meta:contains("李四")  /* 选择作者是李四的meta信息 */

处理动态生成的class：

div[class*="article"]  /* class包含"article" */

处理多层嵌套：

#content > div.article:first-child > div.article-content > p:first-child

提取属性值：

a::attr(href)  /* Scrapy特有语法 */

items

Items 是用于结构化爬取数据的核心组件，它们定义了爬虫将要提取的数据的固定格式和结构。

import scrapy

class Product(scrapy.Item):
    name = scrapy.Field()
    price = scrapy.Field()
    stock = scrapy.Field()
    tags = scrapy.Field()
    last_updated = scrapy.Field(serializer=str)

class ProductItem(scrapy.Item):
    name = scrapy.Field(serializer=str)
    price = scrapy.Field(serializer=float, required=True)

这个部分和Django中的models中定义数据库字段很相似，不过这里的定义会简单很多，大部分都是scrapy.Field()

item loaders

from scrapy.loader import ItemLoader
from myproject.items import Product

def parse(self, response):
    l = ItemLoader(item=Product(), response=response)
    l.add_xpath('name', '//div[@class="product_name"]')
    l.add_xpath('name', '//div[@class="product_title"]')
    l.add_xpath('price', '//p[@id="price"]')
    l.add_css('stock', 'p#stock')
    l.add_value('last_updated', 'today') # you can also use literal values
    return l.load_item()

pipeline

Pipeline(管道)是Scrapy框架中用于处理爬虫抓取到的数据的组件，它构成了Scrapy的数据处理流水线

激活pipeline

ITEM_PIPELINES = {
    'myproject.pipelines.PricePipeline': 300,
    'myproject.pipelines.JsonWriterPipeline': 800,
}

Pipeline的核心作用

数据清洗：验证、过滤和清理爬取的数据

from scrapy.exceptions import DropItem

class ValidationPipeline(object):
    def process_item(self, item, spider):
        if not item.get('price'):
            raise DropItem("Missing price in %s" % item)
        return item

数据存储：将数据保存到数据库或文件中

import pymongo

class MongoPipeline(object):
    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_DATABASE')
        )

    def open_spider(self, spider):
        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):
        self.db[spider.name].insert_one(dict(item))
        return item

去重处理：确保不存储重复数据

from scrapy.exceptions import DropItem

class DuplicatesPipeline(object):
    def __init__(self):
        self.ids_seen = set()

    def process_item(self, item, spider):
        if item['id'] in self.ids_seen:
            raise DropItem("Duplicate item found: %s" % item)
        self.ids_seen.add(item['id'])
        return item

后处理：对数据进行额外处理或转换

自定义逻辑：实现项目特定的数据处理需求

import pymongo
from itemadapter import ItemAdapter

class MongoPipeline:

    collection_name = 'scrapy_items'

    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_DATABASE', 'items')
        )

    def open_spider(self, spider):
        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):
        self.db[self.collection_name].insert_one(ItemAdapter(item).asdict())
        return item

feed export

在Scrapy项目中同时配置了数据库Pipeline和Feed Export时，数据会同时保存到两个地方- 既会保存到配置的数据库中，也会保存到Feed Export指定的输出文件中

Feed Export的核心功能

  1. 多格式支持：JSON, JSON Lines, CSV, XML等
  2. 多种存储方式：本地文件系统、FTP、S3、标准输出等
  3. 灵活配置：通过设置或命令行参数控制导出行为
  4. 分批处理：支持分批存储大型数据集
  5. 数据后处理：支持对导出数据进行编码、格式化等处理

基本使用方法

通过命令行参数使用

```bash  scrapy crawl myspider -o items.json scrapy crawl myspider -o items.jl # JSON Lines格式 scrapy crawl myspider -o items.csv

#### 通过settings.py配置

```bash
FEED_URI = 'file:///path/to/export/%(name)s_%(time)s.json'
FEED_FORMAT = 'json'
FEED_EXPORT_ENCODING = 'utf-8'
FEED_EXPORT_FIELDS = ['field1', 'field2', 'field3']