小白學 Python 爬蟲（31）：自己構建一個簡單的代理池

阿新 • • 發佈：2020-01-03

人生苦短，我用 Python

前文傳送門：

小白學 Python 爬蟲（1）：開篇

小白學 Python 爬蟲（2）：前置準備（一）基本類庫的安裝

小白學 Python 爬蟲（3）：前置準備（二）Linux基礎入門

小白學 Python 爬蟲（4）：前置準備（三）Docker基礎入門

小白學 Python 爬蟲（5）：前置準備（四）資料庫基礎

小白學 Python 爬蟲（6）：前置準備（五）爬蟲框架的安裝

小白學 Python 爬蟲（7）：HTTP 基礎

小白學 Python 爬蟲（8）：網頁基礎

小白學 Python 爬蟲（9）：爬蟲基礎

小白學 Python 爬蟲（10）：Session 和 Cookies

小白學 Python 爬蟲（11）：urllib 基礎使用（一）

小白學 Python 爬蟲（12）：urllib 基礎使用（二）

小白學 Python 爬蟲（13）：urllib 基礎使用（三）

小白學 Python 爬蟲（14）：urllib 基礎使用（四）

小白學 Python 爬蟲（15）：urllib 基礎使用（五）

小白學 Python 爬蟲（16）：urllib 實戰之爬取妹子圖

小白學 Python 爬蟲（17）：Requests 基礎使用

小白學 Python 爬蟲（18）：Requests 進階操作

小白學 Python 爬蟲（19）：Xpath 基操

小白學 Python 爬蟲（20）：Xpath 進階

小白學 Python 爬蟲（21）：解析庫 Beautiful Soup（上）

小白學 Python 爬蟲（22）：解析庫 Beautiful Soup（下）

小白學 Python 爬蟲（23）：解析庫 pyquery 入門

小白學 Python 爬蟲（24）：2019 豆瓣電影排行

小白學 Python 爬蟲（25）：爬取股票資訊

小白學 Python 爬蟲（26）：為啥買不起上海二手房你都買不起

小白學 Python 爬蟲（27）：自動化測試框架 Selenium 從入門到放棄（上）

小白學 Python 爬蟲（28）：自動化測試框架 Selenium 從入門到放棄（下）

小白學 Python 爬蟲（29）：Selenium 獲取某大型電商網站商品資訊

小白學 Python 爬蟲（30）：代理基礎

引言

前面的代理如果有同學動手實踐過，就會發現一個問題，現在網上的免費代理簡直太坑啦！！！

經常一螢幕好多的代理試下來，沒有幾個能用的。

當然，免費的代理嘛，連通率低、延遲高是正常的，人家畢竟是免費的。

但是這件事兒有沒有解決方案呢？

這麼天資聰穎的小編肯定是想到了辦法了呀。

先來屢屢這件事兒，其實我們要的不是連通率高，而是我們在使用的時候，能每次都用到能用的代理。

這件事兒要麼我們每次在用的時候自己手動去試，要麼~~~

我們可以寫程式讓程式自己去尋找合適的代理嘛~~~

其實這一步就是把需要我們手動做的事情變成了程式自動去完成。

代理池

先想一下這個代理池最少需要有哪些功能：

自動獲取代理
定時清除不能用的代理

這兩個是我們的核心訴求，最少要有這兩個功能，不然這個代理池也沒有存在的價值了。

根據上面兩個功能，我們來拆解程式的模組，小編這裡定義了三個模組，獲取模組（獲取代理）、儲存模組（資料庫儲存代理）、檢測模組（定時檢查代理的可用性）。

那麼它們三者的關係就是這樣的：

這裡的儲存模組我們使用 Mysql ，這與儲存模組為什麼選 Mysql ，因為 Mysql 有表結構，給各位同學展示起來比較清晰，如果需要用於生產環境的話，建議是用 Redis ，提高效率。

資料庫

首先還是貼一下資料庫的表結構，之前有同學留言問過小編表結構的事情，是小編偷懶沒有貼。

本次設計使用的還是單表模式，一張表走天下就是小編本人了。

至於欄位的含義小編就不介紹了，後面的註釋已經寫得比較清楚了。

儲存模組

對於儲存模組來講，主要的功能是要將我們獲取到的代理儲存起來，上面的 Mysql 資料庫是儲存模組的一部分。

