前言#
なぜプロキシを使用するのか？ ブルートフォース攻撃やクローリングを行う際、IP がブロックされることがよくあります。このため、信頼できるプロキシプールを自作することが非常に重要です。
プロキシの分類 プロキシの匿名性に基づいて、透明プロキシ、匿名プロキシ、高度匿名プロキシに分類できます。
- 透明プロキシを使用すると、相手のサーバーはプロキシを使用していることを知っており、あなたの実際の IP も知っています。
- 匿名プロキシを使用すると、相手のサーバーはプロキシを使用していることを知っていますが、あなたの実際の IP は知りません。
- 高度匿名プロキシを使用すると、相手のサーバーはプロキシを使用していることも、あなたの実際の IP も知りません。
この記事では、国内で比較的有名な西刺プロキシ（もう一つの有名なプロキシである快プロキシもあり、クローリング手順はほぼ同じです）の高匿名プロキシを取得し、有効性を検証します。
思路#
情報の取得#
まず、requests ライブラリを使用してウェブページをクローリングし、必要な情報がどのタグにあるかを確認します。
各プロキシのすべての情報は tr タグに含まれており、各情報は td タグに格納されています:
したがって、lxml ライブラリを使用して解析し、xpath を使用してタグを探すことができます。簡単なサンプルコードは以下の通りです：
import requests
from lxml import etree
url = 'https://www.xicidaili.com/nn/1' # クローリングするURL
req = requests.get(url) # requestsでリクエスト
content = req.content.decode("utf-8") # コンテンツをデコード
tree = etree.HTML(content) # HTMLファイルを解析
tr_nodes = tree.xpath('.//table[@id="ip_list"]/tr')[1:] # xpathを使用して必要なタグを解析
指定したタグの内容を取得するための xpath の使用方法については、この記事を参照してください：https://www.cnblogs.com/lei0213/p/7506130.html
各ポートの IP、ポート、タイプ、速度、接続時間などの情報を取得した後、フィルタリングを行います：
- http、https タイプのプロキシ IP を分類
- 速度と接続時間が理想的でない IP を除外
有効性の検証#
次に、プロキシ IP を使用して検証を行います。
検証方法としては、インターネット上ではほとんどがプロキシ IP をリクエストヘッダーに追加して特定のウェブサイト（例えば百度）にリクエストを送る方法です。
もし返されたステータスコードが 200 であれば、そのプロキシ IP は有効です。
しかし、これはネットワークに依存しすぎており、効率も良くありません。ある記事では、各 IP に telnet で接続してみて、接続できれば有効であると述べています。
この部分の主要なパラメータはタイムアウト時間、つまり待機時間で、10 秒に設定することをお勧めします。
telnet を使用してプロキシ IP の有効性を検証する簡単な例は以下の通りです：
import telnetlib
def verifyProxy(ip, port):
try:
telnetlib.Telnet(ip, port, timeout=5) # telnetで接続し、接続できればプロキシが使用可能
except:
pass
else:
print("このプロキシは有効であることが確認されました：{}".format(proxies))
if __name__ == "__main__":
verifyProxy("113.13.177.80", "9999")
マルチスレッドによる最適化#
情報の取得と有効性の検証が可能になった後、効率が低すぎることがわかりました。主に時間がかかる部分は以下の通りです：
- 複数ページのリクエストに時間がかかる
- プロキシの検証にはタイムアウト待機時間が必要で、後続の処理が待機する必要がある
したがって、multiprocessing ライブラリの Pool スレッドプールを使用し、apply_async () メソッドを利用します。
このスレッドプールは apply () メソッドと組み合わせて使用できますが、apply () はブロッキングです。
まず、主プロセスが実行を開始し、子プロセスに遭遇すると、オペレーティングシステムは子プロセスに切り替え、子プロセスの実行が終了するまで待機し、次に別の子プロセスに切り替え、すべての子プロセスが完了するまで待機します。その後、主プロセスに戻り、残りの部分を実行します。これはシングルプロセスの直列実行とほとんど変わりません。
apply_async は非同期でブロッキングではありません。
現在のプロセスの実行が完了するのを待つ必要がなく、システムのスケジューリングに応じてプロセスの切り替えを行います。まず、主プロセスが実行を開始し、子プロセスに遭遇した後、主プロセスはそのまま実行を続け、オペレーティングシステムがプロセスの切り替えを行うときに子プロセスに実行を渡します。子プロセスの実行が完了するのを待たずに、主プロセスがすでに完了し、プログラムを終了することができます。
プロキシを取得し、有効性を検証するだけでよく、順序は必要ありません。小説をクローリングするように、章の順序に従ってファイルに保存する必要はありません。
したがって、ここでは apply_async を使用するのが最も適切です。
multiprocessing ライブラリの Pool スレッドプールと apply_async () メソッドの簡単なサンプルコードは以下の通りです：
from multiprocessing import Pool
def example(proxies):
if type(proxies) == 'http':
with open("1.txt", 'a') as f:
f.write(proxies+"\n")
else:
with open("2.txt", 'a') as f:
f.write(proxies+"\n")
if __name__ == "__main__":
proxy = []
with open("tmp.txt", 'r') as f:
proxy = f.read().splitlines()
pool = Pool() # 新しいスレッドプールを作成
for line in proxy:
pool.apply_async(target=example, args=(line,)) # targetパラメータ名は省略可能、最後のパラメータの後にカンマを追加することをお勧めします
pool.close() # プロセスプールを作成した後は必ず閉じる必要があります
pool.join() # ブロッキングキューに追加し、上記のコードが完了した後に次のコードを実行します
具体的なコード実装#
重要な部分にはコメントが付けられています：
#!/usr/bin/python
# -_- coding: utf-8 -_-
'''
@author: soapffz
@function: マルチスレッドで西刺の高匿名プロキシを取得し、有効性を検証する
@time: 2019-01-21
'''
import requests
from lxml import etree
import re
import telnetlib
import threading
from multiprocessing import Pool
import os
import timeit
path = os.path.join(os.path.expanduser("~")+"\\") # クローリングした未検証のプロキシを現在のユーザーのルートディレクトリに保存
http_tmp_path = os.path.join(path+"http_tmp.txt")
https_tmp_path = os.path.join(path+"https_tmp.txt")
# 検証済みのプロキシは現在のユーザーのデスクトップに保存
http_proxy_path = os.path.join(path+"Desktop"+"\\"+"http_proxy.txt")
https_proxy_path = os.path.join(path+"Desktop"+"\\"+"https_proxy.txt")
http_proxy = []
https_proxy = []
def get_nn_proxy(page_num):
url = "https://www.xicidaili.com/nn/{}".format(page_num)
headers = {
"User-Agent": 'Mozilla/5.0 (Windows NT 6.2; WOW64; rv:21.0) Gecko/20100101 Firefox/21.0'} # ランダムなUser-Agentヘッダーをテストしましたが、効果がないことがわかりました。西刺はIPをブロックします。
req = requests.get(url, headers=headers)
print("第{}ページの内容をクローリング中...".format(page_num))
content = req.content.decode("utf-8")
tree = etree.HTML(content) # xpathを使用して全体のip_listを取得
tr_nodes = tree.xpath('.//table[@id="ip_list"]/tr')[1:]
for tr_node in tr_nodes:
td_nodes = tr_node.xpath('./td') # xpathを使用して単一のIPのタグを取得
speed = int(
re.split(r":|%", td_nodes[6].xpath('./div/div/@style')[0])[1]) # 速度の値を取得
conn_time = int(
re.split(r":|%", td_nodes[7].xpath('./div/div/@style')[0])[1]) # 接続時間の値を取得
if(speed <= 95 | conn_time <= 95): # 速度と接続時間が理想的でない場合、このプロキシをスキップ
continue
ip = td_nodes[1].text
port = td_nodes[2].text
proxy_type = td_nodes[4].text
ip_type = td_nodes[5].text.lower()
proxy = "{}:{}".format(ip, port)
if ip_type == 'http':
with open(http_tmp_path, 'a') as f:
f.write("http://{}".format(proxy)+"\n") # クローリングしたhttpプロキシを一時ファイルに保存
else:
with open(https_tmp_path, 'a') as f: # クローリングしたhttpsプロキシを一時ファイルに保存
f.write("https://{}".format(proxy)+"\n")
def verifyProxy(proxies):
ree = re.split(r'//|:', proxies)
ip_type = ree[0]
ip = ree[2]
port = ree[3]
try:
telnetlib.Telnet(ip, port, timeout=5) # telnetで接続し、接続できればプロキシが使用可能
except:
pass
else:
print("このプロキシは有効であることが確認されました：{}".format(proxies))
if ip_type == 'http':
with open(http_proxy_path, 'a') as f:
f.write(proxies+"\n")
else:
with open(https_proxy_path, 'a') as f:
f.write(proxies+"\n")
def clear_cache(path):
if os.path.exists(path):
os.remove(path)
if __name__ == "__main__":
start_time = timeit.default_timer()
clear_cache(http_tmp_path)
clear_cache(https_tmp_path)
clear_cache(http_proxy_path)
clear_cache(https_proxy_path)
pool = Pool()
for i in range(1, 11): # ここでは1から10ページの高匿名プロキシをクローリングしています。必要に応じてパラメータを変更してください。
pool.apply_async(get_nn_proxy, args=(i,))
pool.close()
pool.join()
if not os.path.exists(http_tmp_path): # IPがブロックされている場合、取得した各ページのプロキシは空のリストになり、ファイルは生成されません。
print("あなたのIPは西刺によってブロックされています。ipconfig /releaseを使用した後、ipconfig /renewでIPを変更するか、VPNを使用してください！")
os._exit(0)
with open(http_tmp_path, 'r') as f:
http_proxy = f.read().splitlines()
Unhttp_proxy_num = len(http_proxy)
with open(https_tmp_path, 'r') as f:
https_proxy = f.read().splitlines()
Unhttps_proxy_num = len(https_proxy)
pool2 = Pool()
pool3 = Pool()
for proxies in http_proxy:
pool2.apply_async(verifyProxy, args=(proxies,))
for proxies in https_proxy:
pool3.apply_async(verifyProxy, args=(proxies,))
pool2.close()
pool3.close()
pool2.join()
pool3.join()
http_proxy.clear()
https_proxy.clear()
with open(http_proxy_path, 'r') as f:
http_proxy = f.read().splitlines()
http_proxy_num = len(http_proxy)
with open(https_proxy_path, 'r') as f:
https_proxy = f.read().splitlines()
https_proxy_num = len(https_proxy)
print(http_proxy)
print(https_proxy)
print("http_proxyとhttps_proxyの2つのリストに保存され、txtファイルがデスクトップに保存されました。")
end_time = timeit.default_timer()
print("合計{}個のhttpプロキシをクローリングし、有効なものは{}個、{}個のhttpsプロキシをクローリングし、有効なものは{}個、総時間は{}秒です。".format(Unhttp_proxy_num,
http_proxy_num, Unhttps_proxy_num, https_proxy_num, end_time-start_time))
注意：プロキシの有効性により、長期間の保存は推奨されません。「即用即クローリング」をお勧めします。また、短期間に 2、3 回使用すると IP がブロックされるため、IP がブロックされた場合は VPN を使用してください。
効果の展示#
無料の高匿名 IP の有効数は非常に少なく、376 個の http プロキシのうち 53 個が有効で、409 個の https プロキシのうち 72 個が有効です。