IP_Pool

前言

剛好前段時間突然萌發了自己做一個代理池的想法，於是就用了一些通俗的方法來實現，一來能方便自己理解，二來也加強學習。
這裡開放給大家，給大家提供一點參考，使用前請務必要仔細檢視README.md檔案。
Github:Proxy_IP_Pool

總體構思

• 定期從公開的代理網站上採集ip，在進行初次驗證後進行格式化並儲存到指定檔案；
• 定期檢測已存ip的有效性；
• 提供api介面檢視以存ip及獲取有效代理ip。

應用技術

• python3: requests, flask, gevent，multiprocessing, logging

目錄結構

--[ip_pool]:          
    --[pool]
        --[core]:
            --ip_crawl.py           # 爬蟲程式碼
            --tool_lib.py           # 工具函式集合    
        --items.py                  # 資料結構定義檔案
        --settings.py               # 基礎設定檔案
        --api.py                    # RESTful API介面檔案
    --[data]:
        --ip_data.json              # 有效ip儲存檔案（執行後生成資料夾）
    --[log]:
        --err.log                   # 錯誤日誌（執行後生成資料夾）
    --run.py                        # 爬蟲及驗證程式啟動檔案
    --log.yml                       # logging配置檔案
    --requirements.txt              # python requirments
    --README.md    
 

程式碼片段

僅展示部分程式碼片段，限於篇幅沒有展示自建模組tool_lib，如需瀏覽完整程式碼請移步Github/Proxy_IP_Pool，執行前請務必細看README.md檔案

ip_crawl.py

#!/usr/bin/env python
# -*- coding:utf-8 -*-
"""
Purpose: 從公開網站抓取代理ip，建立ip池，每10分鐘檢測可用性進行更新一次，提供api介面，每次獲取一個ip地址，並在取用之前進行檢測
Author: YajunZheng
Created: 2018/10/25
"""

import datetime
import requests
from
 
 lxml import etree

from pool.items import IpDataFormat
from pool import settings
from pool.core import tool_lib


