I’m scraping data daily using python playwright. On my local Windows 10 machine, I had some issues at first, but I got things working using BrowserForge + residential smart proxy (for fingerprints and legit IPs). That setup worked perfectly but only locally. The problem started when I moved my scraping tasks to the cloud. I’m using AWS Batch with Fargate to run the scripts, and that’s where everything breaks. After hitting 403 errors in the cloud, I tried alternatives like Camoufox and Patchright – they work great locally in headed mode, but as soon as I run them on AWS I am instantly getting blocked and I see 403 and a captcha. The captcha requires you to press and hold a button, and even when I solve it manually, I still get 403s afterward. I also tried xvfb to simulate a display and run in headed mode, but it didn’t help – same result: 403. I also implemented oxymouse to stimulate mouse movements but I am getting blocked immediately after page is loaded so it is useless. At this point I’m out of ideas. Has anyone managed to scrape easypara.fr reliably from AWS (especially with Playwright)? Any tricks, setups, or tools I might’ve missed? I have several other eretailers with cloudflare and advanced captchas protection (eva.ua, walmart.com.mx, chewy.com etc.). Below is my implementation of playwright driver for your reference: from patchright.async_api import async_playwright, TimeoutError as PlaywrightTimeoutError, Page, BrowserContext, Request, Error as PlaywrightError from pathlib import Path import os from io import BytesIO from datetime import datetime import re as reg import random import shutil import asyncio from oxymouse import OxyMouse from logs.logs import module_logger class BrowserObject: def __init__(self, page: Page) -> None: if page: self.browser_context: BrowserContext = page.context self.page: Page = page self.initial_call: bool = True self.count_requests: int = 0 class AsyncPatchRightDriver: def __init__( self, async_playwright: async_playwright, args: list = None, headless: bool = False, proxy: dict = None, route_block: list = [], resource_block: list = [], rotate_every: int = 10, user_data_dir: str = os.path.join(Path.home(), ".patchright_user_data") # Added for persistent context ): self.route_block = route_block self.resource_block = resource_block self.async_playwright = async_playwright self.browser_name_literal = "chromium" # Hardcoded to chromium because it is the only one supported by Patchright self.browser_name = self.async_playwright.chromium self.headless = headless self.args = args self.proxy = proxy self.rotate_every = rotate_every self.logger = module_logger self.s3_utils = S3Utils(logger=self.logger) self.user_data_dir = user_data_dir self.headers_printed = False async def async_init(self): await self.clear_user_data() self.browser_object = await self.create_new_page_with_context() return self def __await__(self): return self.async_init().__await__() async def emulate_human_behavior(self, browser_object: BrowserObject = None): """Emulates human-like browsing behavior using OxyMouse for mouse movements.""" if not browser_object: browser_object = self.browser_object try: # Get current viewport size viewport = await browser_object.page.evaluate("() => ({ width: window.innerWidth, height: window.innerHeight })") mouse = OxyMouse(algorithm="gaussian") self.logger.info("Initialized OxyMouse with gaussian algorithm") await asyncio.sleep(random.uniform(0.05, 0.15)) # Step 1: Initial scroll first_scroll = random.randint(100, 300) await browser_object.page.evaluate(f"window.scrollBy(0, {first_scroll})") self.logger.info(f"Performed initial scroll by {first_scroll} pixels") await asyncio.sleep(random.uniform(0.2, 0.5)) # Step 2: Generate 1-3 random mouse movements num_movements = random.randint(1, 3) self.logger.info(f"Generating {num_movements} mouse movement sequences") for i in range(num_movements): # Generate random coordinates within viewport movements = mouse.generate_random_coordinates( viewport_width=viewport["width"], viewport_height=min(600, first_scroll + 200) )[:5] # Limit to 5 coordinates per sequence for x, y in movements: await browser_object.page.mouse.move(x, y) await asyncio.sleep(random.uniform(0.03, 0.06)) self.logger.info(f"Completed mouse movement sequence {i+1}") await asyncio.sleep(random.uniform(0.15, 0.35)) # Step 3: Occasionally scroll back up if random.random() > 0.8: scroll_back = random.randint(30, 100) await browser_object.page.evaluate(f"window.scrollBy(0, -{scroll_back})") self.logger.info(f"Scrolled back up by {scroll_back} pixels") await asyncio.sleep(random.uniform(0.15, 0.3)) # Step 4: Occasionally scroll further down if random.random() > 0.6: scroll_down = random.randint(100, 400) await browser_object.page.evaluate(f"window.scrollBy(0, {scroll_down})") self.logger.info(f"Scrolled further down by {scroll_down} pixels") await asyncio.sleep(random.uniform(0.15, 0.3)) self.logger.info("Human-like behavior emulation completed with OxyMouse") except Exception as e: self.logger.warning(f"Error during human behavior emulation with OxyMouse: {e}") raise async def create_new_page_with_context(self, browser_object: BrowserObject = None) -> BrowserObject: if not hasattr(self, "browser_context"): self.browser_context = await self.browser_name.launch_persistent_context( user_data_dir=self.user_data_dir, channel="chrome", headless=self.headless, no_viewport=True, args=self.args, proxy=self.proxy, ) module_logger.info("PatchRight browser context launched successfully") page = await self.browser_context.new_page() if browser_object: browser_object.page = page browser_object.browser_context = self.browser_context browser_object.count_requests = 0 browser_object.initial_call = True return browser_object else: browser_object = BrowserObject(page) return browser_object async def async_go(self, url, browser_object: BrowserObject = None): if not browser_object: browser_object = self.browser_object await self.update_page_count(browser_object=browser_object) if self.resource_block: self.logger.info(f"Blocking resources: {self.resource_block}") await browser_object.page.route("**/*", lambda route: self.block_resources(route, self.resource_block)) if self.route_block: self.logger.info(f"Blocking network files: {self.route_block}") await browser_object.page.route("**/*", lambda route: self.block_network_files(route, self.route_block)) if not self.headers_printed: browser_object.page.on("request", lambda request: self.log_browser_info(request=request, url=url)) self.logger.info(f"Navigating to {url}") await self.wait_random(0.5, browser_object=browser_object) try: response = await browser_object.page.goto(url, wait_until="load", timeout=60000) await self.emulate_human_behavior(browser_object) if response and 400 <= response.status <= 599: await self.make_screenshot_to_s3( bucket_name=secret_manager.get_secret("AWS_STORAGE_BUCKET_NAME"), folder="screenshot_errors", browser_object=browser_object ) module_logger.warning(f"Received {response.status} status code for {url}") return browser_object except Exception as e: self.logger.error(f"Navigation failed: {str(e)}") raise async def clear_user_data(self): """Clear the user data directory.""" try: if hasattr(self, "browser_context"): await self.browser_context.close() if os.path.exists(self.user_data_dir): self.logger.info(f"Clearing user data directory: {self.user_data_dir}") shutil.rmtree(self.user_data_dir, ignore_errors=True) self.logger.info("User data directory cleared") except Exception as e: self.logger.error(f"Error clearing user data: {e}") async def block_resources(self, route, resource_types_to_block): if route.request.resource_type in resource_types_to_block: await route.abort() else: await route.continue_() async def block_network_files(self, route, files_to_block): request_url = route.request.url if any(file_to_block in request_url for file_to_block in files_to_block): await route.abort() else: await route.continue_() Thanks in advance! Continue reading...
