معماری کلاستر خزش پرسرعت با Playwright Stealth و مدیریت بهینه رم

import asyncio
import psutil
from typing import Optional, List
from playwright.async_api import async_playwright, Browser, BrowserContext, Route

STEALTH_INIT_SCRIPT = """
Object.defineProperty(navigator, 'webdriver', { get: () => undefined });
window.chrome = { runtime: {}, loadTimes: function() {}, csi: function() {} };
Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5] });
Object.defineProperty(navigator, 'languages', { get: () => ['en-US', 'en'] });
const getParameter = WebGLRenderingContext.prototype.getParameter;
WebGLRenderingContext.prototype.getParameter = function(param) {
    if (param === 37445) return 'Intel Inc.';
    if (param === 37446) return 'Intel Iris OpenGL Engine';
    return getParameter.apply(this, arguments);
};
"""

class PlaywrightStealthCluster:
    BLOCKED_RESOURCE_TYPES = {'image', 'media', 'font', 'stylesheet'}
    BLOCKED_DOMAINS = ('google-analytics.com', 'doubleclick.net', 'hotjar.com', 'clarity.ms')

    def __init__(self, concurrency: int = 8, max_pages_per_browser: int = 250, max_ram_percent: float = 82.0):
        self.semaphore = asyncio.Semaphore(concurrency)
        self.max_pages_per_browser = max_pages_per_browser
        self.max_ram_percent = max_ram_percent
        self._playwright = None
        self._browser: Optional[Browser] = None
        self._processed_pages = 0
        self._lock = asyncio.Lock()

    async def start(self) -> None:
        self._playwright = await async_playwright().start()
        await self._launch_browser()

    async def _launch_browser(self) -> None:
        if self._browser:
            await self._browser.close()
        self._browser = await self._playwright.chromium.launch(
            headless=True,
            args=[
                '--disable-blink-features=AutomationControlled',
                '--disable-dev-shm-usage',
                '--no-sandbox',
                '--disable-gpu',
                '--js-flags=--max-old-space-size=512'
            ]
        )
        self._processed_pages = 0

    async def _intercept_route(self, route: Route) -> None:
        req = route.request
        if req.resource_type in self.BLOCKED_RESOURCE_TYPES or any(d in req.url for d in self.BLOCKED_DOMAINS):
            await route.abort()
        else:
            await route.continue_()

    async def _ensure_healthy_browser(self) -> None:
        async with self._lock:
            ram_usage = psutil.virtual_memory().percent
            if self._processed_pages >= self.max_pages_per_browser or ram_usage > self.max_ram_percent:
                await self._launch_browser()

    async def fetch_rendered_html(self, url: str, wait_selector: str = 'body', proxy: Optional[str] = None) -> str:
        async with self.semaphore:
            await self._ensure_healthy_browser()
            context: BrowserContext = await self._browser.new_context(
                viewport={'width': 1440, 'height': 900},
                user_agent='Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36',
                proxy={'server': proxy} if proxy else None
            )
            await context.add_init_script(STEALTH_INIT_SCRIPT)
            page = await context.new_page()
            await page.route('**/*', self._intercept_route)

            try:
                await page.goto(url, wait_until='domcontentloaded', timeout=25000)
                await page.wait_for_selector(wait_selector, timeout=10000)
                self._processed_pages += 1
                return await page.content()
            finally:
                await page.close()
                await context.close()

    async def close(self) -> None:
        if self._browser:
            await self._browser.close()
        if self._playwright:
            await self._playwright.stop()