AnasSarkiz/muse-voice-remote

scripts/check-stock.py

#!/usr/bin/env python3
"""Fetch supplier inventory evidence; never equate HTTP failure with stock.

JLC search reports cached supplier inventory. Positive values are discovery
evidence only; final stock must be checked at the distributor before ordering.
"""
import concurrent.futures
import collections
import datetime
import json
import pathlib
import urllib.parse
import urllib.request

root = pathlib.Path(__file__).resolve().parents[1]
catalog = json.loads((root / "docs/component-catalog.json").read_text())
mpns = sorted({p["mpn"] for p in catalog if p.get("mpn")})

def lookup(item):
    mpn, supplier = item
    if supplier == "jlcpcb":
        url = "https://jlcsearch.tscircuit.com/api/search?" + urllib.parse.urlencode({"q":mpn,"limit":20})
    else:
        url = f"https://{supplier}search.tscircuit.com/components/list?" + urllib.parse.urlencode({"search":mpn,"json":"true"})
    record = {"mpn":mpn,"url":url,"checked_at":datetime.datetime.now(datetime.timezone.utc).isoformat(),
              "supplier":supplier,"status":"unverified","stock":None,"inventory_kind":"supplier search snapshot"}
    try:
        request = urllib.request.Request(url,headers={"User-Agent":"Muse-Voice-Remote-sourcing/1.0"})
        with urllib.request.urlopen(request,timeout=20) as response:
            data = json.load(response)
        # Match exact orderable MPN, not a similar search hit.
        matches = [p for p in data.get("components",[]) if str(p.get("mfr","")).upper()==mpn.upper()]
        record["meta"] = data.get("meta",{})
        record["exact_matches"] = matches
        quantities = [p["stock"] for p in matches if isinstance(p.get("stock"),(int,float))]
        if quantities:
            record["stock"] = max(quantities)
            record["status"] = "cached-stock-positive" if record["stock"]>0 else "cached-out-of-stock"
        elif not matches:
            record["status"] = "no-exact-match"
    except Exception as error:
        record["error"] = str(error)
    return record

with concurrent.futures.ThreadPoolExecutor(max_workers=6) as pool:
    evidence = list(pool.map(lookup,[(mpn,supplier) for mpn in mpns for supplier in ["digikey","mouser","jlcpcb"]]))
quantities = collections.Counter(p["mpn"] for p in catalog if p.get("mpn"))
positive_mpns = {p["mpn"] for p in evidence if p.get("stock") and p["stock"]>=quantities[p["mpn"]]
                 and not p.get("meta",{}).get("stale",False)
                 and any(not m.get("discontinued",False) and not m.get("marketplace",False) for m in p.get("exact_matches",[]))}
result = {"checked_at":datetime.datetime.now(datetime.timezone.utc).isoformat(),
          "all_parts_verified_in_stock":False,
          "all_populated_pcb_parts_have_positive_snapshot":set(mpns)<=positive_mpns,
          "unique_pcb_mpns":len(mpns),"populated_pcb_components":sum(quantities.values()),
          "required_quantities":dict(quantities),
          "note":"Cached inventory is not a purchase guarantee. Actual ER-TFT028A2-4 panel availability, protected pack/NTC and off-board speaker/motor/antenna qualification remain outside this populated PCB catalog.",
          "parts":evidence}
(root / "docs/stock-evidence.json").write_text(json.dumps(result,indent=2)+"\n")
print(json.dumps({"queried_supplier_records":len(evidence),"unique_pcb_mpns":len(mpns),"positive_mpns":len(positive_mpns),"cached_positive":sum(p["status"]=="cached-stock-positive" for p in evidence),
                  "unverified":sum(p["status"]=="unverified" for p in evidence)},indent=2))