-
Notifications
You must be signed in to change notification settings - Fork 602
Description
I want to update the
-
Auto Login with Multiple Sessions:
- Add an “Auto Login” button.
- Add a box to input how many sessions (browser windows) should be logged in.
- The script should automatically log in to multiple browser sessions using the same credentials.
- This will allow faster purchases by using multiple sessions in parallel. -
Product URL Input and Live Product Fetch:
- Add a box to input the product page URL.
- Add a “Fetch Products” button that opens the product page (without login) and fetches all live products from that URL.
- Display all available products on the page along with their stock status. -
Wait Before Checkout:
- When the product page opens in the browser, add a 1–2 second delay before clicking the checkout button.
- This is to avoid accidental purchase of a previously selected product that might still be in the checkout cart. -
Product Selection and Quantity Handling:
- From interface, allow the user to select a product and enter the quantity they want to purchase.
- Based on the selected quantity, the script should open the product in multiple tabs across available sessions and proceed to checkout for each one.
- All operations should happen using the already logged-in sessions.
...
from flask import Flask, request, jsonify, render_template_string
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.chrome.options import Options
from webdriver_manager.chrome import ChromeDriverManager
from selenium.webdriver.chrome.service import Service
import pyotp
import time
import threading
import json
from flask_cors import CORS
app = Flask(name)
CORS(app)
--- Global State Management for the Selenium Driver ---
This dictionary will hold the driver instance after a successful login.
driver_state = {
'driver': None,
'wait': None,
'session_active': False,
'totp_key': None,
'base_url': "https://gold.razer.com/global/en/gold/catalog/pubg-mobile-uc-code" # Default
}
--- Constants ---
RAZER_LOGIN_URL = "https://razerid.razer.com/?client_id=63c74d17e027dc11f642146bfeeaee09c3ce23d8&redirect=https%3A%2F%2Fgold.razer.com%2Faccount%2Flogin"
--- Helper Function to Initialize WebDriver ---
def init_webdriver(headless=False):
"""Initializes and returns a Chrome WebDriver instance."""
chrome_options = Options()
chrome_options.add_argument("--disable-extensions")
chrome_options.add_argument("--disable-gpu")
chrome_options.add_argument("--no-sandbox")
chrome_options.add_argument("--window-size=1920,1080")
if headless:
chrome_options.add_argument("--headless")
chrome_options.add_argument("user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36")
else:
chrome_options.add_argument("--start-maximized")
chrome_options.page_load_strategy = 'normal'
try:
service = Service(ChromeDriverManager().install())
driver = webdriver.Chrome(service=service, options=chrome_options)
return driver
except Exception as e:
print(f"Error initializing WebDriver: {e}")
return None
--- Helper Function to Fetch Live Product Data (for /get_products endpoint) ---
def fetch_product_denominations(product_url):
"""
Fetches live product denomination and stock status from the Razer Gold page
using a dedicated headless driver instance.
"""
driver = None
MAX_WAIT_TIME = 30
try:
driver = init_webdriver(headless=True)
if not driver:
return {"status": "error", "message": "Failed to initialize Chrome Driver for scraping."}
wait = WebDriverWait(driver, MAX_WAIT_TIME)
driver.get(product_url)
PRODUCT_LABEL_XPATH = "//div[@id='webshop_step_sku']//div[contains(@class, 'selection-tile')]/label"
wait.until(EC.presence_of_all_elements_located((By.XPATH, PRODUCT_LABEL_XPATH)))
labels = driver.find_elements(By.XPATH, PRODUCT_LABEL_XPATH)
product_list = []
for label in labels:
try:
text_div = label.find_element(By.XPATH, ".//div[contains(@class, 'selection-tile__text')]")
uc_text = text_div.text.replace('PUBG', '').replace('UC', '').strip()
denomination = int(uc_text)
is_out_of_stock = False
try:
label.find_element(By.XPATH, ".//span[contains(text(), 'Out of stock')]")
is_out_of_stock = True
except:
pass
price_text = f"USD {round(denomination / 60, 2)}"
product_list.append({
"denomination": denomination,
"price": price_text,
"out_of_stock": is_out_of_stock
})
except Exception as e:
# print(f"Skipping product tile due to error in parsing: {e}")
continue
valid_products = [p for p in product_list if p['denomination'] > 0]
print(f"--- Successfully Fetched Products ({len(valid_products)} items) ---: {valid_products}")
return {"status": "success", "products": valid_products}
except Exception as e:
print(f"\n--- CRITICAL ERROR in fetch_product_denominations ---")
print(f"Error: Failed to fetch products, likely due to timeout or element change. Message: {e}\n")
return {"status": "error", "message": f"Failed to fetch products: {e}"}
finally:
try:
if driver:
driver.quit()
except:
pass
--- Automation Logic: Login Function ---
def start_razer_login(email, password, totp_key, product_url):
"""Performs login and opens the product page, then stores the driver in global state."""
# Clean up previous session if it exists
if driver_state['driver']:
try:
driver_state['driver'].quit()
except:
pass
driver_state['driver'] = None
driver_state['wait'] = None
driver_state['session_active'] = False
driver = init_webdriver(headless=False)
if not driver:
return {"status": "Error", "message": "Could not start Chrome Driver."}
wait = WebDriverWait(driver, 60)
# Update global state with the new driver and other details
driver_state['driver'] = driver
driver_state['wait'] = wait
driver_state['totp_key'] = totp_key
driver_state['base_url'] = product_url
try:
# --- A. Login Phase ---
print("Starting Login...")
driver.get(RAZER_LOGIN_URL)
email_field = wait.until(EC.presence_of_element_located((By.ID, "input-login-email")))
email_field.send_keys(email)
driver.find_element(By.ID, "input-login-password").send_keys(password)
driver.find_element(By.ID, "btn-log-in").click()
print("Login button clicked. Waiting 10 seconds for dashboard to load (Increased timeout)...")
time.sleep(10) # Increased sleep time for stability
# --- B. Navigate to Product Page ---
driver.get(product_url)
print("Redirected to product page. Session is now ready for purchase.")
driver_state['session_active'] = True
# NOTE: Returning status here so the Flask thread can send an immediate response.
return {"status": "Success", "message": "Login successful. Driver is active and on the product page. You can now perform purchases from the web UI."}
except Exception as e:
print(f"An unexpected error occurred during login: {e}")
try:
driver.quit()
except:
pass
driver_state['driver'] = None
driver_state['wait'] = None
driver_state['session_active'] = False
return {"status": "Error", "message": f"Login automation failed. An error occurred: {e}. Check CMD for details."}
--- Automation Logic: Purchase and 2FA (Unchanged) ---
def handle_2fa_input(driver, totp_key):
"""Handles 2FA using JavaScript input and a FRESH TOTP code."""
wait = WebDriverWait(driver, 30)
IFRAME_XPATH = "//iframe[contains(@title, 'Razer OTP') or contains(@src, 'otp?') or @id='otp-iframe-1']"
try:
# 1. Generate FRESH TOTP CODE
otp_code = None
try:
totp = pyotp.TOTP(totp_key)
otp_code = totp.now()
except:
return False
if not otp_code or len(otp_code) != 6:
return False
# 2. Wait for and switch to the specific IFRAME
iframe_element = wait.until(EC.presence_of_element_located((By.XPATH, IFRAME_XPATH)))
driver.switch_to.frame(iframe_element)
# 3. Wait for the first NEW input box
NEW_INPUT_ID = "otp-input-0"
wait.until(EC.presence_of_element_located((By.ID, NEW_INPUT_ID)))
# 4. Input each digit using JavaScript
for i, digit in enumerate(otp_code):
input_id = f"otp-input-{i}"
try:
input_element = driver.find_element(By.ID, input_id)
driver.execute_script("arguments[0].value = arguments[1];", input_element, digit)
driver.execute_script("arguments[0].dispatchEvent(new Event('input', { bubbles: true }));", input_element)
except:
return False
return True
except Exception as e:
print(f"2FA Input Critical Failure: {e}")
return False
finally:
try:
driver.switch_to.default_content()
except:
pass
def perform_purchase_action(driver, wait, product_link, denomination, totp_key):
"""Performs the single purchase action."""
try:
driver.get(product_link)
# --- STEP 1: Select Product (UC Denomination) ---
full_uc_text = f"PUBG {denomination} UC"
uc_label_xpath = f"//div[@id='webshop_step_sku']//div[contains(text(),'{full_uc_text}')]/ancestor::label[1]"
label_element = wait.until(EC.element_to_be_clickable((By.XPATH, uc_label_xpath)))
driver.execute_script("arguments[0].click();", label_element)
# --- STEP 2: Select Payment Method (Razer Gold) ---
razer_gold_xpath = "//div[contains(@class, 'payment-channels-list')]//div[contains(text(), 'Razer Gold')]/ancestor::label[1]"
razer_gold_label = wait.until(EC.element_to_be_clickable((By.XPATH, razer_gold_xpath)))
driver.execute_script("arguments[0].click();", razer_gold_label)
# --- STEP 3: Click Checkout ---
checkout_button_xpath = "//button[@data-cs-override-id='purchase-webshop-checkout-btn']"
checkout_button = wait.until(EC.element_to_be_clickable((By.XPATH, checkout_button_xpath)))
checkout_button.click()
# --- STEP 4: HANDLE 2FA / FINAL STEP (Pass TOTP key) ---
handle_2fa_input(driver, totp_key)
return True
except Exception as e:
print(f"ERROR processing {denomination} UC: Purchase automation failed. Message: {e}")
return False
def _process_purchases_threaded(products_to_buy, product_link):
"""Handles the purchase logic in a separate thread."""
driver = driver_state['driver']
wait = driver_state['wait']
totp_key = driver_state['totp_key']
product_count = 0
total_items = sum(p['quantity'] for p in products_to_buy)
if not driver or not wait or not totp_key:
print("Thread Error: Driver or essential data missing. Purchase aborted.")
return
try:
driver.switch_to.window(driver.window_handles[0])
for product in products_to_buy:
denomination = str(product['denomination'])
quantity = product['quantity']
for i in range(quantity):
if product_count > 0:
driver.execute_script("window.open('');")
driver.switch_to.window(driver.window_handles[-1])
print(f"Processing: {denomination} UC (Item {product_count+1}/{total_items})")
if perform_purchase_action(driver, wait, product_link, denomination, totp_key):
product_count += 1
else:
try:
driver.close()
driver.switch_to.window(driver.window_handles[0])
except:
pass
if product_count > 0:
driver.switch_to.window(driver.window_handles[0])
print(f"--- Purchase Phase Complete ({product_count} successful purchases) ---")
except Exception as e:
print(f"CRITICAL ERROR during purchase phase: {e}")
driver_state['session_active'] = False
print(f"Purchase Thread Finished. Total successful: {product_count}")
--- Flask Server Setup (Updated HTML Template) ---
HTML_FORM_TEMPLATE = """
<!doctype html>
Razer Gold Automation Control
<div id="status-display" class="info">
Driver Status: Not Active.
</div>
<form id="controlForm">
<label for="product_url">Product Page URL:</label>
<input type="text" id="product_url" name="product_url" value="https://gold.razer.com/global/en/gold/catalog/pubg-mobile-uc-code" required>
<button type="button" onclick="fetchProducts()">Refresh / Fetch Products</button>
<div class="note">
<strong>STEP 1: LOGIN (Required for Purchase)</strong>
</div>
<label for="email">Razer ID Email:</label>
<input type="text" id="email" name="email" required>
<label for="password">Razer ID Password:</label>
<input type="password" id="password" name="password" required>
<label for="totp_key">TOTP Secret Key (Base32):</label>
<input type="text" id="totp_key" name="totp_key" required>
<button type="button" onclick="startLogin()">Start Auto Login</button>
<div class="purchase-group">
<div class="note">
<strong>STEP 2: PURCHASE (Requires successful login)</strong><br>
Items marked <span class="out-of-stock">RED</span> are Out of Stock and will be skipped by the script.
</div>
<label>Select Quantities to Purchase:</label>
<div class="product-list-container">
<div id="product-list-area">Click 'Refresh / Fetch Products' to load data.</div>
</div>
<button type="button" onclick="startPurchase()">Perform Purchases</button>
</div>
</form>
<div id="message"></div>
@app.route('/', methods=['GET'])
def index():
return render_template_string(HTML_FORM_TEMPLATE)
@app.route('/get_products', methods=['POST'])
def get_products():
"""Endpoint to fetch and return live product list from the provided URL."""
data = request.json
product_url = data.get('product_url', driver_state['base_url'])
result = fetch_product_denominations(product_url)
return jsonify(result)
@app.route('/start_login', methods=['POST'])
def handle_login_request():
"""Endpoint to start the login process and keep the driver open."""
data = request.json
email = data.get('email')
password = data.get('password')
totp_key = data.get('totp_key')
product_url = data.get('product_url', driver_state['base_url'])
if not all([email, password, totp_key, product_url]):
return jsonify({"status": "Error", "message": "Missing email, password, TOTP key, or product URL."}), 400
# Start the login in a separate thread
# We create a function to manage the thread and return its result for the Flask response
def run_login_and_report():
result = start_razer_login(email, password, totp_key, product_url)
# In a real app, this would use websockets to push status to the front-end.
# Here, we just rely on the main function to start the thread.
print(f"Login Thread Status Report: {result['status']} - {result['message']}")
threading.Thread(target=run_login_and_report).start()
return jsonify({
"status": "Initiated",
"message": "Login script started in the background. Please wait for the Chrome window to open. The Backend session is ready if the console shows success."
})
@app.route('/perform_purchase', methods=['POST'])
def handle_purchase_request():
"""Endpoint to perform the purchase using the existing, logged-in driver."""
data = request.json
products_to_buy = data.get('products')
product_link = data.get('product_link')
# This is the critical BACKEND check.
if not driver_state['session_active'] or not driver_state['driver']:
return jsonify({"status": "Error", "message": "No active session found in the backend. Please restart 'Start Auto Login'."}), 400
if not products_to_buy or not product_link:
return jsonify({"status": "Error", "message": "Missing product data or product link."}), 400
threading.Thread(target=_process_purchases_threaded, args=(products_to_buy, product_link)).start()
return jsonify({
"status": "Initiated",
"message": "Purchase script started in the background. New tabs will open in the existing Chrome window. Monitor the console for 2FA messages."
})
if name == 'main':
driver_state['base_url'] = "https://gold.razer.com/global/en/gold/catalog/pubg-mobile-uc-code"
print("Starting Flask Server... Access http://127.0.0.1:5000 in your browser.")
app.run(host='127.0.0.1', port=5000)