scrapy_redis实战去哪儿旅游信息爬虫(分布式爬虫实例）

前言

在这个信息爆炸的时代，网络上充斥着大量的旅游信息，而其中关于景区的介绍和评论更是琳琅满目。然而，对于想要获取特定景区信息并了解其真实评价的人来说，筛选和获取准确、有用的数据可能是一项极具挑战性的任务。为了解决这一难题，利用网络爬虫技术成为了一个高效的途径。

在这篇笔记中，我们将介绍一个针对去哪儿网（qunar.com）景区信息和评论的网络爬虫。通过 Python 的 Scrapy 框架，结合模糊匹配技术，我们将展示如何从该网站获取特定景区的相关信息，并抓取其评论内容。本文将逐步解析代码，探讨实现爬虫的思路和关键步骤，以帮助读者更好地理解和应用网络爬虫技术。

本文已对部分关键URL进行处理，本文内容仅供参考，请勿用以任何商业、违法行径
本文使用scrapy_redis实现分布式爬虫，若想了解scrapy基础，或遇到反爬较为严重的网站，想使用Selenium技术请参考：网络爬虫 - 冷月半明的专栏 - 掘金 (juejin.cn)

大致思路

整体思路是利用 Scrapy 框架发送请求获取页面信息，通过 CSS 选择器解析页面内容，使用模糊匹配技术对景区名称进行相似度匹配，提取匹配度较高的景区信息和评论内容，并最终以 JSON 格式存储数据。

分布式爬虫简介

Scrapy-Redis 是 Scrapy 框架的一个扩展，用于实现分布式爬取。它基于 Redis 数据库实现了 Scrapy 的调度器、去重集和队列，使得多个爬虫节点可以共享相同的信息，并能够高效地协作。

以下是一些关于 Scrapy-Redis 的要点：

1. 分布式爬取：Scrapy-Redis 允许多个 Scrapy 爬虫实例之间共享爬取队列和去重集合，使得爬取任务可以被多台机器分担，提高爬取效率和速度。
2. 基于 Redis 实现的调度器和去重集：Scrapy-Redis 使用 Redis 数据库作为后端存储，通过 Redis 的数据结构实现了分布式爬取任务的调度和去重。
3. 配置简单：只需简单地配置 Scrapy 项目的 settings.py 文件，即可使用 Scrapy-Redis。需要设置 Redis 的连接信息和调度器的相关参数。
4. 支持优先级队列：可以为爬取请求设置不同的优先级，使得某些请求可以优先被处理。
5. 提供示例代码和文档：Scrapy-Redis 提供了详细的文档和示例代码，便于开发者快速上手并理解其用法。
6. 调试和监控：可以通过监控 Redis 中的键来监视爬虫状态，查看队列情况以及任务分配情况。

使用 Scrapy-Redis 可以实现一个分布式的、高效的爬虫系统，使得多个爬虫节点协同工作，提升了爬取效率和稳定性。

准备工作

import urllib
import scrapy
from fuzzywuzzy import fuzz
from scrapy import Request
import pandas as pd
from ..items import QvnaItem
from scrapy_redis.spiders import RedisSpider
import json

• 模块导入： 代码中导入了必要的 Python 模块。例如 urllib、scrapy、fuzzywuzzy、pandas 等。这些模块用于处理网络请求、数据解析、相似度匹配、数据存储等。

具体实现

定义爬虫类

class QvnaSpider(RedisSpider):
    name = "qvna"
    allowed_domains =["piao.qunar.com"]
    redis_key = 'db:start_urls'

• class QvnaSpider(RedisSpider):：定义了一个名为 QvnaSpider 的 Python 类，并指定它继承自 RedisSpider。这意味着 QvnaSpider 类将继承 RedisSpider 类的所有属性和方法，可以重写或扩展父类的功能。
• name = "qvna"：设置了该爬虫的名称为 "qvna"。Scrapy 中每个爬虫的唯一标识是其名称。
• allowed_domains = ["piao.qunar.com"]：指定了爬取的目标域名。在爬取过程中，Scrapy 将只会爬取这些域名下的页面，其他域名的页面将被忽略。
• redis_key = 'db:start_urls'：指定了 Redis 中存储初始 URL 的键值名称为 'db:start_urls'。这个键值对应着爬虫启动时从 Redis 中读取初始 URL 的地方。在分布式爬取时，起始 URL 会存储在 Redis 的特定键中，供多个爬虫节点共享使用。

生成URL队列

def start_requests(self) :
    # 读取景区文件

    # 用pandas读取.xlsx文件
    df = pd.read_excel("D:\code\Scrapy\scrapy_tour\A级景区（按省份）.xlsx")
    scenic_namelist = df['景区名称']
    dflen = len(scenic_namelist)  # 执行多少行

    for i in range(0,dflen):
        key = scenic_namelist[i]
        newurl = '************' + key + '&region=&from=mpl_search_suggest'
        # print(newurl)
        re=Request(url=newurl,headers=header,meta={"use_selenium":False},callback=self.parse)
        re.meta["data"] = {"oldtitle":key}
        re.meta["data"]["id"]=i
        yield re

这段代码是一个 Scrapy 爬虫的起始方法 start_requests。它从一个名为"A级景区（按省份）.xlsx"的 Excel 文件中读取景区名称，然后构建相应的 URL。接着生成针对这些 URL 的请求，并指定回调函数 parse 来处理响应。每个请求都包含了一个 data 字典，其中包含了景区名称和索引。通过循环处理每个景区名称，生成对应的请求，最终开始爬取。

parse

def parse(self, response):

    def get_similarity(oldtitle, newtitle):
        # 模糊匹配
        # 两个字符串之间的相似度（得分越高表示相似度越高）
        similarity_score = fuzz.ratio(oldtitle, newtitle)
        # 部分字符串匹配的得分
        partial_score = fuzz.partial_ratio(oldtitle, newtitle)
        # 排序匹配得分（处理单词顺序不同的情况）
        token_sort_score = fuzz.token_sort_ratio(oldtitle, newtitle)
        ans = [oldtitle, newtitle, similarity_score, partial_score, token_sort_score,
               similarity_score + partial_score + token_sort_score]
        return ans

    sight_elements = response.css('.search_result .sight_item[data-sight-name]')

    Similarity_score = []
    for element in  sight_elements:
        title = element.attrib['data-sight-name']
        Similarity_score.append(get_similarity(response.meta.get("data")["oldtitle"], title))

    max_score = None
    max_index = None

    if Similarity_score != []:
        for index, score in enumerate(Similarity_score):
            if max_score == None or max_score[-1] < score[-1]:
                max_score = score
                max_index = index

    if max_score != None and max_score[2] >= 50 and max_score[3] >= 50 and max_score[4] >= 50:
        # print('max', max_score)
        # print(sight_elements[max_index].attrib['data-sight-name'])
        newurl = "https://piao.qunar.com" + sight_elements[max_index].css(
        '.sight_item_detail .sight_item_about .sight_item_caption a::attr(href)').get()

        price = sight_elements[max_index].css(
        '.sight_item_detail .sight_item_pop .sight_item_price em::text').get()
        re = Request(url=newurl, headers=header, meta={"use_selenium": False}, callback=self.parse_second)
        response.meta["data"]["title"]=sight_elements[max_index].attrib['data-sight-name']
        re.meta["data"]=response.meta["data"]
        re.meta["data"]["price"]=price
        yield re

这段代码是爬虫中的一个解析方法 parse。它主要执行以下操作：

1. 相似度计算： 使用 fuzz 库中的不同方法来计算传入的景区名称和页面中景区名称的相似度。主要有 fuzz.ratio、fuzz.partial_ratio 和 fuzz.token_sort_ratio。
2. 获取页面元素： 使用 CSS 选择器定位到页面中的景区元素。
3. 计算相似度得分： 对每个页面中的景区名称计算其与目标名称的相似度得分，并将得分存储在列表 Similarity_score 中。
4. 找出最相似的景区： 检查计算得到的相似度得分列表，找到得分最高且满足一定条件（相似度得分都大于等于50）的景区。
5. 构建请求： 如果找到相似度足够高的景区，则构建一个新的请求，访问该景区的详细页面，并传递一些元数据（如价格、标题等），以便下一步处理。

该方法的主要作用是在爬取的页面中查找与目标景区名称相似的景区，然后提取相关信息并构建新的请求进一步获取更详细的数据。

一些优化的相关思考：
如果想要保证数据的准确性，那么一个高效且准确的匹配算法是必要的。
在上述代码中相似度计算只是对于 目标景点的名称和从去哪引擎过滤过一遍的相似景区名称 的比较，之所以选择这种方式是因为经过测试 去哪搜索引擎 的准确率较高，经过匹配后基本上能保证获取较为准确的数据，然而，有些旅游景点的搜索引擎准确率并没有那么高（比如飞猪），此时我们可以配合经纬度进行匹配。在很多景区列表的item里经纬度信息都会作为标签属性存在，而不会直接渲染，我们可以通过css选择器或XPath去获取。然后进行计算。

大致思路如下：
使用了两种不同的相似度度量方式：名称相似度和地理位置相似度，并通过加权计算得到最终的综合相似度。

1. 地理位置相似度计算（Geographic Similarity）：
   计算两个景区之间的地理位置相似度。这是通过计算两个景区之间的地理距离，并将其归一化为 0 到 1 之间的值来实现的。通常会使用球面距离公式（如 Haversine 公式）来计算地理距离。

2. 加权相似度计算（Weighted Similarity）：
   最后，对名称相似度和地理位置相似度进行加权求和，得到最终的综合相似度。这里通过权衡考虑了两个相似度度量的重要性。

3. calculate_distance 函数：

   计算两个经纬度点之间的距离通常会使用球面距离公式，例如 Haversine 公式。下面是一个简化的示例，它使用 Haversine 公式来计算地球表面上两个点之间的球面距离：

示例代码如下：

from math import radians, sin, cos, sqrt, atan2
def calculate_distance(location1, location2):
    lat1, lon1 = radians(location1[0]), radians(location1[1])
    lat2, lon2 = radians(location2[0]), radians(location2[1])
    # Haversine formula
    dlon = lon2 - lon1
    dlat = lat2 - lat1
    a = sin(dlat / 2)**2 + cos(lat1) * cos(lat2) * sin(dlon / 2)**2
    c = 2 * atan2(sqrt(a), sqrt(1 - a))
    radius_earth = 6371  # Earth radius in kilometers
    distance = radius_earth * c
    return distance

这段代码会返回两个经纬度点之间的球面距离，单位为公里。

# 计算经纬度相似度
distance = calculate_distance(location1, location2) 
geographic_similarity = 1 - distance / (max_distance * 1.0) 
# 将距离转化为相似度，最大距离设为max_distance 
print(f"The geographic similarity between {location1} and {location2} is {geographic_similarity}.") 
# 计算加权相似度 
weighted_similarity = 0.3 * name_similarity + 0.7 * geographic_similarity 
# 这里的0.3和0.7是权重，可以根据需要调整

然后根据名称相似度和经纬度得分去进行计算，可根据参数的具体情况去判断权重的选值。
上述思考未进行实践，有兴趣的朋友可以自己试试。

parse_second

def parse_second(self, response):
    # 进入景点详情页，解析景点id，发送第一个请求，获取评论数量以进一步行动
    print(response.meta["data"])
    sightid = response.css("#mp-tickets-new").attrib['data-sightid']
    # print("景点id",sightid)
    newurl= "*************?sightId="+sightid+"&index=1&page=1&pageSize=10&tagType=0"
    re = Request(url=newurl, headers=header, meta={"use_selenium": False}, callback=self.parse_third)
    re.meta["data"] = response.meta["data"]
    re.meta["data"]["sightid"] = sightid
    re.meta["result_flag"] = 1
    re.meta["data"]["all_results"]=[]
    yield re

这段代码实现了在进入景点详情页后，解析景点ID，发送第一个请求以获取评论数量，并进行下一步操作。

1. 通过 CSS 选择器定位景点ID信息：sightid = response.css("#mp-tickets-new").attrib['data-sightid']
2. 构建新的 URL，以获取景点评论的 JSON 数据。这个 URL 包含了景点ID、页码、每页数量等信息。
3. 创建一个新的 Request 对象 re，并使用之前解析的景点ID，配置相关的元数据信息。meta 中包括了爬虫需要的一些信息，如使用 Selenium、结果标志、评论列表等。
4. 最后通过 yield re 将新的请求对象 re 返回，以便进一步进行评论数据的爬取和处理。

这段代码的目的是获取景点的评论信息，准备好第一个请求，并将相关的元数据信息添加到请求中，以便在接下来的请求中使用这些信息进行进一步处理和爬取。

parse_third

def parse_third(self, response):
    if response.status != 200:
        scrapy.Spider.logger.error(f"异常状态码，可能被识别")
        return None
    # 将获取到的数据添加进数组

    # 大于五十条评论，只存储前五十条
    try:
        if response.json()["data"].get("commentCount") is not None and response.json()["data"].get("commentCount") >= 50:
            response.meta["data"]["all_results"].append(response.json()["data"]["commentList"])
            numlist = range(2, 6)
        elif response.json()["data"].get("commentCount") <= 10 and response.json()["data"].get("commentCount") != 0:
            response.meta["data"]["all_results"].append(response.json()["data"]["commentList"])
            numlist = None
        elif response.json()["data"].get("commentCount") ==  None or response.json()["data"].get("commentCount") == 0 :
            return None
        else:
            numlist = range(2, int(response.json()["data"]["commentCount"] / 10) + 1)
        if response.meta["result_flag"] == 1 and numlist != None:
            # 第一页请求，且不仅有一页评论，循环发送请求获取剩下的评论
            for i in numlist:
                newurl = "?sightId=" + \
                         response.meta["data"]["sightid"] + "&index=" + str(i) + "&page=" + str(
                    i) + "&pageSize=10&tagType=0"
                re = Request(url=newurl, headers=header, meta={"use_selenium": False}, callback=self.parse_third)
                re.meta["data"] = response.meta["data"]
                if i != numlist[-1]:
                    re.meta["result_flag"] = i
                else:
                    re.meta["result_flag"] = -1
                yield re
        elif (response.meta["result_flag"] == 1 and numlist == None) or response.meta["result_flag"] == -1:
            # 仅有一页评论，结束爬虫
            if response.meta["data"]["all_results"] == []:
                return None
            json_str = json.dumps(response.meta["data"]["all_results"], ensure_ascii=False)
            qvna_item = QvnaItem()
            qvna_item['Price'] = response.meta['data']['price']
            qvna_item['Title'] = response.meta['data']['oldtitle']
            qvna_item['Id'] = response.meta['data']['id']
            qvna_item['Commentlist'] = json_str
            # 打印转换后的 JSON 字符串
            # print( qvna_item )
            print("进入管道")
            response.meta["data"]["all_results"].clear()
            yield qvna_item
    except Exception as e:
        scrapy.Spider.logger.error(f"Error: qvna爬虫报错，{e}")
        return None

上述这段代码的功能是解析第二个请求的响应，处理景点评论信息，存储获取的评论数据并组装成 QvnaItem 对象返回。

1. 首先检查响应的状态码是否为200，若不是则记录错误并返回None。

2. 尝试解析评论信息并存储到数组all_results中。

   • 若评论数量超过50条，则只存储前50条评论。
   • 若评论数量小于等于10条且不为0，则将所有评论存储。
   • 若评论数量为None或0，则返回None。
   • 否则，根据评论数量计算页数，然后通过循环发送请求获取所有评论。

3. 如果是第一页请求且有多页评论，则循环发送请求获取剩下的评论。

   • 构建请求的 URL，根据评论数量、页码等信息。
   • 创建新的 Request 对象 re，添加了相应的元数据信息，包括result_flag来表示当前处理的页码。

4. 如果只有一页评论或是最后一页评论的请求，则结束爬取并返回 QvnaItem 对象。

   • 若评论为空列表，则返回None。
   • 将所有评论信息转换为 JSON 字符串，创建 QvnaItem 对象，并填充相关字段。
   • 清空all_results列表，并返回 QvnaItem 对象。

5. 如果在处理过程中出现异常，记录错误并返回None。

item

class QvnaItem(scrapy.Item):
    Commentlist = scrapy.Field()
    Price = scrapy.Field()
    Title = scrapy.Field()
    Id = scrapy.Field()

定义了一个 Scrapy Item 类 QvnaItem，它是用于存储爬取到的数据的容器。
在这个 Item 类中，定义了以下字段：

• Commentlist: 用于存储景点评论的信息，通常是一个 JSON 字符串。
• Price: 用于存储景点的价格信息。
• Title: 存储景点的标题或名称。
• Id: 存储景点的唯一标识符或 ID。
  这些字段将会在爬取过程中用来存储相应的数据，每次爬取到新的数据时，会创建一个 QvnaItem 对象，并将数据存储在这些字段中，最终被传递到 Item Pipeline 进行后续处理。

piplines

class MySQLPipeline:
    def __init__(self, mysql_host, mysql_port, mysql_database, mysql_user, mysql_password):
        self.mysql_host = mysql_host
        self.mysql_port = mysql_port
        self.mysql_database = mysql_database
        self.mysql_user = mysql_user
        self.mysql_password = mysql_password
        self.conn = None
        self.cursor = None

        self.data = []

    @classmethod
    def from_crawler(cls, crawler):
        return cls(
            mysql_host=crawler.settings.get('MYSQL_HOST'),
            mysql_port=crawler.settings.get('MYSQL_PORT'),
            mysql_database=crawler.settings.get('MYSQL_DATABASE'),
            mysql_user=crawler.settings.get('MYSQL_USER'),
            mysql_password=crawler.settings.get('MYSQL_PASSWORD'),
        )


    def open_connection(self):
        # 手动开启数据库连接
        # print(self.mysql_port)
        # print(self.mysql_user)
        # print(self.mysql_password)
        # print(self.mysql_host)
        # print(self.mysql_database)
        self.conn = mysql.connector.connect(
            host=self.mysql_host,
            port=self.mysql_port,
            database=self.mysql_database,
            user=self.mysql_user,
            password=self.mysql_password,
        )
        self.cursor = self.conn.cursor()

    def open_spider(self, spider):
        self.conn = mysql.connector.connect(
            host=self.mysql_host,
            port=self.mysql_port,
            database=self.mysql_database,
            user=self.mysql_user,
            password=self.mysql_password,
        )
        self.cursor = self.conn.cursor()

    def close_spider(self, spider):
        self.conn.close()
        if  self.data:
            sql = "INSERT INTO xiecheng_data (id,title, commentlist,averagescore,opentime,number) VALUES (%s,%s, %s,%s, %s,%s)"
            self.write_data(sql=sql)

    def process_item(self, item, spider):
        if not self.conn or not self.cursor:
            # 如果连接或游标未初始化，则手动开启数据库连接
            self.open_connection()
        if isinstance(item, XiechengItem):
            # 在这里执行将item数据存入MySQL的操作
            # 例如，假设你有一个名为 "your_table" 的表，且item中包含字段 "field1" 和 "field2"
            print("执行成功")
            # print(item)
            sql = "INSERT INTO xiecheng_data (id,title, commentlist,averagescore,opentime,number) VALUES (%s,%s, %s,%s, %s,%s)"
            values = (
            item['Id'], item['Title'], item['Commentlist'], item['AverageScore'], item['OpenTime'], item['Number'])
            self.cursor.execute(sql, values)
            self.conn.commit()

        elif isinstance(item, QvnaItem):

            sql = "INSERT INTO qvna_data (id,Title, Commentlist,Price) VALUES (%s,%s,%s,%s)"
            values = (
                item['Id'], item['Title'], item['Commentlist'], item['Price'])
            self.data.append(values)
            print(len(self.data))
            if len(self.data) == 5:
                self.write_data(sql)

        elif isinstance(item, ZhihuItem):

            sql = "INSERT INTO zhihu_data (Id,Title, Commentlist) VALUES (%s,%s,%s)"
            values = (
                item['Id'], item['Title'], item['Commentlist'])
            self.data.append(values)
            print(len(self.data))
            if len(self.data) == 10:
                self.write_data(sql)
        return item

    def write_data(self, sql):

        self.cursor.executemany(sql, self.data)
        self.conn.commit()
        self.data.clear()
        print("提交完成")

这段代码是一个自定义的 Scrapy Pipeline，名为 MySQLPipeline，它用于将爬取到的数据存储到 MySQL 数据库中。

• __init__ 方法用于初始化连接 MySQL 数据库所需的参数，并创建一个空列表 self.data 用于暂存待写入数据库的数据。
• from_crawler 类方法是一个工厂方法，用于从 Scrapy 的配置中获取数据库连接所需的参数。
• open_connection 方法用于手动开启数据库连接。
• open_spider 方法在爬虫启动时调用，用于开启数据库连接。
• close_spider 方法在爬虫关闭时调用，用于关闭数据库连接，并将暂存的数据写入数据库。
• process_item 方法用于处理爬取到的数据。根据不同的 Item 类型，将数据存入相应的数据表中。如果是 XiechengItem，直接将数据插入到名为 xiecheng_data 的表中；如果是 QvnaItem 或 ZhihuItem，则将数据暂存到 self.data 列表中，当列表中的数据达到一定量（5 条或 10 条）时，再进行批量写入数据库操作。