#!/usr/bin/env python3 """ jrvltsql が Windows で作った SQLite keiba.db (NL_RA / NL_SE / NL_HR 等) を MySQL へ投入。 Manus マニュアル経路: 1) Windows: jrvltsql quickstart / jltsql fetch → data/keiba.db 2) keiba.db をサーバーへコピー 3) 本スクリプトで import 例: python3 collectors/import_jrvltsql_db.py --sqlite /tmp/keiba.db --from-year 2020 --to-year 2026 python3 collectors/import_jrvltsql_db.py --sqlite /root/keiba.db --payouts-only --from-year 2020 --to-year 2026 python3 collectors/import_jrvltsql_db.py --sqlite /root/keiba.db --odds-only --from-year 2020 --to-year 2026 """ from __future__ import annotations import argparse import os import re import sqlite3 import sys from datetime import date, time ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, ROOT) from pymysql.err import IntegrityError from collectors.db_util import connect # noqa: E402 COMMIT_EVERY_RACES = 500 JRA_JYO = {"01", "02", "03", "04", "05", "06", "07", "08", "09", "10"} # jrvltsql / Manus サンプルで出てくる列名の別名 ALIASES: dict[str, list[str]] = { "year": ["Year", "JvdYear", "year", "JvdKaisaiYear"], "month_day": ["MonthDay", "JvdMonthDay", "month_day", "KaisaiDate"], "jyo_cd": ["JyoCD", "JvdCourseCd", "JyoCd", "course_cd", "CourseCd"], "race_num": ["RaceNum", "JvdRaceNum", "race_num", "RaceNo"], "race_name": ["Hondai", "JvdRaceName", "RaceName", "race_name"], "distance": ["Kyori", "JvdDistance", "distance", "Distance"], "track_cd": ["TrackCD", "JvdTrackCd", "track_cd", "TrackCd"], "kaiji": ["Kaiji", "Kai", "kai"], "nichiji": ["Nichiji", "Nichi", "nichi"], "head_count": ["SyussoTosu", "TorokuTosu", "head_count"], "post_time": ["HassoTime", "PostTime", "post_time"], "horse_name": ["Bamei", "JvdHorseName", "HorseName", "horse_name"], "horse_num": ["Umaban", "JvdHorseNum", "HorseNum", "horse_num"], "bracket": ["Wakuban", "JvdBracketNum", "BracketNum", "bracket"], "finish": ["KakuteiJyuni", "JvdOrderOfFinish", "Jyuni", "finish_order"], "popularity": ["Ninki", "JvdPopularity", "popularity"], } def _pick(cols: set[str], key: str) -> str | None: for name in ALIASES.get(key, []): if name in cols: return name return None def _table_columns(conn: sqlite3.Connection, table: str) -> set[str]: cur = conn.execute(f"PRAGMA table_info({table})") return {row[1] for row in cur.fetchall()} def _parse_race_date(year_s: str | None, md_s: str | None) -> date | None: if not year_s or not md_s: return None y = str(year_s).strip() md = re.sub(r"\D", "", str(md_s).strip()).zfill(4) if len(y) != 4 or len(md) != 4: return None try: return date(int(y), int(md[:2]), int(md[2:])) except ValueError: return None def _parse_post_time(v: str | None) -> time | None: if not v: return None s = re.sub(r"\D", "", str(v).strip()) if len(s) < 3: return None s = s.zfill(4) try: return time(int(s[:2]), int(s[2:4])) except ValueError: return None def _surface_from_track_cd(track_cd: str | None) -> str | None: if not track_cd: return None t = str(track_cd).strip() if t.startswith("1"): return "芝" if t.startswith("2"): return "ダ" return None def _ensure_horse(cur, name: str) -> int: name = name.strip()[:64] cur.execute("SELECT id FROM horses WHERE name=%s LIMIT 1", (name,)) row = cur.fetchone() if row: return int(row["id"]) cur.execute("INSERT INTO horses (name) VALUES (%s)", (name,)) return int(cur.lastrowid) def _parse_int(v) -> int | None: if v is None or v == "": return None s = re.sub(r"[^\d]", "", str(v).strip()) if not s: return None try: return int(s) except ValueError: return None def _combination_from_number(number: str | None, legs: int) -> str | None: if not number: return None digits = "".join(ch for ch in str(number) if ch.isdigit()) if not digits: return None if legs == 1: n = int(digits) return str(n) if n > 0 else None if legs not in (2, 3): return None part_len = None for c in (2, 1): if len(digits) == c * legs: part_len = c break if part_len is None: return None parts: list[str] = [] for i in range(legs): chunk = digits[i * part_len : (i + 1) * part_len] n = int(chunk) if n <= 0: return None parts.append(str(n)) return "-".join(parts) def _combo_pair_sorted(kumi: str | None) -> str | None: combo = _combination_from_number(kumi, 2) if not combo: return None a, b = combo.split("-", 1) ia, ib = int(a), int(b) return f"{min(ia, ib)}-{max(ia, ib)}" def _source_cname(rd: date, jyo: str, kai: int | None, nichi: int | None, race_number: int) -> str: return f"jrvltsql:{rd.isoformat()}:{jyo}:{kai}:{nichi}:{race_number}" def _find_race_id(cur, rd: date, jyo: str, kai: int | None, nichi: int | None, race_number: int) -> int | None: cur.execute("SELECT id FROM races WHERE source_cname=%s LIMIT 1", (_source_cname(rd, jyo, kai, nichi, race_number),)) row = cur.fetchone() if row: return int(row["id"]) cur.execute( """ SELECT r.id FROM races r JOIN tracks t ON t.id = r.track_id WHERE r.circuit='JRA' AND r.race_date=%s AND t.code=%s AND r.race_number=%s LIMIT 1 """, (rd, jyo, race_number), ) row = cur.fetchone() return int(row["id"]) if row else None def _parse_odds(v) -> float | None: """NL_O1 の TanOdds（確定単勝オッズ）を倍率に変換。""" if v is None or v == "": return None raw = str(v).strip() try: f = float(raw) except (TypeError, ValueError): s = re.sub(r"[^\d.]", "", raw) if not s: return None try: f = float(s) except ValueError: return None if f <= 0: return None # JV 由来の整数10倍表記（583 → 58.3）。SQLite REAL の 5.8 には小数点あり if "." not in raw and f >= 10: f = f / 10.0 return round(f, 1) def _insert_payout(cur, race_id: int, bet_type: str, combo: str | None, payout: int | None, pop: int | None) -> bool: if not combo or payout is None or payout <= 0: return False if pop is not None and not (0 <= pop <= 255): pop = None cur.execute( """ INSERT INTO payouts (race_id, bet_type, combination, payout_yen, popularity) VALUES (%s,%s,%s,%s,%s) """, (race_id, bet_type, combo, payout, pop), ) return True # NL_HR 1件形式（TanUmaban 等）と HARAI 配列形式（PayTansyoUmaban1 等）の両方 HR_FLAT_SPECS: list[tuple[str, str, str, str, int, bool]] = [ ("TanUmaban", "TanPay", "TanNinki", "tansho", 1, False), ("FukuUmaban", "FukuPay", "FukuNinki", "fukusho", 1, False), ("WakuKumi", "WakuPay", "WakuNinki", "wakuren", 2, True), ("UmarenKumi", "UmarenPay", "UmarenNinki", "umaren", 2, True), ("WideKumi", "WidePay", "WideNinki", "wide", 2, True), ("UmatanKumi", "UmatanPay", "UmatanNinki", "umatan", 2, False), ("SanrenfukuKumi", "SanrenfukuPay", "SanrenfukuNinki", "sanrenpuku", 3, False), ("SanrentanKumi", "SanrentanPay", "SanrentanNinki", "sanrentan", 3, False), ] HR_ARRAY_SPECS: list[tuple[str, int, str, int, bool]] = [ ("PayTansyo", 3, "tansho", 1, False), ("PayFukusyo", 5, "fukusho", 1, False), ("PayWakuren", 3, "wakuren", 2, True), ("PayUmaren", 3, "umaren", 2, True), ("PayWide", 7, "wide", 2, True), ("PayUmatan", 6, "umatan", 2, False), ("PaySanrenpuku", 3, "sanrenpuku", 3, False), ("PaySanrentan", 6, "sanrentan", 3, False), ] def _payout_combo(kumi_val: str | None, legs: int, sort_pair: bool) -> str | None: if legs == 1: return _combination_from_number(kumi_val, 1) if sort_pair: return _combo_pair_sorted(kumi_val) return _combination_from_number(kumi_val, legs) def _import_hr_row(cur, row: sqlite3.Row, hr_cols: set[str], stats: dict) -> None: yc = _pick(hr_cols, "year") or "Year" mdc = _pick(hr_cols, "month_day") or "MonthDay" jyoc = _pick(hr_cols, "jyo_cd") or "JyoCD" rnc = _pick(hr_cols, "race_num") or "RaceNum" kaic = _pick(hr_cols, "kaiji") or "Kaiji" nicc = _pick(hr_cols, "nichiji") or "Nichiji" rd = _parse_race_date(str(row[yc]), str(row[mdc])) jyo = str(row[jyoc]).zfill(2) if not rd or jyo not in JRA_JYO: stats["hr_skip"] += 1 return try: race_number = int(row[rnc]) except (TypeError, ValueError): stats["hr_skip"] += 1 return kai = int(row[kaic]) if kaic and row[kaic] not in (None, "") else None nichi = int(row[nicc]) if nicc and row[nicc] not in (None, "") else None race_id = _find_race_id(cur, rd, jyo, kai, nichi, race_number) if not race_id: stats["hr_no_race"] += 1 return stats["hr_rows"] += 1 cur.execute("DELETE FROM payouts WHERE race_id=%s", (race_id,)) n = 0 if "TanUmaban" in hr_cols or "TanPay" in hr_cols: for k_col, p_col, pop_col, bet_type, legs, sort_pair in HR_FLAT_SPECS: if k_col not in hr_cols or p_col not in hr_cols: continue combo = _payout_combo(row[k_col], legs, sort_pair) payout = _parse_int(row[p_col]) pop = _parse_int(row[pop_col]) if pop_col in hr_cols else None if _insert_payout(cur, race_id, bet_type, combo, payout, pop): n += 1 else: for prefix, count, bet_type, legs, sort_pair in HR_ARRAY_SPECS: k_suffix = "Umaban" if legs == 1 else "Kumi" for i in range(1, count + 1): k_col = f"{prefix}{k_suffix}{i}" p_col = f"{prefix}Pay{i}" pop_col = f"{prefix}Ninki{i}" if k_col not in hr_cols or p_col not in hr_cols: continue combo = _payout_combo(row[k_col], legs, sort_pair) payout = _parse_int(row[p_col]) pop = _parse_int(row[pop_col]) if pop_col in hr_cols else None if _insert_payout(cur, race_id, bet_type, combo, payout, pop): n += 1 stats["payout_upsert"] += n if n > 0: stats["hr_races_with_payout"] += 1 def import_odds_from_sqlite( sconn: sqlite3.Connection, cur, *, from_year: int, to_year: int, dry_run: bool, stats: dict, ) -> None: """NL_O1（確定単勝オッズ）→ odds_lines / race_entries.odds_win""" tables = {r[0] for r in sconn.execute("SELECT name FROM sqlite_master WHERE type='table'")} o1_table = None for name in ("NL_O1", "RT_O1", "ODDS_TANPUKU"): if name in tables: o1_table = name break if not o1_table: stats["o1_missing_table"] = 1 return cols = _table_columns(sconn, o1_table) yc = _pick(cols, "year") or "Year" mdc = _pick(cols, "month_day") or "MonthDay" jyoc = _pick(cols, "jyo_cd") or "JyoCD" rnc = _pick(cols, "race_num") or "RaceNum" kaic = _pick(cols, "kaiji") or "Kaiji" nicc = _pick(cols, "nichiji") or "Nichiji" umc = "Umaban" if "Umaban" in cols else _pick(cols, "horse_num") todc = "TanOdds" if "TanOdds" in cols else None tink = "TanNinki" if "TanNinki" in cols else None if not all([yc, mdc, jyoc, rnc, umc, todc]): stats["o1_missing_table"] = 1 return cleared: set[int] = set() o1_sql = ( f"SELECT * FROM {o1_table} WHERE {yc} >= ? AND {yc} <= ? " f"ORDER BY {yc}, {mdc}, {jyoc}, {kaic}, {nicc}, {rnc}, {umc}" ) for row in sconn.execute(o1_sql, (str(from_year), str(to_year))): stats["o1_rows"] += 1 if dry_run: continue rd = _parse_race_date(str(row[yc]), str(row[mdc])) jyo = str(row[jyoc]).zfill(2) if not rd or jyo not in JRA_JYO: stats["o1_skip"] += 1 continue try: race_number = int(row[rnc]) umaban = int(row[umc]) except (TypeError, ValueError): stats["o1_skip"] += 1 continue if umaban <= 0: stats["o1_skip"] += 1 continue kai = int(row[kaic]) if kaic and row[kaic] not in (None, "") else None nichi = int(row[nicc]) if nicc and row[nicc] not in (None, "") else None race_id = _find_race_id(cur, rd, jyo, kai, nichi, race_number) if not race_id: stats["o1_no_race"] += 1 continue odds = _parse_odds(row[todc]) if odds is None: stats["o1_skip"] += 1 continue pop = _parse_int(row[tink]) if tink and tink in cols else None if race_id not in cleared: cur.execute("DELETE FROM odds_lines WHERE race_id=%s", (race_id,)) cleared.add(race_id) stats["o1_races"] += 1 cur.execute( """ INSERT INTO odds_lines (race_id, odds_type, combination, min_odds, max_odds, popularity, source_record_type) VALUES (%s,'tansho',%s,%s,%s,%s,'O1') ON DUPLICATE KEY UPDATE min_odds=VALUES(min_odds), max_odds=VALUES(max_odds), popularity=VALUES(popularity), source_record_type=VALUES(source_record_type) """, (race_id, str(umaban), odds, odds, pop), ) stats["odds_upsert"] += 1 cur.execute( "UPDATE race_entries SET odds_win=%s WHERE race_id=%s AND horse_number=%s", (odds, race_id, umaban), ) if stats["o1_rows"] % 5000 == 0: cur.connection.commit() def import_payouts_from_sqlite( sconn: sqlite3.Connection, cur, *, from_year: int, to_year: int, dry_run: bool, stats: dict, ) -> None: tables = {r[0] for r in sconn.execute("SELECT name FROM sqlite_master WHERE type='table'")} hr_table = None for name in ("NL_HR", "HARAI", "RT_HR"): if name in tables: hr_table = name break if not hr_table: stats["hr_missing_table"] = 1 return hr_cols = _table_columns(sconn, hr_table) yc = _pick(hr_cols, "year") or "Year" hr_sql = f"SELECT * FROM {hr_table} WHERE {yc} >= ? AND {yc} <= ?" for row in sconn.execute(hr_sql, (str(from_year), str(to_year))): if dry_run: stats["hr_rows"] += 1 continue _import_hr_row(cur, row, hr_cols, stats) if stats["hr_rows"] % COMMIT_EVERY_RACES == 0: cur.connection.commit() def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--sqlite", required=True, help="keiba.db のパス") ap.add_argument("--from-year", type=int, default=2020) ap.add_argument("--to-year", type=int, default=2026) ap.add_argument("--dry-run", action="store_true") ap.add_argument( "--payouts-only", action="store_true", help="NL_HR（払戻）のみ取込。レース・着順はスキップ", ) ap.add_argument( "--odds-only", action="store_true", help="NL_O1（確定単勝オッズ）のみ取込。レース・着順はスキップ", ) args = ap.parse_args() if not os.path.isfile(args.sqlite): raise SystemExit(f"sqlite not found: {args.sqlite}") sconn = sqlite3.connect(f"file:{args.sqlite}?mode=ro", uri=True) sconn.row_factory = sqlite3.Row tables = {r[0] for r in sconn.execute("SELECT name FROM sqlite_master WHERE type='table'")} if "NL_RA" not in tables: raise SystemExit(f"NL_RA not in {args.sqlite}. tables={sorted(tables)[:20]}") ra_cols = _table_columns(sconn, "NL_RA") se_cols = _table_columns(sconn, "NL_SE") if "NL_SE" in tables else set() yc = _pick(ra_cols, "year") mdc = _pick(ra_cols, "month_day") jyoc = _pick(ra_cols, "jyo_cd") rnc = _pick(ra_cols, "race_num") if not all([yc, mdc, jyoc, rnc]): raise SystemExit(f"NL_RA column map failed. cols sample={sorted(ra_cols)[:30]}") rname = _pick(ra_cols, "race_name") distc = _pick(ra_cols, "distance") trkc = _pick(ra_cols, "track_cd") kaic = _pick(ra_cols, "kaiji") nicc = _pick(ra_cols, "nichiji") headc = _pick(ra_cols, "head_count") postc = _pick(ra_cols, "post_time") conn = connect() conn.autocommit(False) cur = conn.cursor() cur.execute("SELECT id, code FROM tracks WHERE circuit='JRA'") track_map = {str(r["code"]).zfill(2): int(r["id"]) for r in cur.fetchall()} stats = { "ra_rows": 0, "races_upsert": 0, "se_rows": 0, "entries_upsert": 0, "results_upsert": 0, "results_skip": 0, "entries_skip": 0, "skip": 0, "hr_rows": 0, "hr_skip": 0, "hr_no_race": 0, "payout_upsert": 0, "hr_races_with_payout": 0, "hr_missing_table": 0, "o1_rows": 0, "o1_races": 0, "o1_skip": 0, "o1_no_race": 0, "odds_upsert": 0, "o1_missing_table": 0, } if args.odds_only: import_odds_from_sqlite( sconn, cur, from_year=args.from_year, to_year=args.to_year, dry_run=args.dry_run, stats=stats, ) if not args.dry_run: conn.commit() cur.close() conn.close() sconn.close() print("import_jrvltsql_db", stats, flush=True) return if args.payouts_only: import_payouts_from_sqlite( sconn, cur, from_year=args.from_year, to_year=args.to_year, dry_run=args.dry_run, stats=stats, ) if not args.dry_run: conn.commit() cur.close() conn.close() sconn.close() print("import_jrvltsql_db", stats, flush=True) return ra_sql = f"SELECT * FROM NL_RA WHERE {yc} >= ? AND {yc} <= ?" for row in sconn.execute(ra_sql, (str(args.from_year), str(args.to_year))): stats["ra_rows"] += 1 rd = _parse_race_date(row[yc], row[mdc]) jyo = str(row[jyoc]).zfill(2) if not rd or jyo not in JRA_JYO: stats["skip"] += 1 continue track_id = track_map.get(jyo) if not track_id: stats["skip"] += 1 continue try: race_number = int(row[rnc]) except (TypeError, ValueError): stats["skip"] += 1 continue kai = int(row[kaic]) if kaic and row[kaic] not in (None, "") else None nichi = int(row[nicc]) if nicc and row[nicc] not in (None, "") else None dist = int(row[distc]) if distc and str(row[distc]).strip().isdigit() else None surface = _surface_from_track_cd(str(row[trkc]) if trkc else None) rname_v = str(row[rname]).strip()[:128] if rname and row[rname] else None head_v = int(row[headc]) if headc and str(row[headc]).strip().isdigit() else None post_v = _parse_post_time(str(row[postc]) if postc else None) source_cname = _source_cname(rd, jyo, kai, nichi, race_number) if args.dry_run: stats["races_upsert"] += 1 continue cur.execute("SELECT id FROM races WHERE source_cname=%s", (source_cname,)) ex = cur.fetchone() if ex: race_id = int(ex["id"]) cur.execute( """UPDATE races SET race_date=%s, track_id=%s, kai=%s, nichi=%s, race_number=%s, race_name=%s, surface=%s, distance_m=%s, head_count=%s, start_time=%s, data_source='jrvltsql', result_fetched=1 WHERE id=%s""", (rd, track_id, kai, nichi, race_number, rname_v, surface, dist, head_v, post_v, race_id), ) else: cur.execute( """INSERT INTO races (race_date, track_id, circuit, kai, nichi, race_number, race_name, surface, distance_m, head_count, start_time, data_source, source_cname, result_fetched) VALUES (%s,%s,'JRA',%s,%s,%s,%s,%s,%s,%s,%s,'jrvltsql',%s,1)""", (rd, track_id, kai, nichi, race_number, rname_v, surface, dist, head_v, post_v, source_cname), ) race_id = int(cur.lastrowid) stats["races_upsert"] += 1 if not se_cols: continue sy = _pick(se_cols, "year") or yc smd = _pick(se_cols, "month_day") or mdc sj = _pick(se_cols, "jyo_cd") or jyoc sr = _pick(se_cols, "race_num") or rnc hnc = _pick(se_cols, "horse_name") umnc = _pick(se_cols, "horse_num") wkc = _pick(se_cols, "bracket") fnc = _pick(se_cols, "finish") popc = _pick(se_cols, "popularity") if not all([hnc, umnc]): continue se_q = f"""SELECT * FROM NL_SE WHERE {sy}=? AND {smd}=? AND {sj}=? AND {sr}=?""" md_key = row[mdc] for se in sconn.execute(se_q, (row[yc], md_key, row[jyoc], row[rnc])): stats["se_rows"] += 1 hname = str(se[hnc]).strip() if not hname: continue try: umaban = int(se[umnc]) except (TypeError, ValueError): continue horse_id = _ensure_horse(cur, hname) bracket = int(se[wkc]) if wkc and str(se[wkc]).strip().isdigit() else None pop = int(se[popc]) if popc and str(se[popc]).strip().isdigit() else None finish = int(se[fnc]) if fnc and str(se[fnc]).strip().isdigit() else None cur.execute( """SELECT id FROM race_entries WHERE race_id=%s AND horse_number=%s LIMIT 1""", (race_id, umaban), ) ent = cur.fetchone() if ent: cur.execute( """UPDATE race_entries SET horse_id=%s, bracket_number=%s, popularity=%s WHERE id=%s""", (horse_id, bracket, pop, ent["id"]), ) else: try: cur.execute( """INSERT INTO race_entries (race_id, horse_id, bracket_number, horse_number, popularity) VALUES (%s,%s,%s,%s,%s)""", (race_id, horse_id, bracket, umaban, pop), ) except IntegrityError: stats["entries_skip"] += 1 continue stats["entries_upsert"] += 1 if finish and finish > 0: try: cur.execute( """INSERT INTO race_results (race_id, horse_id, finish_position) VALUES (%s,%s,%s) ON DUPLICATE KEY UPDATE finish_position=VALUES(finish_position)""", (race_id, horse_id, finish), ) stats["results_upsert"] += 1 except IntegrityError: # 同着などで (race_id, finish_position) が衝突する行はスキップ stats["results_skip"] += 1 if not args.dry_run and stats["races_upsert"] % COMMIT_EVERY_RACES == 0: conn.commit() import_payouts_from_sqlite( sconn, cur, from_year=args.from_year, to_year=args.to_year, dry_run=args.dry_run, stats=stats, ) import_odds_from_sqlite( sconn, cur, from_year=args.from_year, to_year=args.to_year, dry_run=args.dry_run, stats=stats, ) if not args.dry_run: conn.commit() cur.close() conn.close() sconn.close() print("import_jrvltsql_db", stats, flush=True) if __name__ == "__main__": main()