9d49a99916
Chrome extension for PokemonCenter monitoring with Discord notifications. Includes Python scripts for Target monitoring. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
184 lines
6.1 KiB
Python
184 lines
6.1 KiB
Python
"""
|
|
Browser automation with Playwright and stealth patches
|
|
Handles bot protection bypass for retail sites
|
|
"""
|
|
|
|
import logging
|
|
from playwright.sync_api import sync_playwright, Browser, Page, BrowserContext
|
|
from playwright_stealth import Stealth
|
|
from config import HEADLESS, SLOW_MO, CHROME_USER_DATA_DIR, USE_REAL_CHROME
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Create stealth instance
|
|
stealth = Stealth()
|
|
|
|
|
|
class StealthBrowser:
|
|
"""Manages a stealth browser instance that bypasses bot detection"""
|
|
|
|
def __init__(self):
|
|
self.playwright = None
|
|
self.browser: Browser = None
|
|
self.context: BrowserContext = None
|
|
self.using_persistent = False
|
|
|
|
def start(self):
|
|
"""Initialize the browser"""
|
|
logger.info("Starting stealth browser...")
|
|
self.playwright = sync_playwright().start()
|
|
|
|
# Check if we should use a persistent Chrome profile
|
|
if CHROME_USER_DATA_DIR:
|
|
logger.info(f"Using Chrome profile from: {CHROME_USER_DATA_DIR}")
|
|
self._start_with_profile()
|
|
else:
|
|
self._start_fresh()
|
|
|
|
logger.info("Browser started successfully")
|
|
|
|
def _start_fresh(self):
|
|
"""Start with a fresh browser profile"""
|
|
# Launch browser - use real Chrome if configured
|
|
launch_options = {
|
|
"headless": HEADLESS,
|
|
"slow_mo": SLOW_MO,
|
|
"args": [
|
|
"--disable-blink-features=AutomationControlled",
|
|
"--disable-dev-shm-usage",
|
|
"--no-sandbox",
|
|
]
|
|
}
|
|
|
|
if USE_REAL_CHROME:
|
|
launch_options["channel"] = "chrome"
|
|
logger.info("Using real Chrome browser")
|
|
|
|
self.browser = self.playwright.chromium.launch(**launch_options)
|
|
|
|
# Create context with realistic viewport and user agent
|
|
self.context = 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/131.0.0.0 Safari/537.36",
|
|
locale="en-US",
|
|
timezone_id="America/New_York",
|
|
)
|
|
|
|
# Apply stealth to context (affects all pages created from it)
|
|
stealth.apply_stealth_sync(self.context)
|
|
|
|
def _start_with_profile(self):
|
|
"""Start using an existing Chrome profile (better for bot detection)"""
|
|
self.using_persistent = True
|
|
|
|
# Launch persistent context with user's REAL Chrome (not Playwright's Chromium)
|
|
# This is needed because Chrome encrypts credentials with its own binary
|
|
self.context = self.playwright.chromium.launch_persistent_context(
|
|
user_data_dir=CHROME_USER_DATA_DIR,
|
|
channel="chrome", # Use real Chrome, not Playwright's Chromium
|
|
headless=HEADLESS,
|
|
slow_mo=SLOW_MO,
|
|
viewport={"width": 1920, "height": 1080},
|
|
locale="en-US",
|
|
timezone_id="America/New_York",
|
|
args=[
|
|
"--disable-blink-features=AutomationControlled",
|
|
"--disable-dev-shm-usage",
|
|
]
|
|
)
|
|
|
|
# Apply stealth
|
|
stealth.apply_stealth_sync(self.context)
|
|
|
|
def new_page(self) -> Page:
|
|
"""Create a new page with stealth patches applied"""
|
|
if not self.context:
|
|
raise RuntimeError("Browser not started. Call start() first.")
|
|
|
|
page = self.context.new_page()
|
|
return page
|
|
|
|
def get_page_content(self, url: str, wait_for_selector: str = None, timeout: int = 30000) -> tuple[Page, str]:
|
|
"""
|
|
Navigate to URL and return page + HTML content
|
|
|
|
Args:
|
|
url: URL to navigate to
|
|
wait_for_selector: CSS selector to wait for before returning
|
|
timeout: Max time to wait in milliseconds
|
|
|
|
Returns:
|
|
Tuple of (page, html_content)
|
|
"""
|
|
page = self.new_page()
|
|
|
|
try:
|
|
logger.info(f"Navigating to: {url}")
|
|
page.goto(url, wait_until="domcontentloaded", timeout=timeout)
|
|
|
|
# Wait for specific element if provided
|
|
if wait_for_selector:
|
|
logger.debug(f"Waiting for selector: {wait_for_selector}")
|
|
page.wait_for_selector(wait_for_selector, timeout=timeout)
|
|
else:
|
|
# Wait for DOM to be ready, then give extra time for JS rendering
|
|
try:
|
|
page.wait_for_load_state("networkidle", timeout=15000)
|
|
except:
|
|
# networkidle can timeout on dynamic sites, that's OK
|
|
logger.debug("networkidle timeout, continuing anyway")
|
|
|
|
# Let dynamic content render
|
|
page.wait_for_timeout(5000)
|
|
|
|
# Scroll down to trigger lazy loading
|
|
page.evaluate("window.scrollTo(0, document.body.scrollHeight / 2)")
|
|
page.wait_for_timeout(2000)
|
|
page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
|
|
page.wait_for_timeout(2000)
|
|
page.evaluate("window.scrollTo(0, 0)")
|
|
page.wait_for_timeout(1000)
|
|
|
|
content = page.content()
|
|
logger.info(f"Page loaded, content length: {len(content)}")
|
|
|
|
return page, content
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error loading page {url}: {e}")
|
|
page.close()
|
|
raise
|
|
|
|
def stop(self):
|
|
"""Clean up browser resources"""
|
|
logger.info("Stopping browser...")
|
|
if self.context:
|
|
self.context.close()
|
|
if self.browser and not self.using_persistent:
|
|
# Only close browser if not using persistent context
|
|
self.browser.close()
|
|
if self.playwright:
|
|
self.playwright.stop()
|
|
logger.info("Browser stopped")
|
|
|
|
|
|
# Global browser instance
|
|
_browser: StealthBrowser = None
|
|
|
|
|
|
def get_browser() -> StealthBrowser:
|
|
"""Get or create the global browser instance"""
|
|
global _browser
|
|
if _browser is None:
|
|
_browser = StealthBrowser()
|
|
_browser.start()
|
|
return _browser
|
|
|
|
|
|
def shutdown_browser():
|
|
"""Shutdown the global browser instance"""
|
|
global _browser
|
|
if _browser:
|
|
_browser.stop()
|
|
_browser = None
|