How to Scrape Amazon

"""
KnoxProxy scraper for Amazon product pages.
Parses Amazon product pages by ASIN (the ten-character product ID in
every /dp/{asin} URL) to extract title, price, rating, review count,
and availability. Includes a batch function for scraping a list of
ASINs without tripping Amazon's WAF, and CAPTCHA-aware retry logic.
"""
import random
import time

import requests
from bs4 import BeautifulSoup

PROXY_HOST = "proxy.knoxproxy.com"
PROXY_PORT = 10000  # residential rotating
PROXY_USER = "your_username"
PROXY_PASS = "your_password"

HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
        "(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36"
    ),
    "Accept-Language": "en-US,en;q=0.9",
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
}


def build_proxy_url(country: str = "us") -> str:
    user = f"{PROXY_USER}-country-{country}"
    return f"http://{user}:{PROXY_PASS}@{PROXY_HOST}:{PROXY_PORT}"


def fetch_amazon_product(asin: str, country: str = "us", max_retries: int = 3) -> dict:
    """Fetch and parse a single Amazon product page by ASIN, retrying
    with a fresh proxy IP whenever the response is a CAPTCHA page."""
    url = f"https://www.amazon.com/dp/{asin}"
    proxy_url = build_proxy_url(country)
    proxies = {"http": proxy_url, "https": proxy_url}

    for attempt in range(1, max_retries + 1):
        try:
            resp = requests.get(url, headers=HEADERS, proxies=proxies, timeout=20)
            if resp.status_code == 200 and "captcha" not in resp.text.lower():
                return parse_amazon_page(resp.text, asin)
            print(f"Attempt {attempt}: status {resp.status_code}, retrying with a new IP")
        except requests.exceptions.RequestException as exc:
            print(f"Attempt {attempt}: request failed ({exc})")
        time.sleep(random.uniform(2, 5))

    raise RuntimeError(f"Failed to fetch Amazon ASIN {asin} after {max_retries} attempts")


def parse_amazon_page(html: str, asin: str) -> dict:
    soup = BeautifulSoup(html, "html.parser")
    title_el = soup.select_one("#productTitle")
    price_el = soup.select_one(".a-price .a-offscreen")
    rating_el = soup.select_one("span.a-icon-alt")
    review_count_el = soup.select_one("#acrCustomerReviewText")
    availability_el = soup.select_one("#availability span")

    return {
        "asin": asin,
        "title": title_el.get_text(strip=True) if title_el else None,
        "price": price_el.get_text(strip=True) if price_el else None,
        "rating": rating_el.get_text(strip=True) if rating_el else None,
        "review_count": review_count_el.get_text(strip=True) if review_count_el else None,
        "availability": availability_el.get_text(strip=True) if availability_el else None,
    }


def scrape_asin_batch(asins: list[str], country: str = "us") -> list[dict]:
    """Batch scrape a list of Amazon ASINs one at a time with randomized
    pacing and a fresh proxy IP per ASIN. This is the pattern that keeps
    an amazon asin batch scraping tool under Amazon's rate-limit radar
    instead of tripping a CAPTCHA wall on the fourth or fifth ASIN."""
    results = []
    for asin in asins:
        try:
            results.append(fetch_amazon_product(asin, country=country))
        except RuntimeError as exc:
            print(f"Skipping {asin}: {exc}")
            results.append({"asin": asin, "error": str(exc)})
        time.sleep(random.uniform(2, 5))  # pace requests between ASINs
    return results


if __name__ == "__main__":
    asin_list = ["B0BSHF7WHW", "B08N5WRWNW", "B0C6KKQ7ND"]
    for row in scrape_asin_batch(asin_list):
        print(row)

"""
KnoxProxy price-history tracker for Amazon products.
Runs the ASIN scraper on a schedule and writes each result to a local
SQLite table so price drops can be charted over days or weeks. Reuses
fetch_amazon_product() from amazon_scraper.py so parsing logic lives
in one place.
"""
import random
import sqlite3
import time
from datetime import datetime, timezone

from amazon_scraper import fetch_amazon_product

DB_PATH = "amazon_price_history.db"


def init_db(path: str = DB_PATH) -> None:
    conn = sqlite3.connect(path)
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS price_snapshots (
            asin TEXT NOT NULL,
            title TEXT,
            price TEXT,
            rating TEXT,
            review_count TEXT,
            availability TEXT,
            checked_at TEXT NOT NULL
        )
        """
    )
    conn.commit()
    conn.close()


def save_snapshot(row: dict, path: str = DB_PATH) -> None:
    conn = sqlite3.connect(path)
    conn.execute(
        """
        INSERT INTO price_snapshots
            (asin, title, price, rating, review_count, availability, checked_at)
        VALUES (?, ?, ?, ?, ?, ?, ?)
        """,
        (
            row.get("asin"),
            row.get("title"),
            row.get("price"),
            row.get("rating"),
            row.get("review_count"),
            row.get("availability"),
            datetime.now(timezone.utc).isoformat(),
        ),
    )
    conn.commit()
    conn.close()


def track_asins(asins: list[str], country: str = "us") -> None:
    """Scrape each ASIN and store a timestamped row, building the
    history an amazon price scraper dashboard reads from later. Run
    this on a daily cron job for ongoing amazon price tracking."""
    init_db()
    for asin in asins:
        try:
            row = fetch_amazon_product(asin, country=country)
            save_snapshot(row)
            print(f"Saved snapshot for {asin}: {row.get('price')}")
        except RuntimeError as exc:
            print(f"Skipping {asin} this run: {exc}")
        time.sleep(random.uniform(2, 5))


if __name__ == "__main__":
    track_asins(["B0BSHF7WHW", "B08N5WRWNW"])

{
  "asin": "B0BSHF7WHW",
  "title": "Echo Dot (5th Gen) Smart Speaker with Alexa",
  "price": "$29.99",
  "rating": "4.7 out of 5 stars",
  "review_count": "89,412",
  "availability": "In Stock"
}