#!/usr/bin/env python3.11 """ jv_raw_records をアプリ参照向けの jv_app_records に展開する。 目的: - 個別テーブル化していない record_type も API で扱えるようにする - レコードの「最新状態」を論理キー単位で保持する """ from __future__ import annotations import argparse import hashlib import json import os import sys import time from datetime import date from typing import Any ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, ROOT) from collectors.db_util import connect # noqa: E402 def _s(v: Any) -> str | None: if v is None: return None t = str(v).strip() return t or None def _i(v: Any) -> int | None: s = _s(v) if not s: return None try: return int(s) except ValueError: return None def _yyyymmdd_to_date(v: Any) -> date | None: s = _s(v) if not s: return None digits = "".join(ch for ch in s if ch.isdigit()) if len(digits) != 8: return None y = int(digits[:4]) m = int(digits[4:6]) d = int(digits[6:8]) try: return date(y, m, d) except ValueError: return None def _meet_date(payload: dict[str, Any]) -> date | None: year = _i(payload.get("meet_year")) mmdd = _s(payload.get("meet_date")) if not year or not mmdd: return None mmdd_digits = "".join(ch for ch in mmdd if ch.isdigit()) if len(mmdd_digits) != 4: return None try: return date(year, int(mmdd_digits[:2]), int(mmdd_digits[2:])) except ValueError: return None def _logical_key(dataspec: str, record_type: str, raw_id: int, payload: dict[str, Any]) -> str: parts: list[str] = [] y = _s(payload.get("meet_year")) md = _s(payload.get("meet_date")) jc = _s(payload.get("racecourse_code")) rn = _s(payload.get("race_number")) if y and md and jc and rn: parts.append(f"R:{y}{md}:{jc.zfill(2)}:{rn.zfill(2)}") if _s(payload.get("pedigree_reg_num")): parts.append(f"P:{_s(payload.get('pedigree_reg_num'))}") if _s(payload.get("horse_number")): parts.append(f"HN:{_s(payload.get('horse_number'))}") if _s(payload.get("data_code")): parts.append(f"DC:{_s(payload.get('data_code'))}") if _s(payload.get("data_creation_date")): parts.append(f"CD:{_s(payload.get('data_creation_date'))}") if _s(payload.get("announcement_datetime")): parts.append(f"AD:{_s(payload.get('announcement_datetime'))}") if not parts: # キー抽出不可のものは raw_id 単位で保持。 parts.append(f"RAW:{raw_id}") return "|".join(parts)[:255] def _log_start(cur, from_id: int, to_id: int | None) -> int: cur.execute( """ INSERT INTO import_logs (source, started_at, status, message) VALUES (%s, NOW(), 'running', %s) """, ("jv_app_materialize", f"from_id={from_id} to_id={to_id if to_id is not None else 'max'}"), ) return int(cur.lastrowid) def _log_finish(cur, log_id: int, status: str, rows: int, msg: str | None) -> None: cur.execute( """ UPDATE import_logs SET finished_at = NOW(), status = %s, rows_inserted = %s, message = %s WHERE id = %s """, (status, rows, msg, log_id), ) def _execute_with_retry(cur, sql: str, params: tuple[Any, ...], *, max_retries: int = 6, retry_sec: float = 0.6) -> None: """ Lock wait timeout / deadlock の場合のみ短時間リトライする。 """ for attempt in range(max_retries + 1): try: cur.execute(sql, params) return except Exception as e: code = None if hasattr(e, "args") and e.args: code = e.args[0] if code not in (1205, 1213) or attempt >= max_retries: raise time.sleep(retry_sec * (attempt + 1)) def run(from_id: int, to_id: int | None, batch_size: int) -> None: conn = connect() conn.autocommit(False) cur = conn.cursor() log_id = _log_start(cur, from_id, to_id) conn.commit() total = 0 current = from_id try: if to_id is None: cur.execute("SELECT MAX(id) AS mx FROM jv_raw_records") to_id = int((cur.fetchone() or {}).get("mx") or 0) while current < to_id: cur.execute( """ SELECT id, fetched_at, dataspec, record_type, key_text, payload_json FROM jv_raw_records WHERE id > %s AND id <= %s ORDER BY id LIMIT %s """, (current, to_id, batch_size), ) rows = cur.fetchall() if not rows: break current = int(rows[-1]["id"]) for r in rows: raw_id = int(r["id"]) dataspec = str(r["dataspec"] or "")[:16] record_type = str(r["record_type"] or "")[:8] key_text = _s(r.get("key_text")) try: payload = json.loads(r["payload_json"]) except Exception: payload = {} if not isinstance(payload, dict): payload = {"_value": payload} logical_key = _logical_key(dataspec, record_type, raw_id, payload) key_hash = hashlib.sha1(f"{dataspec}|{record_type}|{logical_key}".encode("utf-8")).hexdigest() _execute_with_retry( cur, """ INSERT INTO jv_app_records (dataspec, record_type, key_hash, logical_key, key_text, meet_year, meet_date, racecourse_code, race_number, pedigree_reg_num, data_code, data_creation_date, source_raw_id, fetched_at, payload_json) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE source_raw_id = GREATEST(source_raw_id, VALUES(source_raw_id)), key_text = COALESCE(VALUES(key_text), key_text), meet_year = COALESCE(VALUES(meet_year), meet_year), meet_date = COALESCE(VALUES(meet_date), meet_date), racecourse_code = COALESCE(VALUES(racecourse_code), racecourse_code), race_number = COALESCE(VALUES(race_number), race_number), pedigree_reg_num = COALESCE(VALUES(pedigree_reg_num), pedigree_reg_num), data_code = COALESCE(VALUES(data_code), data_code), data_creation_date = COALESCE(VALUES(data_creation_date), data_creation_date), fetched_at = GREATEST(fetched_at, VALUES(fetched_at)), payload_json = VALUES(payload_json) """, ( dataspec, record_type, key_hash, logical_key, key_text, _i(payload.get("meet_year")), _meet_date(payload), _s(payload.get("racecourse_code")), _i(payload.get("race_number")), _s(payload.get("pedigree_reg_num")), _s(payload.get("data_code")), _yyyymmdd_to_date(payload.get("data_creation_date")), raw_id, r["fetched_at"], json.dumps(payload, ensure_ascii=False), ), ) total += 1 conn.commit() print(f"processed<=id:{current} total:{total}", flush=True) _log_finish(cur, log_id, "success", total, json.dumps({"last_id": current, "to_id": to_id}, ensure_ascii=False)) conn.commit() print(f"done total={total} last_id={current}") except Exception as e: conn.rollback() try: _log_finish(cur, log_id, "fail", total, str(e)) conn.commit() except Exception: pass raise finally: cur.close() conn.close() def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--from-id", type=int, default=0) ap.add_argument("--to-id", type=int, default=None) ap.add_argument("--batch-size", type=int, default=5000) args = ap.parse_args() run(from_id=args.from_id, to_id=args.to_id, batch_size=args.batch_size) if __name__ == "__main__": main()