from abc import ABC, abstractmethod from collections.abc import Awaitable, Callable from dataclasses import dataclass from typing import Any from marketing_intelligence.evasion.behavior import BehaviorEmulator from marketing_intelligence.evasion.profiles import DeviceProfile @dataclass class PageSnapshot: url: str html: str title: str | None = None final_url: str | None = None response_time_ms: int = 0 class BrowserAdapter(ABC): def __init__( self, profile: DeviceProfile, behavior: BehaviorEmulator, proxy: dict | None = None, ) -> None: self.profile = profile self.behavior = behavior self.proxy = proxy @abstractmethod async def fetch_with_scroll( self, url: str, scroll_distance: int = 3000 ) -> PageSnapshot: """Navigate to url, wait, scroll to trigger lazy-load, return full HTML.""" ... @abstractmethod async def fetch_interactive( self, url: str, callback: Callable[[Any, str, str, int], Awaitable[Any]], ) -> Any: """Navigate to url, then run callback(page, html, final_url, response_time_ms).""" ... _PLAYWRIGHT_ENGINES = ("camoufox", "chromium", "webkit") _NODRIVER_ENGINES = ("nodriver",) _DRISSIONPAGE_ENGINES = ("drissionpage",) def create_adapter( engine: str = "camoufox", profile_name: str = "desktop", proxy: dict | None = None, human_behavior: bool = True, delay_factor: float = 1.0, scroll_speed: str = "normal", ) -> BrowserAdapter: """Factory — returns the right BrowserAdapter for the given engine.""" from marketing_intelligence.evasion.profiles import get_profile profile = get_profile(profile_name) behavior = BehaviorEmulator( human=human_behavior, delay_factor=delay_factor, scroll_speed=scroll_speed, ) if engine in _PLAYWRIGHT_ENGINES: from marketing_intelligence.browser.playwright import PlaywrightAdapter return PlaywrightAdapter( engine=engine, profile=profile, behavior=behavior, proxy=proxy ) if engine in _NODRIVER_ENGINES: from marketing_intelligence.browser.nodriver import NoDriverAdapter return NoDriverAdapter(profile=profile, behavior=behavior, proxy=proxy) if engine in _DRISSIONPAGE_ENGINES: from marketing_intelligence.browser.drissionpage import DrissionPageAdapter return DrissionPageAdapter(profile=profile, behavior=behavior, proxy=proxy) valid = _PLAYWRIGHT_ENGINES + _NODRIVER_ENGINES + _DRISSIONPAGE_ENGINES raise ValueError(f"Unknown engine: {engine!r}. Valid: {valid}") def create_adapter_from_session( session: object, proxy: dict | None = None ) -> BrowserAdapter: """Create adapter from a BrowserSession config object.""" return create_adapter( engine=getattr(session, "engine", "camoufox"), profile_name=getattr(session, "profile_name", "desktop"), proxy=proxy, human_behavior=getattr(session, "human_behavior", True), delay_factor=getattr(session, "delay_factor", 1.0), scroll_speed=getattr(session, "scroll_speed", "normal"), )