Python3.x：免费代理ip的批量获取并入库

时间：2023-03-08 19:04:47

Python3.x：免费代理ip的批量获取并入库

一、简介

网络爬虫的世界，向来都是一场精彩的攻防战。现在许多网站的反爬虫机制在不断的完善，其中最令人头疼的，莫过于直接*你的ip。但是道高一尺魔高一丈，在爬取网页的时候，使用上代理ip，便可以有效的避免自己的ip被*。

想要使用代理ip，目前你可以去相应的代理网站购买代理ip（如果是大型的项目还是推荐去购买），也可以去使用一些代理网站提供的免费的代理ip，不过这些ip还是存在很多问题的，有些不可用，有些不稳定，有些时效短。不过如果量大的话，还是有不少可以使用的。

基于这个目的，利用Python的requests库写了一个简单的批量获取免费代理ip的程序，其中包括“下载+验证”程序。下面将简单介绍代码思路和使用方法。

二、Python实现思路

1，确定获取免费代理ip的网页

目前有些提供免费代理ip网站有以下三类情况：

  • 所有的免费代理ip信息在网页标签中
  • 所有的免费代理ip信息在网页标签中，不过使用了一些隐藏标签
  • 所有的免费代理ip信息在图片中

本文选择第一类网页进行提取（这一类的网站的数量也是较多的，基本满足小规模的使用需求），此选取了以下5个网站：

  • 无忧代理 : http://www.data5u.com/
  • 快代理 : https://www.kuaidaili.com/
  • 小舒代理 : http://www.xsdaili.com/
  • 西刺代理 : http://www.xicidaili.com/
  • 89免费代理: http://www.89ip.cn/

2，下载代理IP（parse_html.py）

2.1 获取网页

解析一个网页，第一步就是先获取页面。因为有多个页面要获取，为了方便就编写一个获取页面的函数，便于之后进行调用。函数如下：

# 1. 无忧代理  : http://www.data5u.com/

# 2. 快代理   : https://www.kuaidaili.com/

# 3. 小舒代理  : http://www.xsdaili.com/

# 4. 西刺代理  : http://www.xicidaili.com/

# 5. 89免费代理: http://www.89ip.cn/

from bs4 import BeautifulSoup

import requests

import re

def get_html(url, open_proxy=False, ip_proxies=None):

    """

    获取页面的html文件

    :param url: 待获取页面的链接

    :param open_proxy: 是否开启代理，默认为False

    :param ip_proxies: 若开启，代理地址

    :return:

    """

    try:

        pattern = re.compile(r'//(.*?)/')

        host_url = pattern.findall(url)[0]

        headers = {

            "Host": host_url,

            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64; rv:60.0) Gecko/20100101 Firefox/60.0",

            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",

            "Accept-Language": "zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2",

            "Accept-Encoding": "gzip, deflate",

            "Connection": "keep-alive",

        }

        if open_proxy:   # 判断是否开启代理

            proxies = {"http": "http://" + ip_proxies, }  # 设置代理，例如{"http": "http://103.109.58.242:8080", }

            res = requests.get(url, headers=headers, proxies=proxies, timeout=5)

        else:

            res = requests.get(url, headers=headers, timeout=5)

        res.encoding = res.apparent_encoding  # 自动确定html编码

        print("Html页面获取成功 " + url)

        return res.text     # 只返回页面的源码

    except Exception as e:

        print("Html页面获取失败 " + url)

        print(e)

这个函数设置了三个参数，分别是：待获取页面的链接、是否开启代理、代理地址。最终的返回值为网页的源码文本。

其中headers部分是伪装为浏览器请求头，这个函数还设置了利用代理去获取页面，也就是说你可以利用已有的代理去获取这些页面。

2.2 保存IP

当我们对获取的每个页面进行解析时，都会获取页面上的代理ip，同样的，每个页面获取的代理ip都要保存下来，将代理ip写入txt文件。函数如下：

def save_ip(data, save_path):

"""

将获取的ip信息保存到文件中

:param data: 代理ip数据，数据类型为列表

:param save_path: 代理ip保存路径

:return:

"""

