Scraping intelligent avec l'IA

# Scraping classique avec BeautifulSoup
import requests
from bs4 import BeautifulSoup

response = requests.get("https://example.com/produits")
soup = BeautifulSoup(response.text, "html.parser")

produits = []
for card in soup.select("div.product-card"):
    nom = card.select_one("h3.product-title").text.strip()
    prix = card.select_one("span.price").text.strip()
    produits.append({"nom": nom, "prix": prix})

# ❌ Problèmes :
# - Si le site change "div.product-card" → tout casse
# - Pages JavaScript → requests ne suffit pas
# - Données dans des formats variés → regex fragiles
# - Anti-bot → headers, cookies, captchas

# Scraping IA : le LLM comprend la page
from openai import OpenAI

client = OpenAI()

# On récupère le texte de la page (pas besoin de parser le HTML)
page_content = fetch_page_as_text("https://example.com/produits")

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": """Extrais tous les produits 
        de cette page. Pour chaque produit, donne :
        - nom, prix, description, disponibilité
        Réponds en JSON."""},
        {"role": "user", "content": page_content}
    ],
    response_format={"type": "json_object"}
)

produits = json.loads(response.choices[0].message.content)
# ✅ Fonctionne même si la structure HTML change
# ✅ Comprend le contexte ("en rupture", "promo -30%")
# ✅ Extrait des infos implicites

import httpx
from openai import OpenAI

client = OpenAI()

# Étape 1 : Récupérer la page en texte lisible
# (en utilisant un service comme Jina Reader ou un extracteur markdown)
response = httpx.get(
    "https://r.jina.ai/https://example.com/article",
    headers={"Accept": "text/markdown"}
)
page_markdown = response.text

# Étape 2 : Extraire les données avec le LLM
extraction = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": """Tu es un extracteur de données.
        Extrais les informations demandées au format JSON.
        Si une info n'est pas trouvée, mets null."""},
        {"role": "user", "content": f"""Page web :
{page_markdown}

Extrais : titre, auteur, date, résumé (2 phrases), tags"""}
    ]
)
print(extraction.choices[0].message.content)

# OpenClaw peut directement fetch + analyser
# L'agent utilise web_fetch puis le LLM pour structurer
User: "Récupère les 5 derniers articles de TechCrunch sur l'IA"
# → OpenClaw fetch la page, extrait, structure en JSON

from playwright.sync_api import sync_playwright
from openai import OpenAI

client = OpenAI()

