小白学pyton 爬虫五

Tinken · 发表于 2020-10-3 16:09:25

本帖最后由 Tinken 于 2020-10-6 01:27 编辑

ip地址池 —— 爬虫必备
小白学习阶段，首选免费代理ip，需要做如下几部：
选择免费代理网站：西西、快代理...等
安装数据库：Redis（以后再考虑持久化的问题）
然后是爬免费的代理ip、检测、存储、开放接口、然后再反复检测IP有效性

第一步：安装Redis
下载地址：https://github.com/tporadowski/redis/releases
cmd 切换路径到redis目录，然后启动服务器：redis-server.exe redis.windows.conf
服务器cmd窗口保持打开状态，不要关闭
然后另开cmd窗口，启动客户端：redis-cli.exe -h 127.0.0.1 -p 6379
小插曲，客户端报错：Could not connect to Redis at 127.0.0.1:6379: 由于目标计算机积极拒绝，无法连接。
解决办法：
关闭上面的第一个cmd，redis服务器，通过redis目录中的：redis-server.exe 文件启动服务器
启动redis-server.exe 会有一个弹窗，网络是否允许，这个可能就是出现上面计算机拒绝，无法连接的原因。点击允许即可
再通过cmd：redis-cli.exe -h 127.0.0.1 -p 6379
或者直接在文件中双击：redis-cli.exe 均可成功启动客服端

完美解决！

忍不住尝试了一下，set和get，操作很简单

-------------------------------------------
redis 可视化了解工具，如果无法忍受cmd黑白dos窗口，可以自由选择
Redis Desktop Manager（又名 RDM）是一款适用于 Windows、Linux、MacOS 和 iPadOS 的 Redis 数据库快速管理应用程序
下载地址：https://redisdesktop.com/
省略一万字……

第二步 python 安装redis模块
命令行：pip install redis

项目引入：import redis

第三步寻找IP代理网站，爬取IP，写入redis
选择的代理网站：https://www.kuaidaili.com/free/intr/1/
很不幸的是：503错误

应该ip封禁了，原来爬代理ip地址也需要代理ip，可怜我还在测试中

经过分析，requests请求报503的原因是，同一时间内进行了2次请求，请求时间过快
改进后，去掉了第一次对url进行res.status_code == 200的判断，这样就不用请求两次了
然后在每一个循环后增加延时 time.sleep(5)避免再次出现503

最后的结果非常，成功将代理IP数据存入redis

第三步对存入redis的ip进行验证，检测
设置一个定时任务，定时执行：将无法使用ip删除，保留可用IP
安装定时任务模块：pip install apscheduler

# 以后台的方式运行
sched = BlockingScheduler()
sched.add_job(aps_detection_ip, 'interval', seconds=10, args=[redis_conn_1])
sched.start()

复制代码