try:

    print("总共获取 " + str(len(data)) + " 条数据")

    with open(save_path, "a") as f:

        for i in range(len(data)):

            f.write(data[i])

        f.close()

        print("文件保存成功")

except Exception as e:

    print("文件保存失败！！！")

    print(e)

2.3 页面分析

由于不同的网站，代理ip在其中的所在位置也都不同，函数如下：

def get_data5u_free_ip(ip_proxies, save_path, open_proxy):

    """

    获取无忧代理的免费ip

    :param ip_proxies: 要使用的代理ip（这里是用代理ip去爬代理ip）

    :param save_path: 代理ip保存路径

    :param open_proxy: 是否开启代理，默认为False

    :return:

    """

    url_list = [

        "http://www.data5u.com/free/index.shtml",

        "http://www.data5u.com/free/gngn/index.shtml",

        "http://www.data5u.com/free/gnpt/index.shtml",

        "http://www.data5u.com/free/gwgn/index.shtml",

        "http://www.data5u.com/free/gwpt/index.shtml"

    ]

    ip_list_sum = []    # 代理ip列表

    for i in range(5):

        res_text = get_html(url_list[i], open_proxy=open_proxy, ip_proxies=ip_proxies)

        # 抓取错误页面，主动报异常

        if res_text.find("错误") != -1:

            raise AttributeError('错误页面')

        # 页面解析

        soup = BeautifulSoup(res_text, "html.parser")

        tags = soup.find_all("ul", class_="l2")

        for tag in tags:

            ip_list = []

            ip_info_format = ""

            sps = tag.find_all("li")

            for sp in sps:

                ip_info = sp.get_text()

                ip_list.append(ip_info)

            for j in range(len(sps)):

                # 格式化IP信息

                if j == len(sps) - 1:

                    ip_info_format += str(ip_list[j]) + "\n"

                else:

                    ip_info_format += str(ip_list[j]) + "___"

            ip_list_sum.append(ip_info_format)

    save_ip(ip_list_sum, save_path)

def get_kuaidaili_free_ip(ip_proxies, save_path, open_proxy):

    """

    获取快代理的免费ip

    :param ip_proxies: 要使用的代理ip（这里是用代理ip去爬代理ip）

    :param save_path: 代理ip保存路径

    :param open_proxy: 是否开启代理，默认为False

    :return:

    """

    ip_list_sum = []    # 代理ip列表

    for i in range(10):  # 获取页数

        res_text = get_html("https://www.kuaidaili.com/ops/proxylist/" + str(i+1) + "/", open_proxy=open_proxy,

                            ip_proxies=ip_proxies)

        # 页面解析

        soup = BeautifulSoup(res_text, "html.parser")

        tags = soup.find_all("div", id="freelist")

        for tag in tags:

            ip_list = []

            sps = tag.find_all("td")

            for sp in sps:

                ip_info = sp.get_text()

                ip_list.append(ip_info)

            for j in range(10):      # 每页100条数据

                ip_info_format = ""

                for k in range(8):   # 每条6个内容

                    if k == 7:

                        ip_info_format += str(ip_list[(j * 8 + k)]) + "\n"

                    else:

                        ip_info_format += str(ip_list[(j * 8 + k)]) + "___"

                ip_list_sum.append(ip_info_format)

    save_ip(ip_list_sum, save_path)

