"""Espacenet OPS (EPO Open Patent Services) client. https://developers.epo.org/ops-v3-2 - OAuth 2.0 client_credentials — 20dk access_token - Published data search: /rest-services/published-data/search - 500 sorgu/hafta ücretsiz (IP başına) - Response JSON (Accept: application/json) Tasarım: - Global singleton (lru_cache) — credentials .env'den - Token cache + proaktif yenileme (60s buffer) - Response → PriorArtHit listesi (biblio + abstract yalın) """ from __future__ import annotations import time from typing import Any from urllib.parse import quote import httpx from app.services.prior_art.models import PriorArtHit _TOKEN_URL = "https://ops.epo.org/3.2/auth/accesstoken" _BASE_URL = "https://ops.epo.org/3.2/rest-services" class EspacenetError(RuntimeError): """EPO OPS API hatası (HTTP 4xx/5xx, parse hatası).""" class EspacenetAuthError(EspacenetError): """OAuth token alınamadı — credentials yanlış veya EPO rate-limit'te.""" class EspacenetNotConfigured(EspacenetError): """EPO_OPS_KEY / EPO_OPS_SECRET .env'de yok.""" class EspacenetClient: """EPO OPS REST API wrapper — async.""" def __init__( self, consumer_key: str | None, consumer_secret: str | None, *, http_client: httpx.AsyncClient | None = None, ) -> None: self._key = consumer_key self._secret = consumer_secret self._external_client = http_client is not None self._http = http_client or httpx.AsyncClient(timeout=30.0) self._token: str | None = None self._token_expires_at: float = 0.0 def is_configured(self) -> bool: return bool(self._key and self._secret) async def close(self) -> None: if not self._external_client: await self._http.aclose() async def _get_access_token(self) -> str: """OAuth client_credentials — token 60s buffer ile cache'lenir.""" if not self.is_configured(): raise EspacenetNotConfigured("EPO_OPS_KEY / EPO_OPS_SECRET .env'de tanımlı değil.") if self._token and time.time() < self._token_expires_at - 60: return self._token response = await self._http.post( _TOKEN_URL, data={"grant_type": "client_credentials"}, auth=(self._key or "", self._secret or ""), ) if response.status_code != 200: raise EspacenetAuthError( f"EPO token alınamadı (HTTP {response.status_code}): {response.text[:200]}" ) data = response.json() self._token = data["access_token"] # EPO genellikle 1200s (20dk) döndürür self._token_expires_at = time.time() + int(data.get("expires_in", 1200)) return self._token # type: ignore[return-value] async def search( self, query: str, *, limit: int = 10, ) -> list[PriorArtHit]: """Espacenet CQL arama — basit keyword/boolean destekli. Args: query: CQL (ör. 'txt="post-kuantum kriptografi"' veya 'cpc=H04L9/08 AND ta="gateway"'). Basit keyword de kabul eder. limit: Maks. 100. Returns: PriorArtHit listesi (her biri EP/US/WO patent no ile). """ token = await self._get_access_token() url = f"{_BASE_URL}/published-data/search/biblio?q={quote(query)}&Range=1-{min(limit, 100)}" response = await self._http.get( url, headers={"Authorization": f"Bearer {token}", "Accept": "application/json"}, ) if response.status_code == 404: # "no results found" — boş liste dön return [] if response.status_code >= 400: raise EspacenetError( f"EPO search hatası (HTTP {response.status_code}): {response.text[:300]}" ) try: data = response.json() except ValueError as exc: raise EspacenetError(f"EPO JSON parse hatası: {exc}") from exc return list(_parse_biblio_response(data)) # -------------------------------------------------------------------------- # EPO response parser — JSON yapısı oldukça iç içe; burada izole ettik. # -------------------------------------------------------------------------- def _as_list(value: Any) -> list[Any]: """EPO response'unda tek öğe bazen dict, bazen list olur — normalize et.""" if value is None: return [] if isinstance(value, list): return value return [value] def _first_text(value: Any) -> str | None: """EPO '$' field'i ile text'i çıkar (dict or list of dicts).""" if value is None: return None if isinstance(value, str): return value if isinstance(value, dict): return value.get("$") if isinstance(value, list) and value: return _first_text(value[0]) return None def _parse_biblio_response(data: dict) -> list[PriorArtHit]: """EPO /published-data/search/biblio JSON → PriorArtHit listesi.""" try: search_result = data["ops:world-patent-data"]["ops:biblio-search"] entries = _as_list(search_result.get("ops:search-result", {}).get("exchange-documents")) except KeyError: return [] hits: list[PriorArtHit] = [] for entry_wrapper in entries: exchange_docs = _as_list(entry_wrapper.get("exchange-document")) for doc in exchange_docs: hit = _parse_single_document(doc) if hit: hits.append(hit) return hits def _parse_single_document(doc: dict) -> PriorArtHit | None: """Tek bir exchange-document'ı PriorArtHit'e çevir.""" country = doc.get("@country", "") doc_number = doc.get("@doc-number", "") kind = doc.get("@kind", "") if not (country and doc_number): return None patent_no = f"{country}{doc_number}{kind}" biblio = doc.get("bibliographic-data", {}) # Başlık title_text: str | None = None for title_entry in _as_list(biblio.get("invention-title")): if isinstance(title_entry, dict) and title_entry.get("@lang") == "en": title_text = _first_text(title_entry) break if not title_text: title_text = _first_text(biblio.get("invention-title")) # Başvuru tarihi (application-reference altında) filing_date: str | None = None app_refs = _as_list(biblio.get("application-reference")) for ref in app_refs: doc_id_list = _as_list(ref.get("document-id")) for did in doc_id_list: date_val = _first_text(did.get("date")) if date_val and len(date_val) == 8: # YYYYMMDD → YYYY-MM-DD filing_date = f"{date_val[:4]}-{date_val[4:6]}-{date_val[6:8]}" break if filing_date: break # Başvuru sahibi applicant_text: str | None = None parties = biblio.get("parties", {}) applicants = _as_list(parties.get("applicants", {}).get("applicant")) for ap in applicants: name = _first_text(ap.get("applicant-name", {}).get("name")) if name: applicant_text = name break # CPC sınıfları cpc_classes: list[str] = [] pat_class = _as_list(biblio.get("patent-classifications", {}).get("patent-classification")) for pc in pat_class: scheme = _first_text(pc.get("classification-scheme")) if scheme and "CPC" in scheme: section = _first_text(pc.get("section")) or "" cls = _first_text(pc.get("class")) or "" subclass = _first_text(pc.get("subclass")) or "" main_group = _first_text(pc.get("main-group")) or "" subgroup = _first_text(pc.get("subgroup")) or "" code = f"{section}{cls}{subclass}{main_group}/{subgroup}".strip("/") if code: cpc_classes.append(code) # Espacenet URL (görünüm için) url = f"https://worldwide.espacenet.com/patent/search/publication/{patent_no}" return PriorArtHit( source="epo", patent_no=patent_no, title=title_text, abstract=None, # /search/biblio abstract döndürmüyor; ayrı endpoint gerekir applicant=applicant_text, inventors=None, filing_date=filing_date, publication_date=None, cpc_classes=cpc_classes or None, url=url, )