`python-matcher/scripts/ebay_sold_scraper.py`

188 lines
"""
eBay Sold Listings Scraper
Fetch real sales history for handbag valuation and price charts
"""
import os
import sys
import requests
import time
from datetime import datetime, timedelta
from typing import List, Dict, Optional

# eBay Finding API (no auth required for sold listings search)
EBAY_FINDING_API = "https://svcs.ebay.com/services/search/FindingService/v1"
EBAY_APP_ID = os.environ.get("EBAY_APP_ID", "")  # Get free API key from eBay Developers

def search_sold_listings(
    query: str,
    category_id: Optional[str] = None,
    min_price: Optional[float] = None,
    max_price: Optional[float] = None,
    days_back: int = 90,
    max_results: int = 100
) -> List[Dict]:
    """
    Search eBay sold/completed listings

    Args:
        query: Search keywords (e.g., "Hermes Birkin 35")
        category_id: eBay category (e.g., "63852" for Women's Handbags)
        min_price: Minimum price filter
        max_price: Maximum price filter
        days_back: How many days back to search (max 90)
        max_results: Max results to return (max 100 per request)

    Returns:
        List of sold listings with price, date, condition
    """

    if not EBAY_APP_ID:
        raise ValueError("EBAY_APP_ID environment variable not set")

    # Calculate date range
    end_date = datetime.now()
    start_date = end_date - timedelta(days=days_back)

    params = {
        "OPERATION-NAME": "findCompletedItems",
        "SERVICE-VERSION": "1.0.0",
        "SECURITY-APPNAME": EBAY_APP_ID,
        "RESPONSE-DATA-FORMAT": "JSON",
        "REST-PAYLOAD": "",
        "keywords": query,
        "paginationInput.entriesPerPage": min(max_results, 100),
        "itemFilter(0).name": "SoldItemsOnly",
        "itemFilter(0).value": "true",
        "itemFilter(1).name": "EndTimeFrom",
        "itemFilter(1).value": start_date.strftime("%Y-%m-%dT%H:%M:%S.000Z"),
        "itemFilter(2).name": "EndTimeTo",
        "itemFilter(2).value": end_date.strftime("%Y-%m-%dT%H:%M:%S.000Z"),
    }

    # Add category filter
    filter_idx = 3
    if category_id:
        params[f"categoryId"] = category_id

    # Add price filters
    if min_price:
        params[f"itemFilter({filter_idx}).name"] = "MinPrice"
        params[f"itemFilter({filter_idx}).value"] = str(min_price)
        filter_idx += 1

    if max_price:
        params[f"itemFilter({filter_idx}).name"] = "MaxPrice"
        params[f"itemFilter({filter_idx}).value"] = str(max_price)
        filter_idx += 1

    # Make request
    response = requests.get(EBAY_FINDING_API, params=params, timeout=30)
    response.raise_for_status()

    data = response.json()

    # Parse results
    sold_listings = []

    try:
        search_result = data["findCompletedItemsResponse"][0]["searchResult"][0]

        if search_result["@count"] == "0":
            return []

        items = search_result["item"]

        for item in items:
            # Only include sold items (not just ended)
            selling_status = item["sellingStatus"][0]
            selling_state = selling_status["sellingState"][0]

            if selling_state != "EndedWithSales":
                continue

            listing = {
                "item_id": item["itemId"][0],
                "title": item["title"][0],
                "price_usd": float(selling_status["convertedCurrentPrice"][0]["__value__"]),
                "currency": selling_status["convertedCurrentPrice"][0]["@currencyId"],
                "condition": item.get("condition", [{}])[0].get("conditionDisplayName", ["Unknown"])[0],
                "end_time": item["listingInfo"][0]["endTime"][0],
                "location": item.get("location", [""])[0],
                "shipping_type": item.get("shippingInfo", [{}])[0].get("shippingType", [""])[0],
                "url": item["viewItemURL"][0],
                "image_url": item.get("galleryURL", [""])[0],
                "category_name": item.get("primaryCategory", [{}])[0].get("categoryName", [""])[0],
            }

            sold_listings.append(listing)

    except (KeyError, IndexError) as e:
        print(f"Error parsing eBay response: {e}")
        return []

    return sold_listings


def get_sales_statistics(sold_listings: List[Dict]) -> Dict:
    """
    Calculate statistics from sold listings

    Returns:
        Dict with avg, min, max, median prices and sale counts
    """
    if not sold_listings:
        return {
            "count": 0,
            "avg_price": 0,
            "min_price": 0,
            "max_price": 0,
            "median_price": 0,
        }

    prices = [item["price_usd"] for item in sold_listings]
    prices_sorted = sorted(prices)

    median_idx = len(prices_sorted) // 2
    median_price = prices_sorted[median_idx] if len(prices_sorted) % 2 == 1 else (prices_sorted[median_idx - 1] + prices_sorted[median_idx]) / 2

    return {
        "count": len(sold_listings),
        "avg_price": sum(prices) / len(prices),
        "min_price": min(prices),
        "max_price": max(prices),
        "median_price": median_price,
        "listings": sold_listings,
    }


if __name__ == "__main__":
    # Test scraper
    print("Testing eBay sold listings scraper...\n")

    test_query = "Hermes Birkin 35"

    print(f"Searching for: {test_query}")
    print(f"Date range: Last 90 days\n")

    try:
        results = search_sold_listings(test_query, days_back=90, max_results=50)

        print(f"Found {len(results)} sold listings")

        if results:
            stats = get_sales_statistics(results)
            print(f"\n📊 Sales Statistics:")
            print(f"   Count: {stats['count']}")
            print(f"   Average: ${stats['avg_price']:,.2f}")
            print(f"   Median: ${stats['median_price']:,.2f}")
            print(f"   Min: ${stats['min_price']:,.2f}")
            print(f"   Max: ${stats['max_price']:,.2f}")

            print(f"\n📦 Recent sales:")
            for item in results[:5]:
                print(f"   • ${item['price_usd']:,.2f} - {item['condition']} - {item['end_time'][:10]}")
                print(f"     {item['title'][:80]}")

    except Exception as e:
        print(f"❌ Error: {e}")