""" HTG Engine for SL CAT TOPUP Web Store Handles Free Fire Player UID validation and automated order fulfillment on htgtopup.com. """ import json import logging import os import re import datetime import time import requests from typing import Dict, Any, Optional from config import HTG_SESSION_PATH, HTG_ACCOUNTS_PATH, get_product_price logger = logging.getLogger("htg_engine") DEFAULT_HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36", "Accept": "application/json, text/javascript, */*; q=0.01", "Accept-Language": "en-US,en;q=0.9", "Origin": "https://htgtopup.com", "Referer": "https://htgtopup.com/", } # Static catalogue of Free Fire products supported on the store PRODUCT_CATALOG = [ # Memberships { "id": "1", "name": "Weekly Membership", "category": "Memberships", "icon": "🌟", "image": "/static/img/prod_weekly.png", "badge": "Popular", "diamonds": "450 💎 value", }, { "id": "2", "name": "Monthly Membership", "category": "Memberships", "icon": "👑", "image": "/static/img/prod_monthly.png", "badge": "Best Value", "diamonds": "2,600 💎 value", }, { "id": "3", "name": "Weekly Lite Membership", "category": "Memberships", "icon": "⚡", "image": "/static/img/prod_weekly_lite.png", "badge": "Budget", "diamonds": "100 💎 value", }, # VIP Bundles { "id": "bundle_vip", "name": "VIP Bundle (1W + 1M)", "category": "VIP Bundles", "icon": "🔥", "image": "/static/img/prod_bundle_vip.png", "badge": "Hot", "diamonds": "1 Weekly + 1 Monthly", }, { "id": "bundle_svip", "name": "SVIP Bundle (4W + 1M)", "category": "VIP Bundles", "icon": "💎", "image": "/static/img/prod_bundle_svip.png", "badge": "Super VIP", "diamonds": "4 Weekly + 1 Monthly", }, { "id": "bundle_vip_max", "name": "VIP Max Bundle (1W + 1M + 1WL)", "category": "VIP Bundles", "icon": "🚀", "image": "/static/img/prod_bundle_vip_max.png", "badge": "Max Saver", "diamonds": "1W + 1M + 1 Weekly Lite", }, { "id": "bundle_svip_max", "name": "SVIP Max Bundle (4W + 1M + 4WL)", "category": "VIP Bundles", "icon": "👑", "image": "/static/img/prod_bundle_svip_max.png", "badge": "Ultimate", "diamonds": "4W + 1M + 4 Weekly Lite", }, # Diamonds { "id": "6", "name": "25 Diamonds", "category": "Diamonds", "icon": "💎", "image": "/static/img/prod_diamonds_chest.png", "badge": None, "diamonds": "25 💎", }, { "id": "7", "name": "100 Diamonds", "category": "Diamonds", "icon": "💎", "image": "/static/img/prod_diamonds_chest.png", "badge": None, "diamonds": "100 💎", }, { "id": "8", "name": "310 Diamonds", "category": "Diamonds", "icon": "💎", "image": "/static/img/prod_diamonds_chest.png", "badge": None, "diamonds": "310 💎", }, { "id": "9", "name": "520 Diamonds", "category": "Diamonds", "icon": "💎", "image": "/static/img/prod_diamonds_chest.png", "badge": "Popular", "diamonds": "520 💎", }, { "id": "10", "name": "1060 Diamonds", "category": "Diamonds", "icon": "💎", "image": "/static/img/prod_diamonds_chest.png", "badge": None, "diamonds": "1060 💎", }, { "id": "11", "name": "2180 Diamonds", "category": "Diamonds", "icon": "💎", "image": "/static/img/prod_diamonds_chest.png", "badge": "Mega", "diamonds": "2180 💎", }, { "id": "12", "name": "5600 Diamonds", "category": "Diamonds", "icon": "💎", "image": "/static/img/prod_diamonds_chest.png", "badge": None, "diamonds": "5600 💎", }, { "id": "13", "name": "11500 Diamonds", "category": "Diamonds", "icon": "💎", "image": "/static/img/prod_diamonds_chest.png", "badge": "Ultimate", "diamonds": "11500 💎", }, # Level Up Pass (Under Maintenance) { "id": "86", "name": "level up pass 6", "category": "Level Up Pass", "icon": "🎟️", "image": "/static/img/leveluppass/lvl6.png", "badge": "Maintenance", "diamonds": "Level 6 Rewards", "original_price": 90.0, "discount_percent": 10, "maintenance": True, }, { "id": "87", "name": "level up pass 10", "category": "Level Up Pass", "icon": "🎟️", "image": "/static/img/leveluppass/lvl10.png", "badge": "Maintenance", "diamonds": "Level 10 Rewards", "original_price": 195.0, "discount_percent": 10, "maintenance": True, }, { "id": "88", "name": "level up pass 15", "category": "Level Up Pass", "icon": "🎟️", "image": "/static/img/leveluppass/lvl15.png", "badge": "Maintenance", "diamonds": "Level 15 Rewards", "original_price": 195.0, "discount_percent": 10, "maintenance": True, }, { "id": "89", "name": "level up pass 20", "category": "Level Up Pass", "icon": "🎟️", "image": "/static/img/leveluppass/lvl20.png", "badge": "Maintenance", "diamonds": "Level 20 Rewards", "original_price": 195.0, "discount_percent": 10, "maintenance": True, }, { "id": "90", "name": "level up pass 25", "category": "Level Up Pass", "icon": "🎟️", "image": "/static/img/leveluppass/lvl25.png", "badge": "Maintenance", "diamonds": "Level 25 Rewards", "original_price": 195.0, "discount_percent": 10, "maintenance": True, }, { "id": "91", "name": "level up pass 30", "category": "Level Up Pass", "icon": "🎟️", "image": "/static/img/leveluppass/lvl30.png", "badge": "Maintenance", "diamonds": "Level 30 Rewards", "original_price": 310.0, "discount_percent": 10, "maintenance": True, }, ] def parse_cookie_string(raw_input: Any) -> Dict[str, str]: """ Parses cookies from: 1. A dict 2. A JSON string (e.g. {"PHPSESSID": "...", "csrf_token": "..."}) 3. A raw header/cookie string (e.g. "PHPSESSID=xyz; csrf_token=abc") """ cookies = {} if not raw_input: return {} if isinstance(raw_input, dict): if "cookies" in raw_input and isinstance(raw_input["cookies"], dict): cookies = raw_input["cookies"] else: cookies = {str(k): str(v) for k, v in raw_input.items()} else: raw_str = str(raw_input).strip() if not raw_str: return {} # Try JSON if (raw_str.startswith("{") and raw_str.endswith("}")) or (raw_str.startswith("[") and raw_str.endswith("]")): try: data = json.loads(raw_str) if isinstance(data, dict): if "cookies" in data and isinstance(data["cookies"], dict): cookies = data["cookies"] else: cookies = {str(k): str(v) for k, v in data.items()} except Exception: pass # Try raw cookie string if not parsed as JSON if not cookies: pairs = re.split(r"[;\n]+", raw_str) for pair in pairs: pair = pair.strip() if not pair or pair.startswith("#") or pair.startswith("//") or "=" not in pair: continue key, val = pair.split("=", 1) key = key.strip() val = val.strip() if re.match(r"^[a-zA-Z0-9_\-]+$", key): cookies[key] = val # Ensure HTGSESSID and PHPSESSID are synchronized (HTG uses HTGSESSID as session cookie) if "PHPSESSID" in cookies and "HTGSESSID" not in cookies: cookies["HTGSESSID"] = cookies["PHPSESSID"] if "HTGSESSID" in cookies and "PHPSESSID" not in cookies: cookies["PHPSESSID"] = cookies["HTGSESSID"] return cookies def get_default_accounts_structure() -> Dict[str, Any]: """Returns initial multi-account structure.""" return { "active_account": "account_1", "auto_failover": True, "accounts": { "account_1": { "id": "account_1", "name": "HTG Account 1 (Main)", "username": "", "user_id": "", "cookies": {}, "balance": 0.0, "status": "unconfigured", "last_checked": "", "enabled": True, }, "account_2": { "id": "account_2", "name": "HTG Account 2 (Backup)", "username": "", "user_id": "", "cookies": {}, "balance": 0.0, "status": "unconfigured", "last_checked": "", "enabled": True, }, }, } def load_all_accounts() -> Dict[str, Any]: """ Loads all accounts from HTG_ACCOUNTS_PATH. If the file doesn't exist, migrates from existing HTG_SESSION_PATH. """ data = None if os.path.exists(HTG_ACCOUNTS_PATH): try: with open(HTG_ACCOUNTS_PATH, "r", encoding="utf-8") as f: data = json.load(f) except Exception as e: logger.error(f"Error reading {HTG_ACCOUNTS_PATH}: {e}") if not data or not isinstance(data, dict) or "accounts" not in data: data = get_default_accounts_structure() # Migrate from legacy HTG_SESSION_PATH if available if os.path.exists(HTG_SESSION_PATH): try: with open(HTG_SESSION_PATH, "r", encoding="utf-8") as f: legacy_cookies = json.load(f) if isinstance(legacy_cookies, dict): if "cookies" in legacy_cookies and isinstance(legacy_cookies["cookies"], dict): legacy_cookies = legacy_cookies["cookies"] data["accounts"]["account_1"]["cookies"] = legacy_cookies data["accounts"]["account_1"]["username"] = legacy_cookies.get("login-username", "SL-cat-app") data["accounts"]["account_1"]["user_id"] = legacy_cookies.get("login-user_id", "10562") data["accounts"]["account_1"]["status"] = "connected" try: data["accounts"]["account_1"]["balance"] = float(legacy_cookies.get("login-wallet", 0.0)) except Exception: pass except Exception as e: logger.error(f"Error migrating legacy session: {e}") save_all_accounts(data) return data def save_all_accounts(data: Dict[str, Any]) -> bool: """Saves multi-account data and syncs active account to HTG_SESSION_PATH.""" try: with open(HTG_ACCOUNTS_PATH, "w", encoding="utf-8") as f: json.dump(data, f, indent=2) # Sync active account's cookies to HTG_SESSION_PATH for backwards compatibility active_key = data.get("active_account", "account_1") active_acc = data.get("accounts", {}).get(active_key, {}) cookies = active_acc.get("cookies", {}) if cookies: with open(HTG_SESSION_PATH, "w", encoding="utf-8") as f: json.dump(cookies, f, indent=2) return True except Exception as e: logger.error(f"Error saving {HTG_ACCOUNTS_PATH}: {e}") return False def get_active_account_key() -> str: data = load_all_accounts() return data.get("active_account", "account_1") def set_active_account(account_key: str) -> bool: data = load_all_accounts() if account_key in data.get("accounts", {}): data["active_account"] = account_key return save_all_accounts(data) return False def set_auto_failover(enabled: bool) -> bool: data = load_all_accounts() data["auto_failover"] = bool(enabled) return save_all_accounts(data) def load_session_cookies(account_key: Optional[str] = None) -> Dict[str, str]: """Loads HTG cookies for the specified account, or the active account if None.""" data = load_all_accounts() key = account_key or data.get("active_account", "account_1") acc = data.get("accounts", {}).get(key, {}) cookies = acc.get("cookies", {}) res = {} if cookies: res = dict(cookies) elif not account_key and os.path.exists(HTG_SESSION_PATH): try: with open(HTG_SESSION_PATH, "r", encoding="utf-8") as f: raw = json.load(f) res = raw.get("cookies", raw) if isinstance(raw, dict) else {} except Exception: pass if res: if "PHPSESSID" in res and "HTGSESSID" not in res: res["HTGSESSID"] = res["PHPSESSID"] if "HTGSESSID" in res and "PHPSESSID" not in res: res["PHPSESSID"] = res["HTGSESSID"] return res def verify_account_session(cookies: Dict[str, str]) -> Dict[str, Any]: """Verifies cookies against https://htgtopup.com/get_wallet.php and returns parsed balance & username.""" if not cookies: return {"valid": False, "message": "No cookies provided"} session = requests.Session() session.headers.update(DEFAULT_HEADERS) session.cookies.update(cookies) csrf_token = cookies.get("csrf_token") if csrf_token: session.headers["X-CSRF-Token"] = csrf_token try: url = "https://htgtopup.com/get_wallet.php" r = session.get(url, timeout=10) data = r.json() if data.get("success") or data.get("status") == "active" or "balance" in data: bal = float(data.get("balance", 0.0)) user = cookies.get("login-username", data.get("username", "Owner")) uid = str(cookies.get("login-user_id", data.get("user_id", ""))) return { "valid": True, "username": user, "balance": bal, "user_id": uid, } return {"valid": False, "message": data.get("message", "Session expired or invalid")} except Exception as e: return {"valid": False, "message": str(e)} def update_account_cookies(account_key: str, raw_input: Any, account_name: Optional[str] = None) -> Dict[str, Any]: """Updates cookies for a specific account, validates with HTG, and saves.""" cookies = parse_cookie_string(raw_input) if not cookies: return {"success": False, "message": "No valid cookies found in input. Please paste valid cookies or JSON."} v_res = verify_account_session(cookies) data = load_all_accounts() if account_key not in data.get("accounts", {}): data.setdefault("accounts", {})[account_key] = { "id": account_key, "name": account_name or f"HTG Account {account_key}", "enabled": True, } target = data["accounts"][account_key] target["cookies"] = cookies if account_name: target["name"] = account_name now_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") target["last_checked"] = now_str if v_res.get("valid"): target["valid"] = True target["status"] = "connected" target["username"] = v_res.get("username") target["user_id"] = v_res.get("user_id") target["balance"] = v_res.get("balance", 0.0) else: target["valid"] = False target["status"] = "error" target["error_message"] = v_res.get("message") save_all_accounts(data) return { "success": v_res.get("valid", False), "account_id": account_key, "username": target.get("username"), "balance": target.get("balance", 0.0), "status": target.get("status"), "message": "Account session verified and connected successfully!" if v_res.get("valid") else f"Saved, but verification failed: {v_res.get('message')}", } def check_htg_balance(account_key: Optional[str] = None) -> Dict[str, Any]: """Checks HTG balance for a specific account or the active account.""" data = load_all_accounts() key = account_key or data.get("active_account", "account_1") acc = data.get("accounts", {}).get(key, {}) cookies = acc.get("cookies", {}) or load_session_cookies(key) if not cookies: return {"valid": False, "message": f"Account '{key}' has no cookies configured.", "account_id": key} v_res = verify_account_session(cookies) if v_res.get("valid"): acc["balance"] = v_res["balance"] acc["username"] = v_res["username"] acc["user_id"] = v_res["user_id"] acc["valid"] = True acc["status"] = "connected" acc["last_checked"] = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") save_all_accounts(data) v_res["account_id"] = key return v_res else: acc["valid"] = False acc["status"] = "expired" save_all_accounts(data) v_res["account_id"] = key return v_res def get_all_htg_accounts_status() -> Dict[str, Any]: """Returns overview of all configured HTG accounts with live/cached balances.""" data = load_all_accounts() active_key = data.get("active_account", "account_1") auto_failover = data.get("auto_failover", True) accounts_list = [] for acc_id, acc in data.get("accounts", {}).items(): accounts_list.append({ "id": acc_id, "name": acc.get("name", acc_id), "username": acc.get("username", ""), "user_id": acc.get("user_id", ""), "balance": acc.get("balance", 0.0), "balance_formatted": f"LKR {acc.get('balance', 0.0):,.2f}", "is_active": (acc_id == active_key), "has_cookies": bool(acc.get("cookies")), "status": acc.get("status", "unconfigured"), "valid": acc.get("valid", bool(acc.get("cookies"))), "last_checked": acc.get("last_checked", ""), "enabled": acc.get("enabled", True), }) return { "active_account": active_key, "auto_failover": auto_failover, "accounts": accounts_list, } def refresh_all_accounts_balance() -> Dict[str, Any]: """Refreshes balances of all configured accounts directly from HTG.""" data = load_all_accounts() for acc_id in data.get("accounts", {}): check_htg_balance(acc_id) return get_all_htg_accounts_status() def check_player_uid(uid: str, server: str = "SG") -> Dict[str, Any]: """Validates Free Fire Player UID using Garena SG server.""" uid = str(uid).strip() if not uid or not uid.isdigit() or len(uid) < 6: return {"success": False, "message": "කරුණාකර නිවැරදි Free Fire Player ID (UID) එකක් ඇතුළත් කරන්න."} cookies = load_session_cookies() session = requests.Session() session.headers.update(DEFAULT_HEADERS) if cookies: session.cookies.update(cookies) url = "https://htgtopup.com/api/ff.php" data = {"uid": uid, "server": server} try: resp = session.post(url, data=data, timeout=8) res_data = resp.json() if not res_data or not isinstance(res_data, dict): return {"success": False, "message": "Player UID සොයාගත නොහැකි විය. කරුණාකර අංකය නැවත පරීක්ෂා කරන්න."} nickname = res_data.get("nickname", "") region = res_data.get("region", "") level = res_data.get("level", 0) if not nickname: return {"success": False, "message": "වලංගු නොවන Player ID එකකි. නැවත පරීක්ෂා කරන්න."} return { "success": True, "uid": uid, "nickname": nickname, "region": region, "level": level, } except Exception as e: return {"success": False, "message": f"UID පරීක්ෂාවේදී දෝෂයක්: {str(e)}"} def get_store_products() -> list: """Returns product catalog with selling prices.""" products = [] for p in PRODUCT_CATALOG: price = get_product_price(p["id"]) orig_price = p.get("original_price") products.append({ "id": p["id"], "name": p["name"], "category": p["category"], "icon": p["icon"], "image": p.get("image", "/static/img/prod_diamonds_chest.png"), "badge": p["badge"], "diamonds": p["diamonds"], "price": price, "price_formatted": f"Rs. {price:,.2f}", "original_price": orig_price, "original_price_formatted": f"Rs. {orig_price:,.2f}" if orig_price else None, "discount_percent": p.get("discount_percent"), }) return products def execute_htg_topup( player_id: str, nickname: str, product_id: str, quantity: int = 1, preferred_account: Optional[str] = None, account_key: Optional[str] = None, **kwargs, ) -> Dict[str, Any]: """ Executes topup order on HTG Digital with Smart Auto-Failover. - Attempts on the active/preferred account first. - If it fails (e.g. insufficient wallet balance, session expired) and auto_failover is enabled, automatically retries on the backup account. """ accounts_data = load_all_accounts() active_key = preferred_account or account_key or accounts_data.get("active_account", "account_1") auto_failover = accounts_data.get("auto_failover", True) account_keys = [active_key] if auto_failover: for k in accounts_data.get("accounts", {}): if k != active_key and accounts_data["accounts"][k].get("enabled", True): account_keys.append(k) last_error = "No HTG accounts configured." attempt_logs = [] for acc_key in account_keys: acc_info = accounts_data.get("accounts", {}).get(acc_key, {}) cookies = acc_info.get("cookies") or load_session_cookies(acc_key) acc_name = acc_info.get("name", acc_key) if not cookies: attempt_logs.append(f"[{acc_name}] No session cookies") continue session = requests.Session() session.headers.update(DEFAULT_HEADERS) session.cookies.update(cookies) csrf = cookies.get("csrf_token", "") bundle_map = { "bundle_vip": [("1", 1), ("2", 1)], "bundle_svip": [("1", 4), ("2", 1)], "bundle_vip_max": [("1", 1), ("2", 1), ("3", 1)], "bundle_svip_max": [("1", 4), ("2", 1), ("3", 4)], } success = False res = {} if product_id in bundle_map: items = bundle_map[product_id] order_ids = [] bundle_failed = False for sub_pid, sub_qty in items: sub_res = _single_topup(session, csrf, player_id, nickname, sub_pid, sub_qty * quantity) if not sub_res.get("success"): bundle_failed = True res = sub_res break order_ids.extend(sub_res.get("order_ids", [])) if not bundle_failed: success = True res = {"success": True, "order_ids": order_ids} else: res = _single_topup(session, csrf, player_id, nickname, product_id, quantity) success = res.get("success", False) if success: res["fulfilled_account"] = acc_key res["fulfilled_account_name"] = acc_name res["is_failover"] = (acc_key != active_key) if acc_key != active_key: logger.warning(f"Auto-Failover triggered: Order for {player_id} was fulfilled via {acc_name}!") return res else: msg = res.get("message", "Unknown error") attempt_logs.append(f"[{acc_name}] Failed: {msg}") last_error = msg logger.warning(f"HTG account [{acc_name}] failed: {msg}. Checking next fallback account...") return { "success": False, "message": f"HTG Fulfillment Failed. Attempts: {'; '.join(attempt_logs)}", "last_error": last_error, } def check_htg_order_queue(order_id: Optional[str] = None, player_id: Optional[str] = None, account_key: Optional[str] = None) -> Optional[Dict[str, Any]]: """Checks queue status of an order in HTG history.""" try: cookies = load_session_cookies(account_key) session = requests.Session() session.headers.update(DEFAULT_HEADERS) session.cookies.update(cookies) if "csrf_token" in cookies: session.headers["X-CSRF-Token"] = cookies["csrf_token"] session.headers["X-Requested-With"] = "XMLHttpRequest" r = session.get("https://htgtopup.com/history.php?action=listTopUp", timeout=10) data = r.json() if r.text.startswith("{") else {} for item in data.get("history", []): if order_id and item.get("orderid") == order_id: return item if player_id and str(item.get("playerid")) == str(player_id): return item except Exception as e: logger.warning(f"Error querying HTG order queue: {e}") return None def _poll_and_verify_htg_queue( session, player_id: str, expected_order_id: Optional[str] = None, wait_seconds: float = 4.0, ) -> Dict[str, Any]: """ Polls HTG listTopUp history to verify the real execution queue status. - If queue_status == 'done': Garena successfully delivered the items! - If queue_status == 'failed': Garena rejected the request (ineligible, already claimed, or server issue). - If queue_status == 'processing': Remains in queue. """ time.sleep(wait_seconds) for attempt in range(2): try: r = session.get("https://htgtopup.com/history.php?action=listTopUp", timeout=10) data = r.json() if r.text.startswith("{") else {} history = data.get("history", []) target_item = None for item in history[:20]: if expected_order_id and str(item.get("orderid", "")).strip() == str(expected_order_id).strip(): target_item = item break if not expected_order_id and str(item.get("playerid", "")).strip() == str(player_id).strip(): target_item = item break # If still not found and expected_order_id is empty, check latest top item if not target_item and not expected_order_id and history: top_item = history[0] if str(top_item.get("playerid", "")).strip() == str(player_id).strip(): target_item = top_item if target_item: order_id = target_item.get("orderid") or expected_order_id or "HTG-ORDER" h_status = str(target_item.get("status", "")).strip().lower() q_status = str(target_item.get("queue_status", "")).strip().lower() res_id = str(target_item.get("result_id", "")) # Check failure first: either HTG history status or queue status is failed if h_status in ["failed", "refund", "invalid product", "cancel", "cancelled"] or q_status in ["failed", "cancel", "cancelled", "rejected"]: return { "success": False, "message": ( f"HTG/Garena Queue Rejected ({order_id}). " f"Garena Free Fire rejected delivery for UID {player_id}. " f"Account may have already purchased Level Up Pass, level requirement not met, or region mismatch." ), "queue_status": "failed", "status": "failed", "order_id": order_id, "queue_failed": True, } if q_status == "done" or h_status in ["success", "completed", "done"]: display_id = f"{order_id} (Ref: {res_id})" if res_id and res_id != "-" else order_id return { "success": True, "order_ids": [display_id], "queue_status": "done", "status": "completed", "result_id": res_id, "order_id": order_id, } # Otherwise still processing if attempt == 0: time.sleep(2.0) continue return { "success": True, "order_ids": [order_id], "queue_status": "processing", "status": "processing", "pending_queue": True, "order_id": order_id, "message": f"Order {order_id} queued on HTG, delivery in progress.", } except Exception as e: logger.warning(f"Error checking HTG queue: {e}") break fallback_id = expected_order_id or (history[0].get("orderid") if history else None) or "HTG-QUEUED" return { "success": True, "order_ids": [fallback_id], "pending_queue": True, "order_id": fallback_id, "message": "Order queued on HTG Digital.", } def _single_topup(session, csrf: str, player_id: str, nickname: str, product_id: str, quantity: int): # Specialized fulfillment for Level Up Pass products if str(product_id) in ["86", "87", "88", "89", "90", "91"]: lvl_map = {"86": "lv6", "87": "lv10", "88": "lv15", "89": "lv20", "90": "lv25", "91": "lv30"} lvl_target = lvl_map.get(str(product_id), "") # Pre-flight Level Up Pass eligibility verification try: chk_resp = requests.post( "https://htgtopup.com/api/levelup_pass/lvl.php", data={"uid": str(player_id)}, headers={ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Origin": "https://htgtopup.com", "Referer": "https://htgtopup.com/leveluppass", }, timeout=6, ) if chk_resp.status_code == 200: chk_data = chk_resp.json() lvl_info = chk_data.get("levelup_pass", {}) if isinstance(lvl_info, dict) and lvl_target: if lvl_info.get("display") is False: return { "success": False, "message": f"Free Fire UID {player_id}: Level Up Pass is not available for this account region or status.", } if lvl_info.get(lvl_target) is False: return { "success": False, "message": f"Free Fire UID {player_id}: {lvl_target.upper()} is ineligible or already purchased. Level Up Pass can only be bought once per Free Fire account.", } if not nickname and chk_data.get("nickname"): nickname = chk_data.get("nickname") except Exception as ex: logger.warning(f"Level Up Pass pre-check skipped due to network: {ex}") url = "https://htgtopup.com/levelup.php" headers = { "Content-Type": "application/x-www-form-urlencoded", "X-CSRF-Token": csrf, "X-Requested-With": "XMLHttpRequest", "Referer": "https://htgtopup.com/leveluppass", "Origin": "https://htgtopup.com", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", } payload = { "action": "order", "productId": str(product_id), "playerid": str(player_id), "nickname": str(nickname or "Player"), "quantity": "1", "terms_accepted": "1", } try: resp = session.post(url, data=payload, headers=headers, timeout=25) try: data = resp.json() except Exception: clean_t = resp.text.strip() data = json.loads(clean_t) if clean_t.startswith("{") else {} status = str(data.get("status", "")).lower() is_success = ( status in ["success", "ok", "processing", "pending"] or data.get("success") is True or data.get("ok") is True or "order" in data or "order_id" in data or "orderID" in data ) if is_success: order_id = str(data.get("order_id") or data.get("orderID") or (data.get("data") or {}).get("order_id") or "") return _poll_and_verify_htg_queue(session, player_id=player_id, expected_order_id=order_id, wait_seconds=4.0) err_msg = data.get("message") or data.get("error") or f"Level Up Pass order failed on HTG (Status: {status or 'unknown'})" return {"success": False, "message": err_msg} except Exception as e: return {"success": False, "message": str(e)} # Standard diamonds & memberships fulfillment url = "https://htgtopup.com/topup.php" headers = {"X-CSRF-Token": csrf, "X-Requested-With": "XMLHttpRequest"} payload = { "playerid": str(player_id), "productid": str(product_id), "nickname": str(nickname), "quantity": str(quantity), } try: resp = session.post(url, data=payload, headers=headers, timeout=25) data = resp.json() status = data.get("status", "").lower() if status in ["success", "ok", "processing", "pending"] or data.get("ok") is True: order_ids = [] responses = (data.get("data") or {}).get("responses", []) for r in responses: oid = (r.get("topup_response") or {}).get("order_id") if oid: order_ids.append(str(oid)) first_oid = order_ids[0] if order_ids else None return _poll_and_verify_htg_queue(session, player_id=player_id, expected_order_id=first_oid, wait_seconds=4.0) return {"success": False, "message": data.get("message", "Order failed on HTG")} except Exception as e: return {"success": False, "message": str(e)} def create_visa_payment_link(amount: float = 5000.0, account_key: Optional[str] = None) -> Dict[str, Any]: """ Creates an official Visa Pay payment link on HTG Digital for the given amount (e.g. LKR 5000). Can target either account_1 or account_2. """ cookies = load_session_cookies(account_key) if not cookies: return {"success": False, "message": "HTG session cookies not found for the selected account."} amt = float(amount) fee = round(amt * 0.031, 2) # 3.1% processing fee total_amount = round(amt + fee, 2) session = requests.Session() session.headers.update(DEFAULT_HEADERS) session.cookies.update(cookies) csrf = cookies.get("csrf_token") if csrf: session.headers["X-CSRF-Token"] = csrf try: # Step 1: Call https://htgtopup.com/api/pay.php?action=create&amount=... url = f"https://htgtopup.com/api/pay.php?action=create&amount={total_amount}" resp = session.get(url, timeout=15) text = resp.text pay_url_match = re.search(r"Payurl\s*:\s*(.+)", text, re.IGNORECASE) tx_match = re.search(r"TransactionId\s*:\s*(.+)", text, re.IGNORECASE) if not pay_url_match or not tx_match: return {"success": False, "message": f"Payment API returned unexpected response: {text[:150]}"} pay_url = pay_url_match.group(1).strip().rstrip(";\r\n ") tx_id = tx_match.group(1).strip().rstrip(";\r\n ") # Step 2: Register with recharge.php?action=createVisa store_url = "https://htgtopup.com/recharge.php?action=createVisa" payload = { "amount": str(int(amt) if amt.is_integer() else amt), "TransactionId": tx_id, "payurl": pay_url, } headers = { "Content-Type": "application/x-www-form-urlencoded", "X-CSRF-Token": csrf or "", "X-Requested-With": "XMLHttpRequest", "Referer": "https://htgtopup.com/recharge", } store_resp = session.post(store_url, data=payload, headers=headers, timeout=15) try: store_data = store_resp.json() except Exception: store_data = {} if store_data.get("success") is False: err_msg = store_data.get("message", "User not authenticated") return { "success": False, "message": f"⚠️ HTG Session Expired! කරුණාකර htgtopup.com වෙත Login වී නව Cookies Admin Settings හි Update කර Recharge කරන්න. (HTG: {err_msg})" } return { "success": True, "account_id": account_key or get_active_account_key(), "amount": amt, "fee": fee, "total_amount": total_amount, "pay_url": pay_url, "tx_id": tx_id, "store_response": store_data, "message": "Visa Pay payment link generated successfully!", } except Exception as e: return {"success": False, "message": f"Error creating Visa payment: {str(e)}"} def check_visa_transactions(account_key: Optional[str] = None) -> Dict[str, Any]: """Triggers transaction check on HTG to update wallet after payment.""" cookies = load_session_cookies(account_key) if not cookies: return {"success": False} session = requests.Session() session.headers.update(DEFAULT_HEADERS) session.cookies.update(cookies) try: r = session.get("https://htgtopup.com/api/d1/check_transactions.php", timeout=10) return {"success": True, "data": r.text} except Exception as e: return {"success": False, "error": str(e)} def sync_missing_visa_recharge(tx_id: str, amount: float = 5000.0, account_key: Optional[str] = None) -> Dict[str, Any]: """ Manually links a paid GenieBiz transaction to the HTG account and triggers validation. Useful when a payment was completed but not auto-linked to the HTG account. """ clean_tx = str(tx_id).strip() if not clean_tx: return {"success": False, "message": "Transaction reference is required."} cookies = load_session_cookies(account_key) if not cookies: return {"success": False, "message": "HTG session cookies not found."} session = requests.Session() session.headers.update(DEFAULT_HEADERS) session.cookies.update(cookies) csrf = cookies.get("csrf_token", "") store_url = "https://htgtopup.com/recharge.php?action=createVisa" payload = { "amount": str(int(amount) if float(amount).is_integer() else amount), "TransactionId": clean_tx, "payurl": f"https://transaction.geniebiz.lk/{clean_tx}", } headers = { "Content-Type": "application/x-www-form-urlencoded", "X-CSRF-Token": csrf, "X-Requested-With": "XMLHttpRequest", "Referer": "https://htgtopup.com/recharge", } try: # Step 1: Link transaction to HTG account reg_resp = session.post(store_url, data=payload, headers=headers, timeout=20) reg_data = {} try: reg_data = reg_resp.json() except Exception: pass if reg_data.get("success") is False: return { "success": False, "message": f"HTG Error: {reg_data.get('message', 'User not authenticated. Please login.')}" } # Step 2: Trigger transaction check on HTG session.get("https://htgtopup.com/api/d1/check_transactions.php", timeout=25) # Step 3: Check updated balance bal_info = check_htg_balance(account_key) return { "success": True, "message": "Transaction synced successfully! HTG verified payment.", "tx_id": clean_tx, "order_id": reg_data.get("orderID"), "new_balance": bal_info.get("balance", 0.0), "username": bal_info.get("username", "") } except Exception as e: return {"success": False, "message": f"Error syncing transaction: {str(e)}"}