def get_xsdaili_free_ip(ip_proxies, save_path, open_proxy):

    """

    获取小舒代理的免费ip

    :param ip_proxies: 要使用的代理ip（这里是用代理ip去爬代理ip）

    :param save_path: 代理ip保存路径

    :param open_proxy: 是否开启代理，默认为False

    :return:

    """

    url = "http://www.xsdaili.com/"

    url_list = []

    home_page = get_html(url, open_proxy=open_proxy, ip_proxies=ip_proxies)

    # 首页解析

    home_soup = BeautifulSoup(home_page, "html.parser")

    home_tags = home_soup.find_all("div", class_="title")

    for home_tag in home_tags:

        home_url = home_tag.a["href"]

        new_url = "http://www.xsdaili.com" + str(home_url)

        url_list.append(new_url)

    # 页面解析

    ip_list_sum = []

    for i in range(len(url_list)):  # 页面页数

        res_text = get_html(url_list[i], open_proxy=open_proxy, ip_proxies=ip_proxies)

        # 页面解析

        soup = BeautifulSoup(res_text, "html.parser")

        tags = soup.find("div", class_="cont")

        ip_info = tags.get_text()

        ip_info_temp = ip_info.replace("\r\n\t\t\t\t\t\t\t", "")

        ip_list = re.split(r'[:@# \] ]', ip_info_temp)      # 分割字符串

        for j in range(100):    # 每页100条数据

            ip_info_format = ""

            for k in range(6):  # 每条6个内容

                if k == 5:

                    ip_info_format += str(ip_list[(j * 6 + k)]) + "\n"

                else:

                    ip_info_format += str(ip_list[(j * 6 + k)]) + "___"

            ip_list_sum.append(ip_info_format)

    save_ip(ip_list_sum, save_path)

def get_xicidaili_free_ip(ip_proxies, save_path, open_proxy):

    """

    获取西刺代理的免费ip

    :param ip_proxies: 要使用的代理ip（这里是用代理ip去爬代理ip）

    :param save_path: 代理ip保存路径

    :param open_proxy: 是否开启代理，默认为False

    :return:

    """

    ip_list_sum = []

    for i in range(10):  # 获取页数

        res_text = get_html("http://www.xicidaili.com/nn/" + str(i+1), open_proxy=open_proxy, ip_proxies=ip_proxies)

        # 抓取错误页面，主动报异常

        # print(res_text)

        if res_text.find("错误") != -1:         # 错误页面

            raise AttributeError('错误页面')

        elif res_text == "block":              # 空白页面

            raise AttributeError('错误页面')

        # 页面解析

        soup = BeautifulSoup(res_text, "html.parser")

        tags = soup.find_all("tr", class_="")

        for tag in tags:

            ip_list = []

            ip_ths = tag.find_all("td")

            for ip_th in ip_ths:

                ip_info = ip_th.get_text().replace("\n", "")

                if ip_info != "":

                    ip_list.append(ip_info)

            try:

                ip_info_format = ""

                for k in range(7):   # 每条6个内容

                    if k == 6:

                        ip_info_format += str(ip_list[k]) + "\n"

                    else:

                        ip_info_format += str(ip_list[k]) + "___"

                ip_list_sum.append(ip_info_format)

            except Exception as e:

                # print(e)

                pass

    save_ip(ip_list_sum, save_path)

def get_89ip_free_ip(ip_proxies, save_path, open_proxy):

    """

    获取89免费代理的免费ip

    :param ip_proxies: 要使用的代理ip（这里是用代理ip去爬代理ip）

    :param save_path: 代理ip保存路径

    :param open_proxy: 是否开启代理，默认为False

    :return:

    """

    ip_list_sum = []

    for i in range(10):  # 获取页数

        res_text = get_html("http://www.89ip.cn/index_" + str(i+1) + ".html", open_proxy=open_proxy, ip_proxies=ip_proxies)

        # 抓取错误页面，主动报异常

        if res_text.find("错误") != -1:     # 错误页面

            raise AttributeError('错误页面')

        # 页面解析

        soup = BeautifulSoup(res_text, "html.parser")

        tags = soup.find_all("tbody")

        for tag in tags:

            ip_ths = tag.find_all("tr")

            for ip_th in ip_ths:

                ip_tds = ip_th.find_all("td")

                ip_list = []

                for ip_td in ip_tds:

                    ip_info = re.split(r'[\t\n ]', ip_td.get_text())  # 分割字符串

                    for j in range(len(ip_info)):

                        if ip_info[j] != "":

                            ip_list.append(ip_info[j])

                ip_info_format = ""

                for k in range(len(ip_list)):  # 每条6个内容

                    if k == len(ip_list) - 1:

                        ip_info_format += str(ip_list[k]) + "\n"

                    else:

                        ip_info_format += str(ip_list[k]) + "___"

                ip_list_sum.append(ip_info_format)

    save_ip(ip_list_sum, save_path)

