File size: 2,520 Bytes
25ae7fe | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 | """
Free Proxy Fetcher - Automatically fetches and validates free proxies
"""
import aiohttp
import asyncio
import logging
logger = logging.getLogger("proxy_fetcher")
class FreeProxyFetcher:
def __init__(self):
self.proxies = []
self.last_fetch = 0
async def fetch_free_proxies(self):
"""Fetch free proxies from public APIs"""
proxy_sources = [
"https://api.proxyscrape.com/v2/?request=get&protocol=http&timeout=10000&country=all&ssl=all&anonymity=all",
"https://www.proxy-list.download/api/v1/get?type=http",
]
all_proxies = []
async with aiohttp.ClientSession() as session:
for source in proxy_sources:
try:
async with session.get(source, timeout=10) as resp:
if resp.status == 200:
text = await resp.text()
proxies = [f"http://{line.strip()}" for line in text.split('\n') if line.strip()]
all_proxies.extend(proxies[:20]) # Take first 20 from each source
logger.info(f"Fetched {len(proxies)} proxies from {source}")
except Exception as e:
logger.error(f"Failed to fetch from {source}: {e}")
self.proxies = all_proxies
logger.info(f"Total free proxies loaded: {len(self.proxies)}")
return self.proxies
async def validate_proxy(self, proxy, test_url="https://httpbin.org/ip"):
"""Test if a proxy works"""
try:
async with aiohttp.ClientSession() as session:
async with session.get(test_url, proxy=proxy, timeout=5) as resp:
if resp.status == 200:
return True
except:
pass
return False
async def get_working_proxies(self, max_count=10):
"""Get validated working proxies"""
if not self.proxies:
await self.fetch_free_proxies()
working = []
tasks = [self.validate_proxy(p) for p in self.proxies[:30]]
results = await asyncio.gather(*tasks, return_exceptions=True)
for proxy, is_working in zip(self.proxies[:30], results):
if is_working and len(working) < max_count:
working.append(proxy)
logger.info(f"Validated {len(working)} working proxies")
return working
proxy_fetcher = FreeProxyFetcher()
|