"""EPO Open Patent Services (OPS) v3.2 backend. Legitimate REST alternative to scraping Espacenet (which sits behind an interactive Cloudflare CAPTCHA). Provides keyword search and by-number retrieval of the same bibliographic / full-text data. Auth is OAuth2 client-credentials: the consumer key/secret are exchanged for a short-lived (~20 min) bearer token at the token endpoint. Credentials are read from the OPS_CONSUMER_KEY / OPS_CONSUMER_SECRET environment variables. """ import asyncio import base64 import logging import os import re import time from typing import Optional from dotenv import load_dotenv from httpx import AsyncClient, HTTPStatusError from pydantic import BaseModel from scrap import ClassificationCode, PatentScrapResult load_dotenv() OPS_BASE = "https://ops.epo.org/3.2/rest-services" OPS_TOKEN_URL = "https://ops.epo.org/3.2/auth/accesstoken" class OPSNotConfigured(Exception): """Raised when OPS credentials are not set in the environment.""" class OPSError(Exception): """Raised when OPS returns an error that isn't a plain 'not found'.""" class _OPSTokenManager: """Caches and refreshes the OPS OAuth2 bearer token.""" def __init__(self) -> None: self._key = os.environ.get("OPS_CONSUMER_KEY") self._secret = os.environ.get("OPS_CONSUMER_SECRET") self._token: Optional[str] = None self._expiry: float = 0.0 self._lock = asyncio.Lock() @property def configured(self) -> bool: return bool(self._key and self._secret) async def get_token(self, client: AsyncClient, force: bool = False) -> str: if not self.configured: raise OPSNotConfigured( "OPS_CONSUMER_KEY / OPS_CONSUMER_SECRET are not set.") async with self._lock: # 30s safety margin before expiry. if not force and self._token and time.monotonic() < self._expiry - 30: return self._token basic = base64.b64encode( f"{self._key}:{self._secret}".encode()).decode() resp = await client.post( OPS_TOKEN_URL, headers={"Authorization": f"Basic {basic}"}, data={"grant_type": "client_credentials"}, ) resp.raise_for_status() data = resp.json() self._token = data["access_token"] self._expiry = time.monotonic() + int(data.get("expires_in", 1199)) logging.info("Obtained new OPS access token.") return self._token token_manager = _OPSTokenManager() _KIND_SUFFIX_RE = re.compile(r"[A-Z]\d{0,2}$") def _normalize_epodoc(number: str) -> str: """Strip whitespace and a trailing kind code for the epodoc format. OPS epodoc lookups reject the kind suffix (e.g. 'US11930446B2' -> 404), so 'US11930446B2' becomes 'US11930446' and 'EP4760514A1' becomes 'EP4760514'. """ number = number.strip().replace(" ", "") return _KIND_SUFFIX_RE.sub("", number) async def _ops_get(client: AsyncClient, path: str, params: Optional[dict] = None) -> dict: """GET an OPS endpoint as JSON, refreshing the token once on auth failure.""" token = await token_manager.get_token(client) url = f"{OPS_BASE}{path}" headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"} resp = await client.get(url, params=params, headers=headers) # A stale/invalid token surfaces as 400/401/403 mentioning the token; refresh once. if resp.status_code in (400, 401, 403) and "token" in resp.text.lower(): token = await token_manager.get_token(client, force=True) headers["Authorization"] = f"Bearer {token}" resp = await client.get(url, params=params, headers=headers) resp.raise_for_status() return resp.json() # --------------------------- JSON walking helpers --------------------------- # OPS returns XML-to-JSON: text nodes are {"$": "..."}, attributes are "@name", # and repeated elements become lists (but a single element stays a bare dict). def _as_list(node) -> list: if node is None: return [] return node if isinstance(node, list) else [node] def _text(node) -> Optional[str]: if isinstance(node, dict): return node.get("$") if isinstance(node, str): return node return None def _pick_lang(nodes: list, lang: str = "en"): """From a list of {@lang, ...} nodes, prefer the requested language.""" nodes = _as_list(nodes) for n in nodes: if isinstance(n, dict) and n.get("@lang", "").lower() == lang: return n return nodes[0] if nodes else None def _epodoc_number(exchange_doc: dict) -> Optional[str]: """Best-effort human patent number (e.g. 'EP4760514').""" bd = exchange_doc.get("bibliographic-data", {}) ref = bd.get("publication-reference", {}) for doc_id in _as_list(ref.get("document-id")): if doc_id.get("@document-id-type") == "epodoc": return _text(doc_id.get("doc-number")) country = exchange_doc.get("@country", "") number = exchange_doc.get("@doc-number", "") return f"{country}{number}" if (country or number) else None def _extract_title(bibliographic_data: dict) -> Optional[str]: title = _pick_lang(bibliographic_data.get("invention-title")) return _text(title) if title else None def _extract_abstract(exchange_doc: dict) -> Optional[str]: abstract = _pick_lang(exchange_doc.get("abstract")) if not abstract: return None paras = [_text(p) for p in _as_list(abstract.get("p"))] joined = " ".join(p for p in paras if p) return joined or None def _extract_classifications(bibliographic_data: dict) -> Optional[list[ClassificationCode]]: """Assemble CPC symbols from patent-classifications (no descriptions in OPS).""" codes: list[ClassificationCode] = [] seen: set[str] = set() container = bibliographic_data.get("patent-classifications", {}) for pc in _as_list(container.get("patent-classification")): section = _text(pc.get("section")) or "" klass = _text(pc.get("class")) or "" subclass = _text(pc.get("subclass")) or "" main_group = _text(pc.get("main-group")) or "" subgroup = _text(pc.get("subgroup")) or "" if not (section and klass and subclass and main_group): continue code = f"{section}{klass}{subclass}{main_group}/{subgroup}".rstrip("/") if code not in seen: seen.add(code) codes.append(ClassificationCode(code=code, description="")) return codes or None def _fulltext_paragraphs(data: dict, section: str) -> Optional[str]: """Extract text from a fulltext service response (claims / description).""" wpd = data.get("ops:world-patent-data", {}) ftxt_docs = wpd.get("ftxt:fulltext-documents", {}) doc = ftxt_docs.get("ftxt:fulltext-document") doc = doc[0] if isinstance(doc, list) else doc if not doc: return None node = _pick_lang(doc.get(section)) if isinstance( doc.get(section), list) else doc.get(section) if not node: return None if section == "claims": lines: list[str] = [] for claim in _as_list(node.get("claim")): for ct in _as_list(claim.get("claim-text")): t = _text(ct) if t: lines.append(t) return "\n".join(lines) or None # description paras = [_text(p) for p in _as_list(node.get("p"))] return "\n".join(p for p in paras if p) or None # ------------------------------- Public API -------------------------------- def _to_cql(query: str) -> str: """Turn a free-text query into a CQL expression OPS accepts. OPS only auto-wraps a single bare word; multi-word queries like 'agentic ai' are invalid CQL and 404. Queries already containing a relational operator ('=') are assumed to be hand-written CQL and passed through unchanged. """ q = query.strip() if not q or "=" in q: return q escaped = q.replace('"', " ").strip() return f'txt all "{escaped}"' async def ops_search(client: AsyncClient, query: str, n_results: int) -> list[dict]: """Keyword search against OPS published-data, normalized to SERP dicts.""" n_results = max(1, min(n_results, 100)) # OPS caps a page at 100. try: data = await _ops_get( client, "/published-data/search/biblio", params={"q": _to_cql(query), "Range": f"1-{n_results}"}, ) except HTTPStatusError as e: # OPS returns 404 (SERVER.EntityNotFound) when a search has no hits. if e.response.status_code == 404: return [] raise search = data.get("ops:world-patent-data", {}).get("ops:biblio-search", {}) result = search.get("ops:search-result", {}) results: list[dict] = [] for entry in _as_list(result.get("exchange-documents")): ed = entry.get("exchange-document") if isinstance(entry, dict) else None if not ed: continue number = _epodoc_number(ed) bd = ed.get("bibliographic-data", {}) results.append({ "title": _extract_title(bd) or number or "Untitled", "body": _extract_abstract(ed) or "", "href": f"https://worldwide.espacenet.com/patent/search?q={number}" if number else "", "id": number, }) return results async def ops_biblio(client: AsyncClient, number: str, doc_type: str = "publication", fmt: str = "epodoc") -> dict: if fmt == "epodoc": number = _normalize_epodoc(number) data = await _ops_get( client, f"/published-data/{doc_type}/{fmt}/{number}/biblio") return data["ops:world-patent-data"]["exchange-documents"]["exchange-document"] async def _ops_fulltext(client: AsyncClient, number: str, section: str, doc_type: str = "publication", fmt: str = "epodoc") -> Optional[str]: """Fetch claims/description full text; returns None when unavailable (404).""" if fmt == "epodoc": number = _normalize_epodoc(number) try: data = await _ops_get(client, f"/published-data/{doc_type}/{fmt}/{number}/{section}") except HTTPStatusError as e: if e.response.status_code == 404: return None raise return _fulltext_paragraphs(data, section) async def ops_scrap_patent(client: AsyncClient, number: str, doc_type: str = "publication", fmt: str = "epodoc") -> PatentScrapResult: """Retrieve a patent via OPS and shape it like the Google Patents scraper.""" exchange_doc = await ops_biblio(client, number, doc_type, fmt) bd = exchange_doc.get("bibliographic-data", {}) claims, description = await asyncio.gather( _ops_fulltext(client, number, "claims", doc_type, fmt), _ops_fulltext(client, number, "description", doc_type, fmt), return_exceptions=True, ) claims = None if isinstance(claims, Exception) else claims description = None if isinstance(description, Exception) else description return PatentScrapResult( title=_extract_title(bd) or (_epodoc_number(exchange_doc) or number), abstract=_extract_abstract(exchange_doc), description=description, claims=claims, field_of_invention=None, background=None, classifications=_extract_classifications(bd), ) class OPSBulkResponse(BaseModel): patents: list[PatentScrapResult] failed_ids: list[str] async def ops_scrap_patent_bulk(client: AsyncClient, numbers: list[str]) -> OPSBulkResponse: results = await asyncio.gather( *[ops_scrap_patent(client, n) for n in numbers], return_exceptions=True) patents = [r for r in results if not isinstance(r, Exception)] failed = [numbers[i] for i, r in enumerate( results) if isinstance(r, Exception)] return OPSBulkResponse(patents=patents, failed_ids=failed)