基於 OOP 的思想，我們本次寫一個類 MysqlClient 將所有對於 Mysql 的操作封裝起來，其他模組需要和資料庫產生互動的時候只需要呼叫我們在 MysqlClient 中封裝好的方法即可。

示例程式碼如下：

MYSQL_HOST = 'localhost'
MYSQL_PORT = 3306
MYSQL_USER = 'root'
MYSQL_PASSWORD = 'password'
MYSQL_DB ='test'
MYSQL_CHARSET = 'utf8mb4'

import pymysql
import uuid

class MysqlClient(object):
    def __init__(self, host=MYSQL_HOST, port=MYSQL_PORT, user=MYSQL_USER, password=MYSQL_PASSWORD, database=MYSQL_DB, charset=MYSQL_CHARSET):
        """
        初始化 mysql 連線
        :param host: mysql 地址
        :param port: mysql 埠
        :param user: mysql 使用者
        :param password: mysql 密碼
        :param database: mysql scheme
        :param charset: 使用的字符集
        """
        self.conn = pymysql.connect(
            host = host,
            port = port,
            user = user,
            password = password,
            database = database,
            charset = charset
        )

    def add_proxy(self, proxy):
        """
        新增代理
        :param proxy: 代理字典
        :return:
        """
        sql = 'INSERT INTO `proxy_pool` VALUES (%(id)s, %(scheme)s, %(ip)s, %(port)s, %(status)s, %(response_time)s, now(), null )'
        data = {
            "id": str(uuid.uuid1()),
            "scheme": proxy['scheme'],
            "ip": proxy['ip'],
            "port": proxy['port'],
            "status": proxy['status'],
            "response_time": proxy['response_time'],
        }
        self.conn.cursor().execute(sql, data)
        self.conn.commit()

    def find_all(self):
        """
        獲取所有可用代理
        :return:
        """
        sql = 'SELECT * FROM proxy_pool WHERE status = "1" ORDER BY update_date ASC '
        cursor = self.conn.cursor()
        cursor.execute(sql)
        res = cursor.fetchall()
        cursor.close()
        self.conn.commit()
        return res

    def update_proxy(self, proxy):
        """
        更新代理資訊
        :param proxy: 需要更新的代理
        :return:
        """
        sql = 'UPDATE proxy_pool SET scheme = %(scheme)s, ip = %(ip)s, port = %(port)s, status = %(status)s, response_time = %(response_time)s, update_date = now()  WHERE id = %(id)s '
        data = {
            "id": proxy['id'],
            "scheme": proxy['scheme'],
            "ip": proxy['ip'],
            "port": proxy['port'],
            "status": proxy['status'],
            "response_time": proxy['response_time'],
        }
        self.conn.cursor().execute(sql, data)
        self.conn.commit()

在這個類中，我們首先定義了一些常量，都是和資料庫連線有關的常量，如 MYSQL_HOST 資料庫地址、 MYSQL_PORT 資料庫埠、 MYSQL_USER 資料庫使用者名稱、 MYSQL_PASSWORD 資料庫密碼、 MYSQL_DB 資料庫的 scheme 、 MYSQL_CHARSET 字符集。

接下來定義了一個 MysqlClient 類，定義了一些方法用以執行資料庫的相關操作。

init()：初始化方法，在初始化 MysqlClient 這個類時，同時初始化了 Mysql 資料庫的連結資訊，獲得了資料庫連線 connection 。
add_proxy()：向資料庫中新增代理，並新增相關資訊，包括代理響應延時和健康狀況。
find_all()：獲取所有資料庫可用代理，並根據更新時間正序排布，主要用於後續代理檢查。
update_proxy()：更新代理資訊，主要使用者檢查模組檢查完代理後更新代理資訊，根據取出當前代理的主鍵 id 進行更新。

獲取模組

獲取模組相對比較簡單，主要功能就是從各個免費代理網站上將我們所需要的代理資訊抓取下來。示例如下：

import requests
from pyquery import PyQuery
from MysqlClient import MysqlClient
from VerifyProxy import VerifyProxy

