Création d'un robot de scrapping en Python

Apprenez à créer un robot de scrapping en Python. Suivez un guide étape par étape pour comprendre les concepts clés, configurer votre environnement de développement, et déployer un script fonctionnel.

Création du projet

Commencez par créer un nouveau dossier pour votre projet et naviguez à l'intérieur :

mkdir mon-robot-scrapping
cd mon-robot-scrapping

Installation des dépendances

Installez les bibliothèques nécessaires pour le scrapping web avec BeautifulSoup :

uv add playwright simplecron

Si uv n'est pas installé, vous pouvez l'installer en suivant les instructions suivantes à partir de la documentation officielle :

curl -LsSf https://astral.sh/uv/install.sh | sh

Simplecron est utilisé pour planifier des tâches récurrentes dans votre script Python. La documentation officielle est disponible ici.

Création du script de scrapping

app.py

import aiofiles
from playwright.async_api import async_playwright, Page
from simplecron.context import Context
from simplecron.base import default_scheduler, Job

URLS: set[str] = set()

async def scrap_page(job: Job, context: Context | None = None):
    pass


default_scheduler.every(20).seconds.do(scrap_page)


async def main():
    async with async_playwright() as p:
        browser = await p.chromium.launch()
        page = await browser.new_page()

        async with aiofiles.open("urls.csv", "r") as f:
            URLS = set(await f.read().splitlines())
        
        default_scheduler.start_blocking(context={'page': page})
        await browser.close()

Dans le code ci-dessus, nous ouvrons un nouveau navigateur avec Playwright, lisons les URL à partir d'un fichier CSV, et les stockons dans un ensemble pour les visiter ensuite.

La fonction start_blocking démarre le planificateur et bloque l'exécution du programme principal, permettant ainsi aux tâches planifiées de s'exécuter à intervalles réguliers.

Définition de la fonction de scrapping

La fonction scrap_page sera utilisée pour visiter les URL stockées dans l'ensemble URLS. Vous pouvez la définir comme suit :

async def scrap_page(job: Job, context: Context | None = None):
    page = context['page'] if context else None

    await page.wait_for_load_state('domcontentloaded')
    if not page:
        return

    links = await page.query_selector_all("a")
    for link in links:
        href = await link.get_attribute("href")
        if href:
            URLS.add(href)

Cette fonction attend que la page soit complètement chargée avant de commencer à extraire les liens.

await page.wait_for_load_state('domcontentloaded')

Cette section extrait tous les liens présents sur la page et les ajoute à l'ensemble URLS afin de pouvoir visiter toutes les URL du site.

links = await page.query_selector_all("a")
for link in links:
    href = await link.get_attribute("href")
    if href:
        URLS.add(href)

Amélioration

Vous pouvez améliorer le script en ajoutant la gestion des erreurs, la limitation du nombre de requêtes simultanées, et en filtrant les URL pour éviter les doublons ou les liens externes. Par exemple :

async def scrap_page(job: Job, context: Context | None = None):
    page = context['page'] if context else None

    if not page:
        return

    try:
        await page.wait_for_load_state('domcontentloaded')
        links = await page.query_selector_all("a")
        for link in links:
            href = await link.get_attribute("href")
            if href and href not in URLS:
                URLS.add(href)
    except Exception as e:
        print(f"Error occurred while scraping page: {e}")

Nous pouvons aussi nous assurer que seules les URL internes au site sont ajoutées à l'ensemble URLS en filtrant les liens externes. Par exemple :

from urllib.parse import urlparse

async def scrap_page(job: Job, context: Context | None = None):
    page = context['page'] if context else None

    if not page:
        return

    try:
        await page.wait_for_load_state('domcontentloaded')
        links = await page.query_selector_all("a")
        for link in links:
            href = await link.get_attribute("href")
            if href and href not in URLS:
                parsed_url = urlparse(href)
                if parsed_url.netloc == "" or parsed_url.netloc == "votre-domaine.com":
                    URLS.add(href)
    except Exception as e:
        print(f"Error occurred while scraping page: {e}")