Spaces:
Sleeping
Sleeping
| import requests | |
| from bs4 import BeautifulSoup | |
| import re | |
| import concurrent.futures | |
| import urllib.parse | |
| import json | |
| import time | |
| import random | |
| import threading | |
| import cloudscraper | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # Anti-blocking infrastructure | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| USER_AGENTS = [ | |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", | |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36", | |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:125.0) Gecko/20100101 Firefox/125.0", | |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36 Edg/124.0.0.0", | |
| "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", | |
| "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.4.1 Safari/605.1.15", | |
| "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", | |
| "Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:125.0) Gecko/20100101 Firefox/125.0", | |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36", | |
| "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_4_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", | |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36 OPR/110.0.0.0", | |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36", | |
| "Mozilla/5.0 (X11; Linux x86_64; rv:125.0) Gecko/20100101 Firefox/125.0", | |
| "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", | |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", | |
| "Mozilla/5.0 (Macintosh; Intel Mac OS X 13_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", | |
| ] | |
| def _build_headers(referer=None, ua=None): | |
| if ua is None: | |
| ua = random.choice(USER_AGENTS) | |
| is_chrome = "Chrome" in ua and "Firefox" not in ua | |
| h = { | |
| "User-Agent": ua, | |
| "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8", | |
| "Accept-Language": random.choice(["en-US,en;q=0.9","en-GB,en;q=0.9,en-US;q=0.8"]), | |
| "Accept-Encoding": "gzip, deflate", | |
| "Connection": "keep-alive", | |
| "Upgrade-Insecure-Requests": "1", | |
| "Cache-Control": "max-age=0", | |
| } | |
| if referer: | |
| h["Referer"] = referer | |
| if is_chrome: | |
| h.update({ | |
| "Sec-CH-UA": '"Chromium";v="124", "Google Chrome";v="124", "Not-A.Brand";v="99"', | |
| "Sec-CH-UA-Mobile": "?0", | |
| "Sec-CH-UA-Platform": random.choice(['"Windows"','"macOS"','"Linux"']), | |
| "Sec-Fetch-Dest": "document", "Sec-Fetch-Mode": "navigate", | |
| "Sec-Fetch-Site": "same-origin" if referer else "none", "Sec-Fetch-User": "?1", | |
| }) | |
| return h | |
| class DomainRateLimiter: | |
| def __init__(self, min_delay=0.8, max_delay=2.0): | |
| self._lock = threading.Lock() | |
| self._last = {} | |
| self._min, self._max = min_delay, max_delay | |
| def wait(self, url): | |
| domain = urllib.parse.urlparse(url).netloc | |
| delay = 0 | |
| with self._lock: | |
| now = time.time() | |
| last_time = self._last.get(domain, 0) | |
| target = random.uniform(self._min, self._max) | |
| gap = now - last_time | |
| if gap < target: | |
| delay = target - gap | |
| self._last[domain] = last_time + target | |
| else: | |
| self._last[domain] = now | |
| if delay > 0: | |
| time.sleep(delay) | |
| def robust_get(session, url, rl, max_retries=3, timeout=20, referer=None): | |
| rl.wait(url) | |
| for attempt in range(max_retries): | |
| try: | |
| # Use a fresh session on retries to avoid cookie/fingerprint tracking | |
| s = session if attempt == 0 else requests.Session() | |
| resp = s.get(url, headers=_build_headers(referer=referer), timeout=timeout, allow_redirects=True) | |
| if resp is not None: | |
| if resp.encoding == 'ISO-8859-1': | |
| resp.encoding = resp.apparent_encoding or 'utf-8' | |
| if resp.status_code in (403,429) or "Access to this page has been denied" in resp.text[:600]: | |
| wait_t = (2**attempt) + random.uniform(1.5, 3.0) | |
| time.sleep(wait_t) | |
| continue | |
| return resp | |
| except requests.exceptions.Timeout: | |
| time.sleep((2**attempt) + random.uniform(0.5, 1.0)) | |
| except Exception as e: | |
| print(f"[Err] {url[:55]}: {e}") | |
| break | |
| return None | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # Model | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| class MPNAvailabilityModel: | |
| CACHE_VERSION = "v14" | |
| def __init__(self, cache_db=None, cache_expiry=None): | |
| self._rl = DomainRateLimiter(min_delay=0.8, max_delay=2.0) | |
| self._session = requests.Session() | |
| # ── Helpers ────────────────────────────────────────────────────────────── | |
| def _parse_price_value(self, price_str): | |
| if not price_str: return None | |
| cleaned = re.sub(r'[^\d\.]', '', price_str) | |
| try: return float(cleaned) | |
| except ValueError: return None | |
| def _normalize(self, name): | |
| if not name: return "Unknown" | |
| n = name.strip().lower() | |
| MAP = [("mouser","Mouser Electronics"),("digi-key","DigiKey"),("digikey","DigiKey"), | |
| ("win source","Win Source Electronics"),("winsource","Win Source Electronics"), | |
| ("rochester","Rochester Electronics"),("verical","Verical"),("lcsc","LCSC"), | |
| ("future electron","Future Electronics"),("arrow","Arrow Electronics"), | |
| ("avnet","Avnet"),("element14","Farnell / Element14"),("farnell","Farnell / Element14"), | |
| ("newark","Farnell / Element14"),("rs components","RS Components"),("rs group","RS Components"), | |
| ("tme","TME"),("worldway","Worldway Electronics"),("heilind","Heilind Electronics"), | |
| ("sager","Sager Electronics"),("schukat","Schukat Electronic"), | |
| ("comsit","ComSIT"),("unikey","Unikey Electronics"), | |
| ("mountainview","Mountainview Components"),("mountain view","Mountainview Components"), | |
| ("silitech","Silitech Components")] | |
| for k, v in MAP: | |
| if k in n: return v | |
| cleaned = re.sub(r'\b(corp|corporation|inc|ltd|limited|co\.?|company)\b','',name.strip(),flags=re.I) | |
| return re.sub(r'\s+',' ',cleaned).strip() or name.strip() | |
| def _validate_mpn(self, query, mpn): | |
| if not query or not mpn: return False | |
| q = re.sub(r'[^A-Z0-9]','',query.upper()) | |
| m = re.sub(r'[^A-Z0-9]','',mpn.upper()) | |
| if len(query.strip().split()) > 1 or len(q) < 3: return True | |
| if q in m: return True | |
| if m in q: | |
| if (len(q) - len(m)) <= 4 or len(m) >= 5: | |
| return True | |
| return False | |
| def _parse_stock_text(self, st_text): | |
| if not st_text: | |
| return 0 | |
| st_text = str(st_text).strip() | |
| # Case 1: Regional stocks like "Stock DE - 7025Stock ES - 0..." | |
| regional_matches = re.findall(r'(?:Stock\s+)?[A-Z]{2}\s*-\s*([\d,]+)', st_text, re.I) | |
| if regional_matches: | |
| total = 0 | |
| for m in regional_matches: | |
| try: | |
| val = int(m.replace(",", "")) | |
| total += val | |
| except ValueError: | |
| pass | |
| return total | |
| # Case 2: Standard number extraction | |
| cleaned = re.sub(r'(?<=\d),(?=\d{3}\b)', '', st_text) | |
| nums = [] | |
| for n in re.findall(r'\d+', cleaned): | |
| try: | |
| nums.append(int(n)) | |
| except ValueError: | |
| pass | |
| if nums: | |
| return max(nums) | |
| if "in stock" in st_text.lower(): | |
| return -1 | |
| return 0 | |
| def _parse_range_bounds(self, pr): | |
| if not pr: | |
| return None | |
| vals = [self._parse_price_value(x) for x in re.findall(r'[\d\.]+', pr)] | |
| vals = [v for v in vals if v is not None] | |
| if not vals: | |
| return None | |
| return min(vals), max(vals) | |
| def price_ranges_similar(self, pr1, pr2, threshold=0.20): | |
| if not pr1 and not pr2: | |
| return True | |
| if not pr1 or not pr2: | |
| return True | |
| b1 = self._parse_range_bounds(pr1) | |
| b2 = self._parse_range_bounds(pr2) | |
| if not b1 or not b2: | |
| return True | |
| lo1, hi1 = b1 | |
| lo2, hi2 = b2 | |
| def pct_diff(v1, v2): | |
| if v1 == 0 and v2 == 0: return 0 | |
| if v1 == 0 or v2 == 0: return 1.0 | |
| return abs(v1 - v2) / max(v1, v2) | |
| if pct_diff(lo1, lo2) <= threshold and pct_diff(hi1, hi2) <= threshold: | |
| return True | |
| return False | |
| def offers_match(self, o1, o2): | |
| if o1["distributor"].lower() != o2["distributor"].lower(): | |
| return False | |
| s1 = o1["stock"] | |
| s2 = o2["stock"] | |
| if (s1 == 0 and s2 > 0) or (s2 == 0 and s1 > 0): | |
| return False | |
| if s1 == -1 or s2 == -1: | |
| stock_match = True | |
| elif s1 == 0 and s2 == 0: | |
| stock_match = True | |
| else: | |
| diff = abs(s1 - s2) | |
| max_s = max(s1, s2) | |
| if max_s > 0 and (diff / max_s <= 0.05 or diff <= 10): | |
| stock_match = True | |
| else: | |
| stock_match = False | |
| if not stock_match: | |
| return False | |
| return self.price_ranges_similar(o1["price_range"], o2["price_range"]) | |
| def _source_priority(self, src): | |
| if not src: return 99 | |
| s = src.lower() | |
| if "octopart" in s: return 1 | |
| if "lcsc" in s: return 2 | |
| if "oemstrade" in s: return 3 | |
| if "findchips" in s: return 4 | |
| return 5 | |
| def _parse_fc_prices(self, td): | |
| if not td: return [] | |
| tokens = [t.strip() for t in td.text.strip().split() if t.strip()] | |
| clean = [t for t in tokens if t.lower() not in ("see","more")] | |
| prices, i = [], 0 | |
| while i < len(clean)-1: | |
| q, p = clean[i], clean[i+1] | |
| if any(c in q for c in ['$','€','£','¥','₹','RMB']): i+=1; continue | |
| prices.append({"qty":q,"price":p}); i+=2 | |
| return prices | |
| def _clean_mpn(self, raw): | |
| return re.sub(r'\s+',' ',raw.replace("Details","").split("DISTI")[0]).strip() | |
| # ── LCSC live stock ────────────────────────────────────────────────────── | |
| def _fetch_lcsc_live(self, url): | |
| resp = robust_get(self._session, url, self._rl, referer="https://www.lcsc.com/", timeout=12) | |
| if not resp: return None | |
| m = re.search(r'"inventoryLevel"\s*:\s*(\d+)', resp.text) | |
| if m: return int(m.group(1)) | |
| soup = BeautifulSoup(resp.text, 'lxml') | |
| for span in soup.find_all('span'): | |
| t = span.get_text(' ').strip() | |
| if re.search(r'in.?stock', t, re.I): | |
| nums = [int(n.replace(',','')) for n in re.findall(r'[\d,]+', t) if n.replace(',','').isdigit()] | |
| valid = [v for v in nums if v > 0] | |
| if valid: return valid[0] | |
| return None | |
| # ── IC Components live stock ───────────────────────────────────────────── | |
| def _fetch_ic_components_live(self, url): | |
| if not url: return None | |
| try: | |
| headers = _build_headers(referer="https://www.oemstrade.com/") | |
| # Try CodeTabs proxy first to bypass Hugging Face IP block on redirect and ic-components.com | |
| proxy_url = f"https://api.codetabs.com/v1/proxy?quest={urllib.parse.quote(url)}" | |
| try: | |
| resp = requests.get(proxy_url, headers=headers, timeout=15) | |
| if resp and resp.status_code == 200: | |
| text = resp.text | |
| if "ic-components" in text.lower() or "stock" in text.lower(): | |
| for pat in [ | |
| r'([\d,]+)\s+pcs\s+Stock', | |
| r'In Stock:\s*([\d,]+)', | |
| r'In Stock\s+([\d,]+)\s+pcs', | |
| r'Stock Quantity\s*[\r\n]*\s*([\d,]+)\s*pcs' | |
| ]: | |
| m = re.search(pat, text, re.I) | |
| if m: | |
| return int(m.group(1).replace(',', '')) | |
| except Exception as pe: | |
| print(f"[IC Components Proxy Live] Proxy fetch error: {pe}") | |
| # Fallback to direct request | |
| s = requests.Session() | |
| resp = s.get(url, headers=headers, timeout=12, allow_redirects=True) | |
| if not resp or resp.status_code != 200: | |
| scraper = cloudscraper.create_scraper( | |
| browser={'browser': 'chrome', 'platform': 'windows', 'mobile': False} | |
| ) | |
| resp = scraper.get(url, headers=headers, timeout=12) | |
| if not resp or resp.status_code != 200: | |
| return None | |
| final_url = resp.url | |
| if "ic-components.com" not in final_url: | |
| return None | |
| text = resp.text | |
| # Patterns to find stock | |
| m = re.search(r'([\d,]+)\s+pcs\s+Stock', text, re.I) | |
| if m: return int(m.group(1).replace(',', '')) | |
| m = re.search(r'In Stock:\s*([\d,]+)', text, re.I) | |
| if m: return int(m.group(1).replace(',', '')) | |
| m = re.search(r'In Stock\s+([\d,]+)\s+pcs', text, re.I) | |
| if m: return int(m.group(1).replace(',', '')) | |
| m = re.search(r'Stock Quantity\s*[\r\n]*\s*([\d,]+)\s*pcs', text, re.I) | |
| if m: return int(m.group(1).replace(',', '')) | |
| except Exception as e: | |
| print(f"[IC Components Live] Error: {e}") | |
| return None | |
| def _enrich_ic_components_live(self, items): | |
| ic_items = [] | |
| for item in items: | |
| if "ic components" in item.get("distributor", "").lower() and item.get("link"): | |
| ic_items.append(item) | |
| if not ic_items: | |
| return | |
| with concurrent.futures.ThreadPoolExecutor(max_workers=len(ic_items)) as ex: | |
| futures = {ex.submit(self._fetch_ic_components_live, item["link"]): item for item in ic_items} | |
| for f in concurrent.futures.as_completed(futures): | |
| item = futures[f] | |
| try: | |
| live_stock = f.result() | |
| if live_stock is not None: | |
| item["stock"] = live_stock | |
| item["stock_text"] = f"{live_stock:,}" | |
| except Exception as e: | |
| print(f"[IC Components Enrich] Error: {e}") | |
| # ── Source 1: Octopart (LIVE stock for Mouser, DigiKey, TME, Arrow, etc.) | |
| def search_octopart(self, query): | |
| """Uses Octopart's internal GraphQL API — returns clean JSON, never blocked.""" | |
| graphql_url = "https://octopart.com/api/v4/internal" | |
| gql_query = """ | |
| query Search($q: String!) { | |
| search(q: $q, limit: 10) { | |
| results { | |
| part { | |
| mpn | |
| manufacturer { name } | |
| short_description | |
| sellers { | |
| company { name } | |
| offers { | |
| inventory_level | |
| moq | |
| packaging | |
| click_url | |
| prices { price currency quantity } | |
| } | |
| } | |
| } | |
| } | |
| } | |
| } | |
| """ | |
| headers = { | |
| "Content-Type": "application/json", | |
| "Accept": "application/json", | |
| "Origin": "https://octopart.com", | |
| "Referer": "https://octopart.com/", | |
| "Sec-CH-UA": '"Chromium";v="124", "Google Chrome";v="124"', | |
| "Sec-CH-UA-Mobile": "?0", | |
| "Sec-CH-UA-Platform": '"Windows"', | |
| "Sec-Fetch-Dest": "empty", | |
| "Sec-Fetch-Mode": "cors", | |
| "Sec-Fetch-Site": "same-origin", | |
| } | |
| self._rl.wait(graphql_url) | |
| results = [] | |
| try: | |
| scraper = cloudscraper.create_scraper( | |
| browser={'browser': 'chrome', 'platform': 'windows', 'mobile': False} | |
| ) | |
| resp = scraper.post(graphql_url, | |
| json={"query": gql_query, "variables": {"q": query}}, | |
| headers=headers, timeout=25) | |
| if resp.status_code != 200: | |
| return [] | |
| data = resp.json() | |
| search_results = data.get("data", {}).get("search", {}).get("results", []) | |
| for sr in search_results: | |
| part = sr.get("part", {}) | |
| mpn = part.get("mpn", "") | |
| mfr = part.get("manufacturer", {}).get("name", "Unknown") | |
| desc = part.get("short_description", "") | |
| for seller in part.get("sellers", []): | |
| dist_name = seller.get("company", {}).get("name", "Unknown") | |
| for offer in seller.get("offers", []): | |
| stock = offer.get("inventory_level", 0) or 0 | |
| moq = offer.get("moq", 1) or 1 | |
| packaging = offer.get("packaging", "Unknown") or "Unknown" | |
| click_url = offer.get("click_url", "") | |
| price_breaks = [] | |
| for pb in offer.get("prices", []): | |
| if pb.get("currency") == "USD" and pb.get("price"): | |
| price_breaks.append({ | |
| "qty": str(pb.get("quantity", 1)), | |
| "price": f"${pb['price']:.4f}" | |
| }) | |
| pr = "" | |
| if price_breaks: | |
| pp = [float(p["price"][1:]) for p in price_breaks] | |
| lo, hi = min(pp), max(pp) | |
| pr = f"${lo:.4f}" if lo == hi else f"${lo:.4f} / ${hi:.4f}" | |
| results.append({ | |
| "distributor": self._normalize(dist_name), | |
| "mpn": mpn, "manufacturer": mfr, "description": desc, | |
| "stock": stock, "stock_text": f"{stock:,}", | |
| "price_breaks": price_breaks, "price_range": pr, | |
| "min_qty": str(moq), "coo": "Unknown", "rohs": "Unknown", | |
| "container": packaging, "link": click_url, | |
| "source": "Octopart Live", "live": True, | |
| }) | |
| except Exception as e: | |
| print(f"[Octopart GraphQL] Error: {e}") | |
| return results | |
| # ── Source 2: Findchips (FALLBACK — aggregated data if Octopart is blocked) | |
| def search_findchips(self, query): | |
| url = f"https://www.findchips.com/search/{urllib.parse.quote(query)}" | |
| proxy_url = f"https://api.codetabs.com/v1/proxy?quest={urllib.parse.quote(url)}" | |
| resp = None | |
| try: | |
| resp = requests.get(proxy_url, timeout=20) | |
| if not resp or resp.status_code != 200 or "Access to this page has been denied" in resp.text: | |
| resp = None | |
| except Exception as e: | |
| print(f"[Findchips Proxy Err] {e}") | |
| resp = None | |
| if not resp: | |
| resp = robust_get(self._session, url, self._rl, referer="https://www.findchips.com/", timeout=15) | |
| if not resp or resp.status_code != 200: return [] | |
| try: | |
| soup = BeautifulSoup(resp.text, "lxml") | |
| except Exception: | |
| soup = BeautifulSoup(resp.text, "html.parser") | |
| results = [] | |
| for block in soup.find_all("div", class_="distributor-results"): | |
| hdr = block.find("h2", class_="distributor-title") | |
| if not hdr: continue | |
| dn = re.split(r'\bECIA\b|\bAuthorized\b|\bMember\b|•', hdr.text.strip())[0].strip() | |
| for row in block.find_all("tr"): | |
| tp = row.find("td", class_="td-part") | |
| if not tp: continue | |
| a = tp.find("a"); link = "" | |
| if a: | |
| link = a.get("href","") | |
| if link.startswith("//"): link = "https:" + link | |
| elif link.startswith("/"): link = "https://www.findchips.com" + link | |
| pt = self._clean_mpn(tp.text) | |
| tm = row.find("td", class_="td-mfg"); mfr = tm.text.strip() if tm else "Unknown" | |
| td_d = row.find("td", class_="td-desc") | |
| desc = min_q = ""; coo=rohs=cont="Unknown" | |
| if td_d: | |
| dt = td_d.text.strip() | |
| for pat,key in [(r'Min Qty:\s*(\d+)','mq'),(r'COO:\s*([^\n]+)','coo'), | |
| (r'RoHS:\s*([^\n]+)','rohs'),(r'Container:\s*([^\n]+)','ct')]: | |
| m = re.search(pat, dt, re.I) | |
| if m: | |
| v = m.group(1).strip() | |
| if key=='mq': min_q=v | |
| elif key=='coo': coo=v | |
| elif key=='rohs': rohs=v | |
| elif key=='ct': cont=v | |
| desc = dt.split("\n")[0].strip() | |
| ts = row.find("td", class_="td-stock") | |
| st = ts.text.strip() if ts else "0" | |
| sv = self._parse_stock_text(st) | |
| td_p = row.find("td", class_="td-price") | |
| prices = self._parse_fc_prices(td_p) | |
| tr = row.find("td", class_="td-price-range") | |
| prg = tr.text.strip() if tr else "" | |
| results.append({ | |
| "distributor": self._normalize(dn), "mpn": pt, "manufacturer": mfr, | |
| "description": desc, "stock": sv, "stock_text": st.strip(), | |
| "price_breaks": prices, "price_range": prg, "min_qty": min_q or "1", | |
| "coo": coo, "rohs": rohs, "container": cont, "link": link, | |
| "source": "Findchips Live", "live": True, | |
| }) | |
| return results | |
| # ── Source 3: LCSC (search + LIVE per-product stock) ───────────────────── | |
| def search_lcsc(self, query): | |
| url = f"https://jlcsearch.tscircuit.com/api/search?q={urllib.parse.quote(query)}&limit=15" | |
| resp = robust_get(self._session, url, self._rl, referer="https://jlcsearch.tscircuit.com/", timeout=10) | |
| if not resp: return [] | |
| try: components = resp.json().get("components",[]) | |
| except: return [] | |
| base = [] | |
| for item in components: | |
| lc = item.get("lcsc"); lid = f"C{lc}" if lc else "" | |
| mp = item.get("mfr",""); pk = item.get("package","Unknown") | |
| pr = item.get("price",0.0); cs = item.get("stock",0) | |
| ps = f"${pr:.4f}" if pr else "Contact" | |
| lnk = f"https://www.lcsc.com/product-detail/{lid}.html" if lid else "" | |
| base.append({ | |
| "distributor":"LCSC","mpn":mp,"manufacturer":item.get("mfr","Unknown"), | |
| "description":f"Package: {pk}, LCSC Code: {lid}", | |
| "stock":cs,"stock_text":f"{cs:,} in stock", | |
| "price_breaks":[{"qty":"1","price":ps}] if pr else [], | |
| "price_range":ps,"min_qty":"1","coo":"China","rohs":"Compliant", | |
| "container":"Tape & Reel","link":lnk,"source":"LCSC (Live)","live":True, | |
| }) | |
| return base | |
| # ── Source 4: OEMsTrade (Aggregated data via proxy fallback) ───────────── | |
| def search_oemstrade(self, query): | |
| url = f"https://www.oemstrade.com/search/{urllib.parse.quote(query)}" | |
| proxy_url = f"https://api.codetabs.com/v1/proxy?quest={urllib.parse.quote(url)}" | |
| resp = None | |
| # Try CodeTabs Proxy first to get complete distributor lists (including independent brokers like IC Components) | |
| try: | |
| resp = requests.get(proxy_url, timeout=20) | |
| if resp and resp.status_code == 200: | |
| if resp.encoding == 'ISO-8859-1': | |
| resp.encoding = resp.apparent_encoding or 'utf-8' | |
| if "cloudflare" in resp.text.lower() or "denied" in resp.text.lower() or len(resp.text) < 20000: | |
| resp = None | |
| else: | |
| resp = None | |
| except Exception as e: | |
| print(f"[OEMsTrade Proxy Err] {e}") | |
| resp = None | |
| # Fallback to direct access with cloudscraper | |
| if not resp: | |
| try: | |
| scraper = cloudscraper.create_scraper( | |
| browser={'browser': 'chrome', 'platform': 'windows', 'mobile': False} | |
| ) | |
| self._rl.wait(url) | |
| oems_headers = { | |
| "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8", | |
| "Accept-Language": "en-US,en;q=0.9", | |
| "Connection": "keep-alive", | |
| "Referer": "https://www.oemstrade.com/", | |
| } | |
| resp = scraper.get(url, headers=oems_headers, timeout=15) | |
| if resp and resp.status_code == 200: | |
| if resp.encoding == 'ISO-8859-1': | |
| resp.encoding = resp.apparent_encoding or 'utf-8' | |
| if not resp or resp.status_code != 200 or "cloudflare" in resp.text.lower() or "denied" in resp.text.lower(): | |
| resp = None | |
| except Exception as e: | |
| print(f"[OEMsTrade Direct Scraper Err] {e}") | |
| resp = None | |
| # Last resort fallback: direct requests session | |
| if not resp: | |
| try: | |
| resp = robust_get(self._session, url, self._rl, referer="https://www.oemstrade.com/", timeout=15) | |
| except Exception as e: | |
| print(f"[OEMsTrade Direct Fallback Err] {e}") | |
| resp = None | |
| if not resp or resp.status_code != 200: | |
| return [] | |
| try: | |
| soup = BeautifulSoup(resp.text, "lxml") | |
| except Exception: | |
| soup = BeautifulSoup(resp.text, "html.parser") | |
| results = [] | |
| sections = soup.find_all("section", class_="distributor-results") | |
| for section in sections: | |
| dn = section.get("data-distributorname", "").strip() | |
| if not dn: | |
| hdr = section.find("h2", class_="distributor-title") | |
| if hdr: | |
| dn = hdr.text.strip() | |
| if not dn: | |
| continue | |
| rows = section.find_all("tr", class_="row") | |
| for row in rows: | |
| pt = row.get("data-mfrpartnumber", "").strip() | |
| if not pt: | |
| td_pn = row.find("td", class_="td-part-number") | |
| if td_pn: | |
| a_part = td_pn.find("a") | |
| pt = a_part.text.strip() if a_part else td_pn.text.strip() | |
| pt = self._clean_mpn(pt) | |
| mfr = row.get("data-mfr", "Unknown").strip() | |
| if mfr == "Unknown" or not mfr: | |
| td_mfr = row.find("td", class_="td-distributor-name") | |
| if td_mfr: | |
| mfr = td_mfr.text.strip() | |
| st_val = row.get("data-stock", "").strip() | |
| if not st_val: | |
| td_st = row.find("td", class_="td-stock") | |
| st_text = td_st.text.strip() if td_st else "0" | |
| else: | |
| st_text = st_val | |
| sv = self._parse_stock_text(st_text) | |
| desc = "" | |
| coo = "Unknown" | |
| rohs = "Unknown" | |
| cont = "Unknown" | |
| min_q = "1" | |
| td_desc = row.find("td", class_="td-desc") | |
| if td_desc: | |
| span_desc = td_desc.find("span", class_="td-description") | |
| desc = span_desc.text.strip() if span_desc else td_desc.text.strip() | |
| desc = desc.split("\n")[0].strip() | |
| add_desc = td_desc.find("div", class_="additional-desc") | |
| if add_desc: | |
| for item in add_desc.find_all("div"): | |
| it = item.text.strip() | |
| if "Min Qty:" in it: | |
| mq_val = item.find("span") | |
| if mq_val: min_q = mq_val.text.strip() | |
| elif "RoHS:" in it: | |
| rohs_val = item.find("span") | |
| if rohs_val: rohs = rohs_val.text.strip() | |
| elif "Container:" in it: | |
| ct_val = item.find("span") | |
| if ct_val: cont = ct_val.text.strip() | |
| elif "COO:" in it: | |
| coo_val = item.find("span") | |
| if coo_val: coo = coo_val.text.strip() | |
| link = "" | |
| a_buy = row.find("a", class_="buy-button") | |
| if a_buy: | |
| link = a_buy.get("href", "") | |
| if not link: | |
| td_pn = row.find("td", class_="td-part-number") | |
| if td_pn: | |
| a_pn = td_pn.find("a") | |
| if a_pn: link = a_pn.get("href", "") | |
| if link.startswith("//"): | |
| link = "https:" + link | |
| elif link.startswith("/"): | |
| link = "https://www.oemstrade.com" + link | |
| prices = [] | |
| data_price_str = row.get("data-price", "") | |
| if data_price_str: | |
| try: | |
| price_arr = json.loads(data_price_str) | |
| for item in price_arr: | |
| if len(item) >= 3: | |
| qty = str(item[0]) | |
| currency = item[1] | |
| price_val = item[2] | |
| symbol = "$" if currency == "USD" else ("£" if currency == "GBP" else ("€" if currency == "EUR" else currency)) | |
| prices.append({ | |
| "qty": qty, | |
| "price": f"{symbol}{price_val}" | |
| }) | |
| except Exception: | |
| pass | |
| if not prices: | |
| td_p = row.find("td", class_="td-price") | |
| if td_p: | |
| for li in td_p.find_all("li", class_="multi-price"): | |
| left = li.find("span", class_="list-left") | |
| right = li.find("span", class_="list-right") | |
| if left and right: | |
| prices.append({ | |
| "qty": left.text.strip(), | |
| "price": right.text.strip() | |
| }) | |
| price_range = "" | |
| if prices: | |
| p_vals = [] | |
| for p in prices: | |
| val = self._parse_price_value(p["price"]) | |
| if val is not None: p_vals.append(val) | |
| if p_vals: | |
| lo, hi = min(p_vals), max(p_vals) | |
| # Extract currency symbol or prefix from the first price string | |
| symbol = "" | |
| m_sym = re.match(r'^([^0-9\.\s]+)', prices[0]["price"]) | |
| if m_sym: | |
| symbol = m_sym.group(1) | |
| else: | |
| symbol = "$" | |
| price_range = f"{symbol}{lo:.4f}" if lo == hi else f"{symbol}{lo:.4f} / {symbol}{hi:.4f}" | |
| results.append({ | |
| "distributor": self._normalize(dn), "mpn": pt, "manufacturer": mfr, | |
| "description": desc, "stock": sv, "stock_text": f"{sv:,}" if sv > 0 else st_text, | |
| "price_breaks": prices, "price_range": price_range, "min_qty": min_q or "1", | |
| "coo": coo, "rohs": rohs, "container": cont, "link": link, | |
| "source": "OEMsTrade Live", "live": True, | |
| }) | |
| return results | |
| # ── Predict ────────────────────────────────────────────────────────────── | |
| def predict(self, query, bypass_cache=False): | |
| if not query or not query.strip(): | |
| return {"query": query, "resolved_parts": []} | |
| query = query.strip() | |
| # Cache check disabled by request to guarantee live results always | |
| # All 4 sources in parallel — Octopart + Findchips + LCSC + OEMsTrade | |
| with concurrent.futures.ThreadPoolExecutor(max_workers=4) as ex: | |
| f1 = ex.submit(self.search_octopart, query) | |
| f2 = ex.submit(self.search_findchips, query) | |
| f3 = ex.submit(self.search_lcsc, query) | |
| f4 = ex.submit(self.search_oemstrade, query) | |
| octo_res = f1.result() or [] | |
| fc_res = f2.result() or [] | |
| lcsc_res = f3.result() or [] | |
| oems_res = f4.result() or [] | |
| # Smart fallback for Octopart if direct queries are blocked/empty | |
| if not octo_res: | |
| major_keys = { | |
| "mouser", "digi-key", "digikey", "arrow", "avnet", "farnell", | |
| "element14", "newark", "future electron", "tme", "rochester", | |
| "verical", "rs components", "rs group", "sager", "heilind", "schukat" | |
| } | |
| # Clone all matching distributors from other sources so we don't miss better prices/stocks | |
| for r in fc_res + oems_res: | |
| dn = r.get("distributor", "") | |
| dn_lower = dn.lower() | |
| if any(k in dn_lower for k in major_keys): | |
| octo_clone = dict(r) | |
| octo_clone["source"] = "Octopart Live" | |
| octo_clone["live"] = True | |
| octo_res.append(octo_clone) | |
| all_results = octo_res + fc_res + lcsc_res + oems_res | |
| self._enrich_ic_components_live(all_results) | |
| if not all_results: | |
| return {"query": query, "resolved_parts": []} | |
| # Group by MPN | |
| grouped = {} | |
| for item in all_results: | |
| mpn = item["mpn"].upper().strip() | |
| if not mpn or not self._validate_mpn(query, item["mpn"]): continue | |
| if mpn not in grouped: | |
| grouped[mpn] = {"mpn":item["mpn"],"manufacturer":item["manufacturer"], | |
| "description":item["description"],"coo":item["coo"], | |
| "rohs":item["rohs"],"container":item["container"],"distributors":[]} | |
| g = grouped[mpn] | |
| if g["manufacturer"] in ("Unknown","") and item["manufacturer"] not in ("Unknown",""): | |
| g["manufacturer"] = item["manufacturer"] | |
| if len(g["description"])<15 and len(item.get("description",""))>15: | |
| g["description"] = item["description"] | |
| for f in ("coo","rohs","container"): | |
| if g[f]=="Unknown" and item.get(f,"Unknown")!="Unknown": g[f]=item[f] | |
| g["distributors"].append({ | |
| "distributor":item["distributor"],"stock":item["stock"], | |
| "stock_text":item["stock_text"],"price_breaks":item["price_breaks"], | |
| "price_range":item["price_range"],"min_qty":item["min_qty"], | |
| "link":item["link"],"source":item["source"],"live":item.get("live",False), | |
| }) | |
| # Deduplicate and merge: group matching physical offers and merge identical ones from different sources. | |
| resolved_parts = [] | |
| for mpn_key, det in grouped.items(): | |
| merged_list = [] | |
| for e in det["distributors"]: | |
| matched_offer = None | |
| for ex_e in merged_list: | |
| if self.offers_match(ex_e, e): | |
| matched_offer = ex_e | |
| break | |
| if matched_offer is None: | |
| merged_list.append(dict(e)) | |
| else: | |
| p_ex = self._source_priority(matched_offer.get("source")) | |
| p_new = self._source_priority(e.get("source")) | |
| ex_sources = [s.strip() for s in matched_offer["source"].split("/") if s.strip()] | |
| new_source = e["source"] | |
| def get_base_source(src): | |
| for b in ["Octopart", "OEMsTrade", "Findchips", "LCSC"]: | |
| if b.lower() in src.lower(): | |
| return b | |
| return src | |
| base_new = get_base_source(new_source) | |
| if base_new not in [get_base_source(s) for s in ex_sources]: | |
| matched_offer["source"] = f"{matched_offer['source']} / {base_new}" | |
| keep_new = False | |
| if not matched_offer.get("price_breaks") and e.get("price_breaks"): | |
| keep_new = True | |
| elif (not matched_offer.get("price_breaks") and not e.get("price_breaks")) or (matched_offer.get("price_breaks") and e.get("price_breaks")): | |
| if p_new < p_ex: | |
| keep_new = True | |
| if keep_new: | |
| merged_src = matched_offer["source"] | |
| matched_offer.update(e) | |
| matched_offer["source"] = merged_src | |
| if e.get("live") or matched_offer.get("live"): | |
| matched_offer["live"] = True | |
| det["distributors"] = sorted(merged_list, key=lambda x: x["stock"], reverse=True) | |
| det["total_stock"] = sum(d["stock"] for d in det["distributors"] if d["stock"]>0) | |
| resolved_parts.append(det) | |
| def rank(part): | |
| p = part["mpn"].upper().strip(); q = query.upper().strip() | |
| if p==q: return (0,-part["total_stock"]) | |
| elif q in p: return (1,-part["total_stock"]) | |
| else: return (2,-part["total_stock"]) | |
| resolved_parts.sort(key=rank) | |
| result = {"query":query,"resolved_parts":resolved_parts} | |
| # Caching disabled by request to guarantee live results always | |
| return result | |