AnasSarkiz/muse-voice-remote

scripts/audit-jlcpcb.py

#!/usr/bin/env python3
"""Audit the current BOM against JLCPCB's live assembly catalog.

Text search in a third-party index is insufficient proof: compare exact MPNs
and retain the official response, including quantity and manufacturer.
"""
import concurrent.futures
import csv
import datetime
import hashlib
import json
import pathlib
import subprocess

ROOT = pathlib.Path(__file__).resolve().parents[1]
ENDPOINT = "https://jlcpcb.com/api/overseas-pcb-order/v1/shoppingCart/smtGood/selectSmtComponentList/v2"
OUT = ROOT / "docs/jlcpcb-audit"
OUT.mkdir(parents=True, exist_ok=True)


def audit(row):
    mpn = row["MPN"]
    query = {"currentPage": 1, "pageSize": 100, "keyword": mpn,
             "searchSource": "search", "searchType": 2,
             "componentBrandList": [], "componentSpecificationList": [],
             "componentAttributeList": [], "paramList": []}
    response = subprocess.run(
        ["curl", "--silent", "--show-error", "--fail-with-body", "--max-time", "30",
         "-H", "Content-Type: application/json", "--data-binary", json.dumps(query), ENDPOINT],
        capture_output=True, text=True,
    )
    checked = datetime.datetime.now(datetime.timezone.utc).isoformat()
    filename = "".join(c if c.isalnum() or c in "-_." else "_" for c in mpn) + ".json"
    raw = response.stdout
    result = {"mpn": mpn, "quantity": int(row["Quantity"]),
              "references": row["References"].split(), "checked_at": checked,
              "source": ENDPOINT, "raw_response": "docs/jlcpcb-audit/" + filename,
              "raw_sha256": hashlib.sha256(raw.encode()).hexdigest(), "exact_matches": []}
    def retain_failed_lookup():
        # Never replace a verified catalog snapshot with a proxy error page.
        # Keep the failed attempt separately and make its failed status clear.
        previous = OUT / filename
        if previous.exists():
            result["retained_snapshot"] = {"file": str(previous.relative_to(ROOT)),
                "sha256": hashlib.sha256(previous.read_bytes()).hexdigest()}
        failed = OUT / (filename.removesuffix(".json") + ".failed-response.txt")
        failed.write_text(raw)
        result["raw_response"] = str(failed.relative_to(ROOT))
    if response.returncode:
        result["status"] = "lookup_failed"
        result["error"] = response.stderr.strip() or raw[:250]
        retain_failed_lookup()
        return result
    try:
        data = json.loads(raw)
        info = data["data"]["componentPageInfo"]
        if info.get("total") == 0 and info.get("list") is None:
            info["list"] = []
        if data["code"] != 200 or not isinstance(info["list"], list):
            raise ValueError("Unexpected JLCPCB catalog response")
        (OUT / filename).write_text(raw)
        result["total_search_matches"] = info["total"]
        for part in info["list"]:
            if str(part.get("componentModelEn", "")).strip().upper() != mpn.upper():
                continue
            result["exact_matches"].append({
                "lcsc": part["componentCode"], "mpn": part["componentModelEn"],
                "manufacturer": part.get("componentBrandEn"),
                "package": part.get("componentSpecificationEn"),
                "stock": part.get("stockCount"),
                "library_type": part.get("componentLibraryType"),
                "datasheet_url": part.get("dataManualOfficialLink") or part.get("dataManualUrl"),
                "assembly_component": part.get("assemblyComponentFlag"),
            })
        stocked = [p for p in result["exact_matches"]
                   if isinstance(p["stock"], int) and p["stock"] >= result["quantity"]]
        result["status"] = "stocked_exact_candidate" if stocked else "no_stocked_exact_match"
    except (KeyError, TypeError, ValueError) as error:
        result["status"] = "lookup_failed"
        result["error"] = str(error)
        retain_failed_lookup()
    return result


with (ROOT / "docs/procurement.csv").open() as fd:
    bom = list(csv.DictReader(fd))
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
    results = list(executor.map(audit, bom))
report = {"source": "Official JLCPCB live assembly catalog",
          "scope": "Exact current BOM candidates; not replacement approval or import validation",
          "parts": results}
(ROOT / "docs/jlcpcb-stock-audit.json").write_text(json.dumps(report, indent=2) + "\n")
for row in results:
    matches = ", ".join(f"{p['lcsc']} ({p['manufacturer']}, stock {p['stock']})"
                        for p in row["exact_matches"][:3])
    print(row["mpn"], row["status"], matches, flush=True)
print("Exact stocked MPNs:", sum(r["status"] == "stocked_exact_candidate" for r in results), "/", len(results))