ops.py

"""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)