はじめに
Seleniumを使ったWebのスクレーピング中のreCAPTCHA(v2)の回避方法を実装した。
音声テストを選択して、GoogleやGladiaなどSaaS AIに文字起こしさせて回避する方法が知られている。
ここでは軽いAIモデルをローカルで動かして高速に処理できたのでまとめてみた。
概略フロー
Whisperモデルは軽量だがロードには多少の時間がかかるので、reCAPTCHAを検出する前にロードしておきたい。Whipserサーバーとして別のプロセスで動かしておくことにした。
スクレーピングのプロセスからSocketで文字起こしの要求をして、Whisperサーバーのプロセスで処理をしてテキストを返すような作りにする。
Whisperによる文字起こし
Openai/whisperのGithubの手順どおり、pipでwhisperをインストール。
ffmpegも必要なので、Ubuntuならaptで、Windowsなら公式サイトからダウンロードしてインストールする。
pip install -U openai-whisper
sudo apt install ffmpeg
コマンドライン版で試す
- ffmpegのインストール先をパスに追加する。
これをしないと、「FileNotFoundError: [WinError 2] The system cannot find the file specified」という不親切なメッセージのエラーが発生する。
PATH=C:\tools\ffmpeg\bin;%PATH% - 短い音声ファイルを用意する。例えばALCで適当なものが見つけられる。
- 以下のコマンドを実行
モデルはtinyからturboまで6つのサイズがあり、medium以下には英語特化版がある。試した限りでは「base.en」くらいで十分そう。
whisper Downloads\xxx.mp3 --model base.en
Pythonを使った場合のコード
def ask_whisper(audio_url):
import whisper, requests, os
# ffmpegのパスを追加
os.environ['path'] = 'C:/tools/ffmpeg-7.1-essentials_build/bin'
# whisperのbase.enモデルをロード
model = whisper.load_model('base.en')
# requestsを使って音声データをダウンロード
audio_data = requests.get(audio_url).content
# 適当な名前でファイルに保存
TEMP_FILE = os.path.join(os.environ['temp'], 'recaptcha_temp.mp3')
with open(TEMP_FILE, mode='wb') as f:
f.write(audio_data)
# 一応保存が終わるのを待って
while not os.path.isfile(TEMP_FILE):
sleep(1)
# whisperに文字起こししてもらって結果を返す
result = model.transcribe(TEMP_FILE)
return result['text']
if __name__ == '__main__':
text = ask_whisper('https://cdn2.alc.co.jp/sa/ejo/mp3/tabieigo/kotsu/scene14.mp3')
print(text)
実行結果
Excuse me, please let me know when this bus gets to the mall. Sure.
Seleniumでスクレーピング
Googleが提供しているReCAPTCHAデモ・サイトをPython-seleniumで開く。
Seleniumをインストール
pip install selenium
Webdriverをダウンロード
ブラウザがChromeの場合はChrome for Testingから、Chromeと同じバージョンのchromedriverをダウンロードする。
解凍して適当なフォルダに保存する。(以下のコードでは C:\tools\driver\chromedriver.exe)
chromedriverのダウンロードについて
最近のバージョンのSeleniumではこのステップは不要で、自動的に適切なchromedriverをダウンロードしてくれる。だだし時間がかかるので開発中は手動でダウンロードしておくのがお勧め。
一方で手動でダウンロードすると、Chromeがアップデートされる度にchromedriverを入れ替える必要があるので、運用中は自動ダウンロードの方がお勧め。
SeleniumでChromeを起動
以下を実行するとChromeが起動するはず。
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service as ChromeService
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
def create_driver():
options = Options() # 今回使わないが、Headlessモードの指定など可能
# service = ChromeService() # chromedriverを自動でダウンロードする場合
service = ChromeService(executable_path='C:/tools/driver/chromedriver.exe')
# options.add_argument('--window-size=1920,1080') # Chromeのウィンドウサイズの変更
return webdriver.Chrome(service=service, options=options)
if __name__ == '__main__':
driver = create_driver()
ReCAPTCHAデモサイトを開く
import time
def wait_site(driver, wtime=2): # 新しいページがロードされるのを待つため関数
WebDriverWait(driver, 15).until(EC.presence_of_all_elements_located)
time.sleep(wtime)
driver.get('https://www.google.com/recaptcha/api2/demo')
WebDriverWait(driver, 5).until(
EC.frame_to_be_available_and_switch_to_it((By.XPATH, '//*[@id="recaptcha-demo"]/div/div/iframe'))
) # チェックボックスのあるiframeがロードされるのを待って移動する
driver.find_element(by=By.XPATH, value='//*[@id="recaptcha-anchor"]/div[1]').click() # チェックボックスのクリック
wait_site(driver)
driver.switch_to.default_content() # 元の位置に戻る
#
# パズルが出た時の処理がここに入る
#
driver.find_element(by=By.ID, value='recaptcha-demo-submit').click() # Submitボタンのクリック
wait_site(driver)
「I'm not a robot」のチェックボックスをクリックしたいが、その要素は下のようにiframeの中にある。iframeを移動しないと要素を見つけられない。
<iframe title="reCAPTCHA" ....></iframe>
:
<div class="recaptcha-checkbox-border" role="presentation"></div> # クリックしたい要素
以下のようにiframeの要素を指定して移動する方法もあるが、上記のコードではiframeがロードされるのを待って、自動的にiframeに移動してくれる。
iframe = driver.find_element(by=By.XPATH, value='//*[@id="recaptcha-demo"]/div/div/iframe')
driver.switch_to.frame(iframe)
iframe内の要素に対する処理が終わったら、switch_to.default_content()で元の位置(iframe外)に戻っておく。
パズルが出た時の処理
チェックボックスだけで回避できれば上記で問題ないが、パズルが出てしまった場合はヘッドホン型ボタンを押して、音声パズルに変更する。
パズルが出たかどうかはヘッドホン型ボタンの要素が見つかるかどうかで判定できる。このボタンも別のiframeにあるので、移動してから要素があるかを確認する。iframeへの移動はパズルが表示されていなくても可能。
WebDriverWait(driver, 5).until(
EC.frame_to_be_available_and_switch_to_it((By.XPATH, '/html/body/div[2]/div[4]/iframe'))
) # ヘッドホン型ボタンのあるiframeへ移動
if button := driver.find_elements(by=By.ID, value='recaptcha-audio-button'): # ボタンがあるかどうかの判定
button[0].click() # buttonには要素のリストが格納されている
wait_site(driver)
audio_url = driver.find_element(by=By.CLASS_NAME, value='rc-audiochallenge-tdownload-link').get_attribute('href')
# これから作るSocketを使った処理
text = whisper_client(audio_url)
# 文字起こしができたら、テキストボックスに回答とエンター・キーを入力
driver.find_element(by=By.ID, value='audio-response').send_keys(text)
driver.find_element(by=By.ID, value='audio-response').send_keys(Keys.ENTER)
wait_site(driver)
driver.switch_to.default_content() # 元のiframeに戻る
要素があるかどうかの確認方法
要素を見つけるためのメソッドには、find_elementとfind_elementsがある。要素があるかどうか分からない場合には、find_elementsの返り値がTrueかFalseかを利用して判定できる。
# もっとスマートな方法があるかもしれません。。
| メソッド | 要素が見つかった場合 | 要素が見つからなかった場合 |
|---|---|---|
| find_element | 最初に見つかった要素を返す | エラーが発生する |
| find_elements | 見つかった全ての要素をリストで返す | 空のリストを返す |
Socketを使って通信
プロセスの間で通信をさせるのに一番簡単そうだったのでSocketを利用した。
IPを指定して異なるマシンの間でも通信ができるが、今回使うのは同じマシンで動かしている2つのプロセスの間の通信となる。こちらを参考にした。
クライアントのプロセス
音声のURLを受け取ったら、socketでサーバーに接続してエンコードしたURLを送信する。
サーバーが文字起こししたテキストを送ってくるのでデコードして返り値とする。
def whisper_client(audio_url):
import socket
HOST = 'localhost' # 同じマシン上のサーバー・プロセスに通信
PORT = 12345 # ポート名は適当、サーバー・プロセスと同じにする
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sc:
sc.connect((HOST, PORT))
sc.sendall(audio_url.encode())
data = sc.recv(1024)
return data.decode()
サーバーのプロセス
サーバー側はクライアントから何度でもリクエストを受けられるように無限ループ内にsocketの受信処理を記述した。
クライアントから音声のURLを受け取ったらデコードして、最初に作ったask_whisper関数へ渡して文字起こしをさせる。
文字起こしの回答をエンコードしてクライアントへ返信する。
def whisper_server():
os.environ['path'] = 'C:/tools/ffmpeg-7.1-essentials_build/bin'
import whisper, requests, socket
model = whisper.load_model('base.en') # モデルのロードは一度で良いのでask_whisper関数の外に移動した
def ask_whisper(audio_url): # この記事の冒頭で作った関数を使う
TEMP_FILE = os.path.join(os.environ['temp'], 'recaptcha_temp.mp3')
audio_data = requests.get(audio_url).content
with open(TEMP_FILE, mode='wb') as f:
f.write(audio_data)
while not os.path.isfile(TEMP_FILE):
sleep(1)
result = model.transcribe(TEMP_FILE)
return result['text']
#
HOST = 'localhost' # ここはループバックのアドレスで良いらしい
PORT = 12345 # クライアント・プロセスと同じポート番号を指定
while True: # クライアントから何度でも要求を受けられるよう無限ループ
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sc:
sc = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sc.bind((HOST, PORT))
sc.listen()
conn, addr = sc.accept()
with conn:
while True:
data = conn.recv(1024)
if not data:
break
audio_url = data.decode()
text = ask_whisper(audio_url)
conn.sendall(text.encode())
最終的なコード
import sys, os, time, pdb, warnings, argparse
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service as ChromeService
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
def whisper_server():
os.environ['path'] = 'C:/tools/ffmpeg-7.1-essentials_build/bin'
import whisper, requests, socket
model = whisper.load_model('base.en')
def ask_whisper(audio_url):
TEMP_FILE = os.path.join(os.environ['temp'], 'recaptcha_temp.mp3')
audio_data = requests.get(audio_url).content
with open(TEMP_FILE, mode='wb') as f:
f.write(audio_data)
while not os.path.isfile(TEMP_FILE):
sleep(1)
result = model.transcribe(TEMP_FILE)
return result['text']
#
HOST = 'localhost'
PORT = 12345
while True:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sc:
sc = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sc.bind((HOST, PORT))
sc.listen()
conn, addr = sc.accept()
with conn:
while True:
data = conn.recv(1024)
if not data:
break
audio_url = data.decode()
if audio_url == 'QUIT':
return
text = ask_whisper(audio_url)
print(f'\nrequested url: {audio_url}')
print(f'\nrecognized text: {text}')
conn.sendall(text.encode())
def whisper_client(audio_url):
import socket
HOST = 'localhost'
PORT = 12345
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sc:
sc.connect((HOST, PORT))
sc.sendall(audio_url.encode())
data = sc.recv(1024)
return data.decode()
def create_driver(headless=False):
options = Options()
if (headless == True):
options.add_argument('-headless')
# service = ChromeService()
service = ChromeService(executable_path='C:/tools/driver/chromedriver.exe')
# options.add_argument('--window-size=1920,1080')
return webdriver.Chrome(service=service, options=options)
def wait_site(driver, wtime=2):
WebDriverWait(driver, 15).until(EC.presence_of_all_elements_located)
time.sleep(wtime)
def main():
driver = create_driver(headless=False)
driver.get('https://www.google.com/recaptcha/api2/demo')
for _ in range(50):
WebDriverWait(driver, 5).until(
EC.frame_to_be_available_and_switch_to_it((By.XPATH, '//*[@id="recaptcha-demo"]/div/div/iframe'))
)
driver.find_element(by=By.XPATH, value='//*[@id="recaptcha-anchor"]/div[1]').click()
wait_site(driver)
driver.switch_to.default_content()
WebDriverWait(driver, 5).until(EC.frame_to_be_available_and_switch_to_it((By.XPATH, '/html/body/div[2]/div[4]/iframe')))
if button := driver.find_elements(by=By.ID, value='recaptcha-audio-button'):
button[0].click()
wait_site(driver)
audio_url = driver.find_element(by=By.CLASS_NAME, value='rc-audiochallenge-tdownload-link').get_attribute('href')
text = whisper_client(audio_url)
driver.find_element(by=By.ID, value='audio-response').send_keys(text)
time.sleep(1)
driver.find_element(by=By.ID, value='audio-response').send_keys(Keys.ENTER)
wait_site(driver)
driver.switch_to.default_content()
driver.find_element(by=By.ID, value='recaptcha-demo-submit').click()
wait_site(driver)
keyin = input('\nContinue(c) or quit(q)? ')
if keyin in ('q', 'Q'):
driver.quit()
whisper_client('QUIT')
return
driver.refresh()
if __name__ == '__main__':
import argparse
parser = argparse.ArgumentParser(description='Recaptcha solver and socket tests')
parser.add_argument('--server', action='store_true')
args = parser.parse_args()
if args.server:
whisper_server()
else:
main()
exit()
サーバー・プロセスの起動と実行結果
python .\test_socket_recaptcha_solver.py --server
requested url: https://www.google.com/recaptcha/api2/payload/audio.mp3?p=06AFcWeA4vBdmYG2fSgxPe0AU4Y2vZZYnnJMf1YtffOqP621D-4BmUuSLrjaohTIBF8Ll2_5fQWv0Y1PAgbI2tzBb9ajDNJnVnC08PRb154d5skCK0ht4i7eOrgMeltpYR5XkzYP3UAVnN6pij9CH4wPTB1vxfJHPM4QgDTHhPLUW03ZMw_Tp0eVVplpJks-y2i2NPaYz4irMFqVIDV4xWeSVmFo0ujQ8e4w&k=6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-
recognized text: I think it different sizes.
requested url: https://www.google.com/recaptcha/api2/payload/audio.mp3?p=06AFcWeA60qZmzwVsn4Uq9gHNcdtWpyJdu9QVRAdYmOtIFaalV8cJM9toI36BcbmV0FEqEtb8WW5xaYXoXdJDhhaJM3uPCi0Vw87IAiGmBtsvcDEufuzuaIFbWblZfC7sdaInNgp2l6FcyeJ7Zq9Yim_lSVFjNKxhHLzflsMA-sLwaci9AQyh_DSbjjHJYYfE33w2QB51MHan-&k=6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-
recognized text: You might see something.
クライアント・プロセスの起動と実行結果
python .\test_socket_recaptcha_solver.py
Continue(c) or quit(q)?

