競馬予想AIのためのデータ収集！スクレイピングでレース結果を簡単取得する方法

from module import RequestWrapper
import pandas as pd
import io
from bs4 import BeautifulSoup
import re
import time, requests

# リクエスト制御用のラッパー(基本的にnetkeibaしかアクセスしないため簡易的な制御)
class RequestWrapper(object):
    def __init__(self, request_interval_milli = 1000):
        self.request_interval_milli = request_interval_milli
        self.last_requested_at = 0

    def get(self, url):
        target_epochmilli = self.last_requested_at + self.request_interval_milli
        self.sleep_until(target_epochmilli)

        response = requests.get(url)
        self.last_requested_at = time.time()*1000
        
        return response

    def sleep_until(self, target_epochmilli):
        epochmilli = time.time()*1000
        sleep_sec = (target_epochmilli - epochmilli)/1000
        if(sleep_sec > 0):
            time.sleep(sleep_sec)

# レース情報を抽出する
def extract_race_info_from_soup(soup):
    all_info = soup.find("div", attrs={"class": "data_intro"})
    race_data = all_info.find("dl", attrs={"class": "racedata fc"})

    # レース番号の取得
    race_number = race_data.find("dt").get_text().replace("\n", "").replace(" ", "")

    # レースのコンディションを取得
    race_condition1 = race_data.find("dd").find("p").find("span").get_text()
    parsed_race_condition1 = race_condition1.split(u'\xa0/\xa0')

    # レースのコンディションを取得(小さいテキスト)
    race_condition2 = all_info.find("p", attrs={"class":"smalltxt"}).get_text()
    parsed_race_condition2 = race_condition2.split()

    merged =  [race_number] + parsed_race_condition1 + parsed_race_condition2
    return merged

# 結果テーブルを抽出する
def extract_result_table_from_soup(soup):
    # tableタグを抜き出す
    table = soup.find("table", attrs={"summary":"レース結果"})
    rows = table.findAll("tr")

    # カラム名を抽出
    column_names = [column_name.get_text().replace("\n", "") for column_name in rows[0].findAll("th")]

    # 行毎にパースする
    parsed_rows = []
    for row in rows[1:]:
        parsed_row = []
        for cell in row.findAll("td"):
            text = cell.get_text().replace("\n", "")

            a_tag = cell.find("a")
            if(a_tag):
                text += "(" + a_tag.get("href") + ")"
            parsed_row.append(text)
        parsed_rows.append(parsed_row)

    # データフレームに埋め込む
    extracted = pd.DataFrame(parsed_rows, columns = column_names)
    return extracted

# 払い戻しテーブルを抽出する
def extract_payout_table_from_soup(soup):
    # tableタグを抜き出す
    tables = soup.findAll("table", attrs={"class":"pay_table_01","summary":"払い戻し"})

    index_names = []
    parsed_rows = []
    for table in tables:
        for row in table.findAll("tr"):
            index_names += [th.get_text() for th in row.findAll("th")]

            parsed_row = []
            for cell in row.findAll("td"):
                for br in cell.findAll("br"):
                    br.replace_with(";")
                text = cell.get_text()
                parsed_row.append(text)
            parsed_rows.append(parsed_row)

    # データフレームに埋め込む
    extracted = pd.DataFrame(parsed_rows, index = index_names)

    return extracted

# レースIDからレース結果を取り出す
def scrape_race_info(requestWrapper: RequestWrapper, race_id: int):
    # リクエスト
    url = f"https://db.netkeiba.com/race/{race_id}"
    html = requestWrapper.get(url)
    html.encoding = "EUC-JP"
    soup = BeautifulSoup(html.text, "html.parser")

    # レース情報のパース
    extracted_race_info = extract_race_info_from_soup(soup)

    # 結果テーブルのパース
    extracted_result_table = extract_result_table_from_soup(soup)

    # 払い戻しテーブルのパース
    extracted_payout_table = extract_payout_table_from_soup(soup)

    return extracted_race_info, extracted_result_table, extracted_payout_table

requestWrapper = RequestWrapper(request_interval_milli=1000)
result = scrape_race_info(requestWrapper, 202404030303)

print(result[0])
print(result[1].to_csv())
print(result[2].to_csv())

['3R', 'ダ左1200m', '天候 : 晴', 'ダート : 良', '発走 : 11:15', '2024年8月17日', '3回新潟3日目', '2歳未勝利', '(混)[指](馬齢)']