Split fetch from parse in scrap.py and serp.py (item 2 from review)

Both scrap_patent_async and the four Playwright scrapers previously fused
navigation/fetching with DOM parsing in one function, so there was no way
to unit-test "given this page, what does the current selector contract
extract" without also driving a real network request or browser
navigation through the exact same call.

- scrap.py: extracted parse_patent_html(html, patent_url) as a plain
synchronous function; scrap_patent_async is now just fetch + delegate.
- serp.py: extracted _extract_google_scholar_results,
_extract_google_patents_results, _extract_brave_results and
_extract_bing_results, each taking an already-loaded Page. The public
query_* functions are now just navigate + delegate. Also hoisted the
identical `_block_resources` closure (copy-pasted four times) to one
module-level _block_stylesheet_and_image_resources, and PATENT_ID_REGEX
out of query_google_patents's body.

No behavior change - all 67 existing characterization tests pass
unmodified, which is the point: this refactor happened under a green
safety net rather than being verified by hand. Added a few new tests
demonstrating the payoff: parse_patent_html and _extract_bing_results can
now be tested directly (a saved HTML string, or page.set_content()) with
no respx mocking or goto-patching wrapper needed.

scrap.py

@@ -38,7 +38,18 @@ async def scrap_patent_async(client: AsyncClient, patent_url: str) -> PatentScra
     response = await client.get(patent_url, headers=headers)
     response.raise_for_status()
 
-    soup = BeautifulSoup(response.text, "html.parser")
+    return parse_patent_html(response.text, patent_url)
+
+
+def parse_patent_html(html: str, patent_url: str) -> PatentScrapResult:
+    """Parse a Google Patents patent page into a PatentScrapResult.
+
+    Pure function of the page's HTML - no network I/O - so it can be tested
+    directly against saved/fixture HTML without mocking a client.
+    `patent_url` is only used to identify the page in the error message
+    below.
+    """
+    soup = BeautifulSoup(html, "html.parser")
 
     # Abstract
     abstract_div = soup.find("div", {"class": "abstract"})

serp.py

@@ -82,187 +82,193 @@ async def playwright_open_page(browser: Optional[Browser]):
             await context.close()
 
 
-async def query_google_scholar(browser: Browser, q: str, n_results: int = 10):
-    """Queries google scholar for the specified query and number of results. Returns relevant papers"""
+async def _block_stylesheet_and_image_resources(route, request):
+    """Shared `page.route("**/*", ...)` handler for every scraper below: skip
+    fetching stylesheets/images, since only the page's text content matters
+    here and this meaningfully speeds up navigation.
+    """
+    if request.resource_type in ["stylesheet", "image"]:
+        await route.abort()
+    else:
+        await route.continue_()
 
-    async with playwright_open_page(browser) as page:
 
-        async def _block_resources(route, request):
-            if request.resource_type in ["stylesheet", "image"]:
-                await route.abort()
-            else:
-                await route.continue_()
+async def _extract_google_scholar_results(page: Page, n_results: int) -> list[dict]:
+    """Extract results from an already-loaded Google Scholar results page."""
+    await page.wait_for_selector("div.gs_ri")
+
+    items = await page.locator("div.gs_ri").all()
+    results = []
+    for item in items[:n_results]:
+        title = await item.locator("h3").inner_text(timeout=1000)
+        body = await item.locator("div.gs_rs").inner_text(timeout=1000)
+        href = await item.locator("h3 > a").get_attribute("href")
+
+        results.append({
+            "title": title,
+            "body": body,
+            "href": href
+        })
 
-        await page.route("**/*", _block_resources)
+    return results
+
+
+async def query_google_scholar(browser: Browser, q: str, n_results: int = 10):
+    """Queries google scholar for the specified query and number of results. Returns relevant papers"""
+    async with playwright_open_page(browser) as page:
+        await page.route("**/*", _block_stylesheet_and_image_resources)
 
         url = f"https://scholar.google.com/scholar?q={quote_plus(q)}&num={n_results}"
         await page.goto(url)
 
-        await page.wait_for_selector("div.gs_ri")
+        return await _extract_google_scholar_results(page, n_results)
 
