Patent Public Search Support in view of PatFT/AppFT Deprecation

[mustberuss]

Wanted to make sure you knew about this:

In October 2022, other searching platforms (i.e. Pub EAST, Pub WEST, Pat/FT and App/FT) will be retired and replaced with Patent Public Search.

under "What is the timeline for Patent Public Search?" on https://ppubs.uspto.gov/pubwebapp/static/pages/faq.html

I found this page that shows the search fields https://ppubs.uspto.gov/pubwebapp/static/pages/searchable-indexes.html for SEARCH_FIELDS in settings.py PN is still patent number but issue date is now PD etc. and it looks like the response comes back as json.

I guess that's a round about feature request to have fulltext use ppubs when patft goes away. I'm not sure I'm pythonic enough to take this on myself but I could help if needed.

Thanks

[parkerhancock]

Thanks for the notice! I missed that! And man, it will be nice to retire the Pat/FT and App/FT modules. Half of the development time for this library was trying to craft xpath queries to make that all work.

I saw you forked the library. Would love to have your help updating the library to work with Patent Public Search. Adding a new module basically requires basically writing three files here:

• src/patent_client/uspto/public/manager.py
• src/patent_client/uspto/public/model.py
• src/patent_client/uspto/public/schema.py

The manager file is responsible for fetching the data. It should have a subclass of patent_client.util.Manager that implements a _get_results method that can fetch a result based on a ManagerConfig object.

The schema file is responsible for converting the response data provided by the manager into native Python types.

The model file is a dataclass for holding the data produced by the. schema and building links between the various data types.

A good place to start would be figuring out how to write the manager to get data out of Patent Public Search. The schema and model are pretty easy after that!

Let me know if you want/need any help or guidance!

[parkerhancock]

One more note - I was looking at Patent Public Search, and the data it gives back is extremely detailed - providing data in multiple different formats. Don't worry about mapping all of them in a Schema/Model. Instead, Patent Client can be opinionated, and just pull in the important stuff.

Thanks!

Parker

[mustberuss]

I'll give it a go but I haven't done much with python. I'm more of a perl/LWP guy with a java past life. I get that indenting counts but never got along with pandas...

thanks for the hints,
Russ

[parkerhancock]

Hey! If you're still interested in this, let me know and let's set up a time to chat on Zoom or something and go over how to do it. I'd love to get this done sometime before the deprecation date, if we can!

[mustberuss]

Hi Parker,

Zoom would help, there's an email address in my profile. I'm also in the Central time zone and have a flexible schedule.

thanks,
Russ

[parkerhancock]

Hey! On this topic - I've looked into how the API works, and I have the following thoughts.

Public Patents Session ID

First, below is some sample code that provides a Requests Session object customized to work with Patents Public Search. It modifies the basic session in two ways:

1. It automatically replaces any instance of "{{session_id}}" in a URL or request data (e.g. form or json) with the actual session ID.

2. It spawns a "keep-alive" thread that periodically pings the keep-alive endpoint about every 50 seconds.

import requests
from concurrent.futures import ThreadPoolExecutor
import time
import random

def dict_replace(dictionary, text, replacement):
    # Recursively iterate through a dictionary
    # and replace all occurences of "text" with "replacement"
    # Used to put the session ID in request information
    if not dictionary:
        return dictionary
    for k, v in dictionary.items():
        if isinstance(v, dict):
            dictionary[k] = dict_replace(v, text, replacement)
        elif isinstance(v, str) and v == text:
            dictionary[k] = replacement
    return dictionary


class PublicPatentsSession(requests.Session):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.executor = ThreadPoolExecutor(thread_name_prefix="Patent-Public-Search-Keep-Alive")
        self.session_id = None
        
    def request(self, *args, **kwargs):
        # Delayed initiation of session id
        if self.session_id == None:
            self.reset_session()
            
        # Insert the session id into all 
        if len(args) > 1: # URL is passed as a positional argument
            args = list(args)
            args[1] = args[1].replace("{{session_id}}", self.session_id)
        if "url" in kwargs: # URL is passed as a keyword argument
            kwargs['url'] = kwargs['url'].replace("{{session_id}}", self.session_id)        
        if "params" in kwargs:
            kwargs['params'] = dict_replace(kwargs['params'], "{{session_id}}", self.session_id)
        if "data" in kwargs:
            kwargs['data'] = dict_replace(kwargs['data'], "{{session_id}}", self.session_id)
        if "json" in kwargs:
            kwargs['json'] = dict_replace(kwargs['json'], "{{session_id}}", self.session_id)
            
        return super().request(*args, **kwargs)
            

    def reset_session(self):
        print("Resetting Public Search Session!")
        # Create a new session
        url = "https://ppubs.uspto.gov/dirsearch-public/users/me/session"
        response = super().request("post", url, json=str(random.randint(10000, 99999)))
        self.session_data = response.json()
        self.session_id = str(self.session_data['userSessionId'])
        # Kill the existing keep-alive thread
        self.executor.shutdown(wait=False, cancel_futures=True)
        # Create a new keep-alive thread
        self.executor = ThreadPoolExecutor(thread_name_prefix="Patent-Public-Search-Keep-Alive")
        self.executor.submit(self.keep_alive)
        
    def keep_alive(self):
        while True:
            response = self.session.put(f"https://ppubs.uspto.gov/dirsearch-public/users/lastUpdateDate/{self.session_id}")
            time.sleep(50)
            
    def __del__(self):
        self.executor.shutdown(wait=False, cancel_futures=True)

