環境構築と基本設定

ScrapyはPython向けのクローラーフレームワークです。本記事ではBOSS直聘サイトの求人情報を抽出する実装例を示します。

プロジェクト初期化手順

1. プロジェクト作成: scrapy startproject BOSS
2. スパイダー生成: cd BOSS && scrapy genspider job_spider zhipin.com
3. アイテム定義（items.py）:

   import scrapy
   
   class JobItem(scrapy.Item):
       position_id = scrapy.Field()
       title = scrapy.Field()
       company = scrapy.Field()
       salary = scrapy.Field()
       location = scrapy.Field()
       experience = scrapy.Field()
   

データストレージ設定

MongoDBを使用したデータ保存のため、以下の設定を追加:

# settings.py
ITEM_PIPELINES = {'BOSS.pipelines.MongoDBPipeline': 300}
MONGO_URI = "mongodb://localhost:27017"
MONGO_DATABASE = "job_data"

MongoDB接続ヘルパー

# db_handler.py
from pymongo import MongoClient

class MongoDBConnector:
    def __init__(self, host, port, db_name):
        self.client = MongoClient(host, port)
        self.db = self.client[db_name]
    
    def insert_document(self, collection_name, data):
        collection = self.db[collection_name]
        return collection.insert_one(data)

クローラー実装

スパイダー基本構造

# job_spider.py
import scrapy
from BOSS.items import JobItem

class JobSpider(scrapy.Spider):
    name = 'job_spider'
    allowed_domains = ['zhipin.com']
    start_urls = ['https://www.zhipin.com/c101010100/?query=Python']
    
    def parse(self, response):
        listings = response.css('div.job-list ul li')
        for job in listings:
            item = JobItem()
            # データ抽出処理
            yield item

ページネーション処理

def parse(self, response):
    current_data = self.extract_jobs(response)
    
    if self.previous_data == current_data:
        self.logger.info("データ重複のためクローリング終了")
        return
    
    self.previous_data = current_data
    next_page = self.construct_next_page()
    yield scrapy.Request(next_page, callback=self.parse)

データパイプライン

# pipelines.py
from BOSS.db_handler import MongoDBConnector

class MongoDBPipeline:
    def __init__(self, mongo_uri, mongo_db):
        self.connector = MongoDBConnector(mongo_uri, 27017, mongo_db)
        
    @classmethod
    def from_crawler(cls, crawler):
        return cls(
            mongo_uri=crawler.settings.get('MONGO_URI'),
            mongo_db=crawler.settings.get('MONGO_DATABASE')
        )
    
    def process_item(self, item, spider):
        self.connector.insert_document('python_jobs', dict(item))
        return item

実行とデータ検証

クローラー実行コマンド: scrapy crawl job_spider

MongoDBでのデータ確認: db.python_jobs.find().pretty()