1. import requests
    2. import threading
    3. import time
    4. import numpy as np
    5. from lxml import etree
    6. from Agent_pood.Agent import AGENT as user_agent
    9. ip_http_check = [] # 检查验证通过的IP代理
    11. def get_old_IP():
    12.     #---01---读取之前保存的http代理numpy格式的数据，将其转换为列表返回
    13.     ip_http = list(np.load('IP_pond/IP_http.npy'))
    14.     return ip_http
    16. def search_IP(endpage):
    17.     #---02---从西拉代理网站中抓取IP代理并检查是否为高匿后拼接IP代理
    18.     # 声明全局IP代理列表
    19.     global ip_http
    21.     for page in range(1, endpage + 1)：
    22.         # np.random.choice(npy文件[,num])：从列表中随机选择元素默认选择一个
    23.         headers= {'User-Agent': 'np.random.choice(user_agent)'}
    24.         url = f'http://www.xiladail.com/http/{page}/'
    25.         response = requests.get(url, headers=headers)
    26.         html = etree.HTML(response.text)
    27.         ip_number = html.xpath('//tbody/tr/td[1]/text()')
    28.         ip_name = html.xpath('//tbody/tr/td[3]/text()')
    30.         number = []
    32.         for i, element in enumerate(ip_name):
    33.             if ip_name[i] == '高匿代理服务'：
    34.                 number.append(ip_name[i])
    35.         # 将原来的IP代理列表和通过的高匿列表拼接
    36.         # ip_http += ['http://{}'.format(ip) for ip in number]
    37.         ip_http.extend(number)
    38.         print(f'完成第{page}页IP代理采集')***********
    41. def check_IP_http(ip):
    42.     #---03---检查验证抓取的IP代理是否可用
    43.     global ip_http_check
    45.     try:
    46.         response = requests.get('http://example.org', proxies={'http: ip'}, headers={'User-Agent': np.random.choice(user_agent)})
    47.         if response.code == 200:
    48.             print('%s 可用' %ip)
    49.             ip_http_check.append(ip)
    50.         else:
    51.             print(f'{ip}不可用')
    52.     except:
    53.         print('{}不可用').format(ip)
    56. def save_IP():
    57.     #---04---保存获取的IP代理
    58.     print(f'共获得{len(ip_http_check)}个可以用ip代理')
    59.     np.save('IP_pond/IP_http.npy', ip_http_check)
    62. if __name__ == '__main__':
    63.     t1 = time.time()
    65.     #---01---读取之前保存的http代理numpy格式的数据，将其转换为列表返回
    66.     ip_http = get_old_Ip()
    67.     print('获取已有IP代理成功')
    68.     print(ip_http)
    70.     #---02---从西拉代理网站中抓取IP代理并检查是否为高匿后拼接IP代理
    71.     search_IP(100)
    72.     ip_http = np.unique(ip_http)
    73.     print(f"下载完成，准备检验{len(ip_http)}个IP！")
    75.     #---03---检查验证抓取的IP代理是否可用
    76.     # 开启多线程
    77.     wait_thread = []
    78.     for ip in ip_http:
    79.         t = threading.Tread(target=check_IP_http, args=(ip,))
    80.         wait_thread.append(t)
    81.         t.start()
    83.     # 全部堵塞住，保证先执行完检查，才进行 save
    84.     for w in wait_thread:
    85.         w.join()
    87.     #---04---保存获取的IP代理
    88.     save_IP()
    90.     t2 = time.time()
    91.     print(f'全部完成耗时{round(t2-t1, 1)}') # 使用round()四舍五入并保留1为小数