3，验证代理ip（check_ip.py）

3.1 文件查重

写一个简单的函数用于查重，查重思路就是获取txt文件中的每一行元素，组成列表。对列表使用自带的set函数，注意：这种方法会改变列表原来的顺序。函数如下：

from bs4 import BeautifulSoup

import requests

import time

def check_repeat(path):

    """

    检查文件中每一行的内容是否重复，删除重复内容

    :param path: 文件路径

    :return:

    """

    try:

        # 读取文件

        data_list = []

        with open(path, "r") as fr:

            lines = fr.readlines()

            fr.close()

        for line in lines:

            data_list.append(line)

        new_data_list = list(set(data_list))    # 查重

        file_name = path.split("/")

        print(file_name[-1] + "文件共有 " + str(len(data_list)) + " 条数据")

        print("经过查重,现在共有 " + str(len(new_data_list)) + " 条数据")

        # 保存文件

        with open(path, "w") as f:

            for i in range(len(new_data_list)):

                f.write(new_data_list[i])

            f.close()

            print(file_name[-1] + "文件查重成功")

    except Exception as e:

        print("文件查重失败！！！")

        print(e)

3.2 代理IP格式化

网页获取的ip的格式基本为以下格式：

5.135.66.232___8554___透明___http___法国___XXXX___XX___7.869 秒___4秒前

220.230.120.101___8286___高匿___http___韩国___XXXX___XX___4.14 秒___11秒前

从页面解析出来的代理ip的信息中间都是使用“—”进行隔开，为了方便直接使用，在此需要将上述的的格式转换为以下格式：

5.135.66.232:8554

220.230.120.101:8286

函数如下：

def ip_format(read_path, save_path):

"""

将文件中的代理ip进行格式化转换，并进行查重

:param read_path: 读取待转换的代理ip的文件路径

:param save_path: 转换完成的代理ip的保存路径

:return:

"""

data_list = []

with open(read_path, "r") as fr:

    lines = fr.readlines()

    fr.close()

for line in lines:

    new_line = line.split("___")

    ip_format_line = new_line[0].replace(" ", "") + ":" + new_line[1] + "\n"

    data_list.append(ip_format_line)

with open(save_path, "a") as fs:

    for i in range(len(data_list)):

        fs.write(data_list[i])

    fs.close()

    print("文件保存成功")

    fs.close()

3.3 代理IP验证

由于免费的代理ip很多都是无法使用，或是不稳定，或是时效短。所以验证代理ip是否可用，就非常有必要。主要验证原理：使用代理ip去访问网页，判断是否能够正常访问。在此我选择的网站是“站长之家”，这个网站可用直接返回你当前使用的ip以及ip所在地。这里需要注意的是访问前可以设定连接超时的时间如果访问时间超过一定时间，就直接跳过这个代理ip。建议是设定在2秒内，具体的可以看以下函数：

def ip_test(ip_proxies):

"""

验证单个代理ip是否可用

:param ip_proxies: 待验证ip，例如：101.96.10.36:88

:return:

"""

url = "http://ip.chinaz.com/"

headers = {

    "Host": "ip.chinaz.com",

    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64; rv:60.0) Gecko/20100101 Firefox/60.0",

    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",

    "Accept-Language": "zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2",

    "Accept-Encoding": "gzip, deflate",

    "Referer": "https://blog.****.net/Winterto1990/article/details/51220307",

    "Connection": "keep-alive",

    "Upgrade-Insecure-Requests": "",

    "Cache-Control": "max-age=0",

}

proxies = {"http": "http://" + ip_proxies, }   # 设置代理

res = requests.get(url, headers=headers, proxies=proxies, timeout=1)    # timeout为设定的相应时长，建议在2秒内

# 解析网页

soup = BeautifulSoup(res.text, "html.parser")

info_list = soup.find_all("p", {"class": "getlist pl10"})

for info in info_list:

    is_local = info.get_text()

    print(info.get_text())

return is_local.find("XXX.XXX.XXX.XXX")  # 判断是否为本地的地址

3.4. 批量验证代理ip

