In this Article
Que ce soit pour le web scraping ou les tests de sites Web, les proxies sont indispensables. Pour garantir un accès fiable, vous avez toutefois besoin de nombreuses adresses différentes, sans avoir à passer manuellement de l’une à l’autre. C’est précisément le rôle d’un rotateur de proxies. Dans cet article, nous allons créer de A à Z un rotateur de proxies personnalisé en Python.
Pourquoi créer un bon rotateur de proxies ?
Un rotateur de proxies parcourt automatiquement un pool de proxies. Il attribue une adresse IP différente à chaque requête ou selon un intervalle défini. Le serveur cible ne voit donc pas que les requêtes proviennent de la même source : le trafic semble venir de plusieurs adresses et emplacements, ce qui aide à ne pas déclencher les systèmes anti-bot. C’est pourquoi un rotateur de proxies est un outil central pour le scraping et la gestion de plusieurs comptes. Il permet de conserver un flux d’authentification cohérent, de respecter les règles de trafic d’un site Web et d’accéder à du contenu propre à une région, notamment dans les projets à grande échelle où les risques de blocage sont nombreux.
Un rotateur de proxies offre aussi des avantages moins évidents. Il renforce notamment votre confidentialité et votre anonymat : comme votre IP change régulièrement, il est plus difficile pour les sites Web et les trackers d’établir le profil de vos activités. En masquant votre véritable adresse, vous réduisez également le risque que des fraudeurs ciblant certains appareils puissent vous identifier.
De plus, de nombreux sites Web affichent des résultats différents selon votre emplacement ou vos préférences. En utilisant plusieurs IP, vous obtenez des données plus variées et plus précises. La rotation évite aussi l’affichage de contenu personnalisé lorsque vous n’en avez pas besoin.
Les composants d’un rotateur de proxies fiable
Au-delà de l’utilisation d’une nouvelle IP pour chaque requête, un rotateur de proxies doit intégrer quelques fonctionnalités supplémentaires pour être réellement utile et ne pas devenir une source de problèmes :
- Logique de nouvelle tentative : si un proxy ne fonctionne pas, le rotateur en utilise un autre ;
- Rotation des en-têtes User-Agent afin que vos requêtes ne semblent pas provenir de la même source ;
- Gestion des codes d’état HTTP problématiques, tels que 403 (“Forbidden”), 429 (“Too Many Requests”) et 500 (“Internal Server Error”) ;
- Désactivation des proxies qui échouent de manière répétée ;
- Ajout de courts délais entre les requêtes afin d’éviter de déclencher les systèmes anti-bot ;
- Journalisation pour savoir, en cas de nouvelle tentative, pourquoi elle a eu lieu et quel proxy ainsi que quel User-Agent ont été utilisés ;
- Health check avant le scraping afin de tester tous les proxies avant de commencer et de vérifier qu’ils fonctionnent ;
- Circuit breaker pour désactiver temporairement les proxies indisponibles ;
- Sessions persistantes par hôte pour que, dans une même session, le domaine voie la même IP lorsque c’est nécessaire, par exemple pour la gestion de plusieurs comptes.
Nous garderons ces éléments à l’esprit lors de la création de notre rotateur de proxies.
Créer un rotateur de proxies avec Python
Nous allons créer notre rotateur de proxies dans un seul fichier, nommé proxy_rotator.py. Vous trouverez ci-dessous tous les extraits de code, accompagnés de leur explication. Vous pouvez les copier-coller tels quels. Le code ne contient aucune donnée sensible codée en dur, comme des identifiants, vous n’aurez donc rien à remplacer par la suite.
Premiers pas
- Installez Python depuis son site officiel, ou mettez-le à jour. La version 3.8 ou une version ultérieure est requise.
- Exécutez la commande suivante pour installer la bibliothèque requests. Nous en aurons besoin pour effectuer des requêtes HTTP.
pip install requests
- Récupérez une liste de proxies depuis la page correspondant à votre plan. Vous pouvez les copier-coller manuellement dans un fichier .txt depuis votre tableau de bord DataImpulse, ou les télécharger. Nous avons nommé notre fichier proxies.txt. Il doit se présenter ainsi :
http://login:password.host:port
https://login:password.host:port
Vous pouvez modifier le format du proxy si nécessaire.
Importations et configuration
Nous devons d’abord importer les modules dont nous allons nous servir.
import argparse
import logging
import os
import random
import threading
import time
from dataclasses import dataclass
from typing import Dict, Iterable, List, Optional, Tuple
import requests
from requests import Response
requests envoie les requêtes Web, random sélectionne les proxies et les en-têtes, et threading gère plusieurs threads de connexion.
Paramètres par défaut
La section suivante assure plusieurs fonctions essentielles : elle fait tourner les User-Agent, bascule vers un autre proxy en cas de code d’erreur et vérifie qu’un proxy fonctionne.
DEFAULT_USER_AGENTS: List[str] = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:125.0) Gecko/20100101 Firefox/125.0",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36 Edg/124.0.0.0",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 13_4) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.5 Safari/605.1.15",
"Mozilla/5.0 (iPhone; CPU iPhone OS 16_5 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.5 Mobile/15E148 Safari/604.1",
"Mozilla/5.0 (Linux; Android 13; Pixel 7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Mobile Safari/537.36",
]
DEFAULT_ROTATE_ON_STATUS = {403, 407, 409, 418, 421, 425, 429, 500, 502, 503, 504}
DEFAULT_HEALTHCHECK_URL = "http://api.ipify.org"
Vous pouvez remplacer ou ajouter des agents utilisateurs et utiliser un autre site Web pour un bilan de santé si vous le souhaitez. Dans notre cas, nous utilisons httpbin.org/ip. D’autres options décentes sont https://httpbin.org/get (affiche les en-têtes de requête ainsi que IP), https://api.ipify.org (ou https://api.ipify.org?format=json pour retourner votre IP en JSON), https://ifconfig.me ou https://ifconfig.me/all.json (renvoie les en-têtes + IP dans JSON), ou http://ip-api.com/json/ (affiche le pays, la ville, le FAI et IP).
Objet proxy
Cette partie définit la manière dont les proxys sont stockés.
@dataclass(frozen=True)
class Proxy:
url: str
label: str = ""
def to_requests_proxies(self) -> Dict[str, str]:
return {"http": self.url, "https": self.url}
Là encore, vous n’avez rien à modifier.
Lire un fichier de proxies
Cette section lit les proxies depuis un fichier .TXT.
def parse_proxy_file(path: str) -> List[Proxy]:
proxies: List[Proxy] = []
with open(path, "r", encoding="utf-8") as f:
for line in f:
raw = line.strip()
if not raw or raw.startswith("#"):
continue
raw = raw.split(" #", 1)[0].split("\t#", 1)[0].strip()
proxies.append(Proxy(url=raw))
if not proxies:
raise ValueError(f"No proxies found in file: {path}")
return proxies
Délai entre les nouvelles tentatives
Cet extrait met en place une logique de nouvelle tentative. À chaque échec d’une requête, le rotateur attend plus longtemps afin d’éviter la limitation du débit et les blocages par le service.
def _exp_backoff_sleep(base: float, attempt: int, max_sleep: float) -> None:
sleep_s = min(base * (2 ** attempt) + random.random() * base, max_sleep)
time.sleep(sleep_s)
Classe du rotateur de proxies
C’est le cœur de votre rotateur. Il choisit un proxy, relance les requêtes et suit les échecs. Il définit aussi la stratégie à employer, c’est-à-dire choisir les proxies au hasard ou parcourir la liste, configure les délais d’attente et les en-têtes, et conserve le même proxy pour un site Web. Il envoie les requêtes, marque les proxies comme autorisés ou bloqués et exécute les health checks.
class ProxyRotator:
def __init__(
self,
proxies: Iterable[Proxy],
strategy: str = "round_robin",
max_retries: int = 3,
timeout: Tuple[float, float] = (10.0, 30.0),
rotate_on_status: Optional[Iterable[int]] = None,
user_agents: Optional[List[str]] = None,
sticky_per_host: bool = True,
circuit_threshold: int = 2,
circuit_cooldown: float = 60.0,
backoff_base: float = 0.5,
backoff_max: float = 5.0,
healthcheck_url: Optional[str] = None,
logger: Optional[logging.Logger] = None,
) -> None:
self.proxies: List[Proxy] = list(proxies)
if not self.proxies:
raise ValueError("ProxyRotator requires at least one proxy")
if strategy not in {"round_robin", "random"}:
raise ValueError("strategy must be 'round_robin' or 'random'")
self.strategy = strategy
self.max_retries = max_retries
self.timeout = timeout
self.rotate_on_status = set(rotate_on_status or DEFAULT_ROTATE_ON_STATUS)
self.user_agents = user_agents or DEFAULT_USER_AGENTS
self.sticky_per_host = sticky_per_host
self.circuit_threshold = circuit_threshold
self.circuit_cooldown = circuit_cooldown
self.failure_counts: Dict[Proxy, int] = {p: 0 for p in self.proxies}
self.disabled_until: Dict[Proxy, float] = {p: 0.0 for p in self.proxies}
self.backoff_base = backoff_base
self.backoff_max = backoff_max
self.healthcheck_url = healthcheck_url
self._local = threading.local()
self._sticky_map: Dict[str, Proxy] = {}
self._idx = 0
self._lock = threading.Lock()
self.log = logger or logging.getLogger("proxy_rotator")
def get(self, url: str, **kwargs) -> Response:
return self.request("GET", url, **kwargs)
def post(self, url: str, **kwargs) -> Response:
return self.request("POST", url, **kwargs)
def request(self, method: str, url: str, **kwargs) -> Response:
session = getattr(self._local, "session", None)
if session is None:
session = requests.Session()
self._local.session = session
headers = kwargs.pop("headers", {}) or {}
headers.setdefault("User-Agent", random.choice(self.user_agents))
timeout = kwargs.pop("timeout", self.timeout)
last_exc: Optional[Exception] = None
last_response: Optional[Response] = None
for attempt in range(self.max_retries + 1):
proxy = self._choose_proxy(url)
if proxy is None:
raise RuntimeError("No healthy proxies currently available")
try:
self.log.debug("Using proxy %s for %s %s", proxy.url, method, url)
response = session.request(
method,
url,
headers=headers,
proxies=proxy.to_requests_proxies(),
timeout=timeout,
**kwargs,
)
last_response = response
if response.status_code in self.rotate_on_status:
self._mark_failure(proxy, f"HTTP {response.status_code}")
self._maybe_unstick_host(url, proxy)
if attempt < self.max_retries:
_exp_backoff_sleep(self.backoff_base, attempt, self.backoff_max)
continue
else:
self._mark_success(proxy)
return response
except requests.RequestException as exc:
last_exc = exc
self._mark_failure(proxy, repr(exc))
self._maybe_unstick_host(url, proxy)
if attempt < self.max_retries:
_exp_backoff_sleep(self.backoff_base, attempt, self.backoff_max)
continue
break
if last_exc:
raise last_exc
assert last_response is not None
return last_response
def health_check(self, url: Optional[str] = None, sample: Optional[int] = None) -> Dict[str, bool]:
check_url = url or self.healthcheck_url or DEFAULT_HEALTHCHECK_URL
to_test = list(self.proxies)
if sample is not None:
to_test = random.sample(to_test, k=min(sample, len(to_test)))
results: Dict[str, bool] = {}
for p in to_test:
try:
r = requests.get(check_url, proxies=p.to_requests_proxies(), timeout=(5, 10))
healthy = r.ok
except Exception:
healthy = False
results[p.url] = healthy
if healthy:
self._mark_success(p)
else:
self._mark_failure(p, "healthcheck")
return results
def _choose_proxy(self, url: str) -> Optional[Proxy]:
host = requests.utils.urlparse(url).hostname or ""
now = time.time()
with self._lock:
if self.sticky_per_host:
sticky = self._sticky_map.get(host)
if sticky and self.disabled_until.get(sticky, 0.0) <= now:
return sticky
enabled = [p for p in self.proxies if self.disabled_until.get(p, 0.0) <= now]
if not enabled:
return None
if self.strategy == "round_robin":
proxy = enabled[self._idx % len(enabled)]
self._idx = (self._idx + 1) % len(enabled)
else:
proxy = random.choice(enabled)
if self.sticky_per_host:
self._sticky_map[host] = proxy
return proxy
def _mark_failure(self, proxy: Proxy, reason: str) -> None:
with self._lock:
cnt = self.failure_counts.get(proxy, 0) + 1
self.failure_counts[proxy] = cnt
if cnt >= self.circuit_threshold:
self.disabled_until[proxy] = time.time() + self.circuit_cooldown
self.log.warning("Disabling proxy %s for %.1fs (reason: %s)", proxy.url, self.circuit_cooldown, reason)
def _mark_success(self, proxy: Proxy) -> None:
with self._lock:
self.failure_counts[proxy] = 0
self.disabled_until[proxy] = 0.0
def _maybe_unstick_host(self, url: str, proxy: Proxy) -> None:
if not self.sticky_per_host:
return
host = requests.utils.urlparse(url).hostname or ""
with self._lock:
if self._sticky_map.get(host) == proxy:
self._sticky_map.pop(host, None)
Options de ligne de commande
Cette section est nécessaire pour lancer le rotateur depuis un terminal à l’aide d’une seule commande.
def build_arg_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(description="HTTP proxy rotator")
p.add_argument("--url", required=False, help="Target URL to request (for quick testing)")
p.add_argument("--method", default="GET", help="HTTP method to use (default: GET)")
p.add_argument(
"--proxies",
help="Path to proxies.txt file (one proxy URL per line). If omitted, reads PROXY_LIST env (comma-separated)",
)
p.add_argument("--strategy", choices=["round_robin", "random"], default="round_robin")
p.add_argument("--retries", type=int, default=3, help="Max retry attempts across proxies")
p.add_argument(
"--timeout",
type=float,
default=15.0,
help="Read timeout in seconds (connect timeout is fixed at 10s in this simple CLI)",
)
p.add_argument(
"--rotate-on",
default=",".join(str(s) for s in sorted(DEFAULT_ROTATE_ON_STATUS)),
help="Comma-separated status codes that trigger rotation",
)
p.add_argument(
"--no-sticky",
action="store_true",
help="Disable sticky-per-host behavior",
)
p.add_argument(
"--healthcheck",
action="store_true",
help="Run a healthcheck across proxies and print a summary before the request",
)
p.add_argument(
"--debug",
action="store_true",
help="Enable verbose logging",
)
return p
def _load_proxies_from_args(args: argparse.Namespace) -> List[Proxy]:
if args.proxies:
return parse_proxy_file(args.proxies)
env = os.getenv("PROXY_LIST", "").strip()
if env:
return [Proxy(url=p.strip()) for p in env.split(",") if p.strip()]
raise SystemExit("No proxies provided. Use --proxies or set PROXY_LIST env.")
def main_cli() -> None:
parser = build_arg_parser()
args = parser.parse_args()
logging.basicConfig(
level=logging.DEBUG if args.debug else logging.INFO,
format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
log = logging.getLogger("proxy_rotator")
proxies = _load_proxies_from_args(args)
rotate_on_set = {int(x.strip()) for x in args.rotate_on.split(",") if x.strip()}
rotator = ProxyRotator(
proxies=proxies,
strategy=args.strategy,
max_retries=args.retries,
timeout=(10.0, args.timeout),
rotate_on_status=rotate_on_set,
user_agents=DEFAULT_USER_AGENTS,
sticky_per_host=not args.no_sticky,
circuit_threshold=2,
circuit_cooldown=60.0,
backoff_base=0.5,
backoff_max=5.0,
healthcheck_url=DEFAULT_HEALTHCHECK_URL,
logger=log,
)
if args.healthcheck:
results = rotator.health_check()
total = len(results)
healthy = sum(1 for ok in results.values() if ok)
log.info("Healthcheck: %s/%s proxies OK", healthy, total)
for url, ok in results.items():
log.info(" %-40s %s", url, "OK" if ok else "BAD")
if args.url:
try:
resp = rotator.request(args.method.upper(), args.url)
print(f"Status: {resp.status_code}")
text = resp.text
preview = text[:500].replace("\n", " ")
print(f"Body preview (first 500 chars):\n{preview}")
except Exception as e:
log.error("Request failed: %s", e)
raise
else:
log.info("No --url provided; nothing else to do. Add --url to test a request.")
Point d’entrée
C’est ce qui démarre votre rotateur. Il charge les proxies, les vérifie si nécessaire, puis effectue les requêtes.
if __name__ == "__main__":
main_cli()
Exécuter un rotateur de proxies
Vous pouvez exécuter votre rotateur depuis la ligne de commande à l’aide de la commande suivante :
python proxy_rotator.py --proxies ./proxies.txt --url http://api.ipify.org
Remplacez ./proxies.txt par le nom réel de votre fichier de proxies et –url par l’URL cible du site Web que vous souhaitez scraper ou tester.
Vous pouvez également ajouter les indicateurs suivants :
–strategy random – pour que le rotateur sélectionne un proxy aléatoirement dans le pool
–retries 5 – pour définir un autre nombre de tentatives si nécessaire
–healthcheck – pour tester les proxies au préalable
–no-sticky – pour désactiver les proxies persistants. À utiliser seulement si vous n’avez pas besoin de conserver la même session
–debug – pour afficher des journaux détaillés qui vous aideront à résoudre les problèmes.
Remarque: dans certains cas, si Python 3 est installé, vous devrez peut-être utiliser “python3 proxy_rotator.py –-proxies ./proxies.txt –-url http://api.ipify.org” pour lancer le rotateur.
Conclusion
Créer un rotateur de proxies n’est ni long ni difficile, mais cela vous aide à tirer le meilleur parti d’un pool de proxies. Associé à des proxies fiables, un bon rotateur doté d’une logique de nouvelle tentative, d’un health check et de délais peut faire passer votre web scraping ou votre gestion de plusieurs comptes au niveau supérieur. Pour des IP fiables et obtenues de manière éthique, DataImpulse peut vous aider. Nous proposons plus de 90 millions d’IP dans 195 emplacements afin de garantir une connectivité interrégionale ininterrompue. Contactez-nous à l’adresse [email protected] ou cliquez sur le bouton “Essayez maintenant” pour commencer.