with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    page = browser.new_page()

    # Navigation comme un humain
    page.goto("https://example.com/search")
    page.fill("#search-input", "intelligence artificielle")
    page.click("button[type=submit]")
    page.wait_for_selector(".results")

    # Scroll pour charger plus de résultats
    for _ in range(3):
        page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
        page.wait_for_timeout(2000)

    # Récupérer le contenu visible
    content = page.inner_text("body")

    # IA pour extraire les données structurées
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": """Extrais tous les résultats 
            de recherche. Format JSON : [{title, url, snippet, date}]"""},
            {"role": "user", "content": content}
        ]
    )

    browser.close()

# OpenClaw browser automation
# L'agent peut contrôler un navigateur + comprendre les pages
User: "Va sur LinkedIn, cherche les profils 'data engineer Paris'"
# → OpenClaw ouvre le browser, navigue, extrait les profils

# Méthode propre : API + LLM pour structurer
import httpx
from openai import OpenAI

client = OpenAI()

# API publique (exemple : Hacker News)
stories = httpx.get(
    "https://hacker-news.firebaseio.com/v0/topstories.json"
).json()[:10]

articles = []
for story_id in stories:
    story = httpx.get(
        f"https://hacker-news.firebaseio.com/v0/item/{story_id}.json"
    ).json()
    articles.append(story)

# LLM pour analyser et catégoriser
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": """Analyse ces articles tech.
        Pour chacun, ajoute : catégorie (IA, web, hardware, autre),
        pertinence (1-10), résumé en 1 phrase."""},
        {"role": "user", "content": json.dumps(articles, indent=2)}
    ]
)

import base64
from openai import OpenAI
from playwright.sync_api import sync_playwright

client = OpenAI()

# Capture d'écran de la page
with sync_playwright() as p:
    browser = p.chromium.launch()
    page = browser.new_page()
    page.goto("https://example.com/dashboard")
    screenshot = page.screenshot()  # bytes PNG
    browser.close()

# Envoyer au modèle vision
b64_image = base64.b64encode(screenshot).decode()
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{
        "role": "user",
        "content": [
            {"type": "text", "text": "Extrais toutes les données "
             "de ce dashboard : métriques, graphiques, tableaux. "
             "Format JSON structuré."},
            {"type": "image_url", "image_url": {
                "url": f"data:image/png;base64,{b64_image}"
            }}
        ]
    }]
)

import json
import httpx
from openai import OpenAI
from datetime import datetime

client = OpenAI()

CONCURRENTS = [
    {"nom": "Concurrent A", "url": "https://concurrent-a.com/pricing"},
    {"nom": "Concurrent B", "url": "https://concurrent-b.com/tarifs"},
    {"nom": "Concurrent C", "url": "https://concurrent-c.com/plans"},
]

Pour aller plus loin sur ce sujet, consultez notre guide [Cron + IA : automatiser des tâches intelligentes 24/7](/article/cron-ia-automatiser-taches-intelligentes).

def analyser_pricing(nom: str, url: str) -> dict:
    """Scrape et analyse la page pricing d'un concurrent."""
    # Récupérer le contenu
    resp = httpx.get(f"https://r.jina.ai/{url}")

    # Extraction IA
    analysis = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": """Analyse cette page pricing.
            Extrais au format JSON :
            {
              "plans": [{"nom": str, "prix_mensuel": float, 
                         "prix_annuel": float, "features": [str]}],
              "offre_speciale": str ou null,
              "essai_gratuit": bool,
              "changements_notables": str ou null
            }"""},
            {"role": "user", "content": resp.text[:8000]}
        ]
    )

    result = json.loads(analysis.choices[0].message.content)
    result["concurrent"] = nom
    result["date_analyse"] = datetime.now().isoformat()
    return result

Pour aller plus loin sur ce sujet, consultez notre guide [Générer du contenu automatiquement avec l'IA](/article/generer-contenu-automatiquement-ia).

# Lancer l'analyse
rapport = [analyser_pricing(c["nom"], c["url"]) for c in CONCURRENTS]

# Sauvegarder
with open(f"veille_{datetime.now():%Y-%m-%d}.json", "w") as f:
    json.dump(rapport, f, indent=2, ensure_ascii=False)

import feedparser
from openai import OpenAI

client = OpenAI()

SOURCES = [
    "https://techcrunch.com/feed/",
    "https://www.theverge.com/rss/index.xml",
    "https://feeds.arstechnica.com/arstechnica/index",
]

def agreger_actualites(sujet: str = "intelligence artificielle"):
    articles = []

    for feed_url in SOURCES:
        feed = feedparser.parse(feed_url)
        for entry in feed.entries[:10]:
            articles.append({
                "titre": entry.title,
                "lien": entry.link,
                "source": feed.feed.title,
                "date": entry.get("published", ""),
                "resume": entry.get("summary", "")[:500]
            })

    # Filtrage et analyse par IA
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": f"""Tu es un curateur d'actualités.
            Parmi les articles suivants, sélectionne ceux qui concernent 
            "{sujet}". Pour chaque article retenu :
            - Résumé en 2 phrases en français
            - Score de pertinence (1-10)
            - Tags (3 max)

            Trie par pertinence décroissante. Format JSON."""},
            {"role": "user", "content": json.dumps(articles, ensure_ascii=False)}
        ]
    )

    return json.loads(response.choices[0].message.content)

# Utilisation
news = agreger_actualites("intelligence artificielle")
for article in news[:5]:
    print(f"📰 {article['titre']}")
    print(f"   {article['resume']}")
    print(f"   Pertinence: {article['score']}/10")
    print()

import json
import sqlite3
from datetime import datetime
import httpx
from openai import OpenAI

client = OpenAI()

def surveiller_prix(url: str, produit: str) -> dict:
    """Récupère et analyse le prix d'un produit."""
    resp = httpx.get(f"https://r.jina.ai/{url}")

    extraction = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": """Extrais les informations 
            de prix de cette page produit. Format JSON :
            {
              "prix_actuel": float,
              "prix_barre": float ou null,
              "devise": "EUR",
              "en_stock": bool,
              "promotion": str ou null,
              "vendeur": str
            }"""},
            {"role": "user", "content": f"Produit recherché: {produit}"
             f"\n\nPage:\n{resp.text[:6000]}"}
        ]
    )

    data = json.loads(extraction.choices[0].message.content)
    data["url"] = url
    data["produit"] = produit
    data["timestamp"] = datetime.now().isoformat()
    return data

def verifier_alerte(prix_actuel: float, prix_cible: float) -> bool:
    """Vérifie si le prix est passé sous le seuil."""
    return prix_actuel <= prix_cible

# Exemple d'utilisation
produits_suivis = [
    {"url": "https://amazon.fr/dp/XXXXX", "nom": "GPU RTX 5080", "prix_cible": 900},
    {"url": "https://fnac.com/a123456", "nom": "MacBook Air M4", "prix_cible": 1100},
]

for produit in produits_suivis:
    data = surveiller_prix(produit["url"], produit["nom"])
    if verifier_alerte(data["prix_actuel"], produit["prix_cible"]):
        print(f"🚨 ALERTE: {produit['nom']} à {data['prix_actuel']}€ "
              f"(cible: {produit['prix_cible']}€)")

# Exemple : veille automatique quotidienne avec OpenClaw
# L'agent peut être configuré pour :
# 1. Chaque matin à 8h, scraper vos sources
# 2. Extraire et catégoriser les informations
# 3. Vous envoyer un résumé par Telegram/Discord

# Configuration cron dans OpenClaw
# Voir : configurer les tâches planifiées

# Bonnes pratiques de scraping éthique

import time
import httpx

class EthicalScraper:
    def __init__(self, delay: float = 2.0):
        self.delay = delay
        self.session = httpx.Client(
            headers={
                # 1. S'identifier clairement
                "User-Agent": "MonBot/1.0 ([email protected]; "
                              "recherche académique)",
            },
            timeout=30,
            follow_redirects=True,
        )

    def respecte_robots_txt(self, url: str) -> bool:
        """Vérifie robots.txt avant de scraper."""
        from urllib.robotparser import RobotFileParser
        from urllib.parse import urlparse

        parsed = urlparse(url)
        robots_url = f"{parsed.scheme}://{parsed.netloc}/robots.txt"

        rp = RobotFileParser()
        rp.set_url(robots_url)
        rp.read()
        return rp.can_fetch("MonBot", url)

    def fetch(self, url: str) -> str | None:
        # 2. Respecter robots.txt
        if not self.respecte_robots_txt(url):
            print(f"⛔ Bloqué par robots.txt : {url}")
            return None

        # 3. Rate limiting (ne pas surcharger)
        time.sleep(self.delay)

        response = self.session.get(url)

        # 4. Respecter les codes HTTP
        if response.status_code == 429:  # Too Many Requests
            print("⏳ Rate limited, attente...")
            time.sleep(60)
            return self.fetch(url)

        if response.status_code != 200:
            return None

        return response.text

# ✅ Checklist scraping éthique

☐ Vérifié robots.txt du site
☐ Lu les CGU / ToS du site
☐ Pas de données personnelles (ou RGPD respecté)
☐ Rate limiting implémenté (min 1-2 sec entre requêtes)
☐ User-Agent identifiant avec contact
☐ API officielle vérifiée (préférer si disponible)
☐ Usage légitime (recherche, veille, pas de revente)
☐ Pas de contournement de protection (captcha, login)
☐ Stockage sécurisé des données collectées
☐ Politique de rétention définie (pas de stockage infini)

┌──────────────────────────────────────────────────┐
│                  ORCHESTRATEUR                    │
│            (Cron / OpenClaw / Airflow)            │
└──────────────┬───────────────────┬───────────────┘
               │                   │
       ┌───────▼───────┐   ┌──────▼────────┐
       │   COLLECTE    │   │   COLLECTE    │
       │  (web_fetch)  │   │  (browser)    │
       │  Sites simples│   │  Sites JS/SPA │
       └───────┬───────┘   └──────┬────────┘
               │                   │
               └─────────┬─────────┘
                         │ HTML/Markdown brut
                ┌────────▼────────┐
                │   EXTRACTION    │
                │  (LLM / GPT-4o │
                │   mini)         │
                └────────┬────────┘
                         │ JSON structuré
                ┌────────▼────────┐
                │   STOCKAGE      │
                │  (PostgreSQL /  │
                │   SQLite)       │
                └────────┬────────┘
                         │
              ┌──────────┼──────────┐
              │          │          │
       ┌──────▼──┐ ┌────▼────┐ ┌──▼───────┐
       │ ALERTES │ │ ANALYSE │ │ DASHBOARD│
       │ (email, │ │ (trends,│ │ (Grafana,│
       │ Telegram)│ │  LLM)   │ │  custom) │
       └─────────┘ └─────────┘ └──────────┘