আপনি যদি Python-এ বড় পরিসরে কিছু scrape, crawl বা অটোমেট করেন, আগে হোক বা পরে টার্গেট সাইট লক্ষ করবে যে প্রতিটি রিকোয়েস্ট একটি IP থেকে আসছে এবং আপনাকে বন্ধ করে দেবে। Python-এ রোটেটিং প্রক্সি আপনার ট্র্যাফিককে অনেক IP অ্যাড্রেস জুড়ে ছড়িয়ে দেয় যাতে কোনো একটি IP rate limit-এ আটকা না পড়ে বা ban না খায়। এই গাইডে requests, httpx এবং aiohttp-এর জন্য সঠিক কোড দেখানো হয়েছে, সঙ্গে একটি rotating gateway ও একটি managed IP pool-এর মধ্যে কীভাবে বেছে নেবেন তাও।
দুটি বিস্তৃত সিদ্ধান্ত নিচের সবকিছুকে রূপ দেয়। প্রথম: rotating বনাম sticky session — প্রতি রিকোয়েস্টে একটি নতুন IP, নাকি একটি মাল্টি-স্টেপ flow-এর জন্য একটি IP ধরে রাখা। দ্বিতীয়: residential বনাম datacenter IP — দেখুন প্রক্সি-টাইপ সিদ্ধান্ত ম্যাট্রিক্স। এই দুটি ঠিক করুন আর কোড হলো সহজ অংশ।
১. গেটওয়ে রোটেশন (সুপারিশকৃত)। আপনি একটি একক এন্ডপয়েন্টে কানেক্ট করেন এবং প্রোভাইডার প্রতিটি নতুন কানেকশনে আপনাকে একটি নতুন exit IP দেয়। কোনো তালিকা সামলানোর নেই, কোনো মৃত IP ছাঁটার নেই। JIBAO-র ডাইনামিক residential pool এভাবেই কাজ করে: একই host ও port, প্রতি রিকোয়েস্টে একটি নতুন IP।
২. স্ব-পরিচালিত IP তালিকা। আপনি host:port এন্ট্রির একটি তালিকা রাখেন এবং প্রতি রিকোয়েস্টে একটি বেছে নেন। বেশি নিয়ন্ত্রণ, কিন্তু health-checking, retry এবং ban-tracking আপনার দায়িত্ব। যখন আপনার পরিচিত, স্থিতিশীল অ্যাড্রেস দরকার তখন ডেডিকেটেড datacenter IP-র সঙ্গে এটি উপযোগী।
সবচেয়ে সহজ ক্ষেত্র — একটি গেটওয়ের মাধ্যমে একটি রিকোয়েস্ট রুট করুন যা আপনার জন্য রোটেট করে। socks5h:// ব্যবহার করুন (শেষের h মানে DNS প্রক্সির দিকে রিজলভ হয়, যা lookup ফাঁস হওয়া এড়ায়):
import requests
# JIBAO rotating residential gateway: a new exit IP per connection
PROXY = "socks5h://USERNAME:[email protected]:10001"
proxies = {"http": PROXY, "https": PROXY}
for _ in range(5):
r = requests.get("https://httpbin.org/ip", proxies=proxies, timeout=30)
print(r.json()["origin"]) # a different IP each loop
requests-কে SOCKS5 বলতে হবে? এক্সট্রাটি ইনস্টল করুন: pip install "requests[socks]"। আপনি যদি একটি HTTP প্রক্সি এন্ডপয়েন্ট পছন্দ করেন, scheme-টি http://-তে বদলান — বাকিটা অভিন্ন।
import random
import requests
PROXY_POOL = [
"socks5h://USERNAME-country-us:[email protected]:10001",
"socks5h://USERNAME-country-uk:[email protected]:10001",
"socks5h://USERNAME-country-de:[email protected]:10001",
]
def fetch(url):
proxy = random.choice(PROXY_POOL)
return requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=30)
resp = fetch("https://example.com")
print(resp.status_code)
Login flow, cart এবং পেজিনেটেড রেজাল্ট ভেঙে পড়ে যদি টাস্কের মাঝপথে IP বদলায়। username-এ একটি session টোকেন যোগ করে একটি IP পিন করুন; একই টোকেন পুনঃব্যবহার করে তার জীবনকালজুড়ে একই exit IP ধরে রাখুন:
import requests
SESSION_ID = "task-0001"
PROXY = f"socks5h://USERNAME-session-{SESSION_ID}:[email protected]:10001"
s = requests.Session()
s.proxies = {"http": PROXY, "https": PROXY}
s.post("https://example.com/login", data={"u": "me", "p": "secret"})
s.get("https://example.com/dashboard") # same IP, session intact
httpx আপনাকে HTTP/2 এবং নেটিভ async-সহ একটি আধুনিক ক্লায়েন্ট দেয়। লক্ষ করুন যে নতুন ভার্সনগুলো একটি একক proxy= আর্গুমেন্ট নেয়:
import httpx
PROXY = "socks5h://USERNAME:[email protected]:10001"
with httpx.Client(proxy=PROXY, timeout=30) as client:
for _ in range(5):
print(client.get("https://httpbin.org/ip").json()["origin"])
httpx-এ SOCKS সাপোর্টের জন্য pip install "httpx[socks]" দরকার। HTTP/2-র জন্য, ক্লায়েন্টে http2=True যোগ করুন — এটি আপনার ট্র্যাফিককে আরও একটি প্রকৃত ব্রাউজারের মতো দেখায়, যা TLS ফিঙ্গারপ্রিন্ট ম্যাচিং-এর সঙ্গে ভালো জোড় বাঁধে।
উচ্চ থ্রুপুটের জন্য, একসঙ্গে অনেক রিকোয়েস্ট ছুড়ুন। গেটওয়ের মাধ্যমে প্রতিটি কানেকশন নিজস্ব exit IP পায়, তাই concurrency ও রোটেশন একসঙ্গে আসে:
import asyncio
import aiohttp
from aiohttp_socks import ProxyConnector
PROXY = "socks5://USERNAME:[email protected]:10001"
async def fetch(url):
connector = ProxyConnector.from_url(PROXY)
async with aiohttp.ClientSession(connector=connector) as session:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as r:
return await r.text()
async def main():
urls = ["https://httpbin.org/ip"] * 20
results = await asyncio.gather(*(fetch(u) for u in urls))
print(f"fetched {len(results)} pages")
asyncio.run(main())
SOCKS connector ইনস্টল করুন pip install aiohttp_socks দিয়ে। প্রতি টাস্কে একটি নতুন connector তৈরি করুন যাতে প্রতিটি রিকোয়েস্ট একটি নতুন proxied কানেকশন খোলে।
রোটেশন তখনই ফল দেয় যখন আপনি ব্যর্থতাগুলো একটি নতুন IP-তে retry করেন। সূচকীয়ভাবে back off করুন এবং প্রতিটি চেষ্টায় session টোকেন রোটেট করুন:
import requests
from time import sleep
def fetch_with_retry(url, max_retries=4):
for attempt in range(max_retries):
proxy = f"socks5h://USERNAME-session-r{attempt}:[email protected]:10001"
try:
r = requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=30)
if r.status_code in (403, 429):
raise requests.HTTPError(f"blocked: {r.status_code}")
r.raise_for_status()
return r.text
except requests.RequestException:
sleep(2 ** attempt) # 1s, 2s, 4s, 8s
raise RuntimeError(f"failed after {max_retries} tries: {url}")
একটি বড় রান-এর আগে, IP আসলেই বদলায় কিনা নিশ্চিত করুন। একটি echo এন্ডপয়েন্টে কয়েকবার আঘাত করুন এবং দেখুন IP গুলো আলাদা কিনা। আপনার TLS ফিঙ্গারপ্রিন্ট আপনাকে ফাঁস করে দিচ্ছে না তাও নিশ্চিত করতে, রিকোয়েস্টটি JIBAO-র ফ্রি JA3/TLS ফিঙ্গারপ্রিন্ট চেকার-এর বিপরীতে চালান — একটি স্পষ্ট ধরা-পড়া Python ফিঙ্গারপ্রিন্টসহ একটি rotating IP-ও ব্লক হয়।
রোটেশন একবার দৃঢ় হলে, এরপর যে দেয়ালে আঘাত করবেন তা সাধারণত ফিঙ্গারপ্রিন্টিং, IP নয়। পরিচ্ছন্ন residential IP-ও যদি 403 পায়, পড়ুন curl_cffi দিয়ে কীভাবে TLS ফিঙ্গারপ্রিন্টিং বাইপাস করবেন এবং সম্পূর্ণ Cloudflare বাইপাস রেসিপি।
500MB ফ্রি ট্রাফিক এবং একটি rotating residential gateway পান যা প্রতিটি রিকোয়েস্টে আপনাকে একটি নতুন IP দেয়।
ফ্রি ট্রায়াল শুরু করুননতুন ব্যবহারকারীরা নিবন্ধনের সময় 500MB পান, সাথে প্রথম রিচার্জে বোনাস। সীমিত সময়ের অফার।