File size: 2,499 Bytes
0ee2df7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 | """
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}
|