micheleandCursor
dd46ce1da1
Store BOM MPN and orderable code on the same PDF blob, derive TI datasheet slugs from package/Q1 codes, and map LCSC/DigiKey R/C/L parameters to specs without a model call. Co-authored-by: Cursor <cursoragent@cursor.com>
373 lines
12 KiB
Python
373 lines
12 KiB
Python
"""DigiKey API integration — fetch datasheet PDFs and product parameters by MPN.
|
|
|
|
Uses DigiKey Product Information API v4 with OAuth2 client credentials.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
|
|
import httpx
|
|
|
|
from backend.config import settings
|
|
from backend.services.datasheet_finder import (
|
|
_alnum,
|
|
mpn_catalog_match,
|
|
mpn_matches,
|
|
mpn_query_variants,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# OAuth2 token cache
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_token_cache: dict[str, str | float] = {"access_token": "", "expires_at": 0.0}
|
|
|
|
_BASE_URLS = {
|
|
"production": "https://api.digikey.com",
|
|
"sandbox": "https://sandbox-api.digikey.com",
|
|
}
|
|
|
|
|
|
async def _get_access_token() -> str:
|
|
"""Get a DigiKey OAuth2 access token, refreshing if expired."""
|
|
now = time.time()
|
|
if _token_cache["access_token"] and float(_token_cache["expires_at"]) > now + 60:
|
|
return str(_token_cache["access_token"])
|
|
|
|
base = _BASE_URLS.get(settings.digikey_environment, _BASE_URLS["production"])
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
resp = await client.post(
|
|
f"{base}/v1/oauth2/token",
|
|
data={
|
|
"client_id": settings.digikey_client_id,
|
|
"client_secret": settings.digikey_client_secret,
|
|
"grant_type": "client_credentials",
|
|
},
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
|
|
_token_cache["access_token"] = data["access_token"]
|
|
_token_cache["expires_at"] = now + data.get("expires_in", 3600)
|
|
logger.info("DigiKey OAuth token refreshed (expires in %ds)", data.get("expires_in", 3600))
|
|
return str(_token_cache["access_token"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Product search
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _get_mpn(product: dict) -> str:
|
|
return product.get("ManufacturerProductNumber") or product.get("ManufacturerPartNumber") or ""
|
|
|
|
|
|
def _get_ds_url(product: dict) -> str:
|
|
url = product.get("DatasheetUrl") or product.get("PrimaryDatasheet") or ""
|
|
# DigiKey sometimes returns protocol-relative URLs
|
|
if url.startswith("//"):
|
|
url = "https:" + url
|
|
return url
|
|
|
|
|
|
async def _keyword_search(mpn: str) -> list[dict]:
|
|
"""Run a DigiKey keyword search and return the raw products list."""
|
|
base = _BASE_URLS.get(settings.digikey_environment, _BASE_URLS["production"])
|
|
token = await _get_access_token()
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {token}",
|
|
"X-DIGIKEY-Client-Id": settings.digikey_client_id,
|
|
"X-DIGIKEY-Locale-Site": settings.digikey_locale_site,
|
|
"X-DIGIKEY-Locale-Language": settings.digikey_locale_language,
|
|
"X-DIGIKEY-Locale-Currency": settings.digikey_locale_currency,
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
body = {
|
|
"Keywords": mpn,
|
|
"Limit": 5,
|
|
"Offset": 0,
|
|
"ExcludeMarketPlaceProducts": True,
|
|
}
|
|
|
|
async with httpx.AsyncClient(timeout=20) as client:
|
|
resp = await client.post(
|
|
f"{base}/products/v4/search/keyword",
|
|
headers=headers,
|
|
json=body,
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
|
|
return data.get("Products") or data.get("products") or []
|
|
|
|
|
|
def _find_product(mpn: str, products: list[dict]) -> dict | None:
|
|
"""Pick a DigiKey product for ``mpn``.
|
|
|
|
Prefers punctuation-insensitive equality, then packing suffixes, then a
|
|
longer orderable code that starts with the BOM MPN. Does not fall back
|
|
to ``products[0]``. Tries BOM spelling variants (underscore, reel, extra
|
|
description after an em dash).
|
|
"""
|
|
if not products:
|
|
return None
|
|
for query in mpn_query_variants(mpn):
|
|
hit = _find_product_one(query, products)
|
|
if hit:
|
|
return hit
|
|
return None
|
|
|
|
|
|
def _find_product_one(mpn: str, products: list[dict]) -> dict | None:
|
|
exact = None
|
|
loose = None
|
|
family = None
|
|
want = _alnum(mpn)
|
|
for product in products:
|
|
cand = _get_mpn(product)
|
|
if not cand:
|
|
continue
|
|
got = _alnum(cand)
|
|
if got == want:
|
|
exact = product
|
|
break
|
|
if loose is None and mpn_matches(mpn, cand):
|
|
loose = product
|
|
elif family is None and mpn_catalog_match(mpn, cand):
|
|
family = product
|
|
return exact or loose or family
|
|
|
|
|
|
async def _search_mpn(mpn: str) -> tuple[str | None, str | None]:
|
|
"""Search DigiKey; return (datasheet_url, catalog_mpn)."""
|
|
tried: set[str] = set()
|
|
for keyword in mpn_query_variants(mpn):
|
|
key = keyword.upper()
|
|
if key in tried:
|
|
continue
|
|
tried.add(key)
|
|
products = await _keyword_search(keyword)
|
|
product = _find_product(mpn, products)
|
|
if not product:
|
|
continue
|
|
url = _get_ds_url(product)
|
|
if url:
|
|
return url, _get_mpn(product) or None
|
|
return None, None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# PDF download + validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_PDF_MAGIC = b"%PDF-"
|
|
_MIN_PDF_SIZE = 5_000 # 5 KB — anything smaller is probably an error page
|
|
|
|
|
|
async def _download_pdf(url: str) -> bytes:
|
|
"""Download a PDF from a URL and validate it.
|
|
|
|
Raises ValueError if the file isn't a valid PDF or is too small.
|
|
Raises httpx.HTTPStatusError on 4xx/5xx responses.
|
|
"""
|
|
async with httpx.AsyncClient(timeout=15, follow_redirects=True) as client:
|
|
resp = await client.get(url)
|
|
resp.raise_for_status()
|
|
data = resp.content
|
|
|
|
if not data.startswith(_PDF_MAGIC):
|
|
raise ValueError("Downloaded file is not a valid PDF (bad magic bytes)")
|
|
|
|
if len(data) < _MIN_PDF_SIZE:
|
|
raise ValueError(f"PDF too small ({len(data)} bytes) — likely an error page")
|
|
|
|
return data
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Public API
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class DatasheetFetchResult:
|
|
"""Result of a datasheet fetch attempt."""
|
|
|
|
def __init__(
|
|
self,
|
|
mpn: str,
|
|
pdf_bytes: bytes | None = None,
|
|
error: str | None = None,
|
|
url: str | None = None,
|
|
catalog_mpn: str | None = None,
|
|
):
|
|
self.mpn = mpn
|
|
self.pdf_bytes = pdf_bytes
|
|
self.error = error
|
|
self.url = url
|
|
self.catalog_mpn = catalog_mpn
|
|
|
|
@property
|
|
def ok(self) -> bool:
|
|
return self.pdf_bytes is not None
|
|
|
|
|
|
async def fetch_datasheet(mpn: str) -> DatasheetFetchResult:
|
|
"""Fetch a datasheet PDF for the given MPN from DigiKey.
|
|
|
|
Returns a DatasheetFetchResult with either pdf_bytes or an error message.
|
|
The `url` field is set whenever DigiKey returns a datasheet link, even if
|
|
the PDF download itself fails.
|
|
Never raises — all errors are captured in the result.
|
|
"""
|
|
if not settings.use_digikey:
|
|
return DatasheetFetchResult(mpn, error="DigiKey API not configured")
|
|
|
|
try:
|
|
url, catalog_mpn = await _search_mpn(mpn)
|
|
except httpx.HTTPStatusError as e:
|
|
logger.warning("DigiKey search failed for %s: %s", mpn, e)
|
|
return DatasheetFetchResult(mpn, error=f"DigiKey search failed ({e.response.status_code})")
|
|
except Exception as e:
|
|
msg = str(e) or type(e).__name__
|
|
logger.warning("DigiKey search error for %s: %s", mpn, msg)
|
|
return DatasheetFetchResult(mpn, error=f"DigiKey search error: {msg}")
|
|
|
|
if not url:
|
|
return DatasheetFetchResult(mpn, error="No datasheet found on DigiKey")
|
|
|
|
try:
|
|
pdf_bytes = await _download_pdf(url)
|
|
except httpx.HTTPStatusError as e:
|
|
logger.warning("Datasheet download blocked for %s (%s): %s", mpn, url, e)
|
|
return DatasheetFetchResult(
|
|
mpn, error=f"Download blocked ({e.response.status_code})", url=url,
|
|
catalog_mpn=catalog_mpn,
|
|
)
|
|
except ValueError as e:
|
|
logger.warning("Invalid PDF for %s (%s): %s", mpn, url, e)
|
|
return DatasheetFetchResult(mpn, error=str(e), url=url, catalog_mpn=catalog_mpn)
|
|
except httpx.TimeoutException:
|
|
logger.warning("Datasheet download timed out for %s (%s)", mpn, url)
|
|
return DatasheetFetchResult(mpn, error="Download timed out", url=url, catalog_mpn=catalog_mpn)
|
|
except Exception as e:
|
|
msg = str(e) or type(e).__name__
|
|
logger.warning("Datasheet download failed for %s (%s): %s", mpn, url, msg)
|
|
return DatasheetFetchResult(
|
|
mpn, error=f"Download failed: {msg}", url=url, catalog_mpn=catalog_mpn,
|
|
)
|
|
|
|
logger.info("Fetched datasheet for %s (%d KB)", mpn, len(pdf_bytes) // 1024)
|
|
return DatasheetFetchResult(
|
|
mpn, pdf_bytes=pdf_bytes, url=url, catalog_mpn=catalog_mpn,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Product parameters
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class ProductParams:
|
|
"""Structured product parameters from a DigiKey search result."""
|
|
|
|
mpn: str
|
|
parameters: list[dict[str, str]] = field(default_factory=list) # [{"name": ..., "value": ...}]
|
|
category: str = ""
|
|
description: str = ""
|
|
|
|
|
|
class ParamsFetchResult:
|
|
"""Result of a product parameters fetch attempt."""
|
|
|
|
def __init__(self, mpn: str, params: ProductParams | None = None, error: str | None = None):
|
|
self.mpn = mpn
|
|
self.params = params
|
|
self.error = error
|
|
|
|
@property
|
|
def ok(self) -> bool:
|
|
return self.params is not None
|
|
|
|
|
|
def _parse_product_params(mpn: str, product: dict) -> ProductParams:
|
|
"""Extract structured parameters from a DigiKey product dict."""
|
|
raw_params = product.get("Parameters") or product.get("parameters") or []
|
|
parameters = []
|
|
for p in raw_params:
|
|
name = p.get("ParameterText") or p.get("parameterText") or ""
|
|
value = p.get("ValueText") or p.get("valueText") or ""
|
|
if name and value and value != "-":
|
|
parameters.append({"name": name, "value": value})
|
|
|
|
# Category
|
|
cat = product.get("Category") or product.get("category") or {}
|
|
category = cat.get("Name") or cat.get("name") or ""
|
|
|
|
# Description
|
|
desc_obj = product.get("Description") or product.get("description") or {}
|
|
if isinstance(desc_obj, str):
|
|
description = desc_obj
|
|
else:
|
|
description = (
|
|
desc_obj.get("ProductDescription")
|
|
or desc_obj.get("productDescription")
|
|
or desc_obj.get("DetailedDescription")
|
|
or desc_obj.get("detailedDescription")
|
|
or ""
|
|
)
|
|
|
|
return ProductParams(
|
|
mpn=mpn,
|
|
parameters=parameters,
|
|
category=category,
|
|
description=description,
|
|
)
|
|
|
|
|
|
async def fetch_params(mpn: str) -> ParamsFetchResult:
|
|
"""Fetch DigiKey product parameters for the given MPN.
|
|
|
|
Returns structured parameter data (no PDF download needed).
|
|
Never raises — all errors are captured in the result.
|
|
"""
|
|
if not settings.use_digikey:
|
|
return ParamsFetchResult(mpn, error="DigiKey API not configured")
|
|
|
|
try:
|
|
products: list[dict] = []
|
|
tried: set[str] = set()
|
|
for keyword in mpn_query_variants(mpn):
|
|
key = keyword.upper()
|
|
if key in tried:
|
|
continue
|
|
tried.add(key)
|
|
products = await _keyword_search(keyword)
|
|
if _find_product(mpn, products):
|
|
break
|
|
except httpx.HTTPStatusError as e:
|
|
logger.warning("DigiKey search failed for %s: %s", mpn, e)
|
|
return ParamsFetchResult(mpn, error=f"DigiKey search failed ({e.response.status_code})")
|
|
except Exception as e:
|
|
msg = str(e) or type(e).__name__
|
|
logger.warning("DigiKey search error for %s: %s", mpn, msg)
|
|
return ParamsFetchResult(mpn, error=f"DigiKey search error: {msg}")
|
|
|
|
product = _find_product(mpn, products)
|
|
if not product:
|
|
return ParamsFetchResult(mpn, error="No results found on DigiKey")
|
|
|
|
params = _parse_product_params(mpn, product)
|
|
if not params.parameters:
|
|
return ParamsFetchResult(mpn, error="No parameters available on DigiKey")
|
|
|
|
logger.info("Fetched %d params for %s (category: %s)", len(params.parameters), mpn, params.category)
|
|
return ParamsFetchResult(mpn, params=params)
|