最后便是批量的对代理ip进行验证，实际上这就是调用3.3.中验证代理ip中的程序。具体程序如下：

def ip_batch_inspection(read_path, save_path):

"""

 验证多个代理ip是否可用

:param read_path: 代理ip文件路径

:param save_path: 验证可用的代理ip保存路径

:return:

"""

with open(read_path, "r") as fr:

    lines = fr.readlines()

    fr.close()

    count = 0

    file_name = read_path.split("/")

    print(file_name[-1] + "文件共有 " + str(len(lines)) + " 条数据")

    for line in lines:

        count += 1

        ip_proxies = line.replace("\n", "")

        try:

            is_local = ip_test(ip_proxies)  # 如果是本地ip，返回值为大于0数值

            if is_local < 0:

                with open(save_path, "a") as fs:

                    fs.write(ip_proxies + "\n")

        except Exception as e:

            pass

            # print("ip不可用")

        print("验证中......%.2f%%" % (count/len(lines)*100))

    print("验证完毕")

4. 使用的例程（get_proxy.py）

将上述的函数都整合起来，就能够实现批量获取免费的代理ip。

# @File    : get_proxy

# @Description: 下载+验证，获取可用ipfrom check_ip import *

from parse_html import *

def main():

    today = time.strftime("%Y_%m_%d")     # 当前日期

    ip_pools_path = "ip_proxy\\" + today + "_ip_pools.txt"                 # 原始ip保存路径

    ip_format_pools_path = "ip_proxy\\" + today + "_ip_format_pools.txt"   # 格式化后ip保存路径

    ip_use_path = "ip_proxy\\" + today + "_ip_use.txt"                     # 可用ip保存路径

    open_proxy = True  # 是否要开启代理模式

    if not open_proxy:

        # 不开启代理模式，直接获取代理ip

        get_data5u_free_ip(None, ip_pools_path)

        get_kuaidaili_free_ip(None, ip_pools_path)

        get_xsdaili_free_ip(None, ip_pools_path)

        get_xicidaili_free_ip(None, ip_pools_path)

        get_89ip_free_ip(None, ip_pools_path)

    else:

        # 开启代理模式，获取代理ip

        available_ip_path = "ip_proxy\\ip_use.txt"  # 目前可用的代理ip的保存路径

        ip_use_list = []

        with open(available_ip_path, "r") as fr:

            ip_use_lines = fr.readlines()

            for ip_use_line in ip_use_lines:

                ip_use_line_new = ip_use_line.replace("\n", "")

                ip_use_list.append(ip_use_line_new)

        for i in range(len(ip_use_list)):

            # 获取ip建立IP池

            try:

                print("正在使用第" + str(i) + "条代理ip")

                get_data5u_free_ip(ip_use_list[i], ip_pools_path, open_proxy)

                break

            except:

                pass

        for i in range(len(ip_use_list)):

            # 获取ip建立IP池

            try:

                print("正在使用第" + str(i) + "条代理ip")

                get_kuaidaili_free_ip(ip_use_list[i], ip_pools_path, open_proxy)

                break

            except:

                pass

        for i in range(len(ip_use_list)):

            # 获取ip建立IP池

            try:

                print("正在使用第" + str(i) + "条代理ip")

                get_xsdaili_free_ip(ip_use_list[i], ip_pools_path, open_proxy)

                break

            except:

                pass

        for i in range(len(ip_use_list)):

            # 获取ip建立IP池

            try:

                print("正在使用第" + str(i) + "条代理ip")

                get_xicidaili_free_ip(ip_use_list[i], ip_pools_path, open_proxy)

                break

            except:

                pass

        for i in range(len(ip_use_list)):

            # 获取ip建立IP池

            try:

                print("正在使用第" + str(i) + "条代理ip")

                get_89ip_free_ip(ip_use_list[i], ip_pools_path, open_proxy)

                break

            except:

                pass

    # 筛选ip进行查重

    ip_format(ip_pools_path, ip_format_pools_path)

    check_repeat(ip_format_pools_path)

    # 验证ip可用性

    ip_batch_inspection(ip_format_pools_path, ip_use_path)

if __name__ == '__main__':

    main()

相关文章