【競馬AI開発#56】データ収集を「再実行するだけ」で復旧できる仕組みに作り直す

planner.show_status()

=== 2020-01 〜 2025-12 のデータ取得状況 ===

[calendar] 開催カレンダー (kaisai_calendar.csv)
  チェック内容: 期間内の各月について kaisai_calendar.csv に該当月の race_id が登録されているか
    対象: 72 ヶ月 (2020-01 〜 2025-12)
    登録済月数: 72 / 72 (100.00%)
    期間内 race_id: 20,733 件

[race] レースページからのデータ取得状況 (results / race_info / return_tables)
  チェック内容: 期間内の全 race_id が各 csv に登録されているか
    対象 race_id: 20,733 件
    results.csv: 20,733 / 20,733 (100.00%)
    race_info.csv: 20,733 / 20,733 (100.00%)
    return_tables.csv: 20,733 / 20,733 (100.00%)

[horse] 馬の過去成績 (horse_results.csv)
  チェック内容: 期間内に出走する horse_id が最新まで取得済か
    対象 horse_id: 35,891 種
    最新まで取得済: 35,878 / 35,891 (99.96%)
    古い (要再取得): 2 件
    未取得: 11 件

[ped] 血統 (peds.csv)
  チェック内容: 期間内に出走する horse_id が peds.csv に登録済か
    対象 horse_id: 35,891 種
    取得済: 35,891 / 35,891 (100.00%)

[leading] リーディング (jockey / trainer / sire)
  チェック内容: 期間内の各年×タイプ について page_id が rawdf に登録済か
    対象年: 6 年 (2019 〜 2024) ※後段で前年データとして使用
    jockey_leading.csv: 6 / 6 年
    trainer_leading.csv: 6 / 6 年
    sire_leading.csv: 6 / 6 年

[horse plan]
  HTML 取得予定 horse_id: 13 件 (古い 2 + 未取得 11)
  horse_results.csv create 予定: 13 件

# 途中から再開する例。進捗を見て途切れた所からリストを切り出す。
html_paths_horse = scraping.scrape_html_horse(
    horse_id_list=horse_id_list[300:], skip=False
)

# 取得対象とする期間をyyyy-mm形式で指定する。
FROM_YM = "2025-09"
TO_YM = "2025-09"
planner = CollectPlanner(FROM_YM, TO_YM)

# 開催日一覧の取得
kaisai_date_list = scraping.scrape_kaisai_date(from_=FROM_YM, to_=TO_YM)

# race_id 一覧を取得し、kaisai_calendar.csv に追記
kaisai_calendar = scraping.scrape_race_id_list(kaisai_date_list)
create_rawdf.update_rawdf(
    kaisai_calendar, key="kaisai_date", save_filename="kaisai_calendar.csv"
)
kaisai_calendar

    kaisai_date       race_id
0      20250906  202506040101
1      20250906  202506040102
2      20250906  202506040103
3      20250906  202506040104
4      20250906  202506040105
..          ...           ...
235    20250928  202509040908
236    20250928  202509040909
237    20250928  202509040910
238    20250928  202509040911
239    20250928  202509040912

[240 rows x 2 columns]

planner.show_status()

=== 2025-09 〜 2025-09 のデータ取得状況 ===

[calendar] 開催カレンダー (kaisai_calendar.csv)
  チェック内容: 期間内の各月について kaisai_calendar.csv に該当月の race_id が登録されているか
    対象: 1 ヶ月 (2025-09 〜 2025-09)
    登録済月数: 1 / 1 (100.00%)
    期間内 race_id: 240 件

[race] レースページからのデータ取得状況 (results / race_info / return_tables)
  チェック内容: 期間内の全 race_id が各 csv に登録されているか
    対象 race_id: 240 件
    results.csv: 未作成
    race_info.csv: 未作成
    return_tables.csv: 未作成
  → そのまま「4. race ページからのデータ取得」のセルから順に実行してください

planner.plan_race()

[race plan]
  対象期間: 2025-09 〜 2025-09
  対象 race_id (kaisai_calendar から抽出): 240 件
  HTML スクレイピング予定: 240 件
  results.csv create 予定: 240 件
  race_info.csv create 予定: 240 件
  return_tables.csv create 予定: 240 件

# race ページの HTML を取得（既存はスキップ）
_ = scraping.scrape_html_race(race_id_list=planner.race_ids_to_scrape)

