eBay Price Tracker: How to Monitor Prices Automatically

from __future__ import annotations

import os
import random
import time
from dataclasses import dataclass

import requests

TIMEOUT = (10, 30)

DEFAULT_HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/123.0.0.0 Safari/537.36"
    ),
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    "Accept-Language": "en-US,en;q=0.9",
}


@dataclass
class FetchConfig:
    use_proxiesapi: bool = True
    proxiesapi_endpoint: str | None = None
    max_retries: int = 4
    min_sleep: float = 0.8
    max_sleep: float = 1.8


class Fetcher:
    def __init__(self, cfg: FetchConfig):
        self.cfg = cfg
        self.s = requests.Session()
        self.s.headers.update(DEFAULT_HEADERS)

    def _sleep_jitter(self):
        time.sleep(random.uniform(self.cfg.min_sleep, self.cfg.max_sleep))

    def get(self, url: str) -> str:
        last_err = None
        for attempt in range(1, self.cfg.max_retries + 1):
            try:
                self._sleep_jitter()

                proxies = None
                if self.cfg.use_proxiesapi and self.cfg.proxiesapi_endpoint:
                    proxies = {"http": self.cfg.proxiesapi_endpoint, "https": self.cfg.proxiesapi_endpoint}

                r = self.s.get(url, timeout=TIMEOUT, proxies=proxies)

                if r.status_code in (403, 429, 500, 502, 503, 504):
                    raise requests.HTTPError(f"HTTP {r.status_code} for {url}", response=r)

                r.raise_for_status()
                return r.text
            except Exception as e:
                last_err = e
                time.sleep(1.2 ** attempt)

        raise RuntimeError(f"Failed after retries: {url}") from last_err


fetcher = Fetcher(FetchConfig(
    use_proxiesapi=True,
    proxiesapi_endpoint=os.getenv("PROXIESAPI_PROXY_URL"),
))

import re
from bs4 import BeautifulSoup


def parse_money(text: str) -> tuple[float | None, str | None]:
    # Handles "$12.34" or "US $12.34"
    if not text:
        return None, None
    currency = "USD" if "$" in text else None
    m = re.search(r"(\d{1,3}(?:,\d{3})*(?:\.\d{2})?)", text)
    if not m:
        return None, currency
    value = float(m.group(1).replace(",", ""))
    return value, currency


def parse_search_results(html: str) -> list[dict]:
    soup = BeautifulSoup(html, "lxml")

    rows: list[dict] = []

    # Find anchors that look like listing links
    for a in soup.select('a[href*="/itm/"]'):
        href = a.get("href")
        if not href:
            continue

        url = href.split("?")[0]
        title = a.get_text(" ", strip=True) or None
        if not title or len(title) < 4:
            continue

        card = a.find_parent(["li", "div"])  # climb to container
        price_text = None
        if card:
            price_el = card.select_one("span.s-item__price, span[class*='price']")
            price_text = price_el.get_text(" ", strip=True) if price_el else None

        price_value, currency = parse_money(price_text or "")

        rows.append({
            "title": title,
            "url": url,
            "price_value": price_value,
            "price_currency": currency,
        })

    # de-dupe by URL
    dedup = {}
    for r in rows:
        dedup[r["url"]] = r

    return list(dedup.values())

from urllib.parse import urlparse, parse_qs, urlencode, urlunparse


def set_query_param(url: str, key: str, value: str) -> str:
    u = urlparse(url)
    q = parse_qs(u.query)
    q[key] = [value]
    new_query = urlencode(q, doseq=True)
    return urlunparse((u.scheme, u.netloc, u.path, u.params, new_query, u.fragment))


def crawl_search(url: str, pages: int = 2) -> list[dict]:
    out: list[dict] = []
    for p in range(1, pages + 1):
        page_url = set_query_param(url, "_pgn", str(p))
        html = fetcher.get(page_url)
        batch = parse_search_results(html)
        out.extend(batch)
        print("page", p, "batch", len(batch), "total", len(out))
    return out


watch_url = "https://www.ebay.com/sch/i.html?_nkw=ipad+mini+6"
listings = crawl_search(watch_url, pages=2)
print(listings[:3])

import sqlite3
from datetime import datetime, timezone


def init_db(conn: sqlite3.Connection):
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS prices (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            watch_url TEXT NOT NULL,
            listing_url TEXT NOT NULL,
            title TEXT,
            price_value REAL,
            price_currency TEXT,
            scraped_at TEXT NOT NULL
        );
        """
    )
    conn.execute("CREATE INDEX IF NOT EXISTS idx_prices_watch ON prices(watch_url);")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_prices_listing ON prices(listing_url);")


def insert_prices(conn: sqlite3.Connection, watch_url: str, rows: list[dict]):
    now = datetime.now(timezone.utc).isoformat()
    conn.executemany(
        """
        INSERT INTO prices (watch_url, listing_url, title, price_value, price_currency, scraped_at)
        VALUES (?, ?, ?, ?, ?, ?)
        """,
        [
            (
                watch_url,
                r.get("url"),
                r.get("title"),
                r.get("price_value"),
                r.get("price_currency"),
                now,
            )
            for r in rows
            if r.get("url")
        ],
    )
    conn.commit()


conn = sqlite3.connect("ebay_tracker.sqlite")
init_db(conn)
insert_prices(conn, watch_url, listings)
print("inserted", len(listings))

def lowest_price(rows: list[dict]) -> float | None:
    vals = [r["price_value"] for r in rows if isinstance(r.get("price_value"), (int, float))]
    return min(vals) if vals else None

current_low = lowest_price(listings)
print("current low", current_low)