爬取免费代理，拥有自己的代理池

掌控安全学院

2021年10月25日 17:35

作者：掌控安全-秋紫山

很久很久以前，我有个梦想，就是SQL注入不要被ban，于是ip代理成为了首选，但是奈何钱包有限，只能爬取免费代理，于是借鉴了许多文章，形成了今日的ip代理池。

python编写，内容主要涉及多线程问题、以及爬虫。

一、具体实现流程：

包括三个模块： 获取ip代理模块（爬虫获取ip），代理池模块（实现存储爬取的ip，写入有效ip代理），验证模块（对爬取的ip进行验证），获取模块（获取有效ip，并去重！）。

二、代码分析：（按程序执行顺序）代码分析：（按程序执行顺序）

（一）、入口文件：StartApi.py

1.Init: 获取必要的参数。

2.getproxy: 调用module.GetProxy，多线程使用exec函数，实例化

3.module.GetProxy下的类，开始爬虫获取代理ip。

4.isproxy：调用module.IsProxy和module.OptPool类，循环判断获取代理池的ip，并进行判断。

startproxy：程序入口函数，三个模式：

-c api 可获取已爬取存在ips.txt的有效代理0ip；

-t 100 100个多线程各个爬虫爬取一次获取有效代理；

-t 50 -m 1 -time 10 50个多线程无限次循环爬取有效ip，并且每10分钟进行循环。

具体代码如下：

# -*- coding:utf-8 -*-
# Author:qiuzishan
import module.GetProxy
from module.OptPool import ProxyPool
from module.IsProxy import IsProxy
from threading import Thread
import time
from queue import Queue
from argparse import ArgumentParser
from module.GetApi import GetApi
class StartApi:
def __init__(self):
arg = ArgumentParser(description='baidu_url_collection')
arg.add_argument('-t', help='线程数量默认10个线程', type=int, default=10)
arg.add_argument('-c', help='默认爬取ip进行验证 -c api 即可获取ips.txt的ip值！', type=str, default='getproxy')
arg.add_argument('-m', help='默认只爬取一次，-m 1 可实现无线循环，默认8分钟爬取一次', type=int, default=0)
arg.add_argument('-time', help='默认8分钟爬取一次，-m 10 实现每10分钟爬取一次，搭配-m使用，最好不低于5分钟', type=int, default=8)
args = arg.parse_args()
self.que = Queue()
self.c = args.c
self.t = args.t # 获取线程！！！这很重要！
self.m = args.m
self.time = args.time
def getproxy(self):
'''
有几个代理，就会有几个线程爬取代理！
:param msg:
:return: 循环获取代理，这里可以加入获取代理的实例
'''
thread = []
usemodels = ['BeautifulSoup', 'ProxyPool', '__builtins__', '__cached__', '__doc__', '__file__', '__loader__', '__name__', '__package__', '__spec__', 'datetime', 're', 'requests', 'time']
getmodels = dir(module.GetProxy)
for i in getmodels:
if i not in usemodels:
# 用于测试是否存在代理print(i)
self.que.put('import module.GetProxy\nmodule.GetProxy.' + i + '()')
proxyNum = self.que.qsize() # 有几个爬虫就开启几个线程
for i in range(0, proxyNum):
t = Thread(target=exec, args=(self.que.get(),))
thread.append(t)
for i in thread:
i.start()
# print(f'线程数:{i}开始')
for i in thread:
i.join()
def isproxy(self, msg):
"""
:return: 循环判断代理
"""
pp = ProxyPool()
ip = IsProxy(self.t)
print(msg)
while True:
flag = pp.get_length_pool()
if flag == 0:
print("退出循环")
break # 退出循环
ip.startIsProxy()
def startproxy(self):
'''
多进程同时运行
'''
if self.c == 'api':
ga = GetApi()
ga.start(self.t)
else:
if self.m:
while True:
getproxy = Thread(target=self.getproxy)
isproxy = Thread(target=self.isproxy, args=('代理判断中',))
getproxy.start()
time.sleep(5) # IpPool一开始无数据，因此先爬取加入IpPool列表中，防止出错
isproxy.start()
time.sleep(self.time*60) # 每隔多长时间爬取一次
else:
getproxy = Thread(target=self.getproxy)
isproxy = Thread(target=self.isproxy, args=('代理判断中',))
getproxy.start()
time.sleep(5) # IpPool一开始无数据，因此先爬取加入IpPool列表中，防止出错
isproxy.start()
getproxy.join()
isproxy.join()
if __name__ == '__main__':
print('''
III pppppppp lll
III pp pp lll
III pp pp lll
III pppppppp pp pp OOOOOOO OOOOOOO lll
III pp pp pppppppp OO OO OO OO lll
III pp pp pp OO OO OO OO lll
III pp pp pp OO OO OO OO lll
III pppppppp _________ pp OO OO OO OO lll
pp pp OOOOOOOO OOOOOOOO lll
pp
pp {v 1.0 author:秋紫山 QQ:644976520 欢迎加q交流~~}
pp
''')
start = time.perf_counter()
sa = StartApi()
sa.startproxy()
end = time.perf_counter()
print("use time:"+str(end - start) + "s")

