saksham
1a4c80958f
FastAPI + Celery + Next.js + Postgres/Redis app with company monitoring, source collection, LLM-based change analysis, enrichment, and account security (Turnstile, escalating lockout, email verification).
246 lines
11 KiB
Python
246 lines
11 KiB
Python
"""Patent source collector.
|
|
|
|
USPTO's PatentsView data migrated into the Open Data Portal (ODP) in March
|
|
2026; a free API key is available via account registration at
|
|
data.uspto.gov/apis/getting-started (see `Settings.uspto_api_key`). Without
|
|
a key configured (the default), this collector never fabricates patent
|
|
data - it truthfully reports `DISABLED` with an explanation, same as
|
|
before this integration existed, and a fixture adapter remains available
|
|
for local development/testing.
|
|
|
|
With a key configured, `collect()` calls the real ODP Patent Application
|
|
Search API - by INVENTOR NAME, not company name. Confirmed live (2026-08)
|
|
by inspecting a real response's full field list, including one from a
|
|
query that returned 110k+ real results: there is no assignee/company field
|
|
anywhere in this endpoint's data model. Company-name search here always
|
|
returns "no matching records," even for assignees with thousands of real
|
|
patents - it isn't a wrong-field-name bug, the field doesn't exist on this
|
|
dataset. USPTO's Patent Application Search reliably supports inventor-name
|
|
and application-number lookups only.
|
|
|
|
So `collect()` instead searches by each of the company's known leadership
|
|
names (from NinjaPear enrichment, see `CompanyContext.leadership_names` /
|
|
`enrichment_service.py`) and treats a match as a heuristic company signal,
|
|
not a verified one - there is still no way to confirm a given patent
|
|
actually belongs to the monitored company rather than, say, a same-named
|
|
person, or work the person did at a prior employer. Every resulting
|
|
document is trust-scored lower (0.5, vs. a hypothetical verified-assignee
|
|
match) and its content explicitly says which leadership name it matched
|
|
on, so the report LLM's confidence labeling reflects this rather than
|
|
treating it as confirmed fact. With no leadership names available (no
|
|
NinjaPear key, enrichment still pending, or it returned no leadership
|
|
data), this reports an honest empty result without making a network call
|
|
- there's nothing meaningful to search USPTO for.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
|
|
from app.collectors.base import (
|
|
CollectedDocument,
|
|
CollectionResult,
|
|
CompanyContext,
|
|
DiscoveredSource,
|
|
SourceConfig,
|
|
)
|
|
from app.collectors.extraction import compute_content_hash, normalize_whitespace
|
|
from app.core.config import get_settings
|
|
from app.core.logging import get_logger
|
|
from app.models.enums import SourceStatus, SourceType
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
_FIXTURES_DIR = Path(__file__).resolve().parents[2] / "tests" / "fixtures" / "patents"
|
|
_SEARCH_URL = "https://api.uspto.gov/api/v1/patent/applications/search"
|
|
_MAX_INVENTOR_SEARCHES = 5
|
|
_MAX_DOCUMENTS = 25
|
|
|
|
|
|
class PatentSourceCollector:
|
|
source_type = SourceType.PATENT
|
|
|
|
async def discover(self, company: CompanyContext) -> list[DiscoveredSource]:
|
|
# company.uspto_api_key, when set by the caller (see
|
|
# collection_service.to_company_context), is the resolved effective
|
|
# key for whichever user owns this company - their own if they've
|
|
# set one, else the server's global default. None means no
|
|
# per-user resolution happened for this call path, so fall back to
|
|
# the global settings directly.
|
|
api_key = company.uspto_api_key or get_settings().uspto_api_key
|
|
if not api_key:
|
|
return [] # No live discovery without a configured provider.
|
|
return [
|
|
DiscoveredSource(
|
|
source_type=SourceType.PATENT,
|
|
name=f"{company.name} — Patent Filings",
|
|
base_url=None,
|
|
configuration_metadata={"assignee": company.name},
|
|
)
|
|
]
|
|
|
|
async def collect(self, source: SourceConfig, company: CompanyContext) -> CollectionResult:
|
|
api_key = company.uspto_api_key or get_settings().uspto_api_key
|
|
if api_key:
|
|
return await self._collect_live(company.name, company.leadership_names, api_key)
|
|
|
|
fixture_key = source.configuration_metadata.get("fixture_key")
|
|
if not fixture_key:
|
|
return CollectionResult(
|
|
status=SourceStatus.DISABLED,
|
|
error=(
|
|
"No patent data provider is configured. This collector implements the "
|
|
"SourceCollector interface for a live integration (USPTO Open Data Portal) "
|
|
"but does not fabricate results without a configured USPTO_API_KEY."
|
|
),
|
|
)
|
|
|
|
fixture_path = _FIXTURES_DIR / f"{fixture_key}.json"
|
|
if not fixture_path.exists():
|
|
return CollectionResult(
|
|
status=SourceStatus.DISABLED,
|
|
error=f"No fixture found for {fixture_key!r} and no live provider is configured.",
|
|
)
|
|
|
|
payload = json.loads(fixture_path.read_text(encoding="utf-8"))
|
|
documents: list[CollectedDocument] = []
|
|
for entry in payload.get("patents", []):
|
|
text = normalize_whitespace(f"{entry['title']}\n\n{entry.get('abstract', '')}")
|
|
documents.append(
|
|
CollectedDocument(
|
|
url=entry.get("url", fixture_path.as_uri()),
|
|
canonical_url=entry.get("url", fixture_path.as_uri()),
|
|
title=entry["title"],
|
|
author=entry.get("assignee"),
|
|
publication_date=(
|
|
datetime.fromisoformat(entry["filed_date"]).replace(tzinfo=UTC)
|
|
if entry.get("filed_date")
|
|
else None
|
|
),
|
|
retrieved_date=datetime.now(UTC),
|
|
content_text=text,
|
|
content_hash=compute_content_hash(text),
|
|
metadata={
|
|
"is_fixture": True,
|
|
"data_source": "fixture",
|
|
"fixture_key": fixture_key,
|
|
},
|
|
extraction_method="fixture",
|
|
trust_score=0.5,
|
|
)
|
|
)
|
|
return CollectionResult(status=SourceStatus.ACTIVE, documents=documents, pages_attempted=1)
|
|
|
|
async def _collect_live(
|
|
self, company_name: str, leadership_names: list[str], api_key: str
|
|
) -> CollectionResult:
|
|
if not leadership_names:
|
|
# No names to search USPTO's inventor index with - see the
|
|
# module docstring for why company-name search doesn't work on
|
|
# this endpoint at all. Honest empty result, no network call.
|
|
return CollectionResult(status=SourceStatus.ACTIVE, documents=[], pages_attempted=0)
|
|
|
|
documents: list[CollectedDocument] = []
|
|
seen_app_numbers: set[str] = set()
|
|
pages_attempted = 0
|
|
errors: list[str] = []
|
|
|
|
async with httpx.AsyncClient(timeout=20) as client:
|
|
for inventor_name in leadership_names[:_MAX_INVENTOR_SEARCHES]:
|
|
pages_attempted += 1
|
|
body = {
|
|
"q": f'applicationMetaData.inventorBag.inventorNameText:"{inventor_name}"',
|
|
"pagination": {"limit": _MAX_DOCUMENTS},
|
|
"sort": [{"field": "applicationMetaData.filingDate", "order": "desc"}],
|
|
}
|
|
try:
|
|
response = await client.post(
|
|
_SEARCH_URL, json=body, headers={"x-api-key": api_key}
|
|
)
|
|
except httpx.HTTPError as exc:
|
|
errors.append(f"{inventor_name}: {exc}")
|
|
continue
|
|
|
|
if response.status_code == 404:
|
|
# USPTO returns 404 for "no matching records" rather
|
|
# than 200 with an empty array - a real, expected
|
|
# outcome for most names, not a failure.
|
|
continue
|
|
if response.status_code >= 400:
|
|
errors.append(f"{inventor_name}: USPTO API error {response.status_code}")
|
|
continue
|
|
|
|
try:
|
|
payload = response.json()
|
|
except ValueError:
|
|
errors.append(f"{inventor_name}: malformed USPTO response")
|
|
continue
|
|
|
|
entries = payload.get("patentFileWrapperDataBag") or payload.get("results") or []
|
|
for entry in entries:
|
|
metadata = entry.get("applicationMetaData") or {}
|
|
app_number = entry.get("applicationNumberText") or entry.get(
|
|
"applicationNumber"
|
|
)
|
|
if not app_number or app_number in seen_app_numbers:
|
|
continue
|
|
seen_app_numbers.add(app_number)
|
|
|
|
title = metadata.get("inventionTitle") or "Untitled patent filing"
|
|
filing_date = metadata.get("filingDate")
|
|
abstract = metadata.get("abstractText") or ""
|
|
text = normalize_whitespace(
|
|
f"{title}\n\nInventor match: {inventor_name} (leadership-name "
|
|
f"heuristic, not a verified {company_name} assignee - USPTO's "
|
|
"application search has no queryable assignee/company field).\n\n"
|
|
f"{abstract}"
|
|
)
|
|
documents.append(
|
|
CollectedDocument(
|
|
url=(
|
|
f"{_SEARCH_URL}?applicationNumber={app_number}"
|
|
if app_number
|
|
else _SEARCH_URL
|
|
),
|
|
canonical_url=_SEARCH_URL,
|
|
title=title,
|
|
author=inventor_name,
|
|
publication_date=(
|
|
datetime.fromisoformat(filing_date).replace(tzinfo=UTC)
|
|
if filing_date
|
|
else None
|
|
),
|
|
retrieved_date=datetime.now(UTC),
|
|
content_text=text,
|
|
content_hash=compute_content_hash(text),
|
|
metadata={
|
|
"application_number": app_number,
|
|
"data_source": "uspto_odp",
|
|
"matched_inventor_name": inventor_name,
|
|
"match_type": "leadership_name_heuristic",
|
|
},
|
|
extraction_method="uspto_odp_api",
|
|
http_status=response.status_code,
|
|
# Lower than a verified-assignee match would be
|
|
# (was 0.9) - this is a heuristic name match,
|
|
# not confirmed company ownership.
|
|
trust_score=0.5,
|
|
)
|
|
)
|
|
if len(documents) >= _MAX_DOCUMENTS:
|
|
break
|
|
|
|
if errors and not documents:
|
|
return CollectionResult(
|
|
status=SourceStatus.FAILED,
|
|
error="; ".join(errors[:3]),
|
|
pages_attempted=pages_attempted,
|
|
)
|
|
return CollectionResult(
|
|
status=SourceStatus.ACTIVE, documents=documents, pages_attempted=pages_attempted
|
|
)
|