Lors du développement de robots d'indexation (crawlers), l'utilisation de proxys est une pratique courante. Un proxy permet de simuler une connexion depuis une adresse IP tierce. Cette technique est indispensable lorsque la fréquence de vos requêtes risque de déclencher les mécanismes de protection (anti-flood ou anti-bot) d'un serveur, entraînant le bannissement temporaire de votre adresse IP réelle.

Analyse de la source de données

Pour construire notre réservoir, nous devons cibler un fournisseur de listes de proxys gratuits. En examinant le code source de ces sites, on constate généralement que les informations (adresse IP, port, type de protocole) sont structurées au sein de balises HTML <table>, chaque ligne <tr> représentent un serveur proxy distinct.

Extraction des données brutes

Nous utilisons les bibliothèques requests pour la récupération du code HTML et BeautifulSoup pour l'extraction sélective des données. Voici une implémentation permettant de parser les pages et de stocker les résultats dans un fichier local.

import requests
from bs4 import BeautifulSoup
import threading

def extraire_proxys(url_cible):
    # En-tête pour simuler un navigateur réel
    navigateur_headers = {
        'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.114 Safari/537.36'
    }
    
    with open('proxys_bruts.txt', 'a', encoding='utf-8') as flux_sortie:
        # Parcours des 10 premières pages de la source
        for num_page in range(1, 11):
            url_complete = f"{url_cible}{num_page}"
            try:
                reponse = requests.get(url_complete, headers=navigateur_headers, timeout=10)
                soup = BeautifulSoup(reponse.text, 'html.parser')
                
                # Ciblage du tableau contenant les adresses
                tableau_ip = soup.find('table', id='ip_list')
                if not tableau_ip:
                    continue
                    
                lignes = tableau_ip.find_all('tr')
                
                for ligne in lignes[1:]:  # Sauter l'en-tête du tableau
                    cellules = ligne.find_all('td')
                    
                    # Extraction des champs spécifiques
                    adresse_ip = cellules[1].text.strip()
                    port_ip = cellules[2].text.strip()
                    anonymat = cellules[4].text.strip()
                    protocole = cellules[5].text.strip()
                    
                    # Formatage des données pour le stockage
                    entree = f"{adresse_ip}:{port_ip}|{protocole}|{anonymat}\n"
                    flux_sortie.write(entree)
            except Exception as erreur:
                print(f"Erreur lors de la récupération sur {url_complete}: {erreur}")

Validation de la connectivité

Toutes les adresses IP récupérées ne sont pas nécesssairement fonctionnelles ou rapides. Il est crucial d'implémenter un module de vérification qui teste chaque proxy en tentant d'accéder à un service tiers (comme Google ou Baidu).

def tester_validite_ip(proxy_addr):
    """
    Vérifie si le proxy est opérationnel en envoyant une requête de test.
    """
    url_test = "http://www.google.com"
    configuration_proxy = {
        'http': f'http://{proxy_addr}',
        'https': f'http://{proxy_addr}'
    }
    
    try:
        # Tentative de connexion avec un timeout court
        req = requests.get(url_test, proxies=configuration_proxy, timeout=5)
        if req.status_code == 200:
            return True
    except:
        return False
    return False

def traiter_liste_validation(verrou, fichier_entree, fichier_valide):
    """
    Fonction exécutée par les threads pour valider les IPs en masse.
    """
    while True:
        verrou.acquire()
        ligne = fichier_entree.readline()
        verrou.release()
        
        if not ligne:
            break
            
        details = ligne.strip().split('|')
        ip_port = details[0]
        
        if tester_validite_ip(ip_port):
            print(f"[SUCCÈS] Proxy actif : {ip_port}")
            verrou.acquire()
            fichier_valide.write(ligne)
            verrou.release()
        else:
            print(f"[ÉCHEC] Proxy inactif : {ip_port}")

Optimisation via le Multi-threading

La validation unitaire étant lente à cause des délais d'attente réseau, l'utilisation de plusieurs fils d'exécution (threads) permet de traiter des centaines d'IPs simultanément. L'utilisation d'un verrou (Lock) est nécesssaire pour garantir l'intégrité des fichiers lors des accès concurrents.

if __name__ == '__main__':
    # Initialisation des fichiers
    open('proxys_bruts.txt', 'w').close()
    open('proxys_valides.txt', 'w').close()
    
    # Étape 1 : Collecte
    sources = [
        "https://www.exemple-proxy.com/nn/",
        "https://www.exemple-proxy.com/nt/"
    ]
    for s in sources:
        extraire_proxys(s)
    
    # Étape 2 : Validation multi-threadée
    verrou_partage = threading.Lock()
    with open('proxys_bruts.txt', 'r') as f_brut, \
         open('proxys_valides.txt', 'a') as f_ok:
        
        liste_threads = []
        # Lancement de 40 threads simultanés
        for _ in range(40):
            t = threading.Thread(target=traiter_liste_validation, 
                                 args=(verrou_partage, f_brut, f_ok))
            t.start()
            liste_threads.append(t)
            
        for t in liste_threads:
            t.join()
            
    print("Traitement terminé. Le réservoir d'IPs est prêt.")

Cette approche permet d'obtenir un pool de proxys robuste en un temps record. La clé de la réussite réside dans la gestion de la concurrence et la définition de timeouts appropriés pour écarter les serveurs trop lents qui ralentiraient vos futurs projets de scraping.