经过以上几个步骤，最终初步完成ip代理池功能
附上源码

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# @ClassName proxy
# @Description TODO
# @Author lanlo
# @Date 2020-10-03 22:48
# @Version 1.0
import requests
from bs4 import BeautifulSoup
import redis
import time
import random
import datetime
# 存储在redis中
# from apscheduler.jobstores.redis import RedisJobStore
# 存储在mongo中
# from apscheduler.jobstores.mongodb import MongoDBJobStore
# 存储在数据库中
# from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
# 以后台的方式运行
from apscheduler.schedulers.background import BackgroundScheduler
# 以阻塞的方式运行, 前台运行
from apscheduler.schedulers.background import BlockingScheduler
# 快代理国内高匿
url_inha = "https://www.kuaidaili.com/free/inha/"
# 快代理国内普通
url_intr = "https://www.kuaidaili.com/free/intr/"
# 本地redis连接
pool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True)
# 存储爬取的代理ip
redis_conn_0 = redis.Redis(connection_pool=pool, max_connections=10, db=0)
# 存储经过检测后，可用的ip
redis_conn_1 = redis.Redis(connection_pool=pool, max_connections=10, db=1)
# 模拟浏览器的请求头
headers = {
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/67.0.3396.62 Safari/537.36'
}
# 将ip和端口写入redis数据库
def add_ip(ip, port, redis_conn):
redis_conn.zadd("ip", {ip: port}, nx=False)
print("redis 添加 --> {}:{} ".format(ip, port))
def spider_ip(url, redis_conn):
print("从{}中：爬取免费代理ip".format(url))
try:
if "kuaidaili" in url:
for i in range(1, 6):
# 拼接页面后的链接
url_ = url + str(i)
print("循环{}:{}".format(i, url_))
# requests请求，获取html页面
res = requests.get(url_, headers=headers)
print(res.status_code)
# 将html页面转化为BeautifulSoup文件，并设置解析器
soup = BeautifulSoup(res.text, "html.parser")
# 经过查看元素，我们需要的ip数据都在一个表格中，我们查找所有行，对每一行的ip和端口进行单独处理
for child in soup.find_all("tr"):
# 会有页面其他行搜索进来，使用try避免后续错误
try:
# 分析元素得到ip
ip = child.select("td[data-title='IP']")[0].string
# 分析元素得到端口号
port = child.select("td[data-title='PORT']")[0].string
print("以获取：{}:{}".format(ip, port))
# 调用redis写入函数，将数据写入redis
add_ip(str(ip), str(port), redis_conn)
except:
print("*** 找到一个不含IP的tr ***")
print("第{}代理IP获取页完成".format(i))
# 延迟10秒，避免应请求太快被服务器限制IP
time.sleep(5)
else:
print("该网址{}无法获取代理ip地址，请检查源码".format(url))
except:
print("网址{}请求失败".format(url))
# 获取redis 拥有ip的数量
def get_ip_num(redis_conn):
num = redis_conn.zcard("ip")
print("redis 拥有ip的数量：{}个".format(num))
return num
# 获取ip的端口
def get_port(ip,redis_conn):
port = redis_conn.zscore("ip", ip)
port = str(port).replace(".0", "")
return port
# 随机获取一个ip
def get_random_ip(redis_conn):
# 获取ip数量
end_num = get_ip_num(redis_conn)
if end_num == 0:
# ip地址池没有ip了，应该去爬取更多的ip
spider_ip(url_intr, redis_conn_0)
return ""
# 从现有ip数量中，获得一个随机编号
num = random.randint(0, end_num)
# 获得随机的ip
random_ip = redis_conn.zrange("ip", num, num)
# 如果随机ip获得不成功
if not random_ip:
return ""
random_ip = str(random_ip[0]).replace("b", '').replace("'", "")
port = get_port(random_ip, redis_conn)
# 返回ip 和 port
print("获得随机IP:{}，{}".format(random_ip, port))
return random_ip, port
# 删除redis数据库里的ip
def remove_ip(ip,redis_conn):
redis_conn.zrem("ip", ip)
print("已删除ip:{}".format(ip))
# 检测ip是否可用
def aps_detection_ip(redis_conn):
# 获得一个随机ip
res = get_random_ip(redis_conn)
if not res:
print("获得随机ip失败")
return ""
ip = res[0]
port = res[1]
try:
print("{}:{} 检测中...".format(ip, port))
requests.get("http://www.vrpip.com", proxies={'http': '{ip}:{port}'.format(ip=ip, port=port)}, timeout=10)
print("可用ip：{}:{}".format(ip, port))
add_ip(str(ip), str(port), redis_conn_1)
except Exception:
# ip错误失效就删除
remove_ip(ip, redis_conn)
def text_print(int):
print("测试函数:{}".format(int))
# 以后台的方式运行
sched = BlockingScheduler()
sched.add_job(aps_detection_ip, 'interval', seconds=15, args=[redis_conn_1])
if __name__ == '__main__':
# 暂时单独跑高匿分类和普通分类，未做分类的循环
spider_ip(url_inha, redis_conn_0)
# spider_ip(url_intr, redis_conn_0)
print(datetime.datetime.now())
sched.start()

