python3でサイト内リンクを収集してアクセスを判定する　その２

前回記事でサイト内の外部リンクを収集してステータスを確認しCSVにまとめるコードを作りました。しかし対象となるサイトが予想以上にページもリンクも多く、実行に時間がかかったので短縮できるよう改変しました。
前回の実装は内部URLを収集しつつ外部URLのステータスを判定する方法でした。そこから少し手を加えて対象サイトのページツリーから探索対象のURLを収集し、その後各URL内の外部URLを判定する方法になりました。

準備

前回のソースコードを少し変えているだけなのであまり変わりません。ただページツリーのJavaScript制御の折りたたみをすべて開く必要があったため、新たにブラウザを操作できるライブラリのSeleniumを使用しています。

OS:Windows10
言語:python3.7
主な使用ライブラリ:requests,beautifulsoup4,Selenium
必要なライブラリはcondaやpipなどを利用してインストールしてください。

実装

コードの全文がこちら。前回記述した通り、今回の作業が自動化できればいいと思って作ったのであまり汎用的ではありません。コメントアウトされている箇所は前回の実装になったけど改変で不要になった記述です。

import time
import re
import chromedriver_binary
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import os
import sys
import signal
import requests
from requests.exceptions import Timeout
from bs4 import BeautifulSoup
from urllib.parse import urljoin
import datetime
import os.path
from os import makedirs
from urllib.request import urlretrieve
import codecs
from selenium import webdriver
from selenium.webdriver.common.desired_capabilities import DesiredCapabilities

options = webdriver.ChromeOptions()
options.add_argument('--headless')  #don't Display
# enable browser logging
d = DesiredCapabilities.CHROME
d['loggingPrefs'] = {'performance': 'ALL'}
driver = webdriver.Chrome(options=options)

#wait 5seconds
wait = WebDriverWait(driver, 5)

# ID&pass
USER = "devhogepiyo"
PASS = "piyopiyo"

#access
driver.get('http://example.amazonaws.com/login.action?destination=%2Fpages%2hoge&permissionViolation=true')
#login
id = driver.find_element_by_id("username")
id.send_keys(USER)
password = driver.find_element_by_id("password")
password.send_keys(PASS)

time.sleep(1) #write wait

#login send
login_button = driver.find_element_by_id("loginButton")
login_button.click()

#set
h_files = set()
abc_link = set()
xyz_link = set()
base = "http://example.amazonaws.com/"

#open nest
try:
   while True:
       menu_ul = driver.find_element_by_class_name("tree-browser-section")
       if menu_ul:
           print('true')
           try:
               nest = menu_ul.find_element_by_xpath('//a[@class="plugin_pagetree_childtoggle aui-icon aui-icon-small aui-iconfont-chevron-right" and @href="#"]')
           except:
               print('empty')
               break
           driver.execute_script("arguments[0].click();", nest)
           time.sleep(1.3)
except:
   print('open.')
   import traceback
   traceback.print_exc()

try:
   menu_ul = driver.find_element_by_class_name("tree-browser-section")
   if menu_ul:
       try:
           for link in menu_ul.find_elements_by_xpath('//a[contains(@href,"/pages/")]'):
               link_url = link.get_attribute("href")
               url = urljoin(base, link_url)  # 絶対パスに
               abc_link.add(url)
           print('set links.')
       except:
           print('failed.')
except:
   import traceback
   traceback.print_exc()

for a in abc_link:
   print(a)

now = datetime.datetime.now()  # 時間まで
fname = 'test' + now.strftime('%Y%m%d%H%M%S') + '.csv'

#login
login_info = {
   "username": USER,
   "password": PASS,
   "cookie": "true"
}

#start session
session = requests.session()
#セッションの受け渡し
for cookie in driver.get_cookies():
   session.cookies.set(cookie["name"], cookie["value"])

h_files.add('http://example.amazonaws.com/collector/pages.action?key=PROJECT')

