Spaces:
Sleeping
Sleeping
File size: 2,601 Bytes
67aba37 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 | import asyncio
import logging
from typing import Dict, List, Optional
from playwright.async_api import Browser, BrowserContext, Page, async_playwright
logger = logging.getLogger(__name__)
class BrowserManager:
"""
Manages Playwright browser instances and contexts.
Abstraction layer to support future "Remote Cloud PC" connections.
"""
_instance = None
def __init__(self, headless: bool = True):
self.headless = headless
self.playwright = None
self.browser: Optional[Browser] = None
self.active_contexts: List[BrowserContext] = []
@classmethod
def get_instance(cls, headless: bool = True):
if cls._instance is None:
cls._instance = BrowserManager(headless)
return cls._instance
async def start(self):
"""Initialize Playwright and Browser."""
if self.browser:
return
import os
cdp_url = os.getenv("BROWSER_WS_ENDPOINT")
logger.info("Starting Browser Engine...")
self.playwright = await async_playwright().start()
if cdp_url:
logger.info(f"Connecting to remote browser at {cdp_url}...")
self.browser = await self.playwright.chromium.connect_over_cdp(cdp_url)
logger.info("Connected to remote browser.")
else:
# User Agent & Viewport to mimic Desktop
self.browser = await self.playwright.chromium.launch(
headless=self.headless,
args=['--no-sandbox', '--disable-setuid-sandbox', '--disable-gpu', '--disable-dev-shm-usage', '--single-process']
)
logger.info("Browser Engine started (Local).")
async def new_context(self) -> BrowserContext:
"""Create a new isolated browser context."""
if not self.browser:
await self.start()
context = await self.browser.new_context(
viewport={'width': 1280, 'height': 720},
user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
accept_downloads=True
)
self.active_contexts.append(context)
return context
async def close(self):
"""Shutdown all contexts and browser."""
for ctx in self.active_contexts:
await ctx.close()
if self.browser:
await self.browser.close()
if self.playwright:
await self.playwright.stop()
logger.info("Browser Engine stopped.")
|