AWS Rekognition を使う時にクローラーも使ってなんかできないかなと思い scrapy を利用してみました。とりあえず今回はドメインと画像収集のところまで。いかがわしいことには絶対利用しないでください

今回はスタートのページからどんどんリンクを辿り、ドメイン名のフォルダごとに、辿った時のページの画像を保存します。今度そのフォルダごとに画像を AWS Rekognition に投げて、そのドメインがどんなドメインなのかを画像から判別しようと考えています。

前提

scrapy 1.5.0
python3
scrapy インストール済み

参考サイト

Spider のコード

クローラーの肝となる部分です。参考サイトではCrawlSpiderクラスを継承して利用している場合が多かったです。そっちの方が大抵の場合は楽だと思います。

WebSpider.py

# -*- coding: utf-8 -*-
  import scrapy
  from tutorial.items import TutorialItem
  import re
  from scrapy.exceptions import NotSupported
  from urllib.parse import urlparse


  class WebSpider(scrapy.Spider):
      name = 'web'
      # 見つけたドメインを入れる
      tracked_domains = []
      # 全てを対象
      allowed_domains = []
      # 最初に見に行くサイト
      start_urls = ['http://XXXXXXXXXXXXX']

      # response　を毎回処理する関数
      def parse(self, response):
          try:
              # データ処理
              # この関数内の処理が終わると続きを実行する
              # dataPipeline を利用した場合もここに戻って来る
              yield self.parse_items(response)

              # リンクを辿る
              for link in response.xpath('//@href').extract():
                  if re.match(r"^https?://", link):
                      yield scrapy.Request(link, callback=self.parse)
          except NotSupported:
              # GET のレスポンスが txt じゃなかった場合
              # Spiders の logging 機能が用意されているのでそれを利用
              self.logger.info("Raise NotSupported")

      # ドメインごとにページに表示された画像を保存する
      def parse_items(self, response):
          # domain の抽出
          url = response.url
          parsed_url = urlparse(url)
          domain = parsed_url.netloc

          # 同じ Domain は一回しかチェックしない
          if domain in self.tracked_domains:
              return

          self.tracked_domains.append(domain)

          item = TutorialItem()
          item['domain'] = domain

          # title の抽出
          title = response.xpath(r'//title/text()').extract()
          if len(title) > 0:
              item['title'] = title[0]
          else:
              item['title'] = None

          # 画像 URL をセット
          item["image_urls"] = []
          for image_url in response.xpath("//img/@src").extract():
              if "http" not in image_url:
                  item["image_urls"].append(response.url.rsplit("/", 1)[0]
                         + "/" + image_url)
              else:
                  item["image_urls"].append(image_url)

          # item を返すと datapipeline に渡される
          return item

start_urlsに設定したURLを元にクローラーが動きます。
私はそんなことはしませんが、ここにいかがわしいサイトを指定するとリンクを辿って新しいいかがわしいサイトのドメインが見つかるかもしれません。私はそんなことしませんが。

基本的にはparse()関数が scrapy のレスポンスごとに呼ばれて処理を行います。
今回は途中でparse_items()を呼び出し、まだ保存していないドメインであればフォルダを作成してそのページ上の画像を保存します。

settings.py で scrapy の設定を記述

parse_items()でitemを return すると、ImagePipeline に渡されます。
その設定は以下の通りです。自作 Pipeline の説明は後述。

settings.py

  # 自作 pipeline に繋げる
  ITEM_PIPELINES = {'tutorial.pipelines.MyImagesPipeline': 1}
  # データの保存場所
  IMAGES_STORE = '/Users/paper2/scrapy/tutorial/imgs'
  # リンクを辿る深さを指定
  DEPTH_LIMIT = 5
  # LOG_LEVEL = 'ERROR'
  DOWNLOAD_DELAY = 3

pipelines.py で pipeline をカスタマイズ

デフォルトの ImagePipeline ですとドメインごとのフォルダを作成して、、、などといった追加の作業ができないので継承して自分で作成します。

pipeliens.py

# -*- coding: utf-8 -*-
from scrapy.pipelines.images import ImagesPipeline
import os
from tutorial import settings
import shutil


class MyImagesPipeline(ImagesPipeline):
    def item_completed(self, results, item, info):
        # DL できたファイルのパス
        file_paths = [x['path'] for ok, x in results if ok]

        # ドメインごとのフォルダに move
        for file_path in file_paths:
            img_home = settings.IMAGES_STORE
            full_path = img_home + "/" + file_path
            domain_home = img_home + "/" + item['domain']

            os.makedirs(domain_home, exist_ok=True)
            # DL した結果同じファイルのことがある
            if os.path.exists(domain_home + '/' + os.path.basename(full_path)):
                continue
            shutil.move(full_path, domain_home)

        # parse() の続きに戻る
        return item

これで完成です。

実際に回してみる

しばらく回すと色々なドメインから画像が集まりました。

うん？よくみたら**いかがわしいサイトのドメインが混ざっているぞ、、、画像もいかがわしいものが、、、、**ということで次回はこのいかがわしいドメインを取り除く ~~（逆にそれだけ残す??）~~のを AWS Rekognition でやってみようと思います。