Python怎么搭建代理IP池實(shí)現(xiàn)檢測(cè)IP

這篇文章主要介紹Python怎么搭建代理IP池實(shí)現(xiàn)檢測(cè)IP，文中介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們一定要看完！

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

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

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

添加設(shè)置

增加了幾個(gè)測(cè)試用的常量

setting.py

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

# 代理等級(jí)
MAX_SCORE = 30
MIN_SCORE = 0
INITIAL_SCORE = 10

# 代理池?cái)?shù)量界限
POOL_UPPER_THRESHOLD = 1000

VALID_STATUS_CODES = [200, 302]

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

# 最大批測(cè)試量
BATCH_TEST_SIZE = 30

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

VALID_STATUS_CODES 這個(gè)列表里，則代表代理可用

定義方法

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

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()
 
 # 測(cè)試單個(gè)代理
 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('正在測(cè)試', 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('請(qǐng)求響應(yīng)碼不合法 ', response.status, 'IP', ip)
   except (ClientError, aiohttp.client_exceptions.ClientConnectorError, asyncio.TimeoutError, AttributeError):
    self.mysql.decrease(ip)
    print('代理請(qǐng)求失敗', ip)
 
 # 主函數(shù)
 def run(self):
  print('測(cè)試器開(kāi)始運(yùn)行')
  try:
   count = self.mysql.count()
   print('當(dāng)前剩余', count, '個(gè)代理')
   for i in range(0, count, BATCH_TEST_SIZE):
    start = i
    stop = min(i + BATCH_TEST_SIZE, count)
    print('正在測(cè)試第', start + 1, '-', stop, '個(gè)代理')
    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('測(cè)試器發(fā)生錯(cuò)誤', e.args)


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

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

運(yùn)行結(jié)果：

以上是“Python怎么搭建代理IP池實(shí)現(xiàn)檢測(cè)IP”這篇文章的所有內(nèi)容，感謝各位的閱讀！希望分享的內(nèi)容對(duì)大家有幫助，更多相關(guān)知識(shí)，歡迎關(guān)注億速云行業(yè)資訊頻道！