背景：无论是自己爬虫爬取的IP代理还是购买的IP代理服务都会存在IP不稳定的情况。通过不断的校验来判断IP是否可用。

#!/usr/bin/env python2.7
#coding:utf8
import os,time,sys
#import mymodul
#from lxml import etree
import re
import random
import requests,os
import sys

#每10分钟验证一次IP的可用性
def __extract_ipaddress(text_content):
    result_list=[]
    for line in text_content.split(‘\n‘):
        #从每一行提取ip
        m=re.search("((?:(?:25[0-5]|2[0-4]\d|((1\d{2})|([1-9]?\d)))\.){3}(?:25[0-5]|2[0-4]\d|((1\d{2})|([1-9]?\d))))",line)
        ip,port=‘‘,‘‘
        if m:
            ip = m.group(0)
            # 从包含IP的行提取端口
            m=re.search("\D(\d{2,5})\D*",line[line.find(ip)+len(ip):])
            if m:
                port=m.group(1)
                result_list.append((ip, m.group(1)))
    return result_list

#提取IP，验证黄页
def test_poxy((_ip,_port),test_url=‘http://www.yp.net.cn/schinese//about/AboutCnYPol.asp‘):
    import requests
    try:
        session = requests.session()
        session.proxies = {‘http‘: ‘{}:{}‘.format(_ip,_port)}
        headers = {‘User-Agent‘: ‘Mozilla/5.0 (Windows NT 6.1; WOW64; rv:33.0) Gecko/20100101 Firefox/33.0‘}
        resp = session.get(test_url, timeout=10, headers=headers)
        #print len(resp.content)
    except BaseException as e:
        #print ‘test_poxy error:‘,e
        return (_ip,_port,-1,-1)
    # if len(resp.content)<100000:
    #     print ‘######‘,resp.content
    return (_ip,_port,resp.elapsed.seconds,len(resp.content))

if __name__ == ‘__main__‘:
    from multiprocessing.dummy import Pool as ThreadPool
    #while 1:
    #验证Proxy.txt中的IP，将可用IP写入proxy_ip.txt
    with open(‘/root/scrit/Proxy.txt‘,‘r‘) as f , open(‘/root/scrit/proxy_ip.txt‘,‘wb‘) as fout:
        ip_list = __extract_ipaddress(f.read())
        #print(‘ipcount‘,len(ip_list))
        pool=ThreadPool(30)
        result=pool.map(test_poxy, ip_list)
        pool.close()
        pool.join()
        result = sorted(result, key=lambda d:d[3], reverse=True)
        result=set(result)
        for item in  result:
            if int(item[3])>7000:
                #print item
                fout.write(‘{}:{}\n‘.format(item[0],item[1]))
        fout.close()
        #print mymodul.get_time_str(),u‘一次检测结束‘
        #time.sleep(3)
    Data = open("/root/scrit/proxy_ip.txt").read()
    langth = len(Data)
    if not langth:
        get_IP()
    else:
        sys.exit(0)

本文出自 “echo xiayun” 博客，请务必保留此出处http://linuxerxy.blog.51cto.com/10707334/1893906

python--IP代理池验证可用性