Dump suggestions

Author: pogzyb

src/tourist/app/routers/tour/routes.py

@@ -25,11 +25,12 @@ class TouristSerpRequest(BaseRequest):
     search_engine: Literal["brave", "duckduckgo"]
     max_results: int = 5
     exclude_hosts: list[str] | None = []
+    visit_mode: Literal["direct", "click"] = "click"
 
     @field_validator("max_results")
     @classmethod
     def max_results_reasonable(cls, v: int) -> int:
-        if v > 15 and v < 1:
+        if v > 15 or v < 1:
             raise ValueError("max_results must be >= 1 and <= 15")
         return v
 

src/tourist/service/driver.py

@@ -9,12 +9,9 @@
 from typing import Literal
 from urllib.parse import quote_plus
 
-from patchright.async_api import (
-    TimeoutError as PlaywrightTimeoutError,
-    async_playwright,
-)
+from patchright.async_api import TimeoutError as PlaywrightTimeoutError, async_playwright
 
-from .utils import get_links_from_serp, to_markdown
+from .utils import extract_serp_candidates, to_markdown
 
 logger = logging.getLogger("uvicorn.error")
 
@@ -51,6 +48,7 @@ async def chrome(**kws):
                     "--ignore-certificate-errors",
                     "--remote-debugging-port=9222",
                 ],
+                **kws,
             )
             yield context
             await context.close()
@@ -64,7 +62,7 @@ async def handle_cookie_preferences(page) -> None:
 
     for manager in cookie_managers:
         parent_locator = page
-        locator: "Locator" = None
+        locator = None
 
         actions = manager.get("actions", [])
         for action in actions:
@@ -81,48 +79,92 @@ async def handle_cookie_preferences(page) -> None:
                 for selector in action["value"]:
                     if await parent_locator.locator(selector).first.is_visible():
                         locator = parent_locator.locator(selector)
-                        # manager["selector-list-item"] = selector
                         break
 
         if locator is not None:
             try:
-                # explicit wait for navigation as some pages will reload after accepting cookies
-                # async with page.expect_navigation(wait_until="networkidle", timeout=15000):
                 await locator.first.click(delay=10)
-                logger.info(f"Accepted cookie preferences: {manager['name']}")
+                logger.info("Accepted cookie preferences: %s", manager["name"])
                 return
-
             except (PlaywrightTimeoutError, Exception):
-                logger.exception(
-                    f"Could not handle cookie preferences: {manager['name']}"
-                )
+                logger.exception("Could not handle cookie preferences: %s", manager["name"])
                 continue
 
 
-async def scrape(url: str, ctx: "BrowserContext") -> dict[str, str]:
+async def prepare_page(page) -> None:
+    await page.route(
+        "**/*",
+        lambda route: route.abort()
+        if route.request.resource_type in {"image", "media", "font"}
+        else route.continue_(),
+    )
+
+
+async def human_pause(page, low_ms: int = 250, high_ms: int = 900) -> None:
+    await page.wait_for_timeout(random.randint(low_ms, high_ms))
+
+
+async def settle_page(page) -> None:
+    await handle_cookie_preferences(page)
+    for selector in ["main", "article", "[role='main']", "body"]:
+        try:
+            await page.locator(selector).first.wait_for(state="visible", timeout=2500)
+            break
+        except PlaywrightTimeoutError:
+            continue
+    await human_pause(page, 700, 1500)
+    await page.mouse.move(random.randint(200, 700), random.randint(200, 700))
+    await page.mouse.wheel(0, random.randint(200, 1000))
+    await human_pause(page, 200, 700)
+
+
+async def scrape(url: str, ctx, *, serp_title: str = "", serp_rank: int = 0) -> dict[str, str]:
     page = None
     try:
         page = await ctx.new_page()
-        await page.route(
-            "**/*",
-            lambda route: route.abort()
-            if route.request.resource_type == "image"
-            else route.continue_(),
-        )
+        await prepare_page(page)
         await page.goto(url, wait_until="domcontentloaded", timeout=30000)
-        await page.mouse.move(333, 888)
-        await page.mouse.wheel(0, -111)
-        await handle_cookie_preferences(page)
-        await page.wait_for_timeout(random.randint(800, 1600))
+        await settle_page(page)
         scraped_page = {
             "title": await page.title(),
             "html": await page.content(),
             "current_url": page.url,
             "requested_url": url,
+            "serp_title": serp_title,
+            "serp_rank": str(serp_rank),
         }
         return scraped_page
-    except PlaywrightTimeoutError as e:
-        raise e
+    finally:
+        if page is not None:
+            await page.close()
+
+
+async def click_through_result(serp_url: str, candidate, ctx) -> dict[str, str]:
+    page = None
+    try:
+        page = await ctx.new_page()
+        await prepare_page(page)
+        await page.goto(serp_url, wait_until="domcontentloaded", timeout=30000)
+        await settle_page(page)
+
+        link_locator = page.locator(f"a[href='{candidate.url}']").first
+        if await link_locator.count() == 0:
+            raise PlaywrightTimeoutError(f"Could not find SERP link for {candidate.url}")
+
+        await link_locator.hover()
+        await human_pause(page, 200, 600)
+        async with page.expect_navigation(wait_until="domcontentloaded", timeout=30000):
+            await link_locator.click(delay=random.randint(25, 120))
+        await settle_page(page)
+
+        return {
+            "title": await page.title(),
+            "html": await page.content(),
+            "current_url": page.url,
+            "requested_url": candidate.url,
+            "serp_title": candidate.title,
+            "serp_rank": str(candidate.rank),
+        }
     finally:
         if page is not None:
             await page.close()
@@ -133,6 +175,7 @@ async def get_serp_results(
     search_engine: Literal["brave", "duckduckgo"],
     exclude_hosts: list[str],
     max_results: int = 5,
+    visit_mode: Literal["direct", "click"] = "click",
     **chrome_kws,
 ) -> list[dict[str, str]]:
     serp_results: list[dict[str, str]] = []
@@ -143,23 +186,36 @@ async def get_serp_results(
         base_se = "duckduckgo.com"
         path_se = ""
     else:
-        logger.error(f"invalid search_engine: {search_engine}")
+        logger.error("invalid search_engine: %s", search_engine)
         return []
+
     async with chrome(**chrome_kws) as ctx:
         serp_url = f"https://{base_se}/{path_se}?q={quote_plus(search_query)}"
         serp = await scrape(serp_url, ctx)
-        links = get_links_from_serp(serp["html"], search_engine, exclude_hosts)
-        if not links:
-            logger.warning(f"No links were extracted from {serp_url}")
-        tasks = [scrape(link, ctx) for link in links[:max_results]]
-        for task in asyncio.as_completed(tasks):
+        candidates = extract_serp_candidates(serp["html"], base_se, exclude_hosts)
+        if not candidates:
+            logger.warning("No links were extracted from %s", serp_url)
+            return []
+
+        chosen = candidates[:max_results]
+        if visit_mode == "direct":
+            tasks = [
+                scrape(c.url, ctx, serp_title=c.title, serp_rank=c.rank)
+                for c in chosen
+            ]
+            results_iter = asyncio.as_completed(tasks)
+        else:
+            tasks = [click_through_result(serp_url, c, ctx) for c in chosen]
+            results_iter = asyncio.as_completed(tasks)
+
+        for task in results_iter:
             try:
                 result = await task
                 html = result.pop("html")
                 result["contents"] = to_markdown(html)
                 serp_results.append(result)
-            except:
-                logger.exception("Could not individual extract page:")
+            except Exception:
+                logger.exception("Could not extract individual page")
     return serp_results