快捷導(dǎo)航

Python搭建代理IP池實現(xiàn)檢測IP的方法

更新時間：2019年10月27日 14:57:29 作者：Steven·簡談

這篇文章主要介紹了Python搭建代理IP池實現(xiàn)檢測IP的方法，文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

在獲取 IP 時，已經(jīng)成功將各個網(wǎng)站的代理 IP 獲取下來了，然后就需要一個檢測模塊來對所有的代理進行一輪輪的檢測，檢測可用就設(shè)置為滿分，不可用分數(shù)就減 1，這樣就可以實時改變每個代理的可用情況，在獲取有效 IP 的時候只需要獲取分數(shù)高的 IP

代碼地址：https://github.com/Stevengz/Proxy_pool

另外三篇：
Python搭建代理IP池（一）- 獲取 IP
Python搭建代理IP池（二）- 存儲 IP
Python搭建代理IP池（四）- 接口設(shè)置與整體調(diào)度

由于代理 IP 的數(shù)量非常多，為了提高 IP 的檢測效率，這里使用異步請求庫 Aiohttp 來進行檢測。至于為什么不用抓取時用的 Requests 庫，是因為它是一個同步請求庫，在發(fā)出一個請求之后需要等待網(wǎng)頁加載完成之后才能繼續(xù)執(zhí)行程序。這個過程會阻塞在等待響應(yīng)中，如果服務(wù)器響應(yīng)非常慢，一個請求就會需要十幾秒，程序不會繼續(xù)往下執(zhí)行

異步請求庫就解決了這個問題，在請求發(fā)出之后，程序可以繼續(xù)接下去執(zhí)行其他的事情，當(dāng)響應(yīng)到達時會通知程序再去處理這個響應(yīng)，這樣程序就沒有被阻塞，可以充分把時間和資源利用起來

添加設(shè)置

增加了幾個測試用的常量

setting.py

# 數(shù)據(jù)庫地址
HOST = '127.0.0.1'
# MySql端口
MYSQL_PORT = 3306
# MySQl用戶名、密碼
MYSQL_USERNAME = '***'
MYSQL_PASSWORD = '***'
# 數(shù)據(jù)庫名
SQL_NAME = 'test'

# 代理等級
MAX_SCORE = 30
MIN_SCORE = 0
INITIAL_SCORE = 10

# 代理池數(shù)量界限
POOL_UPPER_THRESHOLD = 1000

VALID_STATUS_CODES = [200, 302]

# 測試API，建議抓哪個網(wǎng)站測哪個
TEST_URL = 'http://www.baidu.com'

# 最大批測試量
BATCH_TEST_SIZE = 30

VALID_STATUS_CODES 變量包含了正常的狀態(tài)碼，獲取 Response 后需要判斷響應(yīng)的狀態(tài)，如果狀態(tài)碼在

VALID_STATUS_CODES 這個列表里，則代表代理可用

定義方法

定義了一個類 Tester，init() 方法中建立了一個 MySqlClient 對象，供類中其他方法使用。接下來定義了一個 test_single_proxy() 方法，用來檢測單個代理的可用情況，其參數(shù)就是被檢測的代理

tester.py

import asyncio
import aiohttp
import time
import sys
from aiohttp import ClientError
from db import MySqlClient
from setting import *


class Tester(object):
 def __init__(self):
  self.mysql = MySqlClient()
 
 # 測試單個代理
 async def test_single_ip(self, ip):
  conn = aiohttp.TCPConnector(verify_ssl=False)
  async with aiohttp.ClientSession(connector=conn) as session:
   try:
    if isinstance(ip, bytes):
     ip = ip.decode('utf-8')
    real_ip = 'http://' + ip
    print('正在測試', ip)
    async with session.get(TEST_URL, proxy=real_ip, timeout=15, allow_redirects=False) as response:
     if response.status in VALID_STATUS_CODES:
      self.mysql.max(ip)
      print('代理可用', ip)
     else:
      self.mysql.decrease(ip)
      print('請求響應(yīng)碼不合法 ', response.status, 'IP', ip)
   except (ClientError, aiohttp.client_exceptions.ClientConnectorError, asyncio.TimeoutError, AttributeError):
    self.mysql.decrease(ip)
    print('代理請求失敗', ip)
 
 # 主函數(shù)
 def run(self):
  print('測試器開始運行')
  try:
   count = self.mysql.count()
   print('當(dāng)前剩余', count, '個代理')
   for i in range(0, count, BATCH_TEST_SIZE):
    start = i
    stop = min(i + BATCH_TEST_SIZE, count)
    print('正在測試第', start + 1, '-', stop, '個代理')
    test_ip_group = self.mysql.batch(start, stop)
    loop = asyncio.get_event_loop()
    tasks = [self.test_single_ip(ip_tuple[0]) for ip_tuple in test_ip_group]
    loop.run_until_complete(asyncio.wait(tasks))
    sys.stdout.flush()
    time.sleep(5)
  except Exception as e:
   print('測試器發(fā)生錯誤', e.args)


if __name__ == "__main__":
 test = Tester()
 test.run()

test_single_proxy() 方法前面加了 async 關(guān)鍵詞，代表這個方法是異步的，方法內(nèi)部首先創(chuàng)建了 Aiohttp 的 ClientSession 對象，此對象類似于 Requests 的 Session 對象，可以直接調(diào)用該對象的 get() 方法來訪問頁面

運行結(jié)果：