""" SQLite Database Layer for SL CAT TOPUP Web Store """ import sqlite3 import uuid import hashlib import json from datetime import datetime, timedelta from typing import Optional, List, Dict, Any, Tuple from config import DATABASE_PATH def get_connection(): conn = sqlite3.connect(DATABASE_PATH, timeout=30) conn.row_factory = sqlite3.Row return conn def init_db(): """Initializes SQLite database tables with WAL mode and indexes for production stability.""" with get_connection() as conn: cursor = conn.cursor() try: cursor.execute("PRAGMA journal_mode = WAL") cursor.execute("PRAGMA synchronous = NORMAL") cursor.execute("PRAGMA busy_timeout = 30000") except Exception: pass cursor.execute( """ CREATE TABLE IF NOT EXISTS orders ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_code TEXT UNIQUE NOT NULL, player_id TEXT NOT NULL, nickname TEXT, product_id TEXT NOT NULL, product_name TEXT NOT NULL, quantity INTEGER DEFAULT 1, amount REAL NOT NULL, qr_payload TEXT, status TEXT DEFAULT 'pending_payment', tx_reference TEXT, slip_path TEXT, htg_order_ids TEXT, error_message TEXT, user_id INTEGER, customer_name TEXT, customer_phone TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ) """ ) # Migrate orders table if columns don't exist yet cursor.execute("PRAGMA table_info(orders)") cols = [col[1] for col in cursor.fetchall()] if "user_id" not in cols: cursor.execute("ALTER TABLE orders ADD COLUMN user_id INTEGER") if "customer_name" not in cols: cursor.execute("ALTER TABLE orders ADD COLUMN customer_name TEXT") if "customer_phone" not in cols: cursor.execute("ALTER TABLE orders ADD COLUMN customer_phone TEXT") if "items_json" not in cols: cursor.execute("ALTER TABLE orders ADD COLUMN items_json TEXT") if "slip_hash" not in cols: cursor.execute("ALTER TABLE orders ADD COLUMN slip_hash TEXT") cursor.execute("CREATE INDEX IF NOT EXISTS idx_orders_slip_hash ON orders(slip_hash)") if "slip_dhash" not in cols: cursor.execute("ALTER TABLE orders ADD COLUMN slip_dhash TEXT") cursor.execute("CREATE INDEX IF NOT EXISTS idx_orders_slip_dhash ON orders(slip_dhash)") if "customer_email" not in cols: cursor.execute("ALTER TABLE orders ADD COLUMN customer_email TEXT") if "email_receipt_sent" not in cols: cursor.execute("ALTER TABLE orders ADD COLUMN email_receipt_sent INTEGER DEFAULT 0") cursor.execute( """ CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, phone TEXT UNIQUE NOT NULL, pin_hash TEXT NOT NULL, wallet_balance REAL DEFAULT 0.0, saved_uid TEXT, saved_nickname TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ) """ ) # Migrate users table for wallet_balance & avatar cursor.execute("PRAGMA table_info(users)") u_cols = [col[1] for col in cursor.fetchall()] if "wallet_balance" not in u_cols: cursor.execute("ALTER TABLE users ADD COLUMN wallet_balance REAL DEFAULT 0.0") if "avatar" not in u_cols: cursor.execute("ALTER TABLE users ADD COLUMN avatar TEXT DEFAULT '👤'") if "email" not in u_cols: cursor.execute("ALTER TABLE users ADD COLUMN email TEXT") if "email_verified" not in u_cols: cursor.execute("ALTER TABLE users ADD COLUMN email_verified INTEGER DEFAULT 0") if "email_otp" not in u_cols: cursor.execute("ALTER TABLE users ADD COLUMN email_otp TEXT") if "email_otp_expiry" not in u_cols: cursor.execute("ALTER TABLE users ADD COLUMN email_otp_expiry TEXT") cursor.execute( """ CREATE TABLE IF NOT EXISTS wallet_transactions ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, type TEXT NOT NULL, amount REAL NOT NULL, balance_after REAL NOT NULL, description TEXT NOT NULL, reference_code TEXT, created_at TEXT NOT NULL ) """ ) cursor.execute( """ CREATE TABLE IF NOT EXISTS settings ( key TEXT PRIMARY KEY, value TEXT ) """ ) cursor.execute( """ CREATE TABLE IF NOT EXISTS bank_notifications ( id INTEGER PRIMARY KEY AUTOINCREMENT, source TEXT, amount REAL, reference TEXT, raw_text TEXT, is_matched INTEGER DEFAULT 0, matched_order TEXT, created_at TEXT ) """ ) cursor.execute( """ CREATE TABLE IF NOT EXISTS spins ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER, phone TEXT, spin_date TEXT NOT NULL, reward_type TEXT NOT NULL, reward_label TEXT NOT NULL, reward_amount REAL DEFAULT 0.0, claimed INTEGER DEFAULT 0, created_at TEXT NOT NULL ) """ ) cursor.execute( """ CREATE TABLE IF NOT EXISTS site_visitors ( id INTEGER PRIMARY KEY AUTOINCREMENT, visitor_id TEXT UNIQUE NOT NULL, ip_hash TEXT, user_agent TEXT, last_path TEXT, visit_count INTEGER DEFAULT 1, user_id INTEGER, first_seen TEXT NOT NULL, last_seen TEXT NOT NULL ) """ ) cursor.execute( """ CREATE TABLE IF NOT EXISTS coupons ( id INTEGER PRIMARY KEY AUTOINCREMENT, code TEXT UNIQUE NOT NULL, discount_type TEXT DEFAULT 'fixed', discount_value REAL NOT NULL, min_order_amount REAL DEFAULT 0.0, max_uses INTEGER DEFAULT 1000, used_count INTEGER DEFAULT 0, expires_at TEXT, is_active INTEGER DEFAULT 1, created_at TEXT NOT NULL ) """ ) cursor.execute( """ CREATE TABLE IF NOT EXISTS banned_devices ( id INTEGER PRIMARY KEY AUTOINCREMENT, device_id TEXT UNIQUE NOT NULL, ip_address TEXT, reason TEXT, banned_by TEXT DEFAULT 'admin', created_at TEXT NOT NULL ) """ ) cursor.execute( """ CREATE TABLE IF NOT EXISTS banned_ips ( id INTEGER PRIMARY KEY AUTOINCREMENT, ip_address TEXT UNIQUE NOT NULL, reason TEXT, created_at TEXT NOT NULL ) """ ) cursor.execute( """ CREATE TABLE IF NOT EXISTS chat_ratings ( id INTEGER PRIMARY KEY AUTOINCREMENT, rating INTEGER NOT NULL, comment TEXT, user_id INTEGER, device_id TEXT, created_at TEXT NOT NULL ) """ ) cursor.execute( """ CREATE TABLE IF NOT EXISTS admin_tickets ( id INTEGER PRIMARY KEY AUTOINCREMENT, ticket_code TEXT UNIQUE NOT NULL, user_id INTEGER, player_id TEXT, sender_name TEXT NOT NULL, sender_contact TEXT NOT NULL, subject TEXT, message TEXT NOT NULL, order_code TEXT, status TEXT DEFAULT 'open', admin_reply TEXT, replied_at TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ) """ ) cursor.execute("PRAGMA table_info(admin_tickets)") tkt_cols = [col[1] for col in cursor.fetchall()] if "conversation_json" not in tkt_cols: cursor.execute("ALTER TABLE admin_tickets ADD COLUMN conversation_json TEXT") cursor.execute( """ CREATE TABLE IF NOT EXISTS mystery_box_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER, player_id TEXT, nickname TEXT, order_code TEXT, amount_paid REAL DEFAULT 150.0, won_item_id TEXT NOT NULL, won_item_name TEXT NOT NULL, won_item_type TEXT NOT NULL, won_amount REAL DEFAULT 0.0, is_win INTEGER DEFAULT 1, delivery_status TEXT DEFAULT 'completed', created_at TEXT NOT NULL ) """ ) cursor.execute( """ CREATE TABLE IF NOT EXISTS daily_mystery_luck ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER, device_id TEXT, spin_date TEXT NOT NULL, win_rate INTEGER NOT NULL DEFAULT 80, luck_label TEXT NOT NULL, created_at TEXT NOT NULL ) """ ) cursor.execute( """ CREATE TABLE IF NOT EXISTS push_subscriptions ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER, ticket_code TEXT, device_id TEXT, endpoint TEXT UNIQUE NOT NULL, p256dh TEXT, auth TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ) """ ) # Migrate orders table cursor.execute("PRAGMA table_info(orders)") ord_cols = [col[1] for col in cursor.fetchall()] if "coupon_code" not in ord_cols: cursor.execute("ALTER TABLE orders ADD COLUMN coupon_code TEXT") if "discount_amount" not in ord_cols: cursor.execute("ALTER TABLE orders ADD COLUMN discount_amount REAL DEFAULT 0.0") if "device_id" not in ord_cols: cursor.execute("ALTER TABLE orders ADD COLUMN device_id TEXT") # Migrate users table cursor.execute("PRAGMA table_info(users)") usr_cols = [col[1] for col in cursor.fetchall()] if "is_banned" not in usr_cols: cursor.execute("ALTER TABLE users ADD COLUMN is_banned INTEGER DEFAULT 0") if "banned_reason" not in usr_cols: cursor.execute("ALTER TABLE users ADD COLUMN banned_reason TEXT") # Clean up initial default demo coupons so admin can add them on demand cursor.execute("DELETE FROM coupons WHERE code IN ('SLCAT10', 'WELCOME20', 'FLASH50')") cursor.execute("CREATE INDEX IF NOT EXISTS idx_visitors_last_seen ON site_visitors(last_seen)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_orders_status ON orders(status, created_at)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_orders_code ON orders(order_code)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_orders_txref ON orders(tx_reference)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_notif_matched ON bank_notifications(is_matched, amount, created_at)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_banned_devices ON banned_devices(device_id)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_banned_ips ON banned_ips(ip_address)") conn.commit() def create_order( player_id: str, nickname: str, product_id: str, product_name: str, amount: float, qr_payload: str, quantity: int = 1, user_id: Optional[int] = None, customer_name: Optional[str] = None, customer_phone: Optional[str] = None, coupon_code: Optional[str] = None, discount_amount: float = 0.0, device_id: Optional[str] = None, items_json: Optional[str] = None, ) -> str: """Creates a new topup order and returns order_code.""" order_code = f"CAT-{uuid.uuid4().hex[:8].upper()}" now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with get_connection() as conn: cursor = conn.cursor() cursor.execute( """ INSERT INTO orders ( order_code, player_id, nickname, product_id, product_name, quantity, amount, qr_payload, status, user_id, customer_name, customer_phone, coupon_code, discount_amount, device_id, items_json, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'pending_payment', ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( order_code, player_id, nickname, product_id, product_name, quantity, amount, qr_payload, user_id, customer_name, customer_phone, coupon_code, discount_amount, device_id, items_json, now, now, ), ) conn.commit() return order_code def get_order(order_code: str) -> Optional[Dict[str, Any]]: """Retrieves order by order_code. Seamlessly checks both standard orders and Mystery Box orders.""" if not order_code: return None clean_code = str(order_code).strip() with get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT * FROM orders WHERE UPPER(order_code) = UPPER(?)", (clean_code,)) row = cursor.fetchone() if row: return dict(row) # Fallback for Mystery Diamond Box Orders (MBX-XXXXXXXX) try: cursor.execute("SELECT * FROM mystery_box_logs WHERE UPPER(order_code) = UPPER(?)", (clean_code,)) mbx_row = cursor.fetchone() if mbx_row: m = dict(mbx_row) htg_text = m.get("delivery_status") or "Delivered via HTG" if htg_text == "delivered_htg": htg_text = "Delivered via HTG Digital" elif htg_text == "credited_to_wallet": htg_text = "Prize Credited to Wallet" return { "id": m["id"], "order_code": m["order_code"], "user_id": m.get("user_id"), "player_id": m.get("player_id", "N/A"), "nickname": m.get("nickname") or "Player", "product_id": m.get("won_item_id", "mystery_box"), "product_name": f"Mystery Box: {m.get('won_item_name', 'Prize')}", "quantity": 1, "amount": float(m.get("amount_paid", 150.0)), "status": "completed", "payment_method": "Wallet Balance", "tx_reference": f"MBX-WALLET-{m['order_code']}", "htg_order_ids": htg_text, "error_message": None, "created_at": m.get("created_at"), "updated_at": m.get("created_at"), "is_mystery_box": True, } except Exception: pass return None def compute_image_dhash(image_bytes: bytes) -> Optional[str]: """Computes a 64-bit difference hash (dHash) as a 16-hex character string.""" try: import io from PIL import Image img = Image.open(io.BytesIO(image_bytes)).convert("L") img = img.resize((9, 8), Image.Resampling.LANCZOS) pixels = list(img.getdata()) difference = [] for row in range(8): for col in range(8): pixel_left = pixels[row * 9 + col] pixel_right = pixels[row * 9 + col + 1] difference.append(pixel_left > pixel_right) decimal_val = 0 hex_string = [] for index, value in enumerate(difference): if value: decimal_val += 2 ** (index % 4) if (index % 4) == 3: hex_string.append(hex(decimal_val)[2:]) decimal_val = 0 return "".join(hex_string) except Exception: return None def dhash_hamming_distance(h1: str, h2: str) -> int: """Calculates the hamming distance between two 16-hex dHash strings.""" try: n1 = int(str(h1).strip(), 16) n2 = int(str(h2).strip(), 16) x = n1 ^ n2 return bin(x).count("1") except Exception: return 999 def update_order_payment( order_code: str, tx_reference: Optional[str] = None, slip_path: Optional[str] = None, slip_hash: Optional[str] = None, slip_dhash: Optional[str] = None, ) -> bool: """Updates order when client submits payment verification.""" now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with get_connection() as conn: cursor = conn.cursor() cursor.execute( """ UPDATE orders SET status = CASE WHEN status = 'completed' THEN 'completed' ELSE 'payment_submitted' END, tx_reference = COALESCE(?, tx_reference), slip_path = COALESCE(?, slip_path), slip_hash = COALESCE(?, slip_hash), slip_dhash = COALESCE(?, slip_dhash), updated_at = ? WHERE order_code = ? """, (tx_reference, slip_path, slip_hash, slip_dhash, now, order_code), ) conn.commit() return cursor.rowcount > 0 def check_duplicate_slip( slip_hash: Optional[str] = None, slip_dhash: Optional[str] = None, tx_reference: Optional[str] = None, exclude_order_code: Optional[str] = None, ) -> Tuple[bool, str, Optional[Dict[str, Any]], str]: """ Anti-Fraud & Duplicate Order Guard: Checks if this slip image (by SHA-256 or perceptual dHash) or transaction reference has already been submitted or completed for another order. Returns: (is_duplicate: bool, reason: str, conflicting_order: Optional[dict], duplicate_type: str) where duplicate_type is: - "already_completed": previous order was fulfilled (prevents double diamond claiming). - "pending_reorder": previous order is pending/processing/failed (customer re-ordered because diamonds didn't arrive). - "none": not a duplicate. """ if not slip_hash and not slip_dhash and not tx_reference: return False, "", None, "none" exclude_code = str(exclude_order_code or "").strip().upper() with get_connection() as conn: cursor = conn.cursor() conflicting = None # 1. Exact Slip Image Hash match (SHA-256) if slip_hash and len(str(slip_hash).strip()) > 10: clean_hash = str(slip_hash).strip().lower() cursor.execute( """ SELECT id, order_code, player_id, nickname, amount, status, created_at, tx_reference, slip_path FROM orders WHERE slip_hash = ? AND UPPER(order_code) != ? AND status NOT IN ('cancelled', 'expired') ORDER BY id DESC LIMIT 1 """, (clean_hash, exclude_code), ) row = cursor.fetchone() if row: conflicting = dict(row) # 2. Perceptual dHash Match (catches slightly cropped, resized, re-compressed duplicates) if not conflicting and slip_dhash and len(str(slip_dhash).strip()) >= 12: clean_dhash = str(slip_dhash).strip() cursor.execute( """ SELECT id, order_code, player_id, nickname, amount, status, created_at, tx_reference, slip_path, slip_dhash FROM orders WHERE slip_dhash IS NOT NULL AND UPPER(order_code) != ? AND status NOT IN ('cancelled', 'expired') ORDER BY id DESC LIMIT 50 """, (exclude_code,), ) candidates = cursor.fetchall() for cand in candidates: cand_dhash = cand["slip_dhash"] if cand_dhash: dist = dhash_hamming_distance(clean_dhash, cand_dhash) if dist <= 5: # Virtually identical image fingerprint conflicting = dict(cand) break # 3. Exact Transaction Reference match (if valid length >= 6) if not conflicting and tx_reference: clean_ref = str(tx_reference).strip() if len(clean_ref) >= 6 and not clean_ref.lower().startswith("wallet-bal"): cursor.execute( """ SELECT id, order_code, player_id, nickname, amount, status, created_at, tx_reference, slip_path FROM orders WHERE UPPER(TRIM(tx_reference)) = UPPER(?) AND UPPER(order_code) != ? AND status NOT IN ('cancelled', 'expired') ORDER BY id DESC LIMIT 1 """, (clean_ref, exclude_code), ) row = cursor.fetchone() if row: conflicting = dict(row) if conflicting: st = str(conflicting.get("status", "")).lower() c_code = conflicting.get("order_code") if st == "completed": return ( True, f"මෙම Bank Receipt / Slip පත මීට පෙර Order #{c_code} සඳහා භාවිතා කර Diamonds ලබාගෙන අවසන්ය. එකම Slip එකක් ඇණවුම් කිහිපයකට භාවිතා කළ නොහැක.", conflicting, "already_completed", ) else: return ( True, f"මෙම Payment Slip එක ඔබගේ පෙර ඇණවුම (Order #{c_code}) සඳහා දැනටමත් අප වෙත ලැබී ඇත. ඔබ එක් වරක් මුදල් ගෙවා ඇති බැවින් නැවත Orders දැමීමට අවශ්‍ය නොවේ. ඔබගේ මුල් ඇණවුම (#{c_code}) අපගේ Admin කණ්ඩායම විසින් පරීක්ෂා කර කඩිනමින් Diamonds Deliver කරනු ඇත.", conflicting, "pending_reorder", ) return False, "", None, "none" def mark_order_processing(order_code: str, force: bool = False) -> bool: """Atomically marks order as processing to prevent duplicate fulfillments.""" now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with get_connection() as conn: cursor = conn.cursor() if force: cursor.execute( """ UPDATE orders SET status = 'processing', updated_at = ? WHERE order_code = ? AND status != 'completed' """, (now, order_code), ) else: cursor.execute( """ UPDATE orders SET status = 'processing', updated_at = ? WHERE order_code = ? AND status IN ('pending_payment', 'payment_submitted', 'awaiting_bank_confirmation', 'failed') """, (now, order_code), ) conn.commit() return cursor.rowcount > 0 def mark_order_completed(order_code: str, htg_order_ids: str) -> bool: """Marks order as completed with HTG order IDs. Guarantees receipt email triggers only once.""" now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") was_already_completed = False with get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT status FROM orders WHERE order_code = ?", (order_code,)) row = cursor.fetchone() if row and row["status"] == "completed": was_already_completed = True cursor.execute( """ UPDATE orders SET status = 'completed', htg_order_ids = ?, updated_at = ? WHERE order_code = ? """, (htg_order_ids, now, order_code), ) conn.commit() updated = cursor.rowcount > 0 if updated and not was_already_completed: try: _trigger_order_receipt_email(order_code) except Exception: pass return updated def _trigger_order_receipt_email(order_code: str): """Finds recipient email for the order and triggers email delivery receipt with web promo. Guarantees exactly ONE email receipt is ever sent per order code across all threads/processes. """ try: # Atomic check-and-set to guarantee EXACTLY ONE email receipt per order with get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT status, email_receipt_sent FROM orders WHERE order_code = ?", (order_code,)) row = cursor.fetchone() if not row: return if row["email_receipt_sent"]: # Already sent! Never send a duplicate receipt email. return # Atomically mark email_receipt_sent = 1 cursor.execute( "UPDATE orders SET email_receipt_sent = 1 WHERE order_code = ? AND (email_receipt_sent IS NULL OR email_receipt_sent = 0)", (order_code,) ) conn.commit() if cursor.rowcount == 0: # Another concurrent thread/process already claimed and triggered this order's receipt return order = get_order(order_code) if not order: return recipient_email = order.get("customer_email") if not recipient_email and order.get("user_id"): user = get_user_by_id(order["user_id"]) if user and user.get("email"): recipient_email = user["email"] if not recipient_email and order.get("customer_phone"): with get_connection() as conn: cursor = conn.cursor() cursor.execute( "SELECT email FROM users WHERE phone = ? AND email IS NOT NULL AND email != '' ORDER BY id DESC LIMIT 1", (str(order["customer_phone"]).strip(),) ) urow = cursor.fetchone() if urow and urow["email"]: recipient_email = urow["email"] if recipient_email and "@" in str(recipient_email): from mail_service import send_order_receipt_email_async send_order_receipt_email_async(str(recipient_email).strip(), order) except Exception: pass def mark_order_failed(order_code: str, error_message: str) -> bool: """Marks order as failed with reason.""" now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with get_connection() as conn: cursor = conn.cursor() cursor.execute( """ UPDATE orders SET status = 'failed', error_message = ?, updated_at = ? WHERE order_code = ? """, (error_message, now, order_code), ) conn.commit() return cursor.rowcount > 0 def refund_failed_order(order_code: str, reason: str = "HTG Queue Delivery Failed") -> Dict[str, Any]: """ Marks an order as 'failed' and guarantees 100% full refund back to customer wallet. Idempotent: will never double-refund the same order. """ now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") order = get_order(order_code) if not order: return {"success": False, "message": "Order not found"} amt = float(order.get("amount", 0.0)) user_id = order.get("user_id") # If user_id not on order, try to locate user by player_id or phone if not user_id: with get_connection() as conn: cursor = conn.cursor() if order.get("player_id"): cursor.execute("SELECT id FROM users WHERE saved_uid = ? ORDER BY id DESC LIMIT 1", (str(order["player_id"]).strip(),)) u = cursor.fetchone() if u: user_id = u["id"] if not user_id and order.get("customer_phone"): cursor.execute("SELECT id FROM users WHERE phone = ? ORDER BY id DESC LIMIT 1", (str(order["customer_phone"]).strip(),)) u = cursor.fetchone() if u: user_id = u["id"] with get_connection() as conn: cursor = conn.cursor() cursor.execute( """ UPDATE orders SET status = 'failed', error_message = ?, updated_at = ? WHERE order_code = ? """, (reason, now, order_code), ) conn.commit() refunded = False new_bal = None if user_id and amt > 0: ref_code = f"REFUND-{order_code}" with get_connection() as conn: cursor = conn.cursor() cursor.execute( "SELECT id, balance_after FROM wallet_transactions WHERE user_id = ? AND reference_code = ?", (user_id, ref_code), ) already = cursor.fetchone() if not already: c_res = credit_wallet(user_id, amt, f"Auto-Refund: Order {order_code} failed - {reason[:60]}", ref_code) refunded = c_res.get("success", False) new_bal = c_res.get("new_balance") else: refunded = True new_bal = float(already["balance_after"]) return { "success": True, "order_code": order_code, "refunded": refunded, "user_id": user_id, "amount": amt, "new_balance": new_bal, "message": f"Order {order_code} marked failed." + (f" Rs. {amt:.2f} refunded to wallet!" if refunded else " No user wallet linked.") } def auto_reject_stale_orders(max_age_minutes: int = 15) -> int: """ Automatically rejects orders that have been waiting in 'pending_payment' or unconfirmed 'awaiting_bank_confirmation' without ANY submitted reference or slip for more than max_age_minutes (default: 15 mins). CRUCIAL SAFETY: Orders where client submitted a reference or slip (status 'payment_submitted' or has tx_reference/slip_path) are NEVER auto-rejected! They are preserved safely for Admin manual review or automated Bot verification. """ cutoff_time = (datetime.now() - timedelta(minutes=max_age_minutes)).strftime("%Y-%m-%d %H:%M:%S") now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") rejected_count = 0 with get_connection() as conn: cursor = conn.cursor() # Reject ONLY stale unpaid orders where client never submitted a ref or slip cursor.execute( """ UPDATE orders SET status = 'rejected', error_message = 'ගෙවීම් කාලය ඉකුත් විය (Payment window expired)', updated_at = ? WHERE status IN ('pending_payment', 'awaiting_bank_confirmation') AND (tx_reference IS NULL OR tx_reference = '') AND (slip_path IS NULL OR slip_path = '') AND created_at <= ? """, (now_str, cutoff_time), ) rejected_count += cursor.rowcount conn.commit() return rejected_count def generate_unique_order_amount(base_amount: float) -> float: """ Generates a unique order amount by adding random verification cents (+Rs. 0.11 to +Rs. 0.99) to make every pending order's amount 100% mathematically unique. - Guarantees no two active pending orders share the same amount. - Guarantees personal round-figure bank transfers (e.g. Rs. 350.00) NEVER match. - Commercial Bank ePassbook SMS and LankaQR can both auto-accept with 100% safety! """ import random base_int = int(base_amount) cutoff_time = (datetime.now() - timedelta(minutes=15)).strftime("%Y-%m-%d %H:%M:%S") used_cents = set() with get_connection() as conn: cursor = conn.cursor() cursor.execute( """ SELECT amount FROM orders WHERE status IN ('pending_payment', 'awaiting_bank_confirmation') AND created_at >= ? """, (cutoff_time,), ) rows = cursor.fetchall() for r in rows: amt = float(r["amount"]) if int(amt) == base_int: cent = int(round((amt - int(amt)) * 100)) used_cents.add(cent) # Available cents between 11 and 99 (avoiding .00 or already used cents) available_cents = [c for c in range(11, 99) if c not in used_cents] if not available_cents: available_cents = list(range(11, 99)) chosen_cent = random.choice(available_cents) unique_total = round(base_int + (chosen_cent / 100.0), 2) return unique_total def find_pending_order_by_amount(amount: float, reference: Optional[str] = None, max_age_minutes: int = 15) -> Optional[Dict[str, Any]]: """ Finds the most recent pending order matching the amount and/or reference. STRICT TIME-WINDOW: Order must have been created within max_age_minutes (default: 15 mins). Auto-rejects stale orders before querying to guarantee dead orders are never matched! """ auto_reject_stale_orders(max_age_minutes=max_age_minutes) cutoff_time = (datetime.now() - timedelta(minutes=max_age_minutes)).strftime("%Y-%m-%d %H:%M:%S") with get_connection() as conn: cursor = conn.cursor() if reference: cursor.execute( """ SELECT * FROM orders WHERE (status = 'pending_payment' OR status = 'payment_submitted' OR status = 'processing') AND (order_code LIKE ? OR tx_reference LIKE ?) AND created_at >= ? ORDER BY id DESC LIMIT 1 """, (f"%{reference}%", f"%{reference}%", cutoff_time), ) row = cursor.fetchone() if row: return dict(row) cursor.execute( """ SELECT * FROM orders WHERE (status = 'pending_payment' OR status = 'payment_submitted') AND ABS(amount - ?) < 0.01 AND created_at >= ? ORDER BY id DESC LIMIT 1 """, (amount, cutoff_time), ) row = cursor.fetchone() return dict(row) if row else None def auto_reconcile_delivered_orders() -> int: """ Auto-reconciles any orders that were fulfilled/credited but whose status remained in 'payment_submitted' or 'processing' due to client-side reference submissions. """ now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with get_connection() as conn: cursor = conn.cursor() # 1. Wallet deposits already present in wallet_transactions ledger cursor.execute( """ UPDATE orders SET status = 'completed', htg_order_ids = COALESCE(htg_order_ids, 'WALLET-CREDIT-OK'), updated_at = ? WHERE product_id = 'wallet_deposit' AND status != 'completed' AND order_code IN ( SELECT reference_code FROM wallet_transactions WHERE type = 'deposit' AND reference_code IS NOT NULL AND reference_code != '' ) """, (now,), ) reconciled = cursor.rowcount # 2. HTG orders that already have valid htg_order_ids cursor.execute( """ UPDATE orders SET status = 'completed', updated_at = ? WHERE product_id != 'wallet_deposit' AND status != 'completed' AND htg_order_ids IS NOT NULL AND htg_order_ids != '' AND htg_order_ids NOT LIKE '%FAILED%' """, (now,), ) reconciled += cursor.rowcount conn.commit() return reconciled def get_all_orders(limit: int = 200) -> List[Dict[str, Any]]: """Gets list of all recent orders for admin (auto-expires stale orders and auto-reconciles delivered ones).""" auto_reject_stale_orders(max_age_minutes=5) auto_reconcile_delivered_orders() with get_connection() as conn: cursor = conn.cursor() cursor.execute( """ SELECT o.*, bn.source AS bank_source, bn.amount AS bank_amount, bn.reference AS bank_reference, bn.raw_text AS bank_raw_text, bn.created_at AS bank_notified_at FROM orders o LEFT JOIN bank_notifications bn ON (bn.matched_order = o.order_code OR (o.tx_reference IS NOT NULL AND o.tx_reference != '' AND bn.reference = o.tx_reference)) GROUP BY o.id ORDER BY o.id DESC LIMIT ? """, (limit,), ) return [dict(r) for r in cursor.fetchall()] def record_site_visitor( visitor_id: str, ip: str = "", user_agent: str = "", path: str = "/", user_id: Optional[int] = None, ) -> None: """Records or updates a site visitor for accurate live and historical metrics.""" now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") ip_hash = hashlib.sha256(ip.encode()).hexdigest()[:16] if ip else "" ua_clean = (user_agent or "")[:150] path_clean = (path or "/")[:100] with get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT id FROM site_visitors WHERE visitor_id = ?", (visitor_id,)) existing = cursor.fetchone() if existing: cursor.execute( """ UPDATE site_visitors SET last_seen = ?, last_path = ?, visit_count = visit_count + 1, user_id = COALESCE(?, user_id) WHERE visitor_id = ? """, (now, path_clean, user_id, visitor_id), ) else: cursor.execute( """ INSERT INTO site_visitors ( visitor_id, ip_hash, user_agent, last_path, visit_count, user_id, first_seen, last_seen ) VALUES (?, ?, ?, ?, 1, ?, ?, ?) """, (visitor_id, ip_hash, ua_clean, path_clean, user_id, now, now), ) conn.commit() def get_visitor_stats() -> Dict[str, Any]: """Retrieves live online users, today's unique visitors, and total visitors.""" now = datetime.now() cutoff_live = (now - timedelta(minutes=5)).strftime("%Y-%m-%d %H:%M:%S") today_start = now.strftime("%Y-%m-%d 00:00:00") with get_connection() as conn: cursor = conn.cursor() # Live users active in the last 5 minutes cursor.execute("SELECT COUNT(*) as live_count FROM site_visitors WHERE last_seen >= ?", (cutoff_live,)) live_count = cursor.fetchone()["live_count"] # Today's unique visitors cursor.execute("SELECT COUNT(*) as today_count FROM site_visitors WHERE last_seen >= ?", (today_start,)) today_count = cursor.fetchone()["today_count"] # Total unique visitors all-time & total pageviews cursor.execute("SELECT COUNT(*) as total_count, COALESCE(SUM(visit_count), 0) as total_pageviews FROM site_visitors") tot_row = cursor.fetchone() total_count = tot_row["total_count"] total_pageviews = tot_row["total_pageviews"] # Total registered users cursor.execute("SELECT COUNT(*) as total_users FROM users") total_users = cursor.fetchone()["total_users"] # Total unique player UIDs in orders cursor.execute("SELECT COUNT(DISTINCT player_id) as total_players FROM orders") total_players = cursor.fetchone()["total_players"] return { "live_visitors": live_count, "today_visitors": today_count, "total_visitors": total_count, "total_pageviews": total_pageviews, "total_users": total_users, "total_players": total_players, } def get_admin_stats() -> Dict[str, Any]: """Calculates statistics for admin dashboard.""" with get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT COUNT(*) as total_orders, COALESCE(SUM(amount), 0) as total_volume FROM orders") tot = dict(cursor.fetchone()) today = datetime.now().strftime("%Y-%m-%d") cursor.execute( "SELECT COUNT(*) as today_orders, COALESCE(SUM(amount), 0) as today_volume FROM orders WHERE created_at LIKE ?", (f"{today}%",), ) tod = dict(cursor.fetchone()) cursor.execute("SELECT COUNT(*) as pending_count FROM orders WHERE status = 'payment_submitted'") pen = dict(cursor.fetchone()) v_stats = get_visitor_stats() return { "total_orders": tot["total_orders"], "total_volume": tot["total_volume"], "today_orders": tod["today_orders"], "today_volume": tod["today_volume"], "pending_count": pen["pending_count"], **v_stats, } def record_bank_notification( source: str, amount: float, reference: Optional[str] = None, raw_text: str = "", ) -> Optional[int]: """ Stores incoming bank notification, strictly deduplicating identical alerts in the last 2 hours. Returns notification ID, or None if this notification is a duplicate. """ now = datetime.now() now_str = now.strftime("%Y-%m-%d %H:%M:%S") cutoff_dedup = (now - timedelta(hours=2)).strftime("%Y-%m-%d %H:%M:%S") clean_text = raw_text.strip()[:500] with get_connection() as conn: cursor = conn.cursor() # Deduplication check: Has this exact message text or reference for this amount arrived in the last 2 hours? if reference and len(reference) >= 6: cursor.execute( """ SELECT id FROM bank_notifications WHERE (reference = ? OR raw_text = ?) AND created_at >= ? LIMIT 1 """, (reference, clean_text, cutoff_dedup), ) else: cursor.execute( """ SELECT id FROM bank_notifications WHERE raw_text = ? AND created_at >= ? LIMIT 1 """, (clean_text, cutoff_dedup), ) existing = cursor.fetchone() if existing: # Duplicate alert - do not store again to prevent double fulfillment! return None cursor.execute( """ INSERT INTO bank_notifications (source, amount, reference, raw_text, is_matched, created_at) VALUES (?, ?, ?, ?, 0, ?) """, (source, amount, reference, clean_text, now_str), ) conn.commit() return cursor.lastrowid def find_matching_notification( amount: float, reference: Optional[str] = None, require_reference: bool = False, max_age_minutes: int = 5, min_timestamp: Optional[str] = None, allow_lankaqr_only_without_ref: bool = True, ) -> Optional[Dict[str, Any]]: """ Checks if an unmatched bank notification exists for this amount or reference. STRICT TIME-WINDOW: Only considers notifications received within max_age_minutes (default: 5 mins). If min_timestamp is given, notification must have arrived at or after that time (e.g. order creation time). SECURITY: If require_reference is False, only matches LankaQR/HelaPOS alerts by default so that personal bank deposits (e.g. Commercial Bank ePassbook) are never blindly matched! """ cutoff_time = (datetime.now() - timedelta(minutes=max_age_minutes)).strftime("%Y-%m-%d %H:%M:%S") if min_timestamp and min_timestamp > cutoff_time: cutoff_time = min_timestamp with get_connection() as conn: cursor = conn.cursor() if reference: clean_ref = "".join(c for c in reference if c.isalnum()) if len(clean_ref) >= 4: cursor.execute( """ SELECT * FROM bank_notifications WHERE is_matched = 0 AND (reference LIKE ? OR raw_text LIKE ?) AND created_at >= ? ORDER BY id DESC LIMIT 1 """, (f"%{clean_ref}%", f"%{clean_ref}%", cutoff_time), ) row = cursor.fetchone() if row: return dict(row) # If reference was specifically required to match, do NOT fallback to pure amount if require_reference: return None # Fallback: only if strict reference matching is not requested if not require_reference: if allow_lankaqr_only_without_ref: # If amount has unique verification cents (10 to 99), it is 100% a store order (even from ePassbook SMS!) cents = round((amount - int(amount)) * 100) if 10 <= cents <= 99: cursor.execute( """ SELECT * FROM bank_notifications WHERE is_matched = 0 AND ABS(amount - ?) < 0.01 AND created_at >= ? ORDER BY id DESC LIMIT 1 """, (amount, cutoff_time), ) else: # Round amount without cents: Strictly only match LankaQR/HelaPOS/Seylan alerts. # Personal ePassbook/SMS deposits without cents will NEVER auto-match! cursor.execute( """ SELECT * FROM bank_notifications WHERE is_matched = 0 AND ABS(amount - ?) < 0.01 AND (source LIKE '%HelaPOS%' OR source LIKE '%Seylan%' OR raw_text LIKE '%LankaQR%' OR raw_text LIKE '%HelaPOS%') AND created_at >= ? ORDER BY id DESC LIMIT 1 """, (amount, cutoff_time), ) else: cursor.execute( """ SELECT * FROM bank_notifications WHERE is_matched = 0 AND ABS(amount - ?) < 0.01 AND created_at >= ? ORDER BY id DESC LIMIT 1 """, (amount, cutoff_time), ) row = cursor.fetchone() return dict(row) if row else None return None def mark_notification_matched(notification_id: int, order_code: str) -> bool: """Marks a bank notification as matched with an order.""" with get_connection() as conn: cursor = conn.cursor() cursor.execute( "UPDATE bank_notifications SET is_matched = 1, matched_order = ? WHERE id = ?", (order_code, notification_id), ) conn.commit() return cursor.rowcount > 0 def get_setting(key: str, default: str = "") -> str: """Retrieves a configuration setting value.""" with get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT value FROM settings WHERE key = ?", (key,)) row = cursor.fetchone() return row["value"] if row else default def set_setting(key: str, value: str): """Sets or updates a configuration setting value.""" with get_connection() as conn: cursor = conn.cursor() cursor.execute("INSERT OR REPLACE INTO settings (key, value) VALUES (?, ?)", (key, str(value))) conn.commit() def reject_order(order_code: str, reason: str = "Rejected by Admin") -> bool: """Rejects an order and saves the rejection reason.""" now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with get_connection() as conn: cursor = conn.cursor() cursor.execute( "UPDATE orders SET status = 'rejected', error_message = ?, updated_at = ? WHERE order_code = ?", (reason, now, order_code), ) conn.commit() return cursor.rowcount > 0 # ===================================================================== # CLIENT USER & AUTHENTICATION FUNCTIONS # ===================================================================== def hash_pin(pin: str) -> str: """Hashes 4-digit PIN using sha256 with fixed salt.""" salt = "slcat_topup_pin_salt_998" return hashlib.sha256(f"{salt}{pin}".encode()).hexdigest() def normalize_phone(phone: str) -> str: """Cleans phone numbers into a standard 10-digit format (e.g. 0771234567).""" cleaned = "".join(c for c in str(phone) if c.isdigit()) if cleaned.startswith("94") and len(cleaned) == 11: cleaned = "0" + cleaned[2:] return cleaned def register_user(name: str, phone: str, pin: str, email: str = "") -> Dict[str, Any]: """Registers a new client user with name, phone, 4-digit PIN, and optional email.""" clean_name = str(name).strip() clean_phone = normalize_phone(phone) pin_str = str(pin).strip() clean_email = str(email or "").strip().lower() if not clean_name or len(clean_name) < 2: return {"success": False, "message": "කරුණාකර ඔබගේ නම ඇතුළත් කරන්න (Name required)."} if not clean_phone or len(clean_phone) < 9: return {"success": False, "message": "වලංගු WhatsApp හෝ Mobile අංකයක් ඇතුළත් කරන්න (Valid phone required)."} if not pin_str.isdigit() or len(pin_str) != 4: return {"success": False, "message": "PIN අංකය ඉලක්කම් 4ක් පමණක් විය යුතුය (4-digit PIN required, e.g. 1234)."} now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") pin_hash = hash_pin(pin_str) with get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT id FROM users WHERE phone = ?", (clean_phone,)) if cursor.fetchone(): return {"success": False, "message": "මෙම දුරකථන අංකය දැනටමත් ලියාපදිංචි වී ඇත. කරුණාකර Login වන්න."} cursor.execute( """ INSERT INTO users (name, phone, email, pin_hash, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?) """, (clean_name, clean_phone, clean_email or None, pin_hash, now, now), ) conn.commit() user_id = cursor.lastrowid return { "success": True, "user": { "id": user_id, "name": clean_name, "phone": clean_phone, "email": clean_email or None, "email_verified": 0, "saved_uid": None, "saved_nickname": None, }, } def authenticate_user(phone: str, pin: str) -> Dict[str, Any]: """Authenticates a user with phone and 4-digit PIN.""" clean_phone = normalize_phone(phone) pin_str = str(pin).strip() if not clean_phone: return {"success": False, "message": "දුරකථන අංකය ඇතුළත් කරන්න."} if not pin_str or len(pin_str) != 4: return {"success": False, "message": "ඉලක්කම් 4ක PIN අංකය ඇතුළත් කරන්න."} with get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT * FROM users WHERE phone = ?", (clean_phone,)) row = cursor.fetchone() if not row: return {"success": False, "message": "මෙම අංකයෙන් ගිණුමක් හමු නොවීය. කරුණාකර Register වන්න."} user = dict(row) if user["pin_hash"] != hash_pin(pin_str): return {"success": False, "message": "ඇතුළත් කළ 4-Digit PIN අංකය වැරදියි (Incorrect PIN)."} return { "success": True, "user": { "id": user["id"], "name": user["name"], "phone": user["phone"], "email": user.get("email"), "email_verified": int(user.get("email_verified") or 0), "saved_uid": user.get("saved_uid"), "saved_nickname": user.get("saved_nickname"), }, } def get_user_by_id(user_id: int) -> Optional[Dict[str, Any]]: """Retrieves user profile by ID.""" with get_connection() as conn: cursor = conn.cursor() cursor.execute( "SELECT id, name, phone, email, email_verified, wallet_balance, saved_uid, saved_nickname, avatar, created_at FROM users WHERE id = ?", (user_id,), ) row = cursor.fetchone() return dict(row) if row else None def update_user_profile(user_id: int, name: str, avatar: str = "👤", current_pin: str = "", new_pin: str = "", email: Optional[str] = None) -> Dict[str, Any]: """Updates user's display name, avatar, email, and optionally changes their 4-digit PIN.""" now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") clean_name = (name or "").strip() clean_avatar = (avatar or "👤").strip() if not clean_name: return {"success": False, "message": "නම හිස්ව තැබිය නොහැක (Name cannot be empty)."} with get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT pin_hash, email, email_verified FROM users WHERE id = ?", (user_id,)) row = cursor.fetchone() if not row: return {"success": False, "message": "පරිශීලක ගිණුම හමු නොවීය (User not found)."} current_email = (row["email"] or "").strip().lower() current_verified = int(row["email_verified"] or 0) email_changed = False target_email = current_email target_verified = current_verified if email is not None: clean_email = str(email).strip().lower() if clean_email: if "@" not in clean_email or "." not in clean_email: return {"success": False, "message": "වලංගු ඊමේල් ලිපිනයක් ඇතුළත් කරන්න (Invalid email)."} if clean_email != current_email: email_changed = True target_email = clean_email target_verified = 0 else: if current_email != "": email_changed = True target_email = None target_verified = 0 # If user wants to change PIN if new_pin: clean_curr = str(current_pin).strip() clean_new = str(new_pin).strip() if not clean_curr: return {"success": False, "message": "නව PIN අංකයක් සැකසීමට වත්මන් PIN අංකය ඇතුළත් කරන්න."} if row["pin_hash"] != hash_pin(clean_curr): return {"success": False, "message": "වත්මන් PIN අංකය වැරදියි (Current PIN is incorrect)."} if len(clean_new) != 4 or not clean_new.isdigit(): return {"success": False, "message": "නව PIN අංකය ඉලක්කම් 4කින් සමන්විත විය යුතුය (Must be 4 digits)."} new_hash = hash_pin(clean_new) cursor.execute( """ UPDATE users SET name = ?, avatar = ?, email = ?, email_verified = ?, pin_hash = ?, updated_at = ? WHERE id = ? """, (clean_name, clean_avatar, target_email, target_verified, new_hash, now, user_id), ) else: cursor.execute( """ UPDATE users SET name = ?, avatar = ?, email = ?, email_verified = ?, updated_at = ? WHERE id = ? """, (clean_name, clean_avatar, target_email, target_verified, now, user_id), ) conn.commit() return { "success": True, "message": "ඔබගේ Profile එක සාර්ථකව යාවත්කාලීන විය! (Profile updated successfully)", "name": clean_name, "avatar": clean_avatar, "email": target_email or "", "email_verified": target_verified, "email_changed": email_changed, } def update_user_saved_uid(user_id: int, uid: str, nickname: str = "") -> bool: """Updates user's preferred Free Fire Player ID and nickname for 1-click ordering.""" now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with get_connection() as conn: cursor = conn.cursor() cursor.execute( "UPDATE users SET saved_uid = ?, saved_nickname = ?, updated_at = ? WHERE id = ?", (str(uid).strip(), str(nickname).strip(), now, user_id), ) conn.commit() return cursor.rowcount > 0 def set_user_email_otp(user_id: int, email: str, otp_code: str, expiry_minutes: int = 10) -> bool: """Stores pending email and 6-digit OTP code with expiry time for user.""" from datetime import timedelta clean_email = str(email).strip().lower() now = datetime.now() expiry = (now + timedelta(minutes=expiry_minutes)).strftime("%Y-%m-%d %H:%M:%S") now_str = now.strftime("%Y-%m-%d %H:%M:%S") with get_connection() as conn: cursor = conn.cursor() cursor.execute( """ UPDATE users SET email = ?, email_otp = ?, email_otp_expiry = ?, email_verified = 0, updated_at = ? WHERE id = ? """, (clean_email, str(otp_code).strip(), expiry, now_str, user_id), ) conn.commit() return cursor.rowcount > 0 def verify_user_email_otp(user_id: int, email: str, otp_code: str) -> Dict[str, Any]: """Verifies submitted OTP against saved OTP and marks email as verified.""" clean_email = str(email).strip().lower() clean_otp = str(otp_code).strip() now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with get_connection() as conn: cursor = conn.cursor() cursor.execute( "SELECT id, email, email_otp, email_otp_expiry FROM users WHERE id = ?", (user_id,), ) row = cursor.fetchone() if not row: return {"success": False, "message": "පරිශීලක ගිණුම හමු නොවීය."} saved_otp = str(row["email_otp"] or "").strip() saved_expiry = str(row["email_otp_expiry"] or "").strip() if not saved_otp: return {"success": False, "message": "OTP කේතයක් නිකුත් කර නොමැත. කරුණාකර නැවත OTP එකක් ඉල්ලන්න."} if saved_expiry and now_str > saved_expiry: return {"success": False, "message": "OTP කේතයේ කාලය ඉකුත් වී ඇත (OTP Expired). කරුණාකර නැවත OTP එකක් ඉල්ලන්න."} if clean_otp != saved_otp: return {"success": False, "message": "ඇතුළත් කළ OTP අංකය වැරදියි (Incorrect OTP)."} cursor.execute( """ UPDATE users SET email = ?, email_verified = 1, email_otp = NULL, email_otp_expiry = NULL, updated_at = ? WHERE id = ? """, (clean_email, now_str, user_id), ) conn.commit() return { "success": True, "message": "ඔබගේ Email ලිපිනය සාර්ථකව Verify කරන ලදී! සියලුම Order Receipts සහ Promo Offers මෙම ලිපිනයට ලැබෙනු ඇත.", "email": clean_email, } def reset_user_pin_verified(phone: str, verify_value: str, new_pin: str) -> Dict[str, Any]: """ Allows a customer to self-reset their forgotten 4-digit PIN by verifying either: 1) Their registered/saved Free Fire UID (saved_uid), or 2) The Free Fire Player ID from any order placed with this phone/account, or 3) Any past Order Code (e.g. SLCAT-XXXXX) placed with this phone/account. """ clean_phone = normalize_phone(phone) if not clean_phone or len(clean_phone) < 9: return {"success": False, "message": "වලංගු දුරකථන අංකයක් ඇතුළත් කරන්න (Enter valid phone number)."} clean_pin = str(new_pin).strip() if len(clean_pin) != 4 or not clean_pin.isdigit(): return {"success": False, "message": "නව PIN අංකය ඉලක්කම් 4කින් සමන්විත විය යුතුය (PIN must be 4 digits, e.g. 1234)."} clean_v = str(verify_value).strip() if not clean_v: return {"success": False, "message": "තහවුරු කිරීම සඳහා ඔබගේ Free Fire UID (Player ID) හෝ පෙර Order Code එකක් ඇතුළත් කරන්න."} now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT id, name, phone, saved_uid FROM users WHERE phone = ?", (clean_phone,)) user = cursor.fetchone() if not user: return {"success": False, "message": "මෙම දුරකථන අංකයෙන් ලියාපදිංචි වූ ගිණුමක් හමු නොවීය (Account not found)."} user_id = user["id"] saved_uid = str(user["saved_uid"] or "").strip() matched = False # 1. Match saved_uid if saved_uid and (clean_v == saved_uid or clean_v == saved_uid.lstrip("0")): matched = True # 2. Match player_id from past orders if not matched: cursor.execute( "SELECT id FROM orders WHERE (user_id = ? OR customer_phone = ?) AND (player_id = ? OR player_id = ?) LIMIT 1", (user_id, clean_phone, clean_v, clean_v.lstrip("0")), ) if cursor.fetchone(): matched = True # 3. Match order_code from past orders if not matched: cursor.execute( "SELECT id FROM orders WHERE (user_id = ? OR customer_phone = ?) AND UPPER(order_code) = UPPER(?) LIMIT 1", (user_id, clean_phone, clean_v), ) if cursor.fetchone(): matched = True if not matched: return { "success": False, "message": "තහවුරු කිරීම අසාර්ථකයි! ඇතුළත් කළ Free Fire UID හෝ Order Code එක මෙම දුරකථන අංකයට ගැලපෙන්නේ නැත. කරුණාකර නිවැරදි UID හෝ පෙර Order Code එකක් ඇතුළත් කරන්න.", } # Update PIN new_hash = hash_pin(clean_pin) cursor.execute( "UPDATE users SET pin_hash = ?, updated_at = ? WHERE id = ?", (new_hash, now, user_id), ) conn.commit() return { "success": True, "message": "ඔබගේ 4-Digit PIN අංකය සාර්ථකව වෙනස් කරන ලදී! දැන් නව PIN අංකයෙන් Login විය හැක.", "user": { "id": user["id"], "name": user["name"], "phone": user["phone"], }, } def admin_reset_user_pin(user_id: int, new_pin: str = "1234") -> Dict[str, Any]: """Allows admin to reset a user's PIN directly to default or custom value.""" clean_pin = str(new_pin).strip() if len(clean_pin) != 4 or not clean_pin.isdigit(): return {"success": False, "message": "PIN must be 4 digits"} now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with get_connection() as conn: cursor = conn.cursor() cursor.execute( "UPDATE users SET pin_hash = ?, updated_at = ? WHERE id = ?", (hash_pin(clean_pin), now, user_id), ) conn.commit() return {"success": True, "message": f"User PIN has been reset to {clean_pin}"} def get_user_orders(user_id: int, limit: int = 30) -> List[Dict[str, Any]]: """Retrieves order history for a specific logged in user.""" with get_connection() as conn: cursor = conn.cursor() cursor.execute( """ SELECT order_code, player_id, nickname, product_name, amount, status, htg_order_ids, error_message, created_at FROM orders WHERE user_id = ? ORDER BY id DESC LIMIT ? """, (user_id, limit), ) return [dict(r) for r in cursor.fetchall()] # ===================================================================== # CLIENT WALLET FUNCTIONS # ===================================================================== def get_user_wallet_balance(user_id: int) -> float: """Gets current wallet balance for user.""" with get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT wallet_balance FROM users WHERE id = ?", (user_id,)) row = cursor.fetchone() return float(row["wallet_balance"]) if row and row["wallet_balance"] is not None else 0.0 def credit_wallet(user_id: int, amount: float, description: str, reference_code: str = "") -> Dict[str, Any]: """Adds funds to user wallet and writes ledger record with idempotency protection.""" if amount <= 0: return {"success": False, "message": "Invalid amount"} now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with get_connection() as conn: cursor = conn.cursor() # Idempotency guard: prevent crediting same deposit transaction twice if reference_code: cursor.execute( """ SELECT id, balance_after FROM wallet_transactions WHERE user_id = ? AND reference_code = ? AND type = 'deposit' LIMIT 1 """, (user_id, reference_code), ) existing_tx = cursor.fetchone() if existing_tx: return { "success": True, "already_credited": True, "message": "Deposit already credited", "new_balance": float(existing_tx["balance_after"]), } cursor.execute("SELECT wallet_balance FROM users WHERE id = ?", (user_id,)) row = cursor.fetchone() if not row: return {"success": False, "message": "User not found"} current_balance = float(row["wallet_balance"] or 0.0) new_balance = round(current_balance + amount, 2) cursor.execute( "UPDATE users SET wallet_balance = ?, updated_at = ? WHERE id = ?", (new_balance, now, user_id), ) cursor.execute( """ INSERT INTO wallet_transactions (user_id, type, amount, balance_after, description, reference_code, created_at) VALUES (?, 'deposit', ?, ?, ?, ?, ?) """, (user_id, amount, new_balance, description, reference_code, now), ) conn.commit() return {"success": True, "new_balance": new_balance} def debit_wallet(user_id: int, amount: float, description: str, reference_code: str = "") -> Dict[str, Any]: """Deducts funds from user wallet if balance is sufficient and writes ledger record.""" if amount <= 0: return {"success": False, "message": "Invalid amount"} now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT wallet_balance FROM users WHERE id = ?", (user_id,)) row = cursor.fetchone() if not row: return {"success": False, "message": "User not found"} current_balance = float(row["wallet_balance"] or 0.0) if current_balance < amount: return { "success": False, "insufficient_balance": True, "current_balance": current_balance, "message": f"ප්‍රමාණවත් Wallet Balance එකක් නොමැත (Available: Rs. {current_balance:.2f}). කරුණාකර Wallet එක Top-up කරන්න.", } new_balance = round(current_balance - amount, 2) cursor.execute( "UPDATE users SET wallet_balance = ?, updated_at = ? WHERE id = ?", (new_balance, now, user_id), ) cursor.execute( """ INSERT INTO wallet_transactions (user_id, type, amount, balance_after, description, reference_code, created_at) VALUES (?, 'purchase', ?, ?, ?, ?, ?) """, (user_id, amount, new_balance, description, reference_code, now), ) conn.commit() return {"success": True, "new_balance": new_balance} def get_wallet_transactions(user_id: int, limit: int = 30) -> List[Dict[str, Any]]: """Retrieves wallet ledger transactions for a user.""" with get_connection() as conn: cursor = conn.cursor() cursor.execute( """ SELECT * FROM wallet_transactions WHERE user_id = ? ORDER BY id DESC LIMIT ? """, (user_id, limit), ) return [dict(r) for r in cursor.fetchall()] def get_all_users_with_wallets(search: str = "", limit: int = 100) -> List[Dict[str, Any]]: """Returns list of registered users with wallet balance, phone, and name for admin.""" with get_connection() as conn: cursor = conn.cursor() if search: query = """ SELECT id, name, phone, wallet_balance, saved_uid, saved_nickname, created_at, updated_at FROM users WHERE phone LIKE ? OR name LIKE ? OR saved_uid LIKE ? ORDER BY wallet_balance DESC, id DESC LIMIT ? """ s = f"%{search.strip()}%" cursor.execute(query, (s, s, s, limit)) else: query = """ SELECT id, name, phone, wallet_balance, saved_uid, saved_nickname, created_at, updated_at FROM users ORDER BY wallet_balance DESC, id DESC LIMIT ? """ cursor.execute(query, (limit,)) return [dict(r) for r in cursor.fetchall()] def admin_adjust_wallet(user_id: int, action: str, amount: float, reason: str = "") -> Dict[str, Any]: """Admin function to add (+) or deduct (-) funds from a user's wallet with ledger tracking.""" if amount <= 0: return {"success": False, "message": "Amount must be greater than 0"} user = get_user_by_id(user_id) if not user: return {"success": False, "message": "User not found"} now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") ref_code = f"ADMIN-ADJ-{int(datetime.now().timestamp())}" with get_connection() as conn: cursor = conn.cursor() current_balance = float(user.get("wallet_balance") or 0.0) if action == "add": new_balance = round(current_balance + amount, 2) desc = f"Admin Deposit (+Rs. {amount:.2f})" if reason: desc += f" - {reason}" cursor.execute("UPDATE users SET wallet_balance = ?, updated_at = ? WHERE id = ?", (new_balance, now, user_id)) cursor.execute( """ INSERT INTO wallet_transactions (user_id, type, amount, balance_after, description, reference_code, created_at) VALUES (?, 'deposit', ?, ?, ?, ?, ?) """, (user_id, amount, new_balance, desc, ref_code, now), ) conn.commit() return { "success": True, "action": "add", "amount": amount, "new_balance": new_balance, "message": f"Rs. {amount:.2f} successfully added to wallet! New balance: Rs. {new_balance:.2f}", } elif action == "deduct": new_balance = round(max(0.0, current_balance - amount), 2) desc = f"Admin Deduction (-Rs. {amount:.2f})" if reason: desc += f" - {reason}" cursor.execute("UPDATE users SET wallet_balance = ?, updated_at = ? WHERE id = ?", (new_balance, now, user_id)) cursor.execute( """ INSERT INTO wallet_transactions (user_id, type, amount, balance_after, description, reference_code, created_at) VALUES (?, 'payment', ?, ?, ?, ?, ?) """, (user_id, amount, new_balance, desc, ref_code, now), ) conn.commit() return { "success": True, "action": "deduct", "amount": amount, "new_balance": new_balance, "message": f"Rs. {amount:.2f} deducted from wallet. New balance: Rs. {new_balance:.2f}", } else: return {"success": False, "message": "Invalid action. Must be 'add' or 'deduct'."} # ========================================================================= # Spin & Win Wheel System # ========================================================================= def can_user_spin(user_id: Optional[int] = None, phone: Optional[str] = None) -> Dict[str, Any]: """ Checks if a user is eligible to spin today (once per day). """ today_str = datetime.now().strftime("%Y-%m-%d") with get_connection() as conn: cursor = conn.cursor() if user_id: cursor.execute( "SELECT * FROM spins WHERE (user_id = ? OR (phone = ? AND phone IS NOT NULL)) AND spin_date = ?", (user_id, phone, today_str), ) elif phone: cursor.execute( "SELECT * FROM spins WHERE phone = ? AND spin_date = ?", (phone, today_str), ) else: return {"can_spin": True, "spin_date": today_str} existing = cursor.fetchone() if existing: return { "can_spin": False, "spin_date": today_str, "reward_label": existing["reward_label"], "reward_type": existing["reward_type"], "message": "ඔබ අද දිනට හිමි වාසනා රෝදය කරකවා ඇත. හෙට දින නැවත උත්සාහ කරන්න! (Available Tomorrow)", } return {"can_spin": True, "spin_date": today_str} def record_spin( reward_type: str, reward_label: str, reward_amount: float = 0.0, user_id: Optional[int] = None, phone: Optional[str] = None, ) -> Dict[str, Any]: """ Records a completed spin. If the user won wallet balance, credits their wallet immediately. """ now = datetime.now() now_str = now.strftime("%Y-%m-%d %H:%M:%S") today_str = now.strftime("%Y-%m-%d") with get_connection() as conn: cursor = conn.cursor() cursor.execute( """ INSERT INTO spins (user_id, phone, spin_date, reward_type, reward_label, reward_amount, claimed, created_at) VALUES (?, ?, ?, ?, ?, ?, 1, ?) """, (user_id, phone, today_str, reward_type, reward_label, reward_amount, now_str), ) spin_id = cursor.lastrowid conn.commit() # If won wallet cash and user is logged in, credit their account if reward_type == "wallet_cash" and reward_amount > 0 and user_id: credit_wallet( user_id=user_id, amount=reward_amount, description=f"🎡 Daily Spin & Win Bonus: {reward_label}", reference_code=f"SPIN-{spin_id}", ) return {"success": True, "spin_id": spin_id, "reward_type": reward_type, "reward_label": reward_label} # ========================================================================= # Monthly Top-Up Leaderboard # ========================================================================= def get_monthly_leaderboard(month_str: Optional[str] = None, limit: int = 10) -> List[Dict[str, Any]]: """ Retrieves the top customers/players for the specified month (default: current month YYYY-MM). Calculates total top-up amount, order count, and estimated Free Fire diamonds. """ if not month_str: month_str = datetime.now().strftime("%Y-%m") with get_connection() as conn: cursor = conn.cursor() cursor.execute( """ SELECT player_id, MAX(COALESCE(nickname, customer_name, 'Player')) as player_name, COUNT(id) as total_orders, SUM(amount) as total_spent FROM orders WHERE status = 'completed' AND created_at LIKE ? GROUP BY player_id ORDER BY total_spent DESC LIMIT ? """, (f"{month_str}%", limit), ) rows = cursor.fetchall() leaderboard = [] for rank, r in enumerate(rows, start=1): spent = float(r["total_spent"] or 0.0) # Estimate diamonds based on typical rate (Rs. 2.80 per diamond) est_diamonds = int(spent / 2.8) leaderboard.append({ "rank": rank, "player_id": r["player_id"], "masked_id": r["player_id"][:3] + "****" + r["player_id"][-2:] if len(r["player_id"]) > 5 else r["player_id"], "player_name": r["player_name"], "total_orders": r["total_orders"], "total_spent": round(spent, 2), "diamonds": est_diamonds, "badge": "🥇 Grandmaster" if rank == 1 else ("🥈 Master" if rank == 2 else ("🥉 Diamond King" if rank == 3 else f"Top #{rank} Heroic")), }) return leaderboard # ========================================================================= # Live Recent Purchases Ticker # ========================================================================= def get_recent_public_purchases(limit: int = 10) -> List[Dict[str, Any]]: """ Retrieves recent completed purchases for the social proof live ticker. Anonymizes player IDs and phone numbers for privacy. """ with get_connection() as conn: cursor = conn.cursor() cursor.execute( """ SELECT order_code, player_id, nickname, product_name, amount, created_at FROM orders WHERE status = 'completed' ORDER BY id DESC LIMIT ? """, (limit,), ) rows = cursor.fetchall() purchases = [] for r in rows: uid = r["player_id"] or "" masked = uid[:3] + "****" + uid[-2:] if len(uid) > 5 else uid name = r["nickname"] or "Free Fire Player" purchases.append({ "order_code": r["order_code"], "player_name": name, "masked_uid": masked, "product_name": r["product_name"], "amount": r["amount"], "created_at": r["created_at"], }) return purchases # ========================================================================= # Coupons & Discounts # ========================================================================= def validate_coupon(code: str, amount: float) -> Dict[str, Any]: """Validates coupon code against order amount and usage limits.""" code = (code or "").strip().upper() if not code: return {"valid": False, "message": "කරුණාකර Coupon Code එකක් ඇතුළත් කරන්න."} now_iso = datetime.now().isoformat() with get_connection() as conn: cursor = conn.cursor() cursor.execute( """ SELECT * FROM coupons WHERE code = ? AND is_active = 1 """, (code,), ) row = cursor.fetchone() if not row: return {"valid": False, "message": "වලංගු නොවන Coupon Code එකකි."} coupon = dict(row) if coupon.get("expires_at") and coupon["expires_at"] < now_iso: return {"valid": False, "message": "මෙම Coupon එකේ වලංගු කාලය ඉකුත් වී ඇත."} if coupon.get("max_uses") and coupon.get("used_count", 0) >= coupon["max_uses"]: return {"valid": False, "message": "මෙම Coupon එක භාවිත කිරීමේ සීමාව ඉක්මවා ඇත."} min_amt = float(coupon.get("min_order_amount") or 0.0) if amount < min_amt: return { "valid": False, "message": f"මෙම Coupon එක භාවිත කිරීමට අවම වශයෙන් රු. {min_amt:,.2f} ක ඇණවුමක් තිබිය යුතුය.", } disc_type = coupon.get("discount_type", "fixed") disc_val = float(coupon.get("discount_value") or 0.0) if disc_type == "percent": discount = round((amount * disc_val) / 100.0, 2) else: discount = round(min(disc_val, amount), 2) new_amount = max(1.0, round(amount - discount, 2)) return { "valid": True, "coupon_code": code, "discount_amount": discount, "original_amount": amount, "new_amount": new_amount, "message": f"🎉 Coupon එක සාර්ථකයි! රු. {discount:,.2f} ක වට්ටමක් ලැබුණි.", } def record_coupon_usage(code: str) -> None: """Increments coupon used_count.""" if not code: return with get_connection() as conn: cursor = conn.cursor() cursor.execute( "UPDATE coupons SET used_count = used_count + 1 WHERE code = ?", (code.strip().upper(),), ) conn.commit() def get_all_coupons() -> List[Dict[str, Any]]: """Fetches all coupons for admin management.""" with get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT * FROM coupons ORDER BY id DESC") return [dict(r) for r in cursor.fetchall()] def add_coupon( code: str, discount_type: str = "fixed", discount_value: float = 10.0, min_order_amount: float = 0.0, max_uses: int = 1000, expires_at: Optional[str] = None, is_active: int = 1, ) -> Dict[str, Any]: """Adds a new promo code/coupon.""" code = (code or "").strip().upper() if not code: return {"success": False, "message": "Coupon code is required"} now_iso = datetime.now().isoformat() with get_connection() as conn: cursor = conn.cursor() try: cursor.execute( """ INSERT INTO coupons (code, discount_type, discount_value, min_order_amount, max_uses, expires_at, is_active, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, (code, discount_type, float(discount_value), float(min_order_amount), int(max_uses), expires_at, int(is_active), now_iso) ) conn.commit() return {"success": True, "message": f"Coupon '{code}' added successfully!"} except sqlite3.IntegrityError: return {"success": False, "message": f"Coupon code '{code}' already exists!"} def toggle_coupon_status(coupon_id: int) -> bool: """Toggles active/inactive status of a coupon.""" with get_connection() as conn: cursor = conn.cursor() cursor.execute("UPDATE coupons SET is_active = CASE WHEN is_active = 1 THEN 0 ELSE 1 END WHERE id = ?", (coupon_id,)) conn.commit() return cursor.rowcount > 0 def delete_coupon(coupon_id: int) -> bool: """Deletes a coupon.""" with get_connection() as conn: cursor = conn.cursor() cursor.execute("DELETE FROM coupons WHERE id = ?", (coupon_id,)) conn.commit() return cursor.rowcount > 0 # ========================================================================= # Device & Account Ban System # ========================================================================= def is_device_or_ip_banned( device_id: Optional[str] = None, ip_address: Optional[str] = None, user_id: Optional[int] = None, ) -> Tuple[bool, str]: """Checks if a device, IP, or user is permanently banned.""" with get_connection() as conn: cursor = conn.cursor() # 1. Check user ban if user_id: cursor.execute("SELECT is_banned, banned_reason FROM users WHERE id = ?", (user_id,)) u = cursor.fetchone() if u and u["is_banned"]: reason = u["banned_reason"] or "වංචනික ක්‍රියාවක් හේතුවෙන් ඔබගේ ගිණුම තහනම් කර ඇත." return True, reason # 2. Check device_id ban if device_id: cursor.execute("SELECT reason FROM banned_devices WHERE device_id = ?", (device_id,)) d = cursor.fetchone() if d: reason = d["reason"] or "Security violation / Fraudulent slip submission." return True, reason # 3. Check IP address ban if ip_address: cursor.execute("SELECT reason FROM banned_ips WHERE ip_address = ?", (ip_address,)) ip_row = cursor.fetchone() if ip_row: reason = ip_row["reason"] or "Blacklisted IP address." return True, reason return False, "" def ban_device( device_id: str, ip_address: Optional[str] = None, reason: str = "Fraudulent Activity / Fake Slip", banned_by: str = "admin", ) -> bool: """Permanently bans a device fingerprint and IP.""" if not device_id: return False now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with get_connection() as conn: cursor = conn.cursor() cursor.execute( """ INSERT OR REPLACE INTO banned_devices (device_id, ip_address, reason, banned_by, created_at) VALUES (?, ?, ?, ?, ?) """, (device_id, ip_address, reason, banned_by, now), ) if ip_address and len(ip_address) >= 7: cursor.execute( """ INSERT OR IGNORE INTO banned_ips (ip_address, reason, created_at) VALUES (?, ?, ?) """, (ip_address, reason, now), ) conn.commit() return True def unban_device(device_id: str) -> bool: """Removes a ban on a device.""" with get_connection() as conn: cursor = conn.cursor() cursor.execute("DELETE FROM banned_devices WHERE device_id = ?", (device_id,)) conn.commit() return cursor.rowcount > 0 def get_banned_devices() -> List[Dict[str, Any]]: """Returns list of all banned devices.""" with get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT * FROM banned_devices ORDER BY id DESC") return [dict(r) for r in cursor.fetchall()] def ban_user(user_id: int, reason: str = "Fraudulent Activity") -> bool: """Permanently bans a user account.""" with get_connection() as conn: cursor = conn.cursor() cursor.execute( "UPDATE users SET is_banned = 1, banned_reason = ? WHERE id = ?", (reason, user_id), ) conn.commit() return cursor.rowcount > 0 def unban_user(user_id: int) -> bool: """Unbans a user account.""" with get_connection() as conn: cursor = conn.cursor() cursor.execute( "UPDATE users SET is_banned = 0, banned_reason = NULL WHERE id = ?", (user_id,), ) conn.commit() return cursor.rowcount > 0 # ========================================================================= # Chat Satisfaction Ratings # ========================================================================= def save_chat_rating( rating: int, comment: str = "", user_id: Optional[int] = None, device_id: Optional[str] = None, ) -> bool: """Saves user rating for live support chat.""" now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with get_connection() as conn: cursor = conn.cursor() cursor.execute( """ INSERT INTO chat_ratings (rating, comment, user_id, device_id, created_at) VALUES (?, ?, ?, ?, ?) """, (rating, comment, user_id, device_id, now), ) conn.commit() return True def get_chat_rating_stats() -> Dict[str, Any]: """Retrieves average rating and total ratings.""" with get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT COUNT(*) AS total, AVG(rating) AS avg_rating FROM chat_ratings") row = cursor.fetchone() if row and row["total"]: return { "total": row["total"], "avg_rating": round(row["avg_rating"], 1), } return {"total": 0, "avg_rating": 5.0} # ========================================================================= # 📨 Direct Web Admin / Owner Contact & Support Tickets # ========================================================================= def create_admin_ticket( sender_name: str, sender_contact: str, message: str, subject: Optional[str] = "Customer Inquiry", player_id: Optional[str] = None, order_code: Optional[str] = None, user_id: Optional[int] = None, ) -> str: """Creates a new direct support ticket/message for the Web Owner/Admin.""" ticket_code = f"TKT-{uuid.uuid4().hex[:6].upper()}" now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") initial_conversation = [ { "sender": "customer", "sender_name": sender_name.strip(), "message": message.strip(), "created_at": now } ] with get_connection() as conn: cursor = conn.cursor() cursor.execute( """ INSERT INTO admin_tickets ( ticket_code, user_id, player_id, sender_name, sender_contact, subject, message, order_code, status, created_at, updated_at, conversation_json ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'open', ?, ?, ?) """, ( ticket_code, user_id, player_id, sender_name.strip(), sender_contact.strip(), subject.strip() if subject else "General Help", message.strip(), order_code.strip() if order_code else None, now, now, json.dumps(initial_conversation, ensure_ascii=False) ), ) conn.commit() return ticket_code def get_all_admin_tickets(limit: int = 50) -> List[Dict[str, Any]]: """Returns recent admin support tickets with conversation parsed.""" with get_connection() as conn: cursor = conn.cursor() cursor.execute( """ SELECT * FROM admin_tickets ORDER BY id DESC LIMIT ? """, (limit,) ) tickets = [] for r in cursor.fetchall(): tkt = dict(r) conv = [] if tkt.get("conversation_json"): try: conv = json.loads(tkt["conversation_json"]) except Exception: conv = [] if not conv: if tkt.get("message"): conv.append({ "sender": "customer", "sender_name": tkt.get("sender_name") or "Customer", "message": tkt.get("message"), "created_at": tkt.get("created_at") }) if tkt.get("admin_reply"): conv.append({ "sender": "admin", "sender_name": "Web Owner / Admin", "message": tkt.get("admin_reply"), "created_at": tkt.get("replied_at") or tkt.get("updated_at") }) tkt["conversation"] = conv tickets.append(tkt) return tickets def get_admin_ticket(ticket_code: str) -> Optional[Dict[str, Any]]: """Fetches a specific support ticket by its ticket code with full conversation history.""" with get_connection() as conn: cursor = conn.cursor() cursor.execute( "SELECT * FROM admin_tickets WHERE ticket_code = ? OR id = ?", (ticket_code.strip(), ticket_code.strip()) ) row = cursor.fetchone() if not row: return None tkt = dict(row) conv = [] if tkt.get("conversation_json"): try: conv = json.loads(tkt["conversation_json"]) except Exception: conv = [] if not conv: if tkt.get("message"): conv.append({ "sender": "customer", "sender_name": tkt.get("sender_name") or "Customer", "message": tkt.get("message"), "created_at": tkt.get("created_at") }) if tkt.get("admin_reply"): conv.append({ "sender": "admin", "sender_name": "Web Owner / Admin", "message": tkt.get("admin_reply"), "created_at": tkt.get("replied_at") or tkt.get("updated_at") }) tkt["conversation"] = conv return tkt def reply_admin_ticket(ticket_code: str, admin_reply: str) -> bool: """Saves Admin's response to the ticket and appends to the conversation thread.""" now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") ticket = get_admin_ticket(ticket_code) if not ticket: return False conv = ticket.get("conversation") or [] conv.append({ "sender": "admin", "sender_name": "Web Owner / Admin", "message": admin_reply.strip(), "created_at": now }) with get_connection() as conn: cursor = conn.cursor() cursor.execute( """ UPDATE admin_tickets SET admin_reply = ?, status = 'replied', replied_at = ?, updated_at = ?, conversation_json = ? WHERE ticket_code = ? """, (admin_reply.strip(), now, now, json.dumps(conv, ensure_ascii=False), ticket_code.strip()) ) conn.commit() return cursor.rowcount > 0 def customer_reply_admin_ticket(ticket_code: str, message: str, sender_name: Optional[str] = None) -> bool: """Appends customer's follow-up reply to the ticket and marks status as customer_replied.""" now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") ticket = get_admin_ticket(ticket_code) if not ticket: return False name = (sender_name or ticket.get("sender_name") or "Customer").strip() conv = ticket.get("conversation") or [] conv.append({ "sender": "customer", "sender_name": name, "message": message.strip(), "created_at": now }) with get_connection() as conn: cursor = conn.cursor() cursor.execute( """ UPDATE admin_tickets SET status = 'customer_replied', updated_at = ?, conversation_json = ? WHERE ticket_code = ? """, (now, json.dumps(conv, ensure_ascii=False), ticket_code.strip()) ) conn.commit() return cursor.rowcount > 0 def get_open_tickets_count() -> int: """Returns count of pending unreplied tickets (open or customer_replied).""" with get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT COUNT(*) FROM admin_tickets WHERE status IN ('open', 'customer_replied')") row = cursor.fetchone() return row[0] if row else 0 # ========================================================================= # 🎰 Mystery Diamond Box System # ========================================================================= def record_mystery_box_open( user_id: Optional[int], player_id: str, nickname: str, order_code: str, amount_paid: float, won_item_id: str, won_item_name: str, won_item_type: str, won_amount: float, is_win: int, delivery_status: str = "completed" ) -> int: """Records an unboxing event in mystery_box_logs.""" now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with get_connection() as conn: cursor = conn.cursor() cursor.execute( """ INSERT INTO mystery_box_logs ( user_id, player_id, nickname, order_code, amount_paid, won_item_id, won_item_name, won_item_type, won_amount, is_win, delivery_status, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( user_id, player_id, nickname, order_code, amount_paid, won_item_id, won_item_name, won_item_type, won_amount, is_win, delivery_status, now ) ) # Also create a completed order record in orders table so it is trackable in Live Tracker, My Orders & Admin try: htg_note = delivery_status if htg_note == "delivered_htg": htg_note = "Delivered via Instant Server Direct" elif htg_note == "credited_to_wallet": htg_note = "Prize Credited to Wallet" cursor.execute( """ INSERT OR IGNORE INTO orders ( order_code, user_id, player_id, nickname, product_id, product_name, quantity, amount, status, tx_reference, htg_order_ids, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, 1, ?, 'completed', ?, ?, ?, ?) """, ( order_code, user_id, player_id, nickname, won_item_id, f"Mystery Box: {won_item_name}", amount_paid, f"MBX-WALLET-{order_code}", htg_note, now, now ) ) except Exception as e: print(f"Notice: order insert for mystery box: {e}") conn.commit() return cursor.lastrowid def get_recent_mystery_box_winners(limit: int = 10) -> List[Dict[str, Any]]: """Fetches recent lucky winners of the Mystery Box.""" with get_connection() as conn: cursor = conn.cursor() cursor.execute( """ SELECT id, player_id, nickname, won_item_name, won_item_type, won_amount, is_win, created_at FROM mystery_box_logs ORDER BY id DESC LIMIT ? """, (limit,) ) winners = [] for r in cursor.fetchall(): d = dict(r) d["prize_name"] = d.get("won_item_name") or "Special Reward" winners.append(d) return winners def get_mystery_box_stats() -> Dict[str, Any]: """Returns total opens, total revenue, and win percentage for mystery box.""" with get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT COUNT(*), SUM(amount_paid), SUM(CASE WHEN is_win = 1 THEN 1 ELSE 0 END) FROM mystery_box_logs") row = cursor.fetchone() total_opens = row[0] or 0 total_revenue = row[1] or 0.0 total_wins = row[2] or 0 win_rate = (total_wins / total_opens * 100) if total_opens > 0 else 80.0 return { "total_opens": total_opens, "total_revenue": total_revenue, "total_wins": total_wins, "win_rate": round(win_rate, 1) } def get_user_daily_mystery_luck(user_id: Optional[int], device_id: Optional[str] = None) -> Optional[Dict[str, Any]]: """Returns today's spun mystery box luck rate for user or device.""" today = datetime.now().strftime("%Y-%m-%d") with get_connection() as conn: cursor = conn.cursor() if user_id: cursor.execute( "SELECT * FROM daily_mystery_luck WHERE user_id = ? AND spin_date = ? ORDER BY id DESC LIMIT 1", (user_id, today) ) row = cursor.fetchone() if row: return dict(row) if device_id: cursor.execute( "SELECT * FROM daily_mystery_luck WHERE device_id = ? AND spin_date = ? ORDER BY id DESC LIMIT 1", (device_id, today) ) row = cursor.fetchone() if row: return dict(row) return None def record_user_daily_mystery_luck(user_id: Optional[int], device_id: Optional[str], win_rate: int, luck_label: str) -> Dict[str, Any]: """Records today's spun mystery luck rate.""" today = datetime.now().strftime("%Y-%m-%d") now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with get_connection() as conn: cursor = conn.cursor() cursor.execute( """ INSERT INTO daily_mystery_luck (user_id, device_id, spin_date, win_rate, luck_label, created_at) VALUES (?, ?, ?, ?, ?, ?) """, (user_id, device_id, today, int(win_rate), luck_label, now) ) conn.commit() return { "user_id": user_id, "device_id": device_id, "spin_date": today, "win_rate": int(win_rate), "luck_label": luck_label, "created_at": now } def save_push_subscription( endpoint: str, p256dh: Optional[str] = None, auth: Optional[str] = None, user_id: Optional[int] = None, ticket_code: Optional[str] = None, device_id: Optional[str] = None, ) -> bool: """Saves or updates a Web Push subscription endpoint.""" now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") clean_endpoint = str(endpoint or "").strip() if not clean_endpoint: return False with get_connection() as conn: cursor = conn.cursor() cursor.execute( """ INSERT INTO push_subscriptions (endpoint, p256dh, auth, user_id, ticket_code, device_id, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(endpoint) DO UPDATE SET p256dh = COALESCE(excluded.p256dh, push_subscriptions.p256dh), auth = COALESCE(excluded.auth, push_subscriptions.auth), user_id = COALESCE(excluded.user_id, push_subscriptions.user_id), ticket_code = COALESCE(excluded.ticket_code, push_subscriptions.ticket_code), device_id = COALESCE(excluded.device_id, push_subscriptions.device_id), updated_at = excluded.updated_at """, (clean_endpoint, p256dh, auth, user_id, ticket_code, device_id, now, now), ) conn.commit() return True def get_push_subscriptions_for_ticket(ticket_code: str) -> List[Dict[str, Any]]: """Returns all push subscriptions associated with a given ticket.""" clean_code = str(ticket_code or "").strip().upper() with get_connection() as conn: cursor = conn.cursor() cursor.execute( "SELECT * FROM push_subscriptions WHERE UPPER(ticket_code) = ? OR ticket_code IS NULL ORDER BY id DESC LIMIT 50", (clean_code,), ) rows = cursor.fetchall() return [dict(r) for r in rows] def backup_database(backup_dir: str = "backups") -> Dict[str, Any]: """ Safely creates a consistent SQLite backup of topup_store.db. Retains the last 7 daily backups to prevent filling disk space. """ import os, glob, sqlite3 os.makedirs(backup_dir, exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") backup_filename = f"topup_store_backup_{timestamp}.db" backup_path = os.path.join(backup_dir, backup_filename) with sqlite3.connect(DATABASE_PATH) as src_conn: with sqlite3.connect(backup_path) as dst_conn: src_conn.backup(dst_conn) file_size = os.path.getsize(backup_path) existing = sorted(glob.glob(os.path.join(backup_dir, "topup_store_backup_*.db")), reverse=True) if len(existing) > 7: for old_b in existing[7:]: try: os.remove(old_b) except Exception: pass return { "success": True, "backup_filename": backup_filename, "backup_path": backup_path, "file_size": file_size, "timestamp": timestamp, }