（二）爬取代理ip: module/GetProxy.py

自定义爬虫类：

1.Init：不需要改变，把print的内容稍微改改即可，实现实例化即可启动该爬虫，并把获取的爬虫加入代理池ProxyPool里面

2.get_lxml_by_url：获取需要爬取页面的内容

3.get_proxy_by_lxml：自定义的爬虫，需要返回ip列表。

具体代码：

例如ip88Proxy

# -*- coding:utf-8 -*-
# Author:qiuzishan
import requests
from bs4 import BeautifulSoup
from module.OptPool import ProxyPool
import time
import re
'''
这里可以定义多个代理类，只要保证以“**.**.**.**:**”形式如池即可，然后在getproxy.py中加入多线程即可。
'''
class demoProxy: # demo
'''
Get Agent from web 89ip
'''
def __init__(self): # 自动爬取并且把ip输入代理池！此段函数不需要更改,print 的内容更改一下即可
self.proxy = []
try:
print("demo代理开始爬取")
self.proxy = self.get_proxy_by_lxml(self.get_lxml_by_url())
except Exception as e:
print("出错了：{}\n~aha代理demo凉了，求你了快去看看吧！".format(e))
if self.proxy != []:
pp = ProxyPool()
for p in self.proxy:
pp.get_into_pool(p)
else:
print("demo代理爬取的代理ip为空！代理可能凉了！")
def get_lxml_by_url(self):
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.149 Safari/537.36'
}
r = requests.get('', headers=headers) # 输入需要爬取的网站！
if r.status_code == 200:
response = r.text
return response
else:
exit('Error: Module GetAgent Url Error')
def get_proxy_by_lxml(self, response):
ip = []
# 爬取你的代理包括端口噢，例如，123.12.2.33:8080,然后放进ip列表里。
# 请输入您的爬虫~
# print(ip)
return ip
class ip89Proxy: # 89ip
'''
Get Agent from web 89ip
'''
def __init__(self):
self.proxy = []
try:
print("89ip代理开始爬取")
self.proxy = self.get_proxy_by_lxml(self.get_lxml_by_url())
except Exception as e:
print("出错了：{}\n~aha代理89ip凉了，求你了快去看看吧！".format(e))
if self.proxy != []:
pp = ProxyPool()
for p in self.proxy:
pp.get_into_pool(p)
else:
print("89ip代理爬取的代理ip为空！代理可能凉了！")
def get_lxml_by_url(self):
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.149 Safari/537.36'
}
r = requests.get('https://www.89ip.cn/tqdl.html?num=200&amp;address=&amp;kill_address=&amp;port=&amp;kill_port=&amp;isp=', headers=headers)
if r.status_code == 200:
response = r.text
return response
else:
print('Error: Module GetAgent Url Error')
def get_proxy_by_lxml(self, response):
ip = []
soup = BeautifulSoup(response, 'html5lib')
soup.find_all('div')
for div in soup.find_all('div', style="padding-left:20px;"):
for child in div.children:
if child.string is not None:
if '更好用的代理ip请访问' not in child.string:
ip.append(child.string.strip())
# print(ip)
return ip