88 lines
2.3 KiB
Python
88 lines
2.3 KiB
Python
|
"""Playwright browser management."""
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
from typing import Optional
|
||
|
|
|
||
|
|
from playwright.async_api import Browser, Playwright, async_playwright
|
||
|
|
|
||
|
|
from app.config import settings
|
||
|
|
|
||
|
|
|
||
|
|
class PlaywrightClient:
|
||
|
|
"""
|
||
|
|
Manages the Playwright browser instance.
|
||
|
|
|
||
|
|
Provides lifecycle management for a single Chromium browser
|
||
|
|
that serves all sessions.
|
||
|
|
"""
|
||
|
|
|
||
|
|
def __init__(self):
|
||
|
|
self._playwright: Optional[Playwright] = None
|
||
|
|
self._browser: Optional[Browser] = None
|
||
|
|
|
||
|
|
@property
|
||
|
|
def browser(self) -> Browser:
|
||
|
|
"""Get the browser instance."""
|
||
|
|
if self._browser is None:
|
||
|
|
raise RuntimeError("Browser not started. Call start() first.")
|
||
|
|
return self._browser
|
||
|
|
|
||
|
|
@property
|
||
|
|
def is_running(self) -> bool:
|
||
|
|
"""Check if browser is running."""
|
||
|
|
return self._browser is not None
|
||
|
|
|
||
|
|
async def start(self) -> Browser:
|
||
|
|
"""
|
||
|
|
Start the Playwright browser.
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
The Browser instance
|
||
|
|
"""
|
||
|
|
if self._browser is not None:
|
||
|
|
return self._browser
|
||
|
|
|
||
|
|
self._playwright = await async_playwright().start()
|
||
|
|
|
||
|
|
# Launch Chromium with Docker-compatible settings
|
||
|
|
self._browser = await self._playwright.chromium.launch(
|
||
|
|
headless=settings.browser_headless,
|
||
|
|
args=[
|
||
|
|
"--no-sandbox",
|
||
|
|
"--disable-setuid-sandbox",
|
||
|
|
"--disable-dev-shm-usage",
|
||
|
|
"--disable-gpu",
|
||
|
|
"--single-process",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
|
||
|
|
return self._browser
|
||
|
|
|
||
|
|
async def stop(self) -> None:
|
||
|
|
"""Stop the Playwright browser and cleanup."""
|
||
|
|
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 = None
|
||
|
|
|
||
|
|
|
||
|
|
# Global singleton instance
|
||
|
|
_playwright_client: Optional[PlaywrightClient] = None
|
||
|
|
|
||
|
|
|
||
|
|
def get_playwright_client() -> PlaywrightClient:
|
||
|
|
"""Get the global PlaywrightClient instance."""
|
||
|
|
global _playwright_client
|
||
|
|
if _playwright_client is None:
|
||
|
|
_playwright_client = PlaywrightClient()
|
||
|
|
return _playwright_client
|