# @name: ops.py # @creation_date: 2022-09-08 # @license: The MIT License <https://opensource.org/licenses/MIT> # @author: Simon Bowie <simon.bowie.19@gmail.com> # @purpose: Performs functions against the European Patent Office's Open Patent Services (OPS) API # @acknowledgements: # OPS documented at https://www.epo.org/searching-for-patents/data/web-services/ops.html # OPS RESTful API specification at http://documents.epo.org/projects/babylon/eponet.nsf/0/F3ECDCC915C9BCD8C1258060003AA712/$File/ops_v3.2_documentation_-_version_1.3.18_en.pdf # OPS API functions list at https://developers.epo.org/ops-v3-2/apis import os import requests import base64 from wand.image import Image # get config variables from OS environment variables: set in env file passed through Docker Compose ops_url = os.environ.get('OPS_URL') ops_url_images = os.environ.get('OPS_URL_IMAGES') consumer_key = os.environ.get('CONSUMER_KEY') consumer_secret = os.environ.get('CONSUMER_SECRET') def get_access_token(): # OPS API credentials (details at http://documents.epo.org/projects/babylon/eponet.nsf/0/F3ECDCC915C9BCD8C1258060003AA712/$File/ops_v3.2_documentation_-_version_1.3.18_en.pdf) endpoint_url = ops_url + '3.2/auth/accesstoken' auth = consumer_key + ":" + consumer_secret auth_bytes = auth.encode("ascii") base64_bytes = base64.b64encode(auth_bytes) base64_string = base64_bytes.decode("ascii") # set up API call headers = {"Authorization": "Basic " + base64_string, "Content-Type": "application/x-www-form-urlencoded"} data = "grant_type=client_credentials" # give back result response = requests.post(endpoint_url, headers=headers, data=data) if response.status_code == 200: # turn the API response into useful Json json = response.json() access_token = json['access_token'] return access_token def get_publication_details(doc_ref): access_token = get_access_token() # OPS API credentials (details at http://documents.epo.org/projects/babylon/eponet.nsf/0/F3ECDCC915C9BCD8C1258060003AA712/$File/ops_v3.2_documentation_-_version_1.3.16_en.pdf) endpoint_url = ops_url + 'rest-services/published-data/publication/docdb/' + doc_ref + '/biblio' # set up API call headers = {"Authorization": "Bearer " + access_token, "Accept": "application/json"} # get result response = requests.get(endpoint_url, headers=headers) output = {} if response.status_code == 200: # turn the API response into useful Json json = response.json() # for each invention title, check if it's in the original language try: json['ops:world-patent-data']['exchange-documents']['exchange-document']['bibliographic-data']['invention-title'] invention_titles = json['ops:world-patent-data']['exchange-documents']['exchange-document']['bibliographic-data']['invention-title'] try: invention_titles[1] for invention_title in invention_titles: if invention_title['@lang'] is not None and invention_title['@lang'] != 'en': output['original_title'] = invention_title['$'] except KeyError: if invention_titles['@lang'] is not None and invention_titles['@lang'] != 'en': output['original_title'] = invention_titles['$'] except KeyError: pass # for each abstract, check if it's in the original language try: json['ops:world-patent-data']['exchange-documents']['exchange-document']['abstract'] abstracts = json['ops:world-patent-data']['exchange-documents']['exchange-document']['abstract'] try: abstracts[1] for abstract in abstracts: if abstract['@lang'] is not None and abstract['@lang'] != 'en': output['original_abstract'] = abstract['p']['$'] except KeyError: if abstracts['@lang'] is not None and abstracts['@lang'] != 'en': output['original_abstract'] = abstracts['p']['$'] except KeyError: pass return output def get_images(doc_ref): access_token = get_access_token() # OPS API credentials (details at http://documents.epo.org/projects/babylon/eponet.nsf/0/F3ECDCC915C9BCD8C1258060003AA712/$File/ops_v3.2_documentation_-_version_1.3.16_en.pdf) endpoint_url = ops_url + 'rest-services/published-data/publication/docdb/' + doc_ref + '/images' # set up API call headers = {"Authorization": "Bearer " + access_token, "Accept": "application/json"} # give back result response = requests.get(endpoint_url, headers=headers) if response.status_code == 200: output = {} drawings_url = {} # turn the API response into useful Json json = response.json() try: json['ops:world-patent-data']['ops:document-inquiry']['ops:inquiry-result']['ops:document-instance'] document_instances = json['ops:world-patent-data']['ops:document-inquiry']['ops:inquiry-result']['ops:document-instance'] try: document_instances[0] for document_instance in document_instances: if document_instance['@desc'] == 'Drawing': drawings_url = ops_url_images + '3.2/rest-services/' + document_instance['@link'] + '?Range=1' if drawings_url is None: for document_instance in document_instances: if document_instance['@desc'] == 'FullDocument': drawings_url = ops_url_images + '3.2/rest-services/' + document_instance['@link'] + '?Range=1' except KeyError: pass try: drawings_url[0] # set up API call headers = {"Authorization": "Bearer " + access_token, "Accept": "application/tiff"} # give back result response = requests.get(drawings_url, headers=headers) if response.status_code == 200: with Image(blob = response.content) as image: png_blob = image.make_blob('png') base64_bytes = base64.b64encode(png_blob) output['image'] = base64_bytes.decode("ascii") except KeyError: pass except KeyError: pass else: output = False return output