Tess Trademark Work in Progress

src/patent_client/uspto/tess/manager.py

@@ -0,0 +1,85 @@
+from patent_client.util.base.manager import Manager
+from .session import ExpiredSessionException, session
+from .schema import ResultTableSchema
+import lxml.etree as ET
+
+class TessManager(Manager):
+    result_schema = ResultTableSchema()
+
+    def __init__(self, config=None):
+        super().__init__(config)
+
+    def __len__(self):
+        len = self.execute_query()['num_results'] - self.config.offset
+        if self.config.limit:
+            return min(self.config.limit, len)
+        else:
+            return len
+
+    def last_updated(self):
+        return self.execute_query()['last_updated']
+
+    def _get_results(self):
+        # Set up query
+        data = self.execute_query()
+
+        # yield records up to limit
+        for i, record in enumerate(self.record_iterator(self.config.offset)):
+            if i >= self.config.limit:
+                break
+            yield record
+
+    def record_iterator(self, offset):
+        index = offset + 1
+        while True:
+            try:
+                page_response = session.get("https://tmsearch.uspto.gov/bin/jumpto", params={
+                    "f": "toc", 
+                    "state": self.session.state.string, 
+                    "jumpto": index
+                })
+            except ExpiredSessionException:
+                self.execute_query()
+                page_response = session.get("https://tmsearch.uspto.gov/bin/jumpto", params={
+                    "f": "toc", 
+                    "state": session.state.string, 
+                    "jumpto": index
+                })
+            tree = ET.HTML(page_response.text)
+            results = self.result_schema(tree)
+            for result in results['results']:
+                yield result
+            index += 100
+            if len(results['results'] < 100):
+                break
+
+    # Prohibited methods
+    def order_by(self, *args, **kwargs):
+        raise NotImplementedError("The TESS interface does not permit sorting")
+
+    def execute_query(self):
+        if "query" in self.config.filter:
+            query = self.config.filter['query']
+        else:
+            query = self.get_query(self.config.filter)
+        params = {
+            "f": "toc",
+            "state": session.state.string,
+            "p_search": "search",
+            "p_s_All": None,
+            "BackReference": None,
+            "p_L": "100",
+            "p_plural": "yes",
+            "p_s_ALL": query,
+            "a_search": "Submit Query",
+        }
+        try:
+            response = session.post("https://tmsearch.uspto.gov/bin/gate.exe", data=params)
+        except ExpiredSessionException:
+            session.login()
+            return self.execute_query()
+        return self.parse_page(response.text)
+
+    def parse_page(self, page):
+        tree = ET.HTML(page)
+        return self.result_schema.load(tree)

src/patent_client/uspto/tess/manager_test.py

@@ -0,0 +1,11 @@
+from .manager import TessManager
+
+
+def test_get_length():
+    assert TessManager().filter(query="Nike[FM]").count() == 55
+
+def test_limit():
+    assert TessManager().filter(query="Nike[FM]").limit(50).count() == 50
+
+def test_offset():
+    assert TessManager().filter(query="Nike[FM]").offset(15).count() == 40

src/patent_client/uspto/tess/model.py

@@ -0,0 +1,37 @@
+import datetime
+from patent_client.base.model import Model
+from .session import session
+
+class Result(Model):
+    pos: str = None
+    serial_num: str = None
+    reg_num: str = None
+    word_mark: str = None
+    record_url: str = None
+    status_url: str = None
+    live: str = None
+    classes: str = None
+
+
+
+class TessRecord(Model):
+    img_url: str = None
+    word_mark: str = None
+    goods_and_services: str = None
+    standard_characters_claimed: str = None
+    mark_drawing_code: str = None
+    trademark_search_facility_classification_code: str = None
+    serial_number: str = None
+    filing_date: datetime.date = None
+    current_basis: str = None
+    original_filing_basis: str = None
+    published_for_opposition: str = None
+    registration_number: str = None
+    registration_date: datetime.date = None
+    owner: str = None
+    attorney_of_record: str = None
+    prior_registrations: str = None
+    type_of_mark: str = None
+    register: str = None
+    renewal: str = None
+    live_dead_indicator: str = None

src/patent_client/uspto/tess/schema.py

