"""Browser discovery, launch, and page capture using Playwright.""" from __future__ import annotations import logging import shutil import time from datetime import datetime from pathlib import Path from typing import Optional from playwright.sync_api import ( BrowserContext, Error as PlaywrightError, Page, Playwright, sync_playwright, ) from monitor.models import AppConfig, PageSnapshot, WebsiteConfig from monitor.snapshot import capture_page_snapshot from storage.snapshots import SnapshotStore logger = logging.getLogger(__name__) FATAL_BROWSER_ERRORS = ("TargetClosedError", "BrowserClosedError", "TargetCrashedError") def _is_fatal_browser_error(exc: PlaywrightError) -> bool: """Return True if the error indicates the browser or page is no longer usable.""" return type(exc).__name__ in FATAL_BROWSER_ERRORS # Common browser executable locations on Linux BROWSER_CANDIDATES: dict[str, list[str]] = { "chrome": [ "/usr/bin/google-chrome", "/usr/bin/google-chrome-stable", "/snap/bin/chromium", # some systems ], "chromium": [ "/usr/bin/chromium", "/usr/bin/chromium-browser", "/snap/bin/chromium", ], "edge": [ "/usr/bin/microsoft-edge", "/usr/bin/microsoft-edge-stable", ], "firefox": [ "/usr/bin/firefox", "/snap/bin/firefox", ], } # Playwright channel names for installed browsers PLAYWRIGHT_CHANNELS: dict[str, str] = { "chrome": "chrome", "edge": "msedge", "firefox": "firefox", } def discover_browser_executable(browser_type: str) -> Optional[str]: """Find an installed browser executable for the given type.""" candidates = BROWSER_CANDIDATES.get(browser_type.lower(), []) for candidate in candidates: path = Path(candidate) if path.is_file() and shutil.which(str(path)): return str(path.resolve()) return None def resolve_browser_launch_options(config: AppConfig) -> dict: """Build Playwright launch options from configuration.""" browser_type = config.browser.type.lower() options: dict = { "headless": config.browser.headless, "args": ["--disable-blink-features=AutomationControlled"], } if config.browser.executable_path: options["executable_path"] = config.browser.executable_path logger.info("Using configured browser executable: %s", config.browser.executable_path) else: discovered = discover_browser_executable(browser_type) if discovered: options["executable_path"] = discovered logger.info("Browser detected: %s at %s", browser_type.title(), discovered) elif browser_type in PLAYWRIGHT_CHANNELS: options["channel"] = PLAYWRIGHT_CHANNELS[browser_type] logger.info("Using Playwright channel: %s", options["channel"]) else: logger.warning( "Could not auto-detect %s; Playwright will use bundled Chromium", browser_type, ) return options class BrowserManager: """Manages a persistent browser context with recovery support.""" def __init__(self, config: AppConfig, snapshot_store: SnapshotStore) -> None: self.config = config self.snapshot_store = snapshot_store self._playwright: Optional[Playwright] = None self._context: Optional[BrowserContext] = None self._page: Optional[Page] = None @property def is_running(self) -> bool: return self._context is not None def start(self) -> None: """Launch the browser with a persistent profile.""" self._playwright = sync_playwright().start() profile_dir = Path(self.config.browser.profile_directory) profile_dir.mkdir(parents=True, exist_ok=True) launch_options = resolve_browser_launch_options(self.config) browser_type = self.config.browser.type.lower() try: if browser_type == "firefox": self._context = self._playwright.firefox.launch_persistent_context( user_data_dir=str(profile_dir.resolve()), headless=self.config.browser.headless, executable_path=launch_options.get("executable_path"), ) else: chromium_options = { "user_data_dir": str(profile_dir.resolve()), "headless": self.config.browser.headless, "args": launch_options.get("args", []), } if "executable_path" in launch_options: chromium_options["executable_path"] = launch_options["executable_path"] if "channel" in launch_options: chromium_options["channel"] = launch_options["channel"] self._context = self._playwright.chromium.launch_persistent_context( **chromium_options ) self._page = self._context.pages[0] if self._context.pages else self._context.new_page() logger.info("Browser started with profile: %s", profile_dir) except PlaywrightError as exc: logger.error("Failed to start browser: %s", exc) self.stop() raise def stop(self) -> None: """Close browser and Playwright resources.""" if self._context: try: self._context.close() except PlaywrightError as exc: logger.warning("Error closing browser context: %s", exc) self._context = None self._page = None if self._playwright: try: self._playwright.stop() except PlaywrightError as exc: logger.warning("Error stopping Playwright: %s", exc) self._playwright = None logger.info("Browser stopped") def restart(self) -> None: """Restart the browser after a crash or failure.""" logger.warning("Restarting browser...") self.stop() self.start() def ensure_running(self) -> None: """Ensure browser is running, restarting if needed.""" if not self.is_running: self.start() def visit_and_snapshot(self, website: WebsiteConfig) -> PageSnapshot: """Navigate to a website and capture a page snapshot.""" self.ensure_running() assert self._page is not None start_time = time.monotonic() timestamp = time.time() try: timeout_ms = int(self.config.monitor.page_load_timeout_seconds * 1000) render_wait_ms = int(self.config.monitor.render_wait_seconds * 1000) logger.info("Opening %s (%s)", website.name, website.url) self._page.goto( website.url, wait_until="domcontentloaded", timeout=timeout_ms, ) try: self._page.wait_for_load_state("networkidle", timeout=timeout_ms) except PlaywrightError: logger.debug( "networkidle timeout for %s; continuing after render wait", website.id, ) self._page.wait_for_timeout(render_wait_ms) load_duration = time.monotonic() - start_time snapshot = capture_page_snapshot( page=self._page, website=website, load_duration=load_duration, snapshot_store=self.snapshot_store, timestamp=timestamp, ) logger.info("Snapshot captured: %s", website.name) return snapshot except PlaywrightError as exc: load_duration = time.monotonic() - start_time error_type = type(exc).__name__ if _is_fatal_browser_error(exc): logger.error( "Browser crash detected on %s [%s]: %s", website.id, error_type, exc, ) self._context = None self._page = None raise logger.error( "Website %s navigation failed [%s]: %s", website.id, error_type, exc, ) return PageSnapshot( website_id=website.id, website_name=website.name, requested_url=website.url, final_url=self._page.url if self._page else website.url, timestamp=datetime.fromtimestamp(timestamp), title="", visible_text="", html="", screenshot_path=None, load_duration=load_duration, success=False, error=f"{error_type}: {exc}", ) def open_for_login(self, urls: list[str]) -> None: """ Open the monitor profile for manual login. Keeps the browser open until the user closes it or presses Ctrl+C. Always runs with a visible window regardless of headless config. """ original_headless = self.config.browser.headless self.config.browser.headless = False try: self.start() assert self._context is not None if urls: for index, url in enumerate(urls): page = self._page if index == 0 else self._context.new_page() logger.info("Opening tab %d: %s", index + 1, url) page.goto(url, wait_until="domcontentloaded", timeout=60000) else: logger.info("No URLs provided — browser opened with default tab") profile_dir = Path(self.config.browser.profile_directory).resolve() logger.info("Profile directory: %s", profile_dir) logger.info( "Log in to your websites in the browser window. " "Close the browser or press Ctrl+C when finished." ) while self._context and self._context.pages: time.sleep(0.5) logger.info("Browser window closed") except KeyboardInterrupt: logger.info("Login session ended by user") finally: self.config.browser.headless = original_headless self.stop() def test_browser(self, test_url: str = "https://example.com") -> bool: """Open browser and visit a test URL to verify configuration.""" try: self.start() assert self._page is not None logger.info("Test browser: navigating to %s", test_url) self._page.goto(test_url, wait_until="domcontentloaded", timeout=30000) title = self._page.title() logger.info("Test browser success. Page title: %s", title) return True except PlaywrightError as exc: logger.error("Test browser failed: %s", exc) return False finally: self.stop()