for s in abc_link:
   try:
       res = session.get(s, data=login_info)
   except:
       print('not exist')
       pass
   # BeauttifulSoupを使用してHTML整形
   soup = BeautifulSoup(res.text, 'html.parser')
   main = soup.find('div', id="main")
   #get title
   h1 = main.find('h1')
   if h1:
       title = h1.find('a')
   title_text = ''
   if title is not None:
       title_text = title.string
       if title_text is not None:
           title_text = str(title_text)
       else:
           title_text = ''
   re_url = str(res.url) if res.url is not None else ''
   print(title_text)
   #write CSV
   f = open(fname, 'a',  encoding='cp932', errors='ignore')
   f.write(title_text + ',' + str(re_url) + '\n')
   f.close()
   # aタグからURLを取得し、HTTPリクエストを送る
   for link in main.find_all('a', href=True):
       # URLを取得し、HTTPリクエスト
       url = link.get('href')
       link_text = link.string
       if url is None:
           continue
       elif url == '#':
           continue
       elif link_text is not None:
           link_text = link_text.replace(',', ' ')
       url = urljoin(base, url)  # 絶対パスに
       #内外判定
       if url.startswith(base) or url.startswith("https://example.amazonaws.com/"):
           link_kind = '内部リンク'
       else:
           link_kind = '外部リンク'
       #重複防止
       """if url in h_files:
           continue
       else:"""
       option = ''
       try:
           res = requests.get(url, timeout=(5.0, 7.5))
           st = str(res.status_code)
           if ('docs.google.com' in url) or ('drive.google.com' in url):
               st = '403'
               option = 'googleDrive'
           elif '/cacoo/' in url:
               st = '403'
               option = 'cacoo'
           print(st)
       except requests.exceptions.RequestException:
           print('timeout:' + str(res.status_code))
           st = 'timeout'
           pass
       h_files.add(url)

       # 結果をCSVに出力する
       f = codecs.open(fname, 'a', encoding='cp932', errors='ignore')
       f.write(str(link_text) + ',' + st +
               ',' + url + ',' + link_kind + option + '\n')
       f.close()
#next target url
   """if abc_link != xyz_link:
       sa = list(abc_link - xyz_link)
       for s in sa:
           if bool(s) == False:
               continue
           else:
               try:
                   res = session.get(s, data=login_info)
                   #res.raise_for_status()
                   break
               except requests.exceptions.RequestException:
                   print('timeout.')
                   pass
               finally:
                   xyz_link.add(s)  # 探索済set
   else:
       print('Done')
       sys.exit()"""


driver.quit()

import

import time
import re
import chromedriver_binary
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import os
import sys
import signal
import requests
from requests.exceptions import Timeout
from bs4 import BeautifulSoup
from urllib.parse import urljoin
import datetime
import os.path
from os import makedirs
from urllib.request import urlretrieve
import codecs
from selenium import webdriver
from selenium.webdriver.common.desired_capabilities import DesiredCapabilities

試行錯誤したため使ってないライブラリも含まれていそうですが同じようなことを実現したい人がいたのためにそのまま載せます。

サイトへの接続準備とログイン

options = webdriver.ChromeOptions()
options.add_argument('--headless')  #don't Display
# enable browser logging
d = DesiredCapabilities.CHROME
d['loggingPrefs'] = {'performance': 'ALL'}
driver = webdriver.Chrome(options=options)

今回はGoogleChromeを操作するためChromeのwebdriverをインポートして使用しています。webdriverを配置したディレクトリにパスを通すのを忘れないようにしましょう。2行目の記述は実際にブラウザを立ち上げるか否かの設定ができます。--headlessを指定するとブラウザが立ち上がらずバックグラウンドで処理されます。はじめは挙動を確認するために表示して実行すると良いでしょう。

#wait 5seconds
wait = WebDriverWait(driver, 5)

# ID&pass
USER = "devhogepiyo"
PASS = "piyopiyo"

#access
driver.get('http://example.amazonaws.com/login.action?destination=%2Fpages%2hoge&permissionViolation=true')
#login
id = driver.find_element_by_id("username")
id.send_keys(USER)
password = driver.find_element_by_id("password")
password.send_keys(PASS)

time.sleep(1) #write wait

#login send
login_button = driver.find_element_by_id("loginButton")
login_button.click()

変数waitにはWebを操作した際の待機時間を設定し代入します。サイト側が処理に必要な時間以上を設定しておかないとSelenium側が先走って想定外の動作をしてしまうことがあるので注意しましょう。準備ができたらdriver.get()でログインページを開きます。seleniumでDOM操作ができるので関数でフォームにログインに必要な情報を入力します。入力し終わるのに1秒待ってから、関数でログインボタンを見つけてクリックさせます。

#open nest
try:
   while True:
       menu_ul = driver.find_element_by_class_name("tree-browser-section")
       if menu_ul:
           print('true')
           try:
               nest = menu_ul.find_element_by_xpath('//a[@class="plugin_pagetree_childtoggle aui-icon aui-icon-small aui-iconfont-chevron-right" and @href="#"]')
           except:
               print('empty')
               break
           driver.execute_script("arguments[0].click();", nest)
           time.sleep(1.3)
except:
   print('open.')
   import traceback
   traceback.print_exc()

