- はじめに
- netkeibaのレース結果のURLパターン
- 関数の概要
- 作成プログラムのフローチャート
- 作成したプログラム
- 各関数の解説
- retry_request(self, url)
- get_detail_race_result(self, url)
- get_race_base_info(self, soup)
- reshape_race_results_data(self, race_results, race_info, b_ml_list, b_fml_list)
- get_pedigree(self, soup)
- check_race_info_existence(self, soup)
- extract_race_results(self, url)
- generate_race_urls(self, year)
- process_race_url(self, race_url)
- process_race_urls_multiprocessing(self, race_url_list, year)
- if __name__ == “__main__”
- 実行環境の作成
- プログラムの実行方法
- プログラムの実行方法
- 次回、試したいこと
はじめに
レース結果を取得するプログラムの可読性と保守性の向上させるために改修を行いました。
レース結果のURLは、chromedriverを用いてNetkeibaのWebサイトを操作して取得していました。chromedriverの起動に失敗、Webサイトの操作失敗、が発生してレース結果のURLを取得できないことがありました。
また、レース結果を取得するプログラムは設計をせずに作成したので似たような処理が混在していたため、無駄にコード量やコメントが多くて可読性がありませんでした。
＜主な変更点＞
- レース結果のURLを作成する関数を用意
- chromedriverの使用を廃止
- Webサイトを操作する関数を廃止
- 同じような処理を繰り返していた箇所を１つの関数にまとめる
netkeibaのレース結果のURLパターン
netkeibaのレース結果のURLにはパターンがあることを知りました。
下記の「2023年8月6日 1回札幌6日目 11R 3歳以上オープン」のレース結果URLで解説します。
「https://db.netkeiba.com/race/202301010611」
URLを分解すると下記のようになります。
| https://db.netkeiba.com/race/ | 2023 | 01 | 01 | 06 | 11 |
| 固定 | 西暦 | 競馬場 | 開催回 | 開催日 | レース |
競馬場の番号は下記となっています。
| 01 | 02 | 03 | 04 | 05 | 06 | 07 | 08 | 09 | 10 |
| 札幌 | 函館 | 福島 | 新潟 | 東京 | 中山 | 中京 | 京都 | 阪神 | 小倉 |
開催回の番号は下記となっています。
| 01 | 02 | 03 | 04 | 05 | 06 |
| 1回 | 2回 | 3回 | 3回 | 4回 | 5回 |
開催日は下記となっています。
| 01 | 02 | 03 | 04 | 05 | 06 | 07 | 08 | 09 | 10 | 11 | 12 |
| 1日目 | 2日目 | 3日目 | 4日目 | 5日目 | 6日目 | 7日目 | 8日目 | 9日目 | 10日目 | 11日目 | 12日目 |
レースは下記となっています。
| 01 | 02 | 03 | 04 | 05 | 06 | 07 | 08 | 09 | 10 | 11 | 12 |
| 1レース | 2レース | 3レース | 3レース | 3レース | 3レース | 3レース | 3レース | 3レース | 10レース | 11レース | 12レース |
競馬場の番号、開催回、開催日、レースを結合することでレース結果のURLを生成することができます。
関数の概要
作成した関数の概要を下記に整理しました。
| 関数名 | 概要 |
| retry_request(self, url) | 指定されたURLに対してリクエストを送信し、応答を取得します。リクエストが失敗した場合にリトライを試みます。 |
| get_detail_race_result(self, url) | 指定されたURLから競馬の詳細なレース結果を取得します。 |
| get_race_base_info(self, soup) | スクレイピングしたHTMLから競馬の基本情報を取得します。BeautifulSoupオブジェクトが引数として渡されます。 |
| reshape_race_results_data(self, race_results, race_info, b_ml_list, b_fml_list) | 競馬のレース結果、レース情報、父親の血統、母親の血統などからデータを整形し、加工します。 |
| get_pedigree(self, soup) | BeautifulSoupオブジェクトから競走馬の血統情報を取得します。 |
| check_race_info_existence(self, soup) | スクレイピングしたHTMLからレース情報の存在を確認します。 |
| extract_race_results(self, url) | 指定されたURLから競馬のレース結果を抽出します。 |
| generate_race_urls(self, year) | 指定された年から競馬のレースURLを生成します。 |
| process_race_url(self, race_url) | 特定の競馬レースURLを処理し、結果を取得します。 |
| process_race_urls_multiprocessing(self, race_url_list, year) | 複数の競馬レースURLを並列処理し、指定された年のレース情報を取得します。 |
作成プログラムのフローチャート
作成したプログラムをcode2flowでフローチャートを作成しました。
フローの概要は下記となります。
フロー内の「0:(global)()」は「main」となります。
- 「main」で「RaceScraper」のオブジェクトを作成しています。
- 「RaceScraper」オブジェクトで「generate_race_urls」と「process_race_urls_multiprocessing」を呼び出します。
- 「process_race_urls_multiprocessing」から「process_race_url」を呼び出しています。
- 「process_race_url」から「get_detail_race_result」を呼び出し、レース結果詳細を取得するための各関数を呼び出しています。
作成したプログラム
作成したプログラムは、既存のプログラムをリファクタリングしました。
import requests
import time
import random
import urllib
import re
import pandas as pd
from bs4 import BeautifulSoup
import multiprocessing
from functools import partial
class RaceScraper:
def __init__(self):
self.max_retries = 5
self.retry_delay = 3
def retry_request(self, url):
"""
指定されたURLからデータを取得します。
:param url:[str]取得するデータのURL
:return:取得に成功した場合は requests.Response オブジェクト、失敗した場合は None
"""
retries = 0
while retries < self.max_retries:
try:
time.sleep(random.uniform(1, 10))
res = requests.get(url)
return res
except requests.exceptions.RequestException:
retries += 1
print(f"retry:{retries},retry_delay:{self.retry_delay}")
time.sleep(self.retry_delay)
return None
def get_detail_race_result(self, url):
"""
指定されたURLからレースの詳細情報を取得します。
:param url:[str]レース情報を取得するURL
:return:レースの詳細情報が整形されたデータフレーム、もしくは 'failed_to_get_info'、'no_race_result' の文字列
"""
res = self.retry_request(url)
if res is None:
print("リクエスト失敗:", url)
return 'failed_to_get_info'
res.encoding = 'EUC-JP'
soup = BeautifulSoup(res.text, 'lxml')
# レース情報の有無をチェックする
if not self.check_race_info_existence(soup):
return 'no_race_result'
# 血統情報を取得する
b_ml_list, b_fml_list = self.get_pedigree(soup)
if not isinstance(b_ml_list, list) or len(b_ml_list) == 0:
return 'failed_to_get_info'
if not isinstance(b_fml_list, list) or len(b_fml_list) == 0:
return 'failed_to_get_info'
# レースの基本情報を取得する
race_base_info = self.get_race_base_info(soup)
# HTML内のレース結果のテーブルをデータフレームとして読み込む
race_results = self.extract_race_results(url)
# 取得した情報を整形する
reshaped_race_results = self.reshape_race_results_data(
race_results, race_base_info, b_ml_list, b_fml_list
)
return reshaped_race_results
def get_race_base_info(self, soup):
"""
レースの基本情報をスクレイピングして取得します。
:param soup:[BeautifulSoup]レース情報を含むBeautifulSoupオブジェクト
:return:レースの基本情報を格納した辞書
"""
# HTMLからレースの基本情報を抽出する
def extract_from_soup(tag, class_name=None, index=0, split_text='\n'):
if class_name:
info = soup.find(tag, attrs={'class': class_name})
return info.text.replace('\n', '').split(split_text)[index]
else:
info = soup.find_all(tag)
return info[index].text.replace('\n', '')
# 各要素からレース情報を抽出する
race_name = extract_from_soup(tag='h1', index=1)
race_round = extract_from_soup(tag='dt', index=0)
race_date = extract_from_soup(tag='p', class_name='smalltxt', index=0, split_text=' ')
race_place = extract_from_soup(tag='p', class_name='smalltxt', index=1, split_text=' ')
race_info_text = extract_from_soup(tag='diary_snap_cut', index=0)
race_distance = race_info_text.split('/')[0]
race_distance = re.sub(r'2周', '', race_distance)
race_distance = int(re.sub(r'\D', '', race_distance))
race_weather = race_info_text.split('/')[1].split(':')[1].replace(u'\xa0', u' ') # ノーブレークスペースを削除
race_weather = race_weather.replace(u'\xa5', u' ') # Webページの文字参照を削除
race_condition = race_info_text.split('/')[2].replace(u'\xa0', u' ') # ノーブレークスペースを削除
race_condition = race_condition.replace(u'\xa5', u' ') # Webページの文字参照を削除
return {
"race_name": race_name,
"race_round": race_round,
"race_date": race_date,
"race_place": race_place,
"race_distance": race_distance,
"race_weather": race_weather,
"race_condition": race_condition
}
def reshape_race_results_data(self, race_results, race_info, b_ml_list, b_fml_list):
"""
レース結果のデータを整形し、加工します。
:param race_results:[DataFrame]レース結果のデータを含むデータフレーム
:param race_info:[dict]レースの基本情報を格納した辞書
:param b_ml_list:[list]各出走馬の父親の名前を保持するリスト
:param b_fml_list:[list]各出走馬の母親の名前を保持するリスト
:return:整形されたレース結果のデータを含むデータフレーム
"""
columns = ['馬名', '性齢', '斤量', '騎手', 'タイム', '単勝', '人 気', '馬体重']
race_results = race_results[columns] # 必要な列だけを抽出する
race_results_data = race_results.copy() # 元のデータを保持しておく
# レース基本情報と出走馬の情報をデータフレームに組み込む
try:
race_results_data['距離'] = race_info['race_distance']
race_results_data['天候'] = race_info['race_weather']
race_results_data['馬場'] = race_info['race_condition'].split(':')[0]
race_results_data['状態'] = race_info['race_condition'].split(':')[1]
race_results_data['開催日'] = race_info['race_date']
race_results_data['レース名'] = race_info['race_name']
race_results_data['開催場所'] = race_info['race_place']
race_results_data['ラウンド'] = race_info['race_round']
race_results_data['父親'] = b_ml_list
race_results_data['母親'] = b_fml_list
# 列の順番を整理し、整形されたデータを返す
reshaped_race_results = race_results_data.reindex(
columns=['タイム', '馬名', '父親', '母親', '性齢', '斤量', '騎手', '単勝',
'人 気', '馬体重', '距離', '天候', '馬場', '状態', '開催日', 'レース名',
'開催場所', 'ラウンド'])
except Exception as e:
print('レース情報を取得できませんでした:', e)
return reshaped_race_results
def get_pedigree(self, soup):
"""
出走馬の血統情報を取得します。
:param soup:[BeautifulSoup]レース情報のHTMLを解析したBeautifulSoupオブジェクト
:return:各出走馬の父親と母親の情報を含む辞書
"""
race_table_data = soup.find(class_='race_table_01 nk_tb_common') # 検索結果のテーブルを取得する
b_ml_list = [] # 各出走馬の父親の名前を保持するリスト
b_fml_list = [] # 各出走馬の母親の名前を保持するリスト
href_list = race_table_data.find_all('a')
for href_link in href_list:
if '/horse/' in href_link.get('href'):
# 出走馬情報のリンクページを作成する
link_url = urllib.parse.urljoin('https://db.netkeiba.com', href_link.get('href'))
# 血統情報を取得する
res = self.retry_request(link_url)
if res is not None:
res.encoding = 'EUC-JP'
soup_blood_table = BeautifulSoup(res.text, 'lxml')
else:
print('リクエスト失敗:', link_url)
# 父親の名前を取得する
elements_b_ml = soup_blood_table.find_all(class_='b_ml')
if len(elements_b_ml) == 0:
print('父親の名前の取得に失敗しました。', link_url)
return 'failed_to_get_b_ml'
element_b_ml = elements_b_ml[0].text.replace('\n', '')
b_ml_list.append(element_b_ml)
# 母親の名前を取得する
elements_b_fml = soup_blood_table.find_all(class_='b_fml')
if len(elements_b_ml) == 0:
print('母親の名前の取得に失敗しました。', link_url)
return 'failed_to_get_b_fml'
element_b_fml = elements_b_fml[1].text.replace('\n', '')
b_fml_list.append(element_b_fml)
return b_ml_list, b_fml_list
def check_race_info_existence(self, soup):
"""
BeautifulSoupオブジェクトからレース情報が存在するかどうかをチェックします。
:param soup:[BeautifulSoup]レース情報のHTMLを解析したBeautifulSoupオブジェクト
:return: レース情報が存在する場合はTrue、それ以外はFalse
"""
html_check = soup.find_all('h1')
return len(html_check) == 2
def extract_race_results(self, url):
"""
指定されたURLからレース結果のデータを抽出します。
:param url:[str]レース結果を取得するためのURL
:return: レース結果のデータを含むデータフレーム
"""
table_list = pd.read_html(url)
return table_list[0]
def generate_race_urls(self, year):
"""
指定された年のレースのURLを生成します。
:param year:[str]レースの開催年
:return:レースのURLのリスト
"""
base_url = "https://db.netkeiba.com/race/"
race_id_list = [
f"{year}{str(place).zfill(2)}{str(held).zfill(2)}{str(date_held).zfill(2)}{str(race).zfill(2)}"
for place in range(1, 11) # 競馬場
for held in range(1, 7) # 開催回
for date_held in range(1, 13) # 開催日
for race in range(1, 13) # レース
]
race_url_list = [
f"{base_url}{race_id}"
for race_id in race_id_list
]
return race_url_list
def process_race_url(self, race_url):
"""
指定されたレースのURLからレースの詳細情報を取得します。
:param race_url:[str]レース情報を取得するためのURL
:return:レースの詳細情報が整形されたデータフレーム、もしくは 'failed_to_get_info'、'no_race_result' の文字列
"""
race_result = self.get_detail_race_result(race_url)
if isinstance(race_result, str):
print(f"{race_url} no_race_result")
return None
print(race_url)
return race_result
def process_race_urls_multiprocessing(self, race_url_list, year):
"""
複数のレースURLを並列処理してレースの詳細情報を取得し、結果をCSVファイルに保存します。
:param race_url_list:[list]レースURLのリスト
:return:なし（結果はCSVファイルに保存）
"""
# num_processes = multiprocessing.cpu_count() # 使用可能なCPUコア数を取得
num_processes = 3
# URLリストをプロセスごとに分割
race_url_chunks = [
race_url_list[i::num_processes] for i in range(num_processes)
]
# マルチプロセスでURLを処理し、結果をリストに保存
with multiprocessing.Pool(processes=num_processes) as pool:
partial_process_race_url = partial(self.process_race_url)
# results = list(tqdm(pool.imap(partial_process_race_url, race_url_list), total=len(race_url_list)))
results = pool.map(partial_process_race_url, race_url_list)
# Noneを除外して結果を結合
race_results = pd.concat([res for res in results if res is not None], ignore_index=True)
file_path = year + '_race_results.csv'
race_results.to_csv(file_path, encoding='cp932', header=False, index=False, errors="ignore")
if __name__ == "__main__":
start_year = 2023
end_year = 2023
race_scraper = RaceScraper()
for year in range(start_year, end_year+1):
race_urls = race_scraper.generate_race_urls(str(year))
race_scraper.process_race_urls_multiprocessing(race_urls, str(year))
各関数の解説
プログラムはクラスベースで作成しており、RaceScraperクラスを用意しました。
RaceScraperクラスの各関数について解説します。
retry_request(self, url)
＜プログラム＞
def retry_request(self, url):
"""
指定されたURLからデータを取得します。
:param url:[str]取得するデータのURL
:return:取得に成功した場合は requests.Response オブジェクト、失敗した場合は None
"""
retries = 0
while retries < self.max_retries:
try:
time.sleep(random.uniform(1, 10))
res = requests.get(url)
return res
except requests.exceptions.RequestException:
retries += 1
print(f"retry:{retries},retry_delay:{self.retry_delay}")
time.sleep(self.retry_delay)
return None＜解説＞
- 指定されたURLからデータを取得する際に、requests.get() を使用します。
- リクエストが例外を発生させた場合には、指定された回数だけリトライを試みます。
- リトライ回数が指定された最大回数 (self.max_retries) を超えるか、リクエストが成功するまで試行します。
- リトライ時には、リトライ間の遅延 (self.retry_delay) が設定され、リトライ回数と遅延時間が出力されます。
- 最終的に、成功した場合は requests.Response オブジェクトが返され、失敗した場合は None が返されます。
get_detail_race_result(self, url)
＜プログラム＞
def get_detail_race_result(self, url):
"""
指定されたURLからレースの詳細情報を取得します。
:param url:[str]レース情報を取得するURL
:return:レースの詳細情報が整形されたデータフレーム、もしくは 'failed_to_get_info'、'no_race_result' の文字列
"""
res = self.retry_request(url)
if res is None:
print("リクエスト失敗:", url)
return 'failed_to_get_info'
res.encoding = 'EUC-JP'
soup = BeautifulSoup(res.text, 'lxml')
# レース情報の有無をチェックする
if not self.check_race_info_existence(soup):
return 'no_race_result'
# 血統情報を取得する
b_ml_list, b_fml_list = self.get_pedigree(soup)
if not isinstance(b_ml_list, list) or len(b_ml_list) == 0:
return 'failed_to_get_info'
if not isinstance(b_fml_list, list) or len(b_fml_list) == 0:
return 'failed_to_get_info'
# レースの基本情報を取得する
race_base_info = self.get_race_base_info(soup)
# HTML内のレース結果のテーブルをデータフレームとして読み込む
race_results = self.extract_race_results(url)
# 取得した情報を整形する
reshaped_race_results = self.reshape_race_results_data(
race_results, race_base_info, b_ml_list, b_fml_list
)
return reshaped_race_results＜解説＞
- 指定されたURLからリクエストを送り、その結果を取得します。
- 取得したデータを解析し、レース情報の有無を確認し、血統情報を取得します。
- 特定の条件に合致しない場合は、’failed_to_get_info’ もしくは ‘no_race_result’ のいずれかを返します。
- 取得した情報を整形してデータフレームとして返します。
get_race_base_info(self, soup)
＜プログラム＞
def get_race_base_info(self, soup):
"""
レースの基本情報をスクレイピングして取得します。
:param soup:[BeautifulSoup]レース情報を含むBeautifulSoupオブジェクト
:return:レースの基本情報を格納した辞書
"""
# HTMLからレースの基本情報を抽出する
def extract_from_soup(tag, class_name=None, index=0, split_text='\n'):
if class_name:
info = soup.find(tag, attrs={'class': class_name})
return info.text.replace('\n', '').split(split_text)[index]
else:
info = soup.find_all(tag)
return info[index].text.replace('\n', '')
# 各要素からレース情報を抽出する
race_name = extract_from_soup(tag='h1', index=1)
race_round = extract_from_soup(tag='dt', index=0)
race_date = extract_from_soup(tag='p', class_name='smalltxt', index=0, split_text=' ')
race_place = extract_from_soup(tag='p', class_name='smalltxt', index=1, split_text=' ')
race_info_text = extract_from_soup(tag='diary_snap_cut', index=0)
race_distance = race_info_text.split('/')[0]
race_distance = re.sub(r'2周', '', race_distance)
race_distance = int(re.sub(r'\D', '', race_distance))
race_weather = race_info_text.split('/')[1].split(':')[1].replace(u'\xa0', u' ') # ノーブレークスペースを削除
race_weather = race_weather.replace(u'\xa5', u' ') # Webページの文字参照を削除
race_condition = race_info_text.split('/')[2].replace(u'\xa0', u' ') # ノーブレークスペースを削除
race_condition = race_condition.replace(u'\xa5', u' ') # Webページの文字参照を削除
return {
"race_name": race_name,
"race_round": race_round,
"race_date": race_date,
"race_place": race_place,
"race_distance": race_distance,
"race_weather": race_weather,
"race_condition": race_condition
}＜解説＞
- 与えられた BeautifulSoup オブジェクトから extract_from_soup 関数を使って特定のHTML要素を抽出します。
- 抽出した要素を整形してレースの基本情報を抽出します。
- 抽出された情報は辞書にまとめられ、関数の最後でそれを返します。
reshape_race_results_data(self, race_results, race_info, b_ml_list, b_fml_list)
＜プログラム＞
def reshape_race_results_data(self, race_results, race_info, b_ml_list, b_fml_list):
"""
レース結果のデータを整形し、加工します。
:param race_results:[DataFrame]レース結果のデータを含むデータフレーム
:param race_info:[dict]レースの基本情報を格納した辞書
:param b_ml_list:[list]各出走馬の父親の名前を保持するリスト
:param b_fml_list:[list]各出走馬の母親の名前を保持するリスト
:return:整形されたレース結果のデータを含むデータフレーム
"""
columns = ['馬名', '性齢', '斤量', '騎手', 'タイム', '単勝', '人 気', '馬体重']
race_results = race_results[columns] # 必要な列だけを抽出する
race_results_data = race_results.copy() # 元のデータを保持しておく
# レース基本情報と出走馬の情報をデータフレームに組み込む
try:
race_results_data['距離'] = race_info['race_distance']
race_results_data['天候'] = race_info['race_weather']
race_results_data['馬場'] = race_info['race_condition'].split(':')[0]
race_results_data['状態'] = race_info['race_condition'].split(':')[1]
race_results_data['開催日'] = race_info['race_date']
race_results_data['レース名'] = race_info['race_name']
race_results_data['開催場所'] = race_info['race_place']
race_results_data['ラウンド'] = race_info['race_round']
race_results_data['父親'] = b_ml_list
race_results_data['母親'] = b_fml_list
# 列の順番を整理し、整形されたデータを返す
reshaped_race_results = race_results_data.reindex(
columns=['タイム', '馬名', '父親', '母親', '性齢', '斤量', '騎手', '単勝',
'人 気', '馬体重', '距離', '天候', '馬場', '状態', '開催日', 'レース名',
'開催場所', 'ラウンド'])
except Exception as e:
print('レース情報を取得できませんでした:', e)
return reshaped_race_results＜解説＞
- 与えられた race_results データフレームから特定の列のみを抽出します。
- race_info(競馬場、開催場所、天候など)や b_ml_list(父親の血統)、b_fml_list(母親の血統) の情報を用いて新しい列を作成しています。
- 整形されたデータフレームには、レースの基本情報や出走馬の血統情報、レース結果に関する情報などが含まれます。
get_pedigree(self, soup)
＜プログラム＞
def get_pedigree(self, soup):
"""
出走馬の血統情報を取得します。
:param soup:[BeautifulSoup]レース情報のHTMLを解析したBeautifulSoupオブジェクト
:return:各出走馬の父親と母親の情報を含む辞書
"""
race_table_data = soup.find(class_='race_table_01 nk_tb_common') # 検索結果のテーブルを取得する
b_ml_list = [] # 各出走馬の父親の名前を保持するリスト
b_fml_list = [] # 各出走馬の母親の名前を保持するリスト
href_list = race_table_data.find_all('a')
for href_link in href_list:
if '/horse/' in href_link.get('href'):
# 出走馬情報のリンクページを作成する
link_url = urllib.parse.urljoin('https://db.netkeiba.com', href_link.get('href'))
# 血統情報を取得する
res = self.retry_request(link_url)
if res is not None:
res.encoding = 'EUC-JP'
soup_blood_table = BeautifulSoup(res.text, 'lxml')
else:
print('リクエスト失敗:', link_url)
# 父親の名前を取得する
elements_b_ml = soup_blood_table.find_all(class_='b_ml')
if len(elements_b_ml) == 0:
print('父親の名前の取得に失敗しました。', link_url)
return 'failed_to_get_b_ml'
element_b_ml = elements_b_ml[0].text.replace('\n', '')
b_ml_list.append(element_b_ml)
# 母親の名前を取得する
elements_b_fml = soup_blood_table.find_all(class_='b_fml')
if len(elements_b_ml) == 0:
print('母親の名前の取得に失敗しました。', link_url)
return 'failed_to_get_b_fml'
element_b_fml = elements_b_fml[1].text.replace('\n', '')
b_fml_list.append(element_b_fml)
return b_ml_list, b_fml_list＜解説＞
- race_table_data から各出走馬の情報のリンクを見つけます。
- それぞれのリンクにアクセスして父親と母親の名前を取得します。
- それぞれの名前は b_ml_list(父親の血統) と b_fml_list(母親の血統) に追加されます。
- 最終的に、b_ml_list と b_fml_list を返します。
- リクエスト失敗やデータ取得に失敗した場合、該当するエラーメッセージを返しています。
check_race_info_existence(self, soup)
＜プログラム＞
def check_race_info_existence(self, soup):
"""
BeautifulSoupオブジェクトからレース情報が存在するかどうかをチェックします。
:param soup:[BeautifulSoup]レース情報のHTMLを解析したBeautifulSoupオブジェクト
:return: レース情報が存在する場合はTrue、それ以外はFalse
"""
html_check = soup.find_all('h1')
return len(html_check) == 2
＜解説＞
- 与えられた soup (BeautifulSoupオブジェクト) から ‘h1’ タグを探します。
- ‘h1’タグが2つであるかどうかを確認します。通常、レース情報のページではレース名などの情報を含む ‘h1’ タグが2つ存在します。
- ‘h1’ タグの数が2つならば、レース情報が存在すると判断し、True を返します。それ以外の場合は、False を返します。
extract_race_results(self, url)
＜プログラム＞
def extract_race_results(self, url):
"""
指定されたURLからレース結果のデータを抽出します。
:param url:[str]レース結果を取得するためのURL
:return: レース結果のデータを含むデータフレーム
"""
table_list = pd.read_html(url)
return table_list[0]＜解説＞
- 指定されたURLを使ってデータを読み込みます。
- 指定されたURLからHTMLテーブルを取得して、それをデータフレームのリスト (table_list) として返します。
- HTMLページから取得された複数のテーブルのリストがあるので、そのうちの最初のテーブル（レース結果のテーブル）を返します。
- 指定されたURLから取得したレース結果のデータが含まれたデータフレームが返されます。
generate_race_urls(self, year)
＜プログラム＞
def generate_race_urls(self, year):
"""
指定された年のレースのURLを生成します。
:param year:[str]レースの開催年
:return:レースのURLのリスト
"""
base_url = "https://db.netkeiba.com/race/"
race_id_list = [
f"{year}{str(place).zfill(2)}{str(held).zfill(2)}{str(date_held).zfill(2)}{str(race).zfill(2)}"
for place in range(1, 11) # 競馬場
for held in range(1, 7) # 開催回
for date_held in range(1, 13) # 開催日
for race in range(1, 13) # レース
]
race_url_list = [
f"{base_url}{race_id}"
for race_id in race_id_list
]
return race_url_list＜解説＞
- 与えられた year の年における競馬の全レースのURLを生成します。
- base_url はレース情報が記載されているベースとなるURLです。
- そのURLに年やレースIDを組み合わせて、レースのURLを生成しています。
- race_id_list は、競馬場、開催回、開催日、レース番号を組み合わせて、1年間の全レースのIDのリストを作成しています。
- 各レースIDに基づいて race_url_list が生成されています。
- 最終的に、レースのURLが格納されたリストを返します。
process_race_url(self, race_url)
＜プログラム＞
def process_race_url(self, race_url):
"""
指定されたレースのURLからレースの詳細情報を取得します。
:param race_url:[str]レース情報を取得するためのURL
:return:レースの詳細情報が整形されたデータフレーム、もしくは 'failed_to_get_info'、'no_race_result' の文字列
"""
race_result = self.get_detail_race_result(race_url)
if isinstance(race_result, str):
print(f"{race_url} no_race_result")
return None
print(race_url)
return race_result＜解説＞
- 与えられた race_url を引数に取り、get_detail_race_result メソッドを使ってそのURLからレースの詳細情報を取得しています。
- race_result の戻り値が文字列型の場合、具体的には ‘failed_to_get_info’ もしくは ‘no_race_result’ のいずれかである場合、そのURLとメッセージを標準出力して、None を返します。
- 上記以外は、race_result を返し、その内容を標準出力します。
- get_detail_race_result メソッドはレースの詳細情報を整形し、’failed_to_get_info’ もしくは ‘no_race_result’ の文字列、または整形されたデータフレームを返すようです。
process_race_urls_multiprocessing(self, race_url_list, year)
＜プログラム＞
def process_race_urls_multiprocessing(self, race_url_list, year):
"""
複数のレースURLを並列処理してレースの詳細情報を取得し、結果をCSVファイルに保存します。
:param race_url_list:[list]レースURLのリスト
:return:なし（結果はCSVファイルに保存）
"""
# num_processes = multiprocessing.cpu_count() # 使用可能なCPUコア数を取得
num_processes = 3
# URLリストをプロセスごとに分割
race_url_chunks = [
race_url_list[i::num_processes] for i in range(num_processes)
]
# マルチプロセスでURLを処理し、結果をリストに保存
with multiprocessing.Pool(processes=num_processes) as pool:
partial_process_race_url = partial(self.process_race_url)
# results = list(tqdm(pool.imap(partial_process_race_url, race_url_list), total=len(race_url_list)))
results = pool.map(partial_process_race_url, race_url_list)
# Noneを除外して結果を結合
race_results = pd.concat([res for res in results if res is not None], ignore_index=True)
file_path = year + '_race_results.csv'
race_results.to_csv(file_path, encoding='cp932', header=False, index=False, errors="ignore")＜解説＞
- 与えられた race_url_list に含まれる複数のレースURLを、マルチプロセスを使って並列で処理しています。
- num_processes = 3 のように3つのプロセスを利用するようにしています。
実行環境に応じてこちらの数字を変えてください。 - race_url_list を num_processes の数に応じて分割して、それぞれのプロセスで扱うための race_url_chunks を作成しています。
- multiprocessing.Pool を使用してプロセスプールを作成し、pool.map() を使って self.process_race_url メソッドを複数のURLに対してマップしています。各プロセスは self.process_race_url を呼び出してレースの詳細情報を取得します。
- race_results を構築し、その結果をCSVファイルに保存しています。
- CSVファイルの名前は year に基づいて作成されます。
- 得られた結果が None でない場合、その結果をCSVに書き込みます。
if __name__ == “__main__”
＜プログラム＞
if __name__ == "__main__":
start_year = 2023
end_year = 2023
race_scraper = RaceScraper()
for year in range(start_year, end_year+1):
race_urls = race_scraper.generate_race_urls(str(year))
race_scraper.process_race_urls_multiprocessing(race_urls, str(year))＜解説＞
- start_year と end_year を定義しており、それぞれ 2023 年と設定しています。
start_year と end_yearを変更することで、複数年のレース結果を取得することが可能です。 - RaceScraper クラスのオブジェクト race_scraper を生成しています。
- for ループでは、range(start_year, end_year+1) によって、指定された範囲の年に対してループが実行されます。
- 各イテレーションでは、year をその年の文字列に変換しrace_scraper.generate_race_urls() メソッドを呼び出して、その年のレースURLのリスト race_urls を取得します。
- race_scraper.process_race_urls_multiprocessing() メソッドを使用して、取得したレースURLのリストとその年の文字列を引数として渡します。並列処理を利用してレースの詳細情報を取得し、その結果をCSVファイルに保存する処理を行います。
実行環境の作成
下記のコマンドを実行して仮想環境を作成します。
> python3 -m venv venv下記のコマンドを実行して仮想環境を起動します。
> .\venv\Scripts\activate下記のコマンドを実行して仮想環境にパッケージをインストールします。
(venv) > .\venv\Scripts\python.exe -m pip install --upgrade pip
(venv) > pip install requests==2.31.0
(venv) > pip install pandas==2.0.3
(venv) > pip install bs4==0.0.1
(venv) > pip install lxml==4.9.3＜主要パッケージのバージョン＞
| Python | 3.10.8 |
| requests | 2.31.0 |
| pandas | 2.0.3 |
| bs4 | 0.0.1 |
| lxml | 4.9.3 |
プログラムの実行方法
プログラムを実行する前に下記を実施しておきます。
- マルチプロセスで実行する数の設定
- レース結果を取得する年の設定
プロセスの数は「process_race_urls_multiprocessing」の下記の箇所で設定します。
下記の場合は、プロセス数を「3」としています。数を増やすと同時に取得できるレース結果の情報は増えます。ただし、実行環境の負荷が高まるのと、netkeibaのサーバに負荷を掛けてしまうので、2あるいは3が安定して稼働できる数値です。
num_processes = 3レース結果を取得する年は、main処理の下記の箇所で設定します。
下記の場合は、2023年のレース結果を取得することになります。start_yearを「2021」とすると、2021年から2023年のレース結果を取得することになります。
start_year = 2023
end_year = 2023プログラムの実行方法
仮想環境で下記のようにプログラムを実行してください。
引数などは不要です。
(venv) python get_raceResults.py標準出力で下記のような表示がされてたら、実行が開始しています。
https://db.netkeiba.com/race/202301011001 no_race_resulthttps://db.netkeiba.com/race/202301011002 no_race_result https://db.netkeiba.com/race/202301010701 no_race_result https://db.netkeiba.com/race/202301010702 no_race_result https://db.netkeiba.com/race/202301011003 no_race_result https://db.netkeiba.com/race/202301010703 no_race_result https://db.netkeiba.com/race/202301011004 no_race_result https://db.netkeiba.com/race/202301010704 no_race_result https://db.netkeiba.com/race/202301011005 no_race_result https://db.netkeiba.com/race/202301010705 no_race_result https://db.netkeiba.com/race/202301011006 no_race_result https://db.netkeiba.com/race/202301010706 no_race_result https://db.netkeiba.com/race/202301010707 no_race_result https://db.netkeiba.com/race/202301011007 no_race_result https://db.netkeiba.com/race/202301010708 no_race_result https://db.netkeiba.com/race/202301010101 https://db.netkeiba.com/race/202301011008 no_race_result https://db.netkeiba.com/race/202301010709 no_race_resultURLのあとに「no_race_result」となっている場合は、そのURLにはレース結果が存在しません。
URLのあとに何も表示されない場合は、そのURLのレース結果を取得しています。
プログラムの実行が完了すると、年ごとのレース結果のCSVファイルが作成されます。
CSVファイル名の例：2023_race_results.csv
＜レース結果CSVファイルの内容＞
次回、試したいこと
予測処理を複数用意して選択できるようにしたり、UIを用意して使い勝手を上げたい、など試したいことは色々ありますが、まずは既存のプログラムの可読性と保守性を向上させます。
コメント