代理Ip的抓取与验证

Zss 发表于:

很早之前就有写过代理Ip的爬取,但是没有去验证过这些ip到底是否可不可以使用,网上免费的代理ip基本都是失效的

这次就给西刺的免费代理ip坑惨了,所有抓取到的代理ip全部返回原始ip,搞得我一直怀疑自己使用requests使用代理的方式写错了,

在此纠结了挺久的,各种百度谷歌无解。。。为此,嫩是花了一上午在搞这个问题,唉。。。

后来换了网站验证了一下,发现代理的写法是没有错的,我就说嘛,就这么几句话,怎么可能错了。。。

requests使用代理ip访问的方法,使用代理ip来访问http://ip.cip.cc/,若返回的是此ip则代理成功了,为源ip的话那么失败了

import requests
ip = '103.70.59.140'
port = '22991'
proxy = {'http':'http://%s:%s'%(ip,port),'https':'http://%s:%s'%(ip,port)}
r = requests.get('http://ip.cip.cc/',proxies=proxy,timeout=15)
print(r.text)

后来换了个网站叫做快代理。。。发免费的代理IP还有几个可以用,所以抓取了前面的五页,依次先使用ping来判断ip是否正常,若正常去除延时大于40ms的ip

那么此时的75个ip就不剩多少了,再使用上面的方法依次来判断是否可以使用,因为太慢了,所以使用多线程来测试是否可用,最后将可用的返回出来

抓了75个ip可以使用的每次估计就只有四个左右,免费的果然不靠谱,不过花钱的是真特么的贵。。。想想专门写爬虫的总是要花这些钱的

将这些可用的ip,最后再次访问qq,发现还是一些不能用的。。。我去

[('180.118.134.232', '9000'), ('139.196.51.201', '8118'), ('180.118.134.232', '9000'), ('42.51.206.213', '1080'), ('112.16.28.103', '8060')]
503
200

总共五个可用的,访问三个没响应了,一个成功的,一个服务器错误。。。 好吧

#coding:utf-8
import requests,time
import subprocess
from lxml import etree
from threading import Thread

class Proxy_ip():
    def __init__(self):
        self.all_proxy_ip_port = []
        self.ping_pass_ip_port = []
        self.good_ip_port = []

    def get_proxy_ip(self):
        url_list = ['https://www.kuaidaili.com/free/inha/{}'.format(str(i)) for i in range(1,6)]
        headers = {'Host':'www.kuaidaili.com',
                   'User-Agent':'Mozilla/5.0 (Windows NT 6.3; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/67.0.3396.99 Safari/537.36'
}
        for i in url_list:
            rsp = requests.get(url=i,headers=headers).content
            xml = etree.HTML(rsp)
            ip_info = xml.xpath('//*[@id="list"]/table/tbody/tr/td[1]/text()')
            port_info = xml.xpath('//*[@id="list"]/table/tbody/tr/td[2]/text()')
            for t in zip(ip_info,port_info):
                self.all_proxy_ip_port.append(t)
            time.sleep(1)
        return self.all_proxy_ip_port

    def ping_take(self,ip_port): #传入ip port数组
        ip,port = ip_port
        cmd = 'ping %s'%ip
        reslut = subprocess.Popen(cmd,shell=True,stdout=subprocess.PIPE)
        reslut = reslut.stdout.readlines()
        if len(reslut) == 11: #十一行为通
            reslut = reslut[-1].decode('gbk').strip().split('=')
            ping_delay = int(reslut[-1].strip('ms').strip())
            if ping_delay <= 40:
                self.ping_pass_ip_port.append(ip_port)


    def check_ip_ping(self): #筛选延时小于40ms的ip
        T_list = []
        for i in self.all_proxy_ip_port:
            T = Thread(target=self.ping_take,args=(i,))
            T.start()
            T_list.append(T)
        for i in T_list:
            i.join()

    def ip_valid_task(self,ip_port):
        ip,port = ip_port
        proxy = {'http':'http://%s:%s'%(ip,port),'https':'http://%s:%s'%(ip,port)}
        try:
            requests.get('http://ip.cip.cc/',proxies=proxy,timeout=15)
            self.good_ip_port.append(ip_port)
        except :
            pass

    def check_ip_valid(self):
        S_list = []
        for i in self.ping_pass_ip_port:
            S = Thread(target=self.ip_valid_task,args=(i,))
            S.start()
            S_list.append(S)
        for s in S_list:
            s.join()
        print(self.good_ip_port)

    def main(self):
        self.get_proxy_ip()
        self.check_ip_ping()
        self.check_ip_valid()
        return self.good_ip_port

if __name__ == '__main__':
    proxy_ip = Proxy_ip()
    ip_list = proxy_ip.main()
    for ip,port in ip_list:
        try:
            proxy = {'http':'http://%s:%s'%(ip,port),'https':'http://%s:%s'%(ip,port)}
            r = requests.get('http://www.qq.com',proxies=proxy,timeout=None)
            print(r.status_code)
        except:pass