#!/usr/bin/env python3 """Live landing jackpot + star soak scan (no archive).""" from __future__ import annotations import json import re from collections import Counter from datetime import datetime, timezone from pathlib import Path BASE = Path(__file__).resolve().parents[1] / "logs" # Night of Aug 3: 22:30 EDT = 02:30 UTC Aug 4 W0 = datetime.fromisoformat("2026-08-04T02:15:00+00:00") W1 = datetime.fromisoformat("2026-08-04T03:00:00+00:00") N0 = datetime.fromisoformat("2026-08-03T22:00:00+00:00") N1 = datetime.fromisoformat("2026-08-04T13:00:00+00:00") JP_EV = re.compile( r"jpot|jackpot|fireworks|early_win|skill_claim|progressive|" r"mrt_|status_0d_synth|x9_no_local|lab_gameplay", re.I, ) def utc_parse(s: str | None) -> datetime | None: if not s: return None s = s.replace("Z", "+00:00") try: return datetime.fromisoformat(s) except Exception: return None def event_files(pos: int) -> list[Path]: d = BASE / str(pos) / "data" if not d.is_dir(): return [] return sorted(d.glob("events.jsonl*"), key=lambda p: p.name) def scan(pos: int, t0: datetime, t1: datetime) -> dict: mrt: Counter[str] = Counter() jp_hits: list[dict] = [] stale_by_tid: Counter[str] = Counter() lab_wager = lab_ack = 0 errors: Counter[str] = Counter() sessions: list[tuple] = [] eth_rx_lab = 0 eth_tx_lab = 0 game_msg: Counter[str] = Counter() parse_builds: Counter[str] = Counter() inject_fail = 0 win_msgs: list[dict] = [] lines = 0 for fp in event_files(pos): try: f = open(fp, "r", encoding="utf-8", errors="replace") except OSError as exc: print(f" open fail {fp}: {exc}") continue with f: for line in f: if not line.strip(): continue lines += 1 try: e = json.loads(line) except Exception: continue t = utc_parse(e.get("utc")) if t is None or not (t0 <= t <= t1): continue ev = e.get("ev") or "" tid = str(e.get("tid") or "") is_lab = len(tid) == 4 and tid.upper().startswith("E0") if "mrt" in ev or ev in ("x9_no_local_x1", "mrt_risk_late_0d_after_x1"): mrt[ev] += 1 jp_hits.append( { "utc": e.get("utc"), "ev": ev, "tid": tid, "game": e.get("game"), "reason": e.get("reason"), "file": fp.name, } ) if ev == "trunk_round_stale_abandon": stale_by_tid[tid or "?"] += 1 if ev == "lab_gameplay_synth_wager": lab_wager += 1 if ev == "lab_gameplay_synth_ack": lab_ack += 1 if ev == "session_start": sessions.append( ( e.get("utc"), e.get("parser_build") or e.get("eg_version"), e.get("pi_label"), ) ) if e.get("parser_build"): parse_builds[str(e.get("parser_build"))] += 1 if ev == "error" or "inject_failed" in str(e.get("message", "")): errors[f"{ev}:{e.get('where')}:{e.get('message')}"] += 1 if "inject" in str(e.get("where", "")) or "inject" in str( e.get("message", "") ): inject_fail += 1 if is_lab and ev == "eth_rx": eth_rx_lab += 1 if is_lab and ev == "eth_tx": eth_tx_lab += 1 if ev in ("packet_complete", "eth_rx") and e.get("game"): game_msg[f"{e.get('game')}:{e.get('msg')}"] += 1 # Jackpot / win interest if JP_EV.search(ev) or str(e.get("trigger", "")).lower().find("jpot") >= 0: jp_hits.append( { "utc": e.get("utc"), "ev": ev, "tid": tid, "game": e.get("game"), "msg": e.get("msg"), "reason": e.get("reason"), "where": e.get("where"), "amount": e.get("amount") or e.get("jpot_amount") or e.get("win_usd") or e.get("win_amount"), "trigger": e.get("trigger"), "qualifies": e.get("qualifies"), "file": fp.name, } ) # High win / jpot opcodes style markers in payload fields if e.get("is_jackpot") or e.get("jackpot") or e.get("jpot"): win_msgs.append( { "utc": e.get("utc"), "ev": ev, "tid": tid, "game": e.get("game"), "extra": { k: e.get(k) for k in ( "is_jackpot", "jackpot", "jpot", "amount", "win_usd", ) if k in e }, } ) # fireworks IPC if "fireworks" in ev.lower() or e.get("ev") == "fireworks": win_msgs.append( { "utc": e.get("utc"), "ev": ev, "tid": tid, "game": e.get("game"), "msg": e.get("msg"), "amount": e.get("amount") or e.get("win_usd"), } ) return { "lines_scanned_files_approx": lines, "mrt": dict(mrt), "mrt_total": sum(mrt.values()), "stale_total": sum(stale_by_tid.values()), "stale_top": stale_by_tid.most_common(12), "lab_wager": lab_wager, "lab_ack": lab_ack, "jp_n": len(jp_hits), "jp_hits": jp_hits, "errors_top": errors.most_common(12), "inject_fail_keys": inject_fail, "sessions": sessions[:8], "parse_builds": dict(parse_builds), "eth_rx_lab": eth_rx_lab, "eth_tx_lab": eth_tx_lab, "game_msg_top": game_msg.most_common(15), "win_msgs": win_msgs[:40], } def main() -> None: print(f"BASE={BASE}") print(f"Jackpot window UTC {W0.isoformat()} .. {W1.isoformat()} (EDT 22:15-23:00)") print() print("=== JACKPOT WINDOW by seat ===") for pos in (1, 11, 5, 51, 2, 21, 3, 31, 4, 41): r = scan(pos, W0, W1) print( f"\n--- seat {pos}: mrt={r['mrt_total']} stale={r['stale_total']} " f"jp_hits={r['jp_n']} lab_w/a={r['lab_wager']}/{r['lab_ack']} " f"lab_eth_rx/tx={r['eth_rx_lab']}/{r['eth_tx_lab']} ---" ) if r["mrt"]: print(" mrt", r["mrt"]) if r["sessions"]: print(" sessions", r["sessions"][:3]) # print interesting jp only shown = 0 for h in r["jp_hits"]: ev = h.get("ev") or "" if any( x in ev for x in ( "jpot", "jackpot", "fireworks", "early_win", "skill", "mrt_", "x9_no", "status_0d_synth", "prog", ) ): print( f" {h.get('utc')} {ev} tid={h.get('tid')} game={h.get('game')} " f"msg={h.get('msg')} reason={h.get('reason')} amt={h.get('amount')} " f"trig={h.get('trigger')} q={h.get('qualifies')}" ) shown += 1 if shown >= 40: print(" ...truncated") break if r["win_msgs"]: print(" win_msgs", r["win_msgs"][:10]) if r["game_msg_top"][:8]: print(" games", r["game_msg_top"][:8]) print("\n=== NIGHTWIDE (N0-N1) MRT / stale / lab ===") for pos in (1, 2, 3, 4, 5, 11, 21, 31, 41, 51): r = scan(pos, N0, N1) print( f"seat{pos}: mrt_total={r['mrt_total']} {r['mrt']} " f"stale={r['stale_total']} lab_w/a={r['lab_wager']}/{r['lab_ack']} " f"jp_n={r['jp_n']}" ) if r["stale_total"] and pos in (3, 31): print(" stale_top", r["stale_top"][:8]) if __name__ == "__main__": main()