import requests
from lxml import etree
import time
import threading
class DaliiSpdier():
def __init__(self):
self.url_temp=r'https://www.89ip.cn/index_{}.html'
self.headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.87 Safari/537.36 SE 2.X MetaSr 1.0'}
def get_url_list(self):#构造翻页
url_list=[self.url_temp.format(i) for i in range(20)]
return url_list
def get_resp(self,url):#获取页面 响应
print('正在获取',url)
response=requests.get(url,headers=self.headers)
# response.encoding('utf-8')
time.sleep(2)
return response.content.decode()
def parse(self,response):
html=etree.HTML(response)
ip_list=[]
trs=html.xpath('//table[@class="layui-table"]//tr//td[1]/text()')
for tr in trs:
tr=tr.strip()#替换空格
# tr = etree.tostring(tr,encoding='utf-8').decode('utf-8')
ip="http://"+tr
ip_list.append(ip)
#print(ip_list)
return ip_list
def check_ok(self,ip):#验证
ip_list_ok=[]
try:
r=requests.get('https://baidu.com',proxies={'https':ip},timeout=5)
print(r.status_code)
if r.status_code==200:
ip_list_ok.appednd(ip)
else:
print(f'{ip}不可用')
except:
print(f'{ip}不可用')
return ip_list_ok
def thread_check(self,ip_list):
check_thread=[]
for ip in ip_list:
t=threading.Thread(target=self.check_ok,args=(ip,))
check_thread.append(t)
t.start()
for t in check_thread:
t.join()
def save_ip(self,ip_list_ok):
with open('ip_Ok.txt',"a",encoding='utf-8') as f:
for ip_Ok in ip_list_ok:
f.write(ip_Ok)
f.write('\n')
def run(self):
url_list=self.get_url_list()
for url in url_list:
response=self.get_resp(url)
ip_list=self.parse(response)
ip_list_ok=self.check_ok(ip_list)
self.thread_check(ip_list)
self.save_ip(ip_list_ok)
if __name__ == '__main__':
spider=DaliiSpdier()
spider.run()
|