[Python] Selenium サイトからCSVファイルをダウンロードするツールを作る 6～CSVファイルのデータをSqliteでDBに登録する
はじめに
前回は、Python＋Seleniumで、WebページからCSVファイルをダウンロードし、ダウンロードしたCSVファイルを読み込んで、Pandasを使って必要なデータの抽出する処理を追加しました。
今回は、ダウンロードしたCSVファイルから抽出したデータを、データベースに登録する処理を追加します。
使用するデータベースは、Pythonの標準ライブラリにある、SQLiteを使用します。
使用するライブラリ
・SQLite3
標準ライブラリ内にある、SQLiteのデータベース用ライブラリ。
動作環境
・windows10
・Python 3.7.6
・Chrome バージョン: 87.0.4280.141
実装
from selenium import webdriver
from selenium.webdriver import ChromeOptions
import os
import glob
import time
import pandas as pd
import sqlite3
downloadsFilePath = 'downloads'
# Selenium初期化
def initialize():
options = webdriver.ChromeOptions()
prefs = {
"download.default_directory": path,
}
options.add_experimental_option("prefs", prefs)
driver_file = r'C:\Users\XXXXX\Documents\webdriver\chromedriver.exe' # ドライバのPathを設定
return webdriver.Chrome(driver_file,options=options)
# 指定したフォルダの最新のファイルパスを取得する
def get_latest_file_path(path):
file_path = ''
if len(os.listdir(path)) != 0:
file_path = max (
[ os.path.join(path, f) for f in os.listdir(path)],
key = os.path.getctime
)
return file_path
# 指定フォルダに、ファイルのダウンロードが完了するまで待機する
def wait_file_download(path):
# 待機タイムアウト時間(秒)設定
timeout_second = 10
# 指定時間分待機
for i in range(timeout_second + 1):
# ファイル一覧取得
match_file_path = os.path.join(path,'*.*')
files = glob.glob(match_file_path)
# ファイルが存在する場合
if files:
# ファイル名の拡張子に、'.crdownload'が含むかを確認
extensions = [file_name for file_name in files if '.crdownload' in os.path.splitext(file_name)]
# '.crdownload'が見つからなかったら抜ける
if not extensions : break
# 指定時間待っても .crdownload 以外のファイルが確認できない場合 エラー
if i >= timeout_second:
# 終了処理
raise Exception('Csv file cannnot be finished downloading!')
# 一秒待つ
time.sleep(1)
return
URL = 'https://www.data.jma.go.jp/obd/stats/data/mdrr/docs/csv_dl_format_mxtem.html'
browser = initialize()
browser.implicitly_wait(10)
# ブラウザでurlを開く
browser.get(URL)
# csvファイルをダウンロード
element = browser.find_element_by_partial_link_text('ダウンロード（最高気温）')
element.click()
# Chromeのダウンロード先に指定したパスを取得
path = os.path.join(os.getcwd(), downloadsFilePath)
# csvダウンロード完了まで処理を待たせる
try:
wait_file_download(path)
except Exception:
sys.exit()
browser.close()
browser.quit()
# ダウンロードしたフォルダから最新のファイルを取得
file_path = get_latest_file_path(path)
# csvファイル読み込み
df = pd.read_csv(file_path,encoding='shift_jis')
# ①特定行を抽出
df_pref = df[df['都道府県'].str.contains('北海道|沖縄県')]
# ②列を追加（年、月、日に分かれているデータを日付型に変換し、新規で、日付の列を追加
date = datetime.date(df.loc[0, '現在時刻(年)'],df.loc[0, '現在時刻(月)'],df.loc[0, '現在時刻(日)'])
df_pref['日付'] = date
# ③特定列を抽出
# 日付、都道府県、地点、今日の最高気温を抽出
df_data = df_pref[['日付','都道府県','地点','今日の最高気温(℃)']]
# 「weather.db」に接続（自動コミットする）
conn = sqlite3.connect('weather.db',isolation_level=None)
# カーソル取得
cur = conn.cursor()
for index, row in df_data.iterrows():
sql = f"INSERT INTO HiTemp values(null,'{row['日付']}', '{row['都道府県']}', '{row['地点']}', '{row['今日の最高気温(℃)']}');"
cur.execute(sql)
# データベースを閉じる
conn.close()追加した処理は、DB登録の処理です。
# 「weather.db」に接続（自動コミットする）
conn = sqlite3.connect('weather.db',isolation_level=None)DBに接続する処理です。
指定したファイルがない場合は、新規作成、存在する場合は、指定データベースを開きます。
# カーソル取得
cur = conn.cursor()
for index, row in df_data.iterrows():
sql = f"INSERT INTO HiTemp values(null,'{row['日付']}', '{row['都道府県']}', '{row['地点']}', '{row['今日の最高気温(℃)']}');"
cur.execute(sql)カーソルを取得して、データを登録するSQL文を実行しています。
# データベースを閉じる
conn.close()DBの接続を終了する処理です。接続と終了はセットで使用します。
登録したデータを表示した結果は下記です。
まとめ
今回は、Python＋Seleniumで、WebページからCSVファイルをダウンロードし、ダウンロードしたCSVファイルを読み込んで、Pandasを使って必要なデータの抽出し、データベースへ登録する処理を追加する処理をお伝えしました。こちらでツールは完成です。
全6回は下記から確認できます！