# レース結果テーブルの作成・更新
results = create_rawdf.create_results(
    race_id_list=planner.race_ids_to_create_results
)

# レース情報テーブルの作成・更新
race_info = create_rawdf.create_race_info(
    race_id_list=planner.race_ids_to_create_race_info
)

# 払い戻しテーブルの作成・更新
return_tables = create_rawdf.create_return_tables(
    race_id_list=planner.race_ids_to_create_return_tables
)

planner.plan_horse()

[horse plan]
  HTML 取得予定 horse_id: 249 件 (古い 0 + 未取得 249)
  horse_results.csv create 予定: 249 件

# stale な馬は HTML を再取得する必要があるので skip=False
_ = scraping.scrape_html_horse(
    horse_id_list=planner.horse_ids_to_update, skip=False
)

# 馬の過去成績テーブルの作成
horse_results = create_rawdf.create_horse_results(
    horse_id_list=planner.horse_ids_to_update
)

planner.plan_ped()

[ped plan]
  HTML スクレイピング予定: 249 件
  peds.csv create 予定: 249 件

_ = scraping.scrape_html_ped(horse_id_list=planner.horse_ids_to_scrape_peds)

peds = create_rawdf.create_peds(
    horse_id_list=planner.horse_ids_to_create_peds
)

planner.plan_jockey_leading()

[jockey leading plan]
  scrape 予定 year: 1 年 [2024]
  jockey_leading.csv: html_dir 配下の全 .bin から再生成

for year in planner.jockey_years_to_scrape:
    _ = scraping.scrape_html_leading(leading_type="jockey", year=year)

scrape_html_leading[jockey, 2024]: 取得 4 件, スキップ 0 件

jockey_leading = create_rawdf.create_jockey_leading()

=== 2025-09 〜 2025-09 のデータ取得状況 ===

[calendar] 登録済月数: 1 / 1 (100.00%) / 期間内 race_id: 240 件
[race] results.csv: 240 / 240 (100.00%)
       race_info.csv: 240 / 240 (100.00%)
       return_tables.csv: 240 / 240 (100.00%)
[horse] 最新まで取得済: 2,784 / 2,784 (100.00%) / 古い 0 件 / 未取得 0 件
[ped] 取得済: 2,784 / 2,784 (100.00%)
[leading] jockey/trainer/sire 全 1/1 年

[race plan]   HTML スクレイピング予定: 0 件 / create 予定: 0 件
[horse plan]  HTML 取得予定 horse_id: 0 件 (古い 0 + 未取得 0)
[ped plan]    HTML スクレイピング予定: 0 件 / peds.csv create 予定: 0 件
[jockey leading plan]  scrape 予定 year: 0 年 [] / jockey_leading.csv create 予定: 0 件
[trainer leading plan] scrape 予定 year: 0 年 [] / trainer_leading.csv create 予定: 0 件
[sire leading plan]    scrape 予定 year: 0 年 [] / sire_leading.csv create 予定: 0 件

=== 2025-08 〜 2025-09 のデータ取得状況 ===

[calendar] 開催カレンダー (kaisai_calendar.csv)
    対象: 2 ヶ月 (2025-08 〜 2025-09)
    登録済月数: 2 / 2 (100.00%)
    期間内 race_id: 600 件

[race] レースページからのデータ取得状況 (results / race_info / return_tables)
    対象 race_id: 600 件
    results.csv: 240 / 600 (40.00%)
    race_info.csv: 240 / 600 (40.00%)
    return_tables.csv: 240 / 600 (40.00%)
  → そのまま「4. race ページからのデータ取得」のセルから順に実行してください

[race plan]
  対象 race_id (kaisai_calendar から抽出): 600 件
  HTML スクレイピング予定: 360 件
  results.csv create 予定: 360 件
  race_info.csv create 予定: 360 件
  return_tables.csv create 予定: 360 件

[horse plan]
  HTML 取得予定 horse_id: 2,647 件 (古い 0 + 未取得 2,647)
  horse_results.csv create 予定: 2,647 件

[jockey leading plan]   scrape 予定 year: 0 年 [] / jockey_leading.csv create 予定: 0 件
[trainer leading plan]  scrape 予定 year: 0 年 [] / trainer_leading.csv create 予定: 0 件
[sire leading plan]     scrape 予定 year: 0 年 [] / sire_leading.csv create 予定: 0 件

