競馬予測AIの作成⑱（発走時刻の取得処理の改修）

(venv)> pip install pandas
(venv)> pip install requests
(venv)> pip install beautifulsoup4
(venv)> pip install tqdm
(venv)> pip install selenium
(venv)> pip install webdriver_manager

# 開催場所ごとの発走時刻情報を保存しているパス
cell_time_path = ./data/celltime/

# JRAから開催場所ごとのレースの発走時刻を取得する
# 呼び出し元：
#   main.py
# 実行前の設定：
#   config.iniのcell_time_pathで取得した情報の保存先を指定しておく
# 実行方法：
#   main.pyから呼び出されるが、個別に実行することも可能
#   個別に実行する場合は、仮想環境で下記のコマンドを実行する
#   (venv) python get_cellTime.py
# 出力内容：
#   location_list.pkl
#   開催場所ごとの発走時刻を記載したcsvファイル

import os.path
import os
import time
import pickle
import glob
import pandas as pd
import requests
import re
import urllib
import configparser

import selenium
from bs4 import BeautifulSoup
from tqdm import tqdm

from selenium import webdriver
from selenium.webdriver.chrome.service import Service as ChromeService
from selenium.webdriver.common.by import By
from webdriver_manager.chrome import ChromeDriverManager


class JraRaceDataFetcher:
    def __init__(self, inifile):
        # 開催場所のローマ字表記と日本語表記の組み合わせ
        self.loc_dict = dict(inifile['loc_romaji_jp'])
        self.cell_time_path = inifile['settings']['cell_time_path']
        self.error_count = 0
        self.max_errors = 3  # 最大エラー許容回数

    def setup_directory(self):
        """
        発走時刻を保存するディレクトリを設定し、古いCSVファイルを削除する。
        :return: None
        """
        # 発走時刻を保存するディレクトリがなければ作成する
        os.makedirs(self.cell_time_path, exist_ok=True)

        # 過去に取得した発走時刻のCSVを消去する
        for i in glob.glob(self.cell_time_path + '*.csv'):
            os.remove(i)

    def setup_webdriver(self):
        """
        WebDriverをセットアップし、Chromeブラウザのインスタンスを生成する。
        :return:ChromeブラウザのWebDriverインスタンス
        :rtype:selenium.webdriver.Chrome
        """
        # ChromeDriverを自動インストールし、サービスとして開始する
        service = ChromeService(executable_path=ChromeDriverManager().install())
        browser = webdriver.Chrome(service=service)
        browser.implicitly_wait(20)  # 最大待機時間（秒）
        return browser

    def navigate_to_jra_page(self, access_type):
        """
        JRAの出馬表またはレース結果へのナビゲーションし、BeautifulSoupオブジェクトを返す。
        1. WebDriverをセットアップして、JRAのトップページを開く。
        2. 指定されたページタイプ（出馬表またはレース結果）に応じて、該当ページにナビゲートする。
        3. ナビゲーション後のページを取得し、BeautifulSoupオブジェクトを生成して返す。

        :param str access_type: アクセスするページの種類。'entries'（出馬表）または 'results'（レース結果）のいずれか。
        :return: ナビゲート後のページのBeautifulSoupオブジェクト
        :rtype: bs4.BeautifulSoup
        """
        try:
            browser = self.setup_webdriver()
            browser.get('https://www.jra.go.jp/')
            # 出馬表またはレース結果ページへのナビゲーション
            navigation_steps = {
                'entries': ('//*[@id="quick_menu"]/div/ul/li[2]/a',
                            '//*[@id="main"]/div[2]/div/div/div[1]/a',
                            '//*[@id="race_list"]/tbody/tr[1]/th/a'),
                'results': ('//*[@id="quick_menu"]/div/ul/li[2]/a',
                            '//*[@id="main"]/div[2]/div/div/div[1]/a',
                            '//*[@id="race_list"]/tbody/tr[1]/td[8]/a')
            }
            for step in navigation_steps[access_type]:
                browser.find_element(By.XPATH, step).click()
            # BeautifulSoupオブジェクトの作成
            res = requests.get(browser.current_url)
            browser.quit()
            return BeautifulSoup(res.content, 'lxml')
        except selenium.common.exceptions.NoSuchElementException as err:
            self.error_count += 1
            print(f'エラーが発生しました: {err}')
            if self.error_count >= self.max_errors:
                print('最大エラー回数に達したため、処理を終了します。')
                exit(1)
            else:
                print(f'リトライします ({self.error_count}/{self.max_errors})')
                return self.navigate_to_jra_page(access_type)
        except Exception as err:
            print(f'予期しないエラーが発生しました: {err}')
            exit(1)

    def extract_location_info(self, soup):
        """
        開催場所情報を抽出する。
        1. 指定されたBeautifulSoupオブジェクトから開催場所の情報を含むテキストを取得する。
        2. 不要な文字（数字、回、日）を削除し、開催場所をリストとして返す。

        :param bs4.BeautifulSoup soup: BeautifulSoupオブジェクト
        :return: 開催場所のリスト
        :rtype: list
        """
        locations = soup.find('div', class_='link_list multi div3 center mid narrow').text
        return re.sub(r'[0-9回日]', '', locations).split()

    def fetch_location_urls(self, soup):
        """
        開催場所のURLを取得する。
        1. 指定されたBeautifulSoupオブジェクトから開催場所のリンクを含むaタグを全て取得する。
        2. 各リンクのhref属性を基にフルURLを生成し、リストとして返す。

        :param bs4.BeautifulSoup soup: BeautifulSoupオブジェクト
        :return: 開催場所のURLリスト
        :rtype: list
        """
        base_url = 'https://www.jra.go.jp'
        return [urllib.parse.urljoin(base_url, link['href']) for link in soup.select('.link_list.multi.div3.center.mid.narrow a')]

    def preparing_raceinfo_data(self, url):
        """
        レース情報ページから不要なタグや改行を削除し、BeautifulSoupオブジェクトを返す。
        1. 指定されたURLからページを取得し、BeautifulSoupオブジェクトを生成する。
        2. ブリンカー着用の馬のタグ（horse_icon blinker）を削除する。
        3. 不要な改行を削除する。

        :param str url: レース情報ページのURL
        :return: 整形されたレース情報のBeautifulSoupオブジェクト
        :rtype: bs4.BeautifulSoup
        """
        response = requests.get(url)
        response.encoding = response.apparent_encoding
        soup = BeautifulSoup(response.text, 'html5lib')

        # ブリンカー着用の馬のタグを削除
        for tag in soup.find_all('span', class_='horse_icon blinker'):
            tag.decompose()

        # 不要な改行を削除
        for tag in soup.find_all('td', class_='num'):
            tag.string = tag.get_text(strip=True)

        return soup

    def extract_round_urls(self, soup):
        """
        ラウンドのURLを抽出する。
        1. 指定されたBeautifulSoupオブジェクトからラウンド情報を含むulタグを取得する。
        2. 各ラウンドのリンクを含むaタグのhref属性を基にフルURLを生成し、リストとして返す。

        :param bs4.BeautifulSoup soup: BeautifulSoupオブジェクト
        :return: ラウンドのURLリスト
        :rtype: list
        """
        base_url = 'https://www.jra.go.jp'
        rounds_info = soup.find('ul', attrs={'class': 'nav race-num mt15'})
        return [urllib.parse.urljoin(base_url, a['href']) for a in rounds_info.find_all('a')]

    def get_race_times(self, soup, url):
        """
        レースの発走時刻を抽出し、データフレームとして返す。
        1. 指定されたBeautifulSoupオブジェクトから開催場所と日付の情報を抽出する。
        2. 発走時刻を抽出し、整形する。
        3. 開催場所、日付、発走時刻、レース情報のURLを含むデータフレームを作成する。

        :param bs4.BeautifulSoup soup: BeautifulSoupオブジェクト
        :param str url: レース情報ページのURL
        :return: 開催場所、日付、発走時刻、レース情報のURLを含むデータフレーム
        :rtype: pandas.DataFrame
        """
        # 開催場所と日付の情報を抽出
        race_base_info = soup.find('div', class_='cell date').text.split()
        race_location, race_date = race_base_info[1].replace('日', '日目'), race_base_info[0].split('（')[0]

        # 発走時刻の抽出と整形
        cell_time = (soup.find('div', class_='cell time').text.replace('発走時刻：', '').
                     replace('\n', '').replace('分', '').replace(' ', '').
                     replace('時', ':'))

        # データフレームの作成
        cell_time_info = pd.DataFrame([[race_location, race_date, cell_time, url]],
                                      columns=['location', 'race_date', 'cell_time', 'race_info_url'])

        return cell_time_info

    def save_cell_time(self, cell_time_info, loc_name):
        """
        発走時刻を抽出してCSVに保存する。
        1. 指定された開催場所の発走時刻情報をフィルタリングする。
        2. フィルタリングされた発走時刻情報をCSVファイルとして保存する。

        :param pandas.DataFrame cell_time_info: 発走時刻情報を含むデータフレーム
        :param str loc_name: 保存する開催場所の名前
        :return: None
        """
        # 発走時刻を抽出してCSVに保存
        csv_file_path = os.path.join(self.cell_time_path, f'cell_time_info_{loc_name}.csv')
        filtered_info = cell_time_info[cell_time_info['location'].str.contains(loc_name)][['race_date', 'cell_time', 'race_info_url']]
        filtered_info.to_csv(csv_file_path, encoding='cp932', index=False, errors='ignore')

    def get_URL_of_each_racetrack_round(self):
        """
        開催場所ごとのラウンドのURLを取得する。
        1. 出走表ページにナビゲートし、開催場所の情報とそのURLを取得する。
        2. 各開催場所ごとのラウンドのURLを取得する。

        :return: ラウンドのURLリスト、開催場所のリスト、開催場所のURLリスト
        :rtype: tuple (list, list, list)
        """
        # 出走表に取得する
        soup = self.navigate_to_jra_page('entries')

        # レース開催の場所、レース開催場所のURLを取得する
        location_list = self.extract_location_info(soup)
        location_url = self.fetch_location_urls(soup)

        # 開催場所ごとのラウンドのURLを取得する
        round_url = []
        for i in location_url:
            res = requests.get(i)
            res.encoding = res.apparent_encoding  # resに含まれるテキストから文字コードを自動設定する
            soup = BeautifulSoup(res.text, 'html5lib')
            round_url.extend(self.extract_round_urls(soup))

        return round_url, location_list, location_url

    def extract_and_format_of_race_info(self, round_url):
        """
        レース情報を抽出し、整形してデータフレームに追記保存する。
        1. 指定されたラウンドURLリストを順に処理し、各URLからレース情報を抽出する。
        2. 抽出した情報を整形し、発走時刻情報をデータフレームに追加する。

        :param list round_url: ラウンドのURLリスト
        :return: 発走時刻情報を含むデータフレーム
        :rtype: pandas.DataFrame
        """
        # レース情報の抽出と整形を行い、データフレームに追記保存する
        cell_time_info = pd.DataFrame()
        print('発走時刻の取得を開始します')

        for url in tqdm(round_url):
            soup = self.preparing_raceinfo_data(url)
            # 発走時刻を取得する
            try:
                cell_time_info = pd.concat([cell_time_info, self.get_race_times(soup, url)])
            except BaseException as err:
                print(f'予測に必要なデータを取得できません:{err}')

        return cell_time_info

    def save_location_name(self, location_list):
        """
        開催場所の名前を保存する。
        1. 開催場所のリストからローマ字表記のリストを作成する。
        2. ローマ字表記のリストをpickleファイルとして保存する。

        :param list location_list: 開催場所のリスト
        :return: None
        """
        # 開催場所(中山、東京など)のloc_dictのvalue(nakayama,tokyoなど)を取得する
        loc_list = []
        for location in location_list:
            key = [k for k, v in self.loc_dict.items() if v == location][0]
            loc_list.append(key)

        # 開催場所のリストを保存する
        with open(self.cell_time_path + 'location_list.pkl', 'wb') as f:
            pickle.dump(loc_list, f)


