Lors de l'extraction de données à partir de sites web modernes, les développeurs rencontrent fréquemment des pages rendues dynamiquement via JavaScript. Les requêtes HTTP standards de Scrapy ne peuvent pas exécuter ce code client. Pour contourner cette limitation, il est nécessaire d'intégrer un navigateur automatisé, comme Selenium, directement dans le flux de téléchargement de Scrapy via les Download Middlewares. Cette approche permet d'intercepter les requêtes, de les rendre via un navigateur headless, et de retourner le HTML final au moteur de Scrapy.

Modélisation des Données

La première étape consiste à définir la structure des données à extraire en créant un modèle d'Item robuste. Ce modèle servira de contrat pour les données transitant dans le pipeline.

import scrapy

class EnvironmentalDataItem(scrapy.Item):
    location_name = scrapy.Field()
    observation_month = scrapy.Field()
    observation_date = scrapy.Field()
    air_quality_index = scrapy.Field()
    quality_category = scrapy.Field()
    pm25_concentration = scrapy.Field()
    pm10_concentration = scrapy.Field()
    so2_concentration = scrapy.Field()
    co_concentration = scrapy.Field()
    no2_concentration = scrapy.Field()
    o3_concentration = scrapy.Field()
    origin_source = scrapy.Field()
    extraction_timestamp = scrapy.Field()

Logique du Spider

Le spider est conçu pour naviguer à travers les différentes couches de l'application : il récupère d'abord la liste des localités, puis les mois disponibles pour chaque localité, et enfin les détails quotidiens. L'utilisation des sélecteurs CSS simplifie l'extraction des attributs et du texte.

import scrapy
from myproject.items import EnvironmentalDataItem

class EnvironmentalDataSpider(scrapy.Spider):
    name = 'env_data_spider'
    allowed_domains = ['environmental-data-example.com']
    base_endpoint = 'https://www.environmental-data-example.com/historical/'
    start_urls = [base_endpoint]

    def parse(self, response):
        locations = response.css('div.region-list ul li a::text').getall()[10:12]
        location_links = response.css('div.region-list ul li a::attr(href)').getall()[10:12]

        for loc_name, loc_href in zip(locations, location_links):
            full_url = response.urljoin(loc_href)
            yield scrapy.Request(
                url=full_url,
                callback=self.parse_monthly_data,
                meta={'location': loc_name}
            )

    def parse_monthly_data(self, response):
        location = response.meta['location']
        months = response.css('table.monthly-table tr td a::text').getall()[1:2]
        month_links = response.css('table.monthly-table tr td a::attr(href)').getall()[1:2]

        for month_name, month_href in zip(months, month_links):
            next_url = response.urljoin(month_href)
            yield scrapy.Request(
                url=next_url,
                callback=self.parse_daily_details,
                meta={'location': location, 'month': month_name}
            )

    def parse_daily_details(self, response):
        location = response.meta['location']
        month = response.meta['month']

        data_rows = response.css('table.daily-details tbody tr')[1:]

        for row in data_rows:
            item = EnvironmentalDataItem()
            item['location_name'] = location
            item['observation_month'] = month
            item['observation_date'] = row.css('td:nth-child(1)::text').get()
            item['air_quality_index'] = row.css('td:nth-child(2)::text').get()
            item['quality_category'] = row.css('td:nth-child(3) span::text').get()
            item['pm25_concentration'] = row.css('td:nth-child(4)::text').get()
            item['pm10_concentration'] = row.css('td:nth-child(5)::text').get()
            item['so2_concentration'] = row.css('td:nth-child(6)::text').get()
            item['co_concentration'] = row.css('td:nth-child(7)::text').get()
            item['no2_concentration'] = row.css('td:nth-child(8)::text').get()
            item['o3_concentration'] = row.css('td:nth-child(9)::text').get()

            yield item

Implémentation des Download Middlewares

Pour gérer les pages dynamiques et éviter les blocages, deux middlewares sont implémentés. Le premier assure la rotation des User-Agents. Le second intercepte les requêtes ciblant les pages dynamiques, utilise Selenium en mode headlses pour rendre le JavaScript, et attend explicitement que l'élément DOM cible soit présent avant de retourner la réponse HTML à Scrapy.

import random
from scrapy import signals
from scrapy.http import HtmlResponse
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from myproject.settings import USER_AGENT_LIST

class RandomUserAgentMiddleware:
    def process_request(self, request, spider):
        agent = random.choice(USER_AGENT_LIST)
        request.headers.setdefault('User-Agent', agent)

class SeleniumDynamicMiddleware:
    @classmethod
    def from_crawler(cls, crawler):
        middleware = cls()
        crawler.signals.connect(middleware.spider_closed, signal=signals.spider_closed)
        return middleware

    def __init__(self):
        chrome_options = Options()
        chrome_options.add_argument("--headless")
        chrome_options.add_argument("--disable-gpu")
        chrome_options.add_argument("--no-sandbox")
        self.driver = webdriver.Chrome(options=chrome_options)

    def spider_closed(self, spider):
        self.driver.quit()

    def process_request(self, request, spider):
        if 'daily-details' in request.url:
            self.driver.get(request.url)
            
            WebDriverWait(self.driver, 15).until(
                EC.presence_of_element_located(("css selector", "table.daily-details"))
            )
            
            page_source = self.driver.page_source
            return HtmlResponse(
                url=request.url,
                body=page_source,
                encoding='utf-8',
                request=request
            )

Pipelines de Traitement des Données

Les pipelines permettent d'enrichir les items avec des métadonnées conetxtuelles avant de les persister sur le disque au format JSONL, un format idéal pour le traitement ultérieur de grands volumes de données.

import json
from datetime import datetime, timezone

class JsonStoragePipeline:
    def open_spider(self, spider):
        self.file_handle = open('environmental_data.jsonl', 'w', encoding='utf-8')

    def process_item(self, item, spider):
        line = json.dumps(dict(item), ensure_ascii=False) + '\n'
        self.file_handle.write(line)
        return item

    def close_spider(self, spider):
        self.file_handle.close()

class MetadataEnrichmentPipeline:
    def process_item(self, item, spider):
        item['origin_source'] = spider.name
        item['extraction_timestamp'] = datetime.now(timezone.utc).isoformat()
        return item

Configuration du Projet

L'activation des middlewares et des pipelines se fait dans le fichier settings.py. Les priorités (nombres entiers) définissent l'ordre d'exécution. Il est également nécessaire de définir la liste des User-Agents utilisée par le middleware de rotation.

DOWNLOADER_MIDDLEWARES = {
    'myproject.middlewares.RandomUserAgentMiddleware': 400,
    'myproject.middlewares.SeleniumDynamicMiddleware': 500,
}

ITEM_PIPELINES = {
    'myproject.pipelines.MetadataEnrichmentPipeline': 300,
    'myproject.pipelines.JsonStoragePipeline': 800,
}

USER_AGENT_LIST = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/115.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.5 Safari/605.1.15",
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/115.0"
]