#!/usr/bin/env python3 """ IPAT購入 HTTPサーバー（進捗トラッキング付き） Docker内で常駐し、HTTP APIで馬券購入を受け付ける。 SSE (Server-Sent Events) でリアルタイム進捗を配信。 API: POST /purchase - 馬券購入（JSON body） POST /dry-run - ドライラン GET /health - ヘルスチェック GET /status - 現在の状態（JSON） GET /progress - SSEで進捗をリアルタイム配信 """ import argparse import json import os import re import subprocess import sys import threading import time from collections import deque from datetime import datetime from flask import Flask, request, jsonify, Response app = Flask(__name__) SCRIPT_DIR = os.environ.get("IPAT_SCRIPT_DIR", os.path.join(os.path.dirname(os.path.abspath(__file__)), "scripts")) STORAGE_DIR = os.environ.get("STORAGE_PATH", "/data") # 進捗状態 progress = { "running": False, "dry_run": False, "started_at": None, "finished_at": None, "current_step": 0, "total_steps": 0, "current_bet": None, "phase": "idle", # idle, login, betting, confirm, done, error "message": "", "logs": deque(maxlen=200), "results": None, "bets_success": 0, "bets_failed": 0, } # SSE用のイベントキュー（複数クライアント対応） sse_clients = [] def broadcast_progress(): """全SSEクライアントに現在の進捗を送信""" data = { "running": progress["running"], "phase": progress["phase"], "current_step": progress["current_step"], "total_steps": progress["total_steps"], "current_bet": progress["current_bet"], "message": progress["message"], "percent": round(progress["current_step"] / max(progress["total_steps"], 1) * 100), "bets_success": progress["bets_success"], "bets_failed": progress["bets_failed"], "elapsed": (datetime.now() - datetime.fromisoformat(progress["started_at"])).total_seconds() if progress["started_at"] else 0, } event = f"data: {json.dumps(data, ensure_ascii=False)}\n\n" dead = [] for q in sse_clients: try: q.append(event) except Exception: dead.append(q) for q in dead: sse_clients.remove(q) def add_log(line: str): """ログ追加 & SSEブロードキャスト""" progress["logs"].append(line) # ログからフェーズ・進捗を自動検出 if "ログイン開始" in line: progress["phase"] = "login" progress["message"] = "IPATにログイン中..." elif "ログイン成功" in line or "ログイン済み" in line: progress["phase"] = "betting" progress["message"] = "ログイン完了" elif "馬券購入:" in line: m = re.search(r'bet_id=(\d+)\s+(.+?)\s+(\S+)\s+(\S+)\s+(\d+)円', line) if m: progress["current_step"] += 1 progress["current_bet"] = f"{m.group(2)} {m.group(3)} {m.group(4)} {m.group(5)}円" progress["message"] = f"馬券セット中: {progress['current_bet']}" elif "セットボタンクリック" in line: progress["message"] = f"セット完了: {progress['current_bet']}" progress["bets_success"] += 1 elif "購入確認・実行" in line: progress["phase"] = "confirm" progress["message"] = "投票確定中..." elif "購入成功" in line: progress["phase"] = "done" progress["message"] = "購入成功！" elif "dry_run モード" in line: progress["phase"] = "done" progress["message"] = "ドライラン完了" elif "[ERROR]" in line or "[FATAL]" in line: progress["bets_failed"] += 1 progress["message"] = line.split("]")[-1].strip()[:100] elif "[WARN] セットボタン" in line or "[WARN] 馬番" in line: progress["bets_failed"] += 1 broadcast_progress() @app.route("/health") def health(): return jsonify({"status": "ok", "time": datetime.now().isoformat()}) @app.route("/status") def get_status(): return jsonify({ "running": progress["running"], "phase": progress["phase"], "current_step": progress["current_step"], "total_steps": progress["total_steps"], "current_bet": progress["current_bet"], "message": progress["message"], "started_at": progress["started_at"], "finished_at": progress["finished_at"], "bets_success": progress["bets_success"], "bets_failed": progress["bets_failed"], "results": progress["results"], "logs": list(progress["logs"])[-50:], }) @app.route("/progress") def progress_stream(): """SSE (Server-Sent Events) で進捗をリアルタイム配信""" def event_stream(): q = deque(maxlen=50) sse_clients.append(q) try: # 初回: 現在の状態を送信 data = { "running": progress["running"], "phase": progress["phase"], "current_step": progress["current_step"], "total_steps": progress["total_steps"], "current_bet": progress["current_bet"], "message": progress["message"], "percent": round(progress["current_step"] / max(progress["total_steps"], 1) * 100), "bets_success": progress["bets_success"], "bets_failed": progress["bets_failed"], "elapsed": 0, } yield f"data: {json.dumps(data, ensure_ascii=False)}\n\n" while True: if q: yield q.popleft() else: time.sleep(0.5) yield ": keepalive\n\n" finally: if q in sse_clients: sse_clients.remove(q) return Response(event_stream(), mimetype="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}) @app.route("/purchase", methods=["POST"]) def purchase(): return _run_purchase(dry_run=False) @app.route("/dry-run", methods=["POST"]) def dry_run(): return _run_purchase(dry_run=True) def _run_purchase(dry_run: bool): if progress["running"]: return jsonify({"error": "既に実行中です"}), 409 data = request.get_json() if not data: return jsonify({"error": "JSONボディが必要です"}), 400 params = { "inet_id": data.get("inet_id", ""), "ipat_id": data.get("ipat_id", ""), "pin": data.get("pin", ""), "pars": data.get("pars", ""), "bets": data.get("bets", []), "headless": False, "dry_run": dry_run, "discord_webhook": data.get("discord_webhook", ""), } # 進捗リセット progress["running"] = True progress["dry_run"] = dry_run progress["started_at"] = datetime.now().isoformat() progress["finished_at"] = None progress["current_step"] = 0 progress["total_steps"] = len(params["bets"]) progress["current_bet"] = None progress["phase"] = "login" progress["message"] = "起動中..." progress["logs"].clear() progress["results"] = None progress["bets_success"] = 0 progress["bets_failed"] = 0 broadcast_progress() thread = threading.Thread(target=_execute, args=(params,)) thread.start() return jsonify({ "message": f"{'ドライラン' if dry_run else '購入'}を開始しました", "bets_count": len(params["bets"]), }) def _run_script(script_name: str, params: dict) -> dict | None: """Pythonスクリプトを実行し、ログをSSEに流しつつJSON結果を返す""" script_path = os.path.join(SCRIPT_DIR, script_name) proc = subprocess.Popen( [sys.executable, script_path, json.dumps(params, ensure_ascii=False)], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, encoding="utf-8", errors="replace", env={**os.environ, "STORAGE_PATH": STORAGE_DIR}, ) for line in proc.stdout: line = line.rstrip() add_log(line) proc.wait() # ログから最後のJSON行を抽出 for line in reversed(list(progress["logs"])): try: return json.loads(line) except (json.JSONDecodeError, ValueError): continue return None def _verify_purchase(params: dict, bets: list) -> dict: """購入後にIPAT投票照会で実際にbetされたか検証する""" add_log("[照会検証] 投票照会で購入確認を開始...") progress["phase"] = "verify" progress["message"] = "投票照会で購入確認中..." broadcast_progress() check_params = { "inet_id": params.get("inet_id", ""), "ipat_id": params.get("ipat_id", ""), "pin": params.get("pin", ""), "pars": params.get("pars", ""), "date": datetime.now().strftime("%Y-%m-%d"), "headless": False, } check_output = _run_script("ipat_check_votes.py", check_params) if not check_output or not check_output.get("success"): error = check_output.get("error", "照会失敗") if check_output else "照会出力解析失敗" add_log(f"[照会検証] 照会失敗: {error}") return {"success": False, "error": error, "verified": [], "not_found": []} ipat_votes = check_output.get("votes", []) add_log(f"[照会検証] IPAT投票照会: {len(ipat_votes)}件の投票を確認") verified = [] not_found = [] for b in bets: matched = any( v.get("venue") == b.get("venue") and int(v.get("race_number", 0)) == int(b.get("race_number", 0)) and int(v.get("amount", 0)) == int(b.get("amount", 0)) for v in ipat_votes ) if matched: verified.append(b) add_log(f"[照会検証] ✅ 確認OK: {b.get('venue')}{b.get('race_number')}R {b.get('amount')}円") else: not_found.append(b) add_log(f"[照会検証] ❌ 未確認: {b.get('venue')}{b.get('race_number')}R {b.get('amount')}円") return {"success": True, "verified": verified, "not_found": not_found} def _execute(params: dict): try: # 購入実行 output = _run_script("ipat_purchase.py", params) progress["results"] = output confirmed = output.get("confirmed", False) if output else False dry_run = params.get("dry_run", False) # 購入成功 かつ 本番の場合: 照会検証を実行 if confirmed and not dry_run: verify_result = _verify_purchase(params, params.get("bets", [])) if output: output["verify"] = verify_result progress["results"] = output if verify_result.get("success") and verify_result["verified"]: verified_count = len(verify_result["verified"]) not_found_count = len(verify_result["not_found"]) progress["message"] = f"購入成功（照会確認: {verified_count}件OK" if not_found_count > 0: progress["message"] += f", {not_found_count}件未確認" progress["message"] += "）" elif verify_result.get("success") and not verify_result["verified"]: progress["phase"] = "error" progress["message"] = "購入失敗: 照会で1件もbetが確認できませんでした" else: progress["message"] = "購入完了（照会検証失敗 — 手動確認してください）" if progress["phase"] not in ("done", "error"): if output and output.get("success"): progress["phase"] = "done" if not progress["message"] or progress["message"] == "投票確定中...": progress["message"] = "購入成功" if not dry_run else "ドライラン完了" else: progress["phase"] = "error" progress["message"] = output.get("error", "不明なエラー") if output else "出力解析失敗" except Exception as e: progress["phase"] = "error" progress["message"] = str(e) progress["results"] = {"error": str(e)} finally: progress["running"] = False progress["finished_at"] = datetime.now().isoformat() broadcast_progress() if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--bind", default="0.0.0.0") parser.add_argument("--port", type=int, default=5200) args = parser.parse_args() print(f"IPAT購入サーバー起動: {args.bind}:{args.port}") app.run(host=args.bind, port=args.port, debug=False, threaded=True)