""" HEAD OR TAIL — central server (Contabo). Stage 1 foundation. Runs ONE shared round for all shops: - betting window (config.BETTING_TIME) with last BET_LOCK_SECONDS locked - sequential coin reveal - settlement via engine.choose_outcome (margin guaranteed) - jackpot accumulation (+JACKPOT_RATE) and house profit (+HOUSE_RATE) Bets, balances and ledger are all in PostgreSQL. Web sessions (admin/agent logins) are kept in memory; terminal tokens live in the DB. NOTE: this stage exposes the engine + money chain + round loop. The kiosk, cashier, agent and admin pages are Stages 2-4. The server already serves any .html placed beside it (so you can drop pages in as they are built). """ import asyncio, json, secrets, hashlib, os, logging, threading import urllib.parse from http.server import HTTPServer, SimpleHTTPRequestHandler import websockets from websockets.asyncio.server import serve from psycopg2.extras import RealDictCursor import config import db import engine logging.getLogger("websockets").setLevel(logging.CRITICAL) # ---------------- in-memory round state ---------------- current_round = 1000 # set in main() betting_phase = False countdown = 0 coins = [{"id": i + 1, "result": None, "status": "idle"} for i in range(config.NUM_COINS)] connected = set() SESSIONS = {} # web token -> {role,user_id,username} live_multipliers = dict(config.BASE_MULTIPLIERS) liability_by_size = {s: 0.0 for s in range(1, config.NUM_COINS + 1)} total_pool = 0.0 # ---------------- web sessions ---------------- def make_session(uid, role, username): tok = secrets.token_hex(32) SESSIONS[tok] = {"user_id": uid, "role": role, "username": username} return tok def sess(tok): return SESSIONS.get(tok) def role_ok(tok, roles): s = sess(tok) return s and s["role"] in roles def set_reserve(payout, house, jackpot): """Adjust the payout/house/jackpot split live (in-memory). Must sum to 1.0. Resets to config.py defaults on server restart.""" total = payout + house + jackpot if abs(total - 1.0) > 1e-6: return {"success": False, "error": "Rates must sum to 1.0 (got %.3f)" % total} if min(payout, house, jackpot) < 0: return {"success": False, "error": "Rates cannot be negative"} config.PAYOUT_RATE = payout config.HOUSE_RATE = house config.JACKPOT_RATE = jackpot return {"success": True, "payout": payout, "house": house, "jackpot": jackpot} # ---------------- broadcast ---------------- def broadcast(msg): asyncio.ensure_future(_bcast(json.dumps(msg, default=str))) async def _bcast(s): for ws in list(connected): try: await ws.send(s) except Exception: pass # ---------------- HTTP ---------------- def safe_read(fname, default=""): path = os.path.join(os.path.dirname(os.path.abspath(__file__)), fname) try: return open(path, encoding="utf-8").read() except Exception: return default def jdump(handler, data, status=200): body = json.dumps(data, default=str).encode() handler.send_response(status) handler.send_header("Content-type", "application/json") handler.send_header("Content-Length", str(len(body))) handler.send_header("Access-Control-Allow-Origin", "*") handler.end_headers() try: handler.wfile.write(body) except Exception: pass def params(path): if "?" not in path: return {} return dict(urllib.parse.parse_qsl(path.split("?", 1)[1])) class Handler(SimpleHTTPRequestHandler): def log_message(self, *a): pass def do_GET(self): p = self.path.split("?")[0] pr = params(self.path) if p == "/favicon.ico": self.send_response(204); self.end_headers(); return pages = {"/kiosk": "game_kiosk.html", "/cashier": "cashier.html", "/agent": "ad.html", "/admin": "super.html", "/activation": "activation.html"} if p in pages: body = safe_read(pages[p], "

" + p + " (page not built yet)

