310 lines
11 KiB
Python
310 lines
11 KiB
Python
"""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()
|