[horse] 馬の過去成績 (horse_results.csv)
  チェック内容: 期間内に出走する horse_id が最新まで取得済か
    対象 horse_id: 2,784 種
    最新まで取得済: 2,779 / 2,784 (99.82%)
    古い (要再取得): 5 件
    未取得: 0 件

[horse plan]
  HTML 取得予定 horse_id: 5 件 (古い 5 + 未取得 0)
  horse_results.csv create 予定: 5 件

分類          条件                                        次の処理
──────────────────────────────────────────────────────────────
up_to_date    horse_results.csv の最新日 ≧ 出走日         再取得不要
stale         horse_results.csv の最新日 < 出走日         再scrape
missing       horse_results.csv に未登録                 新規 scrape

[race plan]
  対象 race_id (kaisai_calendar から抽出): 240 件
  HTML スクレイピング予定: 120 件
  results.csv create 予定: 240 件
  race_info.csv create 予定: 240 件
  return_tables.csv create 予定: 240 件

# 残り100件で途切れてしまい、再度上から実行したケース
[horse] 馬の過去成績 (horse_results.csv)
    最新まで取得済: 2,684 / 2,784 (96.41%)
    古い (要再取得): 0 件
    未取得: 100 件

[horse plan]
  HTML 取得予定 horse_id: 100 件 (古い 0 + 未取得 100)
  horse_results.csv create 予定: 100 件

# race ページの HTML を取得（既存はスキップ）
_ = scraping.scrape_html_race(race_id_list=planner.race_ids_to_scrape)
# レース結果テーブルの作成・更新
results = create_rawdf.create_results(
    race_id_list=planner.race_ids_to_create_results
)

変数名                                対応する処理
──────────────────────────────────────────────────────────────
race_ids_in_period                    期間内 race_id 一覧
race_ids_to_scrape                    HTML 未取得の race_id
race_ids_to_create_results            results.csv 未登録の race_id
race_ids_to_create_race_info          race_info.csv 未登録の race_id
race_ids_to_create_return_tables      return_tables.csv 未登録の race_id
horse_ids_in_period                   期間内に出走した horse_id 一覧
horse_ids_to_update                   古い + 未取得の horse_id
horse_ids_to_create_peds              peds.csv 未登録の horse_id
horse_ids_to_scrape_peds              上記のうち ped HTML が未取得の horse_id
{jockey,trainer,sire}_years_to_scrape 各リーディングの scrape 対象 year

旧:
    html_paths = scraping.scrape_html_race(race_id_list)
    results = create_rawdf.create_results(html_path_list=html_paths)

新:
    scraping.scrape_html_race(race_id_list=planner.race_ids_to_scrape)
    results = create_rawdf.create_results(
        race_id_list=planner.race_ids_to_create_results
    )

def scrape_race_id_list(
    kaisai_date_list: list[str],
    sleep_time: float = 1.0,
    skip: bool = True,
    kaisai_calendar_path: Path = CommonPaths.rawdf / "kaisai_calendar.csv",
) -> pd.DataFrame:
    ...

# 旧来の呼び出し (引き続き動作する)
scraping.scrape_html_leading(leading_type="jockey", year=2024, pages=range(1, 6))

# 新しい標準
for year in planner.jockey_years_to_scrape:
    scraping.scrape_html_leading(leading_type="jockey", year=year)

import pandas as pd

from .paths import CommonPaths
from .scraping import has_next_leading_page


class CollectPlanner:
    """
    指定期間 (from_ym 〜 to_ym) のデータ取得を計画する Planner。

    - `show_status()`: 現在のデータ取得状況をサマリ表示する。
    - `plan_xxx()`: 各セクションで「これからスクレイピング・rawdf作成すべき ID 一覧」を確定し、
      インスタンス変数に格納する。
    """

    def __init__(
        self,
        from_ym: str,
        to_ym: str,
        kaisai_calendar_filename: str = "kaisai_calendar.csv",
    ) -> None:
        self.from_ym = from_ym
        self.to_ym = to_ym
        self.period_start = pd.Timestamp(f"{from_ym}-01")
        self.period_end = pd.Timestamp(f"{to_ym}-01") + pd.DateOffset(months=1)
        self.target_months = [str(p) for p in pd.period_range(from_ym, to_ym, freq="M")]
        # リーディングデータは前年までのデータを使用するため n-1 まで
        self.leading_years = list(range(int(from_ym[:4]) - 1, int(to_ym[:4])))
        self.kaisai_calendar_path = CommonPaths.rawdf / kaisai_calendar_filename

