IP代理池学习记录

大家好，欢迎来到IT知识分享网。

免责声明

        本文仅供学习和研究目的使用。所提供的信息和技术仅限于合规和合法的使用场景。请读者在应用相关技术时遵守法律法规，尊重他人的数据隐私和网站使用条款。本文作者对因使用本文信息而产生的任何法律责任或损失不承担责任。

1、初识IP代理池

概述：在爬虫业务中，为了防止本机IP被封，可以使用IP代理来隐藏真实IP地址。因此代理的IP对于爬虫业务是非常的珍贵，对此，我们就尝试利用已学的知识白嫖免费的代理，由此产生了IP代理池。

简易IP代理池设计：

• DB 数据管理
  • Mongo
• Fetcher 数据下载
  • 云代理
  • 快代理
• Validata 数据验证
  • httpbin.org
  • 指定网站
• Request 下载器
  • 专门下载工具
• Scheduler 调度器
  • 协调各工具的使用
• API API接口开发
  • FastAPI
    • 获取IP
    • 获取IP个数

2、日志模块(logging)

概述：logging模块是Python内置的标准模块，主要用于输出运行日志，可以设置输出日志的等级、日志保存路径、日志文件回滚等；

优点：

①、通过设置不同的日志等级，在release版本中只输出重要信息，而不必显示大量的调试信息；

②、开发者可以决定将logging信息输出到什么地方，以及怎么输出；

日志级别从低到高：debug、info、warning、error，critical；

知识点补充：logging中默认只显示warning级别及以上等级的信息，如果想要显示debug或info级别的信息可以通过basicConfig调整默认等级，basicConfig除了能调整等级外还能控制日志的输出格式，将日志写入指定文件的功能：

# 演示的是将默认等级调整到info,不会显示debug级别的信息,format是设置日志的输出格式,可以根据需要自行调整 logging.basicConfig(level=logging.INFO,format='%(asctime)s %(levelname)s %(message)s',filename='文件名')

格式化设置：

%(acstime)s 时间

%(filename)s 日志文件名

%(funcName)s 调用日志的函数名

%(levelname)s 日志的级别

%(module)s 调用日志的模块名

%(message)s 日志信息

%(name)s logger的name

实操：通过本实操掌握logging模块的基本用法(写入文件时是追加模式)。

import logging formats = '%(asctime)s %(levelname)s %(message)s' filenames = './mylog.log' # 配置logging模块 logging.basicConfig(level=logging.INFO,format=formats,filename=filenames) def log_print(): logging.debug('debug ...') logging.info('info ...') logging.warning('warn ...') logging.error('error ...') logging.critical('critical ...') if __name__ == '__main__': log_print()

运行后的mylog.log文件内容如下：

3、开发一个IP代理池

概述：本节将通过从零搭建IP代理池让大家了解搭建过程和思路。

文件结构：

3.1、开发IP代理池的下载器

概述：在开发下载器前先创建一个名为config的python文件用于配置logging的相关信息，方便后面的开发直接使用：

config.py内容如下：

import logging level = logging.DEBUG format = '%(asctime)s --- %(filename)s --- %(levelname)s --- %(message)s' logging.basicConfig(level=level,format=format)

解决完日志的相关配置后，直接上手下载器的开发：

request_util.py内容如下：

import requests from fake_useragent import UserAgent from config import logging from lxml import etree class WebRequest: """ 这是IP代理池的下载器 """ def __init__(self): self.resp = None self.header = {'User-Agent':UserAgent().chrome} self.url = None def get(self,url,header=None,*args,kwargs): """ 本函数主要用于发送get请求 """ try: # 如果调用该方法时没有传值就使用类属性值中的请求头 if not header: header = self.header resp = requests.get(url,headers=header,*args,kwargs) if resp.status_code == 200: resp.encoding = 'utf-8' self.resp = resp return self.resp except Exception as e: logging.warning(f'捕获的异常为{e}') def post(self, url, header=None, *args, kwargs): """ 本函数主要用于发送post请求 """ try: # 如果调用该方法时没有传值就使用类属性值中的请求头 if not header: header = self.header resp = requests.get(url, headers=header, *args, kwargs) if resp.status_code == 200: resp.encoding = 'utf-8' self.resp = resp return self.resp except Exception as e: logging.warning(f'捕获的异常为{e}') def analy_data(self): """ 该函数用于解析数据 """ # 解析响应数据 e = etree.HTML(self.resp.content) return e def get_text(self): """ 该函数用于获取响应的文本信息 """ return self.resp.text def get_json(self): """ 该函数用于获取json数据 """ return self.resp.json() def get_content(self): """ 该函数用于获取数据流 """ return self.resp.content

