Spaces:
Paused
Paused
| # browser/manager.py | |
| import asyncio | |
| import logging | |
| from playwright.async_api import async_playwright, Browser, BrowserContext | |
| from config.settings import settings | |
| from browser.stealth import ( | |
| get_stealth_launch_args, | |
| get_stealth_context_options, | |
| block_trackers_and_media, | |
| STEALTH_INIT_SCRIPT | |
| ) | |
| logger = logging.getLogger(__name__) | |
| class BrowserManager: | |
| """ | |
| Verwaltet Playwright Browser-Instanzen. | |
| Bietet persistente (für eingeloggte Sessions) und ephemere (Incognito) Kontexte. | |
| """ | |
| def __init__(self): | |
| self._playwright = None | |
| self._browser: Browser | None = None | |
| self._contexts: dict[str, BrowserContext] = {} | |
| self._initialize_lock = asyncio.Lock() | |
| def is_ready(self) -> bool: | |
| return bool(self._browser and self._browser.is_connected()) | |
| async def initialize(self): | |
| async with self._initialize_lock: | |
| if self.is_ready: | |
| return | |
| if self._browser: | |
| try: | |
| await self._browser.close() | |
| except Exception: | |
| pass | |
| self._browser = None | |
| if self._playwright: | |
| try: | |
| await self._playwright.stop() | |
| except Exception: | |
| pass | |
| self._playwright = await async_playwright().start() | |
| launch_args = get_stealth_launch_args() | |
| self._browser = await self._playwright.chromium.launch( | |
| headless=True, | |
| args=launch_args, | |
| ignore_default_args=["--enable-automation"] | |
| ) | |
| logger.info("Chromium Browser initialized.") | |
| async def get_ephemeral_context(self, connection_mode: str = "normal", state_data: dict | None = None) -> BrowserContext: | |
| """Erstellt einen neuen, isolierten Incognito-Kontext (z.B. für Recherche oder injizierte Session).""" | |
| if not self.is_ready: | |
| await self.initialize() | |
| options = get_stealth_context_options(connection_mode) | |
| if state_data: | |
| options["storage_state"] = state_data | |
| try: | |
| context = await self._browser.new_context(**options) | |
| except Exception as exc: | |
| if "closed" not in str(exc).lower() and "target" not in str(exc).lower(): | |
| raise | |
| logger.warning("Chromium connection died; recreating browser once") | |
| await self.initialize() | |
| context = await self._browser.new_context(**options) | |
| # Anti-Bot Stealth Init Script injizieren | |
| await context.add_init_script(STEALTH_INIT_SCRIPT) | |
| # Security & Stealth Routes einhängen | |
| await context.route("**/*", block_trackers_and_media) | |
| return context | |
| async def get_persistent_context(self, session_id: str, connection_mode: str = "normal") -> BrowserContext: | |
| """ | |
| Erstellt oder lädt einen persistenten Kontext (inkl. Cookies/LocalStorage). | |
| Gut für Workflows, die Logins erfordern. | |
| """ | |
| if session_id in self._contexts: | |
| return self._contexts[session_id] | |
| if not self._playwright: | |
| self._playwright = await async_playwright().start() | |
| options = get_stealth_context_options(connection_mode) | |
| user_data_dir = settings.session_dir / session_id | |
| launch_args = get_stealth_launch_args() | |
| context = await self._playwright.chromium.launch_persistent_context( | |
| user_data_dir=str(user_data_dir), | |
| headless=True, | |
| args=launch_args, | |
| ignore_default_args=["--enable-automation"], | |
| **options | |
| ) | |
| # Anti-Bot Stealth Init Script injizieren | |
| await context.add_init_script(STEALTH_INIT_SCRIPT) | |
| await context.route("**/*", block_trackers_and_media) | |
| self._contexts[session_id] = context | |
| return context | |
| async def close_all(self): | |
| for context in self._contexts.values(): | |
| await context.close() | |
| self._contexts.clear() | |
| if self._browser: | |
| await self._browser.close() | |
| self._browser = None | |
| if self._playwright: | |
| await self._playwright.stop() | |
| self._playwright = None | |
| logger.info("BrowserManager closed.") | |
| # Singleton-Instanz exportieren | |
| browser_manager = BrowserManager() | |