class XicidailiCrawl(object):
    """西刺代理"""
    def __init__(self):
        self.init_url = "http://www.xicidaili.com/nn/{0}"
        self.s = requests.session()
        self.s.keep_alive = False
        self.proxies = tool_lib.get_valid_ip()

    def start_requests(self):
        """獲取第一頁中包含的ip列表"""
        for i in range(1, 2):
            url = self.init_url.format(str(i))
            headers = settings.HEADERS
            res = self.s.get(url=url, headers=headers, proxies=self.proxies, timeout=60)
            selector_1 = etree.HTML(res.text)
            ip_table = selector_1.xpath('//*[@id="ip_list"]//tr')
            print("Xicidaili: {0} ALL:{1} TIME:{2}".format(url, len(ip_table), datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")))
            yield ip_table

    def parse_ip_info(self, each_tr):
        """解析ip列表中的每一條ip資訊，驗證新增至儲存檔案"""
        ip_info = IpDataFormat()
        try:
            ip = each_tr.xpath('./td[2]/text()')[0]
            port = each_tr.xpath('./td[3]/text()')[0]
            ip_type = each_tr.xpath('./td[6]/text()')[0].casefold()
            ip_info['ip'] = str(ip)
            ip_info['port'] = str(port)
            ip_info['type'] = str(ip_type)
            # print(ip_info.__dict__.values())

            ip_proxies_str = "{'%s':'%s:%s'}" % (ip_info['type'], ip_info['ip'], ip_info['port'])
            ip_proxies = eval(ip_proxies_str)                                 # 字串轉列表
            if tool_lib.verify_ip(ip_proxies=ip_proxies, ip=ip):             # 驗證ip的有效性
                valid_ip_info = dict(ip=ip, proxies=ip_proxies)
                tool_lib.add_ip(new_ip_info=valid_ip_info)                 # 新增ip
        except:
            pass

run.py

#!usr/bin/env python
# -*- coding:utf-8 -*-
"""
Purpose: 啟動檔案，通過執行該檔案啟動爬蟲及週期校驗程式
Author: YajunZheng
Created: 2018/10/25
"""

import logging
import time
from multiprocessing import Pool

import gevent
from gevent import monkey; monkey.patch_all()
from gevent.pool import Pool

from pool import settings
from pool.core import ip_crawl, tool_lib

log_pause = logging.getLogger('Crawl_Pause')

def xicidaili():
    while True:
        try:
            th_pool = Pool(settings.VERIFY_MAX_CONCURRENT)           # 初始化協程池
            xici = ip_crawl.XicidailiCrawl()                        # 例項化西刺代理爬蟲物件
            for each_table in xici.start_requests():
                threads = []
                for each_tr in each_table:
                    threads.append(th_pool.spawn(xici.parse_ip_info, each_tr))           # 構造協程列表
                gevent.joinall(threads)                                                 # 啟動協程
                time.sleep(settings.PAGE_SLEEP_TIME)
            log_pause.info('XICI FINISHED')
            time.sleep(settings.CRAWL_SLEEP_TIME)
        except:
            log_pause.warning('ERROR 503')                 # 利用代理ip訪問西刺可能遇到封ip的情況，此時列印503錯誤，切換ip重連
            time.sleep(settings.CRAWL_SLEEP_TIME/100)
            
if __name__ == '__main__':
    """程式啟動後不必關閉，放置後臺執行，爬蟲及驗證週期時長設定請參照settings.py"""
    p = Pool(3)                                                  # 初始化程序池
    p.apply_async(xicidaili)                                    # 將xicidaili新增到程序池
    p.apply_async(tool_lib.periodic_verify)                   # 將迴圈驗證函式新增到程序池
    p.join()                                                 # 啟動

api.py

#!flask/bin/python
"""
Purpose: 提供有效的ip獲取RESTful API介面
API-Example(127.0.0.1:5000):
    獲取單個有效ip(有驗證): 127.0.0.1:5000/ip_pool/api/v1.0/get_ip
    檢視檔案中所有ip(無驗證): 127.0.0.1:5000/ip_pool/api/v1.0/get_ip_pool
    檢視檔案中指定ip(無驗證): 127.0.0.1:5000/ip_pool/api/v1.0/get_a_ip/0
Author: YajunZheng
Created: 2018/10/29
"""

import json
from os.path import dirname, abspath

from flask import Flask, jsonify
from flask import abort

from pool import settings
from pool.core.tool_lib import get_valid_ip


app = Flask(__name__)
abs_path = dirname(abspath(__file__))
data_path = abs_path + "/data/" + settings.DATA_FILE_NAME


def load_json_file(data_path):
    """讀取json檔案"""
    with open(data_path, 'r') as fp:
        res = json.load(fp)
        fp.close()
    return res


@app.route('/ip_pool/api/v1.0/get_ip', methods=['GET'])
def get_ip():
    """獲取單個ip，驗證有效後返回該ip"""
    ip = get_valid_ip()
    if ip:
        return jsonify(ip)
    else:
        abort(404)


@app.route('/ip_pool/api/v1.0/get_ip_pool', methods=['GET'])
def get_all_ip():
    """獲取整個ip池，不進行驗證，返回json檔案內容"""
    ip_pool = load_json_file(data_path=data_path)
    return jsonify({'get_ip_pool': ip_pool})


@app.route('/ip_pool/api/v1.0/get_ip/<int:ip_id>', methods=['GET'])
def get_a_ip(ip_id):
    """從json檔案中獲取指定ip,不進行驗證，直接返回結果"""
    ip_pool = load_json_file(data_path=data_path)
    if ip_id > len(ip_pool):
        abort(404)
    else:
        return jsonify(ip_pool[ip_id])


if __name__ == '__main__':
    app.run(debug=True)