Spaces:
Sleeping
Sleeping
File size: 2,578 Bytes
ebab0e2 | 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 76 | from typing import Callable, Optional
from playwright.async_api import Browser, async_playwright
from playwright.async_api import TimeoutError as PlaywrightTimeoutError
from playwright_stealth import stealth_async
from app.errors import ScrapeError
class BrowserManager:
"""Owns the single, long-lived Playwright browser process for the app's
lifetime, started/stopped via FastAPI's lifespan hook. Launching a fresh
browser per request is prohibitively slow and memory-hungry; requests
instead get an isolated `new_context()` from this shared browser."""
def __init__(self):
self._playwright = None
self._browser: Optional[Browser] = None
async def start(self) -> None:
self._playwright = await async_playwright().start()
self._browser = await self._playwright.chromium.launch(
headless=True,
args=["--disable-blink-features=AutomationControlled"],
)
async def stop(self) -> None:
if self._browser is not None:
await self._browser.close()
if self._playwright is not None:
await self._playwright.stop()
def get_browser(self) -> Optional[Browser]:
return self._browser
class PlaywrightScraper:
def __init__(
self,
browser_provider: Callable[[], Optional[Browser]],
*,
user_agent: str,
timeout_ms: int,
stealth: Callable = stealth_async,
):
self._browser_provider = browser_provider
self._user_agent = user_agent
self._timeout_ms = timeout_ms
self._stealth = stealth
async def scrape(self, url: str) -> str:
browser = self._browser_provider()
if browser is None:
raise ScrapeError(503, "Scraper is not ready yet, try again shortly.")
# A fresh context per request keeps cookies/storage isolated between
# callers while reusing the shared browser process.
context = await browser.new_context(
user_agent=self._user_agent,
viewport={"width": 1920, "height": 1080},
locale="en-US",
timezone_id="America/New_York",
)
try:
page = await context.new_page()
await self._stealth(page)
try:
await page.goto(url, wait_until="domcontentloaded", timeout=self._timeout_ms)
except PlaywrightTimeoutError:
raise ScrapeError(504, "Timed out loading the page.")
return await page.content()
finally:
await context.close()
|