复制代码

虽然代码还有很多不完善，还有很多漏洞，但不重要
重要的是，我们有了一个清晰的概率，可以自己写出来基本的东西，还怕完善不完善吗？再到后面实战、实操，将ip代理池应用到正式项目中的时候再来改。

Tinken · 发表于 2020-10-6 01:23:22

本帖最后由 Tinken 于 2020-10-6 02:05 编辑

新增 89代理的ip获取功能
新增西拉代理的ip获取功能
新增 ip地址获取接口
优化从redis获取地址为空的问题

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# @ClassName proxy
# @Description TODO
# @Author lanlo
# @Date 2020-10-03 22:48
# @Version 1.0
import requests
from bs4 import BeautifulSoup
import redis
import time
import random
import datetime
# 存储在redis中
# from apscheduler.jobstores.redis import RedisJobStore
# 存储在mongo中
# from apscheduler.jobstores.mongodb import MongoDBJobStore
# 存储在数据库中
# from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
# 以后台的方式运行
from apscheduler.schedulers.background import BackgroundScheduler
# 以阻塞的方式运行, 前台运行
from apscheduler.schedulers.background import BlockingScheduler
# 快代理国内高匿
url_inha = "https://www.kuaidaili.com/free/inha/"
# 快代理国内普通
url_intr = "https://www.kuaidaili.com/free/intr/"
# 89代理
url_89 = "https://www.89ip.cn/index_.html"
# 西拉代理
url_xl = "http://www.xiladaili.com/gaoni/"
# 本地redis连接
pool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True)
# 存储爬取的代理ip
redis_conn = redis.Redis(connection_pool=pool, max_connections=10, db=0)
# 模拟浏览器的请求头
headers = {
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/67.0.3396.62 Safari/537.36'
}
# 将ip和端口写入redis数据库
def add_ip(ip, port, redis_conn):
redis_conn.zadd("ip", {ip: port}, nx=False)
print("redis 添加 --> {}:{} ".format(ip, port))
def spider_ip(url, redis_conn):
print("从{}中：爬取免费代理ip".format(url))
try:
# 从kuaidaili获得免费代理ip
if "kuaidaili" in url:
for i in range(1, 6):
# 拼接页面后的链接
url_ = url + str(i)
print("循环{}:{}".format(i, url_))
# requests请求，获取html页面
res = requests.get(url_, headers=headers)
print(res.status_code)
# 将html页面转化为BeautifulSoup文件，并设置解析器
soup = BeautifulSoup(res.text, "html.parser")
# 经过查看元素，我们需要的ip数据都在一个表格中，我们查找所有行，对每一行的ip和端口进行单独处理
for child in soup.find_all("tr"):
# 会有页面其他行搜索进来，使用try避免后续错误
try:
# 分析元素得到ip
ip = child.select("td[data-title='IP']")[0].string
# 分析元素得到端口号
port = child.select("td[data-title='PORT']")[0].string
print("以获取：{}:{}".format(ip, port))
# 调用redis写入函数，将数据写入redis
add_ip(str(ip), str(port), redis_conn)
except:
print("*** 找到一个不含IP的tr ***")
print("第{}页代理IP获取完成".format(i))
# 延迟10秒，避免应请求太快被服务器限制IP
time.sleep(5)
# 从89ip获得免费代理ip
elif "89ip" in url:
for i in range(1, 6):
# 拼接页面后的链接
url_ = "https://www.89ip.cn/index_{}.html".format(i)
print("循环{}:{}".format(i, url_))
# requests请求，获取html页面
res = requests.get(url_, headers=headers)
print("页面响应：", res.status_code)
# 将html页面转化为BeautifulSoup文件，并设置解析器
soup = BeautifulSoup(res.text, "html.parser")
# 经过查看元素，我们需要的ip数据都在一个表格中，我们查找所有行，对每一行的ip和端口进行单独处理
for child in soup.find_all("tr"):
# 会有页面其他行搜索进来，使用try避免后续错误
try:
# 分析元素得到ip
ip = child.select("td")[0].string.replace("\n", '').replace("\t", "")
# 分析元素得到端口号
port = child.select("td")[1].string.replace("\n", '').replace("\t", "")
print("已获取：{}:{}".format(ip, port))
# 调用redis写入函数，将数据写入redis
add_ip(str(ip), str(port), redis_conn)
except:
print("*** 找到一个不含IP的tr ***")
print("第{}页代理IP获取完成".format(i))
# 延迟10秒，避免应请求太快被服务器限制IP
time.sleep(5)
# 从西拉获取代理ip
elif "xiladaili" in url:
for i in range(1, 2):
# 拼接页面后的链接
url_ = url + str(i)
print("循环{}:{}".format(i, url_))
# requests请求，获取html页面
res = requests.get(url_, headers=headers)
print("页面响应：", res.status_code)
# 将html页面转化为BeautifulSoup文件，并设置解析器
soup = BeautifulSoup(res.text, "html.parser")
# 经过查看元素，我们需要的ip数据都在一个表格中，我们查找所有行，对每一行的ip和端口进行单独处理
for child in soup.find_all("tr"):
# 会有页面其他行搜索进来，使用try避免后续错误
try:
# 分析元素得到ip
ip = child.select("td")[0].string.split(":")[0]
port = child.select("td")[0].string.split(":")[1]
print("已获取：{}:{}".format(ip, port))
# 调用redis写入函数，将数据写入redis
add_ip(str(ip), str(port), redis_conn)
except:
print("*** 找到一个不含IP的tr ***")
print("第{}页代理IP获取完成".format(i))
# 延迟10秒，避免应请求太快被服务器限制IP
time.sleep(5)
else:
print("该网址{}无法获取代理ip地址，请检查源码".format(url))
except:
print("网址{}请求失败".format(url))
# 获取redis 拥有ip的数量
def get_ip_num(redis_conn):
num = redis_conn.zcard("ip")
print("redis 拥有ip的数量：{}个".format(num))
return num
# 获取ip的端口
def get_port(ip,redis_conn):
port = redis_conn.zscore("ip", ip)
port = str(port).replace(".0", "")
return port
# 随机获取一个ip
def get_random_ip(redis_conn):
# 获取ip数量
end_num = get_ip_num(redis_conn)
if end_num == 0:
# ip地址池没有ip了，应该去爬取更多的ip
spider_ip(url_intr, redis_conn)
return ""
# 从现有ip数量中，获得一个随机编号
num = random.randint(0, end_num-1)
# 获得随机的ip
random_ip = redis_conn.zrange("ip", num, num)
# 如果随机ip获得不成功
if not random_ip:
return ""
random_ip = str(random_ip[0]).replace("b", '').replace("'", "")
port = get_port(random_ip, redis_conn)
# 返回ip 和 port
print("获得随机IP:{}，{}".format(random_ip, port))
return random_ip, port
# 获得代理ip的接口
def get_proxy_ip():
# 获取ip数量
end_num = get_ip_num(redis_conn)
if end_num == 0:
return ""
# 从现有ip数量中，获得一个随机编号
num = random.randint(0, end_num-1)
print("获得redis的ip编号为:{}".format(num))
# 获得随机的ip
random_ip = redis_conn.zrange("ip", num, num)
# 如果随机ip获得不成功
if len(random_ip) == 0:
return ""
random_ip = str(random_ip[0]).replace("b", '').replace("'", "")
port = get_port(random_ip, redis_conn)
# 返回ip 和 port
print("获得随机IP:{}，{}".format(random_ip, port))
return {'http': '{}:{}'.format(random_ip, port)}
# 删除redis数据库里的ip
def remove_ip(ip,redis_conn):
redis_conn.zrem("ip", ip)
print("已删除ip:{}".format(ip))
# 检测ip是否可用
def aps_detection_ip(redis_conn):
# 获得一个随机ip
res = get_random_ip(redis_conn)
if not res:
print("获得随机ip失败")
return ""
ip = res[0]
port = res[1]
try:
print("{}:{} 检测中...".format(ip, port))
requests.get("http://www.vrpip.com", proxies={'http': '{}:{}'.format(ip, port)}, timeout=5)
print("可用ip：{}:{}".format(ip, port))
except Exception:
# ip错误失效就删除
remove_ip(ip, redis_conn)
def text_print(int):
print("测试函数:{}".format(int))
# 以后台的方式运行、15秒检测一个
sched = BlockingScheduler()
sched.add_job(aps_detection_ip, 'interval', seconds=6, args=[redis_conn])
if __name__ == '__main__':
# 暂时单独跑高匿分类和普通分类，未做分类的循环
# spider_ip(url_inha, redis_conn)
# spider_ip(url_intr, redis_conn)
# spider_ip(url_89, redis_conn)
spider_ip(url_xl, redis_conn)
print(datetime.datetime.now())
sched.start()

