In this Article
Ob Web Scraping oder Website-Tests: Proxys sind unverzichtbar. Für einen zuverlässigen Zugriff benötigen Sie jedoch viele verschiedene Adressen. Gleichzeitig möchten Sie nicht manuell zwischen Proxys wechseln. Ein Proxy-Rotator ist in dieser Situation die Lösung. In diesem Artikel erstellen wir von Grund auf einen individuellen Proxy-Rotator mit Python.
Warum ist es wichtig, einen guten Proxy-Rotator zu erstellen?
Ein Rotator durchläuft automatisch einen Proxy-Pool. Er weist jeder Anfrage oder in festgelegten Zeitabständen eine andere IP-Adresse zu. Der Zielserver erkennt nicht, dass die Anfragen von derselben Quelle stammen, und der Traffic scheint von unterschiedlichen Adressen und Standorten auszugehen. So bleiben Sie für Anti-Bot-Systeme unauffällig. Deshalb steht ein Proxy-Rotator im Mittelpunkt von Scraping oder Multiaccounting. Der Rotator sorgt für einen kontinuierlichen Authentifizierungsablauf, die Einhaltung der Vorgaben für Website-Traffic und den Zugriff auf regionsspezifische Inhalte, insbesondere bei umfangreichen Projekten mit mehreren Verbindungsrisiken.
Die Nutzung eines Proxy-Rotators bietet außerdem einige weniger offensichtliche Vorteile. So verbessert er etwa Ihre Privatsphäre und Anonymität: Da sich Ihre IP fortlaufend ändert, können Websites und Tracker Ihre Aktivitäten schwerer profilieren. Auch Hacker sind Ihnen einen Schritt weniger nahe, denn wenn Ihre echte Adresse verborgen bleibt, können Betrüger, die bestimmte Geräte ins Visier nehmen, Sie schlechter identifizieren.
Darüber hinaus liefern viele Websites je nach Standort oder Präferenzen unterschiedliche Ergebnisse. Mit mehreren IPs erhalten Sie vielfältigere und präzisere Daten. Die Rotation verhindert außerdem, dass Sie personalisierte Inhalte sehen, wenn Sie diese nicht benötigen.
Zuverlässiger Proxy-Rotator: seine Komponenten
Neben der einfachen Verwendung einer neuen IP für jede Anfrage sollten Sie weitere Funktionen einbinden, damit ein Proxy-Rotator Ihnen wirklich nützt und nicht selbst zur Problemquelle wird:
- Wiederholungslogik: Falls ein Proxy nicht funktioniert, verwendet der Rotator einen anderen;
- rotierende User-Agent-Header, damit Ihre Anfragen nicht aussehen, als kämen sie von derselben Quelle;
- Behandlung von “schlechten” HTTP-Statuscodes wie 403 (“Forbidden”), 429 (“Too many requests”), 500 (“Internal server error”);
- Deaktivieren von Proxys, die wiederholt fehlschlagen;
- kleine Verzögerungen zwischen Anfragen, um Anti-Bot-Systeme nicht auszulösen;
- Protokollierung, damit Sie bei einer Wiederholung den Grund dafür sowie den verwendeten Proxy und User-Agent sehen;
- Health Check vor dem Scraping, um alle Proxys vorab zu testen und sicherzustellen, dass sie funktionieren;
- Circuit Breaker, um ausgefallene Proxys für eine gewisse Zeit zu deaktivieren;
- Sticky Sessions pro Host, damit die Domain innerhalb einer Sitzung dieselbe IP sieht, falls Sie das benötigen, etwa beim Multiaccounting, wenn Sie bei derselben Sitzung bleiben müssen.
Diese Punkte berücksichtigen wir bei der Erstellung unseres Proxy-Rotators.
Erstellung eines Proxy-Rotators mit Python
Wir erstellen einen Proxy-Rotator in einer einzigen Datei. In unserem Beispiel nennen wir sie proxy_rotator.py. Nachfolgend finden Sie alle Codeabschnitte mit einer Erklärung ihrer Funktion. Sie können sie kopieren und einfügen. Wir erstellen den Code ohne Daten wie Zugangsdaten fest einzutragen, sodass Sie später nichts ersetzen müssen.
Erste Schritte
- Installieren Sie Python von der offiziellen Website oder aktualisieren Sie es. Sie benötigen Version 3.8 oder höher.
- Führen Sie den folgenden Befehl aus, um die Bibliothek requests zu installieren. Wir benötigen sie, um HTTP-Anfragen zu senden.
pip install requests
- Rufen Sie auf der Seite des benötigten Tarifs eine Liste von Proxys ab. Sie können sie über Ihr DataImpulse-Dashboard manuell kopieren und in eine .txt-Datei einfügen oder herunterladen. Unsere Datei nennen wir proxies.txt. Sie sollte so aussehen:
http://login:password.host:port
https://login:password.host:port
Bei Bedarf können Sie das Proxy-Format ändern.
Importe und Einrichtung
Zunächst müssen wir die Werkzeuge, sogenannte “Module”, einbinden, die wir verwenden werden.
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
request sendet Webanfragen, random wählt Proxys und Header aus, und threading ist für die Verarbeitung mehrerer Verbindungs-Threads zuständig.
Standardeinstellungen
Der nächste Teil erfüllt mehrere wichtige Funktionen: Er rotiert User Agents, wechselt bei einem Fehlercode zu einem anderen Proxy und prüft, ob ein Proxy funktioniert.
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"
Sie können User Agents ersetzen oder ergänzen und bei Bedarf eine andere Website für einen Health Check verwenden. In unserem Fall verwenden wir httpbin.org/ip. Weitere geeignete Optionen sind https://httpbin.org/get (zeigt Anfrage-Header sowie die IP), https://api.ipify.org (oder https://api.ipify.org?format=json, um Ihre IP im JSON-Format zurückzugeben), https://ifconfig.me oder https://ifconfig.me/all.json (gibt Header + IP im JSON-Format zurück) oder http://ip-api.com/json/ (zeigt Land, Stadt, ISP und IP).
Proxy-Objekt
Dieser Teil definiert, wie Proxys gespeichert werden.
@dataclass(frozen=True)
class Proxy:
url: str
label: str = ""
def to_requests_proxies(self) -> Dict[str, str]:
return {"http": self.url, "https": self.url}
Auch hier müssen Sie nichts ersetzen.
Eine Proxy-Datei einlesen
Dieser Teil liest Proxys aus einer .txt-Datei ein.
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
Verzögerung bei Wiederholungen
Dieser Codeabschnitt implementiert eine Wiederholungslogik. Damit wartet ein Rotator bei jedem Fehlschlagen einer Anfrage länger, um Drosselung und Dienstsperren zu vermeiden.
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)
Die Proxy-Rotator-Klasse
Das Herzstück Ihres Rotators. Sie wählt einen Proxy aus, wiederholt Anfragen und verfolgt Fehler. Außerdem entscheidet sie über eine Strategie, also ob Proxys zufällig ausgewählt oder die Liste durchlaufen wird, legt Timeouts und Header fest und verwendet für eine Website denselben Proxy. Sie sendet Anfragen, markiert Proxys als zugelassen oder gesperrt und führt Health Checks aus.
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)
Kommandozeilenoptionen
Sie benötigen diesen Teil, um einen Rotator mit einem einzeiligen Befehl im Terminal auszuführen.
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.")
Einstiegspunkt
Damit wird Ihr Rotator gestartet. Er lädt Proxys, prüft sie bei Bedarf und sendet Anfragen.
if __name__ == "__main__":
main_cli()
Einen Proxy-Rotator ausführen
Sie können Ihren Rotator mit dem folgenden Befehl über die Kommandozeile ausführen:
python proxy_rotator.py --proxies ./proxies.txt --url http://api.ipify.org
Ersetzen Sie ./proxies.txt durch den tatsächlichen Namen Ihrer Datei mit Proxys und – – URL durch die Ziel-URL einer Website, die Sie scrapen oder testen möchten.
Außerdem können Sie weitere Flags hinzufügen:
– – strategy random – damit ein Rotator einen Proxy zufällig aus einem Pool auswählt
– – retries 5 – bei Bedarf können Sie eine andere Anzahl von Wiederholungen festlegen
– – healthcheck – um Proxys vorab zu testen
– – no-sticky – um Sticky Proxys zu deaktivieren. Verwenden Sie dies nur, wenn Sie nicht bei derselben Sitzung bleiben müssen
– – debug – Sie sehen detaillierte Logs, die Ihnen bei der Fehlerbehebung helfen.
Hinweis: Manchmal müssen Sie, wenn Python 3 installiert ist, möglicherweise “python3 proxy_rotator.py –-proxies ./proxies.txt –-url http://api.ipify.org” als Befehl zum Ausführen eines Rotators verwenden.
Fazit
Die Erstellung eines Proxy-Rotators ist weder langwierig noch schwierig, hilft Ihnen aber, einen Proxy-Pool optimal zu nutzen. Ein guter Rotator mit Wiederholungslogik, Health Check und zeitlichen Pausen kann Ihr Web Scraping oder Multiaccounting zusammen mit vertrauenswürdigen Proxys auf die nächste Stufe heben. Bei zuverlässigen, ethisch beschafften IPs hilft Ihnen DataImpulse gerne weiter. Bei uns erhalten Sie über 90 Millionen IPs aus 195 Standorten für eine unterbrechungsfreie, regionsübergreifende Konnektivität. Kontaktieren Sie uns unter [email protected] oder klicken Sie zum Starten auf die Schaltfläche “Jetzt ausprobieren”.