#!/usr/bin/env python3 """ProEyeKyuu 年別試合結果CSVを取得して games テーブルへ投入する。""" from __future__ import annotations import argparse import csv import io import os import re import sys import urllib.error import urllib.request from collections import defaultdict from datetime import datetime from typing import Any sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from collectors.db_util import connect # noqa: E402 from config import PROEYEKYUU_GAME_CSV_URL # noqa: E402 # 現行12球団（ProEyeKyuu表記 → リーグ） CURRENT_TEAMS: dict[str, str] = { "Yomiuri Giants": "CL", "Hanshin Tigers": "CL", "Yokohama DeNA Baystars": "CL", "Hiroshima Toyo Carp": "CL", "Chunichi Dragons": "CL", "Tokyo Yakult Swallows": "CL", "Fukuoka Softbank Hawks": "PL", "Orix Buffaloes": "PL", "Chiba Lotte Marines": "PL", "Tohoku Rakuten Golden Eagles": "PL", "Saitama Seibu Lions": "PL", "Hokkaido Nippon-Ham Fighters": "PL", } GAME_LENGTH_RE = re.compile(r"(?:(\d+):)?(\d+):(\d+)") def _fetch_csv(year: int, timeout: int = 60) -> str: url = PROEYEKYUU_GAME_CSV_URL.format(year=year) req = urllib.request.Request( url, headers={"User-Agent": "BaseballAI/0.1 (+https://baseball.maspis.com)"}, ) with urllib.request.urlopen(req, timeout=timeout) as resp: return resp.read().decode("utf-8-sig", errors="replace") def _parse_game_length(raw: str) -> int | None: raw = (raw or "").strip() if not raw: return None m = GAME_LENGTH_RE.match(raw) if not m: return None h, mi, _sec = m.groups() hours = int(h or 0) minutes = int(mi or 0) return hours * 60 + minutes def _parse_game_start(raw: str) -> str | None: raw = (raw or "").strip() if not raw: return None try: return datetime.strptime(raw, "%H:%M").strftime("%H:%M:%S") except ValueError: return None def _parse_int(raw: str) -> int | None: raw = (raw or "").strip().replace(",", "") if raw == "": return None return int(raw) def _ensure_team(cur, cache: dict[str, int], name: str) -> int: name = name.strip() if name in cache: return cache[name] league = CURRENT_TEAMS.get(name, "OTHER") cur.execute( "INSERT INTO teams (name, league, active) VALUES (%s, %s, %s) " "ON DUPLICATE KEY UPDATE league = IF(league = 'OTHER', VALUES(league), league)", (name, league, 1 if name in CURRENT_TEAMS else 0), ) cur.execute("SELECT id FROM teams WHERE name = %s", (name,)) row = cur.fetchone() assert row cache[name] = int(row["id"]) return cache[name] def _rows_to_games(rows: list[dict[str, str]]) -> list[dict[str, Any]]: by_gid: dict[str, list[dict[str, str]]] = defaultdict(list) for row in rows: gid = (row.get("GameID") or "").strip() if not gid: continue by_gid[gid].append(row) games: list[dict[str, Any]] = [] for gid, parts in by_gid.items(): if len(parts) != 2: continue home_row = next((p for p in parts if p.get("HomeAway") == "Home"), None) away_row = next((p for p in parts if p.get("HomeAway") == "Away"), None) if not home_row or not away_row: continue try: season = int(home_row.get("Season") or away_row.get("Season") or 0) game_date = datetime.strptime(home_row["Date"], "%Y-%m-%d").date() except (ValueError, KeyError): continue games.append( { "game_id": gid, "season": season, "game_date": game_date, "game_type": (home_row.get("GameType") or "Reg Season").strip(), "home_team": home_row["Team"].strip(), "away_team": away_row["Team"].strip(), "home_score": int(home_row["Score"]), "away_score": int(away_row["Score"]), "home_hits": _parse_int(home_row.get("H", "")), "away_hits": _parse_int(away_row.get("H", "")), "home_errors": _parse_int(home_row.get("E", "")), "away_errors": _parse_int(away_row.get("E", "")), "ballpark": (home_row.get("Ballpark") or "").strip() or None, "audience": _parse_int(home_row.get("Audience", "")), "game_start": _parse_game_start(home_row.get("Game Start", "")), "game_length_min": _parse_game_length(home_row.get("Game Length", "")), } ) return games def import_year(year: int, *, reg_season_only: bool = True) -> dict[str, int]: text = _fetch_csv(year) reader = csv.DictReader(io.StringIO(text)) rows = list(reader) games = _rows_to_games(rows) if reg_season_only: games = [g for g in games if g["game_type"] == "Reg Season"] team_cache: dict[str, int] = {} inserted = 0 updated = 0 with connect() as conn: with conn.cursor() as cur: for g in games: home_id = _ensure_team(cur, team_cache, g["home_team"]) away_id = _ensure_team(cur, team_cache, g["away_team"]) cur.execute( """ INSERT INTO games ( game_id, season, game_date, game_type, home_team_id, away_team_id, home_score, away_score, home_hits, away_hits, home_errors, away_errors, ballpark, audience, game_start, game_length_min ) VALUES ( %s,%s,%s,%s, %s,%s, %s,%s, %s,%s,%s,%s, %s,%s,%s,%s ) ON DUPLICATE KEY UPDATE season = VALUES(season), game_date = VALUES(game_date), game_type = VALUES(game_type), home_team_id = VALUES(home_team_id), away_team_id = VALUES(away_team_id), home_score = VALUES(home_score), away_score = VALUES(away_score), home_hits = VALUES(home_hits), away_hits = VALUES(away_hits), home_errors = VALUES(home_errors), away_errors = VALUES(away_errors), ballpark = VALUES(ballpark), audience = VALUES(audience), game_start = VALUES(game_start), game_length_min = VALUES(game_length_min) """, ( g["game_id"], g["season"], g["game_date"], g["game_type"], home_id, away_id, g["home_score"], g["away_score"], g["home_hits"], g["away_hits"], g["home_errors"], g["away_errors"], g["ballpark"], g["audience"], g["game_start"], g["game_length_min"], ), ) if cur.rowcount == 1: inserted += 1 elif cur.rowcount == 2: updated += 1 conn.commit() return { "year": year, "csv_rows": len(rows), "games": len(games), "inserted": inserted, "updated": updated, } def main() -> None: p = argparse.ArgumentParser(description="ProEyeKyuu 試合CSVインポート") p.add_argument("--from-year", type=int, default=2016) p.add_argument("--to-year", type=int, default=2025) p.add_argument("--year", type=int, help="単年指定（from/toより優先）") p.add_argument("--include-playoffs", action="store_true") args = p.parse_args() if args.year: years = [args.year] else: years = list(range(args.from_year, args.to_year + 1)) for y in years: try: stats = import_year(y, reg_season_only=not args.include_playoffs) print( f"{y}: games={stats['games']} inserted={stats['inserted']} " f"updated={stats['updated']} csv_rows={stats['csv_rows']}" ) except urllib.error.HTTPError as e: print(f"{y}: HTTP {e.code} skip", file=sys.stderr) except Exception as e: print(f"{y}: ERROR {e}", file=sys.stderr) if __name__ == "__main__": main()