复制代码

Tinken · 发表于 2020-10-6 15:51:22

本帖最后由 Tinken 于 2020-10-6 22:17 编辑

新增多线程代理ip获取功能：加快免费代理ip获取速度
新增多线程ip检测功能：加快ip可用性的检测速度

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# @ClassName proxy_pool
# @Description TODO 代理ip 地址池
# @Author lanlo
# @Date 2020-10-03 22:48
# @Version 1.0
import requests
from bs4 import BeautifulSoup
import redis
import time
import random
import datetime
# 多线程模块
import threading
# 定时任务模块
from apscheduler.schedulers.background import BlockingScheduler
# 线程池
get_ip_threads = []
ip_threads = []
# 快代理国内高匿
url_inha = "https://www.kuaidaili.com/free/inha/"
# 快代理国内普通
url_intr = "https://www.kuaidaili.com/free/intr/"
# 89代理
url_89 = "https://www.89ip.cn/index_.html"
# 西拉代理
url_xl = "http://www.xiladaili.com/gaoni/"
# 本地redis连接
pool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True)
# 存储爬取的代理ip
redis_conn = redis.Redis(connection_pool=pool, max_connections=10, db=0)
# 模拟浏览器的请求头
headers = {
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/67.0.3396.62 Safari/537.36'
}
# 将ip和端口写入redis数据库
def add_ip(ip, port, redis_conn):
redis_conn.zadd("ip", {ip: port}, nx=False)
print("redis 添加 --> {}:{} ".format(ip, port))
def spider_ip(url, redis_conn):
print("从{}中：爬取免费代理ip".format(url))
try:
# 从kuaidaili获得免费代理ip
if "kuaidaili" in url:
print("从kuaidaili获取代理ip")
for i in range(1, 6):
# 拼接页面后的链接
url_ = url + str(i)
print("循环{}:{}".format(i, url_))
# requests请求，获取html页面
res = requests.get(url_, headers=headers)
print(res.status_code)
# 将html页面转化为BeautifulSoup文件，并设置解析器
soup = BeautifulSoup(res.text, "html.parser")
# 经过查看元素，我们需要的ip数据都在一个表格中，我们查找所有行，对每一行的ip和端口进行单独处理
for child in soup.find_all("tr"):
# 会有页面其他行搜索进来，使用try避免后续错误
try:
# 分析元素得到ip
ip = child.select("td[data-title='IP']")[0].string
# 分析元素得到端口号
port = child.select("td[data-title='PORT']")[0].string
print("以获取：{}:{}".format(ip, port))
# 调用redis写入函数，将数据写入redis
add_ip(str(ip), str(port), redis_conn)
except:
print("*** 找到一个不含IP的tr ***")
print("快代理免费代理第{}页代理IP 获取完成".format(i))
# 延迟10秒，避免应请求太快被服务器限制IP
time.sleep(5)
# 从89ip获得免费代理ip
elif "89ip" in url:
print("从89ip获取代理ip")
for i in range(1, 6):
# 拼接页面后的链接
url_ = "https://www.89ip.cn/index_{}.html".format(i)
print("循环{}:{}".format(i, url_))
# requests请求，获取html页面
res = requests.get(url_, headers=headers)
print("页面响应：", res.status_code)
# 将html页面转化为BeautifulSoup文件，并设置解析器
soup = BeautifulSoup(res.text, "html.parser")
# 经过查看元素，我们需要的ip数据都在一个表格中，我们查找所有行，对每一行的ip和端口进行单独处理
for child in soup.find_all("tr"):
# 会有页面其他行搜索进来，使用try避免后续错误
try:
# 分析元素得到ip
ip = child.select("td")[0].string.replace("\n", '').replace("\t", "")
# 分析元素得到端口号
port = child.select("td")[1].string.replace("\n", '').replace("\t", "")
print("已获取：{}:{}".format(ip, port))
# 调用redis写入函数，将数据写入redis
add_ip(str(ip), str(port), redis_conn)
except:
print("*** 找到一个不含IP的tr ***")
print("89ip免费代理第{}页代理IP 获取完成".format(i))
# 延迟10秒，避免应请求太快被服务器限制IP
time.sleep(5)
# 从西拉获取代理ip
elif "xiladaili" in url:
print("从西拉获取代理ip")
for i in range(1, 2):
# 拼接页面后的链接
url_ = url + str(i)
print("循环{}:{}".format(i, url_))
# requests请求，获取html页面
res = requests.get(url_, headers=headers)
print("页面响应：", res.status_code)
# 将html页面转化为BeautifulSoup文件，并设置解析器
soup = BeautifulSoup(res.text, "html.parser")
# 经过查看元素，我们需要的ip数据都在一个表格中，我们查找所有行，对每一行的ip和端口进行单独处理
for child in soup.find_all("tr"):
# 会有页面其他行搜索进来，使用try避免后续错误
try:
# 分析元素得到ip
ip = child.select("td")[0].string.split(":")[0]
port = child.select("td")[0].string.split(":")[1]
print("已获取：{}:{}".format(ip, port))
# 调用redis写入函数，将数据写入redis
add_ip(str(ip), str(port), redis_conn)
except:
print("*** 找到一个不含IP的tr ***")
print("西拉免费代理第{}页代理IP 获取完成".format(i))
# 延迟10秒，避免应请求太快被服务器限制IP
time.sleep(5)
else:
print("该网址{}无法获取代理ip地址，请检查源码".format(url))
except:
print("网址{}请求失败".format(url))
# 获取redis 拥有ip的数量
def get_ip_num(redis_conn):
num = redis_conn.zcard("ip")
print("redis 拥有ip的数量：{}个".format(num))
return num
# 获取ip的端口
def get_port(ip,redis_conn):
port = redis_conn.zscore("ip", ip)
port = str(port).replace(".0", "")
return port
# 随机获取一个ip
def get_random_ip(redis_conn):
# 获取ip数量
end_num = get_ip_num(redis_conn)
if end_num == 0:
# ip地址池没有ip了，应该去爬取更多的ip
spider_ip(url_intr, redis_conn)
return ""
# 从现有ip数量中，获得一个随机编号
num = random.randint(0, end_num-1)
# 获得随机的ip
random_ip = redis_conn.zrange("ip", num, num)
# 如果随机ip获得不成功
if not random_ip:
return ""
random_ip = str(random_ip[0]).replace("b", '').replace("'", "")
port = get_port(random_ip, redis_conn)
# 返回ip 和 port
print("获得随机IP:{}，{}".format(random_ip, port))
return random_ip, port
# 获得代理ip的接口
def get_proxy_ip():
# 获取ip数量
end_num = get_ip_num(redis_conn)
if end_num == 0:
return ""
# 从现有ip数量中，获得一个随机编号
num = random.randint(0, end_num-1)
print("获得redis的ip编号为:{}".format(num))
# 获得随机的ip
random_ip = redis_conn.zrange("ip", num, num)
# 如果随机ip获得不成功
if len(random_ip) == 0:
return ""
random_ip = str(random_ip[0]).replace("b", '').replace("'", "")
port = get_port(random_ip, redis_conn)
# 返回ip 和 port
print("获得随机IP:{}，{}".format(random_ip, port))
return {'http': '{}:{}'.format(random_ip, port)}
# 删除redis数据库里的ip
def remove_ip(ip,redis_conn):
redis_conn.zrem("ip", ip)
print("已删除ip:{}".format(ip))
# 检测ip是否可用
def aps_detection_ip(redis_conn):
# 获得一个随机ip
res = get_random_ip(redis_conn)
if not res:
print("获得随机ip失败")
return ""
ip = res[0]
port = res[1]
try:
print("{}:{} 检测中...".format(ip, port))
requests.get("http://www.vrpip.com", proxies={'http': '{}:{}'.format(ip, port)}, timeout=5)
print("可用ip：{}:{}".format(ip, port))
except Exception:
# ip错误失效就删除
remove_ip(ip, redis_conn)
# 以后台的方式运行、15秒检测一个
sched = BlockingScheduler()
sched.add_job(aps_detection_ip, 'interval', seconds=6, args=[redis_conn])
# 多线程获取ip
def ip_is_enable(ip, port):
try:
print("{}:{} 检测中...".format(ip, port))
requests.get("http://www.vrpip.com", proxies={'http': '{}:{}'.format(ip, port)}, timeout=5)
print("可用ip：{}:{}".format(ip, port))
except Exception:
# ip错误失效就删除
remove_ip(ip, redis_conn)
# 多线程开启检测ip
def all_ip_thread():
end_num = get_ip_num(redis_conn)
for i in range(0, end_num):
if i == end_num:
i = i - 1
ip = redis_conn.zrange("ip", i, i)
ip = str(ip[0]).replace("b", '').replace("'", "")
port = get_port(ip, redis_conn)
ip_thread = threading.Thread(target=ip_is_enable, args=(ip, port), name="all_ip_thread_".format(i))
ip_thread.start()
ip_threads.append(ip_thread)
def get_all_ip_thread():
get_ip_thread_inha = threading.Thread(target=spider_ip, args=(url_inha, redis_conn))
get_ip_thread_intr = threading.Thread(target=spider_ip, args=(url_intr, redis_conn))
get_ip_thread_89 = threading.Thread(target=spider_ip, args=(url_89, redis_conn))
get_ip_thread_xl = threading.Thread(target=spider_ip, args=(url_xl, redis_conn))
get_ip_thread_inha.start()
get_ip_thread_intr.start()
get_ip_thread_89.start()
get_ip_thread_xl.start()
get_ip_threads.append(get_ip_thread_inha)
get_ip_threads.append(get_ip_thread_intr)
get_ip_threads.append(get_ip_thread_89)
get_ip_threads.append(get_ip_thread_xl)
if __name__ == '__main__':
print(datetime.datetime.now())
# 快代理高匿分类
# spider_ip(url_inha, redis_conn)
# 快代理普通分类
# spider_ip(url_intr, redis_conn)
# 89代理
# spider_ip(url_89, redis_conn)
# 西拉代理
# spider_ip(url_xl, redis_conn)
# 多线程获取免费代理ip
get_all_ip_thread()
for t in get_ip_threads:
t.join()
# 多线程检测ip是否可用
all_ip_thread()
for t in ip_threads:
t.join()
# 定时任务检测ip
# sched.start()

复制代码

		自动登录	找回密码
密码			注册VRPIP

小白 学pyton 爬虫五

本帖子中包含更多资源

小白学pyton 爬虫五