import asyncio import logging from playwright.async_api import async_playwright from playwright_stealth import Stealth import os import json from dotenv import load_dotenv from datetime import datetime # Setup logging logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class AviatorBot: def __init__(self, config): self.config = config self.browser = None self.context = None self.page = None self.game_frame = None self.is_running = False self.dry_run = config.get('DRY_RUN', True) # Default to True for safety async def setup(self): """Initialize browser, stealth mode and session state""" # Performance/Server args launch_args = ["--disable-web-security", "--disable-features=IsolateOrigins,site-per-process"] # Configure Proxy if available in config proxy_settings = None if self.config.get('PROXY_SERVER'): proxy_settings = { "server": self.config['PROXY_SERVER'] } if self.config.get('PROXY_USER'): proxy_settings["username"] = self.config['PROXY_USER'] proxy_settings["password"] = self.config['PROXY_PASS'] logger.info(f"Using Proxy: {self.config['PROXY_SERVER']}") self.browser = await self.playwright.chromium.launch( headless=False, proxy=proxy_settings, args=["--disable-blink-features=AutomationControlled"] ) # Load session state if exists state_path = "state.json" if os.path.exists(state_path): logger.info("Loading existing session state from state.json") self.context = await self.browser.new_context( storage_state=state_path, viewport={'width': 1920, 'height': 1080}, user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36" ) else: self.context = await self.browser.new_context( viewport={'width': 1920, 'height': 1080}, user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36" ) self.page = await self.context.new_page() try: await Stealth().apply_stealth_async(self.page) logger.info("Stealth applied.") except Exception as e: logger.warning(f"Stealth application failed: {e}") logger.info("Browser setup complete.") async def login(self): """Use state.json if it exists, otherwise do fresh login via visible browser""" if os.path.exists("state.json"): logger.info("state.json found — checking session validity...") # Navigate to site to verify session await self.page.goto("https://luckyeie.com/", wait_until="domcontentloaded", timeout=60000) await asyncio.sleep(3) # If Login button is still visible, session is invalid login_btn = self.page.locator("button:visible:has-text('Log in'), button:visible:has-text('Login')").first is_logged_out = await login_btn.count() > 0 if is_logged_out: logger.warning("Session is invalid (Login button visible). Re-logging in...") if os.path.exists("state.json"): os.remove("state.json") await self._do_login() else: logger.info("Session is valid. Proceeding to game.") return await self._do_login() async def _do_login(self): """Perform actual browser login and save HttpOnly cookies""" logger.info("Opening login page in browser...") await self.page.goto("https://luckyeie.com/", wait_until="domcontentloaded", timeout=60000) await asyncio.sleep(2) try: # Click login button on header logger.info("Step 1: Clicking header login button...") login_btn = self.page.locator("button:visible:has-text('Log in'), button:visible:has-text('Login'), a:visible:has-text('Log in'), a:visible:has-text('Login')").first await login_btn.click() await asyncio.sleep(3) await self.page.screenshot(path="login_step1_modal.png") # Switch to E-mail/Email tab in login modal logger.info("Step 2: Switching to Email tab...") # Use the specific data-test-id we found: signInByPhone-form-tab-email email_tab = self.page.locator("[data-test-id*='tab-email'], button:has-text('Email'), [role='tab']:has-text('Email')").locator("visible=true").first try: await email_tab.wait_for(state='visible', timeout=7000) await email_tab.click(force=True) logger.info("Email tab clicked (force=True).") await asyncio.sleep(2) # Double check if we switched - if not try clicking again or using dispatch_event # Look for an email input to confirm switch if await self.page.locator("input[type='email'], input[placeholder*='mail' i]").locator("visible=true").count() == 0: logger.warning("Email input not visible after click, trying dispatch_event('click')...") await email_tab.evaluate("el => el.click()") await asyncio.sleep(2) except Exception as e: logger.warning(f"Email tab switch failed: {e}") await self.page.screenshot(path="login_step2_email_tab.png") # Fill email logger.info("Step 3: Filling email...") email_input = self.page.locator("input[type='email'], input[name='email'], input[placeholder*='mail' i]").locator("visible=true").first await email_input.wait_for(state='visible', timeout=10000) await email_input.fill(self.config['EMAIL']) # Fill password logger.info("Step 4: Filling password...") pass_input = self.page.locator("input[type='password'], input[name='password']").locator("visible=true").first await pass_input.fill(self.config['PASSWORD']) await self.page.screenshot(path="login_step3_filled.png") # Submit login logger.info("Step 5: Submitting login form...") # Target the green 'Log in' button inside the modal specifically submit_btn = self.page.locator("[class*='modal'] button:has-text('Log in'), [class*='modal'] button[type='submit']").locator("visible=true").first await submit_btn.click() logger.info("Credentials submitted. Waiting 15s for login to finalize and redirect...") await asyncio.sleep(15) await self.page.screenshot(path="login_step4_after_submit.png") # Save full state including HttpOnly cookies from Playwright context await self.context.storage_state(path="state.json") logger.info("SUCCESS: Full session saved to state.json (includes HttpOnly cookies)") except Exception as e: logger.error(f"Login error: {e}. Please login manually in the open browser window.") logger.info("Waiting 30s for manual login...") await asyncio.sleep(30) await self.context.storage_state(path="state.json") logger.info("Session saved after manual window.") async def enter_game(self): """Navigate to Aviator and find the game iframe""" logger.info(f"Navigating to: {self.config['BASE_URL']}") await self.page.goto(self.config['BASE_URL'], wait_until="domcontentloaded", timeout=60000) logger.info("Game page loaded. Waiting up to 30s for game iframe to initialize...") for _ in range(30): for f in self.page.frames: logger.debug(f" Frame URL: {f.url}") if "spribe" in f.url or "aviator" in f.url: # Check if it's the demo game if "demo" in f.url: logger.warning("⚠️ Demo game loaded instead of real game!") self.game_frame = f logger.info(f"✅ Aviator iframe found: {f.url}") return True # Wait 1s and check again await asyncio.sleep(1) logger.error("❌ Could not find Aviator game iframe after scanning all frames.") return False async def monitor_results(self): """Monitor game events by polling the DOM inside the Aviator iframe""" logger.info("Monitoring results started via DOM polling...") last_result = "" last_active = "" while self.is_running: try: if not self.game_frame: await asyncio.sleep(1) continue # Execute JS inside the iframe to grab the current state state = await self.game_frame.evaluate('''() => { let result = null; let active = null; // Historical payout bubbles (most recent is usually first or last depending on layout) let bubbles = Array.from(document.querySelectorAll('.payouts-block app-bubble-multiplier, .bubble-multiplier, app-bubble-multiplier, .payout')); if (bubbles.length > 0) { // Usually the first one in the list is the most recent completed result = bubbles[0].innerText.trim(); } // The large active flying multiplier on screen let main_text = document.querySelector('.fliers-block .fliers-x, .current-multiplier, [class*="fliers-x"], app-multiplier'); if (main_text) { active = main_text.innerText.trim(); } return {result: result, active: active}; }''') result = state.get('result') active = state.get('active') # Detect active flight (Round Start) if active and active != last_active and "x" in active.lower() and "wait" not in active.lower(): # We only log round start if it's a small multiplier indicating takeoff val_str = active.lower().replace("x", "").strip() try: if float(val_str) < 1.10 and last_active == "": logger.info(">>> NEW ROUND STARTING...") await self.on_round_start() except: pass last_active = active # Detect when flight ends (active disappears or says 'flew away') if not active and last_active: last_active = "" # Detect new completed result if result and result != last_result and "x" in result.lower(): last_result = result logger.info(f">>> ROUND RESULT: {result}") self.log_result(result) # Poll every 0.3 seconds await asyncio.sleep(0.3) except Exception as e: # Ignore expected errors like frame detached during reload if "Target page, context or browser has been closed" in str(e): break await asyncio.sleep(1) async def on_round_start(self): """Logic to execute when a new round begins""" if self.dry_run: logger.info("[DRY-RUN] Round started. Skipping bet.") return # 1. Place Bet bet_success = await self.place_bet(self.config['BASE_BET']) if bet_success: # 2. Start looking for target multiplier to Cash Out asyncio.create_task(self.auto_cash_out()) async def place_bet(self, amount): """Interact with the Bet button""" if self.dry_run: return False try: if self.game_frame: # Set amount in input field first # Selector: .input .bet-input await self.game_frame.fill(".input .bet-input", str(amount)) # Click 'BET' button # Selector: button.bet-button.green bet_btn = self.game_frame.locator("button.bet-button").first await bet_btn.click() logger.info(f"Bet placed: {amount}") return True except Exception as e: logger.error(f"Failed to place bet: {e}") return False async def auto_cash_out(self): """Monitor live multiplier and click Cash Out at target""" if self.dry_run: return target = self.config['TARGET_MULTIPLIER'] logger.info(f"Waiting for {target}x to cash out...") # ... logic for actual cashout ... def log_result(self, result): """Save multiplier results to a local file for strategy analysis""" try: timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') # Check if file needs header needs_header = not os.path.exists("game_history.csv") with open("game_history.csv", "a") as f: if needs_header: f.write("timestamp,multiplier\n") f.write(f"{timestamp},{result}\n") logger.info(f"Result logged: {result}") except Exception as e: logger.error(f"Failed to log result: {e}") async def start(self): self.is_running = True logger.info("[START] Initializing Playwright...") async with async_playwright() as self.playwright: await self.setup() await self.login() if await self.enter_game(): logger.info("Game entered. Starting monitor loop...") await self.monitor_results() async def main_bot(): load_dotenv() config = { "EMAIL": os.getenv("EMAIL"), "PASSWORD": os.getenv("PASSWORD"), "BASE_URL": os.getenv("BASE_URL"), "DRY_RUN": os.getenv("DRY_RUN", "True").lower() == "true", "BASE_BET": float(os.getenv("BASE_BET", 10.0)), "TARGET_MULTIPLIER": float(os.getenv("TARGET_MULTIPLIER", 1.5)), "PROXY_SERVER": os.getenv("PROXY_SERVER"), "PROXY_USER": os.getenv("PROXY_USER"), "PROXY_PASS": os.getenv("PROXY_PASS") } bot = AviatorBot(config) await bot.start() if __name__ == "__main__": asyncio.run(main_bot())