Python proxy pool management artifacts! Asynchronous crawling & IP melting mechanism details

Hands-On Tutorial: Building Python Agent Pools by Hand

Crawler old iron know that the proxy IP is our "golden bell". But the proxy pool management tools on the market are either too complex or expensive. Today we use pure Python to get aAsynchronous Grabbing + Intelligent Fusingof the agent pool, the key code is ready for you!

Why does your agent pool always flip?

I've seen too many people treating proxy pools as "trash cans" and using them without thinking when they catch IPs. As a result, either the IP is blocked, or the speed is slow as a turtle. The real proxy pool should be likesmart butlerThe IP quality can be checked automatically, bad IPs can be fused in seconds, and good IPs can be prioritized. Here we recommend the use of ipipgo's dynamic residential proxy, his IP survival time can be customized, with our fusion mechanism is a perfect match.

 Asynchronous IP Crawl Template
import aiohttp
from datetime import datetime

async def fetch_ip(api_url):: async with aiohttp.
    async with aiohttp.ClientSession() as session.
        async with session.get(api_url, proxy="") as session.
            proxy="http://user:pass@ipipgo-proxy.com:port") as resp.
            return await resp.json()

 Example API call for ipipgo (remember to replace your account)
IP_API = "https://api.ipipgo.com/dynamic?country=US&duration=15min"

The meltdown is the soul.

This design is likeCircuit FusesIf you find that an IP response times out or returns an exception status code, immediately fuse the IP. here is a simple version of the implementation:

class IPCircuitBreaker.
    class IPCircuitBreaker. def __init__(self).
        self.broken_ips = {}

    async def check_ip(self, ip).
         Check if an IP is available
        try: async with aiohttp.
            async with aiohttp.Timeout(5): async with session.get('', proxy=ip)
                async with session.get('http://test.com', proxy=ip) as r:: return r.status == 200
                    return r.status == 200
        async with session.get('') as r: return r.status == 200
            self.broken_ips[ip] = datetime.now() Record the time of failure
            return False

How do I choose my IPIPGO package?

Guide to avoiding the pit (QA)

Q: How many times do I have to use the proxy IP before it expires?
A：Check if the IP survival time is not set, it is recommended to set it in the background of ipipgo.Customizing the statute of limitations, in conjunction with the fuse detection cycle in the code.

Q: How do asynchronous requests control concurrency?
A: Use asyncio's Semaphore for flow control, don't let the server think you're in a DDOS attack!

Q: What protocols does ipipgo support?
A: HTTP/HTTPS/SOCKS5 all handle, climbing tubes, ins these need socks5 scene remember to choose the corresponding protocol.

Upgrade Play: IP Health Check

A private tip for the guys: use theMulti-dimensional scoring mechanismRate the IP. 60 points for responsiveness, 30 points for success rate, and 10 points for geolocation. Eliminate IPs with scores below 80 every week so that the quality of the proxy pool rubs off!

def ip_score(ip).
    speed_score = min(60, 60 - (ip.response_time 10))
    success_score = 30 ip.success_rate
    location_score = 10 if ip.country == target else 0
    return speed_score + success_score + location_score