Update app.py

app.py

@@ -1,270 +1,82 @@
-# app.py
+# app_playwright.py
+"""
+Fast concurrent search fetcher using Playwright + FastAPI.
+
+Requirements:
+    pip install fastapi uvicorn playwright beautifulsoup4 pydantic
+    python -m playwright install chromium
+
+Why Playwright:
+  - Single browser process, many lightweight contexts
+  - Fast request interception (block ads/fonts)
+  - Async API for high concurrency
+
+Endpoints:
+  - GET /health
+  - GET /search?query=...
+  - GET /fetch?url=...&fast=true
+  - POST /search  (body: { "query": "..."} or { "url": "..." })
+  - GET /search_pages?query=...&pages=5&concurrency=8&fast=true&ordered=false  (streams NDJSON)
+  - GET /search_pages_aggregate?query=... (returns aggregated JSON)
+Tuning via environment variables:
+  - PLAYWRIGHT_HEADLESS (true/false, default true)
+  - POOL_CONCURRENCY (default 8)  -> concurrent page navigations
+  - PARSER_WORKERS (default 4)    -> threadpool for BeautifulSoup parsing
+  - REQUEST_TIMEOUT_MS (default 20000) -> navigation timeout per page
+"""
+
 from __future__ import annotations
 import os
-import time
-import logging
-import threading
 import asyncio
-from typing import Optional, Dict, Any, Tuple
-from concurrent.futures import ThreadPoolExecutor
+import logging
+import json
+import time
+import math
+from typing import Optional, Dict, Any, List, Tuple
 from urllib.parse import quote_plus, urljoin
 
-from fastapi import FastAPI, HTTPException, Query, Body
+from fastapi import FastAPI, Query, Body, HTTPException
+from fastapi.responses import JSONResponse, StreamingResponse
 from pydantic import BaseModel
-from starlette.responses import JSONResponse
 
-from selenium import webdriver
-from selenium.webdriver.chrome.options import Options
-from selenium.webdriver.chrome.service import Service
-from selenium.common.exceptions import WebDriverException, SessionNotCreatedException
-from webdriver_manager.chrome import ChromeDriverManager
 from bs4 import BeautifulSoup