@@ -0,0 +1,67 @@
+import re
+import lxml.etree as ET
+from yankee.xml import Schema, fields as f
+
+classification_re = re.compile(r"IC (\d{3}).")
+
+def fmt_classification(string):
+    match = classification_re.search(string)
+    return match.group(1) if match else None
+
+class ResultSchema(Schema):
+    pos = f.Int(".//td[1]")
+    serial_num = f.Str(".//td[2]")
+    reg_num = f.Str(".//td[3]")
+    word_mark = f.Str(".//td[4]")
+    record_url = f.Str(".//td[2]/a/@href", formatter=lambda s: "https://tmsearch.uspto.gov" + s)
+    status_url = f.Str(".//td[5]/a/@href")
+    live = f.Str(".//td[6]")#, true_value="LIVE")
+    classes = f.Str(".//td[7]", formatter=fmt_classification)
+
+num_results_re = re.compile(r"docs: (\d*)")
+
+def fmt_num_results(string):
+    match = num_results_re.search(string)
+    return match.group(1) if match else None
+
+occ_re = re.compile(r"occ: (\d*)")
+
+def fmt_occ(string):
+    match = occ_re.search(string)
+    return match.group(1) if match else None
+
+last_updated_re = re.compile(r"TESS was last updated on (.*$)")
+
+def fmt_last_updated(string):
+    match = last_updated_re.search(string)
+    return match.group(1) if match else None
+
+class ResultTableSchema(Schema):
+    num_results = f.Int(".//table[5]//td[4]", formatter=fmt_num_results)
+    num_occurences = f.Int(".//table[5]//td[4]", formatter=fmt_occ)
+    last_updated = f.DateTime(".//table[2]//i", formatter=fmt_last_updated)
+    results = f.List(ResultSchema, './/table[@id="searchResultTable"]/tr')
+
+table_xpath = './/td/b[contains(text(), "{}")]/following::td'
+
+class RecordSchema(Schema):
+    img_url = f.Str('.//img[@alt="Mark Image"]/@src', formatter=lambda s: "https://tmsearch.uspto.gov" + s)
+    word_mark = f.Str(table_xpath.format("Word Mark"))
+    goods_and_services = f.Str(table_xpath.format("Goods and Services"))
+    standard_characters_claimed = f.Str(table_xpath.format("Standard Characters Claimed"))
+    mark_drawing_code = f.Str(table_xpath.format("Mark Drawing Code"))
+    trademark_search_facility_classification_code = f.Str(table_xpath.format("Trademark Search Facility Classification Code"))
+    serial_number = f.Str(table_xpath.format("Serial Number"))
+    filing_date = f.Date(table_xpath.format("Filing Date"))
+    current_basis = f.Str(table_xpath.format("Current Basis"))
+    original_filing_basis = f.Str(table_xpath.format("Original Filing Basis"))
+    published_for_opposition = f.Date(table_xpath.format("Published for Opposition"))
+    registration_number = f.Str(table_xpath.format("Registration Number"))
+    registration_date = f.Date(table_xpath.format("Registration Date"))
+    owner = f.Str(table_xpath.format("Owner"))
+    attorney_of_record = f.Str(table_xpath.format("Attorney of Record"))
+    prior_registrations = f.Str(table_xpath.format("Prior Registrations"))
+    type_of_mark = f.Str(table_xpath.format("Type of Mark"))
+    register = f.Str(table_xpath.format("Register"))
+    renewal = f.Str(table_xpath.format("Renewal"))
+    live_dead_indicator = f.Str(table_xpath.format("Live/Dead Indicator"))

src/patent_client/uspto/tess/session.py

@@ -0,0 +1,57 @@
+import re
+import requests
+
+
+state_re = re.compile(r"state=(?P<session_id>\d{4}:\w+).(?P<query_id>\d+).(?P<record_id>\d+)")
+
+class TessState():    
+    def __init__(self, response_body=None):
+        if response_body is None:
+            self.session_id = self.query_id = self.record_id = ''
+        else:
+            state = state_re.search(response_body).groupdict()
+            self.session_id = state['session_id']
+            self.query_id = state['query_id']
+            self.record_id = state['record_id']
+
+    def __str__(self):
+        return f"{self.session_id}.{self.query_id}.{self.record_id}"
+
+    @property
+    def string(self):
+        return str(self)
+
+class ExpiredSessionException(Exception):
+    pass
+
+class TessSession(requests.Session):
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.state = TessState()
+        self.headers['user-agent'] = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/105.0.0.0 Safari/537.36"
+
+    def request(self, *args, **kwargs):
+        state = bool(self.state)
+        if not state:
+            self.login()
+        response = super().request(*args, **kwargs)
+        if state_re.search(response.text):
+            self.state = TessState(response.text)
+        if "This search session has expired." in response.text or "Bad query - no database specified" in response.text:
+            raise ExpiredSessionException()
+        return response
+
+    def get_state(self):
+        return self.state
+
+    def login(self):
+        login_response = super().request("get", "https://tmsearch.uspto.gov/bin/gate.exe", params={"f": "login", "p_lang": "english", "p_d": "trmk"})
+        self.state = TessState(login_response.text)
+        print(f"Logged in! Current State: {self.state}")
+
+    def __del__(self):
+        super().request("post", "https://tmsearch.uspto.gov/bin/gate.exe", data={"state": str(self.state), "f": "logout", "a_logout": "Logout"})
+        print("Logged out!")
+
+session = TessSession()

src/patent_client/util/base/collections.py

@@ -18,6 +18,12 @@ def __iter__(self):
     # def __len__(self):
     #    return len(self.iterable)
 
+    def len(self):
+        return len(self)
+
+    def count(self):
+        return len(self)
+
     def __repr__(self):
         if hasattr(self, "iterable"):
             return f"Collection({repr(self.iterable)})"