3.2、开发IP代理池的数据库模块

概述：我们将用mongo作为IP代理池的数据库，首先在db_util文件中创建一个名为db_util的类，然后在构造函数中连接数据库并指定要使用的集合，然后在这个类中创建多个类方法，用于实现①、插入数据；②、获取任意一条ip数据；③、统计某个请求类型的数量；④、删除IP及其对应的指纹；⑤、检测指纹是否存在。

db_util.py内容如下：

import hashlib import pymongo from config import logging from random import randint class db_util(): def __init__(self,host='localhost',port=27017): """ 本函数用于连接客户端并获取指定数据库和集合 """ # 连接Mongo客户端 self.client = pymongo.MongoClient(host=host,port=port) # 选择数据库 self.db = self.client.proxy # 选择ip集合 self.ip = self.db.ip # 选择指纹集合 self.finger = self.db.finger def insert(self,data): """ 本函数用于向Mongo中插入数据 data结构：{'type':'https','ip':'192.168.1.2','port':'4567'} """ # 获取data中的ip ip = data.get('ip') # 对ip进行加密 ip_finger = hashlib.md5(ip.encode()).hexdigest() # 在插入数据前判断该数据是否已经保存到数据库中 if not self.exist_finger(ip_finger): # 插入数据到ip集合中 self.ip.insert_one(data) # 将加密的ip插入到指纹集合中 self.finger.insert_one({'ip':ip_finger}) logging.info(f'成功保存了{data}') else: logging.info(f'该数据已经在数据库中啦：{data}') def get_one_ip(self,type='https'): """ 本函数用于获取Mongo数据库中的任意一条数据 """ # 获取当前数据库中有多少条指定类型的数据 count = self.count(type) if count == 0: return None elif count == 1: one_data = self.ip.find_one({'type':type}) else: # 如果有多条数据 # 做分页处理,每页只有一条数据 num = randint(0,count-1) one_data = self.ip.find({'type':type}).skip(num).limit(1)[0] return one_data def count(self,type='https'): """ 该函数用于获取数据库中指定类型的数据数量, 起初设置该函数是为解决get_ip函数中find_one()方法默认只获取找到的第一条数据,导致可能会一直获取同一条数据的尴尬 """ return self.ip.count_documents({'type':type}) def delete(self,data): """ 删除Mongo中指定的IP """ # 删除ip表中的指定数据 ip_data = data.get('ip') self.ip.delete_one({'ip':ip_data}) logging.info(f'{ip_data}删除成功') # 删除finger表中的指纹 ip_finger = hashlib.md5(ip_data.encode()).hexdigest() self.finger.delete_one({'ip':ip_finger}) def exist_finger(self,hash_id): """ 判断指纹是否存在 hash_id:是已经加密好的ip """ if self.finger.count_documents({'ip':hash_id}): logging.info(f'{id}已存在') return True else: return False def __del__(self): """ 如果没有人调用数据库将自动销毁 """ self.client.close() if __name__ == '__main__': mongo = db_util() # 向数据库中插入多条数据 for i in range(10): mongo.insert({'type':'https','ip':f'192.168.1.{i}','port':'4567'}) # 获取数据库中任意一条数据 get_one_data = mongo.get_one_ip() print(f'获取的数据：{get_one_data}') # 删除数据库中的任意一条数据 mongo.delete({'type':'https','ip':'192.168.1.4','port':'4567'})

运行结果如下：

集合ip的内容如下：

finger的内容如下：

3.3、开发IP代理池的验证模块

概述：验证IP地址是否有效主要是向httpbin.org/get发送请求，该网站会返回请求主机的IP，如果使用了代理去访问，返回的就是代理主机的IP。

validate.py内容如下：