-        items = await page.locator("div.gs_ri").all()
-        results = []
-        for item in items[:n_results]:
-            title = await item.locator("h3").inner_text(timeout=1000)
-            body = await item.locator("div.gs_rs").inner_text(timeout=1000)
-            href = await item.locator("h3 > a").get_attribute("href")
 
-            results.append({
-                "title": title,
-                "body": body,
-                "href": href
-            })
+# regex to locate a patent id, e.g. "US11930446B2" or "EP4760514A1"
+PATENT_ID_REGEX = r"\b[A-Z]{2}\d{6,}(?:[A-Z]\d?)?\b"
 
-    return results
 
+async def _extract_google_patents_results(page: Page, n_results: int) -> list[dict]:
+    """Extract results from an already-loaded Google Patents search page."""
+    # Wait for at least one search result item to appear
+    # This ensures the page has loaded enough to start scraping
+    await page.wait_for_function(
+        f"""() => document.querySelectorAll('search-result-item').length >= 1""",
+        timeout=30_000
+    )
 
-async def query_google_patents(browser: Browser, q: str, n_results: int = 10):
-    """Queries google patents for the specified query and number of results. Returns relevant patents"""
+    items = await page.locator("search-result-item").all()
+    results = []
+    for item in items:
+        text = " ".join(await item.locator("span").all_inner_texts())
+        match = re.search(PATENT_ID_REGEX, text)
+        if not match:
+            continue
 
-    # regex to locate a patent id
-    PATENT_ID_REGEX = r"\b[A-Z]{2}\d{6,}(?:[A-Z]\d?)?\b"
+        patent_id = match.group()
 
-    async with playwright_open_page(browser) as page:
+        try:
+            title = await item.locator("h3, h4").first.inner_text(timeout=1000)
+            body = await item.locator("div.abstract, div.result-snippet, .snippet, .result-text").first.inner_text(timeout=1000)
+        except Exception:
+            continue  # If we can't get title or body, skip this item
+
+        results.append({
+            "id": patent_id,
+            "href": f"https://patents.google.com/patent/{patent_id}/en",
+            "title": title,
+            "body": body
+        })
+
+    return results[:n_results]
 
-        async def _block_resources(route, request):
-            if request.resource_type in ["stylesheet", "image"]:
-                await route.abort()
-            else:
-                await route.continue_()
 
-        await page.route("**/*", _block_resources)
+async def query_google_patents(browser: Browser, q: str, n_results: int = 10):
+    """Queries google patents for the specified query and number of results. Returns relevant patents"""
+    async with playwright_open_page(browser) as page:
+        await page.route("**/*", _block_stylesheet_and_image_resources)
 
         url = f"https://patents.google.com/?q={quote_plus(q)}&num={n_results}"
         await page.goto(url)
 
-        # Wait for at least one search result item to appear
-        # This ensures the page has loaded enough to start scraping
-        await page.wait_for_function(
-            f"""() => document.querySelectorAll('search-result-item').length >= 1""",
-            timeout=30_000
-        )
-
-        items = await page.locator("search-result-item").all()
-        results = []
-        for item in items:
-            text = " ".join(await item.locator("span").all_inner_texts())
-            match = re.search(PATENT_ID_REGEX, text)
-            if not match:
-                continue
-
-            patent_id = match.group()
-
-            try:
-                title = await item.locator("h3, h4").first.inner_text(timeout=1000)
-                body = await item.locator("div.abstract, div.result-snippet, .snippet, .result-text").first.inner_text(timeout=1000)
-            except Exception:
-                continue  # If we can't get title or body, skip this item
+        return await _extract_google_patents_results(page, n_results)
 
-            results.append({
-                "id": patent_id,
-                "href": f"https://patents.google.com/patent/{patent_id}/en",
-                "title": title,
-                "body": body
-            })
 
-    return results[:n_results]
+async def _extract_brave_results(page: Page, n_results: int) -> list[dict]:
+    """Extract results from an already-loaded Brave Search results page.
 
+    Raises BraveSearchBlockedException if the page looks like Brave's
+    anti-bot interstitial rather than a results page.
+    """
+    results_cards = await page.locator('.snippet').all()
 
