import json import logging import random from playwright.sync_api import Browser, BrowserContext, Playwright try: from playwright_stealth import stealth_sync except ImportError: stealth_sync = None from ..core.config import Settings logger = logging.getLogger("MOBILEDE_scraper.browser") def _build_init_script() -> str: # Маскировка браузера. hardware_concurrency = random.choice([4, 8, 12, 16]) device_memory = random.choice([4, 8, 16]) languages = ["en-US", "en"] return f""" (() => {{ const define = (obj, prop, value) => {{ try {{ Object.defineProperty(obj, prop, {{ get: () => value, configurable: true }}); }} catch (e) {{}} }}; define(navigator, 'webdriver', undefined); define(navigator, 'platform', 'Win32'); define(navigator, 'vendor', 'Google Inc.'); define(navigator, 'language', '{languages[0]}'); define(navigator, 'languages', {json.dumps(languages)}); define(navigator, 'hardwareConcurrency', {hardware_concurrency}); define(navigator, 'deviceMemory', {device_memory}); define(navigator, 'maxTouchPoints', 0); if (!window.chrome) {{ Object.defineProperty(window, 'chrome', {{ value: {{ runtime: {{}}, app: {{}}, csi: () => ({{}}), loadTimes: () => ({{}}) }}, configurable: true }}); }} const originalQuery = navigator.permissions && navigator.permissions.query; if (originalQuery) {{ navigator.permissions.query = (params) => ( params && params.name === 'notifications' ? Promise.resolve({{ state: Notification.permission }}) : originalQuery(params) ); }} const originalGetParameter = WebGLRenderingContext.prototype.getParameter; WebGLRenderingContext.prototype.getParameter = function(parameter) {{ if (parameter === 37445) return 'Intel Inc.'; if (parameter === 37446) return 'Intel Iris OpenGL Engine'; return originalGetParameter.call(this, parameter); }}; }})(); """ class BrowserFactory: def __init__(self, settings: Settings) -> None: self.settings = settings def _resolve_engine(self) -> str: # Выбор движка. engine = self.settings.browser_engine.strip().lower() if engine == "auto": # Для MOBILEDE стабильнее Chromium. return "chromium" if engine in ("firefox", "chromium"): return engine logger.warning("Unknown MOBILEDE_BROWSER_ENGINE=%r, falling back to auto", engine) return "chromium" def create_browser(self, playwright: Playwright) -> Browser: engine = self._resolve_engine() proxy_dict = self.settings.proxy.to_playwright_dict() logger.info("Resolved browser engine: requested=%s resolved=%s", self.settings.browser_engine, engine) if engine == "firefox": launch_kwargs: dict = {"headless": self.settings.headless} if proxy_dict: launch_kwargs["proxy"] = proxy_dict logger.info("Using proxy: %s", self.settings.proxy.server) # Настройки Firefox. launch_kwargs["firefox_user_prefs"] = { "dom.webdriver.enabled": False, "useAutomationExtension": False, # Базовые оптимизации. "media.autoplay.default": 5, "media.volume_scale": "0.0", "media.audio.playback.standalone": False, "dom.ipc.processCount": 1, "dom.ipc.plugins.enabled": False, "browser.cache.disk.enable": False, "browser.cache.memory.enable": True, "browser.cache.memory.max_entry_size": 8192, "network.prefetch-next": False, "network.dns.disablePrefetch": True, "permissions.default.image": 2, "javascript.options.mem.gc_incremental_mark_slice_ms": 20, } logger.info("Launching Firefox (headless=%s)", self.settings.headless) return playwright.firefox.launch(**launch_kwargs) # Запуск Chromium. args = [ "--disable-blink-features=AutomationControlled", "--no-default-browser-check", "--disable-dev-shm-usage", "--disable-features=IsolateOrigins,site-per-process", ] if self.settings.headless: args.append("--headless=new") pw_headless = False logger.info("Using Chromium new-headless mode (--headless=new)") else: pw_headless = False launch_kwargs = {"headless": pw_headless, "args": args} if proxy_dict: launch_kwargs["proxy"] = proxy_dict logger.info("Using proxy: %s", self.settings.proxy.server) try: logger.info("Trying to launch real Chrome channel") return playwright.chromium.launch(channel="chrome", **launch_kwargs) except Exception: logger.warning("Chrome channel launch failed, falling back to Chromium") return playwright.chromium.launch(**launch_kwargs) def create_context(self, browser: Browser) -> BrowserContext: viewport = random.choice(self.settings.fingerprint.viewport_presets) timezone_id = random.choice(self.settings.fingerprint.timezone_candidates) color_scheme = random.choice(["light", "dark"]) is_firefox = browser.browser_type.name == "firefox" ctx_kwargs: dict = { "viewport": viewport, "screen": viewport, "locale": self.settings.fingerprint.locale, "timezone_id": timezone_id, "color_scheme": color_scheme, "java_script_enabled": True, "ignore_https_errors": False, } if is_firefox: # Заголовки Firefox. ctx_kwargs["user_agent"] = ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:128.0) " "Gecko/20100101 Firefox/128.0" ) ctx_kwargs["extra_http_headers"] = { "Accept-Language": "en-US,en;q=0.5", "DNT": "1", "Upgrade-Insecure-Requests": "1", } else: ctx_kwargs["user_agent"] = self.settings.fingerprint.user_agent ctx_kwargs["device_scale_factor"] = random.choice([1, 1.25]) ctx_kwargs["is_mobile"] = False ctx_kwargs["has_touch"] = False ctx_kwargs["extra_http_headers"] = { "Accept-Language": "en-US,en;q=0.9", "DNT": "1", "Upgrade-Insecure-Requests": "1", "Sec-CH-UA": self.settings.fingerprint.sec_ch_ua, "Sec-CH-UA-Mobile": "?0", "Sec-CH-UA-Platform": '"Windows"', } context = browser.new_context(**ctx_kwargs) context.set_default_timeout(self.settings.default_timeout_ms) context.set_default_navigation_timeout(self.settings.default_timeout_ms) if not is_firefox: # Маскировка Chromium. context.add_init_script(_build_init_script()) if stealth_sync: context.on("page", lambda page: stealth_sync(page)) logger.debug("playwright-stealth attached to context") return context @staticmethod def enable_resource_blocking(page) -> None: # Блокируем тяжёлые ресурсы. BLOCKED_TYPES = {"image", "stylesheet", "font", "media"} BLOCKED_URL_PATTERNS = ( "google-analytics", "googletagmanager", "facebook.net", "doubleclick.net", "hotjar", "newrelic", ".woff", ".woff2", "analytics", "tracking", "adservice", ) def _handle_route(route): req = route.request if req.resource_type in BLOCKED_TYPES: route.abort() return url = req.url.lower() if any(pat in url for pat in BLOCKED_URL_PATTERNS): route.abort() return route.continue_() page.route("**/*", _handle_route)