import binascii import json import re from pathlib import Path from typing import Optional from urllib.parse import urlencode import requests from selenium import webdriver import ee._utils from ee.part import PartDb, save_db from ee.xml import types _title_re = re.compile(r"^([^(]*)\( *[0-9]* *\) *") __all__ = [ "SeeedClient", "download_opls", ] class SeeedClient(object): def __init__(self, cache_dir: Path = None): self.cache = ee._utils.maybe_cache(cache_dir) self.driver: Optional[webdriver.Chrome] = None def get(self, url: str): cache_key = abs(binascii.crc32(url.encode("utf-8"))) cached = self.cache.lookup(cache_key) if cached: return cached if self.driver is None: options = webdriver.ChromeOptions() self.driver = webdriver.Chrome(chrome_options=options) self.driver.get(url) src = self.driver.page_source self.cache.save(cache_key, src) return src def _checksum_opl(tree): for table in tree.xpath("//*[contains(@class, 'fusion-opl-list')]"): rows = table.xpath(".//tr[contains(@class, 'f12')]") for r in rows: print(r) break def _get(obj, key): value = obj.get(key, None) if value is None: return "" return value.strip() def download_opls(out_dir: Path, cache_dir: Path): opls_types = ["SEEED", "HQCHIP"] s = requests.Session() from ee.money import get_default_context money = get_default_context() for opls_type in opls_types: page_offset = 1 page_length = 30 count = 0 cd = cache_dir / opls_type if cache_dir else cache_dir cache = ee._utils.maybe_cache(cd, ext="json") db = PartDb() while True: cache_key = "{}-{}".format(opls_type, page_offset) content = cache.lookup(cache_key) if content is None: form = { "page_offset": page_offset, "page_length": page_length, "keyword": "", "category": "", "type": opls_type, } query = { "guid": "A8D502008EB2E4FD3FE9CE5E0855E8E4", "appid": "en.pc.bazaar", } r = s.get("https://sapi.seeedstudio.com/fusion/opl/list", params=query, data=form) content = r.text cache.save(cache_key, content) obj = json.loads(content) supplier_uri = "http://purl.org/ee/supplier/seeed?{}".format(urlencode({"opl": opls_type})) parts = obj["data"]["list"] for p in parts: # print(json.dumps(p, indent=4)) mpn = _get(p, "mpn") sku = _get(p, "sku") datasheet = _get(p, "datasheet") desc = _get(p, "desc") package = _get(p, "package") if mpn is None and sku is None: continue ladder_price = p["ladder_price"] part = types.Part(references=types.ReferencesList(), price_breaks=types.PriceBreakList()) part_numbers = part.referencesProp.part_numberProp supplier_part_numbers = part.referencesProp.supplier_part_numberProp pbs = part.price_breaksProp.price_breakProp if desc: part.descriptionProp = desc uri_params = { "opl": opls_type, } if sku: uri_params["sku"] = sku if mpn: uri_params["mpn"] = mpn part.uriProp = "http://purl.org/ee/supplier/seeed?{}".format(urlencode(uri_params)) if mpn: part_numbers.append(types.PartNumber(mpn)) if sku: supplier_part_numbers.append(types.SupplierPartNumber(value=mpn, supplier=supplier_uri)) for item in ladder_price: price = money.parse(item["price"], currency="USD") amount = types.Amount(value=price.amount, currency=price.currency) pbs.append(types.PriceBreak(quantity=item["qty"], amount=amount)) db.add_entry(part, True) page_offset += 1 count += len(parts) if len(parts) != page_length: break print("Imported {} parts from Seeed's {} library".format(count, opls_type)) save_db(out_dir / "{}.xml".format(opls_type), db, sort=True)