-async def query_brave_search(browser: Browser, q: str, n_results: int = 10):
-    """Queries Brave Search for the specified query."""
+    if len(results_cards) == 0:
+        page_content = await page.content()
 
-    async with playwright_open_page(browser) as page:
+        if "suspicious" in page_content:
+            raise BraveSearchBlockedException()
 
-        async def _block_resources(route, request):
-            if request.resource_type in ["stylesheet", "image"]:
-                await route.abort()
-            else:
-                await route.continue_()
+    results = []
 
-        await page.route("**/*", _block_resources)
+    for result in results_cards:
+        title = await result.locator('.title').all_inner_texts()
+        description = await result.locator('.snippet-description').all_inner_texts()
+        url = await result.locator('a').nth(0).get_attribute('href')
 
-        url = f"https://search.brave.com/search?q={quote_plus(q)}"
-        await page.goto(url)
+        # Filter out results with no URL or brave-specific URLs
+        if url is None or url.startswith('/'):
+            continue
 
-        results_cards = await page.locator('.snippet').all()
+        results.append({
+            "title": title[0] if title else "",
+            "body": description[0] if description else "",
+            "href": url
+        })
 
-        if len(results_cards) == 0:
-            page_content = await page.content()
+        if len(results) >= n_results:
+            break
 
-            if "suspicious" in page_content:
-                raise BraveSearchBlockedException()
+    return results
 
-        results = []
 
-        for result in results_cards:
-            title = await result.locator('.title').all_inner_texts()
-            description = await result.locator('.snippet-description').all_inner_texts()
-            url = await result.locator('a').nth(0).get_attribute('href')
+async def query_brave_search(browser: Browser, q: str, n_results: int = 10):
+    """Queries Brave Search for the specified query."""
+    async with playwright_open_page(browser) as page:
+        await page.route("**/*", _block_stylesheet_and_image_resources)
+
+        url = f"https://search.brave.com/search?q={quote_plus(q)}"
+        await page.goto(url)
+
+        return await _extract_brave_results(page, n_results)
 
-            # Filter out results with no URL or brave-specific URLs
-            if url is None or url.startswith('/'):
-                continue
 
+async def _extract_bing_results(page: Page, n_results: int) -> list[dict]:
+    """Extract results from an already-loaded Bing results page."""
+    await page.wait_for_selector("li.b_algo")
+
+    results = []
+
+    items = await page.query_selector_all("li.b_algo")
+    for item in items[:n_results]:
+        title_el = await item.query_selector("h2 > a")
+        url = await title_el.get_attribute("href") if title_el else None
+        title = await title_el.inner_text() if title_el else ""
+
+        snippet = ""
+
+        # Try several fallback selectors
+        for selector in [
+            "div.b_caption p",  # typical snippet
+            "div.b_caption",    # sometimes snippet is here
+            "div.b_snippet",    # used in some result types
+            "div.b_text",       # used in some panels
+            "p"                 # fallback to any paragraph
+        ]:
+            snippet_el = await item.query_selector(selector)
+            if snippet_el:
+                snippet = await snippet_el.inner_text()
+                if snippet.strip():
+                    break
+
+        if title and url:
             results.append({
-                "title": title[0] if title else "",
-                "body": description[0] if description else "",
-                "href": url
+                "title": title.strip(),
+                "href": url.strip(),
+                "body": snippet.strip()
             })
 
-            if len(results) >= n_results:
-                break
-
-        return results
+    return results
 
 
 async def query_bing_search(browser: Browser, q: str, n_results: int = 10):
     """Queries bing search for the specified query"""
     async with playwright_open_page(browser) as page:
-        async def _block_resources(route, request):
-            if request.resource_type in ["stylesheet", "image"]:
-                await route.abort()
-            else:
-                await route.continue_()
-
-        await page.route("**/*", _block_resources)
+        await page.route("**/*", _block_stylesheet_and_image_resources)
 
         url = f"https://www.bing.com/search?q={quote_plus(q)}"
         await page.goto(url)
 