from request_util import WebRequest from config import logging def validate_ip(proxy:dict) -> bool: http_url = 'http://www.httpbin.org/get' https_url = 'https://www.httpbin.org/get' ip_type = proxy.get('type') # 获取IP的请求类型 ip_addr = proxy.get('ip') # 获取ip ip_port = proxy.get('port') # 获取端口号 # 构建代理 proxies = {ip_type:f'{ip_type}://{ip_addr}:{ip_port}'} # 创建请求对象 web = WebRequest() try: # 根据请求类型发送请求 if ip_type == 'http': web.get(http_url,proxies=proxies,timeout=12) else: web.get(https_url,proxies=proxies,timeout=12) except: logging.info(F'该代理请求超时/请求失败：{ip_addr}') return False # 判断代理是否有效 if web.resp: resp_ip = web.get_json().get("origin") # 防止origen有多个返回值 if resp_ip.find(ip_addr) != -1: logging.info(f'该代理有效:{ip_addr}') return True else: logging.info(f'该代理无效：{ip_addr}') return False else: return False

3.4、开发解析模块

概述：本模块调用前面开发的下载器向目标代理网站发送请求，解析模块主要的任务是将代理网站中的指定数据获取出来并返回，这里我只解析了一个网站，实际开发中当然是多多益善了。fetcher_util.py内容如下：

from request_util import WebRequest
from lxml import html
from time import sleep
import re

def yun(page=2):
    """
    本函数主要用于解析某代理网站的数据并将数据返回
    """
    # 创建请求对象
    web = WebRequest()

    for i in range(1,page+1):
        sleep(3)
        # 目标网站
        url = (f'http://www.ip3366.net/free/?stype=3&page={page}')

        # 发送请求
        web.get(url)
        # 解析数据
        e = web.analy_data()

        for i in range(1,16):
            ips = e.xpath(f'//tr[{i}]/td[1]')[0]
            ip_about = html.tostring(ips, encoding='utf-8').decode('utf-8')
            # 这里尝试用切片直接截取,但只能切头不能去尾,最后还是采用正则去拿数据
            ip = re.search('<td>(.+)</td>', ip_about).group(1)

            ports = e.xpath(f'//tr[{i}]/td[2]')[0]
            port_about = html.tostring(ports, encoding='utf-8').decode('utf-8')
            # 这里尝试用切片直接截取,但只能切头不能去尾,最后还是采用正则去拿数据
            port = re.search('<td>(.+)</td>', port_about).group(1)

            types = e.xpath(f'//tr[{i}]/td[4]')[0]
            type_about = html.tostring(types, encoding='utf-8').decode('utf-8')
            # 这里尝试用切片直接截取,但只能切头不能去尾,最后还是采用正则去拿数据
            type = re.search('<td>(.+)</td>', type_about).group(1)

            yield {'ip':ip.lower(),'port':port,'type':type}
            # print({'ip':ip,'port':port,'type':type})

if __name__ == '__main__':
    # 填入要获取的界面数量
    yun(1)

3.5、开发调度模块

dispatch_util.html内容如下：

from db_util import db_util # 导入数据库模块 from fetcher_util import yun # 导入代理数据网站解析 from validate_util import validate_ip # 导入验证ip的模块 def start(): """ 本模块用于调度其它模块实现ip的获取,验证,保存/丢弃 """ # 链接数据库 mongo = db_util() # 下载IP for proxy in yun(6): # 验证IP if validate_ip(proxy): # 保存IP mongo.insert(proxy) if __name__ == '__main__': start()

运行结果如下：

总结：通过IP代理池获取免费代理的结果相当炸裂，能用的极少，大家可以多解析几个免费网站尝试一下。

3.6、获取收费IP

概述：既然白嫖失败了就考虑为我们的IP代理池添加一个获取收费IP的函数，由于是收费IP，所以基本都没什么问题，我们直接获取收费IP的地址，然后直接获取即可：

fetcher_util文件添加如下内容：

def money_player(num=2): """ 本函数用于获取收费代理 """ # 创建请求对象 web = WebRequest() for i in range(num): url = '由代理网站提供,进入该地址就能看到网站提供的(多条)代理信息了,至少包含ip和port' resp = web.get(url) # 这里假设收费代理为我们提供了以下两条代理数据 # 193.48.34.74:4789 # 192.167.48.59:8874 ip,port = resp.split(':') {'ip':ip,'type':'http','port':port} # 不同的网站可能还存在休眠要求,避免高频访问造成IP被封 sleep(10)