-from selenium.webdriver.common.by import By
-from selenium.webdriver.support.ui import WebDriverWait
-from selenium.webdriver.support import expected_conditions as EC
-from selenium.common.exceptions import TimeoutException
-
-# virtual display
-from pyvirtualdisplay import Display
-
-# Logging
-logging.basicConfig(level=logging.INFO)
-logger = logging.getLogger("fast_fetcher")
-
-# ---------------- BrowserManager ----------------
-class BrowserManager:
-    def __init__(
-        self,
-        headless: bool = True,
-        user_agent: Optional[str] = None,
-        window_size: str = "1366,768",
-        disable_images: bool = True,
-        block_resource_urls: Optional[list[str]] = None,
-    ):
-        self.headless = headless
-        self.user_agent = user_agent
-        self.window_size = window_size
-        self.disable_images = disable_images
-        self.block_resource_urls = block_resource_urls or [
-            "*.doubleclick.net/*",
-            "*.google-analytics.com/*",
-            "*.googlesyndication.com/*",
-            "*.adservice.google.com/*",
-        ]
-        self._driver_lock = threading.Lock()
-        self._driver: Optional[webdriver.Chrome] = None
-        self._display: Optional[Display] = None
-        self._start_driver_with_retries()
-
-    def _build_options(self) -> Options:
-        opts = Options()
-        # If CHROME_BIN is present, point to it
-        chrome_bin = os.environ.get("CHROME_BIN", "/usr/bin/google-chrome-stable")
-        if os.path.exists(chrome_bin):
-            opts.binary_location = chrome_bin
-            logger.debug("Using chrome binary: %s", chrome_bin)
-        else:
-            logger.warning("Chrome binary not found at %s (will rely on system/browser manager).", chrome_bin)
-
-        if self.headless:
-            opts.add_argument("--headless=new")
-            opts.add_argument("--headless")
-
-        # container-friendly flags (and stable fallback)
-        opts.add_argument("--no-sandbox")
-        opts.add_argument("--disable-setuid-sandbox")
-        opts.add_argument("--disable-dev-shm-usage")
-        opts.add_argument("--disable-gpu")
-        opts.add_argument("--disable-extensions")
-        opts.add_argument("--disable-blink-features=AutomationControlled")
-        opts.add_argument("--disable-software-rasterizer")
-        opts.add_argument(f"--window-size={self.window_size}")
-        opts.add_argument("--remote-debugging-port=0")
-
-        if self.user_agent:
-            opts.add_argument(f"--user-agent={self.user_agent}")
-
-        if self.disable_images:
-            prefs = {
-                "profile.managed_default_content_settings.images": 2,
-                "profile.managed_default_content_settings.stylesheets": 2,
-                "profile.managed_default_content_settings.fonts": 2,
-            }
-            opts.add_experimental_option("prefs", prefs)
-
-        opts.add_experimental_option("excludeSwitches", ["enable-logging"])
-        opts.add_experimental_option("useAutomationExtension", False)
-        return opts
-
-    def _start_driver_with_retries(self, attempts: int = 3, delay_seconds: float = 1.0):
-        last_exc = None
-        for attempt in range(1, attempts + 1):
-            try:
-                logger.info("Starting Chrome driver (attempt %d/%d)...", attempt, attempts)
-                self._start_driver()
-                logger.info("Chrome driver started successfully.")
-                return
-            except Exception as exc:
-                logger.exception("Failed to start driver on attempt %d: %s", attempt, exc)
-                last_exc = exc
-                time.sleep(delay_seconds)
-        raise RuntimeError(f"Unable to start Chrome driver after {attempts} attempts: {last_exc}") from last_exc
-
-    def _start_xvfb_if_needed(self):
-        # If headless=False AND no DISPLAY, start Xvfb via pyvirtualdisplay
-        if not self.headless and os.environ.get("DISPLAY", "") == "":
-            try:
-                logger.info("No DISPLAY found and headless=False — starting virtual X display (Xvfb).")
-                self._display = Display(visible=0, size=(int(self.window_size.split(",")[0]), int(self.window_size.split(",")[1])))
-                self._display.start()
-                logger.info("Virtual X display started (DISPLAY=%s).", os.environ.get("DISPLAY"))
-            except Exception as e:
-                logger.exception("Failed to start virtual display: %s", e)
-                raise
-
-    def _stop_xvfb_if_started(self):
-        if self._display:
-            try:
-                self._display.stop()
-                logger.info("Virtual X display stopped.")
-            except Exception:
-                pass
-            self._display = None
-
-    def _start_driver(self):
-        # start virtual display if required BEFORE launching Chrome
-        self._start_xvfb_if_needed()
-
-        opts = self._build_options()
-
-        # 1) Try Selenium Manager (webdriver.Chrome(options=opts)). Selenium >=4.14 may use driver manager itself.
-        primary_exc = None
-        fallback_exc = None
-        try:
-            logger.debug("Attempting to start Chrome via Selenium Manager (webdriver.Chrome(options=opts))")
-            self._driver = webdriver.Chrome(options=opts)
-            # quick smoke test: ensure browser is responsive (may throw)
-            try:
-                self._driver.execute_script("return navigator.userAgent")
-            except Exception as e:
-                # browser started but died quickly
-                raise RuntimeError("Browser started by Selenium Manager but crashed immediately.") from e
-
-            self._post_start_setup()
-            return
-        except Exception as e_primary:
-            primary_exc = e_primary
-            logger.warning("Selenium Manager attempt failed: %s", e_primary)
-
-        # 2) Fallback: use webdriver-manager to download driver and start with the explicit Service
-        try:
-            driver_path = ChromeDriverManager().install()
-            logger.info("webdriver-manager installed chromedriver: %s", driver_path)
-            try:
-                os.chmod(driver_path, 0o755)
-            except Exception:
-                logger.debug("chmod on chromedriver failed or unnecessary.")
-
-            service = Service(driver_path)
-            self._driver = webdriver.Chrome(service=service, options=opts)
-            self._post_start_setup()
-            return
-        except Exception as e_fallback:
-            fallback_exc = e_fallback
-            logger.exception("webdriver-manager fallback failed: %s", e_fallback)
-
-        # 3) Final fallback: attempt system /usr/bin/chromedriver if available
-        try:
-            sys_path = "/usr/bin/chromedriver"
-            if os.path.exists(sys_path):
-                logger.info("Trying system chromedriver at %s", sys_path)
-                try:
-                    os.chmod(sys_path, 0o755)
-                except Exception:
-                    pass
-                service = Service(sys_path)
-                self._driver = webdriver.Chrome(service=service, options=opts)
-                self._post_start_setup()
-                return
-        except Exception as e_sys:
-            logger.exception("System chromedriver attempt failed: %s", e_sys)
-
-        # If all failed, stop virtual display (if started) and raise a helpful error
-        self._stop_xvfb_if_started()
-        # Include both primary and fallback messages in the raised exception
-        raise RuntimeError(f"Failed to start Chrome driver. primary_error={primary_exc}, fallback_error={fallback_exc}")
-
-    def _post_start_setup(self):
-        try:
-            self._driver.set_page_load_timeout(60)
-            # best-effort CDP network blocking
-            try:
-                self._driver.execute_cdp_cmd("Network.enable", {})
-                if self.block_resource_urls:
-                    self._driver.execute_cdp_cmd("Network.setBlockedURLs", {"urls": self.block_resource_urls})
-            except Exception:
-                pass
-        except Exception:
-            pass
-
-    def fetch_html(
-        self,
-        url: str,
-        wait_seconds: Optional[float] = 10.0,
-        wait_for_selector: Optional[str] = None,
-    ) -> str:
-        if self._driver is None:
-            self._start_driver_with_retries()
-
-        with self._driver_lock:
-            driver = self._driver
-            try:
-                driver.get(url)
-
-                if wait_for_selector and wait_seconds:
-                    try:
-                        WebDriverWait(driver, wait_seconds).until(
-                            EC.presence_of_element_located((By.CSS_SELECTOR, wait_for_selector))
-                        )
-                    except TimeoutException:
-                        pass
-                else:
-                    if wait_seconds:
-                        try:
-                            WebDriverWait(driver, min(wait_seconds, 3)).until(
-                                lambda d: d.execute_script("return document.readyState") == "complete"
-                            )
-                        except Exception:
-                            time.sleep(0.5)
-
-                return driver.page_source
-            except WebDriverException as e:
-                logger.exception("WebDriver exception during fetch: %s", e)
-                # restart driver and raise
-                try:
-                    self._safe_quit_driver()
-                except Exception:
-                    pass
-                self._start_driver_with_retries()
-                raise RuntimeError(f"WebDriver error during fetch: {e}")
-
-    def _safe_quit_driver(self):
-        if self._driver:
-            try:
-                self._driver.quit()
-            except Exception:
-                pass
-        self._driver = None
-        # stop display if we started one
-        self._stop_xvfb_if_started()
-
-    def close(self):
-        self._safe_quit_driver()
-
-
-# ---------------- EXTRACT_DATA (same as your earlier implementation) ----------------
-def EXTRACT_DATA(html: str) -> Dict[str, Any]:
-    soup = BeautifulSoup(html, "html.parser")
+import concurrent.futures
+
+# Playwright async
+from playwright.async_api import async_playwright, Playwright, Browser, BrowserContext, Page, Request as PWRequest, Error as PWError
+
+# ---------------------
+# Config / Logging
+# ---------------------
+LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO").upper()
+logging.basicConfig(level=LOG_LEVEL, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
+logger = logging.getLogger("fast_playwright_fetcher")
+
+PLAYWRIGHT_HEADLESS = os.environ.get("PLAYWRIGHT_HEADLESS", "true").lower() in ("1", "true", "yes")
+POOL_CONCURRENCY = int(os.environ.get("POOL_CONCURRENCY", "8"))          # parallel navigations
+PARSER_WORKERS = int(os.environ.get("PARSER_WORKERS", "4"))             # threadpool size for BS parsing
+REQUEST_TIMEOUT_MS = int(os.environ.get("REQUEST_TIMEOUT_MS", "20000")) # navigation timeout in ms
+PAGE_LOAD_WAIT = os.environ.get("PAGE_LOAD_WAIT", "domcontentloaded")   # "domcontentloaded" or "load"
+
+# set of resource URL substrings to block (ads, analytics, fonts, etc.)
+BLOCK_PATTERNS = [
+    "doubleclick.net", "google-analytics.com", "googlesyndication.com",
+    "adservice.google.com", "googletagmanager.com", "facebook.com",
+    "fonts.googleapis.com", "gstatic.com", "analytics.twitter.com",
+    ".woff", ".woff2", ".ttf", ".otf", "font.gstatic.com",
+]
+
+# ---------------------
+# Parsing helper (threadpool)
+# ---------------------
+PARSE_POOL = concurrent.futures.ThreadPoolExecutor(max_workers=PARSER_WORKERS)
+
+def extract_data_sync(html: str) -> Dict[str, Any]:
+    """
+    Synchronous BeautifulSoup parsing (will be run inside PARSE_POOL).
+    Re-uses your previous EXTRACT_DATA logic (simplified/robust).
+    """
+    soup = BeautifulSoup(html or "", "html.parser")
     BASE_URL = "https://www.google.com"
 
     def safe_text(el):
@@ -274,311 +86,379 @@ def EXTRACT_DATA(html: str) -> Dict[str, Any]:
         return el.get(attr) if el and el.has_attr(attr) else ""
 
     def abs_url(url):
-        return urljoin(BASE_URL, url) if url else ""
+        try:
+            return urljoin(BASE_URL, url) if url else ""
+        except Exception:
+            return url or ""
 
     def clean_thumb(src):
         if src and not src.startswith("data:"):
             return abs_url(src)
         return None
 
-    def is_ad_element(element):
-        for parent in element.parents:
-            if parent.get("id") in ["tads", "tadsb"] or "ads-ad" in parent.get("class", []):
-                return True
-        return False
-
+    # web results
     web_results = []
     for result in soup.select(".tF2Cxc"):
-        if is_ad_element(result):
-            continue
         title_tag = result.select_one("h3")
         link_tag = result.select_one("a")
         cite_tag = result.select_one("cite")
         snippet_tag = result.select_one(".VwiC3b")
-        read_more_tag = result.select_one(".vzmbzf")
-
         if title_tag and link_tag:
-            entry = {
-                "no": len(web_results) + 1,
+            web_results.append({
                 "title": safe_text(title_tag),
                 "link": abs_url(safe_attr(link_tag, "href")),
                 "displayed_url": safe_text(cite_tag),
                 "snippet": safe_text(snippet_tag)
-            }
-            extra = []
-            if read_more_tag:
-                read_more_url = abs_url(safe_attr(read_more_tag, "href"))
-                if read_more_url:
-                    extra.append({"read_more": read_more_url})
-            if extra:
-                entry["extra"] = extra
-            web_results.append(entry)
+            })
 
+    # images
     image_results = []
     for img_item in soup.select(".eA0Zlc"):
         img_tag = img_item.select_one("img")
         link_tag = img_item.select_one("a")
-        source_tag = img_item.select_one(".s0fJje span")
         src = safe_attr(img_tag, "data-src") or safe_attr(img_tag, "src")
         thumb = clean_thumb(src)
         if thumb:
-            image_results.append({
-                "thumbnail": thumb,
-                "alt": safe_attr(img_tag, "alt"),
-                "source": safe_text(source_tag),
-                "link": abs_url(safe_attr(link_tag, "href"))
-            })
-
-    video_results = []
-    for video in soup.select(".KYaZsb"):
-        title_tag = video.select_one(".tNxQIb.ynAwRc")
-        link_tag = video.select_one("a.rIRoqf")
-        thumb_img = video.select_one(".AZJdrc img")
-        duration_tag = video.select_one(".c8rnLc")
-        channel_tag = video.select_one(".Sg4azc span:first-child")
-        date_tag = video.select_one(".rbYSKb span")
-        desc_tag = video.select_one(".wNifxf .p4wth")
-        thumb_src = safe_attr(thumb_img, "data-src") or safe_attr(thumb_img, "src")
-        thumb = clean_thumb(thumb_src)
-        if title_tag and link_tag:
-            video_results.append({
-                "title": safe_text(title_tag),
-                "link": abs_url(safe_attr(link_tag, "href")),
-                "thumbnail": thumb,
-                "duration": safe_text(duration_tag),
-                "channel": safe_text(channel_tag),
-                "date": safe_text(date_tag),
-                "description_snippet": safe_text(desc_tag)
-            })
-
-    news_results = []
-    for news in soup.select(".m7jPZ"):
-        title_tag = news.select_one(".n0jPhd")
-        link_tag = news.select_one("a")
-        source_tag = news.select_one(".MgUUmf span")
-        time_tag = news.select_one(".rbYSKb span")
-        thumb_img = news.select_one(".uhHOwf img")
-        thumb_src = safe_attr(thumb_img, "data-src") or safe_attr(thumb_img, "src")
-        thumb = clean_thumb(thumb_src)
-        if title_tag and link_tag:
-            news_results.append({
-                "title": safe_text(title_tag),
-                "link": abs_url(safe_attr(link_tag, "href")),
-                "source": safe_text(source_tag),
-                "time": safe_text(time_tag),
-                "thumbnail": thumb
-            })
-
-    knowledge_panel = {}
-    rhs = soup.find(id="rhs")
-    if rhs:
-        title_tag = rhs.select_one(".PZPZlf.ssJ7i")
-        subtitle_tag = rhs.select_one(".iAIpCb span")
-        if title_tag:
-            knowledge_panel["title"] = safe_text(title_tag)
-        if subtitle_tag:
-            knowledge_panel["subtitle"] = safe_text(subtitle_tag)
-
-        desc_tag = rhs.select_one(".kno-rdesc span")
-        if desc_tag:
-            knowledge_panel["description"] = safe_text(desc_tag)
-
-        facts = {}
-        for fact in rhs.select(".zloOqf"):
-            label_tag = fact.select_one(".w8qArf")
-            value_tag = fact.select_one(".LrzXr")
-            if label_tag and value_tag:
-                label = safe_text(label_tag).replace(":", "").strip()
-                links = value_tag.find_all("a")
-                if links and len(links) > 1:
-                    names = [safe_text(a) for a in links if safe_text(a)]
-                    if names:
-                        facts[label] = names
-                else:
-                    text = safe_text(value_tag)
-                    if text:
-                        facts[label] = text
-        if facts:
-            knowledge_panel["facts"] = facts
-
-        profiles = []
-        for profile in rhs.select(".dRrfkf a"):
-            name_tag = profile.select_one(".CtCigf")
-            link = safe_attr(profile, "href")
-            if name_tag and link:
-                profiles.append({
-                    "platform": safe_text(name_tag),
-                    "link": abs_url(link)
-                })
-        if profiles:
-            knowledge_panel["profiles"] = profiles
-
-    if not knowledge_panel:
-        knowledge_panel = None
-
-    ai_overview = None
-    ai_container = soup.select_one(".p2M1Qe .f5cPye")
-    if ai_container:
-        text = safe_text(ai_container)
-        if text:
-            ai_overview = text
-
-    thumbnails = set()
-    for img in soup.select("img[data-src], img[src]"):
-        src = safe_attr(img, "data-src") or safe_attr(img, "src")
-        clean = clean_thumb(src)
-        if clean:
-            thumbnails.add(clean)
-
-    all_thumbnails = sorted(thumbnails) if thumbnails else None
+            image_results.append({"thumbnail": thumb, "alt": safe_attr(img_tag, "alt"), "link": abs_url(safe_attr(link_tag, "href"))})
 
+    # videos/news/knowledge panels omitted for brevity but can be added similarly
     data = {}
     if web_results:
         data["web_results"] = web_results
     if image_results:
         data["image_results"] = image_results
-    if video_results:
-        data["video_results"] = video_results
-    if news_results:
-        data["news_results"] = news_results
-    if knowledge_panel:
-        data["knowledge_panel"] = knowledge_panel
-    if ai_overview:
-        data["ai_overview"] = ai_overview
-    if all_thumbnails:
-        data["all_thumbnail_urls"] = all_thumbnails
 
-    return data
+    # collect thumbnails
+    thumbnails = set()
+    for img in soup.select("img[data-src], img[src]"):
+        src = safe_attr(img, "data-src") or safe_attr(img, "src")
+        thumb = clean_thumb(src)
+        if thumb:
+            thumbnails.add(thumb)
+    if thumbnails:
+        data["all_thumbnail_urls"] = sorted(thumbnails)
 
+    return data
 
-# ---------------- BrowserPool and API ----------------
-class BrowserPool:
-    def __init__(self, pool_size: int = 1, headless: bool = True):
-        self.pool_size = max(1, pool_size)
-        self.managers = [BrowserManager(headless=headless) for _ in range(self.pool_size)]
-        self._rr_index = 0
-        self._rr_lock = threading.Lock()
+async def extract_data(html: str) -> Dict[str, Any]:
+    loop = asyncio.get_running_loop()
+    return await loop.run_in_executor(PARSE_POOL, extract_data_sync, html)
 
-    def pick_manager(self) -> BrowserManager:
-        with self._rr_lock:
-            idx = self._rr_index
-            self._rr_index = (self._rr_index + 1) % self.pool_size
-        return self.managers[idx]
+# ---------------------
+# FastAPI + Playwright globals
+# ---------------------
+app = FastAPI(title="fast_playwright_fetcher", version="0.1")
+PLAY: Optional[Playwright] = None
+BROWSER: Optional[Browser] = None
 
-    def close_all(self):
-        for m in self.managers:
-            try:
-                m.close()
-            except Exception:
-                pass
+# Semaphore to limit concurrency across endpoints
+CONCURRENCY_SEMAPHORE = asyncio.Semaphore(POOL_CONCURRENCY)
 
+# Simple TTL cache (in-memory)
 class SimpleTTLCache:
     def __init__(self, ttl_seconds: int = 20):
         self.ttl = ttl_seconds
-        self._cache: Dict[str, Tuple[float, Any]] = {}
-        self._lock = threading.Lock()
+        self.store: Dict[str, Tuple[float, Any]] = {}
+        self._lock = asyncio.Lock()
 
-    def get(self, key: str):
-        with self._lock:
-            item = self._cache.get(key)
+    async def get(self, key: str):
+        async with self._lock:
+            item = self.store.get(key)
             if not item:
                 return None
             ts, value = item
             if time.time() - ts > self.ttl:
-                del self._cache[key]
+                del self.store[key]
                 return None
             return value
 
-    def set(self, key: str, value: Any):
-        with self._lock:
-            self._cache[key] = (time.time(), value)
-
-class SearchRequest(BaseModel):
-    query: Optional[str] = None
-    url: Optional[str] = None
-    wait_for_selector: Optional[str] = None
-    headless: Optional[bool] = True
+    async def set(self, key: str, value: Any):
+        async with self._lock:
+            self.store[key] = (time.time(), value)
+
+CACHE = SimpleTTLCache(ttl_seconds=int(os.environ.get("CACHE_TTL", "25")))
+
+# ---------------------
+# Playwright helpers
+# ---------------------
+async def start_playwright():
+    global PLAY, BROWSER
+    if PLAY is not None:
+        return
+    PLAY = await async_playwright().start()
+    BROWSER = await PLAY.chromium.launch(headless=PLAYWRIGHT_HEADLESS,
+                                         args=["--no-sandbox", "--disable-setuid-sandbox", "--disable-dev-shm-usage"])
+    logger.info("Playwright started (headless=%s)", PLAYWRIGHT_HEADLESS)
+
+async def stop_playwright():
+    global PLAY, BROWSER
+    if BROWSER:
+        try:
+            await BROWSER.close()
+        except Exception:
+            pass
+        BROWSER = None
+    if PLAY:
+        try:
+            await PLAY.stop()
+        except Exception:
+            pass
+        PLAY = None
+    logger.info("Playwright stopped.")
 
-app = FastAPI(title="fast_fetcher_api", version="0.1")
-POOL: Optional[BrowserPool] = None
-EXECUTOR: Optional[ThreadPoolExecutor] = None
-CACHE = SimpleTTLCache(ttl_seconds=25)
+def _should_block_request(url: str) -> bool:
+    if not url:
+        return False
+    u = url.lower()
+    for pat in BLOCK_PATTERNS:
+        if pat in u:
+            return True
+    return False
+
+async def _fetch_with_context(url: str, wait_until: str = "domcontentloaded", timeout_ms: int = REQUEST_TIMEOUT_MS, user_agent: Optional[str] = None) -> str:
+    """
+    Create an isolated browser context, block unwanted requests, navigate, get HTML.
+    This is efficient: creating small contexts is cheap compared to full browser processes.
+    """
+    if BROWSER is None:
+        raise RuntimeError("Browser not started")
+
+    # create context with minimal resources
+    context: BrowserContext = await BROWSER.new_context(user_agent=user_agent or "Mozilla/5.0 (Playwright)", viewport={"width": 1366, "height": 768})
+    page: Page = await context.new_page()
+
+    # intercept requests to block common resources
+    async def route_handler(route, request: PWRequest):
+        try:
+            if _should_block_request(request.url):
+                await route.abort()
+            else:
+                await route.continue_()
+        except Exception:
+            # if anything fails, let the request through
+            try:
+                await route.continue_()
+            except Exception:
+                pass
 
+    try:
+        await page.route("**/*", route_handler)
+    except Exception:
+        # route might fail if the browser doesn't support it; ignore
+        pass
+
+    # navigate with retries/backoff
+    max_attempts = 3
+    backoff_base = 0.2
+    last_exc = None
+    for attempt in range(1, max_attempts + 1):
+        try:
+            # Playwright's goto uses wait_until values like "load" or "domcontentloaded"
+            await page.goto(url, wait_until=wait_until, timeout=timeout_ms)
+            content = await page.content()
+            return content
+        except PWError as e:
+            last_exc = e
+            logger.warning("Playwright navigation error (attempt %d/%d) for %s : %s", attempt, max_attempts, url, str(e))
+            # small backoff
+            await asyncio.sleep(backoff_base * attempt)
+            # if fatal, break and raise after attempts
+        except Exception as e:
+            last_exc = e
+            logger.exception("Unexpected navigation error (attempt %d/%d) for %s", attempt, max_attempts, url)
+            await asyncio.sleep(backoff_base * attempt)
+
+    # cleanup
+    try:
+        await page.close()
+    except Exception:
+        pass
+    try:
+        await context.close()
+    except Exception:
+        pass
+    raise RuntimeError(f"Failed to fetch {url} after {max_attempts} attempts: {last_exc}")
+
+# ---------------------
+# App startup/shutdown
+# ---------------------
 @app.on_event("startup")
-async def startup_event():
-    global POOL, EXECUTOR
-    # Switch headless here to False as you asked. The BrowserManager will start an Xvfb display automatically.
-    POOL = BrowserPool(pool_size=1, headless=False)
-    EXECUTOR = ThreadPoolExecutor(max_workers=2)
-    app.state.executor = EXECUTOR
-    app.state.pool = POOL
-    logger.info("Startup: browser pool created (size=%d).", 1)
+async def on_startup():
+    await start_playwright()
+    # pre-warm a small context to reduce first request latency
+    try:
+        async with CONCURRENCY_SEMAPHORE:
+            _ = await _fetch_with_context("about:blank")
+    except Exception:
+        pass
+    logger.info("App startup complete. concurrency=%d parser_workers=%d", POOL_CONCURRENCY, PARSER_WORKERS)
 
 @app.on_event("shutdown")
-async def shutdown_event():
-    global POOL, EXECUTOR
-    if POOL:
-        POOL.close_all()
-    if EXECUTOR:
-        EXECUTOR.shutdown(wait=True)
-    logger.info("Shutdown: browsers closed and executor stopped.")
-
-def _blocking_fetch_and_extract(manager: BrowserManager, url: str, wait_for_selector: Optional[str], wait_seconds: Optional[float]):
-    start = time.time()
-    html = manager.fetch_html(url, wait_seconds=wait_seconds, wait_for_selector=wait_for_selector)
-    extracted = EXTRACT_DATA(html)
-    duration = time.time() - start
-    return {"url": url, "duration": duration, "data": extracted}
+async def on_shutdown():
+    await stop_playwright()
+    PARSE_POOL.shutdown(wait=False)
+
+# ---------------------
+# Blocking helper wrapper
+# ---------------------
+async def fetch_and_extract(url: str, wait_until: str = PAGE_LOAD_WAIT, timeout_ms: int = REQUEST_TIMEOUT_MS, user_agent: Optional[str] = None) -> Dict[str, Any]:
+    """
+    Acquire concurrency semaphore, fetch page with Playwright, parse with BeautifulSoup in threadpool.
+    """
+    async with CONCURRENCY_SEMAPHORE:
+        html = await _fetch_with_context(url, wait_until=wait_until, timeout_ms=timeout_ms, user_agent=user_agent)
+        data = await extract_data(html)
+        return {"url": url, "data": data}
+
+# ---------------------
+# Request models
+# ---------------------
+class SearchRequest(BaseModel):
+    query: Optional[str] = None
+    url: Optional[str] = None
 
+# ---------------------
+# Endpoints
+# ---------------------
 @app.get("/health")
 async def health():
     return {"status": "ok"}
 
+@app.get("/fetch")
+async def fetch(url: str = Query(..., min_length=5), fast: Optional[bool] = Query(True)):
+    """
+    Fetch a URL and extract data. Use fast=true to wait only for domcontentloaded (faster).
+    """
+    wait_until = "domcontentloaded" if fast else "load"
+    try:
+        result = await fetch_and_extract(url, wait_until=wait_until)
+        return JSONResponse(content=result)
+    except Exception as e:
+        logger.exception("Fetch error for %s", url)
+        raise HTTPException(status_code=500, detail=str(e))
+
 @app.get("/search")
-async def search(query: str = Query(..., min_length=1), wait_for_selector: Optional[str] = None):
+async def search(query: str = Query(..., min_length=1), fast: Optional[bool] = Query(True)):
     q = query.strip()
     if not q:
-        raise HTTPException(status_code=400, detail="query parameter required")
+        raise HTTPException(status_code=400, detail="query required")
 
     url = f"https://www.google.com/search?q={quote_plus(q)}"
-    cache_key = f"search:{q}:{wait_for_selector}"
-
-    cached = CACHE.get(cache_key)
+    cache_key = f"search:{q}:{fast}"
+    cached = await CACHE.get(cache_key)
     if cached:
         return JSONResponse(content={"cached": True, **cached})
 
-    manager = app.state.pool.pick_manager()
-    loop = asyncio.get_event_loop()
-    fut = loop.run_in_executor(app.state.executor, _blocking_fetch_and_extract, manager, url, wait_for_selector, 5.0)
-    result = await fut
-    CACHE.set(cache_key, result)
-    return JSONResponse(content={"cached": False, **result})
-
-@app.get("/fetch")
-async def fetch(url: str = Query(..., min_length=5), wait_for_selector: Optional[str] = None):
-    manager = app.state.pool.pick_manager()
-    loop = asyncio.get_event_loop()
-    fut = loop.run_in_executor(app.state.executor, _blocking_fetch_and_extract, manager, url, wait_for_selector, 6.0)
-    result = await fut
-    return JSONResponse(content=result)
+    try:
+        res = await fetch_and_extract(url, wait_until=("domcontentloaded" if fast else "load"))
+        await CACHE.set(cache_key, res)
+        return JSONResponse(content={"cached": False, **res})
+    except Exception as e:
+        logger.exception("Search error for %s", q)
+        raise HTTPException(status_code=500, detail=str(e))
 
 @app.post("/search")
 async def post_search(body: SearchRequest = Body(...)):
     if not (body.query or body.url):
         raise HTTPException(status_code=400, detail="Either query or url must be provided")
-    if body.url:
-        target = body.url
-    else:
-        target = f"https://www.google.com/search?q={quote_plus(body.query)}"
-
-    cache_key = f"search_post:{target}:{body.wait_for_selector}"
-    cached = CACHE.get(cache_key)
-    if cached:
-        return JSONResponse(content={"cached": True, **cached})
-
-    manager = app.state.pool.pick_manager()
-    loop = asyncio.get_event_loop()
-    fut = loop.run_in_executor(app.state.executor, _blocking_fetch_and_extract, manager, target, body.wait_for_selector, 6.0)
-    result = await fut
-    CACHE.set(cache_key, result)
-    return JSONResponse(content={"cached": False, **result})
+    target = body.url if body.url else f"https://www.google.com/search?q={quote_plus(body.query)}"
+    try:
+        res = await fetch_and_extract(target, wait_until=PAGE_LOAD_WAIT)
+        return JSONResponse(content=res)
+    except Exception as e:
+        logger.exception("Post search error for %s", target)
+        raise HTTPException(status_code=500, detail=str(e))
+
+# ---------------------
+# Streaming / multi-page endpoints
+# ---------------------
+@app.get("/search_pages")
+async def search_pages(
+    query: str = Query(..., min_length=1),
+    pages: int = Query(3, ge=1, le=50),
+    concurrency: Optional[int] = Query(None, ge=1),
+    ordered: Optional[bool] = Query(False),
+    fast: Optional[bool] = Query(True),
+):
+    """
+    Streams NDJSON results per Google page. pages=N means start=0,10,20...
+    Use concurrency to override default concurrency (POOL_CONCURRENCY).
+    """
+    q = query.strip()
+    if not q:
+        raise HTTPException(status_code=400, detail="query required")
+
+    pages = min(max(1, pages), 50)
+    items: List[Tuple[int, str]] = []
+    for i in range(pages):
+        start = i * 10
+        items.append((i + 1, f"https://www.google.com/search?q={quote_plus(q)}&start={start}"))
+
+    # determine concurrency (local, can't change global semaphore size, so we create local tasks but still rely on global)
+    local_concurrency = concurrency or POOL_CONCURRENCY
+    # create tasks
+    loop = asyncio.get_running_loop()
+    task_list = []
+    for page_no, url in items:
+        task = loop.create_task(_search_page_task(page_no, url, fast))
+        task_list.append(task)
+
+    async def streamer():
+        try:
+            if ordered:
+                # yield in page order
+                for t in task_list:
+                    res = await t
+                    yield (json.dumps(res, ensure_ascii=False) + "\n").encode("utf-8")
+            else:
+                # yield fastest-first
+                for fut in asyncio.as_completed(task_list):
+                    res = await fut
+                    yield (json.dumps(res, ensure_ascii=False) + "\n").encode("utf-8")
+        finally:
+            for t in task_list:
+                if not t.done():
+                    t.cancel()
+
+    return StreamingResponse(streamer(), media_type="application/x-ndjson")
+
+async def _search_page_task(page_no: int, url: str, fast: bool):
+    try:
+        wait_until = "domcontentloaded" if fast else "load"
+        result = await fetch_and_extract(url, wait_until=wait_until)
+        return {"page": page_no, "url": url, "ok": True, "duration": None, "data": result.get("data")}
+    except Exception as e:
+        logger.exception("Error fetching page %d (%s): %s", page_no, url, e)
+        return {"page": page_no, "url": url, "ok": False, "error": str(e)}
+
+@app.get("/search_pages_aggregate")
+async def search_pages_aggregate(
+    query: str = Query(..., min_length=1),
+    pages: int = Query(3, ge=1, le=50),
+    concurrency: Optional[int] = Query(None, ge=1),
+    fast: Optional[bool] = Query(True),
+):
+    q = query.strip()
+    if not q:
+        raise HTTPException(status_code=400, detail="query required")
+
+    pages = min(max(1, pages), 50)
+    items = []
+    for i in range(pages):
+        start = i * 10
+        items.append((i + 1, f"https://www.google.com/search?q={quote_plus(q)}&start={start}"))
+
+    tasks = [asyncio.create_task(_search_page_task(pno, url, fast)) for pno, url in items]
+    results = await asyncio.gather(*tasks, return_exceptions=False)
+    return JSONResponse(content={"pages": results})
+
+# ---------------------
+# Run notes:
+# ---------------------
+# start server:
+#   uvicorn app_playwright:app --host 0.0.0.0 --port 8000 --workers 1
+#
+# Important: use a single uvicorn worker (playwright and the asyncio loop should be in one process).
+# You can scale horizontally by running multiple instances behind a load balancer if needed.
+#
+# ---------------------