অ্যাসিঙ্ক স্ক্র্যাপিং হলো সেই জায়গা যেখানে প্রক্সি সেটআপ নীরবে ভেঙে পড়ে। requests দিয়ে আপনি প্রতি কলে একটি IP ঘোরান আর এগিয়ে যান; asyncio দিয়ে হঠাৎ আপনার হাতে একসাথে ২০০টি রিকোয়েস্ট চলে আসে, যাদের প্রত্যেকটির নিজের এক্সিট, নিজের সেশন অ্যাফিনিটি এবং টার্গেট ৪০৩ ছুড়লে নিজের রিট্রাই দরকার। ভুল করলে হয় আপনি একটি IP-কে পিটিয়ে ব্যান করিয়ে ফেলবেন, নয়তো প্রবাহের মাঝপথে ঘুরিয়ে প্রতিটি লগইন সেশন ছিন্নভিন্ন করে ফেলবেন।
এই গাইডটি দেখায় সেইসব প্যাটার্ন যেগুলো আসলে কনকারেন্সিতে টিকে থাকে: httpx ও aiohttp-তে প্রতি-রিকোয়েস্ট রোটেশন, একটি সীমাবদ্ধ ওয়ার্কার পুল, স্টেটফুল প্রবাহের জন্য স্টিকি সেশন এবং ব্লক-সচেতন রিট্রাই — সঙ্গে কার্যকর কোড যা আপনি পেস্ট করে দিতে পারেন।
একটি সিঙ্ক্রোনাস স্ক্র্যাপার একবারে একটি প্রক্সি স্পর্শ করে, তাই "প্রতিটি রিকোয়েস্টে ঘোরাও" সহজেই সঠিক। অ্যাসিঙ্ক একসাথে তিনটি অনুমান ভেঙে দেয়:
httpx.AsyncClient প্রতি ক্লায়েন্টে একটি প্রক্সি বাঁধে, তাই ঘোরাতে হলে আপনি প্রতি রিকোয়েস্টে প্রক্সি বেছে নেন এবং এক্সিট অনুযায়ী কী করা ক্লায়েন্টের একটি ছোট পুলের মধ্য দিয়ে রাউট করেন:
import asyncio, itertools, httpx
PROXIES = [
"socks5h://USERNAME:[email protected]:913",
"socks5h://USERNAME:[email protected]:913",
"socks5h://USERNAME:[email protected]:913",
]
pool = itertools.cycle(PROXIES)
# One reusable client per exit (connection pooling stays intact)
clients = {p: httpx.AsyncClient(proxy=p, timeout=30) for p in PROXIES}
async def fetch(url):
proxy = next(pool)
r = await clients[proxy].get(url)
return r.status_code, r.text
async def main(urls):
results = await asyncio.gather(*(fetch(u) for u in urls))
for client in clients.values():
await client.aclose()
return resultsপ্রতি এক্সিটে একটি ক্লায়েন্ট পুনর্ব্যবহার করলে প্রতি রিকোয়েস্টে নতুন হ্যান্ডশেক দেওয়ার বদলে HTTP/2 ও কানেকশন পুলিং সচল থাকে। লক্ষ্য করুন httpx>=0.28 ক্লায়েন্টে proxies= নাম বদলে proxy= করেছে।
aiohttp এখানে আরও সহজ — একটি একক ClientSession প্রতি রিকোয়েস্টে একটি proxy= আর্গুমেন্ট নেয়, তাই আপনি ইনলাইনে ঘোরাতে পারেন:
import asyncio, itertools, aiohttp
pool = itertools.cycle(PROXIES) # http:// or socks5h:// (needs aiohttp-socks for SOCKS)
async def fetch(session, url):
proxy = next(pool)
async with session.get(url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=30)) as r:
return r.status, await r.text()
async def main(urls):
async with aiohttp.ClientSession() as session:
return await asyncio.gather(*(fetch(session, u) for u in urls))aiohttp নেটিভভাবে HTTP প্রক্সি বোঝে; socks5h://-এর জন্য aiohttp-socks যোগ করুন এবং একটি ProxyConnector পাস করুন।
সীমাহীন gather হলো আপনার পুলের প্রতিটি IP একসাথে ফ্ল্যাগ করানোর দ্রুততম উপায়। এটিকে সীমাবদ্ধ করুন যাতে প্রতিটি এক্সিট মানুষসুলভভাবে বিশ্বাসযোগ্য সংখ্যক সমান্তরাল রিকোয়েস্ট বহন করে:
sem = asyncio.Semaphore(10) # at most 10 in flight
async def fetch_capped(session, url):
async with sem:
return await fetch(session, url)সাধারণ নিয়ম: কনকারেন্সিকে আপনার থাকা স্বতন্ত্র এক্সিটের সংখ্যার সমান বা তার নিচে রাখুন, যাতে আপনি একটি একক আবাসিক IP-তে অনেকগুলো সমান্তরাল রিকোয়েস্ট স্তূপীকৃত না করেন।
রোটেশন হলো ভেতরে ঢোকার জন্য; স্টিকি সেশন হলো ভেতরে থাকার জন্য। একটি লগইন, একটি কার্ট, কুকি-বাঁধা যেকোনো কিছু পুরো ক্রমের জন্য একটি এক্সিট ধরে রাখতে হবে। প্রতি রিকোয়েস্টের বদলে প্রতি টাস্কে প্রক্সিটি পিন করুন:
async def run_account(account, proxy):
# Same exit IP for every step of this account's flow
async with httpx.AsyncClient(proxy=proxy, timeout=30) as client:
await client.post("https://target.site/login", data=account.creds)
await client.get("https://target.site/dashboard")
await client.post("https://target.site/cart", json=account.order)
async def main(accounts):
# One sticky exit per account, accounts run concurrently
await asyncio.gather(*(run_account(a, p)
for a, p in zip(accounts, itertools.cycle(PROXIES))))এমন একটি প্রোভাইডার ব্যবহার করুন যা স্টিকি আবাসিক সেশন সমর্থন করে, যাতে টাস্কের পুরো আয়ুষ্কালের জন্য একই এক্সিট ধরে রাখা হয়, আপনার অজান্তে নীরবে ঘোরানো না হয়।
কনকারেন্সিতে আপনি একটি 200-কে বিশ্বাস করতে পারেন না — অ্যান্টি-বট সিস্টেম একটি চ্যালেঞ্জ বডিসহ 200 ফেরত দেয়। ব্লক শনাক্ত করুন এবং কেবল ব্যর্থ হওয়া একক টাস্কটিকে একটি নতুন এক্সিটে রিট্রাই করুন:
BLOCK_MARKERS = ("just a moment", "/cdn-cgi/challenge-platform",
"datadome", "px-captcha", "access denied")
def looks_blocked(status, text):
return status in (403, 429, 503) or any(m in text[:4000].lower() for m in BLOCK_MARKERS)
async def fetch_retry(url, attempts=4):
for _ in range(attempts):
proxy = next(pool)
async with httpx.AsyncClient(proxy=proxy, timeout=30) as c:
r = await c.get(url)
if not looks_blocked(r.status_code, r.text):
return r
await asyncio.sleep(0.5)
raise RuntimeError(f"blocked after {attempts} exits: {url}")একটি সমস্যা অ্যাসিঙ্ক সমাধান করে না: httpx ও aiohttp দুটিই পাইথনের TLS স্ট্যাকের উপর বসে, তাই তারা এমন একটি JA3 পাঠায় যা কোনো আসল ব্রাউজার তৈরি করে না। একটি পরিচ্ছন্ন আবাসিক IP-তে একটি নিখুঁত রোটেশন পুলও হ্যান্ডশেকের সময় অটোমেশন হিসেবে ফিঙ্গারপ্রিন্ট হয়ে যায়। অ্যাসিঙ্ক রোটেশনকে TLS ইমপারসোনেশনের সঙ্গে জোড়া লাগান — দেখুন curl_cffi দিয়ে TLS ফিঙ্গারপ্রিন্টিং বাইপাস করুন, এবং www.jibaoproxy.com/tools/fingerprint.html-তে আপনার এক্সিটের JA3 + ASN যাচাই করুন।
নতুন ব্যবহারকারীরা নিবন্ধনের সময় 500MB পান, সাথে প্রথম রিচার্জে বোনাস। সীমিত সময়ের অফার।