如何制作一个自己的IP代理池
发布时间
阅读量：
阅读量
开始前的准备
在开始这个项目之前，请确保你已经掌握了基础的编程逻辑思维能力，并具备一定的Python编程经验。不过这些知识点的具体应用方法与实现细节将在后续的文章中进行深入讲解，并会提供详细的代码示例以帮助你更好地理解和实践。
第一步：IP代理池的功能实现
框架
一个IP代理池的主要功能包括IP获取、IP检测两大核心模块，并提供便捷的取用接口。具体来说：主要功能
class IP代理池(object):
'''
IP代理池
用于获取代理IP以及维护代理IP池
'''
def __init__(self):
pass
def 代理爬取(self,num):
pass
def 代理检测(self):
pass
def 代理取用(self):
pass
if __name__ == '__main__':
IPAP = IP代理池()IP的爬取
通过爬虫技术实现网络代理的获取
import time
import httpx
from lxml import etree
headers = {
'User-Agent': "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/104.0.0.0 Safari/537.36"
}
class IP代理池(object):
'''
IP代理池
'''
def __init__(self):
self.IPPool = []
self.快代理链接 = "https://www.kuaidaili.com/free/"
def 代理爬取(self,num):
for i in range(1,num):
url = self.快代理链接+"inha/"+str(i)
res = httpx.get(url, headers=headers)
if(res.status_code == 200):
con = etree.HTML(res.text)
ip_list = con.xpath('//*[@id="list"]/table/tbody/tr/td/text()')
for i in range(len(ip_list)//8):
self.IPPool.append({"IP":ip_list[i*8+0],"Port":ip_list[i*8+1],"匿名度":ip_list[i*8+2],"类型":ip_list[i*8+3]})
print(self.IPPool)
time.sleep(1)
def 代理检测(self):
pass
def 代理取用(self):
pass
if __name__ == '__main__':
IPAP = IP代理池()
IPAP.代理爬取(2)IP的检测
该系统集成了一个IP可用性检测模块，在获取百度首页数据的基础上进行分析，并验证代理服务器的有效性；随后系统会筛选出响应速度优于0.3秒的代理服务器，并将这些结果保存为CSV格式文件中。
import time,os,csv
import httpx
from lxml import etree
headers = {
'User-Agent': "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/104.0.0.0 Safari/537.36"
}
class IP代理池(object):
'''
IP代理池
'''
def __init__(self):
self.IP池 = []
self.可用IP = []
self.快代理链接 = "https://www.kuaidaili.com/free/"
# 如果存在IP代理池文件，则进行读取
if(os.path.exists("IPPool.csv")):
# 读取相关文档
with open("IPPool.csv",'r') as csv_f:
reader = csv.DictReader(csv_f)
for cvs_row in reader:
self.IP池.append({'IP':cvs_row['IP'],'Port':cvs_row['Port'],'匿名度':cvs_row['匿名度'],'类型':cvs_row['类型']})
def 代理爬取(self,num):
for i in range(1,num+1):
url = self.快代理链接+"inha/"+str(i)
res = httpx.get(url, headers=headers)
if(res.status_code == 200):
con = etree.HTML(res.text)
ip_list = con.xpath('//*[@id="list"]/table/tbody/tr/td/text()')
for i in range(len(ip_list)//8):
ipcon = {"IP":ip_list[i*8+0],"Port":ip_list[i*8+1],"匿名度":ip_list[i*8+2],"类型":ip_list[i*8+3]}
if(not ipcon in self.IP池):
self.IP池.append(ipcon)
print("添加IP代理:"+ipcon["IP"]+":"+ipcon["Port"])
else:
print("IP代理:"+ipcon["IP"]+":"+ipcon["Port"]+"已存在")
time.sleep(1)
def 代理检测(self):
for each in self.IP池:
proxies = {each["类型"]+"://":each["类型"]+"://"+each["IP"]+":"+each["Port"]}
res = httpx.get("https://www.baidu.com/", headers=headers, proxies=proxies, timeout=0.3)
if(res.status_code == 200):
if(not each in self.可用IP):
self.可用IP.append(each)
print("IP代理:"+each["IP"]+":"+each["Port"]+"可使用")
else:
print("IP代理:"+each["IP"]+":"+each["Port"]+"已存在")
else:
print("IP代理:"+each["IP"]+":"+each["Port"]+"无效")
with open("IPPool.csv",'w',newline='') as csv_f:
write=csv.writer(csv_f)
# 写入头
row=['IP','Port','匿名度','类型']
write.writerow(row)
for each in self.可用IP:
# 写入数据
row=[each["IP"],each["Port"],each["匿名度"],each["类型"]]
write.writerow(row)
print("代理IP写入完毕！")
def 代理取用(self):
pass
if __name__ == '__main__':
IPAP = IP代理池()
index = input('1.爬取代理\n2.检测代理\n3.取用代理\n请选择:')
while(True):
if(index == '1'):
IPAP.代理爬取(2)
IPAP.代理检测()
elif(index == '2'):
IPAP.代理检测()
elif(index == '3'):
IPAP.代理取用()IP的取用
本来也旨在将IP提取功能封装至类中进行管理与维护。然而Sanic目前强制运行在主线程环境中导致无法实现该功能的一键化流程操作一时半会儿未发现有效的解决方案鉴于能力有限只好采用当前的方法继续运行。为此我们主要目标是通过API接口实现IP获取功能目前采用的是GET方式用于随机获取IP地址未来计划进一步优化该接口以提升效率。
import time,os,csv,random
import httpx
from sanic import Sanic
from lxml import etree
import threading
from sanic.response import json
headers = {
'User-Agent': "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/104.0.0.0 Safari/537.36"
}
class IP代理池(object):
'''
IP代理池
'''
def __init__(self):
self.IP池 = []
self.可用IP = []
self.快代理链接 = "https://www.kuaidaili.com/free/"
# 如果存在IP代理池文件，则进行读取
if(os.path.exists("IPPool.csv")):
# 读取相关文档
with open("IPPool.csv",'r') as csv_f:
reader = csv.DictReader(csv_f)
for cvs_row in reader:
self.IP池.append({'IP':cvs_row['IP'],'Port':cvs_row['Port'],'匿名度':cvs_row['匿名度'],'类型':cvs_row['类型']})
def 代理爬取(self,num):
for i in range(num):
url = self.快代理链接+"inha/"+str(i+1)
try:
res = httpx.get(url, headers=headers)
except:
print("爬取超时")
time.sleep(1)
if(res.status_code == 200):
con = etree.HTML(res.text)
ip_list = con.xpath('//*[@id="list"]/table/tbody/tr/td/text()')
for i in range(len(ip_list)//8):
ipcon = {"IP":ip_list[i*8+0],"Port":ip_list[i*8+1],"匿名度":ip_list[i*8+2],"类型":ip_list[i*8+3]}
if(not (ipcon in self.IP池)):
self.IP池.append(ipcon)
print("添加IP代理:"+ipcon["IP"]+":"+ipcon["Port"])
else:
print("IP代理:"+ipcon["IP"]+":"+ipcon["Port"]+"已存在")
def 代理检测(self):
for each in self.IP池:
proxies = {each["类型"]+"://":each["类型"]+"://"+each["IP"]+":"+each["Port"]}
try:
res = httpx.get("https://www.baidu.com/", headers=headers, proxies=proxies, timeout=0.3)
except:
print("检测超时，检测未通过")
if(res.status_code == 200):
if(not each in self.可用IP):
self.可用IP.append(each)
print("IP代理:"+each["IP"]+":"+each["Port"]+"可使用")
else:
print("IP代理:"+each["IP"]+":"+each["Port"]+"已存在")
else:
print("IP代理:"+each["IP"]+":"+each["Port"]+"无效")
with open("IPPool.csv",'w',newline='') as csv_f:
write=csv.writer(csv_f)
# 写入头
row=['IP','Port','匿名度','类型']
write.writerow(row)
for each in self.可用IP:
# 写入数据
row=[each["IP"],each["Port"],each["匿名度"],each["类型"]]
write.writerow(row)
print("代理IP写入完毕！")
IPAP = IP代理池()
app = Sanic("IPPoll")
@app.route("/")
async def IP代理取用(request):
num = random.randint(0,len(IPAP.IP池)-1)
return json({
'IP': IPAP.IP池[num]['IP'],
'Port': IPAP.IP池[num]['Port'],
'Type': IPAP.IP池[num]['类型']
})
def IP爬取():
index = input('1.爬取代理\n2.检测代理\n请选择:\n')
while(True):
if(index == '1'):
IPAP.代理爬取(50)
IPAP.代理检测()
elif(index == '2'):
IPAP.代理检测()
if __name__ == '__main__':
threading.Thread(target=IP爬取).start()
app.run(host="localhost", port=8000)
使用方式：
直接访问 “http://localhost:8000/” 即可。
至此为止已经完成了整个项目的开发基础架构设计工作。同样可以进一步优化系统性能和用户体验功能模块。这些功能大家自行探索配置吧。或者如果需要引入一个用户友好的界面的话相应地增加相应的配置项除此之外还有一些其他功能可选
最后提醒大家：千万记得点赞、关注和收藏哦~
如果有任何其他问题，请在评论区留言询问。之后若有时间也会陆续为大家带来优化后的资源，这一切都是为了让大家使用起来更加便捷。
全部评论 (0)
还没有任何评论哟~