def main(inifile):
    """
    プログラムのメイン処理を実行する。
    1. プログラムの開始時刻を取得する。
    2. JraRaceDataFetcherオブジェクトを作成する。
    3. 発走時刻を保存するフォルダがなければ作成し、過去のCSVファイルを消去する。
    4. 開催場所ごとのラウンドのURLを取得する。
    5. ラウンドURLからレース情報を抽出し、整形する。
    6. 各開催場所の発走時刻をCSVファイルとして保存する。
    7. 開催場所の名前をpickleファイルとして保存する。
    8. プログラムの実行時間を表示する。

    :param dict inifile: 初期設定ファイル
    :return: None
    """
    # プログラムの開始時刻を取得する
    start_time = time.time()

    jra_data_fetcher = JraRaceDataFetcher(inifile)

    # 発走時刻を保存するフォルダがなければ作成する
    # 過去に取得した発走時刻のCSVを消去する
    jra_data_fetcher.setup_directory()

    # 開催場所ごとのラウンドのURLを取得する
    round_url, location_list, location_url = jra_data_fetcher.get_URL_of_each_racetrack_round()

    # レース情報の抽出と整形を行う
    cell_time_info = jra_data_fetcher.extract_and_format_of_race_info(round_url)

    # 開催場所の発走時刻を保存する
    for location in location_list:
        jra_data_fetcher.save_cell_time(cell_time_info, location)

    # 開催場所の名前を保存する
    jra_data_fetcher.save_location_name(location_list)

    # プログラムの実行時間を表示する
    print(os.path.basename(__file__), f'実行時間: {time.time() - start_time:.2f}秒')


if __name__ == "__main__":
    # 設定ファイルを読み込む
    inifile = configparser.ConfigParser(interpolation=None)
    inifile.read('./config.ini', 'UTF-8')
    main(inifile)