ごみの掃き溜めは美味ぬ

ごみの掃き溜めです

ScrapyでPOSTDの記事情報を抽出した

Python

動機

最近「そうだWebアプリを作ろう」とぼんやり思っていたのですが、そのためにはPythonのWebクローラフレームワークであるScrapyが必要だと分かりました。

今回はScrapyの勉強のために、最近発見し読み始めたPOSTDさんの記事をクローリングしてみました。

やったこと

Scrapyで記事一覧情報(タイトル、タグなど)を抽出し、JSONL形式でアウトプットした。

参考にしたWebサイト

Scrapy Tutorial 10分で理解する Scrapy

環境

$ python --version
Python 3.6.8 :: Anaconda custom (64-bit)
$ scrapy version
Scrapy 1.5.1

コーディング

実際にScrapyでクローラを作成してみます。

プロジェクトの作成

まずはプロジェクトを作ります。

$ scrapy startproject postd

Spiderの実装

Itemsの作成

まずitems.pyを編集します。 今回は記事の - URL - タイトル - 日にち - タグ

を抽出することにしたので、以下のようにItemsを定義します。

# -*- coding: utf-8 -*-
import scrapy


class ArticleItem(scrapy.Item):
    url = scrapy.Field()
    title = scrapy.Field()
    date = scrapy.Field()
    tags = scrapy.Field()

Spiderの追加

以下のようにSpiderを作成します。

scrapy genspider article postd.cc

作成されたarticle.pyを編集してparseをいじります。

# -*- coding: utf-8 -*-
import scrapy
from postd.items import ArticleItem


class ArticleSpider(scrapy.Spider):
    name = 'article'
    allowed_domains = ['postd.cc']
    start_urls = ['https://postd.cc/']

    def parse(self, response):
        for article in response.css('.block-titles-wrap'):
            yield ArticleItem(
                url=article.css('div.block-titles a::attr(href)').get(),
                title=article.css('div.block-titles a::text').get(),
                date=article.css('div.block-titles div::text').get(),
                tags=article.css('div.block-text div div a::text').getall(),
            )

        next_page = [i.split(' ')[1].split('\"')[1] for i in response.css(
            'div.pagination div a.pagination-jump').getall() if 'Next' in i]
        if next_page != []:
            yield response.follow(next_page[0], callback=self.parse)

設定の編集

最後にクロールを実行する前にsettings.pyを編集します。

リクエスト間隔の設定

以下のようにコメントを外す。

# Configure a delay for requests for the same website (default: 0)
# See https://doc.scrapy.org/en/latest/topics/settings.html#download-delay
# See also autothrottle settings and docs
DOWNLOAD_DELAY = 3

レスポンスのキャッシュの設定

以下のようにコメントを外す。

# Enable and configure HTTP caching (disabled by default)
# See https://doc.scrapy.org/en/latest/topics/downloader-middleware.html#httpcache-middleware-settings
HTTPCACHE_ENABLED = True
HTTPCACHE_EXPIRATION_SECS = 0
HTTPCACHE_DIR = 'httpcache'
HTTPCACHE_IGNORE_HTTP_CODES = []
HTTPCACHE_STORAGE = 'scrapy.extensions.httpcache.FilesystemCacheStorage'

エンコードの設定

以下を追記する。

FEED_EXPORT_ENCODING='utf-8'

Spiderの実行

JSONL形式でアウトプットします。

$ scrapy crawl article -o articles.jl

これで無事にarticles.jlに記事情報がアウトプットされました。

感想

今後やりたいこと

クロールを定期的に行う

今回クローリングすることができましたが、コマンドを入力しクローリングを開始している状態です。

「ある時間が来たらクローラを起動する」という風にしたいと思います。

アウトプットの形式

まだデータベースについて詳しくないのですが、データベースとして使いやすいようなアウトプットを調べたいです。

記事の書き方

とても久しぶりの記事で書く気力もあまりありませんでした。

気が向いたらもうちょっと追記したいと思います。

Scrapyについて

settings.pyの記法

デフォルトで生成されるsettings.pyの行の長さやコメントの表記が、Pythonのコーディング規約のPEP8に準拠していないのはPythonフレームワークとしてどうなのかなぁと思いました。

遺伝的アルゴリズムを可視化してみる

Python 
%matplotlib notebook
import numpy as np
import matplotlib.pyplot as plt
import matplotlib.animation as animation
import seaborn as sns
import math

# パラメータ
N = 100
LENGTH = 100
ELITE_NUMBER = 20
CHILDREN_NUMBER = math.ceil(2 * N / ELITE_NUMBER)
ERROR_RATE = 0.01

# 初期配列生成
def generate_first_men():
    np.random.seed(seed=0)
    return sort_men(np.random.choice(2, (N, LENGTH)))

# 優秀な順番に並べる
def sort_men(data):
    return sorted(data, key=lambda x: sum(x), reverse=True)

# 交叉する
def crossing(data):
    elites = data[:ELITE_NUMBER]
    #np.random.seed(seed=0)
    elites_pairs = np.random.choice(ELITE_NUMBER, (int(ELITE_NUMBER / 2), 2), replace=False)
    children = []
    for i, pair in enumerate(elites_pairs):
        #np.random.seed(seed=i)
        positions = np.random.choice(LENGTH, CHILDREN_NUMBER)
        children += [np.hstack((elites[pair[0]][:position], elites[pair[1]][position:])) for position in positions]
    return sort_men(children)[:N]

# エラーを起こす
def error_occuring(data):
    #np.random.seed(seed=0)
    ERROR_loc = np.random.choice(2, (N, LENGTH), p=[ERROR_RATE, 1 - ERROR_RATE])
    return sort_men(np.logical_not(np.logical_xor(data, ERROR_loc)).astype(int))

fig = plt.figure()
men = generate_first_men()
def init():
    global men
    men = generate_first_men()
    plt.title("Generation: 0")
    sns.heatmap(men, 
                cmap="Blues", 
                linewidths=0.01,
                xticklabels=False,
                yticklabels=False,
                cbar=False,
               )

def animate(i):
    plt.clf()
    global men
    men = crossing(men)
    men = error_occuring(men)
    plt.title("Generation: " + str(i + 1))
    sns.heatmap(men, 
                cmap="Blues", 
                linewidths=0.01,
                xticklabels=False,
                yticklabels=False,
                cbar=False,
               )
    
ani = animation.FuncAnimation(fig, animate, 
                              interval=500, 
                              init_func=init,
                              frames=40,
                             )

#plt.show()
ani.save("onemax_error.gif", writer="imagemagick")