← back to Uspto Trtyrap
scripts/serve_trtyrap.py
162 lines
#!/usr/bin/env python3
"""
serve_trtyrap.py — public search over the USPTO TRTYRAP trademark DB.
Serves db/trtyrap.sqlite on http://127.0.0.1:8800 : a search box (matches either the
word mark OR the owner name) + a results table with the mark, owner, live status,
int'l class, and a goods-&-services excerpt.
Unlike the California unclaimed-property server there is NO masking layer — a trademark
IS a public government registration (the whole point is public notice), so showing the
full mark + owner + goods is correct, not a privacy leak.
Stdlib only. Read-only against the DB. Usage: python3 scripts/serve_trtyrap.py [port]
"""
from __future__ import annotations
import html
import json
import sqlite3
import sys
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlparse, parse_qs
ROOT = Path(__file__).resolve().parents[1]
DB = ROOT / "db" / "trtyrap.sqlite"
PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 8800
MAX_ROWS = 100
# USPTO status codes are 3-digit; we translate only the handful that dominate the
# applications file into a plain-English label, and fall back to the raw code.
STATUS = {
"681": "Registered / renewed",
"700": "Registered",
"710": "Cancelled",
"606": "Abandoned",
"641": "Opposition pending",
"638": "Suspended",
"930": "Live / pending",
}
PAGE = """<!doctype html><html><head><meta charset=utf-8>
<title>USPTO Trademark Search</title>
<meta name=viewport content="width=device-width,initial-scale=1">
<style>
body{{font:15px/1.5 -apple-system,Segoe UI,Roboto,sans-serif;margin:0;background:#0d1117;color:#e6edf3}}
header{{padding:22px 26px;border-bottom:1px solid #30363d;background:#161b22}}
h1{{margin:0;font-size:19px;font-weight:650}} .sub{{color:#8b949e;font-size:13px;margin-top:4px}}
main{{max-width:1040px;margin:0 auto;padding:22px 26px}}
form{{display:flex;gap:10px;margin-bottom:18px}}
input[type=text]{{flex:1;padding:11px 13px;border:1px solid #30363d;border-radius:8px;background:#0d1117;color:#e6edf3;font-size:15px}}
button{{padding:11px 20px;border:0;border-radius:8px;background:#238636;color:#fff;font-weight:600;cursor:pointer}}
table{{width:100%;border-collapse:collapse;font-size:14px}}
th,td{{text-align:left;padding:9px 11px;border-bottom:1px solid #21262d;vertical-align:top}}
th{{color:#8b949e;font-weight:600;font-size:12px;text-transform:uppercase;letter-spacing:.04em}}
td.mark{{font-weight:650;color:#e6edf3}}
.cls{{font-family:ui-monospace,Menlo,monospace;color:#58a6ff}}
.gs{{color:#8b949e;font-size:13px;max-width:360px}}
.status{{display:inline-block;background:#1f6feb22;color:#58a6ff;border:1px solid #1f6feb55;border-radius:5px;padding:1px 7px;font-size:11px}}
.meta{{color:#8b949e;font-size:13px;margin:10px 0 16px}}
.empty{{color:#8b949e;padding:30px 0}}
.badge{{display:inline-block;background:#8957e522;color:#a371f7;border:1px solid #8957e555;border-radius:5px;padding:1px 7px;font-size:11px;margin-left:8px}}
</style></head><body>
<header><h1>USPTO Trademark Search <span class=badge>PUBLIC · TRTYRAP</span></h1>
<div class=sub>{count} trademarks · full-text application/registration data from the USPTO</div></header>
<main>
<form method=get action="/"><input type=text name=q placeholder="Search a mark or owner (e.g. APPLE, or a company name)" value="{q}" autofocus>
<button type=submit>Search</button></form>
{results}
</main></body></html>"""
def _status(code: str | None) -> str:
if not code:
return "—"
return STATUS.get(code, f"Code {code}")
class Handler(BaseHTTPRequestHandler):
def log_message(self, *a): # quiet
pass
def _db(self):
c = sqlite3.connect(f"file:{DB}?mode=ro", uri=True)
c.row_factory = sqlite3.Row
return c
# one WHERE for both the mark and the owner, so a query hits either field
_WHERE = "mark_norm LIKE ? OR owner_name LIKE ?"
def do_GET(self):
u = urlparse(self.path)
if u.path == "/api/search":
return self._api(parse_qs(u.query).get("q", [""])[0].strip())
if u.path != "/":
self.send_error(404)
return
q = parse_qs(u.query).get("q", [""])[0].strip()
c = self._db()
total = c.execute("SELECT COUNT(*) FROM trademark").fetchone()[0]
results = ""
if q:
like = f"%{q.upper()}%"
rows = c.execute(
f"""SELECT mark, owner_name, owner_state, status_code, primary_class,
filing_date, goods_services
FROM trademark WHERE {self._WHERE}
ORDER BY (mark_norm = ?) DESC, filing_date DESC LIMIT ?""",
(like, like, q.upper(), MAX_ROWS)).fetchall()
hits = c.execute(
f"SELECT COUNT(*) FROM trademark WHERE {self._WHERE}",
(like, like)).fetchone()[0]
if rows:
body = "".join(
f"<tr><td class=mark>{html.escape(r['mark'] or '—')}</td>"
f"<td>{html.escape((r['owner_name'] or '').title())}"
f"{(', ' + html.escape(r['owner_state'])) if r['owner_state'] else ''}</td>"
f"<td><span class=status>{html.escape(_status(r['status_code']))}</span></td>"
f"<td class=cls>{html.escape(r['primary_class'] or '—')}</td>"
f"<td class=gs>{html.escape((r['goods_services'] or '')[:160])}</td></tr>"
for r in rows)
more = f" (showing top {MAX_ROWS})" if hits > MAX_ROWS else ""
results = (f"<div class=meta>{hits:,} match(es) for “{html.escape(q)}”{more}.</div>"
"<table><tr><th>Mark</th><th>Owner</th><th>Status</th>"
"<th>Class</th><th>Goods & services</th></tr>" + body + "</table>")
else:
results = f"<div class=empty>No trademarks match “{html.escape(q)}”.</div>"
c.close()
out = PAGE.format(count=f"{total:,}", q=html.escape(q), results=results)
self._send(out.encode(), "text/html; charset=utf-8")
def _api(self, q: str):
c = self._db()
rows = []
if q:
like = f"%{q.upper()}%"
rows = [dict(r) for r in c.execute(
f"""SELECT mark, owner_name, owner_city, owner_state, owner_country,
status_code, primary_class, intl_classes, filing_date,
registration_number, goods_services
FROM trademark WHERE {self._WHERE}
ORDER BY (mark_norm = ?) DESC, filing_date DESC LIMIT ?""",
(like, like, q.upper(), MAX_ROWS))]
c.close()
self._send(json.dumps({"query": q, "count": len(rows), "results": rows}).encode(),
"application/json")
def _send(self, body: bytes, ctype: str):
self.send_response(200)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
if __name__ == "__main__":
if not DB.exists():
print(f"DB not found: {DB} — run scripts/load_trtyrap.py first", file=sys.stderr)
raise SystemExit(1)
print(f"Trademark search → http://127.0.0.1:{PORT} (DB: {DB.name})")
ThreadingHTTPServer(("127.0.0.1", PORT), Handler).serve_forever()