Spaces:
Running
Running
| 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()) | |