AWS Rekognition を使う時にクローラーも使ってなんかできないかなと思い scrapy を利用してみました。とりあえず今回はドメインと画像収集のところまで。いかがわしいことには絶対利用しないでください
今回はスタートのページからどんどんリンクを辿り、ドメイン名のフォルダごとに、辿った時のページの画像を保存します。今度そのフォルダごとに画像を AWS Rekognition に投げて、そのドメインがどんなドメインなのかを画像から判別しようと考えています。
前提
- scrapy 1.5.0
- python3
- scrapy インストール済み
参考サイト
Spider のコード
クローラーの肝となる部分です。参考サイトではCrawlSpiderクラスを継承して利用している場合が多かったです。そっちの方が大抵の場合は楽だと思います。
# -*- 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 の説明は後述。
# 自作 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 ですとドメインごとのフォルダを作成して、、、などといった追加の作業ができないので継承して自分で作成します。
# -*- 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 でやってみようと思います。