").encode() self.send_response(200) self.send_header("Content-type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() try: self.wfile.write(body) except Exception: pass return if p.startswith("/assets/"): full = os.path.join(os.path.dirname(os.path.abspath(__file__)), "player-kiosk", "public", p.lstrip("/")) if os.path.exists(full): ext = full.rsplit(".", 1)[-1].lower() ct = {"png": "image/png", "jpg": "image/jpeg", "jpeg": "image/jpeg", "gif": "image/gif", "svg": "image/svg+xml", "ogg": "audio/ogg", "mp3": "audio/mpeg", "wav": "audio/wav"}.get( ext, "application/octet-stream") with open(full, "rb") as f: data = f.read() self.send_response(200) self.send_header("Content-type", ct) self.send_header("Content-Length", str(len(data))) self.end_headers() try: self.wfile.write(data) except Exception: pass else: self.send_error(404) return if p.endswith((".png", ".jpg", ".jpeg", ".ogg", ".mp3", ".wav", ".gif", ".svg")): fname = os.path.basename(p) full = os.path.join(os.path.dirname(os.path.abspath(__file__)), fname) if os.path.exists(full): ext = full.rsplit(".", 1)[-1].lower() ct = {"png": "image/png", "jpg": "image/jpeg", "jpeg": "image/jpeg", "gif": "image/gif", "svg": "image/svg+xml", "ogg": "audio/ogg", "mp3": "audio/mpeg", "wav": "audio/wav"}.get( ext, "application/octet-stream") with open(full, "rb") as f: data = f.read() self.send_response(200) self.send_header("Content-type", ct) self.send_header("Content-Length", str(len(data))) self.end_headers() try: self.wfile.write(data) except Exception: pass else: self.send_error(404) return # API endpoints (also reachable via POST with a JSON body) if self.handle_api(p, pr): return super().do_GET() def do_POST(self): p = self.path.split("?")[0] # read JSON body try: length = int(self.headers.get("Content-Length", 0)) except (TypeError, ValueError): length = 0 raw = self.rfile.read(length) if length else b"" pr = {} if raw: try: pr = json.loads(raw.decode("utf-8")) except Exception: # fall back to form-encoded try: pr = dict(urllib.parse.parse_qsl(raw.decode("utf-8"))) except Exception: pr = {} # merge any query-string params too pr.update(params(self.path)) if self.handle_api(p, pr): return self.send_error(404) def handle_api(self, p, pr): """Shared API routing for GET and POST. Returns True if handled.""" # ---- AUTH ---- if p == "/auth/admin": d = db.admin_login(pr.get("username", ""), pr.get("password", "")) if d: tok = make_session(d["id"], "admin", d["username"]) jdump(self, {"success": True, "token": tok, "balance": float(d["balance"]), "username": d["username"]}) else: jdump(self, {"success": False, "error": "Invalid credentials"}, 401) return True if p == "/auth/agent": d = db.agent_login(pr.get("username", ""), pr.get("password", "")) if d: tok = make_session(d["id"], "agent", d["username"]) jdump(self, {"success": True, "token": tok, "balance": float(d["balance"]), "username": d["username"]}) else: jdump(self, {"success": False, "error": "Invalid credentials"}, 401) return True if p == "/auth/cashier": d = db.cashier_login(pr.get("username", ""), pr.get("password", "")) if d: tok = make_session(d["id"], "cashier", d["username"]) jdump(self, {"success": True, "token": tok, "balance": float(d["real_balance"]), "username": d["username"], "cashier_id": d["id"]}) else: jdump(self, {"success": False, "error": "Invalid credentials"}, 401) return True if p == "/activate": jdump(self, db.activate_terminal(pr.get("code", "").strip())); return True if p == "/cashier/summary": if not role_ok(pr.get("token", ""), ["cashier"]): jdump(self, {"error": "Unauthorized"}, 401); return True cid = sess(pr["token"])["user_id"] jdump(self, build_shift_summary(cid)); return True # ---- ADMIN ---- if p == "/admin/create_agent": if not role_ok(pr.get("token", ""), ["admin"]): jdump(self, {"error": "Unauthorized"}, 401); return True jdump(self, db.create_agent(sess(pr["token"])["user_id"], pr.get("username", "").strip(), pr.get("password", "").strip(), pr.get("phone", ""))); return True if p == "/admin/topup_agent": if not role_ok(pr.get("token", ""), ["admin"]): jdump(self, {"error": "Unauthorized"}, 401); return True jdump(self, db.topup_agent(sess(pr["token"])["user_id"], pr.get("agent_id", ""), float(pr.get("amount", 0) or 0))); return True if p == "/admin/reassign_cashier": if not role_ok(pr.get("token", ""), ["admin"]): jdump(self, {"error": "Unauthorized"}, 401); return True jdump(self, db.reassign_cashier(sess(pr["token"])["user_id"], pr.get("cashier_id", ""), pr.get("to_agent_id", ""))); return True if p == "/admin/jackpot": jdump(self, {"jackpot": db.get_jackpot()}); return True # ---- AGENT ---- if p == "/agent/create_cashier": if not role_ok(pr.get("token", ""), ["agent"]): jdump(self, {"error": "Unauthorized"}, 401); return True jdump(self, db.create_cashier(sess(pr["token"])["user_id"], pr.get("username", "").strip(), pr.get("password", "").strip(), pr.get("shop", ""), pr.get("phone", ""))); return True if p == "/agent/topup_cashier": if not role_ok(pr.get("token", ""), ["agent"]): jdump(self, {"error": "Unauthorized"}, 401); return True jdump(self, db.topup_cashier(sess(pr["token"])["user_id"], pr.get("cashier_id", ""), float(pr.get("amount", 0) or 0))); return True if p == "/agent/gen_code": if not role_ok(pr.get("token", ""), ["agent"]): jdump(self, {"error": "Unauthorized"}, 401); return True jdump(self, db.gen_activation_code(sess(pr["token"])["user_id"], pr.get("cashier_id", ""))); return True # ---- REPORTS / READ ---- if p == "/agent/overview": if not role_ok(pr.get("token", ""), ["agent"]): jdump(self, {"error": "Unauthorized"}, 401); return True aid = sess(pr["token"])["user_id"] jdump(self, {"overview": db.agent_overview(aid), "cashiers": db.agent_cashiers(aid)}); return True if p == "/admin/overview": if not role_ok(pr.get("token", ""), ["admin"]): jdump(self, {"error": "Unauthorized"}, 401); return True adid = sess(pr["token"])["user_id"] jdump(self, {"overview": db.admin_overview(), "agents": db.admin_agents(adid), "cashiers": db.admin_all_cashiers(), "reserve": {"payout": config.PAYOUT_RATE, "house": config.HOUSE_RATE, "jackpot": config.JACKPOT_RATE}}); return True if p == "/admin/set_reserve": if not role_ok(pr.get("token", ""), ["admin"]): jdump(self, {"error": "Unauthorized"}, 401); return True jdump(self, set_reserve(float(pr.get("payout", 0) or 0), float(pr.get("house", 0) or 0), float(pr.get("jackpot", 0) or 0))); return True return False def start_http(): HTTPServer(("0.0.0.0", config.HTTP_PORT), Handler).serve_forever() # ---------------- bet placement (WebSocket) ---------------- def place_bet(cashier_id, session_id, picks, stake): """Fund-first wallet draw, persists the bet, snapshots the live multiplier.""" size = len(picks) if size < 1 or size > config.NUM_COINS or stake <= 0: return {"success": False, "error": "Invalid bet"} multi = live_multipliers.get(size, config.BASE_MULTIPLIERS[size]) with db.conn_ctx() as c: cur = c.cursor(cursor_factory=RealDictCursor) cur.execute("SELECT real_balance,fund_balance FROM player_sessions " "WHERE id=%s FOR UPDATE", (session_id,)) ps = cur.fetchone() if not ps: return {"success": False, "error": "Unknown player"} avail = float(ps["real_balance"]) + float(ps["fund_balance"]) if stake > avail + 1e-9: return {"success": False, "error": "Insufficient player balance"} fund_used = min(float(ps["fund_balance"]), stake) # fund-first real_used = stake - fund_used cur.execute("SELECT agent_id FROM cashiers WHERE id=%s", (cashier_id,)) agent_id = cur.fetchone()["agent_id"] ticket_id = "T" + secrets.token_hex(5).upper() barcode = (config.HOUSE_CODE + ticket_id + str(current_round)).upper() raw = ticket_id + str(current_round) + json.dumps(picks, sort_keys=True) + \ str(stake) + barcode + config.HOUSE_CODE thash = hashlib.sha256(raw.encode()).hexdigest() group_key = session_id + ":" + str(current_round) cur.execute( "UPDATE player_sessions SET real_balance=real_balance-%s," "fund_balance=fund_balance-%s," "fund_rollover_rem=GREATEST(0,fund_rollover_rem-%s) WHERE id=%s", (real_used, fund_used, fund_used, session_id), ) cur.execute( "INSERT INTO bets (ticket_id,round_no,session_id,cashier_id,agent_id,picks," "size,stake,fund_used,real_used,multiplier,group_key,barcode,hash) " "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", (ticket_id, current_round, session_id, cashier_id, agent_id, json.dumps(picks), size, stake, fund_used, real_used, multi, group_key, barcode, thash), ) db.ledger_move(cur, "bet", "player", session_id, "pool_round", None, stake, current_round, ref=ticket_id) global total_pool total_pool += stake liability_by_size[size] += stake * multi _recalc_multipliers() return {"success": True, "ticket_id": ticket_id, "barcode": barcode, "fund_used": fund_used, "real_used": real_used, "multiplier": multi} def place_cash_bet(cashier_id, picks, stake): """Single-step cash bet: cashier takes cash and places the bet in one action. The stake is debited from the cashier's own balance (the shift float). No player_session is involved. Returns ticket info for the local printer.""" size = len(picks) if size < 1 or size > config.NUM_COINS or stake <= 0: return {"success": False, "error": "Invalid bet"} multi = live_multipliers.get(size, config.BASE_MULTIPLIERS[size]) with db.conn_ctx() as c: cur = c.cursor(cursor_factory=RealDictCursor) cur.execute("SELECT username,real_balance,agent_id FROM cashiers " "WHERE id=%s FOR UPDATE", (cashier_id,)) ci = cur.fetchone() if not ci: return {"success": False, "error": "Unknown cashier"} if stake > float(ci["real_balance"]) + 1e-9: return {"success": False, "error": "Insufficient cashier balance"} agent_id = ci["agent_id"] ticket_id = "T" + secrets.token_hex(5).upper() barcode = (config.HOUSE_CODE + ticket_id + str(current_round)).upper() raw = ticket_id + str(current_round) + json.dumps(picks, sort_keys=True) + \ str(stake) + barcode + config.HOUSE_CODE thash = hashlib.sha256(raw.encode()).hexdigest() # debit the cashier float cur.execute("UPDATE cashiers SET real_balance=real_balance-%s WHERE id=%s", (stake, cashier_id)) # persist the bet (session_id NULL = direct cash bet) cur.execute( "INSERT INTO bets (ticket_id,round_no,session_id,cashier_id,agent_id,picks," "size,stake,fund_used,real_used,multiplier,group_key,barcode,hash) " "VALUES (%s,%s,NULL,%s,%s,%s,%s,%s,0,%s,%s,%s,%s,%s)", (ticket_id, current_round, cashier_id, agent_id, json.dumps(picks), size, stake, stake, multi, ticket_id + ":" + str(current_round), barcode, thash), ) db.ledger_move(cur, "bet", "cashier", cashier_id, "pool_round", None, stake, current_round, ref=ticket_id) global total_pool total_pool += stake liability_by_size[size] += stake * multi _recalc_multipliers() return {"success": True, "ticket_id": ticket_id, "barcode": barcode, "round_no": current_round, "multiplier": multi, "cashier_name": ci["username"], "stake": stake, "picks": picks} def _recalc_multipliers(): global live_multipliers live_multipliers = engine.recalc_live_multipliers(total_pool, liability_by_size) async def handle(ws): # NOT added to `connected` (the broadcast set) until this connection # authenticates — either as a cashier (login token) or an activated # kiosk terminal (activation token). This stops anyone from just opening # the kiosk URL and seeing the live game. auth = {"cashier_id": None, "authed": False} try: async for raw in ws: data = json.loads(raw) t = data.get("type") if t == "auth": token = data.get("token", "") row = db.cashier_by_terminal(token) cid = row["cashier_id"] if row else None if not cid: s = sess(token) if s and s["role"] == "cashier": cid = s["user_id"] if not cid: await ws.send(json.dumps({"type": "error", "message": "Invalid token"})) continue auth["cashier_id"] = cid auth["authed"] = True connected.add(ws) # now receives game broadcasts sh = db.open_shift(cid) # auto-start shift on login auth["shift_open"] = bool(sh.get("success")) cinfo = db.q("SELECT username,real_balance FROM cashiers WHERE id=%s", (cid,), fetch="one") await ws.send(json.dumps({ "type": "auth_ok", "cashier_id": cid, "username": cinfo["username"], "cashier_balance": float(cinfo["real_balance"]), "shift_open": auth["shift_open"], "round": current_round, "betting_open": betting_phase, "countdown": countdown, "live_multipliers": live_multipliers, "jackpot": db.get_jackpot()})) elif t == "kiosk_auth": # A TV/kiosk presents its stored terminal token. Valid token = # this TV was activated by an agent's code, so stream the game. token = data.get("token", "") row = db.cashier_by_terminal(token) if not row: await ws.send(json.dumps({"type": "kiosk_denied"})) continue auth["authed"] = True connected.add(ws) # now receives game broadcasts await ws.send(json.dumps({"type": "game_state", "round": current_round, "betting_open": betting_phase, "countdown": countdown, "coins": coins, "bet_lock": config.BET_LOCK_SECONDS, "live_multipliers": live_multipliers, "jackpot": db.get_jackpot()})) elif t == "fund_player": if not auth["cashier_id"]: await ws.send(json.dumps({"type": "error", "message": "Not authed"})); continue r = db.fund_player(auth["cashier_id"], data.get("session", ""), float(data.get("amount", 0))) await ws.send(json.dumps({"type": "fund_result", **r})) elif t == "place_bet": if not auth["cashier_id"]: await ws.send(json.dumps({"type": "error", "message": "Not authed"})); continue if not db.current_shift(auth["cashier_id"]): await ws.send(json.dumps({"type": "bet_result", "success": False, "error": "No open shift — start a shift to bet"})); continue if not betting_phase or countdown <= config.BET_LOCK_SECONDS: await ws.send(json.dumps({"type": "error", "message": "Betting closed" if not betting_phase else "Locked"})) continue r = place_cash_bet(auth["cashier_id"], data.get("picks", []), float(data.get("amount", 0))) if r["success"]: broadcast({"type": "live_multipliers", "live_multipliers": live_multipliers}) # send updated cashier balance back bal = db.q("SELECT real_balance FROM cashiers WHERE id=%s", (auth["cashier_id"],), fetch="one") r["cashier_balance"] = float(bal["real_balance"]) await ws.send(json.dumps({"type": "bet_result", **r})) elif t == "start_shift": if not auth["cashier_id"]: await ws.send(json.dumps({"type": "error", "message": "Not authed"})); continue sh = db.open_shift(auth["cashier_id"]) auth["shift_open"] = bool(sh.get("success")) await ws.send(json.dumps({"type": "shift_status", "shift_open": auth["shift_open"]})) elif t == "end_shift": if not auth["cashier_id"]: await ws.send(json.dumps({"type": "error", "message": "Not authed"})); continue r = db.close_shift(auth["cashier_id"]) auth["shift_open"] = False await ws.send(json.dumps({"type": "shift_ended", **r})) elif t == "lookup_ticket": # verify a ticket by typed number or scanned barcode (read-only) if not auth["cashier_id"]: await ws.send(json.dumps({"type": "error", "message": "Not authed"})); continue r = lookup_ticket(data.get("code", "")) await ws.send(json.dumps({"type": "lookup_result", **r})) elif t == "redeem_ticket": # pay out a winning cash ticket; credit cashier float (they pay from drawer) if not auth["cashier_id"]: await ws.send(json.dumps({"type": "error", "message": "Not authed"})); continue r = redeem_ticket(auth["cashier_id"], data.get("code", "")) if r.get("success"): bal = db.q("SELECT real_balance FROM cashiers WHERE id=%s", (auth["cashier_id"],), fetch="one") r["cashier_balance"] = float(bal["real_balance"]) await ws.send(json.dumps({"type": "redeem_result", **r})) elif t == "cancel_ticket": if not auth["cashier_id"]: await ws.send(json.dumps({"type": "error", "message": "Not authed"})); continue r = cancel_ticket(auth["cashier_id"], data.get("code", "")) if r.get("success"): bal = db.q("SELECT real_balance FROM cashiers WHERE id=%s", (auth["cashier_id"],), fetch="one") r["cashier_balance"] = float(bal["real_balance"]) await ws.send(json.dumps({"type": "cancel_result", **r})) except websockets.exceptions.ConnectionClosed: pass finally: connected.discard(ws) # ---------------- ticket lookup / redeem / cancel ---------------- def _find_ticket(cur, code): """Resolve a ticket by ticket_id OR barcode. Returns row or None.""" code = (code or "").strip().upper() if not code: return None # strip the {B Code128 prefix if a scanner included it if code.startswith("{B"): code = code[2:] cur.execute("SELECT * FROM bets WHERE UPPER(ticket_id)=%s OR UPPER(barcode)=%s", (code, code)) row = cur.fetchone() if not row and code.startswith(config.HOUSE_CODE): # barcode payload is HOUSE_CODE + ticket_id + round; pull the ticket_id out inner = code[len(config.HOUSE_CODE):] cur.execute("SELECT * FROM bets WHERE UPPER(barcode)=%s OR %s LIKE UPPER(ticket_id)||'%%'", (code, inner)) row = cur.fetchone() return row def lookup_ticket(code): with db.conn_ctx() as c: cur = c.cursor(cursor_factory=RealDictCursor) b = _find_ticket(cur, code) if not b: return {"success": False, "error": "Ticket not found"} return {"success": True, "ticket_id": b["ticket_id"], "round_no": b["round_no"], "status": b["status"], "stake": float(b["stake"]), "win_amount": float(b["win_amount"] or 0), "picks": b["picks"], "redeemed": bool(b.get("redeemed", False)), "multiplier": float(b["multiplier"])} def redeem_ticket(cashier_id, code): with db.conn_ctx() as c: cur = c.cursor(cursor_factory=RealDictCursor) b = _find_ticket(cur, code) if not b: return {"success": False, "error": "Ticket not found"} if b["status"] not in ("win", "near_miss"): return {"success": False, "error": "Not a winning ticket"} if b.get("redeemed"): return {"success": False, "error": "Already paid"} payout = float(b["win_amount"] or 0) cur.execute("UPDATE bets SET redeemed=TRUE,redeemed_at=NOW(),redeemed_by=%s " "WHERE ticket_id=%s", (cashier_id, b["ticket_id"])) # cashier pays the player from the drawer, so credit the float back to balance the books cur.execute("UPDATE cashiers SET real_balance=real_balance+%s WHERE id=%s", (payout, cashier_id)) db.ledger_move(cur, "redeem", "pool_round", None, "cashier", cashier_id, payout, b["round_no"], ref=b["ticket_id"]) return {"success": True, "ticket_id": b["ticket_id"], "paid": payout, "round_no": b["round_no"], "picks": b["picks"], "stake": float(b["stake"]), "multiplier": float(b["multiplier"])} def cancel_ticket(cashier_id, code): with db.conn_ctx() as c: cur = c.cursor(cursor_factory=RealDictCursor) b = _find_ticket(cur, code) if not b: return {"success": False, "error": "Ticket not found"} if b["cashier_id"] != cashier_id: return {"success": False, "error": "Not your ticket"} if b["round_no"] != current_round or not betting_phase: return {"success": False, "error": "Can only cancel during the same betting round"} if b["status"] != "pending": return {"success": False, "error": "Ticket already settled"} if b.get("redeemed") or b["status"] == "cancelled": return {"success": False, "error": "Already voided"} stake = float(b["stake"]) cur.execute("UPDATE bets SET status='cancelled',cancelled=TRUE,cancelled_at=NOW() WHERE ticket_id=%s", (b["ticket_id"],)) cur.execute("UPDATE cashiers SET real_balance=real_balance+%s WHERE id=%s", (stake, cashier_id)) db.ledger_move(cur, "cancel", "pool_round", None, "cashier", cashier_id, stake, b["round_no"], ref=b["ticket_id"]) # remove from live pool global total_pool total_pool = max(0.0, total_pool - stake) liability_by_size[b["size"]] = max(0.0, liability_by_size[b["size"]] - stake * float(b["multiplier"])) _recalc_multipliers() return {"success": True, "ticket_id": b["ticket_id"], "refunded": stake, "round_no": b["round_no"], "picks": b["picks"], "stake": stake, "multiplier": float(b["multiplier"])} def build_shift_summary(cashier_id): """Compute the cashier's running shift totals for the summary ticket. Shapes the dict to match the print helper's build_summary_ticket().""" with db.conn_ctx() as c: cur = c.cursor(cursor_factory=RealDictCursor) cur.execute("SELECT username,real_balance FROM cashiers WHERE id=%s", (cashier_id,)) ci = cur.fetchone() if not ci: return {"error": "Unknown cashier"} # find the current open shift, if any cur.execute("SELECT id,start_time,start_balance FROM shifts " "WHERE cashier_id=%s AND status='open' ORDER BY start_time DESC LIMIT 1", (cashier_id,)) sh = cur.fetchone() since = sh["start_time"] if sh else None start_balance = float(sh["start_balance"]) if sh else 0.0 # aggregate bets in this shift window where = "cashier_id=%s" + (" AND created_at>=%s" if since else "") args = (cashier_id,) + ((since,) if since else ()) cur.execute(f"SELECT COALESCE(SUM(stake),0) s FROM bets WHERE {where} AND status<>'cancelled'", args) bets_total = float(cur.fetchone()["s"]) cur.execute(f"SELECT COALESCE(SUM(stake),0) s FROM bets WHERE {where} AND status='cancelled'", args) cancels_total = float(cur.fetchone()["s"]) cur.execute(f"SELECT COALESCE(SUM(win_amount),0) s FROM bets WHERE {where} AND redeemed=TRUE", args) redeemed_total = float(cur.fetchone()["s"]) end_balance = float(ci["real_balance"]) now = __import__("datetime").datetime.now() return { "username": ci["username"], "cashier_id_label": "#" + str(cashier_id)[:6], "start_time": (since.strftime("%Y/%m/%d %H:%M") if since else "—"), "end_time": now.strftime("%Y/%m/%d %H:%M"), "start_balance": start_balance, "deposits": 0.0, "bets": bets_total, "cancellations": cancels_total, "redeemed": redeemed_total, "commission": 0.0, "withdraws": 0.0, "end_balance": end_balance, } # ---------------- settlement ---------------- def settle_round(outcome_faces): """Apply the revealed board to the DB; returns a summary dict.""" with db.conn_ctx() as c: cur = c.cursor(cursor_factory=RealDictCursor) cur.execute("SELECT ticket_id,session_id,cashier_id,picks,size,stake,multiplier " "FROM bets WHERE round_no=%s AND cancelled=FALSE", (current_round,)) rows = cur.fetchall() bets = [{"ticket_id": r["ticket_id"], "session_id": r["session_id"], "cashier_id": r["cashier_id"], "picks": r["picks"], "size": r["size"], "stake": float(r["stake"]), "multiplier": float(r["multiplier"])} for r in rows] pool = sum(b["stake"] for b in bets) total, results, cashback = engine.evaluate_outcome(tuple(outcome_faces), bets) ceiling = pool * config.PAYOUT_RATE scale = (ceiling / total) if (total > ceiling and total > 0) else 1.0 sess_cashier = {b["session_id"]: b["cashier_id"] for b in bets} winner_count = 0 for b in bets: status, payout = results[b["ticket_id"]] payout = round(payout * scale, 2) if status == "win": winner_count += 1 cur.execute("UPDATE bets SET status='win',is_winner=TRUE,win_amount=%s " "WHERE ticket_id=%s", (payout, b["ticket_id"])) if b["session_id"]: cur.execute("UPDATE player_sessions SET real_balance=real_balance+%s " "WHERE id=%s", (payout, b["session_id"])) db.ledger_move(cur, "win", "pool_round", None, "player", b["session_id"], payout, current_round, ref=b["ticket_id"]) # cash bets (NULL session): paid out by cashier on redemption elif status == "near_miss": cur.execute("UPDATE bets SET status='near_miss',near_miss=TRUE,win_amount=%s " "WHERE ticket_id=%s", (payout, b["ticket_id"])) if b["session_id"]: cur.execute("UPDATE player_sessions SET real_balance=real_balance+%s " "WHERE id=%s", (payout, b["session_id"])) db.ledger_move(cur, "near_miss", "pool_round", None, "player", b["session_id"], payout, current_round, ref=b["ticket_id"]) else: cur.execute("UPDATE bets SET status='loss' WHERE ticket_id=%s", (b["ticket_id"],)) cashback_total = 0.0 for sid, amt in cashback.items(): amt = round(amt * scale, 2) if amt <= 0: continue cashback_total += amt cur.execute("UPDATE player_sessions SET fund_balance=fund_balance+%s," "fund_rollover_rem=fund_rollover_rem+%s WHERE id=%s", (amt, amt * config.CASHBACK_ROLLOVER, sid)) cur.execute("UPDATE promo_pool SET balance=balance-%s WHERE id=1", (amt,)) cid = sess_cashier.get(sid) cur.execute("UPDATE cashiers SET real_balance=real_balance+%s WHERE id=%s", (amt, cid)) db.ledger_move(cur, "cashback", "pool_promo", None, "player", sid, amt, current_round, ref="fund credit") db.ledger_move(cur, "reimburse", "pool_promo", None, "cashier", cid, amt, current_round, ref="cashback reimbursement") total_payout = sum(round(p * scale, 2) for (_, p) in results.values()) + cashback_total house_cut = round(pool * config.HOUSE_RATE, 2) jackpot_cut = round(pool * config.JACKPOT_RATE, 2) cur.execute("UPDATE house SET profit=profit+%s WHERE id=1", (house_cut,)) cur.execute("UPDATE jackpot SET amount=amount+%s WHERE id=1", (jackpot_cut,)) cur.execute("UPDATE promo_pool SET balance=balance+%s WHERE id=1", (jackpot_cut,)) db.ledger_move(cur, "house_take", "pool_round", None, "pool_house", None, house_cut, current_round) db.ledger_move(cur, "jackpot_take", "pool_round", None, "pool_jackpot", None, jackpot_cut, current_round) cur.execute( "INSERT INTO rounds (round_no,status,result_sequence,total_pool,total_payout," "house_cut,jackpot_cut,scale_applied,ticket_count,winner_count,settled_at) " "VALUES (%s,'settled',%s,%s,%s,%s,%s,%s,%s,%s,now()) " "ON CONFLICT (round_no) DO UPDATE SET status='settled'," "result_sequence=EXCLUDED.result_sequence,total_pool=EXCLUDED.total_pool," "total_payout=EXCLUDED.total_payout,house_cut=EXCLUDED.house_cut," "jackpot_cut=EXCLUDED.jackpot_cut,scale_applied=EXCLUDED.scale_applied," "ticket_count=EXCLUDED.ticket_count,winner_count=EXCLUDED.winner_count," "settled_at=now()", (current_round, " ".join(outcome_faces), pool, total_payout, house_cut, jackpot_cut, scale, len(bets), winner_count), ) db.save_last_round(current_round) return {"pool": pool, "payout": total_payout, "house": house_cut, "jackpot_cut": jackpot_cut, "winners": winner_count, "scale": scale, "cashback": cashback_total} # ---------------- game loop ---------------- async def game_loop(): global current_round, betting_phase, countdown, coins, total_pool, liability_by_size await asyncio.sleep(2) while True: try: current_round += 1 for cc in coins: cc["result"] = None; cc["status"] = "idle" total_pool = 0.0 liability_by_size = {s: 0.0 for s in range(1, config.NUM_COINS + 1)} _recalc_multipliers() betting_phase = True countdown = config.BETTING_TIME db.q("INSERT INTO rounds (round_no,status) VALUES (%s,'betting') " "ON CONFLICT (round_no) DO UPDATE SET status='betting'", (current_round,)) broadcast({"type": "betting_start", "round": current_round, "coins": coins, "countdown": countdown, "bet_lock": config.BET_LOCK_SECONDS, "live_multipliers": live_multipliers, "jackpot": db.get_jackpot()}) while countdown > 0: await asyncio.sleep(1) countdown -= 1 broadcast({"type": "countdown", "countdown": countdown, "round": current_round, "live_multipliers": live_multipliers, "jackpot": db.get_jackpot()}) betting_phase = False broadcast({"type": "betting_closed", "round": current_round}) await asyncio.sleep(2) rows = db.q("SELECT ticket_id,session_id,picks,size,stake,multiplier " "FROM bets WHERE round_no=%s AND cancelled=FALSE", (current_round,), fetch="all") or [] bets = [{"ticket_id": r["ticket_id"], "session_id": r["session_id"], "picks": r["picks"], "size": r["size"], "stake": float(r["stake"]), "multiplier": float(r["multiplier"])} for r in rows] pool = sum(b["stake"] for b in bets) if bets: pick = engine.choose_outcome(bets, pool) board = list(pick["outcome"]) else: import random board = [random.choice(["HEADS", "TAILS"]) for _ in range(config.NUM_COINS)] if config.TEST_MODE and bets: board = ["HEADS"] * config.NUM_COINS for pk in bets[0]["picks"]: board[pk["coin"] - 1] = pk["pick"] for idx in range(config.NUM_COINS): coins[idx]["status"] = "flipping" broadcast({"type": "coin_flipping", "coin_index": idx, "coin": coins[idx], "round": current_round}) await asyncio.sleep(config.COIN_SPIN_SECONDS) coins[idx]["result"] = board[idx]; coins[idx]["status"] = "result" broadcast({"type": "coin_result", "coin_index": idx, "coin": coins[idx], "round": current_round}) await asyncio.sleep(config.COIN_SHOW_SECONDS) summary = settle_round(board) if bets else { "pool": 0, "payout": 0, "house": 0, "jackpot_cut": 0, "winners": 0, "scale": 1.0, "cashback": 0} broadcast({"type": "winners", "round": current_round, "coins": [dict(cc) for cc in coins], "result_sequence": " ".join(board), "summary": summary, "jackpot": db.get_jackpot()}) await asyncio.sleep(config.WINNER_HOLD_SECS) broadcast({"type": "round_end", "round": current_round}) for i in range(config.INTRO_COUNTDOWN, 0, -1): broadcast({"type": "countdown", "countdown": i, "round": current_round, "intro": True, "jackpot": db.get_jackpot()}) await asyncio.sleep(1) except Exception as e: import traceback print("game_loop error:", e, flush=True) traceback.print_exc() await asyncio.sleep(2) async def main(): global current_round db.init_pool() db.bootstrap_admin() current_round = db.load_last_round() threading.Thread(target=start_http, daemon=True).start() print("HEAD OR TAIL central server | HTTP :%d WS :%d" % (config.HTTP_PORT, config.WS_PORT), flush=True) print("Reserve: payout %.0f%% / house %.0f%% / jackpot %.0f%%" % (config.PAYOUT_RATE * 100, config.HOUSE_RATE * 100, config.JACKPOT_RATE * 100), flush=True) async with serve(handle, "0.0.0.0", config.WS_PORT): asyncio.ensure_future(game_loop()) await asyncio.Future() if __name__ == "__main__": asyncio.run(main())