使用 Python 和 Scrapy 半小时爬了10 个在线商店的网页

#安装 Scrapy
$ pip install scrapy
#安装用于下载产品图片的图像
$ pip install image
#使用 Scrapy 开始网页抓取项目
$ scrapy startproject fashionWebScraping
$ cd fashionWebScraping
$ ls
#创建项目文件夹，如下所述
$ mkdir csvFiles
$ mkdir images_scraped
$ mkdir jsonFiles
$ mkdir utilityScripts

https://www.derimod.com.tr/kadin-canta-aksesuar/?page=1

# fashionWebScraping 文件夹中的 items.py
import scrapy
from scrapy.item import Item, Field
class FashionwebscrapingItem(scrapy.Item):
 
 #与产品相关的项，如 Id、名称、价格等
 gender=Field()
 productId=Field()
 productName=Field()
 priceOriginal=Field()
 priceSale=Field()
#要存储链接的项
 imageLink = Field()
 productLink=Field()
#公司名称项
 company = Field()
pass
class ImgData(Item):
#用于下载产品图像的图像管道项
 image_urls=scrapy.Field()
 images=scrapy.Field()

# fashionWebScraping 文件夹中的 settings.py
# fashionWebScraping 项目的 Scrapy 设置
# 为简单起见，此文件仅包含被认为重要或常用的设置。你可以参考文档找到更多的设置：
# https://doc.scrapy.org/en/latest/topics/settings.html
# https://doc.scrapy.org/en/latest/topics/downloader-middleware.html
# https://doc.scrapy.org/en/latest/topics/spider-middleware.html
BOT_NAME = 'fashionWebScraping'
SPIDER_MODULES = ['fashionWebScraping.spiders']
NEWSPIDER_MODULE = 'fashionWebScraping.spiders'
# 通过在用户代理上标识自己（和你的网站），负责任的抓取。
USER_AGENT = 'fashionWebScraping'
# 遵守 rebots.txt 规则
ROBOTSTXT_OBEY = True
# 参阅 https://doc.scrapy.org/en/latest/topics/settings.html
# 下载延迟
# 另请参阅 autothrottle 的设置和文档
# 这样可以避免对服务器造成太大的压力
DOWNLOAD_DELAY = 1
# 重写默认的请求报头：
DEFAULT_REQUEST_HEADERS = {
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
'Accept-Language': 'tr',
}
# 配置项目管道
# 参阅 https://doc.scrapy.org/en/latest/topics/item-pipeline.html
ITEM_PIPELINES = {'scrapy.pipelines.images.ImagesPipeline': 1}
IMAGES_STORE = '/Users/erdemisbilen/Angular/fashionWebScraping/images_scraped'

fashionWebScraping $ scrapy genspider fashionBOYNER boyner.com 

# fashionWebScraping/Spiders 文件夹中的 'fashionBOYNER.py' 文件
# 导入 scrapy 和 scrapy 项
import scrapy
from fashionWebScraping.items import FashionwebscrapingItem
from fashionWebScraping.items import ImgData
from scrapy.http import Request
# 从 csv 文件读取
import csv
class FashionboynerSpider(scrapy.Spider):
 name = 'fashionBOYNER'
 allowed_domains = ['BOYNER.com']
 start_urls = ['http://BOYNER.com/']
