| """ |
| Amazon search scraper across marketplaces. No API key. |
| """ |
| import time |
| import httpx |
| from bs4 import BeautifulSoup |
|
|
| DOMAINS = { |
| "US": "amazon.com", "UK": "amazon.co.uk", "DE": "amazon.de", "FR": "amazon.fr", |
| "CA": "amazon.ca", "AU": "amazon.com.au", "IN": "amazon.in", "JP": "amazon.co.jp", |
| "BR": "amazon.com.br", "MX": "amazon.com.mx", "ES": "amazon.es", "IT": "amazon.it", |
| } |
| HEADERS = {"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)"} |
|
|
|
|
| def run(params: dict) -> dict: |
| keyword = params.get("keyword") |
| if not keyword: |
| raise ValueError("keyword is required") |
| country = params.get("country", "US").upper() |
| max_results = int(params.get("max_results", 25)) |
| domain = DOMAINS.get(country, DOMAINS["US"]) |
|
|
| url = f"https://www.{domain}/s?k={keyword}" |
|
|
| last_err = None |
| for attempt in range(3): |
| try: |
| resp = httpx.get(url, headers=HEADERS, timeout=20.0) |
| resp.raise_for_status() |
| break |
| except Exception as e: |
| last_err = e |
| time.sleep(1.5 * (attempt + 1)) |
| else: |
| raise ValueError(f"Amazon request failed after retries: {last_err}") |
|
|
| soup = BeautifulSoup(resp.text, "html.parser") |
| cards = soup.select('div[data-component-type="s-search-result"]') |
|
|
| results = [] |
| for card in cards[:max_results]: |
| asin = card.get("data-asin") |
| title_el = card.select_one("h2 span") |
| price_whole = card.select_one(".a-price-whole") |
| price_frac = card.select_one(".a-price-fraction") |
| rating_el = card.select_one("span.a-icon-alt") |
| reviews_el = card.select_one("span.a-size-base.s-underline-text") |
| link_el = card.select_one("h2 a") |
|
|
| price = None |
| if price_whole: |
| price = f"{price_whole.get_text(strip=True)}{price_frac.get_text(strip=True) if price_frac else ''}" |
|
|
| results.append( |
| { |
| "asin": asin, |
| "title": title_el.get_text(strip=True) if title_el else None, |
| "price": price, |
| "rating": rating_el.get_text(strip=True) if rating_el else None, |
| "review_count": reviews_el.get_text(strip=True) if reviews_el else None, |
| "is_prime": bool(card.select_one("[aria-label='Amazon Prime']")), |
| "url": f"https://www.{domain}{link_el['href']}" if link_el and link_el.get("href") else None, |
| } |
| ) |
| return {"count": len(results), "results": results} |
|
|
