scrap.py

import asyncio
import re
from typing import Optional
from httpx import AsyncClient
from bs4 import BeautifulSoup
from pydantic import BaseModel


class ClassificationCode(BaseModel):
    """A single CPC or IPC classification code with its description."""
    code: str
    description: str


class PatentScrapResult(BaseModel):
    """Schema for the result of scraping a google patents page."""
    # The title of the patent.
    title: str
    # The abstract of the patent, if available.
    abstract: Optional[str] = None
    # The full description of the patent containing the field of the invention, background, summary, etc.
    description: Optional[str] = None
    # The full claims of the patent.
    claims: Optional[str] = None
    # The field of the invention, if available.
    field_of_invention: Optional[str] = None
    # The background of the invention, if available.
    background: Optional[str] = None
    # CPC and IPC classification codes with descriptions.
    classifications: Optional[list[ClassificationCode]] = None


async def scrap_patent_async(client: AsyncClient, patent_url: str) -> PatentScrapResult:
    headers = {
        "User-Agent": "Mozilla/5.0 (compatible; GPTBot/1.0; +https://openai.com/gptbot)"
    }
    response = await client.get(patent_url, headers=headers)
    response.raise_for_status()

    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"})
    abstract = abstract_div.get_text(
        strip=True) if abstract_div else None

    # Description
    description_section = soup.find("section", itemprop="description")
    description = description_section.get_text(
        separator="\n", strip=True) if description_section else None

    # Field of the Invention
    invention_field_match = re.findall(
        r"(FIELD OF THE INVENTION|TECHNICAL FIELD)(.*?)(?:(BACKGROUND|BACKGROUND OF THE INVENTION|SUMMARY|BRIEF SUMMARY|DETAILED DESCRIPTION|DESCRIPTION OF THE RELATED ART))", description, re.IGNORECASE | re.DOTALL) if description_section else None
    invention_field = invention_field_match[0][1].strip(
    ) if invention_field_match else None

    # Background of the Invention
    invention_background_match = re.findall(
        r"(BACKGROUND OF THE INVENTION|BACKGROUND)(.*?)(?:(SUMMARY|BRIEF SUMMARY|DETAILED DESCRIPTION|DESCRIPTION OF THE PREFERRED EMBODIMENTS|DESCRIPTION))", description, re.IGNORECASE | re.DOTALL) if description_section else None
    invention_background = invention_background_match[0][1].strip(
    ) if invention_background_match else None

    # Claims
    claims_section = soup.find("section", itemprop="claims")
    claims = claims_section.get_text(
        separator="\n", strip=True) if claims_section else None

    # Patent Title
    meta_title_tag = soup.find("meta", {"name": "DC.title"})
    if meta_title_tag is None or not meta_title_tag.get("content"):
        # Unexpected page layout (interstitial, non-patent page, markup
        # change) rather than a normal 4xx/5xx - raise so callers (the OPS
        # fallback in app.py, or scrap_patent_bulk_async's gather) treat it
        # as a failed scrape instead of crashing on `None.get(...)`.
        raise ValueError(
            f"Could not find a patent title (meta[name=DC.title]) at {patent_url}; "
            "the page may not be a valid Google Patents patent page.")
    meta_title = meta_title_tag.get("content").strip()

    # Patent publication number
    # pub_num = soup.select_one("h2#pubnum").get_text(strip=True)
    # get the h2 with id ="pubnum" and extract the text

    # Classification codes (CPC + IPC, flat list, deduplicated by code).
    # Google Patents renders each code in a <span> inside a <li>. Leaf entries
    # (no nested <ul>) carry exactly one code; parent entries are breadcrumbs.
    leaf_code_re = re.compile(r'^[A-Z]\d{2}[A-Z]\d+/\d+$')
    classifications = []
    seen_codes: set[str] = set()
    for li in soup.find_all("li"):
        if li.find("ul"):
            continue
        span = li.find("span")
        if not span:
            continue
        code = span.get_text(strip=True)
        if leaf_code_re.match(code) and code not in seen_codes:
            seen_codes.add(code)
            full_text = li.get_text(separator=" ", strip=True)
            desc = full_text.replace(code, "", 1).strip().lstrip("—").lstrip("-").strip()
            classifications.append(ClassificationCode(code=code, description=desc))

    return PatentScrapResult(
        # publication_number=pub_num,
        abstract=abstract,
        description=description,
        claims=claims,
        title=meta_title,
        field_of_invention=invention_field,
        background=invention_background,
        classifications=classifications or None
    )


class PatentScrapBulkResponse(BaseModel):
    """Response model for bulk patent scraping."""
    patents: list[PatentScrapResult]
    failed_ids: list[str]


# Cap how many patent pages are fetched at once. Without this a single
# accepted request opened one outbound scrape per id, all at the same time;
# the Playwright scrapers have had an equivalent limit from the start and
# this brings the HTTP path in line with them.
BULK_SCRAP_CONCURRENCY_LIMIT = 10


async def scrap_patent_bulk_async(client: AsyncClient, patent_ids: list[str]) -> PatentScrapBulkResponse:
    """Scrape multiple patents asynchronously, a bounded number at a time."""
    urls = [
        f"https://patents.google.com/patent/{pid}/en" for pid in patent_ids]
    semaphore = asyncio.Semaphore(BULK_SCRAP_CONCURRENCY_LIMIT)

    async def scrap_one(url: str):
        async with semaphore:
            return await scrap_patent_async(client, url)

    results = await asyncio.gather(*[scrap_one(url) for url in urls], return_exceptions=True)

    filtered_results = [
        res for res in results if not isinstance(res, Exception)]

    failed_ids = [
        patent_ids[i] for i, res in enumerate(results) if isinstance(res, Exception)
    ]

    return PatentScrapBulkResponse(
        patents=filtered_results,
        failed_ids=failed_ids
    )