Data Flow

The API has four endpoints that will need to be wrapped/automated (that I've found so far). The process for executing a query is as follows:

1. Pre-flight the request to the Counts API
2. Obtain the Query ID from the Queries API
3. Retrieve a list of results using the Query ID from the Search API
4. Retrieve an individual record using the Highlight API using the Query ID

The Counts API

This is a query "pre-flight" endpoint that sends a query, and gives back some interesting information about the query. The main purpose of this endpoint for our purposes is that it will assign us a "query id". It looks like this:

url = "https://ppubs.uspto.gov/dirsearch-public/searches/counts"

query = "Some query to be sent"

data = {
    "caseId":"{{session_id}}",
    "hl_snippets":"2",
    "op":"OR",
    "q":query,
    "queryName":query,
    "highlights":"1",
    "qt":"brs",
    "spellCheck":False,
    "viewName":"tile",
    "plurals":True,
    "britishEquivalents":True,
    "databaseFilters":[
        {"databaseName":"US-PGPUB","countryCodes":[]},
        {"databaseName":"USPAT","countryCodes":[]},
        {"databaseName":"USOCR","countryCodes":[]}],
    "searchType":1,
    "ignorePersist":False,
    "userEnteredQuery":query
}

response = session.post(url, json=data)
response.json()

We actually don't care about what data it spits out. But if you don't pre-flight the query, it doesn't get assigned a query ID.

The Queries API

Once a query has been pre-flighted, you can hit the queries endpoint to get a list of queries that the current session has submitted, along with a query ID that is necessary to get complete records.

response = session.get("https://ppubs.uspto.gov/dirsearch-public/cases/{{session_id}}/queries")
response.json()
>>>[
    {'id': 1720226,
     'caseId': 737615,
     'numResults': 3007758,
  'ignorePersist': False,
  'queryName': 'Boppy Company',
  #(more omitted lines)
}]

The Search API

After the query has been pre-flighted, then stub records for the results can be retrieved from the "searchWithBeFamily endpoint":

data = {
    "start":0,
    "pageCount":500,
    "sort":"date_publ desc",
    "docFamilyFiltering":"familyIdFiltering",
    "searchType":1,
    "familyIdEnglishOnly":True,
    "familyIdFirstPreferred":"US-PGPUB",
    "familyIdSecondPreferred":"USPAT",
    "familyIdThirdPreferred":"FPRS",
    "showDocPerFamilyPref":"showEnglish",
    "queryId": "query ID", # The Query ID goes here 
    "tagDocSearch":False,
    "query":{
        "caseId": "{{session_id}},
        "hl_snippets":"2",
        "op":"OR",
        "q": query,
        "queryName":query,
        "highlights":"1",
        "qt":"brs",
        "spellCheck":False,
        "viewName":"tile",
        "plurals":True,
        "britishEquivalents":True,
        "databaseFilters":[
            {"databaseName":"US-PGPUB","countryCodes":[]},
            {"databaseName":"USPAT","countryCodes":[]},
            {"databaseName":"USOCR","countryCodes":[]}
        ],
        "searchType":1,
        "ignorePersist":True,
        "userEnteredQuery":query
    }
}

response = session.post(url, json=data)

That then gives you a list of "stub records" that has a ton of very interesting and useful information, but doesn't give you everything related to a publication record, for that you need the:

Highlight API

This API gives back the complete set of information for a publication. It looks like this:

url = "https://ppubs.uspto.gov/dirsearch-public/patents/{{reference_id}}/highlight"
params = {
    "queryId": "123123" # The query ID from the Queries API *MUST* go here,
    "source": "USPAT",
    "includeSections": True,
    "uniqueId": None
}

response = session.get(url, params=params)

And that completes the process.

[mustberuss]

I'm struggling to get this to work. Any additional help would welcome! Feel free to take it over or reassign.

[parkerhancock]

Hey Russ! I actually have a working proof-of-concept. The trick is to realize that the session ID that the API seems to require actually still works if you pass it any integer greater than zero. Just set it to 1 and then ignore all the complicated state stuff.

The only holdup from my end right now is that I'm trying to do some code-generation to get the model / schema files, since the schemas are very complex. I'm having the same issue over on wrapping the EP register, so I'm hoping once I figure out a way to infer schema from a pile of JSON / read from an XML schema, I can just auto-generate starting model.py and schema.py. There will still be some hand-editing to get where I want to go, but that will save a lot of tedium.

Thanks for your efforts though!

[parkerhancock]

See #89! Almost there! Just need to wait for PEDS to be back up to refresh the US Application tests...

[parkerhancock]

New Public Search classes are added in v3.1.0!