很早之前就有写过代理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