[Python] Selenium サイトからCSVファイルをダウンロードするツールを作る 6～CSVファイルのデータをSqliteでDBに登録する

はじめに

前回は、Python＋Seleniumで、WebページからCSVファイルをダウンロードし、ダウンロードしたCSVファイルを読み込んで、Pandasを使って必要なデータの抽出する処理を追加しました。

今回は、ダウンロードしたCSVファイルから抽出したデータを、データベースに登録する処理を追加します。

使用するデータベースは、Pythonの標準ライブラリにある、SQLiteを使用します。

使用するライブラリ

・SQLite3

sqlite3 --- DB-API 2.0 interface for SQLite databases

標準ライブラリ内にある、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回は下記から確認できます！