インスタンス変数         意味
──────────────────────────────────────────────────────────────
period_start            期間の開始 (例: 2025-09-01 00:00:00)
period_end              期間の終了 (排他的, 例: 2025-10-01 00:00:00)
target_months           期間内の月リスト (例: ["2025-09"])
leading_years           リーディング対象年リスト ([n-1, ..., n-1])
                        ※ from_ym[:4]-1 〜 to_ym[:4]-1
kaisai_calendar_path    開催カレンダーCSVのパス

公開メソッド             内部メソッド
──────────────────────────────────────────────────────────────
show_status           _status_calendar / _status_race / _status_horse /
                      _status_ped / _status_leading
plan_race             _read_race_ids_in_period / _ids_to_create
plan_horse            _read_horse_ids_in_period / _classify_horses
plan_ped              _ids_to_create
plan_jockey_leading   _find_years_to_scrape
plan_trainer_leading  _find_years_to_scrape
plan_sire_leading     _find_years_to_scrape

def _classify_horses(self, target_horse_ids: list[str]) -> dict[str, list[str]]:
    """
    target_horse_ids を以下の 3 つに分類する。
    - up_to_date: horse_results.csv の最終日付が results.csv での最新出走日以上 (再取得不要)
    - stale: horse_results.csv に登録はあるが、results.csv での最新出走日のほうが新しい (再取得が必要)
    - missing: horse_results.csv に未登録 (新規取得が必要)
    """
    targets = list(dict.fromkeys(str(horse_id) for horse_id in target_horse_ids))
    horse_results_path = CommonPaths.rawdf / "horse_results.csv"
    if not horse_results_path.exists():
        return {"up_to_date": [], "stale": [], "missing": targets}
    horse_results = pd.read_csv(
        horse_results_path,
        sep="\t",
        usecols=["horse_id", "日付"],
        dtype={"horse_id": str},
    )
    horse_results_latest = (
        pd.to_datetime(horse_results["日付"], errors="coerce")
        .groupby(horse_results["horse_id"])
        .max()
    )

    results = pd.read_csv(
        CommonPaths.rawdf / "results.csv",
        sep="\t",
        usecols=["race_id", "horse_id"],
        dtype={"race_id": str, "horse_id": str},
    )
    kaisai_calendar = pd.read_csv(
        self.kaisai_calendar_path,
        sep="\t",
        dtype={"race_id": str, "kaisai_date": str},
    )
    kaisai_calendar["kaisai_date"] = pd.to_datetime(
        kaisai_calendar["kaisai_date"], format="%Y%m%d"
    )
    results_latest = (
        results.merge(kaisai_calendar, on="race_id", how="left")
        .groupby("horse_id")["kaisai_date"]
        .max()
    )

    classified: dict[str, list[str]] = {
        "up_to_date": [],
        "stale": [],
        "missing": [],
    }
    for horse_id in targets:
        horse_results_date = horse_results_latest.get(horse_id)
        if horse_results_date is None or pd.isna(horse_results_date):
            classified["missing"].append(horse_id)
            continue
        results_date = results_latest.get(horse_id)
        if results_date > horse_results_date:
            classified["stale"].append(horse_id)
        else:
            classified["up_to_date"].append(horse_id)
    return classified

def _find_years_to_scrape(
    self, leading_type: str, max_pages: int = 15
) -> list[int]:
    """
    self.leading_years のうち scrape が未完了の year を返す。

    判定 (year × leading_type):
      - .bin が無い → 未着手なので scrape 対象
      - 最終ページ番号が max_pages 以上 → 上限到達済みとみなして対象外
        （これ以上 scrape しても scrape_html_leading の auto_paginate で
        同じ上限に到達するだけなので無駄）
      - 最終ページに「次」リンクが残る → 未完なので scrape 対象
      - 「次」リンクが無ければ完了済みで対象外

    max_pages のデフォルトは scraping.scrape_html_leading の同名引数と
    一致させること。
    """
    leading_dir = CommonPaths.html_leading / leading_type
    years_to_scrape: list[int] = []
    for year in self.leading_years:
        if not leading_dir.exists():
            years_to_scrape.append(year)
            continue
        bin_files = sorted(
            leading_dir.glob(f"{year}_*.bin"),
            key=lambda p: int(p.stem.split("_")[1]),
        )
        if not bin_files:
            years_to_scrape.append(year)
            continue
        last_page_num = int(bin_files[-1].stem.split("_")[1])
        if last_page_num >= max_pages:
            continue
        with open(bin_files[-1], "rb") as f:
            last_html = f.read()
        if has_next_leading_page(last_html):
            years_to_scrape.append(year)
    return years_to_scrape

