前言#
暴力破解や反クローリングに関わらず、代理プールは不可欠な条件です。
代理プールの取得については、前回の記事「マルチスレッドで西刺高匿名代理を取得し、有効性を検証する」で紹介しました。
この文章で取得した検証済みの代理を使用して、アクセス数を増やしてみましょう。
醜いコード実装#
ほとんどのコード実装は、上記の文章をそのままコピーしたものです。
唯一の違いは、fake_useragent というライブラリを学んだことです。これを使ってランダムな User-Agent を生成できます。簡単なサンプルコードは以下の通りです：
from fake_useragent import UserAgent
ua = UserAgent() # User-Agentを生成するために使用
headers = {"User-Agent": ua.random} # ランダムなUser-Agentを取得
print(headers)
ブラウザの User-Agent を指定する必要がある場合は、この記事を参考にしてください：https://blog.csdn.net/qq_29186489/article/details/78496747
全コード：
#!/usr/bin/python
# -*- coding: utf-8 -*-
'''
@author: soapffz
@function: 代理IPプールを使用してアクセス数を増やす
@time: 2019-01-23
'''
import os
import requests
from fake_useragent import UserAgent
from multiprocessing import Pool
import timeit
url = "https://soapffz.com/" # アクセスするウェブページ
ua = UserAgent() # User-Agentを生成するために使用
path = os.path.join(os.path.expanduser("~")+"\\")
http_proxy_path = os.path.join(
path+"Desktop"+"\\"+"http_proxy.txt") # 自分の代理IPの位置
https_proxy_path = os.path.join(
path+"Desktop"+"\\"+"https_proxy.txt") # 自分の位置に応じて変更
http_proxy = [] # http代理を保存
https_proxy = [] # https代理を保存
def brush_visits(proxies):
proxy_type = proxies.split(":")[0]
if proxy_type == 'http':
proxy = {'http': proxies}
else:
proxy = {'https': proxies}
headers = {"User-Agent": ua.random} # ランダムなUser-Agentを取得
req = requests.get(url, headers=headers, proxies=proxy, timeout=10)
if req.status_code == 200:
print("この代理は成功しました：{}".format(proxies))
if __name__ == "__main__":
start_time = timeit.default_timer()
if not os.path.exists(http_proxy_path):
print("代理が準備できていないのにアクセス数を増やしたいですか？出て行け！")
os._exit(0)
with open(http_proxy_path, 'r') as f:
http_proxy = f.read().splitlines()
with open(https_proxy_path, 'r') as f:
https_proxy = f.read().splitlines()
pool1 = Pool()
pool2 = Pool()
for proxies in http_proxy:
pool1.apply_async(brush_visits, args=(proxies,))
for proxies in https_proxy:
pool2.apply_async(brush_visits, args=(proxies,))
pool1.close()
pool2.close()
pool1.join()
pool2.join()
end_time = timeit.default_timer()
print("代理プールは使い果たされました。合計で{}秒かかりました".format(end_time-start_time))
効果の表示は以下の通りです：
更新#
[2019-08-27 更新] 前に書いたのはファイルの読み込みに基づいていましたが、今考えると少し馬鹿げています。最近この需要があったので、更新します。
前準備として、Mongodbデータベースをインストールしておけば大丈夫です。私が以前書いた記事「Win10 に mongodb をインストールする」を参考にしてください。
その後、このコードを直接実行し、取得したいページを変更すれば OK です。全コードは以下の通りです：
#!/usr/bin/python
# -*- coding: utf-8 -*-
'''
@author: soapffz
@function: 代理IPプールを使用してウェブサイトのアクセス数を一括で増やす（ローカルmongodbデータベースの代理コレクションを読み取る）
@time: 2019-08-27
'''
from os import popen
from threading import Thread # マルチスレッド
from pymongo import MongoClient
from requests import get
from lxml import etree # サイトを解析
from re import split # re解析ライブラリ
from telnetlib import Telnet # telnet接続で代理の有効性をテスト
from fake_useragent import UserAgent
from multiprocessing import Pool
def brush_visits(url, proxy):
# 単一の代理で単一のページにアクセス
try:
ua = UserAgent() # User-Agentを生成するために使用
headers = {"User-Agent": ua.random} # ランダムなUser-Agentを取得
req = get(url, headers=headers, proxies=proxy, timeout=0.3)
if req.status_code == 200:
print("この代理{}は一回のアクセスを貢献しました".format(proxy))
except Exception as e:
pass
class batch_brush_visits(object):
def __init__(self, url):
self.url = url
self.ua = UserAgent() # User-Agentを生成するために使用
self.run_time = 0 # 実行回数
self.conn_db()
self.get_proxies_l()
def conn_db(self):
# データベースに接続
try:
# mongodbに接続し、接続オブジェクトを取得
self.client = MongoClient('mongodb://localhost:27017/').proxies
self.db = self.client.xici # proxiesデータベースを指定。このクラスはこのデータベースを使用し、データがない場合は自動的に作成されません。
self.refresh_proxies()
except Exception as e:
print("データベース接続エラー")
exit(0)
def refresh_proxies(self):
if "xici" in self.client.list_collection_names():
# 初期化のたびに既存のコレクションを削除
self.db.drop()
t = Thread(target=self.xici_nn_proxy_start)
t.start()
t.join()
def get_proxies_l(self):
# 対応するコレクションからIPとポート情報を読み取る
collection_dict_l = self.db.find(
{}, {"_id": 0, "ip": 1, "port": 1, "type": 1})
self.proxies_l = []
for item in collection_dict_l:
if item["type"] == 'http':
self.proxies_l.append(
{"http": "{}:{}".format(item['ip'], item['port'])})
else:
self.proxies_l.append(
{"https": "{}:{}".format(item['ip'], item['port'])})
if len(self.proxies_l) == 0:
print("代理を取得できませんでした。何もできません...")
exit(0)
else:
self.brush_visits_start()
def xici_nn_proxy_start(self):
xici_t_cw_l = [] # クローリングスレッドリスト
self.xici_crawled_proxies_l = [] # クローリング前に毎回クリア
for i in range(1, 11): # 10ページの代理をクローリング
t = Thread(target=self.xici_nn_proxy, args=(i,))
xici_t_cw_l.append(t) # スレッドリストにスレッドを追加
t.start()
for t in xici_t_cw_l: # すべてのスレッドが完了するのを待ってメインスレッドを終了
t.join()
# データベースに挿入
self.db_insert(self.xici_crawled_proxies_l)
def xici_nn_proxy(self, page):
# 西刺代理クローリング関数
url = "https://www.xicidaili.com/nn/{}".format(page)
# ここにUser-Agentを加えないと503のステータスコードが返されます
req = get(url, headers={"User-Agent": self.ua.random})
if req.status_code == 200:
# 他のステータスコードはIPが封鎖されていることを示します
content = req.content.decode("utf-8")
tree = etree.HTML(content)
# xpathを使用して全IPリストを取得
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(split(r":|%", td_nodes[6].xpath(
'./div/div/@style')[0])[1]) # 速度の値を取得
conn_time = int(split(r":|%", td_nodes[7].xpath(
'./div/div/@style')[0])[1]) # 接続時間の値を取得
if(speed <= 85 | conn_time <= 85): # 速度と接続時間が理想的でない場合、この代理をスキップ
continue
ip = td_nodes[1].text
port = td_nodes[2].text
ip_type = td_nodes[5].text.lower()
self.get_usable_proxy(ip, port, ip_type, "xici")
def get_usable_proxy(self, ip, port, ip_type, proxy_name):
proxies_l = {"ip": ip, "port": port, "type": ip_type}
if proxies_l not in self.xici_crawled_proxies_l:
try:
# telnetで接続して、接続できれば代理が有効であることを示します
Telnet(ip, port, timeout=0.3)
except:
pass
else:
self.xici_crawled_proxies_l.append(proxies_l)
def db_insert(self, proxies_list):
if proxies_list:
# 受け取ったリストが空の場合は終了
self.db.insert_many(proxies_list) # 辞書のリストを挿入
print(
"データの挿入が完了しました！\n今回、{}個の代理をxiciコレクションに挿入しました！".format(len(proxies_list)))
else:
print(
"クローリングした代理リストが空です！\nこのメッセージをすぐに見た場合\nおそらくIPが封鎖されています\nVPNを使用してください！\nさもなければ最新の代理はすでにデータベースにあります\n再度クローリングする必要はありません！")
def brush_visits_start(self):
# 取得した代理を使用してページをクローリング
try:
self.run_time += 1
pool = Pool()
for proxy in self.proxies_l:
pool.apply_async(brush_visits(self.url, proxy))
pool.close()
pool.join()
if self.run_time == 100:
print("100回クローリングしました。再度代理を取得します。")
self.refresh_proxies()
print("一巡クローリングが完了しました。")
self.brush_visits_start()
except Exception as e:
print("プログラム実行中にエラーが発生しました:{}".format(e))
exit(0)
if __name__ == "__main__":
url = "https://soapffz.com/3.html" # ここにクローリングしたいURLを入力
batch_brush_visits(url)
効果は以下の通りです：
また、この記事のコードは 1 つのurlをクローリングするように設計されていますが、複数のurlを同時にクローリングする必要がある場合は、デカルト積を使用する必要があります。使用例は以下の通りです：
from itertools import product
urls_l = ["https://soapffz.com/279.html",
"https://soapffz.com/timeline.html", "https://soapffz.com/372.html"]
proxies_l = ["120.120.120.121", "1.1.1.1", "231.220.15.2"]
paramlist = list(product(urls_l, proxies_l))
for i in range(len(paramlist)):
print(paramlist[i])
効果は以下の通りです：
したがって、次のように変更する必要があります：
- このライブラリをインポートします：
from itertools import product
main関数内の URL をリストに変更します：
if __name__ == "__main__":
urls_l = ["https://soapffz.com/3.html"] # ここにクローリングしたいURLのリストを入力
batch_brush_visits(urls_l)
- 目標をデカルト積に変更します。
batch_brush_visitsクラスのbrush_visits_start部分のコードを以下のように変更します：
pool = Pool()
for proxy in self.proxies_l:
pool.apply_async(brush_visits(self.url, proxy))
pool.close()
pool.join()
を
paramlist = list(product(self.urls_l, self.proxies_l))
pool = Pool()
for i in range(len(paramlist)):
pool.apply_async(brush_visits(paramlist[i]))
pool.close()
pool.join()
- アクセス関数を変更します。
brush_visits関数を以下のように変更します：
def brush_visits(url, proxy):
# 単一の代理で単一のページにアクセス
try:
ua = UserAgent() # User-Agentを生成するために使用
headers = {"User-Agent": ua.random} # ランダムなUser-Agentを取得
req = get(url, headers=headers, proxies=proxy, timeout=0.3)
if req.status_code == 200:
print("この代理:{}は一回のアクセスを貢献しました".format(proxy))
except Exception as e:
pass
を：
def brush_visits(paramlist):
# 単一の代理で単一のページにアクセス
try:
url = paramlist[0]
proxy = paramlist[1]
ua = UserAgent() # User-Agentを生成するために使用
headers = {"User-Agent": ua.random} # ランダムなUser-Agentを取得
req = get(url, headers=headers, proxies=proxy, timeout=0.3)
if req.status_code == 200:
print("この代理:{}は一回のアクセスを貢献しました".format(proxy))
except Exception as e:
pass
これで、各代理が各urlに一度アクセスする効果を実現できます。
この記事は以上です。