-        await page.wait_for_selector("li.b_algo")
-
-        results = []
-
-        items = await page.query_selector_all("li.b_algo")
-        for item in items[:n_results]:
-            title_el = await item.query_selector("h2 > a")
-            url = await title_el.get_attribute("href") if title_el else None
-            title = await title_el.inner_text() if title_el else ""
-
-            snippet = ""
-
-            # Try several fallback selectors
-            for selector in [
-                "div.b_caption p",  # typical snippet
-                "div.b_caption",    # sometimes snippet is here
-                "div.b_snippet",    # used in some result types
-                "div.b_text",       # used in some panels
-                "p"                 # fallback to any paragraph
-            ]:
-                snippet_el = await item.query_selector(selector)
-                if snippet_el:
-                    snippet = await snippet_el.inner_text()
-                    if snippet.strip():
-                        break
-
-            if title and url:
-                results.append({
-                    "title": title.strip(),
-                    "href": url.strip(),
-                    "body": snippet.strip()
-                })
-
-        return results
+        return await _extract_bing_results(page, n_results)
 
 
 async def query_ddg_search(q: str, n_results: int = 10):

tests/test_scrap.py

@@ -3,12 +3,35 @@ import pytest
 import respx
 
 from helpers import load_fixture
-from scrap import scrap_patent_async, scrap_patent_bulk_async
+from scrap import parse_patent_html, scrap_patent_async, scrap_patent_bulk_async
 
 FULL_PATENT_HTML = load_fixture("patents", "full_patent.html")
 MISSING_TITLE_HTML = load_fixture("patents", "missing_title.html")
 
 
+def test_parse_patent_html_extracts_all_fields_with_no_network_involved():
+    """`scrap_patent_async` is now a thin fetch that delegates to this - a
+    plain, synchronous function of the HTML string, so the parsing logic
+    (including the regex-based section splitting, which is the part most
+    likely to break when Google Patents changes its markup) can be tested
+    directly against a saved page with no client/respx/event loop needed.
+    """
+    result = parse_patent_html(FULL_PATENT_HTML, "https://patents.google.com/patent/US11930446B2/en")
+
+    assert result.title == "Widget with improved gadget mechanism"
+    codes = {c.code: c.description for c in result.classifications}
+    assert codes == {
+        "G06F17/30": "Database structures therefor",
+        "G06F17/50": "Other database related",
+        "H04L9/00": "Cryptographic mechanisms",
+    }
+
+
+def test_parse_patent_html_raises_when_page_has_no_title():
+    with pytest.raises(ValueError):
+        parse_patent_html(MISSING_TITLE_HTML, "https://patents.google.com/patent/BOGUS/en")
+
+
 async def test_scrap_patent_async_extracts_all_fields():
     url = "https://patents.google.com/patent/US11930446B2/en"
     with respx.mock:

tests/test_serp_playwright.py

@@ -13,6 +13,7 @@ from helpers import load_fixture, make_fixture_browser
 from serp import (
     BraveSearchBlockedException,
     BrowserUnavailableError,
+    _extract_bing_results,
     playwright_open_page,
     query_bing_search,
     query_brave_search,
@@ -21,6 +22,26 @@ from serp import (
 )
 
 
+async def test_extract_bing_results_works_on_an_already_loaded_page(browser):
+    """Demonstrates the payoff of splitting fetch (navigate) from parse
+    (extract): `_extract_bing_results` takes a Page that's already loaded,
+    so this needs nothing beyond `set_content` - no goto-patching wrapper,
+    no browser-as-a-parameter indirection.
+    """
+    context = await browser.new_context()
+    page = await context.new_page()
+    await page.set_content(load_fixture("bing_results.html"))
+
+    results = await _extract_bing_results(page, 10)
+
+    assert results == [
+        {"title": "Result A", "href": "https://example.org/a", "body": "Snippet A"},
+        {"title": "Result B", "href": "https://example.org/b", "body": "Snippet B"},
+    ]
+
+    await context.close()
+
+
 async def test_bing_search_parses_results(browser):
     fixture_browser = make_fixture_browser(browser, load_fixture("bing_results.html"))