Python搭建代理IP池实现检测IP的方法

 更新时间:2019年10月27日 14:57:29   作者:Steven·简谈  
这篇文章主要介绍了Python搭建代理IP池实现检测IP的方法,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧

在获取 IP 时,已经成功将各个网站的代理 IP 获取下来了,然后就需要一个检测模块来对所有的代理进行一轮轮的检测,检测可用就设置为满分,不可用分数就减 1,这样就可以实时改变每个代理的可用情况,在获取有效 IP 的时候只需要获取分数高的 IP

代码地址:https://github.com/Stevengz/Proxy_pool

另外三篇:
Python搭建代理IP池(一)- 获取 IP
Python搭建代理IP池(二)- 存储 IP
Python搭建代理IP池(四)- 接口设置与整体调度

由于代理 IP 的数量非常多,为了提高 IP 的检测效率,这里使用异步请求库 Aiohttp 来进行检测。至于为什么不用抓取时用的 Requests 库,是因为它是一个同步请求库,在发出一个请求之后需要等待网页加载完成之后才能继续执行程序。这个过程会阻塞在等待响应中,如果服务器响应非常慢,一个请求就会需要十几秒,程序不会继续往下执行

异步请求库就解决了这个问题,在请求发出之后,程序可以继续接下去执行其他的事情,当响应到达时会通知程序再去处理这个响应,这样程序就没有被阻塞,可以充分把时间和资源利用起来

添加设置

增加了几个测试用的常量

setting.py

# 数据库地址
HOST = '127.0.0.1'
# MySql端口
MYSQL_PORT = 3306
# MySQl用户名、密码
MYSQL_USERNAME = '***'
MYSQL_PASSWORD = '***'
# 数据库名
SQL_NAME = 'test'

# 代理等级
MAX_SCORE = 30
MIN_SCORE = 0
INITIAL_SCORE = 10

# 代理池数量界限
POOL_UPPER_THRESHOLD = 1000

VALID_STATUS_CODES = [200, 302]

# 测试API,建议抓哪个网站测哪个
TEST_URL = 'http://www.baidu.com'

# 最大批测试量
BATCH_TEST_SIZE = 30

VALID_STATUS_CODES 变量包含了正常的状态码,获取 Response 后需要判断响应的状态,如果状态码在

VALID_STATUS_CODES 这个列表里,则代表代理可用

定义方法

定义了一个类 Tester,init() 方法中建立了一个 MySqlClient 对象,供类中其他方法使用。接下来定义了一个 test_single_proxy() 方法,用来检测单个代理的可用情况,其参数就是被检测的代理

tester.py

import asyncio
import aiohttp
import time
import sys
from aiohttp import ClientError
from db import MySqlClient
from setting import *


class Tester(object):
 def __init__(self):
  self.mysql = MySqlClient()
 
 # 测试单个代理
 async def test_single_ip(self, ip):
  conn = aiohttp.TCPConnector(verify_ssl=False)
  async with aiohttp.ClientSession(connector=conn) as session:
   try:
    if isinstance(ip, bytes):
     ip = ip.decode('utf-8')
    real_ip = 'http://' + ip
    print('正在测试', ip)
    async with session.get(TEST_URL, proxy=real_ip, timeout=15, allow_redirects=False) as response:
     if response.status in VALID_STATUS_CODES:
      self.mysql.max(ip)
      print('代理可用', ip)
     else:
      self.mysql.decrease(ip)
      print('请求响应码不合法 ', response.status, 'IP', ip)
   except (ClientError, aiohttp.client_exceptions.ClientConnectorError, asyncio.TimeoutError, AttributeError):
    self.mysql.decrease(ip)
    print('代理请求失败', ip)
 
 # 主函数
 def run(self):
  print('测试器开始运行')
  try:
   count = self.mysql.count()
   print('当前剩余', count, '个代理')
   for i in range(0, count, BATCH_TEST_SIZE):
    start = i
    stop = min(i + BATCH_TEST_SIZE, count)
    print('正在测试第', start + 1, '-', stop, '个代理')
    test_ip_group = self.mysql.batch(start, stop)
    loop = asyncio.get_event_loop()
    tasks = [self.test_single_ip(ip_tuple[0]) for ip_tuple in test_ip_group]
    loop.run_until_complete(asyncio.wait(tasks))
    sys.stdout.flush()
    time.sleep(5)
  except Exception as e:
   print('测试器发生错误', e.args)


if __name__ == "__main__":
 test = Tester()
 test.run()

test_single_proxy() 方法前面加了 async 关键词,代表这个方法是异步的,方法内部首先创建了 Aiohttp 的 ClientSession 对象,此对象类似于 Requests 的 Session 对象,可以直接调用该对象的 get() 方法来访问页面

运行结果:

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持脚本之家。

相关文章

  • Python Django教程之实现新闻应用程序

    Python Django教程之实现新闻应用程序

    Django是一个用Python编写的高级框架,它允许我们创建服务器端Web应用程序。在本文中,我们将了解如何使用Django创建新闻应用程序,感兴趣的可以尝试一下
    2022-10-10
  • 神经网络python源码分享

    神经网络python源码分享

    这篇文章主要介绍了神经网络python源码分享,具有一定借鉴价值,需要的朋友可以参考下。
    2017-12-12
  • 使用Python给PDF添加目录书签的实现方法

    使用Python给PDF添加目录书签的实现方法

    有时下载到扫描版的 PDF 是不带书签目录的,这样阅读起来很不方便,下面通过 python 实现一个半自动化添加书签目录的脚本,文中通过代码介绍的非常详细,具有一定的参考价值,需要的朋友可以参考下
    2023-10-10
  • tensorflow 打印内存中的变量方法

    tensorflow 打印内存中的变量方法

    今天小编就为大家分享一篇tensorflow 打印内存中的变量方法,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2018-07-07
  • django之如何按日期查询数据

    django之如何按日期查询数据

    这篇文章主要介绍了django之如何按日期查询数据问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
    2023-08-08
  • python中的常量和变量代码详解

    python中的常量和变量代码详解

    这篇文章主要介绍了python中的常量和变量,通过实例代码给大家介绍了局部和全局变量,需要的朋友参考下吧
    2018-07-07
  • Python大数据之网络爬虫的post请求、get请求区别实例分析

    Python大数据之网络爬虫的post请求、get请求区别实例分析

    这篇文章主要介绍了Python大数据之网络爬虫的post请求、get请求区别,结合具体实例形式分析了Python网页爬虫post请求与get请求相关使用技巧,需要的朋友可以参考下
    2019-11-11
  • Django中日期时间型字段进行年月日时分秒分组统计

    Django中日期时间型字段进行年月日时分秒分组统计

    这篇文章主要介绍了Django中日期时间型字段进行年月日时分秒分组统计,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
    2020-11-11
  • pytorch进行上采样的种类实例

    pytorch进行上采样的种类实例

    今天小编就为大家分享一篇pytorch进行上采样的种类实例,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2020-02-02
  • 一文教你用python编写Dijkstra算法进行机器人路径规划

    一文教你用python编写Dijkstra算法进行机器人路径规划

    迪杰斯特拉(Dijkstra)算法是典型最短路径算法,用于计算一个节点到其他节点的最短路径,这篇文章主要给大家介绍了关于利用python编写Dijkstra算法进行机器人路径规划的相关资料,需要的朋友可以参考下
    2021-08-08

最新评论