`scripts/load_trtyrap.py`

164 lines
#!/usr/bin/env python3
"""
load_trtyrap.py — parse a USPTO TRTYRAP trademark-applications XML file into a
searchable SQLite DB (db/trtyrap.sqlite).

Streaming (xml.etree.iterparse on </case-file>) so a 377MB file never lands in RAM.
Extracts the core fields per trademark: serial/registration #, the word mark, filing +
registration dates, live status, owner (name/city/state/country), primary int'l class,
and the goods-&-services text. Idempotent (serial_number PRIMARY KEY, upsert).

Usage:
    python3 scripts/load_trtyrap.py /path/to/apc*.xml [more.xml ...]
"""
from __future__ import annotations

import re
import sqlite3
import sys
import xml.etree.ElementTree as ET
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
DB = ROOT / "db" / "trtyrap.sqlite"
BATCH = 2000

DDL = """
CREATE TABLE IF NOT EXISTS trademark (
    serial_number       TEXT PRIMARY KEY,
    registration_number TEXT,
    mark                TEXT,
    mark_norm           TEXT,
    filing_date         TEXT,
    registration_date   TEXT,
    status_code         TEXT,
    status_date         TEXT,
    attorney_name       TEXT,
    owner_name          TEXT,
    owner_city          TEXT,
    owner_state         TEXT,
    owner_country       TEXT,
    primary_class       TEXT,
    intl_classes        TEXT,
    goods_services      TEXT
) WITHOUT ROWID;
"""
UPSERT = """
INSERT INTO trademark
 (serial_number,registration_number,mark,mark_norm,filing_date,registration_date,
  status_code,status_date,attorney_name,owner_name,owner_city,owner_state,owner_country,
  primary_class,intl_classes,goods_services)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
ON CONFLICT(serial_number) DO UPDATE SET
  registration_number=excluded.registration_number, mark=excluded.mark,
  mark_norm=excluded.mark_norm, filing_date=excluded.filing_date,
  registration_date=excluded.registration_date, status_code=excluded.status_code,
  status_date=excluded.status_date, attorney_name=excluded.attorney_name,
  owner_name=excluded.owner_name, owner_city=excluded.owner_city,
  owner_state=excluded.owner_state, owner_country=excluded.owner_country,
  primary_class=excluded.primary_class, intl_classes=excluded.intl_classes,
  goods_services=excluded.goods_services;
"""


def _t(el, path):
    """First matching descendant's text, or None."""
    f = el.find(path)
    return f.text.strip() if f is not None and f.text else None


def _regnum(s):  # 0000000 = not registered
    return None if not s or set(s) == {"0"} else s


def extract(cf) -> tuple:
    hdr = cf.find("case-file-header")
    mark = _t(hdr, "mark-identification") if hdr is not None else None
    # goods & services = concat of statement texts whose type-code starts with GS
    gs = []
    for st in cf.findall("case-file-statements/case-file-statement"):
        code = _t(st, "type-code") or ""
        if code.startswith("GS"):
            txt = _t(st, "text")
            if txt:
                gs.append(txt)
    # first owner
    ow = cf.find("case-file-owners/case-file-owner")
    intl = [c.text.strip() for c in cf.findall("classifications/classification/international-code")
            if c.text and c.text.strip()]
    return (
        _t(cf, "serial-number"),
        _regnum(_t(cf, "registration-number")),
        mark,
        (mark or "").upper() or None,
        _t(hdr, "filing-date") if hdr is not None else None,
        _t(hdr, "registration-date") if hdr is not None else None,
        _t(hdr, "status-code") if hdr is not None else None,
        _t(hdr, "status-date") if hdr is not None else None,
        _t(hdr, "attorney-name") if hdr is not None else None,
        _t(ow, "party-name") if ow is not None else None,
        _t(ow, "city") if ow is not None else None,
        _t(ow, "state") if ow is not None else None,
        _t(ow, "country") if ow is not None else None,
        _t(cf, "classifications/classification/primary-code"),
        ";".join(dict.fromkeys(intl)) or None,
        " | ".join(gs) or None,
    )


def load_file(conn, path: Path) -> int:
    cur = conn.cursor()
    batch, n = [], 0
    # iterparse streams; we clear each <case-file> after use to keep memory flat.
    for _ev, cf in ET.iterparse(str(path), events=("end",)):
        if cf.tag != "case-file":
            continue
        row = extract(cf)
        if row[0]:  # has a serial number
            batch.append(row)
        cf.clear()
        if len(batch) >= BATCH:
            cur.executemany(UPSERT, batch)
            conn.commit()
            n += len(batch)
            batch.clear()
            if n % 20000 == 0:
                print(f"    …{n:,} trademarks", flush=True)
    if batch:
        cur.executemany(UPSERT, batch)
        conn.commit()
        n += len(batch)
    return n


def main(argv):
    files = [Path(a) for a in argv]
    files = [f for f in files if f.exists()]
    if not files:
        print("usage: load_trtyrap.py <file.xml> [...]", file=sys.stderr)
        return 1
    DB.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(DB)
    conn.execute("PRAGMA journal_mode=WAL;")
    conn.execute("PRAGMA synchronous=NORMAL;")
    conn.executescript(DDL)
    total = 0
    for f in files:
        print(f"Parsing {f.name} …", flush=True)
        c = load_file(conn, f)
        total += c
        print(f"  {f.name}: +{c:,} trademarks", flush=True)
    print("Indexing (mark + owner + status) …", flush=True)
    conn.execute("CREATE INDEX IF NOT EXISTS idx_mark ON trademark(mark_norm);")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_owner ON trademark(owner_name);")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_status ON trademark(status_code);")
    conn.commit()
    tot = conn.execute("SELECT COUNT(*) FROM trademark").fetchone()[0]
    conn.close()
    print(f"\nDONE. db/trtyrap.sqlite holds {tot:,} trademarks (this run +{total:,}).")
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))