# 此函数通过跟踪 csv 文件中的起始 URL 来帮助我们抓取网站的全部内容
def start_requests(self):
# 从 csv 文件读取主类别 URL
with open ("/Users/erdemisbilen/Angular/fashionWebScraping/
  csvFiles/SpiderMainCategoryLinksBOYNER.csv", "rU") as f:
  
    reader=csv.DictReader(f)
for row in reader:
      url=row['url']
# 以增量方式更改 page 值来浏览产品列表
      # 你可以根据最大值来调整产品数量的范围值，默认是抓取 30 个网页
      link_urls = [url.format(i) for i in range(1,30)]
for link_url in link_urls:
        print(link_url)
# 将包含产品的每个链接传递给带有性别元数据的 parse_ product_pages 函数
request=Request(link_url, callback=self.parse_product_pages,
        meta={'gender': row['gender']})
yield request
# 此函数在 xpath 的帮助下抓取网页数据
 def parse_product_pages(self,response):
 
  item=FashionwebscrapingItem()
  
  # 获取 HTML 块，其中列出了所有的产品。
  # <div> HTML 元素，具有"product-list-item" 类名
content=response.xpath('//div[starts-with(@class,"product-list-
  item")]')
# 循环遍历内容中的每个<div>元素
  for product_content in content:
image_urls = []
# 获取产品详细信息并填充项
   item['productId']=product_content.xpath('.//a/@data
   -id').extract_first()
item['productName']=product_content.xpath('.//img/@title').
   extract_first()
item['priceSale']=product_content.xpath('.//ins[@class=
   "price-payable"]/text()').extract_first()
item['priceOriginal']=product_content.xpath('.//del[@class=
   "price-psfx"]/text()').extract_first()
if item['priceOriginal']==None:
    item['priceOriginal']=item['priceSale']
item['imageLink']=product_content.xpath('.//img/
   @data-original').extract_first()
   
   item['productLink']="https://www.boyner.com.tr"+
   product_content.xpath('.//a/@href').extract_first()
image_urls.append(item['imageLink'])
item['company']="BOYNER"
   item['gender']=response.meta['gender']
if item['productId']==None:
    break
yield (item)
# 下载 image_urls 中包含的图像
   yield ImgData(image_urls=image_urls)
def parse(self, response):
  pass

# // 从当前节点中选择与所选内容匹配的节点，无论它们在何处
#//'//div[starts-with(@class,"product-list-item")]'选择所有的 div 元素，这些元素具有类值 start
content = response.xpath('//div[starts-with(@class,"product-list-item")]')

# 遍历内容中的每个 <div> 元素
  for product_content in content:
  
  image_urls = []
  
# 获取产品详细信息并填充项
   # ('.//a/@data-id') 提取 product_content 中的 <a> 元素的 'data-id' 值
 item['productId']=product_content.xpath('.//a/@data
   -id').extract_first()
# ('.//img/@title') 提取 product_content 中的 <img> 元素的 `title` 值 
   item['productName']=product_content.xpath('.//img/@title').
   extract_first()
# ('.//ins[@class= "price-payable"]/text()') 提取 <ins> 元素的文本值，该元素具有 product_content 中的 `price-payable` 类别属性。
   item['priceSale']=product_content.xpath('.//ins[@class=
   "price-payable"]/text()').extract_first()
# ('.//del[@class="price-psfx"]/text()') 提取 <del> 元素的文本值，该元素具有 product_content 中的 `price-psfx` 类属性。
   item['priceOriginal']=product_content.xpath('.//del[@class=
   "price-psfx"]/text()').extract_first()
   
if item['priceOriginal']==None:
     item['priceOriginal']=item['priceSale']
     
# ('.//img/@data-original') 提取 product_content 中的 <img> 元素的 `data-original` 值
   item['imageLink']=product_content.xpath('.//img/
   @data-original').extract_first()
# ('.//a/@href') 提取 product_content 中的 <a> 元素的 `href`值
   item['productLink']="https://www.boyner.com.tr"+
   product_content.xpath('.//a/@href').extract_first()
   
# 将产品图片链接分配到图像管道定义的 `image_urls` 中
   image_urls.append(item['imageLink'])
   
   item['company']="BOYNER"
    item['gender']=response.meta['gender']
   
   if item['productId']==None:
    break
    
   yield (item)
   **# 下载 image_urls 中包含的图像
   yield ImgData(image_urls=image_urls)

fashionWebScraping $ scrapy crawl -o rawdata_BOYNER.json -t jsonlines fashionBOYNER

# fashionWebScraping/utilityScripts  文件夹中的 `jsonPrep.py`文件
import json
import sys
from collections import OrderedDict
import csv
import os
# 从 jsonFiles.csv 文件中读取需要验证的所有 json 文件的名称和位置
with open("/Users/erdemisbilen/Angular/fashionWebScraping/csvFiles/ jsonFiles.csv", "rU") as f:reader=csv.DictReader(f)
# 迭代 jsonFiles.csv 中列出的 json 文件
 for row in reader:
 # 从 jsonFiles.csv 文件中读取 jsonFile_raw 列
  jsonFile=row['jsonFile_raw']
  
# 打开 the jsonFile
   with open(jsonFile) as json_file:
    data = []
    i = 0
    
   seen = OrderedDict()
   # 对 json 文件中的行进行迭代
    for d in json_file:
     seen = json.loads(d)
     
# 如果产品 Id 为空，则不包括行项
    try:
      if seen["productId"] != None:
       for key, value in seen.items():
        print("ok")
        i = i + 1
        data.append(json.loads(d))
     except KeyError:
      print("nok")
    print (i)
    baseFileName=os.path.splitext(jsonFile)[0]
    
# 通过从 `file_name_prep` 行读取 filename，将结果写成 json 文件
 with open('/Users/erdemisbilen/Angular/fashionWebScraping/
    jsonFiles/'+row['file_name_prep'], 'w') as out:
    
json.dump(data, out)

# fashionWebScraping/utilityScripts 文件夹中的 'deldub.py' 文件
import json
import sys
from collections import OrderedDict
import csv
import os
# 从 jsonFiles.csv 文件中读取所有 json 文件的名称和位置，需要验证重复的行项
with open("/Users/erdemisbilen/Angular/fashionWebScraping/csvFiles/ jsonFiles.csv", newline=None) as f:
 reader=csv.DictReader(f)
 
# 迭代 jsonFiles.csv 中列出的 json 文件
 for row in reader:
 
# 从 jsonFile.csv 文件中读取 jsonFile_raw 列
  jsonFile=row['jsonFile_prep']
  
# 打开 jsonFile
  with open(jsonFile) as json_file:
   data = json.load(json_file)
   
seen = OrderedDict()
   dubs = OrderedDict()
   
# 对 json 文件的行进行迭代
   for d in data:
    oid = d["productId"]
    
# 如果产品 Id 具有重复值，则不包括该项
    if oid not in seen:
     seen[oid] = d
     
    else:
     dubs[oid]=d
     
   baseFileName=os.path.splitext(jsonFile)[0]
   
# 通过从 `file_name_final` 读取 filename，将结果写成 json 文件
with open('/Users/erdemisbilen/Angular/fashionWebScraping/
     jsonFiles/'+row['file_name_final'], 'w') as out:
      json.dump(list(seen.values()), out)
      
with open('/Users/erdemisbilen/Angular/fashionWebScraping/
     jsonFiles/'+'DELETED'+row['file_name_final'], 'w') as out:
      json.dump(list(dubs.values()), out)

$ python3 --version
Python 3.7.3$ virtualenv --version
15.2.0$ cd /path/to/my/airflow/workspace
$ virtualenv -p `which python3` venv
$ source venv/bin/activate
(venv) $ pip install apache-airflow
(venv) $ mkdir airflow_home
(venv) $ export AIRFLOW_HOME=`pwd`/airflow_home
(venv) $ airflow initdb
(venv) $ airflow webserver

(venv) $ mkdir dags

# Airflow dag 文件夹中的'fashionsearch_dag.py'文件
import datetime as dt
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
default_args = {
'owner': 'airflow',
'depends_on_past': False,
'start_date': datetime(2019, 11, 23),
'retries': 1,
'retry_delay': timedelta(minutes=5),
# 'queue': 'bash_queue',
# 'pool': 'backfill',
# 'priority_weight': 10,
# 'end_date': datetime(2016, 1, 1),
}
dag = DAG(dag_id='fashionsearch_dag', default_args=default_args, schedule_interval=timedelta(days=1))
#此任务将删除在以前抓取会话中生成的所有 json 文件
t1 = BashOperator(
task_id='delete_json_files',
bash_command='run_delete_files',
dag=dag)
# 此任务运行 www.boyner.com 的网络蜘蛛，并使用所抓取的数据填充相关的 json 文件
t2 = BashOperator(
task_id='boyner_spider',
bash_command='run_boyner_spider',
dag=dag)
# 此任务运行 www.derimod.con 的网络蜘蛛，并使用抓取的数据填充相关的 json 文件
t3 = BashOperator(
task_id='derimod_spider',
bash_command='run_derimod_spider',
dag=dag)
# 此任务运行 www.hepsiburada.com 的网络蜘蛛，并使用抓取的数据填充相关的 json 文件
t4 = BashOperator(
task_id='hepsiburada_spider',
bash_command='run_hepsiburada_spider',
dag=dag)
# 此任务运行 www.hm.com 的网络蜘蛛，并使用抓取的数据填充相关的 json 文件
t5 = BashOperator(
task_id='hm_spider',
bash_command='run_hm_spider',
dag=dag)
# 此任务运行 www.koton.com 的网络蜘蛛，并使用抓取的数据填充相关的 json 文件
t6 = BashOperator(
task_id='koton_spider',
bash_command='run_koton_spider',
dag=dag)
# 此任务运行 www.lcwaikiki.com 的网络蜘蛛，并使用抓取的数据填充相关的 json 文件
t7 = BashOperator(
task_id='lcwaikiki_spider',
bash_command='run_lcwaikiki_spider',
dag=dag)
# 此任务运行 www.matmazel.com 的网络蜘蛛，并使用抓取的数据填充相关的 json 文件
t8 = BashOperator(
task_id='matmazel_spider',
bash_command='run_matmazel_spider',
dag=dag)
# 此任务运行 www.modanisa.com 的网络蜘蛛，并使用抓取的数据填充相关的 json 文件
t9 = BashOperator(
task_id='modanisa_spider',
bash_command='run_modanisa_spider',
dag=dag)
# 此任务运行 www.morhipo.com 的网络蜘蛛，并使用抓取的数据填充相关的 json 文件
t10 = BashOperator(
task_id='morhipo_spider',
bash_command='run_morhipo_spider',
dag=dag)
# 此任务运行 www.mudo.com 的网络蜘蛛，并使用抓取的数据填充相关的 json 文件
t11 = BashOperator(
task_id='mudo_spider',
bash_command='run_mudo_spider',
dag=dag)
# 此任务运行 www.trendyol.com 的网络蜘蛛，并使用抓取的数据填充相关的 json 文件
t12 = BashOperator(
task_id='trendyol_spider',
bash_command='run_trendyol_spider',
dag=dag)
# 此任务运行 www.yargici.com 的网络蜘蛛，并使用抓取的数据填充相关的 json 文件
t13 = BashOperator(
task_id='yargici_spider',
bash_command='run_yargici_spider',
dag=dag)
# 此任务检查并删除 json 文件中的空行项
t14 = BashOperator(
task_id='prep_jsons',
bash_command='run_prep_jsons',
dag=dag)
# 此任务检查并删除 json 文件中的重复行项
t15 = BashOperator(
task_id='delete_dublicate_lines',
bash_command='run_del_dub_lines',
dag=dag)
# 此任务使用 JSON 文件中的数据填充远程 ES 集群
t16 = BashOperator(
task_id='json_to_elasticsearch',
bash_command='run_json_to_es',
dag=dag)
# 对于顺序执行器，所有任务都依赖于前一个任务
# 不可能并行执行任务
# 至少在执行并行任务时使用本地执行器
t1 >> t2 >> t3 >> t4 >> t5 >> t6 >> t7 >> t8 >> t9 >> t10 >> t11 >> t12 >> t13 >> t14 >> t15 >> t16

(venv) $ airflow scheduler