In this Article
Baik untuk web scraping maupun pengujian situs web, kebutuhan akan proxy tidak dapat disangkal. Namun, Anda memerlukan banyak alamat berbeda untuk memastikan akses yang konsisten. Pada saat yang sama, Anda tentu tidak akan berpindah antarproxy secara manual. Proxy rotator adalah jawabannya dalam situasi ini. Dalam artikel ini, kami membangun proxy rotator kustom menggunakan Python dari awal.
Mengapa penting membangun proxy rotator yang baik
Rotator secara otomatis bergilir di dalam pool proxy. Rotator menetapkan alamat IP lain untuk setiap request atau pada interval waktu tertentu. Server target tidak melihat bahwa request berasal dari sumber yang sama, dan lalu lintas tampak berasal dari alamat serta lokasi yang berbeda, sehingga Anda dapat lolos dari perhatian sistem anti-bot. Itulah sebabnya membangun proxy rotator menjadi inti dari scraping atau multiaccounting. Rotator bertanggung jawab menjaga alur autentikasi yang stabil, menyesuaikan diri dengan panduan lalu lintas situs web, dan mengakses konten khusus wilayah, terutama dalam proyek berskala besar dengan beragam ancaman koneksi.
Selain itu, ada beberapa keuntungan menggunakan proxy rotator yang tidak begitu jelas. Contohnya, rotator meningkatkan privasi dan anonimitas Anda karena IP yang terus berubah menyulitkan situs web dan pelacak untuk membuat profil aktivitas Anda. Peretas pun semakin jauh dari Anda karena alamat asli Anda tersembunyi, sehingga penipu yang menargetkan perangkat tertentu lebih sulit mengenali Anda.
Selain itu, banyak situs web menyajikan hasil yang berbeda berdasarkan lokasi atau preferensi Anda. Dengan menggunakan beberapa IP, Anda memperoleh data yang lebih beragam dan akurat. Rotation juga mencegah munculnya konten yang dipersonalisasi jika Anda tidak membutuhkannya.
Proxy Rotator Andal: Komponennya
Selain sekadar menggunakan IP baru untuk setiap request, Anda sebaiknya menyertakan beberapa fitur lagi agar proxy rotator benar-benar bermanfaat dan tidak menjadi sumber masalah:
- Logika retry – jika satu proxy tidak berfungsi, rotator menggunakan proxy lain;
- Header User-Agent rotating agar request Anda tidak tampak berasal dari sumber yang sama;
- Menangani kode status HTTP yang buruk seperti 403 (“Dilarang”), 429 (“Terlalu banyak request”), 500 (“Kesalahan server internal”);
- Menonaktifkan proxy yang terus gagal;
- Menambahkan jeda singkat di antara request agar tidak memicu sistem anti-bot;
- Pencatatan log agar, jika terjadi retry, Anda dapat melihat penyebabnya serta proxy dan User-Agent yang digunakan;
- Health check sebelum scraping untuk menguji semua proxy sebelum memulai dan memastikan semuanya berfungsi;
- Circuit breaker untuk menonaktifkan proxy mati selama beberapa waktu;
- Sticky session per host agar dalam satu session, domain melihat IP yang sama jika Anda memerlukannya, misalnya untuk multiaccounting ketika Anda perlu tetap menggunakan session yang sama.
Kami akan mengingat poin-poin tersebut saat membuat proxy rotator kami.
Membangun proxy rotator menggunakan Python
Kami akan membuat proxy rotator dalam satu file. Dalam contoh ini, kami menamainya proxy_rotator.py. Di bawah ini, Anda akan menemukan semua bagian kode beserta penjelasan fungsinya. Anda dapat menyalin dan menempelkannya. Kami membuat kode tanpa menetapkan data seperti kredensial secara langsung dalam kode, jadi Anda tidak perlu mengganti apa pun nantinya.
Memulai
- Instal Python dari situs web resminya (atau lakukan upgrade). Anda memerlukan versi 3.8 atau lebih tinggi.
- Jalankan perintah berikut untuk menginstal pustaka requests. Pustaka ini diperlukan untuk membuat request HTTP.
pip install requests
- Dapatkan daftar proxy dari halaman paket yang diperlukan. Anda dapat menyalin dan menempelkannya secara manual ke dalam file .txt dari dasbor DataImpulse atau mengunduhnya. Kami menamai file kami proxies.txt. Tampilannya seharusnya seperti ini:
http://login:password.host:port
https://login:password.host:port
Anda dapat mengubah format proxy jika diperlukan.
Import dan penyiapan
Pertama, kita perlu memasukkan alat yang disebut “modul” dan akan kita gunakan.
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 mengirim request web, random memilih proxy dan header, sedangkan threading bertanggung jawab menangani beberapa thread koneksi.
Pengaturan default
Bagian berikut menjalankan beberapa fungsi penting: melakukan rotating User-Agent, beralih ke proxy lain jika terjadi kode error, dan memeriksa apakah proxy berfungsi.
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"
Anda dapat mengganti atau menambahkan User-Agent dan menggunakan situs web lain untuk health check jika mau. Dalam contoh ini, kami menggunakan httpbin.org/ip. Opsi lain yang juga layak adalah https://httpbin.org/get (menampilkan header request dan IP), https://api.ipify.org (atau https://api.ipify.org?format=json untuk menampilkan IP Anda dalam JSON), https://ifconfig.me atau https://ifconfig.me/all.json (menampilkan header + IP dalam JSON), atau http://ip-api.com/json/ (menampilkan negara, kota, ISP, dan IP).
Objek proxy
Bagian ini menentukan cara proxy disimpan.
@dataclass(frozen=True)
class Proxy:
url: str
label: str = ""
def to_requests_proxies(self) -> Dict[str, str]:
return {"http": self.url, "https": self.url}
Sekali lagi, Anda tidak perlu mengganti apa pun di sini.
Membaca file proxy
Bagian ini membaca proxy dari file .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
Jeda retry
Bagian kode ini membentuk logika retry. Dengannya, rotator menunggu lebih lama setiap kali request gagal untuk mencegah throttling dan penguncian layanan.
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)
Kelas proxy rotator
Inilah inti rotator Anda. Kelas ini memilih proxy, mencoba request kembali, dan melacak kegagalan. Kelas ini juga menentukan strategi, apakah menggunakan proxy secara acak atau bergilir melalui daftar, menetapkan timeout dan header, serta memakai proxy yang sama untuk suatu situs web. Kelas ini mengirim request, menandai proxy sebagai masuk whitelist atau diblokir, dan menjalankan health check.
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)
Opsi command line
Anda memerlukan bagian ini untuk menjalankan rotator dengan perintah satu baris melalui terminal.
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.")
Titik masuk
Bagian inilah yang memulai rotator Anda. Bagian ini memuat proxy, memeriksanya jika Anda memerlukannya, dan membuat request.
if __name__ == "__main__":
main_cli()
Menjalankan proxy rotator
Anda dapat menjalankan rotator dari command line dengan perintah berikut:
python proxy_rotator.py --proxies ./proxies.txt --url http://api.ipify.org
Ubah ./proxies.txt menjadi nama sebenarnya dari file proxy Anda dan – – URL menjadi URL target situs web yang ingin Anda scrape atau uji.
Anda juga dapat menambahkan beberapa flag berikut:
– – strategy random agar rotator memilih proxy dari pool secara acak
– – retries 5 untuk menetapkan jumlah retry lain jika diperlukan
– – healthcheck untuk menguji proxy terlebih dahulu
– – no-sticky untuk menonaktifkan sticky proxy. Gunakan hanya jika Anda tidak perlu mempertahankan session yang sama
– – debug untuk menampilkan log terperinci yang akan membantu Anda mengatasi masalah.
Catatan: terkadang, jika Python 3 telah terinstal, Anda mungkin perlu menggunakan “python3 proxy_rotator.py –-proxies ./proxies.txt –-url http://api.ipify.org” sebagai perintah untuk menjalankan rotator.
Kesimpulan
Membangun proxy rotator tidak membutuhkan waktu lama dan tidak sulit. Namun, rotator membantu Anda memaksimalkan proxy pool. Rotator yang baik, dengan logika retry, health check, dan jeda waktu, serta dipadukan dengan proxy tepercaya, dapat meningkatkan web scraping atau multiaccounting Anda ke tingkat berikutnya. Untuk IP andal yang diperoleh secara etis, DataImpulse siap membantu Anda. Bersama kami, Anda bisa mendapatkan lebih dari 90 juta IP dari 195 lokasi untuk memastikan konektivitas lintas wilayah yang tanpa gangguan. Hubungi kami di [email protected] atau tekan tombol “Coba sekarang” untuk memulai.