class CrawlProxy(object):

    def __init__(self):
        self.mysql = MysqlClient()
        self.verify = VerifyProxy()

    def get_page(self, url, charset):
        response = requests.get(url)
        response.encoding = charset
        return response.text

    def crawl_ip3366(self, page_num = 3):
        """
        獲取代理 ip3366
        :param page_num:
        :return:
        """
        start_url = 'http://www.ip3366.net/?stype=1&page={}'
        urls = [start_url.format(page) for page in range(1, page_num + 1)]
        for url in urls:
            print('crawl:', url)
            html = self.get_page(url, 'gb2312')
            if html:
                d = PyQuery(html)
                trs = d('.table-bordered tbody tr').items()
                for tr in trs:
                    scheme = tr.find('td:nth-child(4)').text().lower()
                    ip = tr.find('td:nth-child(1)').text()
                    port = tr.find('td:nth-child(2)').text()
                    verify_result = self.verify.verify_proxy(scheme, ip, port)

                    if verify_result["status"] == '1':
                        proxy = {
                            "scheme": scheme,
                            "ip": ip,
                            "port": port,
                            "status": verify_result["status"],
                            "response_time": verify_result["response_time"],
                        }
                        # 存入資料庫
                        self.mysql.add_proxy(proxy)
                        print('代理', ip, '連通測試已通過，已儲存 Mysql')
                    else:
                        print('代理', ip, '連通測試未通過')

if __name__ == '__main__':
    CrawlProxy().crawl_ip3366()

小編這裡出於示例只演示了從 ip3366 上抓取免費代理，並且在抓取到代理後，呼叫檢查模組的檢查方法對當前的代理進行連通性檢查，如果連通性測試未通過則不會寫入資料庫中。

檢查模組

檢查模組相對也會簡單一些，功能是從資料庫中取出所有可以用的代理，進行輪詢檢查，看看是不是有代理是連不通的，如果連不通則修改連通性標記位，將此代理標記為不可用。示例程式碼如下：

import requests
from MysqlClient import MysqlClient

class VerifyProxy(object):
    def __init__(self):
        self.mysql = MysqlClient()

    def verify_proxy(self, scheme, ip, port):
        """
        使用百度測試代理的連通性，並返回響應時長（單位：ms）
        :param scheme:
        :param ip:
        :param port:
        :return:
        """
        proxies = {
            scheme: scheme + '://' + ip + ':' + port + '/'
        }
        response_time = 0
        status = '0'
        try:
            response = requests.get(scheme + '://www.baidu.com/get', proxies=proxies)
            if response.ok:
                response_time = round(response.elapsed.total_seconds() * 1000)
                status = '1'
            else:
                response_time = 0
                status = '0'
        except:
            pass
        return {"response_time" : response_time, "status" : status}

    def verify_all(self):
        """
        驗證住方法，從資料庫中獲取所有代理進行驗證
        :return:
        """
        results = self.mysql.find_all()
        for result in results:
            res = self.verify_proxy(result[1], result[2], result[3])
            proxy = {
                "id": result[0],
                "scheme": result[1],
                "ip": result[2],
                "port": result[3],
                "status": res["status"],
                "response_time": res["response_time"],
            }
            self.mysql.update_proxy(proxy)
            print('代理驗證成功')

if __name__ == '__main__':
    VerifyProxy().verify_all()

小編這裡使用的是度娘進行連通性測試，如果各位同學有特殊的需要，可以使用特定的網站進行連通性測試。

小結

本篇的內容到這裡就結束了，不過有一點要說明，本篇的示例內容只能作為 DEMO 來進行測試使用，對於一個連線池來講，還有很多不完善的地方。

例如檢測模組應該是定時啟動，自行檢測，現在是靠人手動啟動，不過這個可以使用各種系統上的定時任務來解決。

還有，現在要獲取連線資訊只能自己開啟資料庫從中 Copy ，這裡其實還可以加一個 API 模組，寫成一個介面，供其他有需要使用代理的系統進行呼叫。

獲取模組現在小編也只是簡單的有一個網站寫一個方法，其實可以使用 Python 高階用法，獲取到類中所有的方法名，然後呼叫所需要的方法。

總之，這個 DEMO 非常不完善，等小編下次有空的時候完善下，到時候還可以再來一個推送。

示例程式碼

本系列的所有程式碼小編都會放在程式碼管理倉庫 Github 和 Gitee 上，方便大家取用。

示例程式碼-Github

示例程式碼-Gi

小白學 Python 爬蟲（31）：自己構建一個簡單的代理池

引言

代理池

資料庫

儲存模組

獲取模組

檢查模組

小結

示例程式碼

相關推薦