段階  条件                                  結果
──────────────────────────────────────────────────────────────
1     .bin が 1 個も無い year                未着手 → scrape 対象
2     最終ページ番号 ≧ max_pages (=15)       上限到達済 → 対象外
3     最終ページの HTML に「次」リンクあり      未完 → scrape 対象
4     「次」リンク無し                        完了済 → 対象外

def has_next_leading_page(html: bytes) -> bool:
    """leadingページのHTMLに「次」リンクが存在するかを判定する。"""
    soup = BeautifulSoup(html, "lxml")
    for a in soup.find_all("a"):
        # 騎手名などにも「次」を含む文字列があるので、完全一致のリンクを探す
        if a.get_text(strip=True) == "次":
            return True
    return False

def scrape_html_leading(
    leading_type: str,
    year: int,
    pages: list[int] | range | None = None,
    save_dir: Path = CommonPaths.html_leading,
    skip: bool = True,
    sleep_time: float = 1.0,
    max_pages: int = 15,
) -> list[Path]:
    """
    リーディングページのhtmlをスクレイピングしてsave_dirに保存する関数。

    pages=None を指定すると、HTMLの「次」リンクが無くなるまで自動的に
    ページネーションを進める（上限は max_pages）。明示的にページ範囲を
    指定したい場合は list[int] や range で渡す（max_pages は無視される）。
    """
    updated_html_path_list = []
    save_dir = save_dir / leading_type
    save_dir.mkdir(parents=True, exist_ok=True)

    auto_paginate = pages is None
    if auto_paginate:
        page_iter = list(range(1, max_pages + 1))
    else:
        page_iter = list(pages)

    skipped = 0
    last_html: bytes | None = None
    hit_limit = auto_paginate
    for page in tqdm(page_iter):
        filepath = save_dir / f"{year}_{str(page).zfill(2)}.bin"
        if skip and filepath.is_file():
            skipped += 1
            if auto_paginate:
                # 次へリンクの有無を判定するため既存HTMLを読む
                with open(filepath, "rb") as f:
                    last_html = f.read()
        else:
            if leading_type in ["jockey", "trainer", "owner", "breeder"]:
                url = f"https://db.netkeiba.com/{leading_type}/{leading_type}_leading_jra.html&year={year}&page={page}"
            elif leading_type in ["sire", "bms"]:
                url = f"https://db.netkeiba.com/horse/{leading_type}_leading_jra.html&year={year}&page={page}"
            else:
                raise ValueError(f"不正な leading_type: {leading_type}")
            request = Request(url, headers={"User-Agent": random.choice(USER_AGENTS)})
            html = urlopen(request).read()
            time.sleep(sleep_time)
            with open(filepath, "wb") as f:
                f.write(html)
            updated_html_path_list.append(filepath)
            last_html = html

        if auto_paginate:
            if last_html is None or not has_next_leading_page(last_html):
                hit_limit = False
                break

    print(
        f"scrape_html_leading[{leading_type}, {year}]: "
        f"取得 {len(updated_html_path_list)} 件, スキップ {skipped} 件"
    )
    # 2024 × sire は netkeiba 側ページ構造の不具合により最終ページに「次」リンクが
    # 残り続けるため、auto_paginate で必ず上限到達となるが原因はサイト側で確認済み。
    # WARNING を抑制する
    is_known_buggy = leading_type == "sire" and year == 2024
    if auto_paginate and hit_limit and not is_known_buggy:
        print(
            f"[WARNING] scrape_html_leading[{leading_type}, {year}]: "
            f"自動ページネーションが上限 {max_pages} に到達しました。"
            "データが欠落している可能性があります。"
        )
    return updated_html_path_list