対象サイトのログイン後ページにはページツリーがあり、このツリーがアイコンをクリックするとJSが動作して下位ページリンクが出てくるネスト構造だったので、Seleniumを使って全て開いていきます。上記のコードはツリーの中からアイコンをclass要素で探してなくなるまでクリックしていく処理です。

try:
   menu_ul = driver.find_element_by_class_name("tree-browser-section")
   if menu_ul:
       try:
           for link in menu_ul.find_elements_by_xpath('//a[contains(@href,"/pages/")]'):
               link_url = link.get_attribute("href")
               url = urljoin(base, link_url)  # 絶対パスに
               abc_link.add(url)
           print('set links.')
       except:
           print('failed.')
except:
   import traceback
   traceback.print_exc()

ツリーのネストを全て開いたら含まれるリンクをsetに収集します。条件は/pages/が含まれているhttpURLで探しています。Seleniumの活躍はここまで。

サイト内外部URL収集

now = datetime.datetime.now()  # 時間まで
fname = 'test' + now.strftime('%Y%m%d%H%M%S') + '.csv'

#login
login_info = {
   "username": USER,
   "password": PASS,
   "cookie": "true"
}

#start session
session = requests.session()
#セッションの受け渡し
for cookie in driver.get_cookies():
   session.cookies.set(cookie["name"], cookie["value"])

h_files.add('http://example.amazonaws.com/collector/pages.action?key=PROJECT')

for s in abc_link:
   try:
       res = session.get(s, data=login_info)
   except:
       print('not exist')
       pass
   # BeauttifulSoupを使用してHTML整形
   soup = BeautifulSoup(res.text, 'html.parser')
   main = soup.find('div', id="main")
   #get title
   h1 = main.find('h1')
   if h1:
       title = h1.find('a')
   title_text = ''
   if title is not None:
       title_text = title.string
       if title_text is not None:
           title_text = str(title_text)
       else:
           title_text = ''
   re_url = str(res.url) if res.url is not None else ''
   print(title_text)
   #write CSV
   f = open(fname, 'a',  encoding='cp932', errors='ignore')
   f.write(title_text + ',' + str(re_url) + '\n')
   f.close()
   # aタグからURLを取得し、HTTPリクエストを送る
   for link in main.find_all('a', href=True):
       # URLを取得し、HTTPリクエスト
       url = link.get('href')
       link_text = link.string
       if url is None:
           continue
       elif url == '#':
           continue
       elif link_text is not None:
           link_text = link_text.replace(',', ' ')
       url = urljoin(base, url)  # 絶対パスに
       #内外判定
       if url.startswith(base) or url.startswith("https://example.amazonaws.com/"):
           link_kind = '内部リンク'
       else:
           link_kind = '外部リンク'
       #重複防止
       """if url in h_files:
           continue
       else:"""
       option = ''
       try:
           res = requests.get(url, timeout=(5.0, 7.5))
           st = str(res.status_code)
           if ('docs.google.com' in url) or ('drive.google.com' in url):
               st = '403'
               option = 'googleDrive'
           elif '/cacoo/' in url:
               st = '403'
               option = 'cacoo'
           print(st)
       except requests.exceptions.RequestException:
           print('timeout:' + str(res.status_code))
           st = 'timeout'
           pass
       h_files.add(url)

       # 結果をCSVに出力する
       f = codecs.open(fname, 'a', encoding='cp932', errors='ignore')
       f.write(str(link_text) + ',' + st +
               ',' + url + ',' + link_kind + option + '\n')
       f.close()
#next target url
   """if abc_link != xyz_link:
       sa = list(abc_link - xyz_link)
       for s in sa:
           if bool(s) == False:
               continue
           else:
               try:
                   res = session.get(s, data=login_info)
                   #res.raise_for_status()
                   break
               except requests.exceptions.RequestException:
                   print('timeout.')
                   pass
               finally:
                   xyz_link.add(s)  # 探索済set
   else:
       print('Done')
       sys.exit()"""


driver.quit()

サイト内外部リンク収集はすでに前回のRequestsを使ったコードがあったので再利用しました。一部削ってますがほとんど同じです。SeleniumからRequestsへサイト操作を交代するのにcookie情報を渡しています。

JS制御のあるサイトも構造がパターン化されているものならSeleniumで少し手を加えるだけで収集できました。

参考:
Python + Selenium で Chrome の自動操作を一通り
PythonでSeleniumを使ってスクレイピング (基礎)
SeleniumからStableになったHeadless Chrome/Firefoxを使ってみる
XPath (XML Path Language) とは